if finished, then move this batchTask to his channel and then will be processed as regular task
5135
void processBatchWaitTasks() {
		Channel channel = context.registry.getChannelOrSystem(Model.CHANNEL_BATCH); // channel TedBW or TedSS
		int maxTask = context.taskManager.calcChannelBufferFree(channel);
		Map<String, Integer> channelSizes = new HashMap<String, Integer>();
		channelSizes.put(Model.CHANNEL_BATCH, maxTask);
		List<TaskRec> batches = context.tedDao.reserveTaskPortion(channelSizes);
		if (batches.isEmpty())
			return;

		for (final TaskRec batchTask : batches) {
			channel.workers.execute(new TedRunnable(batchTask) {
				@Override
				public void run() {
					processBatchWaitTask(batchTask);
				}
			});
		}
	}
643373
public function execute()
    {
        $batch  = $this->getBatch();
        $import = $batch->getVariable('import');

        if (! (isset($import['trackId']) && $import['trackId'])) {
            // Do nothing
            return;
        }

        $tracker     = $this->loader->getTracker();
        $trackEngine = $tracker->getTrackEngine($import['trackId']);
        $trackEngine->updateRoundCount($this->currentUser->getUserLoginId());
        
        // Now cleanup the cache
        $this->cache->clean(\Zend_Cache::CLEANING_MODE_MATCHING_TAG, array('tracks'));
    }
675753
private void transitionToFinalBatchStatus() {
		BatchStatus currentBatchStatus = stepContext.getBatchStatus();
		if (currentBatchStatus.equals(BatchStatus.STARTED)) {
			updateBatchStatus(BatchStatus.COMPLETED);
		} else if (currentBatchStatus.equals(BatchStatus.STOPPING)) {
			updateBatchStatus(BatchStatus.STOPPED);
		} else if (currentBatchStatus.equals(BatchStatus.FAILED)) {
			updateBatchStatus(BatchStatus.FAILED);           // Should have already been done but maybe better for possible code refactoring to have it here.
		} else {
			throw new IllegalStateException("Step batch status should not be in a " + currentBatchStatus.name() + " state");
		}
	}
1144022
void doBatchStart(String batch, Object args) {
    checkOpen();
    JsonObject message = createMessage(args)
        .putString("batch", batch)
        .putString("action", "startBatch");
    if (open && !paused) {
      if (log.isDebugEnabled()) {
        log.debug(String.format("%s - Batch start: Batch[batch=%s]", this, batch));
      }
      eventBus.send(inAddress, message);
    }
    checkFull();
  }
421959
def run(self):
        '''
        '''
        if self.verbose:
            pbar = tqdm(total=100)
        while True:
            task_remain = self._task_queue.qsize()
            task_finished = int((float(self.total_task - task_remain) /
                                 float(self.total_task)) * 100)
            if task_finished % 20 == 0 and task_finished != self.current_state:
                self.current_state = task_finished
                logger.info('{0}% done'.format(task_finished))
                if self.verbose and task_finished > 0:
                    pbar.update(20)
            if task_remain == 0:
                break
        logger.debug('All task data cleared')
11828
def _run(self):
        ''''''
      
        while self._is_running:
            yield from self._task()

        # Signal that the loop has finished.
        self._run_complete.set_result(True)
254793
private void awaitRequestsCompletion() throws IOException {
    // Don't wait until all requests will be completed if enough requests are pending for full batch
    while (!responseFutures.isEmpty() && pendingRequests.size() < maxRequestsPerBatch) {
      try {
        responseFutures.remove().get();
      } catch (InterruptedException | ExecutionException e) {
        if (e.getCause() instanceof IOException) {
          throw (IOException) e.getCause();
        }
        throw new RuntimeException("Failed to execute batch", e);
      }
    }
  }
1162378
def batch!(operations, request_options = {})
      res = batch(operations, request_options)
      res['taskID'].each do |index, taskID|
        wait_task(index, taskID, WAIT_TASK_DEFAULT_TIME_BEFORE_RETRY, request_options)
      end
    end
758969
private void transitionToFinalBatchStatus() {
		BatchStatus currentBatchStatus = jobContext.getBatchStatus();
		if (currentBatchStatus.equals(BatchStatus.STARTED)) {
			updateJobBatchStatus(BatchStatus.COMPLETED);
		} else if (currentBatchStatus.equals(BatchStatus.STOPPING)) {
			updateJobBatchStatus(BatchStatus.STOPPED);
		} else if (currentBatchStatus.equals(BatchStatus.FAILED)) {
			updateJobBatchStatus(BatchStatus.FAILED);  // Should have already been done but maybe better for possible code refactoring to have it here.
		} else {
			throw new IllegalStateException("Step batch status should not be in a " + currentBatchStatus.name() + " state");
		}
	}
