When starting a new job, or moving to a new stage, split up the inputs
 into WorkUnits, and queue them. Workers will start picking them up right
 away.
974186
def queue_for_workers(input=nil)
      input ||= JSON.parse(self.inputs)
      input.each {|i| WorkUnit.start(self, action, i, status) }
      self
    end
1259274
def start_workunit(self, workunit):
    """"""
    if not self.is_under_main_root(workunit):
      return

    label_format = self._get_label_format(workunit)

    if label_format == LabelFormat.FULL:
      if not WorkUnitLabel.SUPPRESS_LABEL in workunit.labels:
        self._emit_indented_workunit_label(workunit)
      # Start output on a new line.
      tool_output_format = self._get_tool_output_format(workunit)
      if tool_output_format == ToolOutputFormat.INDENT:
        self.emit(self._prefix(workunit, '\n'))
      elif tool_output_format == ToolOutputFormat.UNINDENTED:
        self.emit('\n')
    elif label_format == LabelFormat.DOT:
      self.emit('.')

    self.flush()
71246
function _go() {
        if(workers < concurrency) {
            var job = queue.shift();
            if(job !== undefined) {
                var worker = job.shift();
                job.push(over);
                workers++;
                process.nextTick(function nextTickWorker() {
                    worker.apply(null, job);
                });
                return true;
            }
        }
        return false;
    }
220487
final void runWorker(WorkQueue w) {
        w.growArray(false);         // initialize queue array in this thread
        do { w.runTask(scan(w)); } while (w.runState >= 0);
    }
24018
public void initTasks() throws IOException {
    // log job info
    jobHistory.logSubmitted(jobFile.toString(), this.startTime, this.jobTrackerId);
    // log the job priority
    JobClient.RawSplit[] splits = null;
    splits = JobClient.getAndRemoveCachedSplits(jobId);
    if (splits == null) {
      FileSystem fs = jobFile.getFileSystem(jobConf);
      Path splitFile = new Path(jobFile.getParent(), "job.split");
      LOG.info("Reading splits from " + splitFile);
      DataInputStream splitFileIn = fs.open(splitFile);
      try {
        splits = JobClient.readSplitFile(splitFileIn);
      } finally {
        splitFileIn.close();
      }
    }
    initTasksFromSplits(splits);
    jobHistory.logInited(this.launchTime,
                         numMapTasks, numReduceTasks);
  }
36511
def _setup_workers(self, initial_states=None):
        
        if initial_states is not None:
            require_listlike(initial_states)

        self.tasks_queues = []
        self.results_queue = mp.Queue()
        self.workers = []

        common_kwargs = dict(tokenizer=self.tokenizer,
                             stemmer=self.stemmer,
                             lemmata_dict=self.lemmata_dict,
                             pos_tagger=self.pos_tagger)

        if initial_states is not None:
            logger.info('setting up %d worker processes with initial states' % len(initial_states))

            for i_worker, w_state in enumerate(initial_states):
                task_q = mp.JoinableQueue()
                w = _PreprocWorker(i_worker, w_state.pop('docs'), self.language, task_q, self.results_queue,
                                   name='_PreprocWorker#%d' % i_worker, **common_kwargs)
                w.start()

                task_q.put(('set_state', w_state))

                self.workers.append(w)
                self.tasks_queues.append(task_q)

            [q.join() for q in self.tasks_queues]

        else:
            # distribute work evenly across the worker processes
            # we assume that the longer a document is, the longer the processing time for it is
            # hence we distribute the work evenly by document length
            logger.info('distributing work via greedy partitioning')

            docs_and_lengths = {dl: len(dt) for dl, dt in self.docs.items()}
            docs_per_worker = greedy_partitioning(docs_and_lengths, k=self.n_max_workers)

            logger.info('setting up %d worker processes' % len(docs_per_worker))

            for i_worker, doc_labels in enumerate(docs_per_worker):
                if not doc_labels: continue
                task_q = mp.JoinableQueue()
                w_docs = {dl: self.docs.get(dl) for dl in doc_labels}

                w = _PreprocWorker(i_worker, w_docs, self.language, task_q, self.results_queue,
                                   name='_PreprocWorker#%d' % i_worker, **common_kwargs)
                w.start()

                self.workers.append(w)
                self.tasks_queues.append(task_q)

        self.n_workers = len(self.workers)
