(internal) Intended to be run in a separate thread and take results from
        the pool and put them into queues depending on the task of the result. 
        It finishes if it receives termination-sentinels from all pool workers.
498659
def _pool_get(get, results, next_available, task_next_lock, to_skip, \
                  task_num, pool_size, id_self):
        
        log.debug('NuMap(%s) started pool_getter' % id_self)
        # should return when all workers have returned, each worker sends a 
        # sentinel before returning. Before returning it should send sentinels to
        # all tasks but the next available queue should be released only if we 
        # know that no new results will arrive.
        sentinels = 0
        result_ids, last_result_id, very_last_result_id = {}, {}, {}
        for i in xrange(task_num):
            last_result_id[i] = -1
            very_last_result_id[i] = -2
            result_ids[i] = set()

        while True:
            try:
                log.debug('NuMap(%s) pool_getter waits for a result.' % id_self)
                #gc.disable()
                result = get()
                #gc.enable()
            except (IOError, EOFError):
                log.error('NuMap(%s) pool_getter has a pipe problem.' % id_self)
                break

            # got a sentinel?
            if result is None:
                sentinels += 1
                log.debug('NuMap(%s) pool_getter got a sentinel.' % id_self)
                if sentinels == pool_size:
                    log.debug('NuMap(%s) pool_getter got all sentinels.' % \
                              id_self)
                    # here we are escaping.
                    break
                else:
                    # waiting for more sentinels or results to come.
                    continue

            # got a sentinel for a task?
            # only one sentinel per task will be received
            if result[1] is None:
                task = result[0]
                very_last_result_id[task] = result[2]
                if last_result_id[task] == very_last_result_id[task]:
                    results[task].put(('stop', False, 'stop'))
                    next_available[task].put(True)
                    log.debug('NuMap(%s) pool_getter sent sentinel for task %s.'\
                                                            % (id_self, task))
                continue

            # got some result for some task, which might be an exception
            task, i, is_valid, real_result = result
            # locked if next for this task is in 
            # the process of raising a TimeoutError
            task_next_lock[task].acquire()
            log.debug('NuMap(%s) pool_getter received result %s for task %s)' % \
                      (id_self, i, task))

            if to_skip[task]:
                log.debug('NuMap(%s) pool_getter skips results: %s' % (id_self, \
                range(last_result_id[task] + 1, last_result_id[task] + \
                      to_skip[task] + 1)))
                last_result_id[task] += to_skip[task]
                to_skip[task] = 0

            if i > last_result_id[task]:
                result_ids[task].add(i)
                results[task].put((i, is_valid, real_result))
                log.debug('NuMap(%s) pool_getter put result %s for task %s to queue' % \
                          (id_self, i, task))
            else:
                log.debug('NuMap(%s) pool_getter skips result %s for task %s' % \
                          (id_self, i, task))

            # this releases the next method for each ordered result in the queue
            # if the NuMap instance is ordered =False this information is 
            # ommitted.
            while last_result_id[task] + 1 in result_ids[task]:
                next_available[task].put(True)
                last_result_id[task] += 1
                log.debug('NuMap(%s) pool_getter released task: %s' % \
                          (id_self, task))
            if last_result_id[task] == very_last_result_id[task]:
                results[task].put(('stop', False, 'stop'))
                next_available[task].put(True)
            # release the next method
            task_next_lock[task].release()
        log.debug('NuMap(%s) pool_getter returns' % id_self)
7438
public static boolean shutdownAndAwaitTermination(ExecutorService pool, long waitSeconds) {
		pool.shutdown(); // Disable new tasks from being submitted
		try {
			// Wait a while for existing tasks to terminate
			if(!pool.awaitTermination(waitSeconds, TimeUnit.SECONDS)) {
				//logger.warn("Thread pool is still running: {}", pool);
				pool.shutdownNow(); // Cancel currently executing tasks
				// Wait a while for tasks to respond to being cancelled
				if(!pool.awaitTermination(waitSeconds, TimeUnit.SECONDS)){
					//logger.warn("Thread pool is not terminating: {}", pool);
					return false;
				}else{
					return true;
				}
			}else{
				return true;
			}
		} catch(InterruptedException ie) {
			// (Re-)Cancel if current thread also interrupted
			pool.shutdownNow();
			// Preserve interrupt status
			Thread.currentThread().interrupt();
			return false;
		}
	}
