保证一个TaskTracker只能有一个Logger, 因为一个jvm可以有多个TaskTracker
172181
public static BizLogger getLogger(Level level, RemotingClientDelegate remotingClient, TaskTrackerAppContext appContext) {

        // 单元测试的时候返回 Mock
        if (Environment.UNIT_TEST == LTSConfig.getEnvironment()) {
            return new MockBizLogger(level);
        }

        String key = appContext.getConfig().getIdentity();
        BizLogger logger = BIZ_LOGGER_CONCURRENT_HASH_MAP.get(key);
        if (logger == null) {
            synchronized (BIZ_LOGGER_CONCURRENT_HASH_MAP) {
                logger = BIZ_LOGGER_CONCURRENT_HASH_MAP.get(key);
                if (logger != null) {
                    return logger;
                }
                logger = new BizLoggerImpl(level, remotingClient, appContext);

                BIZ_LOGGER_CONCURRENT_HASH_MAP.put(key, logger);
            }
        }
        return logger;
    }
276423
def full_tasktrace(self):
        
        if self.prev_error:
            return self.prev_error.tasktrace + self.tasktrace
        else:
            return self.tasktrace
1096397
void lostTaskTracker(TaskTracker taskTracker) {
    String trackerName = taskTracker.getTrackerName();
    LOG.info("Lost tracker '" + trackerName + "'");

    // remove the tracker from the local structures
    synchronized (trackerToJobsToCleanup) {
      trackerToJobsToCleanup.remove(trackerName);
    }

    synchronized (trackerToTasksToCleanup) {
      trackerToTasksToCleanup.remove(trackerName);
    }

    Set<TaskAttemptIDWithTip> lostTasks = trackerToTaskMap.get(trackerName);
    trackerToTaskMap.remove(trackerName);

    if (lostTasks != null) {
      // List of jobs which had any of their tasks fail on this tracker
      Set<JobInProgress> jobsWithFailures = new HashSet<JobInProgress>();
      for (TaskAttemptIDWithTip oneTask : lostTasks) {
        TaskAttemptID taskId = oneTask.attemptId;
        TaskInProgress tip = oneTask.tip;
        JobInProgress job = (JobInProgress) tip.getJob();

        // Completed reduce tasks never need to be failed, because
        // their outputs go to dfs
        // And completed maps with zero reducers of the job
        // never need to be failed.
        if (!tip.isComplete() ||
            (tip.isMapTask() && !tip.isJobSetupTask() &&
             job.desiredReduces() != 0)) {
          // if the job is done, we don't want to change anything
          if (job.getStatus().getRunState() == JobStatus.RUNNING ||
              job.getStatus().getRunState() == JobStatus.PREP) {
            // the state will be KILLED_UNCLEAN, if the task(map or reduce)
            // was RUNNING on the tracker
            TaskStatus.State killState = (tip.isRunningTask(taskId) &&
              !tip.isJobSetupTask() && !tip.isJobCleanupTask()) ?
              TaskStatus.State.KILLED_UNCLEAN : TaskStatus.State.KILLED;
            job.failedTask(tip, taskId,
                          ("Lost task tracker: " + trackerName +
                            " at " + new Date()),
                           (tip.isMapTask() ?
                               TaskStatus.Phase.MAP :
                               TaskStatus.Phase.REDUCE),
                            killState,
                            trackerName);
            jobsWithFailures.add(job);
          }
        } else {
          // Completed 'reduce' task and completed 'maps' with zero
          // reducers of the job, not failed;
          // only removed from data-structures.
          markCompletedTaskAttempt(trackerName, taskId);
        }
      }

      // Penalize this tracker for each of the jobs which
      // had any tasks running on it when it was 'lost'
      // Also, remove any reserved slots on this tasktracker
      for (JobInProgress job : jobsWithFailures) {
        String reason = "Tracker went down";
        job.addTrackerTaskFailure(trackerName, taskTracker, reason);
      }

      // Cleanup
      taskTracker.cancelAllReservations();

      // Purge 'marked' tasks, needs to be done
      // here to prevent hanging references!
      removeMarkedTasks(trackerName);
    }
  }