198620
def setup(job, input_file_id, n, down_checkpoints):
    
    # Write the input file to the file store
    job.fileStore.logToMaster("Starting the merge sort")
    return job.addChildJobFn(down,
                             input_file_id, n,
                             down_checkpoints=down_checkpoints,
                             memory='600M').rv()
97311
def run_job(self):
        """"""
        try:
            self.load_parsers()
            self.load_filters()
            self.load_outputs()
            self.config_args()
            if self.args.list_parsers:
                self.list_parsers()
            if self.args.verbosemode: print('Loading input files')
            self.load_inputs()
            if self.args.verbosemode: print('Running parsers')
            self.run_parse()
            if self.args.verbosemode: print('Merging data')
            self.data_set['finalized_data'] = \
                    logdissect.utils.merge_logs(
                            self.data_set['data_set'], sort=True)
            if self.args.verbosemode: print('Running filters')
            self.run_filters()
            if self.args.verbosemode: print('Running output')
            self.run_output()
        except KeyboardInterrupt:
            sys.exit(1)
994987
public synchronized void initTasks()
  throws IOException, KillInterruptedException {
    if (tasksInited.get() || isComplete()) {
      return;
    }
    synchronized(jobInitKillStatus){
      if(jobInitKillStatus.killed || jobInitKillStatus.initStarted) {
        return;
      }
      jobInitKillStatus.initStarted = true;
    }

    LOG.info("Initializing " + jobId);

    // log job info
    JobHistory.JobInfo.logSubmitted(getJobID(), conf, jobFile.toString(),
                                    this.startTime, hasRestarted());
    // log the job priority
    setPriority(this.priority);

    //
    // read input splits and create a map per a split
    //
    String jobFile = profile.getJobFile();

    Path sysDir = new Path(this.jobtracker.getSystemDir());
    FileSystem fs = sysDir.getFileSystem(conf);
    DataInputStream splitFile =
      fs.open(new Path(conf.get("mapred.job.split.file")));
    JobClient.RawSplit[] splits;
    try {
      splits = JobClient.readSplitFile(splitFile);
    } finally {
      splitFile.close();
    }
    numMapTasks = splits.length;


    // if the number of splits is larger than a configured value
    // then fail the job.
    int maxTasks = jobtracker.getMaxTasksPerJob();
    if (maxTasks > 0 && numMapTasks + numReduceTasks > maxTasks) {
      throw new IOException(
                "The number of tasks for this job " +
                (numMapTasks + numReduceTasks) +
                " exceeds the configured limit " + maxTasks);
    }
    jobtracker.getInstrumentation().addWaitingMaps(getJobID(), numMapTasks);
    jobtracker.getInstrumentation().addWaitingReduces(getJobID(), numReduceTasks);

    maps = new TaskInProgress[numMapTasks];
    for(int i=0; i < numMapTasks; ++i) {
      inputLength += splits[i].getDataLength();
      maps[i] = new TaskInProgress(jobId, jobFile,
                                   splits[i],
                                   conf, this, i, numSlotsPerMap);
    }
    LOG.info("Input size for job " + jobId + " = " + inputLength
        + ". Number of splits = " + splits.length);
    if (numMapTasks > 0) {
      nonRunningMapCache = createCache(splits, maxLevel);
    }

    // set the launch time
    this.launchTime = JobTracker.getClock().getTime();
    jobtracker.getInstrumentation().addLaunchedJobs(
      this.launchTime - this.startTime);

    //
    // Create reduce tasks
    //
    this.reduces = new TaskInProgress[numReduceTasks];
    for (int i = 0; i < numReduceTasks; i++) {
      reduces[i] = new TaskInProgress(jobId, jobFile,
                                      numMapTasks, i,
                                      conf, this, numSlotsPerReduce);
      nonRunningReduces.add(reduces[i]);
    }

    // Calculate the minimum number of maps to be complete before
    // we should start scheduling reduces
    completedMapsForReduceSlowstart =
      (int)Math.ceil(
          (conf.getFloat("mapred.reduce.slowstart.completed.maps",
                         DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART) *
           numMapTasks));
    // The thresholds of total maps and reduces for scheduling reducers
    // immediately.
    rushReduceMaps =
      conf.getInt(RUSH_REDUCER_MAP_THRESHOLD, rushReduceMaps);
    rushReduceReduces =
      conf.getInt(RUSH_REDUCER_REDUCE_THRESHOLD, rushReduceReduces);
    maxFetchFailuresPerMapper = conf.getInt(MAX_FETCH_FAILURES_PER_MAP_KEY,
        MAX_FETCH_FAILURES_PER_MAP_DEFAULT);

    initSetupCleanupTasks(jobFile);

    synchronized(jobInitKillStatus){
      jobInitKillStatus.initDone = true;
      if(jobInitKillStatus.killed) {
        throw new KillInterruptedException("Job " + jobId + " killed in init");
      }
    }

    tasksInited.set(true);
    JobHistory.JobInfo.logInited(profile.getJobID(), this.launchTime,
        numMapTasks, numReduceTasks);

    // Log the number of map and reduce tasks
    LOG.info("Job " + jobId + " initialized successfully with " + numMapTasks
        + " map tasks and " + numReduceTasks + " reduce tasks.");
    refreshIfNecessary();
  }