1351016
def enqueue(self, future):
    

    future.enqueue()
    with self._lock:
      if self._shutdown:
        raise RuntimeError('ThreadPool has been shut down and can no '
          'longer accept futures.')
      self._queue.append(future)
      if len(self._running) == len(self._workers):
        self._new_worker()
      self._lock.notify_all()
188607
protected void startQueue() {
      if (executor == null) {
          logger.debug("Starting the receiving thread pool.");
          executor = new ThreadPoolExecutor(
                  numberOfWorkerThreads,
                  numberOfWorkerThreads,
                  threadKeepAliveTime, TimeUnit.SECONDS,
                  new LinkedBlockingQueue<Runnable>(),
                  getThreadFactory());
          executor.allowCoreThreadTimeOut(true);
          //Start with one thread, and build it up gradually as it needs.
          executor.prestartCoreThread();
          logger.info("ReceiveQueue started! Current pool size: {}", executor.getPoolSize());
      } else {
          if (executor.getCorePoolSize() < numberOfWorkerThreads) {
              //If the number has increased we need to set the max first, or we'll get an IllegalArgumentException
              executor.setMaximumPoolSize(numberOfWorkerThreads);
              executor.setCorePoolSize(numberOfWorkerThreads);
          } else if (executor.getCorePoolSize() > numberOfWorkerThreads) {
              //If the number has decreased we need to set the core first.
              executor.setCorePoolSize(numberOfWorkerThreads);
              executor.setMaximumPoolSize(numberOfWorkerThreads);
          }
          logger.info("ReceiveQueue running. Current pool size: {}. Current Queue size: {}",
                  executor.getPoolSize(), getQueueSize());
          logger.info("Nr of active pool-threads: {}", executor.getActiveCount());
      }
    }
482262
boolean manageIdlePool(ThreadPoolExecutor threadPool, long intervalCompleted) {

        // Manage the intervalCompleted count
        if (intervalCompleted == 0 && threadPool.getActiveCount() == 0) {
            consecutiveIdleCount++;
        } else {
            consecutiveIdleCount = 0;
        }

        if (consecutiveIdleCount >= IDLE_INTERVALS_BEFORE_PAUSE) {
            pause();
            lastAction = LastAction.PAUSE;
            return true;
        }

        return false;
    }