1289809
def _complete_task(self, task_name, **kwargs):
        

        logger.debug('Job {0} marking task {1} as completed'.format(self.name, task_name))
        self.run_log['tasks'][task_name] = kwargs

        for node in self.downstream(task_name, self.snapshot):
            self._start_if_ready(node)

        try:
            self.backend.acquire_lock()
            self._commit_run_log()
        except:
            logger.exception("Error in handling events.")
        finally:
            self.backend.release_lock()

        if kwargs.get('success', None) == False:
            task = self.tasks[task_name]
            try:
                self.backend.acquire_lock()
                if self.event_handler:
                    self.event_handler.emit('task_failed',
                                            task._serialize(include_run_logs=True))
            except:
                logger.exception("Error in handling events.")
            finally:
                self.backend.release_lock()

        self._on_completion()
1106193
protected synchronized void postTaskExecution() {
		if (stopped) {
			// Stopped, so don't count the task(s) as done.
			// (worker threads call this method even if the task was not really executed.)
			return;
		}
		tasksDoneCount++;
		int percentageComplete = tasksDoneCount * 100 / tasksTotalCount;

		// Compute the progress and notify the listeners
		this.notifyListenersSpiderProgress(percentageComplete, tasksDoneCount, tasksTotalCount - tasksDoneCount);

		// Check for ending conditions
		if (tasksDoneCount == tasksTotalCount && initialized) {
			this.complete();
		}
	}
1035213
function() {
        if(!this._isStopped) {
            return;
        }

        this._isStopped = false;
        var processedBuffer = this._processedBuffer;
        if(processedBuffer.length) {
            this._processedBuffer = [];
            nextTick(function() {
                while(processedBuffer.length) {
                    processedBuffer.shift()();
                }
            });
        }

        this._hasPendingTasks() && this._scheduleRun();
    }
277017
function() {
            if (_.size(this.tasks) > 0) {
                this.empty = false;
                var task = this.tasks.shift();
                this.startTask(task);
                this.trigger("tasks:next");
            } else {
                this.empty = true;
                this.trigger("tasks:finish");
            }      
        }
1066653
public void postTask( Runnable task )
  {
    synchronized( _queue )
    {
      _queue.add( task );

      if( !isAlive() )
      {
        start();
      }

      _queue.notifyAll();
    }
  }
1140132
def process_task(self):
        """"""
        pdu = RegisterForeignDevice(self.bbmdTimeToLive)
        pdu.pduDestination = self.bbmdAddress

        # send it downstream
        self.request(pdu)
858187
void doBatchEnd(String batch, Object args) {
    checkOpen();
    JsonObject message = createMessage(args)
        .putString("action", "endBatch")
        .putString("batch", batch);
    if (open && !paused) {
      if (log.isDebugEnabled()) {
        log.debug(String.format("%s - Batch end: Batch[batch=%s, args=%s]", this, batch, args));
      }
      eventBus.send(inAddress, message);
    }
    if (currentBatch != null && currentBatch.id().equals(batch)) {
      currentBatch = null;
    }
  }
1367758
def finish(self):
        """"""
        if self._finished:
            raise RuntimeError("finish() called twice.  May be caused "
                               "by using async operations without the "
                               "@asynchronous decorator.")
        if not hasattr(self, '_stream_queue') or not self._stream_queue:
            raise RuntimeError("`_stream_queue` was not assigned, you should"
                               "call `write_stream_queue` to set.")

        # === Replace `if not self._headers_written` === #
        self.set_status(200)
        self.set_header("Content-Type", "text/event-stream")
        self.set_header("Cache-Control", "no-cache")
        self.set_header("Access-Control-Allow-Origin", "*")
        # ============================================== #

        self.request.connection.set_close_callback(None)

        # === Add before self.flush === #
        # Reset buffer
        self._write_buffer = []
        self._headers_written = False
        # ============================= #

        self.flush(include_footers=True)

        # === Add after self.flush === #
        self._write_buffer = self._stream_queue
        self.request.connection._write_buffer = self._stream_queue
        # ============================ #

        self.request.finish()
        self._log()
        self._finished = True
        self.on_finish()
        # Break up a reference cycle between this handler and the
        # _ui_module closures to allow for faster GC on CPython.
        self.ui = None