called after steals are added to the pending queue
493426
function(){
			// if we don't have a current 'top' steal
			// we create one and set it up
			// to start loading its dependencies (the current pending steals)
			if(! currentCollection ){
				currentCollection = new steal.p.init();
				
				// keep a reference in case it disappears 
				var cur = currentCollection,
					// runs when a steal is starting
					go = function(){
						// indicates that a collection of steals has started
						steal.trigger("start", cur);
						when(cur,"complete", function(){
							steal.trigger("end", cur);
						});
						cur.loaded();
					};
				// if we are in rhino, start loading dependencies right away
				if(!win.setTimeout){
					go()
				}else{
					// otherwise wait a small timeout to make 
					// sure we get all steals in the current file
					setTimeout(go,0)
				}
			}
		}
909558
final boolean acquireQueued(final Node node, int arg) {
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } catch (Throwable t) {
            cancelAcquire(node);
            throw t;
        }
    }
1199374
synchronized void garbageCollect() {
    // Cancel task tracker reservation
    cancelReservedSlots();

    // Remove the remaining speculative tasks counts
    totalSpeculativeReduceTasks.addAndGet(-speculativeReduceTasks);
    totalSpeculativeMapTasks.addAndGet(-speculativeMapTasks);
    garbageCollected = true;

    // Let the JobTracker know that a job is complete
    jobtracker.getInstrumentation().decWaitingMaps(getJobID(), pendingMaps());
    jobtracker.getInstrumentation().decWaitingReduces(getJobID(), pendingReduces());
    jobtracker.storeCompletedJob(this);
    jobtracker.finalizeJob(this);

    try {
      // Definitely remove the local-disk copy of the job file
      if (localJobFile != null) {
        localFs.delete(localJobFile, true);
        localJobFile = null;
      }

      // clean up splits
      for (int i = 0; i < maps.length; i++) {
        maps[i].clearSplit();
      }

      // JobClient always creates a new directory with job files
      // so we remove that directory to cleanup
      // Delete temp dfs dirs created if any, like in case of
      // speculative exn of reduces.
      Path tempDir = jobtracker.getSystemDirectoryForJob(getJobID());
      new CleanupQueue().addToQueue(new PathDeletionContext(
          FileSystem.get(conf), tempDir.toUri().getPath()));
    } catch (IOException e) {
      LOG.warn("Error cleaning up "+profile.getJobID()+": "+e);
    }

    cleanUpMetrics();
    // free up the memory used by the data structures
    this.nonRunningMapCache = null;
    this.runningMapCache = null;
    this.nonRunningReduces = null;
    this.runningReduces = null;
    this.trackerMapStats = null;
    this.trackerReduceStats = null;
  }
1122358
def create_pending_after_task(self):
        
        for task in self.settings.tasks[self.after_tasks_key]:
            self.after_tasks.append(task)
            self.after_results[task] = Result(task)
743507
function _drainQueueStep(queue) {
    var fn = queue.shift();
    if (typeof fn !== "function") {
        fn._settlePromises();
    } else {
        var receiver = queue.shift();
        var arg = queue.shift();
        fn.call(receiver, arg);
    }
}
876638
function runQueue() {
    // Flag whether locker is busy (i.e. has current or pending processes)
    this.busy = (this.running || this.queue.length);

    // Run all items in queue until item requiring exclusive lock
    while (true) {
        var again = runNext.call(this);
		if (!again) return;
    }
}
433009
def _flush_queue(self, q, ignore_priority=False):
        
        assert isinstance(q, PriorityQueue)

        current_timestamp = compute_release_time(lag_in_minutes=0)
        for _ in range(len(q)):
            entry = q.pop()
            assert isinstance(entry, PriorityEntry)

            if ignore_priority or entry.release_time < current_timestamp:
                self._resubmit_uow(entry.entry)
            else:
                q.put(entry)
                break
957915
async def purgeQueues(self, *args, **kwargs):
        

        return await self._makeApiCall(self.funcinfo["purgeQueues"], *args, **kwargs)
1084686
public void restoreAfterSafeModeRestart() {
    for (Integer pendingRequestId : pendingRequestsList) {
      ResourceRequestInfo request = idToRequest.get(pendingRequestId);
      incrementRequestCount(request.getType(), 1);
      addPendingRequest(request);
    }

    for (Integer grantedRequestId : idToGrant.keySet()) {
      ResourceRequestInfo request = idToRequest.get(grantedRequestId);
      incrementRequestCount(request.getType(), 1);
      addGrantedRequest(request);
    }
  }
125027
function flushEvents() {
  var tmp = PENDING_QUEUE;
  PENDING_QUEUE = [];
  for (var i = 0; i < tmp.length; i++) {
    clientObj.send('log', tmp[i].item, tmp[i].fn);
  }
}
227906
def flush(self, queue_name):
        
        for name in (queue_name, dq_name(queue_name)):
            self.do_purge(name)
948072
function afterProcessReq(request, response) {
    defaults.afterExecuteOfflineItem(response, request.payload);
    return processReqs(utils.storage.get('queue'));
  }
74581
public synchronized void doneTask(int stealerId, int donorId) {
        removeNodesFromWorkerList(Arrays.asList(stealerId, donorId));
        numTasksExecuting--;
        doneSignal.countDown();
        // Try and schedule more tasks now that resources may be available to do
        // so.
        scheduleMoreTasks();
    }
1161814
def queue_pull(object_class, remote_id)
      queue_item = QueueItem.new( :kind => "pull",
                                  :object_type => object_class.name,
                                  :object_remote_id => remote_id,
                                  :state => "new" )
      queue_item.save
    end
99459
def on_finish(self, exc=None):
        """"""
        super(GarbageCollector, self).on_finish(exc)
        self._cycles_left -= 1
        if self._cycles_left <= 0:
            num_collected = gc.collect()
            self._cycles_left = self.collection_cycle
            LOGGER.debug('garbage collection run, %d objects evicted',
                         num_collected)
254370
def put(self, message):
    
    with self._outgoing_lock:
      self._outgoing.append(message)
      self._outgoing_counter += 1

      # Check to see if there are pending queues waiting for the item.
      if self._outgoing_counter in self._outgoing_pending_queues:
        for q in self._outgoing_pending_queues[self._outgoing_counter]:
          q.put(message)
        del self._outgoing_pending_queues[self._outgoing_counter]
1278567
void waitForResuming() throws IOException
   {
      if (isSuspended.get())
      {
         try
         {
            latcher.get().await();
         }
         catch (InterruptedException e)
         {
            throw new IOException(e);
         }
      }
   }