614937
public void updateTaskTrackerAvailableThreads(
            String nodeGroup,
            String identity,
            Integer availableThreads,
            Long timestamp) {

        Set<TaskTrackerNode> taskTrackerNodes = NODE_MAP.get(nodeGroup);

        if (taskTrackerNodes != null && taskTrackerNodes.size() != 0) {
            for (TaskTrackerNode trackerNode : taskTrackerNodes) {
                if (trackerNode.getIdentity().equals(identity) && (trackerNode.getTimestamp() == null || trackerNode.getTimestamp() <= timestamp)) {
                    trackerNode.setAvailableThread(availableThreads);
                    trackerNode.setTimestamp(timestamp);
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("更新节点线程数: {}", trackerNode);
                    }
                }
            }
        }
    }
847987
def task_log(self):
        
        if self.task_id is None:
            raise ValueError('task_id is None')
        return self.get_task_log(self.task_id, self.session, self.request_kwargs)
1150351
protected void log( Throwable t )
  {
    if( _logger == null )
    {
      t.printStackTrace();
    }
    else
    {
      _logger.warn( "Error running job.", t );
    }
  }
1100662
def tasktiger_processor(logger, method_name, event_dict):
    

    if g['current_tasks'] is not None and not g['current_task_is_batch']:
        event_dict['task_id'] = g['current_tasks'][0].id

    return event_dict
1247387
synchronized public TaskTrackerStatus getTaskTrackerStatus(String trackerID) {
    TaskTracker taskTracker;
    synchronized (taskTrackers) {
      taskTracker = taskTrackers.get(trackerID);
    }
    return (taskTracker == null) ? null : taskTracker.getStatus();
  }
878839
static void writeTracker(JSONWriter writer, Tracker tracker)
			throws JSONException {
		writer.key("id");
		writer.value(tracker.getId());
		writer.key("name");
		writer.value(tracker.getName());
	}
572553
synchronized Long getMemReservedForTasks(
      TaskTrackerStatus taskTracker, TaskType taskType) {
    long vmem = 0;

    for (TaskStatus task : taskTracker.getTaskReports()) {
      // the following task states are one in which the slot is
      // still occupied and hence memory of the task should be
      // accounted in used memory.
      if ((task.getRunState() == TaskStatus.State.RUNNING) ||
          (task.getRunState() == TaskStatus.State.UNASSIGNED) ||
          (task.inTaskCleanupPhase())) {
        // Get the memory "allotted" for this task based on number of slots
        long myVmem = 0;
        if (task.getIsMap() && taskType == TaskType.MAP) {
          long memSizePerMapSlot = scheduler.getMemSizeForMapSlot(); 
          myVmem = 
            memSizePerMapSlot * task.getNumSlots();
        } else if (!task.getIsMap()
            && taskType == TaskType.REDUCE) {
          long memSizePerReduceSlot = scheduler.getMemSizeForReduceSlot(); 
          myVmem = memSizePerReduceSlot * task.getNumSlots();
        }
        vmem += myVmem;
      }
    }

    return Long.valueOf(vmem);
  }
1238206
@Override
  public void register(TaskExecutor taskExecutor) {
	lock.lock();
	try {
	  Iterator<TaskExecutor> it = taskExecutors.iterator();
	  while (it.hasNext()) {
		TaskExecutor e = it.next();
		if (e.isShutdown() && e.hasTask() == false) {
		  it.remove();
		}
	  }

	  if (taskExecutors.contains(taskExecutor) == false) {
		taskExecutors.add(taskExecutor);
	  }
	} finally {
	  lock.unlock();
	}
  }
902845
def _save_trace(self):
        
        stack_trace = stack()
        try:
            self.trace = []
            for frm in stack_trace[5:]: # eliminate our own overhead
                self.trace.insert(0, frm[1:])
        finally:
            del stack_trace
214859
public static void logAllThreads() {
    StringBuilder sb = new StringBuilder("Dumping all threads:\n");
    for (Thread t : Thread.getAllStackTraces().keySet()) {
      sb.append(formatStackTrace(t));
    }
    LOG.info(sb.toString());
  }
