Remove a reduce TIP from the list for running-reduces
Called when a reduce fails/completes
@param tip the tip that needs to be retired
771514
private synchronized void retireReduce(TaskInProgress tip) {
    if (runningReduces == null) {
      LOG.warn("Running list for reducers missing!! "
               + "Job details are missing.");
      return;
    }
    runningReduces.remove(tip);
  }
1197317
@Override
  public void run() {
    while (true) {
      // Wait to get a TIP
      TipToFinish ttf = null;
      try {
        LOG.debug("Waiting for a TIP");
        ttf = tipQueue.take();
      } catch (InterruptedException e) {
        LOG.info("Got interrupted exception while waiting to take()");
        continue;
      }
      LOG.debug(" Got a TIP " + ttf.getTip().getTask().getTaskID() +
          " at time " + System.currentTimeMillis() + " with finish time " +
          ttf.getTimeToFinish());
      // Wait until it's time to finish the task. Since the TIP was pulled from
      // the priority queue, this should be the first task in the queue that
      // needs to be finished. If we get interrupted, that means that it's
      // possible that we added a TIP that should finish earlier
      boolean interrupted = false;
      while (true) {
        long currentTime = System.currentTimeMillis();
        if (currentTime < ttf.getTimeToFinish()) {
          try {
            long sleepTime = ttf.getTimeToFinish() - currentTime;
            LOG.debug("Sleeping for " + sleepTime + " ms");
            Thread.sleep(sleepTime);
          } catch (InterruptedException e) {
            LOG.debug("Finisher thread was interrupted", e);
            interrupted = true;
            break;
          }
        } else {
          break;
        }
      }

      // Wait was interrupted, then it could mean that we added a task
      // that needs to finish sooner. Put that task back and start again
      if (interrupted) {
        LOG.info("Putting back TIP " + ttf.getTip().getTask().getTaskID() +
            " for job " + ttf.getTip().getTask().getJobID());
        tipQueue.put(ttf);
        continue;
      }

      // Finish the task
      TaskInProgress tip = ttf.getTip();
      ttf.finishTip();

      // Also clean up the mapper wait thread map for reducers. It should exist
      // for reduce tasks that are not cleanup tasks
      if (!tip.getTask().isMapTask() &&
          !tip.getTask().isTaskCleanupTask() &&
          !tip.getTask().isJobCleanupTask() &&
          !tip.getTask().isJobSetupTask()) {
        if (!mapperWaitThreadMap.containsKey(tip)) {
          throw new RuntimeException("Unable to find mapper wait thread for " +
              tip.getTask().getTaskID() + " job " + tip.getTask().getJobID());
        }
        LOG.debug("Removing mapper wait thread for " +
            tip.getTask().getTaskID() + " job " + tip.getTask().getJobID());
        mapperWaitThreadMap.remove(tip);
      } else if (mapperWaitThreadMap.containsKey(tip)) {
        throw new RuntimeException("Mapper wait thread exists for" +
            tip.getTask().getTaskID() + " job " + tip.getTask().getJobID() +
            " when it shouldn't!");
      }
    }
  }
752430
public void remove(PropertyKey key) {
    // remove is a nop if the key doesn't already exist
    if (mUserProps.containsKey(key)) {
      mUserProps.remove(key);
      mSources.remove(key);
    }
  }
198761
public void delete(int key) {
        int i = ContainerHelpers.binarySearch(mKeys, mSize, key);

        if (i >= 0) {
            if (mValues[i] != DELETED) {
                mValues[i] = DELETED;
                mGarbage = true;
            }
        }
    }
739623
private synchronized void retireMap(TaskInProgress tip) {
    if (runningMapCache == null) {
      LOG.warn("Running cache for maps missing!! "
               + "Job details are missing.");
      return;
    }

    String[] splitLocations = tip.getSplitLocations();

    // Remove the TIP from the list for running non-local maps
    if (splitLocations.length == 0) {
      nonLocalRunningMaps.remove(tip);
      return;
    }

    // Remove from the running map caches
    for(String host: splitLocations) {
      Node node = jobtracker.getNode(host);

      for (int j = 0; j < maxLevel; ++j) {
        Set<TaskInProgress> hostMaps = runningMapCache.get(node);
        if (hostMaps != null) {
          hostMaps.remove(tip);
          if (hostMaps.size() == 0) {
            runningMapCache.remove(node);
          }
        }
        node = node.getParent();
      }
    }
  }