871255
public function cleanUp()
    {
        // create a separate queue for each priority
        foreach (PriorityKeys::getAll() as $priorityKey) {
            $this->workers[$this->uniqueWorkerName($priorityKey)]->stop();
        }
    }
67527
def queue(self, new_job_id = None, new_job_name = None, queue_name = None):
    """"""
    # update the job id (i.e., when the job is executed in the grid)
    if new_job_id is not None:
      self.id = new_job_id

    if new_job_name is not None:
      self.name = new_job_name

    if queue_name is not None:
      self.queue_name = queue_name

    new_status = 'queued'
    self.result = None
    # check if we have to wait for another job to finish
    for job in self.get_jobs_we_wait_for():
      if job.status not in ('success', 'failure'):
        new_status = 'waiting'
      elif self.stop_on_failure and job.status == 'failure':
        new_status = 'failure'

    # reset the queued jobs that depend on us to waiting status
    for job in self.get_jobs_waiting_for_us():
      if job.status == 'queued':
        job.status = 'failure' if new_status == 'failure' else 'waiting'

    self.status = new_status
    for array_job in self.array:
      if array_job.status not in ('success', 'failure'):
        array_job.status = new_status
185876
def enqueue_jobs(self, jobs: Iterable[Job]):
        """"""
        for job in jobs:
            if job.should_start:
                job.status = JobStatus.QUEUED
                queue = self._get_queue(job.queue)
                queue.put(job.serialize())
            else:
                with self._lock:
                    job.status = JobStatus.WAITING
                    self._future_jobs.append(job.serialize())
                    self._future_jobs.sort(key=lambda j: Job.deserialize(j).at)
        self._something_happened.set()