122635
def task(name=None, t=INFO, *args, **kwargs):
    

    def c_run(name, f, t, args, kwargs):
        def run(*largs, **lkwargs):
            thread = __get_current_thread()
            old_name = __THREAD_PARAMS[thread][__THREAD_PARAMS_FNAME_KEY]
            __THREAD_PARAMS[thread][__THREAD_PARAMS_FNAME_KEY] = name

            r = log(name, f, t, largs, lkwargs, *args, **kwargs)

            __THREAD_PARAMS[thread][__THREAD_PARAMS_FNAME_KEY] = old_name
            return r

        return run

    if callable(name):
        f = name
        name = f.__name__

        return c_run(name, f, t, args, kwargs)

    if name == None:
        def wrapped(f):
            name = f.__name__
            return c_run(name, f, t, args, kwargs)

        return wrapped
    else:
        return lambda f: c_run(name, f, t, args, kwargs)
644425
public static task_log[] get(nitro_service client) throws Exception
	{
		task_log resource = new task_log();
		resource.validate("get");
		return (task_log[]) resource.get_resources(client);
	}
1086362
public static void main(String argv[]) throws Exception {
    StringUtils.startupShutdownMessage(TaskTracker.class, argv, LOG);
    try {
      if (argv.length == 0) {
        JobConf conf = new JobConf();
        ReflectionUtils.setContentionTracing
        (conf.getBoolean("tasktracker.contention.tracking", false));
        new TaskTracker(conf).run();
        return;
      }
      if ("-instance".equals(argv[0]) && argv.length == 2) {
        int instance = Integer.parseInt(argv[1]);
        if (instance == 0 || instance == 1) {
          JobConf conf = new JobConf();
          JobConf.overrideConfiguration(conf, instance);
          // enable the server to track time spent waiting on locks
          ReflectionUtils.setContentionTracing
          (conf.getBoolean("tasktracker.contention.tracking", false));
          new TaskTracker(conf).run();
          return;
        }
      }
      System.out.println("usage: TaskTracker [-instance <0|1>]");
      System.exit(-1);
    } catch (Throwable e) {
      LOG.error("Can not start task tracker because "+
                StringUtils.stringifyException(e));
      System.exit(-1);
    }
  }
1032356
List<Task> getSetupAndCleanupTasks(
    TaskTrackerStatus taskTracker) throws IOException {
    int maxMapTasks = taskScheduler.getMaxSlots(taskTracker, TaskType.MAP);
    int maxReduceTasks =
        taskScheduler.getMaxSlots(taskTracker, TaskType.REDUCE);
    int numMaps = taskTracker.countOccupiedMapSlots();
    int numReduces = taskTracker.countOccupiedReduceSlots();
    int numTaskTrackers = getClusterStatus().getTaskTrackers();
    int numUniqueHosts = getNumberOfUniqueHosts();

    List<JobInProgress> cachedJobs = new ArrayList<JobInProgress> ();

    // get a snapshot of all the jobs in the system
    synchronized (jobs) {
      cachedJobs.addAll(jobs.values());
    }

    Task t = null;

    if (numMaps < maxMapTasks) {
      for (JobInProgress job: cachedJobs) {
        t = job.obtainJobCleanupTask(taskTracker, numTaskTrackers,
                                     numUniqueHosts, true);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
      for (JobInProgress job: cachedJobs) {
        t = job.obtainTaskCleanupTask(taskTracker, true);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
      for (JobInProgress job: cachedJobs) {
        t = job.obtainJobSetupTask(taskTracker, numTaskTrackers,
                                   numUniqueHosts, true);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
    }
    if (numReduces < maxReduceTasks) {
      for (JobInProgress job: cachedJobs) {
        t = job.obtainJobCleanupTask(taskTracker, numTaskTrackers,
                                     numUniqueHosts, false);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
      for (JobInProgress job: cachedJobs) {
        t = job.obtainTaskCleanupTask(taskTracker, false);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
      for (JobInProgress job: cachedJobs) {
        t = job.obtainJobSetupTask(taskTracker, numTaskTrackers,
                                   numUniqueHosts, false);
        if (t != null) {
          return Collections.singletonList(t);
        }
      }
    }

    return null;
  }