678735
public void remove(K key) {

        V removedValue = m_forwardMap.remove(key);
        if (removedValue != null) {
            m_reverseMap.remove(removedValue, key);
        }
    }
563774
public void reduceThis(){
		if (elements==null || elements.isEmpty())
			throw new AssertionError("Can't reduce this environment");
		elements.remove(elements.size()-1);
	}
169059
@Override
	public final void reduce(Iterable<IN> values, Collector<OUT> out) throws Exception {
		for (Iterator<OUT> iter = reduceGroup(values); iter.hasNext(); ) {
			out.collect(iter.next());
		}
	}
1013279
private void removePoint(Point point, Collidable collidable)
    {
        final Integer group = collidable.getGroup();
        if (collidables.containsKey(group))
        {
            final Map<Point, Set<Collidable>> elements = collidables.get(group);
            if (elements.containsKey(point))
            {
                removePoint(point, collidable, elements, group);
            }
        }
    }
1116975
public void truncate(long newTruncationPoint) {
        if (newTruncationPoint < getFirstDrId()) {
            return;
        }
        final Iterator<Range<Long>> iter = m_map.asRanges().iterator();
        while (iter.hasNext()) {
            final Range<Long> next = iter.next();
            if (end(next) < newTruncationPoint) {
                iter.remove();
            } else if (next.contains(newTruncationPoint)) {
                iter.remove();
                m_map.add(range(newTruncationPoint, end(next)));
                return;
            } else {
                break;
            }
        }
        m_map.add(range(newTruncationPoint, newTruncationPoint));
    }
88982
private static TaskInProgress findTaskFromList(
      Collection<TaskInProgress> tips, String hostName,
      boolean removeFailedTip) {
    Iterator<TaskInProgress> iter = tips.iterator();
    while (iter.hasNext()) {
      TaskInProgress tip = iter.next();

      // Select a tip if
      //   1. runnable   : still needs to be run and is not completed
      //   2. ~running   : no other node is running it
      //   3. earlier attempt failed : has not failed on this host
      //                               and has failed on all the other hosts
      // A TIP is removed from the list if
      // (1) this tip is scheduled
      // (2) if the passed list is a level 0 (host) cache
      // (3) when the TIP is non-schedulable (running, killed, complete)
      if (tip.isRunnable() && !tip.isRunning()) {
        // check if the tip has failed on this host
        if (!tip.hasFailedOnMachine(hostName)) {
          // TODO: check if the tip has failed on all the nodes
          iter.remove();
          return tip;
        } else if (removeFailedTip) {
          // the case where we want to remove a failed tip from the host cache
          // point#3 in the TIP removal logic above
          iter.remove();
        }
      } else {
        // see point#3 in the comment above for TIP removal logic
        iter.remove();
      }
    }
    return null;
  }
