Drains the event queue via the scheduler
161716
function drainQueue(jobInfo) {
        function drainNext() {
            return drainQueue;
        }

        var queue = jobInfo.job._queue;

        if (queue.length === 0 && eventQueue.length > 0) {
            queue = jobInfo.job._queue = copyAndClearQueue();
        }

        jobInfo.setPromise(drainOneEvent(queue).then(drainNext, drainNext));
    }
1306282
protected void drain() {
        Message msg;
        while((msg=queue.poll()) != null)
            addAndSendIfSizeExceeded(msg);
        _sendBundledMessages();
    }
283335
function flushQueue()
    {
        var queued = WebInspector.log.queued;
        if (!queued)
            return;

        for (var i = 0; i < queued.length; ++i)
            logMessage(queued[i]);

        delete WebInspector.log.queued;
    }
703879
private void fireOnPurgeQueue(ChannelEvent e) {
        List<EventListener> listeners = changes.getListenerList(AMQP);
        for (EventListener listener : listeners) {
            ChannelListener amqpListener = (ChannelListener)listener;
            amqpListener.onPurgeQueue(e);
        }
    }
413378
private void ensureCleanStartup() {
        forceShutdown();
        ScheduledThreadPoolExecutor oldPool = pool.getAndSet(createScheduledThreadExecutor());
        oldPool.shutdownNow();
        claimedForHandoff.clear();
        workUnitsPeggedToMe.clear();
        state.set(NodeState.Fresh);
    }
192151
def _socket_close(self):
        """"""
        callback = self.__callback
        self.__callback = None
        try:
            if callback:
                callback(None, InterfaceError('connection closed'))
        finally:
            # Flush the job queue, don't call the callbacks associated with the remaining jobs
            # since they have already been called as error callback on connection closing
            self.__job_queue = []
            self.__alive = False
            self.__pool.cache(self)
828662
def unschedule(event)
            @critical.synchronize {
                # Only call delete and update the timer when required
                if @schedules.include?(event)
                    @schedules.delete(event)
                    remove(event)
                    check_timer
                end
            }
        end
578499
protected void drainToLoad(long targetLoad, int time, boolean useHandoff)
    {
        final double startingLoad = myLoad();
        double currentLoad = startingLoad;
        List<String> drainList = new LinkedList<String>();
        Set<String> eligibleToDrop = new LinkedHashSet<String>(cluster.myWorkUnits);
        eligibleToDrop.removeAll(cluster.workUnitsPeggedToMe);

        for (String workUnit : eligibleToDrop) {
            if (currentLoad <= targetLoad) {
                break;
            }
            double workUnitLoad = cluster.getWorkUnitLoad(workUnit);

            if (workUnitLoad > 0 && (currentLoad - workUnitLoad) > targetLoad) {
                drainList.add(workUnit);
                currentLoad -= workUnitLoad;
            }
        }

        int drainInterval = (int) (((double) config.drainTime / drainList.size()) * 1000);
        TimerTask drainTask = buildDrainTask(drainList, drainInterval, useHandoff, currentLoad);

        if (!drainList.isEmpty()) {
            LOG.info("Releasing work units over {} seconds. Current load: {}. Target: {}. Releasing: {}",
                    time, startingLoad, targetLoad, Strings.mkstring(drainList, ", "));
            cluster.schedule(drainTask, 0, TimeUnit.SECONDS);
        }
    }
96726
private void startPurgeTimer() {
        if (null != this.future || null == this.scheduler) {
            return;
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "Creating repeating purge event");
        }
        this.future = this.scheduler.schedule(SessionEventHandler.PURGE_EVENT, this.purgeInterval, this.purgeInterval, TimeUnit.SECONDS);
    }
371900
void enqueue(Runnable runnable, boolean flush) {
    queue.add(new RunnableCommand(runnable));
    if (flush) {
      scheduleFlush();
    }
  }
785047
def clear(self):
    """"""
    while self.current or self.idlers or self.queue or self.rpcs:
      current = self.current
      idlers = self.idlers
      queue = self.queue
      rpcs = self.rpcs
      _logging_debug('Clearing stale EventLoop instance...')
      if current:
        _logging_debug('  current = %s', current)
      if idlers:
        _logging_debug('  idlers = %s', idlers)
      if queue:
        _logging_debug('  queue = %s', queue)
      if rpcs:
        _logging_debug('  rpcs = %s', rpcs)
      self.__init__()
      current.clear()
      idlers.clear()
      queue[:] = []
      rpcs.clear()
      _logging_debug('Cleared')
977212
def purge(self):
        
        self.sender.send_QueuePurge(self.name)
        yield from self.synchroniser.wait(spec.QueuePurgeOK)
        self.reader.ready()
232859
private void checkDrain() {
    if (full && !paused && messages.size() < maxQueueSize / 2) {
      full = false;
      log.debug(String.format("%s - Connection to %s is drained", this, context.target()));
      if (drainHandler != null) {
        drainHandler.handle((Void) null);
      }
    }
  }
256347
def stop(self):
        
        with self._lock:
            if self._stopped:
                _logger.debug(u'%s is already stopped', self)
                return

            self._flush_all_reports()
            self._stopped = True
            if self._run_scheduler_directly:
                self._cleanup_if_stopped()

            if self._scheduler and self._scheduler.empty():
                # if there are events scheduled, then _running will subsequently
                # be set False by the scheduler thread.  This handles the
                # case where there are no events, e.g because all aggreagation
                # was disabled
                self._running = False
            self._scheduler = None
173688
public void flushPendingSessionEvents() throws Exception {
        synchronized (sessionEventsQueue) {
            IoSessionEvent evt;
            
            while ((evt = sessionEventsQueue.poll()) != null) {
                logger.debug(" Flushing buffered event: {}", evt);    
                evt.deliverEvent();
            }
        }
    }
183279
private void drainPendingCalls() {
    assert realStream != null;
    assert !passThrough;
    List<Runnable> toRun = new ArrayList<>();
    DelayedStreamListener delayedListener = null;
    while (true) {
      synchronized (this) {
        if (pendingCalls.isEmpty()) {
          pendingCalls = null;
          passThrough = true;
          delayedListener = this.delayedListener;
          break;
        }
        // Since there were pendingCalls, we need to process them. To maintain ordering we can't set
        // passThrough=true until we run all pendingCalls, but new Runnables may be added after we
        // drop the lock. So we will have to re-check pendingCalls.
        List<Runnable> tmp = toRun;
        toRun = pendingCalls;
        pendingCalls = tmp;
      }
      for (Runnable runnable : toRun) {
        // Must not call transport while lock is held to prevent deadlocks.
        // TODO(ejona): exception handling
        runnable.run();
      }
      toRun.clear();
    }
    if (delayedListener != null) {
      delayedListener.drainPendingCallbacks();
    }
  }
678615
void drain() {
        lock.lock();
        try {
            LOG.finest("Waiting while draining...");
            while (isWorking()
                    && (emptyRequestFuture == null
                    || emptyRequestFuture.isDone())) {
                try {
                    drained.await();
                } catch (InterruptedException intx) {
                    LOG.log(Level.FINEST, INTERRUPTED, intx);
                }
            }
            LOG.finest("Drained");
        } finally {
            lock.unlock();
        }
    }