1. 将 channel 纳入管理中(不存在就加入)
2. 更新 TaskTracker 节点信息(可用线程数)
344184
private void offerHandler(Channel channel, RemotingCommand request) {
        AbstractRemotingCommandBody commandBody = request.getBody();
        String nodeGroup = commandBody.getNodeGroup();
        String identity = commandBody.getIdentity();
        NodeType nodeType = NodeType.valueOf(commandBody.getNodeType());

        // 1. 将 channel 纳入管理中(不存在就加入)
        appContext.getChannelManager().offerChannel(new ChannelWrapper(channel, nodeType, nodeGroup, identity));
    }
164924
function addTaskToFluence(task, done) {
        if (done) {
            done = "1"
        } else {
            done = "0"
        }
        const query = `INSERT INTO ${todoTable} VALUES ('${task}', ${done})`;
        session.request(query).result().then((r) => {
            console.log("task inserted")
        })
    }
703900
public void trackTask(final ServiceTaskBase<?> task) {

        LOGGER.trace("track a Task");

        this.serviceTasks.add(task);

        task.setOnCancelled(this.workerHandler);
        task.setOnSucceeded(this.workerHandler);
        task.setOnFailed(this.workerHandler);
    }
693389
def task_transaction(channel):
    """"""
    with channel.lock:
        if channel.poll(0):
            task = channel.recv()
            channel.send(Acknowledgement(os.getpid(), task.id))
        else:
            raise RuntimeError("Race condition between workers")

    return task
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);
    }
  }