647039
private void shutdownAndAwaitTermination( ExecutorService pool, int queueSize )
    {
        log.info( "Shutting down executor service; queue size={}.", queueSize );

        pool.shutdown(); // Disable new tasks from being submitted

        log.info( "Waiting for executor service to finish all queued tasks." );

        try
        {
            // Wait a while for existing tasks to terminate
            if ( ! pool.awaitTermination( 60, TimeUnit.SECONDS ) )
            {
                log.info( "Executor service did not terminate before timeout; forcing shutdown." );

                List<Runnable> queuedTasks = pool.shutdownNow(); // Cancel currently executing tasks

                log.info( "Discarded {} tasks that were still queued; still waiting for executor service to terminate."
                        , queuedTasks.size() );

                // Wait a while for tasks to respond to being cancelled
                if ( ! pool.awaitTermination( 60, TimeUnit.SECONDS ) )
                {
                    log.error( "Pool did not terminate." );
                }
                else
                {
                    log.info( "Executor service terminated after forced shutdown." );
                }
            }
            else
            {
                log.info( "Executor service finished all queued tasks and terminated." );
            }
        }
        catch ( InterruptedException ie )
        {
            log.error( "Interrupted while waiting for pool to terminate; forcing shutdown.", ie );

            // (Re-)Cancel if current thread also interrupted
            List<Runnable> queuedTasks = pool.shutdownNow();

            log.info( "Discarded {} tasks that were still queued; no longer waiting for executor service to terminate."
                    , queuedTasks.size() );

            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }
673874
def execute(self, requests, resp_generator, *args, **kwargs):
        '''
            
        '''
        result_futures = [self.executor_pool.submit(resp_generator, req, *args, **kwargs) for req in requests]
        resp = [res_future.result() for res_future in result_futures]
        return resp
150782
def add_jobs(jobs)
      raise PoolClosed.new("Pool #{self} is currently closed") if pool_closed?
      raise ArgumentError.new("Expecting an array but received: #{jobs.class}") unless jobs.is_a?(Array)
      @queue.pause
      begin
        jobs.each do |job|
          case job
          when Proc
            @queue << [job, []]
          when Array
            raise ArgumentError.new('Jobs to be processed by the pool must be a proc/lambda or [proc/lambda, [*args]]') unless job.size == 2 and job[0].is_a?(Proc) and job[1].is_a?(Array)
            @queue << [job.shift, job]
          else
            raise ArgumentError.new('Jobs to be processed by the pool must be a proc/lambda or [proc/lambda, [*args]]')
          end
        end
      ensure
        num = jobs.size - @threads.select{|t|t.waiting?}.size
        @lock.synchronize{ num.times{ create_thread(:nowait) } if num > 0 }
        @queue.unpause
      end
      true
    end
82175
public void run() {
    if (workQueue.array == null) { // only run once
      Throwable exception = null;
      try {
        onStart();
        pool.runWorker(workQueue);
      } catch (Throwable ex) {
        exception = ex;
      } finally {
        try {
          onTermination(exception);
        } catch (Throwable ex) {
          if (exception == null) {
            exception = ex;
          }
        } finally {
          pool.deregisterWorker(this, exception);
        }
      }
    }
  }
175022
function run(pool, idx, executionFunction) {
  var p = Promise.resolve()
    .then(executionFunction)
    .then(function() {
      delete pool.promises[idx];
      var next = pool.queue.pop();
      if (next) {
        pool.execute(next);
      }
    });
  pool.promises[idx] = p;
  return p;
}
1194087
def process_queue(self):
        
        m = pycurl.CurlMulti()
        m.handles = []

        # Loop the queue and create Curl objects for processing
        for item in self._queue:
            c = self._create_curl(*item)
            m.add_handle(c)
            m.handles.append(c)

        # Process the collected Curl handles
        num_handles = len(m.handles)
        while num_handles:
            while 1:
                # Perform the calls
                ret, num_handles = m.perform()
                if ret != pycurl.E_CALL_MULTI_PERFORM:
                    break
            m.select(1.0)

        # Collect data
        results = []
        for c in m.handles:
            c.status = c.getinfo(c.HTTP_CODE)
            if 'Content-Encoding: gzip' in c.response_headers.getvalue():
                c.body = cStringIO.StringIO(self._gunzip_body(c.body.getvalue()))
            result = { "data" : self._digest_result(c.body.getvalue()), "code": c.status }
            if not c.status or c.status >= 400:
                # Don't throw the exception because some might have succeeded
                result['exception'] = HapiThreadedError(c)

            results.append(result)

            
        # cleanup
        for c in m.handles:
            if hasattr(c, "data_out"):
                c.data_out.close()

            c.body.close()
            c.response_headers.close()
            c.close()
            m.remove_handle(c)

        m.close()
        del m.handles
        self._queue = []

        return results
1090028
def do_work
      done = false

      while not done
        queue_runner(@concurrency, @delay, @retry_count, @progress)

        if @interactive and @verify
          done = verify_results(@group)
          # a retry means we are not done yet.
          done = false if done == 'retry'
        else
          done = true
        end
      end

      if not @verify
        failed_workers, results_display = get_results(group)
        exit 1 if failed_workers.size > 0
      end

      puts 'chimp run complete'
    end
174532
def run(self) -> None:
        
        while not self.cleanup:
            try:
                result, test, additional_info = self.result_queue.get(timeout=1)
            except queue.Empty:
                continue

            self.result_queue.task_done()

            if result == TestState.serialization_failure:
                test = self.tests[test]
                warnings.warn("Serialization error: {} on test {}".format(
                    additional_info, test), SerializationWarning)
                test(self)

            else:
                self.testsRun += 1

                if result == TestState.success:
                    self.addSuccess(test)
                elif result == TestState.failure:
                    self.addFailure(test, additional_info)
                elif result == TestState.error:
                    self.addError(test, additional_info)
                elif result == TestState.skipped:
                    self.addSkip(test, additional_info)
                elif result == TestState.expected_failure:
                    self.addExpectedFailure(test, additional_info)
                elif result == TestState.unexpected_success:
                    self.addUnexpectedSuccess(test)
                else:
                    raise Exception("This is not a valid test type :", result)
365888
synchronized String evaluateInterval() {
        // During shutdown, threadpool may have been nulled out. In that case, don't bother analyzing.
        // (We could log this in FFDC, but it isn't clear that it's worth doing so.)
        if (threadPool == null)
            return "threadPool == null";

        int poolSize = threadPool.getPoolSize();

        // Bail when there are no threads in the pool (unlikely)
        if (poolSize <= 0) {
            return "poolSize <= 0";
        }

        // we can't even think about adjusting the pool size until the underlying executor has aggressively
        // grown the pool to the coreThreads value, so if that hasn't happened yet we should just bail
        if (poolSize < coreThreads) {
            return "poolSize < coreThreads";
        }

        long currentTime = System.currentTimeMillis();
        long completedWork = threadPool.getCompletedTaskCount();

        // Calculate work done for the current interval
        long deltaTime = Math.max(currentTime - lastTimerPop, interval);
        long deltaCompleted = completedWork - previousCompleted;
        double throughput = 1000.0 * deltaCompleted / deltaTime;
        try {
            queueDepth = threadPool.getQueue().size();
            boolean queueEmpty = (queueDepth <= 0);

            // Count the number of consecutive times we've seen an empty queue
            if (!queueEmpty) {
                consecutiveQueueEmptyCount = 0;
            } else if (lastAction != LastAction.SHRINK) { // 9/5/2012
                consecutiveQueueEmptyCount++;
            }

            // update cpu utilization info
            boolean cpuHigh = false;

            processCpuUtil = CpuInfo.getJavaCpuUsage();
            systemCpuUtil = CpuInfo.getSystemCpuUsage();
            cpuUtil = Math.max(systemCpuUtil, processCpuUtil);

            if (cpuUtil > highCpu) {
                cpuHigh = true;
            }

            // Handle pausing the task if the pool has been idle
            if (manageIdlePool(threadPool, deltaCompleted)) {
                return "monitoring paused";
            }

            if (resolveHang(deltaCompleted, queueEmpty, poolSize)) {
                /**
                 * Sleep the controller thread briefly after increasing the poolsize
                 * then update task count before returning to reduce the likelihood
                 * of a false negative hang check next cycle due to a few non-hung
                 * tasks executing on the newly created threads
                 */
                try {
                    Thread.sleep(10);
                } catch (Exception ex) {
                    // do nothing
                }
                completedWork = threadPool.getCompletedTaskCount();
                return "action take to resolve hang";
            }

            if (checkTargetPoolSize(poolSize)) {
                return "poolSize != targetPoolSize";
            }

            controllerCycle++;
            ThroughputDistribution currentStats = getThroughputDistribution(poolSize, true);

            // handleOutliers will mark this 'true' if it resets the distribution
            distributionReset = false;

            // Reset statistics based on abnormal data points
            if (handleOutliers(currentStats, throughput)) {
                return "aberrant workload";
            }

            // If the distribution was reset we don't need to add the datapoint because
            // handleOutliers already did that.
            // If throughput was 0 we will not include that datapoint
            if (!distributionReset && throughput > 0) {
                currentStats.addDataPoint(throughput, controllerCycle);
            }

            boolean lowTput = false;
            if (queueDepth == 0 && throughput < poolSize * lowTputThreadsRatio) {
                lowTput = true;
                if (tc.isEventEnabled()) {
                    Tr.event(tc, "low tput flag set: throughput: " + throughput + ", poolSize: "
                                 + poolSize + ", queueDepth: " + queueDepth,
                             threadPool);
                }
            }

            setPoolIncrementDecrement(poolSize);

            double forecast = currentStats.getMovingAverage();
            double shrinkScore = getShrinkScore(poolSize, forecast, throughput, cpuHigh, lowTput);
            double growScore = getGrowScore(poolSize, forecast, throughput, cpuHigh, lowTput);

            // Adjust the poolsize only if one of the scores is both larger than the scoreFilterLevel
            // and sufficiently larger than the other score. These conditions reduce poolsize fluctuation
            // which might arise due to a weak or noisy signal from the historical throughput data.
            int poolAdjustment = 0;
            if (growScore >= growScoreFilterLevel && (growScore - growShrinkDiffFilter) > shrinkScore) {
                poolAdjustment = poolIncrement;
            } else if (shrinkScore >= shrinkScoreFilterLevel && (shrinkScore - growShrinkDiffFilter) > growScore) {
                poolAdjustment = -poolDecrement;
            }

            // Force some random variation into the pool size algorithm
            poolAdjustment = forceVariation(poolSize, poolAdjustment, deltaCompleted, lowTput);

            // Format an event level trace point with the most useful data
            if (tc.isEventEnabled()) {
                Tr.event(tc, "Interval data", toIntervalData(throughput, forecast, shrinkScore, growScore, poolSize, poolAdjustment));
            }

            // Change the pool size and save the result, will check it at start of next control cycle
            targetPoolSize = adjustPoolSize(poolSize, poolAdjustment);

        } finally {
            lastTimerPop = currentTime;
            previousCompleted = completedWork;
            previousThroughput = throughput;
        }
        return "";
    }
1032598
def done(self, result, noraise=False):
        
        # [??] Should this be in a critical section?

        # Has done() already been called on this task?
        if self.ev_done.is_set():
            # ??
            if isinstance(self.result, Exception) and (not noraise):
                raise self.result
            return self.result

        # calculate running time and other finalization
        self.endtime = time.time()
        try:
            self.totaltime = self.endtime - self.starttime
        except AttributeError:
            # task was not initialized properly
            self.totaltime = 0.0
        self.result = result

        # Release thread waiters
        self.ev_done.set()

        # Perform callbacks for event-style waiters
        self.make_callback('resolved', self.result)

        # If the result is an exception, then our final act is to raise
        # it in the caller, unless the caller explicitly supressed that
        if isinstance(result, Exception) and (not noraise):
            raise result

        return result
658624
def perform
      unless @threads.any? { |t| t.alive? }
        @values.start!
        @size.times do
          @threads << Thread.new do
            while job = @queue.pop
              if job == :END_OF_WORK
                break
              else
                @values.push *job[1], &job[0]
              end
            end
          end
        end
      end
    end
549554
def _thread_main(self):
        
        quit_ = False

        while True:
            items = self._get_items()
            data = []

            for item in items:
                if item is _WORKER_TERMINATOR:
                    quit_ = True
                    # Continue processing items, don't break, try to process
                    # all items we got back before quitting.
                else:
                    data.extend(item)

            if data:
                try:
                    self.exporter.emit(data)
                except Exception:
                    logging.exception(
                        '%s failed to emit data.'
                        'Dropping %s objects from queue.',
                        self.exporter.__class__.__name__,
                        len(data))
                    pass

            for _ in range(len(items)):
                self._queue.task_done()

            # self._event is set at exit, at which point we start draining the
            # queue immediately. If self._event is unset, block for
            # self.wait_period between each batch of exports.
            self._event.wait(self._wait_period)

            if quit_:
                break