Create a local reduce input file name.
@param mapTaskId a map task id
@param reduceTaskId a reduce task id
@param size the size of the file
976950
public Path getInputFileForWrite(TaskID mapId, TaskAttemptID reduceTaskId, 
                                   long size)
    throws IOException {
    // TODO *oom* should use a format here
    return lDirAlloc.getLocalPathForWrite(TaskTracker.getIntermediateOutputDir(
                       jobId.toString(), reduceTaskId.toString())
                       + "/map_" + mapId.getId() + ".out", 
                       size, conf);
  }
1074322
@Override
	public void open(int taskNumber, int numTasks) throws IOException {
		this.fileOutputCommitterWrapper.setupJob(this.jobConf);
		if (Integer.toString(taskNumber + 1).length() <= 6) {
			this.jobConf.set("mapred.task.id", "attempt__0000_r_" + String.format("%" + (6 - Integer.toString(taskNumber + 1).length()) + "s"," ").replace(" ", "0") + Integer.toString(taskNumber + 1) + "_0");
			//compatible for hadoop 2.2.0, the temporary output directory is different from hadoop 1.2.1
			this.jobConf.set("mapreduce.task.output.dir", this.fileOutputCommitterWrapper.getTempTaskOutputPath(this.jobConf,TaskAttemptID.forName(this.jobConf.get("mapred.task.id"))).toString());
		} else {
			throw new IOException("task id too large");
		}
		this.recordWriter = this.hadoopOutputFormat.getRecordWriter(null, this.jobConf, Integer.toString(taskNumber + 1), new DummyHadoopProgressable());
	}
1232423
public ScanWriter createScanWriter(int taskId, ScanDestination destination) {
        if (destination.isDiscarding()) {
            return _scanWriterFactory.createDiscardingScanWriter(taskId, Optional.<Integer>absent());
        }

        URI uri = destination.getUri();
        String scheme = uri.getScheme();

        if ("file".equals(scheme)) {
            return _scanWriterFactory.createFileScanWriter(taskId, uri, Optional.<Integer>absent());
        }

        if ("s3".equals(scheme)) {
            return _scanWriterFactory.createS3ScanWriter(taskId, uri, Optional.<Integer>absent());
        }

        throw new IllegalArgumentException("Unsupported destination: " + destination);
    }
236101
public void putMemorySize(String key, MemorySize size) {
		checkNotNull(key);
		checkNotNull(size);
		put(key, size.toString());
	}
504987
public void setTextSize(final TextSize SIZE) {
        if (null == textSize) {
            _textSize = SIZE;
            fireTileEvent(REDRAW_EVENT);
        } else {
            textSize.set(SIZE);
        }
    }
209251
public void removeAll(TaskAttemptID taskId) throws IOException {
    String toBeDeleted =
      TaskTracker.getIntermediateOutputDir(jobId.toString(), taskId.toString());
    if (asyncDiskService != null) {
      asyncDiskService.moveAndDeleteFromEachVolume(toBeDeleted);
      LOG.info("Move and then delete map ouput " +
               toBeDeleted + " for task " + taskId);
      return;
    }
    LOG.info("Delete map ouput " + toBeDeleted + " for task " + taskId);
    conf.deleteLocalFiles(toBeDeleted);
  }
1254758
private void updateTotalTaskCapacity(TaskTrackerStatus status) {
    int mapSlots = taskScheduler.getMaxSlots(status, TaskType.MAP);
    String trackerName = status.getTrackerName();
    Integer oldMapSlots = trackerNameToMapSlots.get(trackerName);
    if (oldMapSlots == null) {
      oldMapSlots = 0;
    }
    int delta = mapSlots - oldMapSlots;

    if (delta != 0) {
      totalMapTaskCapacity += delta;
      trackerNameToMapSlots.put(trackerName, mapSlots);

      LOG.info("Changing map slot count due to " + trackerName + " from " +
               oldMapSlots + " to " + mapSlots + ", totalMap = " + totalMapTaskCapacity);
    }

    int reduceSlots = taskScheduler.getMaxSlots(status, TaskType.REDUCE);
    Integer oldReduceSlots = trackerNameToReduceSlots.get(trackerName);
    if (oldReduceSlots == null) {
      oldReduceSlots = 0;
    }

    delta = reduceSlots - oldReduceSlots;

    if (delta != 0) {
      totalReduceTaskCapacity += delta;
      trackerNameToReduceSlots.put(trackerName, reduceSlots);

      LOG.info("Changing reduce slot count due to " + trackerName + " from " +
               oldReduceSlots + " to " + reduceSlots +
               ", totalReduce = " + totalReduceTaskCapacity);
    }
  }
458189
public static StaticMapsRequest newRequest(GeoApiContext context, Size size) {
    return new StaticMapsRequest(context).size(size);
  }