218954
def _update_project_watch(config, task_presenter, results,
                          long_description, tutorial):  # pragma: no cover
    """"""
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    path = os.getcwd()
    event_handler = PbsHandler(config, task_presenter, results,
                               long_description, tutorial)
    observer = Observer()
    # We only want the current folder, not sub-folders
    observer.schedule(event_handler, path, recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()
414607
def task_add(self, t, periodic=None):
        
        name = t.get_name()
        if name in self._tasknames:
            raise TaskError(name, 'Task already exists with %d daemon%s active' %
                            (len(self._tasknames), ses(len(self._tasknames))))
        self._tasknames[name] = (t, periodic)
        self._tasks.add(t)
1068709
synchronized void addTrackerTaskFailure(String trackerName,
                                          TaskTracker taskTracker,
                                          String lastFailureReason) {
    if (flakyTaskTrackers < (clusterSize * CLUSTER_BLACKLIST_PERCENT)) {
      String trackerHostName = convertTrackerNameToHostName(trackerName);

      List<String> trackerFailures = trackerToFailuresMap.get(trackerHostName);

      if (trackerFailures == null) {
        trackerFailures = new LinkedList<String>();
        trackerToFailuresMap.put(trackerHostName, trackerFailures);
      }
      trackerFailures.add(lastFailureReason);

      // Check if this tasktracker has turned 'flaky'
      if (trackerFailures.size() == maxTaskFailuresPerTracker) {
        ++flakyTaskTrackers;

        // Cancel reservations if appropriate
        if (taskTracker != null) {
          if (trackersReservedForMaps.containsKey(taskTracker)) {
            taskTracker.unreserveSlots(TaskType.MAP, this);
          }
          if (trackersReservedForReduces.containsKey(taskTracker)) {
            taskTracker.unreserveSlots(TaskType.REDUCE, this);
          }
        }
        LOG.info("TaskTracker at '" + trackerHostName + "' turned 'flaky'");
      }
    }
  }
210913
public Task forceNewReduceTaskForTip(String taskTrackerName, String hostName,
      TaskInProgress forcedTip) {
    synchronized (lockObject) {
      Task result = obtainTaskCleanupTask(taskTrackerName, forcedTip);
      if (result != null) {
        return result;
      }

      removeMatchingTipUnprotectedUnconditional(nonRunningMaps, forcedTip);
      LOG.info("Running task " + forcedTip.getTIPId() + " on "
          + taskTrackerName + "(" + hostName + ")");
      scheduleReduceUnprotected(forcedTip);
      result = forcedTip.getTaskToRun(taskTrackerName);
      if (result != null) {
        addRunningTaskToTIPUnprotected(forcedTip, result.getTaskID(),
            taskTrackerName, hostName, true);
        // Handle cleanup task
        setJobCleanupTaskState(result);
      }
      return result;
    }
  }
617093
function addTask(taskData, subTaskData) {
			var subTaskList,
				parsedData,
				subTask = false;

			if(taskData !== undefined) {
				if(parseInt(taskData,10) > -1) {
					subTaskList = this.getTaskByID(parseInt(taskData,10)).subTasks;

					parsedData = this.parseInputData(subTaskData);
					subTask = true;
				}
				else {
					parsedData = this.parseInputData(taskData);
				}


				var newTask = _.clone(defaultTask)
				_.extend(newTask, parsedData);

				newTask.dateAdded = Date.now();
				newTask.id = largestID + 1;
				newTask.subTask = subTask;

				this.tasks.unshift( newTask );

				if(subTask) {
					subTaskList.push(newTask.id);
				}

				console.log('Added:');
				return newTask;
			}
			else {
				console.log('no task name given');
				return false;
			}

		}
882041
private void updateTaskStatuses(TaskTrackerStatus status) {
    TaskTrackerInfo trackerInfo = TaskTrackerInfo.fromStatus(status);
    String trackerName = status.getTrackerName();
    for (TaskStatus report : status.getTaskReports()) {
      // Ensure that every report has information about task tracker
      report.setTaskTracker(trackerName);
      LOG.debug("Task status report: " + report);
      updateTaskStatus(trackerInfo, report);
      // Take any actions outside updateTaskStatus()
      setupReduceRequests(job);
      processFetchFailures(report);
    }
  }
1054111
private void checkTasks() {
    if (currentTask == null) {
      Handler<Task> task = queue.poll();
      if (task != null) {
        currentTask = new Task(this);
        task.handle(currentTask);
      }
    }
  }
332192
synchronized List<TaskTrackerAction> getTasksToKill(String taskTracker) {

    Set<TaskAttemptIDWithTip> taskset = trackerToTaskMap.get(taskTracker);
    List<TaskTrackerAction> killList = new ArrayList<TaskTrackerAction>();
    if (taskset != null) {
      for (TaskAttemptIDWithTip onetask : taskset) {
        TaskAttemptID killTaskId = onetask.attemptId;
        TaskInProgress tip = onetask.tip;

        if (tip == null) {
          continue;
        }
        if (tip.shouldClose(killTaskId)) {
          //
          // This is how the JobTracker ends a task at the TaskTracker.
          // It may be successfully completed, or may be killed in
          // mid-execution.
          //
          if (!((JobInProgress)tip.getJob()).isComplete()) {
            killList.add(new KillTaskAction(killTaskId));
            LOG.debug(taskTracker + " -> KillTaskAction: " + killTaskId);
          }
        }
      }
    }

    // add the stray attempts for uninited jobs
    synchronized (trackerToTasksToCleanup) {
      Set<TaskAttemptID> set = trackerToTasksToCleanup.remove(taskTracker);
      if (set != null) {
        for (TaskAttemptID id : set) {
          killList.add(new KillTaskAction(id));
        }
      }
    }
    return killList;
  }
708808
void removeMarkedTasks(String taskTracker) {
    // Purge all the 'marked' tasks which were running at taskTracker
    Set<TaskAttemptID> markedTaskSet =
      trackerToMarkedTasksMap.get(taskTracker);
    if (markedTaskSet != null) {
      for (TaskAttemptID taskid : markedTaskSet) {
        removeTaskEntry(taskid);
        if (LOG.isDebugEnabled()) {
          LOG.debug("Removed marked completed task '" + taskid + "' from '" +
                    taskTracker + "'");
        }
      }
      // Clear
      trackerToMarkedTasksMap.remove(taskTracker);
    }
  }
564812
public void setChannel(@Nonnull InputStream in, @Nonnull OutputStream out,
                           @Nonnull TaskListener taskListener,
                           @CheckForNull Channel.Listener listener) throws IOException, InterruptedException {
        setChannel(in,out,taskListener.getLogger(),listener);
    }
1272495
public function onSwooleTask($serv, $task_id, $from_id, $data)
    {
        $type = $data['type'] ?? '';
        $message = $data['message'] ?? '';
        switch ($type) {
            case SwooleMarco::MSG_TYPE_SEND_BATCH://发送消息
                foreach ($message['fd'] as $fd) {
                    $this->send($fd, $message['data'], true);
                }

                return null;
            case SwooleMarco::MSG_TYPE_SEND_ALL://发送广播
                foreach ($this->uid_fd_table as $row) {
                    $this->send($row['fd'], $message['data'], true);
                }

                return null;
            case SwooleMarco::MSG_TYPE_SEND_ALL_FD://发送广播
                foreach ($serv->connections as $fd) {
                    $this->send($fd, $message['data'], true);
                }

                return null;
            case SwooleMarco::SERVER_TYPE_TASK://task任务
                $task_name = $message['task_name'];
                $task = $this->loader->task($task_name, $this);
                $task_fuc_name = $message['task_fuc_name'];
                $task_data = $message['task_fuc_data'];
                $task_context = $message['task_context'];
                $call = [$task->getProxy(), $task_fuc_name];
                if (is_callable($call)) {
                    //给task做初始化操作
                    $task->initialization($task_id, $from_id, $this->server->worker_pid, $task_name, $task_fuc_name, $task_context);
                    $result = sd_call_user_func_array($call, $task_data);
                } else {
                    throw new SwooleException("method $task_fuc_name not exist in $task_name");
                }
                $task->destroy();

                return $result;
            default:
                return parent::onSwooleTask($serv, $task_id, $from_id, $data);
        }
    }
271000
async def addSignalHandlers(self):
        '''
        
        '''

        def sigterm():
            print('Caught SIGTERM, shutting down.')
            asyncio.create_task(self.fini())

        def sigint():
            print('Caught SIGINT, shutting down.')
            asyncio.create_task(self.fini())

        loop = asyncio.get_running_loop()
        loop.add_signal_handler(signal.SIGINT, sigint)
        loop.add_signal_handler(signal.SIGTERM, sigterm)