432590
public synchronized boolean completedTask(TaskInProgress tip,
                                            TaskStatus status)
  {
    TaskAttemptID taskid = status.getTaskID();
    int oldNumAttempts = tip.getActiveTasks().size();
    final JobTrackerInstrumentation metrics = jobtracker.getInstrumentation();

    // Metering
    meterTaskAttempt(tip, status);

    // Sanity check: is the TIP already complete?
    // It _is_ safe to not decrement running{Map|Reduce}Tasks and
    // finished{Map|Reduce}Tasks variables here because one and only
    // one task-attempt of a TIP gets to completedTask. This is because
    // the TaskCommitThread in the JobTracker marks other, completed,
    // speculative tasks as _complete_.
    if (tip.isComplete()) {
      // Mark this task as KILLED
      tip.alreadyCompletedTask(taskid);

      // Let the JobTracker cleanup this taskid if the job isn't running
      if (this.status.getRunState() != JobStatus.RUNNING) {
        jobtracker.markCompletedTaskAttempt(status.getTaskTracker(), taskid);
      }
      return false;
    }


    LOG.info("Task '" + taskid + "' has completed " + tip.getTIPId() +
             " successfully.");

    // Mark the TIP as complete
    tip.completed(taskid);
    resourceEstimator.updateWithCompletedTask(status, tip);

    // Update jobhistory
    TaskTrackerStatus ttStatus =
      this.jobtracker.getTaskTrackerStatus(status.getTaskTracker());
    String trackerHostname = jobtracker.getNode(ttStatus.getHost()).toString();
    String taskType = getTaskType(tip);
    if (status.getIsMap()){
      JobHistory.MapAttempt.logStarted(status.getTaskID(), status.getStartTime(),
                                       status.getTaskTracker(),
                                       ttStatus.getHttpPort(),
                                       taskType);
      JobHistory.MapAttempt.logFinished(status.getTaskID(), status.getFinishTime(),
                                        trackerHostname, taskType,
                                        status.getStateString(),
                                        status.getCounters());
    }else{
      JobHistory.ReduceAttempt.logStarted( status.getTaskID(), status.getStartTime(),
                                          status.getTaskTracker(),
                                          ttStatus.getHttpPort(),
                                          taskType);
      JobHistory.ReduceAttempt.logFinished(status.getTaskID(), status.getShuffleFinishTime(),
                                           status.getSortFinishTime(), status.getFinishTime(),
                                           trackerHostname,
                                           taskType,
                                           status.getStateString(),
                                           status.getCounters());
    }
    JobHistory.Task.logFinished(tip.getTIPId(),
                                taskType,
                                tip.getExecFinishTime(),
                                status.getCounters());

    int newNumAttempts = tip.getActiveTasks().size();
    if (tip.isJobSetupTask()) {
      // setup task has finished. kill the extra setup tip
      killSetupTip(!tip.isMapTask());
      setupComplete();
    } else if (tip.isJobCleanupTask()) {
      // cleanup task has finished. Kill the extra cleanup tip
      if (tip.isMapTask()) {
        // kill the reduce tip
        cleanup[1].kill();
      } else {
        cleanup[0].kill();
      }

      //
      // The Job is done
      // if the job is failed, then mark the job failed.
      if (jobFailed) {
        terminateJob(JobStatus.FAILED);
      }
      // if the job is killed, then mark the job killed.
      if (jobKilled) {
        terminateJob(JobStatus.KILLED);
      }
      else {
        jobComplete();
      }
      // The job has been killed/failed/successful
      // JobTracker should cleanup this task
      jobtracker.markCompletedTaskAttempt(status.getTaskTracker(), taskid);
    } else if (tip.isMapTask()) {
      // check if this was a sepculative task
      if (oldNumAttempts > 1) {
        speculativeMapTasks -= (oldNumAttempts - newNumAttempts);
        if (!garbageCollected) {
          totalSpeculativeMapTasks.addAndGet(newNumAttempts - oldNumAttempts);
        }
      }
      if (tip.isSpeculativeAttempt(taskid)) {
        metrics.speculativeSucceededMap(taskid,
            tip.isUsingProcessingRateForSpeculation());
      }
      int level = getLocalityLevel(tip, ttStatus);
      long inputBytes = tip.getCounters()
                .getGroup("org.apache.hadoop.mapred.Task$Counter")
                .getCounter("Map input bytes");
      switch (level) {
      case 0: jobCounters.incrCounter(Counter.LOCAL_MAP_INPUT_BYTES,
                inputBytes);
              metrics.addLocalMapInputBytes(inputBytes);
              break;
      case 1: jobCounters.incrCounter(Counter.RACK_MAP_INPUT_BYTES,
                inputBytes);
              metrics.addRackMapInputBytes(inputBytes);
              break;
      default:metrics.addMapInputBytes(inputBytes);
              break;
      }
      finishedMapTasks += 1;
      metrics.completeMap(taskid);
      if (!garbageCollected) {
        if (!tip.isJobSetupTask() && hasSpeculativeMaps) {
          updateTaskTrackerStats(tip,ttStatus,trackerMapStats,mapTaskStats);
        }
      }
      // remove the completed map from the resp running caches
      retireMap(tip);
      if ((finishedMapTasks + failedMapTIPs) == (numMapTasks)) {
        this.status.setMapProgress(1.0f);
      }
    } else {
      if (oldNumAttempts > 1) {
        speculativeReduceTasks -= (oldNumAttempts - newNumAttempts);
        if (!garbageCollected) {
          totalSpeculativeReduceTasks.addAndGet(newNumAttempts - oldNumAttempts);
        }
      }
      if (tip.isSpeculativeAttempt(taskid)) {
        metrics.speculativeSucceededReduce(taskid,
            tip.isUsingProcessingRateForSpeculation());
      }
      finishedReduceTasks += 1;
      metrics.completeReduce(taskid);
      if (!garbageCollected) {
        if (!tip.isJobSetupTask() && hasSpeculativeReduces) {
          updateTaskTrackerStats(tip,ttStatus,trackerReduceStats,reduceTaskStats);
        }
      }
      // remove the completed reduces from the running reducers set
      retireReduce(tip);
      if ((finishedReduceTasks + failedReduceTIPs) == (numReduceTasks)) {
        this.status.setReduceProgress(1.0f);
      }
    }

    // is job complete?
    if (!jobSetupCleanupNeeded && canLaunchJobCleanupTask()) {
      jobComplete();
    }

    return true;
  }