1237420
private void setupTaskCacheFileAccess(TaskControllerContext context) {
    String taskId = context.task.getTaskID().toString();
    JobID jobId = JobID.forName(getJobId(context));
    //Change permission for the task across all the disks
    for(String localDir : mapredLocalDirs) {
      File f = new File(localDir);
      File taskCacheDir = new File(f,TaskTracker.getLocalTaskDir(
          jobId.toString(), taskId, context.task.isTaskCleanupTask()));
      if(taskCacheDir.exists()) {
        changeDirectoryPermissions(taskCacheDir.getPath(), 
            FILE_PERMISSIONS, true);
      }          
    }//end of local directory Iteration 
  }
473418
public int addTask(Task task, Object remoteTask)
    {
        if (m_taskMain == null)
        {
            if (task != null)
                if (task.isMainTaskCandidate())
                    m_taskMain = task;
        }
        m_mapTasks.put(task, (RemoteTask)remoteTask);
        return m_mapTasks.size();
    }
583665
public static <K, V> HadoopInputFormat<K, V> readHadoopFile(org.apache.hadoop.mapred.FileInputFormat<K, V> mapredInputFormat, Class<K> key, Class<V> value, String inputPath) {
		return readHadoopFile(mapredInputFormat, key, value, inputPath, new JobConf());
	}
40196
public static void setPreferredRoadBorderSize(Integer size) {
		final Preferences prefs = Preferences.userNodeForPackage(RoadNetworkLayerConstants.class);
		if (prefs != null) {
			if (size == null || size.intValue() == DEFAULT_BORDER_SIZE) {
				prefs.remove("ROAD_BORDER_SIZE"); //$NON-NLS-1$
			} else {
				prefs.putInt("ROAD_BORDER_SIZE", size.intValue()); //$NON-NLS-1$
			}
			try {
				prefs.flush();
			} catch (BackingStoreException exception) {
				//
			}
		}
	}
1050998
private static void fillInMissingMapOutputs(FileSystem fs, 
                                              TaskAttemptID taskId,
                                              int numMaps,
                                              JobConf conf) throws IOException {
    Class<? extends WritableComparable> keyClass
        = conf.getMapOutputKeyClass().asSubclass(WritableComparable.class);
    Class<? extends Writable> valueClass
        = conf.getMapOutputValueClass().asSubclass(Writable.class);
    MapOutputFile namer = new MapOutputFile(taskId.getJobID());
    namer.setConf(conf);
    for(int i=0; i<numMaps; i++) {
      Path f = namer.getInputFile(i, taskId);
      if (!fs.exists(f)) {
        LOG.info("Create missing input: " + f);
        SequenceFile.Writer out =
          SequenceFile.createWriter(fs, conf, f, keyClass, valueClass);
        out.close();
      }
    }    
  }
229797
@Override
	public void setup(TaskContext<GroupReduceFunction<IT, OT>, OT> context) {
		this.taskContext = context;
	}
186117
void writeMapHeader(int size) throws IOException {
        if (size < 16) {
            output.writeByte(MIN_FIX_MAP + size);
        } else if (size < 65536) {
            output.writeByte(MAP_16);
            output.writeShort(size);
        } else {
            output.writeByte(MAP_32);
            output.writeInt(size);
        }
    }
181133
@SuppressWarnings("rawtypes")
  public ReducedMetric registerMetric(String name, IReducer reducer, int timeBucketSizeInSecs) {
    return registerMetric(name, new ReducedMetric(reducer), timeBucketSizeInSecs);
  }
1131607
private synchronized void buildMaps() {
    if (loggedTaskMap == null) {
      loggedTaskMap = new HashMap<TaskID, LoggedTask>();
      loggedTaskAttemptMap = new HashMap<TaskAttemptID, LoggedTaskAttempt>();
      
      for (LoggedTask map : job.getMapTasks()) {
        map = sanitizeLoggedTask(map);
        if (map != null) {
          loggedTaskMap.put(maskTaskID(TaskID.forName(map.taskID)), map);

          for (LoggedTaskAttempt mapAttempt : map.getAttempts()) {
            mapAttempt = sanitizeLoggedTaskAttempt(mapAttempt);
            if (mapAttempt != null) {
              TaskAttemptID id = TaskAttemptID.forName(mapAttempt
                  .getAttemptID());
              loggedTaskAttemptMap.put(maskAttemptID(id), mapAttempt);
            }
          }
        }
      }
      for (LoggedTask reduce : job.getReduceTasks()) {
        reduce = sanitizeLoggedTask(reduce);
        if (reduce != null) {
          loggedTaskMap.put(maskTaskID(TaskID.forName(reduce.taskID)), reduce);

          for (LoggedTaskAttempt reduceAttempt : reduce.getAttempts()) {
            reduceAttempt = sanitizeLoggedTaskAttempt(reduceAttempt);
            if (reduceAttempt != null) {
              TaskAttemptID id = TaskAttemptID.forName(reduceAttempt
                  .getAttemptID());
              loggedTaskAttemptMap.put(maskAttemptID(id), reduceAttempt);
            }
          }
        }
      }

      // TODO: do not care about "other" tasks, "setup" or "clean"
    }
  }