524373
public static Job startOneJob(Worker newWorker, 
      Priority pri, Set<String> jobFiles, long detectTime, 
      AtomicLong numFilesSubmitted, AtomicLong lastCheckingTime,
      long maxPendingJobs) 
          throws IOException, InterruptedException, ClassNotFoundException {
    if (lastCheckingTime != null) {
      lastCheckingTime.set(System.currentTimeMillis());
    }
    String startTimeStr = dateFormat.format(new Date());
    String jobName = newWorker.JOB_NAME_PREFIX + "." + newWorker.jobCounter + 
        "." + pri + "-pri" + "." + startTimeStr;
    Job job = null;
    synchronized(jobFiles) {
      if (jobFiles.size() == 0) {
        return null;
      }
      newWorker.jobCounter++;
      
      synchronized(newWorker.jobIndex) {
        if (newWorker.jobIndex.size() >= maxPendingJobs) {
          // full 
          return null;
        }
        job = newWorker.startJob(jobName, jobFiles, pri, detectTime);
      }
      numFilesSubmitted.addAndGet(jobFiles.size());
      jobFiles.clear();
      
    }
    return job;
  }
1154079
private void buildSubJobBatchWorkUnits() {

		List<Flow> flows = this.split.getFlows();

		parallelBatchWorkUnits = new ArrayList<BatchFlowInSplitWorkUnit>();

		// Build all sub jobs from flows in split
		synchronized (subJobs) {
			for (Flow flow : flows) {
				subJobs.add(PartitionedStepBuilder.buildFlowInSplitSubJob(jobContext, this.split, flow));
			}
			// Go back to earlier idea that we may have seen this id before, and need a special "always restart" behavior
			// for split-flows.
			for (JSLJob job : subJobs) {				
				int count = batchKernel.getJobInstanceCount(job.getId());
				FlowInSplitBuilderConfig config = new FlowInSplitBuilderConfig(job, completedWorkQueue, rootJobExecutionId);
				if (count == 0) {
					parallelBatchWorkUnits.add(batchKernel.buildNewFlowInSplitWorkUnit(config));
				} else if (count == 1) {
					parallelBatchWorkUnits.add(batchKernel.buildOnRestartFlowInSplitWorkUnit(config));
				} else {
					throw new IllegalStateException("There is an inconsistency somewhere in the internal subjob creation");
				}
			}
		}
	}
62864
private void makeJobRunning(JobInProgress job, JobSchedulingInfo oldInfo, 
                              QueueInfo qi) {
    // Removing of the job from job list is responsibility of the
    //initialization poller.
    // Add the job to the running queue
    qi.addRunningJob(job);
  }
639425
protected function startWorkers(): void
    {
        $wrapper = realpath(__DIR__ . '/../../../bin/phpunit-sqlite-wrapper');

        for ($i = 1; $i <= $this->options->processes; ++$i) {
            $worker = new SqliteWorker($this->dbFileName);
            if ($this->options->noTestTokens) {
                $token = null;
                $uniqueToken = null;
            } else {
                $token = $i;
                $uniqueToken = uniqid();
            }
            $worker->start($wrapper, $token, $uniqueToken);
            $this->workers[] = $worker;
        }
    }
906658
private void initJobs() {
        List<MasterSlaveNodeData> masterSlaveNodeDataList = masterSlaveApiFactory.nodeApi().getAllNodes();
        if (!ListHelper.isEmpty(masterSlaveNodeDataList)) {
            return;
        }
        List<MasterSlaveJobData> masterSlaveJobDataList = new ArrayList<>();
        try {
            masterSlaveJobDataList = masterSlaveApiFactory.jobApi().getAllJobs();
        } catch (Throwable e) {
            if (e instanceof NiubiException) {
                e = e.getCause();
            }
            if (e instanceof KeeperException.NoNodeException) {
                LoggerHelper.info("job path not found. skip init jobs.");
            } else {
                LoggerHelper.warn("get jobs failed. ", e);
            }
        }
        for (MasterSlaveJobData masterSlaveJobData : masterSlaveJobDataList) {
            MasterSlaveJobData.Data data = masterSlaveJobData.getData();
            data.init();
            masterSlaveApiFactory.jobApi().updateJob(data.getGroupName(), data.getJobName(), data);
        }
    }