51178
public void cancel(TaskInProgress tip) {
    LOG.info("Canceling task "  + tip.getTask().getTaskID() + " of job " +
        tip.getTask().getJobID());
    // Cancel & remove the map completion finish thread for reduce tasks.
    if (!tip.getTask().isMapTask() && !tip.getTask().isTaskCleanupTask()) {
      if (!mapperWaitThreadMap.containsKey(tip)) {
        throw new RuntimeException("Mapper wait thread doesn't exist " +
            "for " + tip.getTask().getTaskID());
      }
      LOG.debug("Interrupting mapper wait thread for " +
          tip.getTask().getTaskID() + " job " +
          tip.getTask().getJobID());
      mapperWaitThreadMap.get(tip).interrupt();
      LOG.debug("Removing mapper wait thread for " +
          tip.getTask().getTaskID() + " job " + tip.getTask().getJobID());
      mapperWaitThreadMap.remove(tip);
    } else {
      LOG.debug(tip.getTask().getTaskID() + " is not a reduce task, so " +
          "not canceling mapper wait thread");
    }
    removeFromFinishingQueue(tip);
  }
391076
public void failUnacknowledgedPendingCheckpointsFor(ExecutionAttemptID executionAttemptId, Throwable cause) {
		synchronized (lock) {
			Iterator<PendingCheckpoint> pendingCheckpointIterator = pendingCheckpoints.values().iterator();

			while (pendingCheckpointIterator.hasNext()) {
				final PendingCheckpoint pendingCheckpoint = pendingCheckpointIterator.next();

				if (!pendingCheckpoint.isAcknowledgedBy(executionAttemptId)) {
					pendingCheckpointIterator.remove();
					discardCheckpoint(pendingCheckpoint, cause);
				}
			}
		}
	}
130193
private boolean remove(List<String> list, String item) {
        if (null != list) {
            if (list.remove(item)) {
                this.num_items--;
                return true;
            }
        }
        return false;
    }
911328
@UiThread
  public void delete(List<T> annotationList) {
    for (T annotation : annotationList) {
      annotations.remove(annotation.getId());
    }
    updateSource();
  }
1258537
private void failMapUnprotected(TaskInProgress tip) {
    if (nonRunningMaps == null) {
      LOG.warn("Non-running cache for maps missing!! "
               + "Job details are missing.");
      return;
    }
    nonRunningMaps.add(tip);
  }