As used by the dispatch pool and the worker.
967449
def msg_handle(msg, err)

      fexp = Ruote::Exp::FlowExpression.fetch(
        @context, msg['fei'] || msg['workitem']['fei']
      ) rescue nil

      handle(msg, fexp, err)
    end
1158572
def _dispatch(self, textgroup, directory):
        
        self.dispatcher.dispatch(textgroup, path=directory)
1040407
def run(self):
        
        _logger.info('Start dispatcher')
        if dispatcher_env_vars.NNI_MODE == 'resume':
            self.load_checkpoint()

        while True:
            command, data = receive()
            if data:
                data = json_tricks.loads(data)

            if command is None or command is CommandType.Terminate:
                break
            if multi_thread_enabled():
                result = self.pool.map_async(self.process_command_thread, [(command, data)])
                self.thread_results.append(result)
                if any([thread_result.ready() and not thread_result.successful() for thread_result in self.thread_results]):
                    _logger.debug('Caught thread exception')
                    break
            else:
                self.enqueue_command(command, data)
                if self.worker_exceptions:
                    break

        _logger.info('Dispatcher exiting...')
        self.stopping = True
        if multi_thread_enabled():
            self.pool.close()
            self.pool.join()
        else:
            self.default_worker.join()
            self.assessor_worker.join()

        _logger.info('Terminated by NNI manager')
988252
def dispatch_in_thread(channel, message)
      callback_id, class_name, method_name, meta_data, *args = message
      method_name = method_name.to_sym

      # Get the class
      klass = Object.send(:const_get, class_name)

      promise = Promise.new
      cookies = nil

      start_time = Time.now.to_f

      # Check that we are calling on a Task class and a method provide at
      # Task or above in the ancestor chain. (so no :send or anything)
      if safe_method?(klass, method_name)
        promise.resolve(nil)

        # Init and send the method
        promise = promise.then do
          result = nil
          Timeout.timeout(klass.__timeout || @worker_timeout) do
            Thread.current['meta'] = meta_data
            begin
              klass_inst = klass.new(@volt_app, channel, self)
              result = klass_inst.send(method_name, *args)
              cookies = klass_inst.fetch_cookies
            ensure
              Thread.current['meta'] = nil
            end
          end

          result
        end

      else
        # Unsafe method
        promise.reject(RuntimeError.new("unsafe method: #{method_name}"))
      end

      # Called after task runs or fails
      finish = proc do |error|
        if error.is_a?(Timeout::Error)
          # re-raise with a message
          error = Timeout::Error.new("Task Timed Out after #{@worker_timeout} seconds: #{message}")
        end

        run_time = ((Time.now.to_f - start_time) * 1000).round(3)
        Volt.logger.log_dispatch(class_name, method_name, run_time, args, error)
      end

      # Run the promise and pass the return value/error back to the client
      promise.then do |result|
        reply = EJSON.stringify(['response', callback_id, result, nil, cookies])
        channel.send_string_message(reply)

        finish.call
      end.fail do |error|
        begin
          finish.call(error)

          begin
            # Try to send, handle error if we can't convert the result to EJSON
            reply = EJSON.stringify(['response', callback_id, nil, error, cookies])

            # use send_string_message, since we stringify here, not on the other
            # side of Drb.
            channel.send_string_message(reply)
          rescue EJSON::NonEjsonType => e
            # Convert the error into a string so it can be serialized to
            # something.
            error = "#{error.class.to_s}: #{error.to_s}"
            channel.send_message('response', callback_id, nil, error, cookies)
          end

        rescue => e
          Volt.logger.error "Error in fail dispatch: #{e.inspect}"
          Volt.logger.error(e.backtrace.join("\n")) if e.respond_to?(:backtrace)
          raise
        end
      end

    end
1097740
def _dispatch(self, textgroup, directory):
        
        if textgroup.id in self.dispatcher.collection:
            self.dispatcher.collection[textgroup.id].update(textgroup)
        else:
            self.dispatcher.dispatch(textgroup, path=directory)

        for work_urn, work in textgroup.works.items():
            if work_urn in self.dispatcher.collection[textgroup.id].works:
                self.dispatcher.collection[work_urn].update(work)
775780
def dispatch(channel, message)
      # Dispatch the task in the worker pool.  Pas in the meta data
      @worker_pool.post do
        begin
          dispatch_in_thread(channel, message)
        rescue => e
          err = "Worker Thread Exception for #{message}\n"
          err += e.inspect
          err += e.backtrace.join("\n") if e.respond_to?(:backtrace)

          Volt.logger.error(err)
        end
      end
    end
576325
private function pull(): Worker
    {
        if (!$this->isRunning()) {
            throw new StatusError("The pool was shutdown");
        }

        do {
            if ($this->idleWorkers->isEmpty()) {
                if ($this->getWorkerCount() >= $this->maxSize) {
                    // All possible workers busy, so shift from head (will be pushed back onto tail below).
                    $worker = $this->busyQueue->shift();
                } else {
                    // Max worker count has not been reached, so create another worker.
                    $worker = $this->factory->create();
                    $this->workers->attach($worker, 0);
                    break;
                }
            } else {
                // Shift a worker off the idle queue.
                $worker = $this->idleWorkers->shift();
            }

            if ($worker->isRunning()) {
                break;
            }

            $this->workers->detach($worker);
        } while (true);

        $this->busyQueue->push($worker);
        $this->workers[$worker] += 1;

        return $worker;
    }
456714
function (job_name, job_obj) {

        //If we're capturing output, we need to explicitly override job.output()
        if (isMaster && capture_output) {
            var output = [];
            job_obj.output = function (out) {
                if (out instanceof Array && job_obj.options.flatten) {
                    for (var i = 0, l = out.length; i < l; i++) {
                        output.push(out[i]);
                    }
                } else {
                    output.push(out);
                }
            };
            var old_callback = callback;
            callback = function (err) {
                old_callback(err, output);
            };
        }

        //Are we distributing work?
        fork = job_obj.options.fork;

        //Initialise the processor
        processor.init(function () {

            //Start the job
            if (isMaster) {
                processor.startJob(job_name, job_obj, callback);
            }

        });
    }
935455
function Dispatcher(req, res, next) {
            _classCallCheck(this, Dispatcher);

            this.req = req;
            this.res = res;
            this.next = next;
         }
1250316
def worker_thread(context):
    """"""
    queue = context.task_queue
    parameters = context.worker_parameters

    if parameters.initializer is not None:
        if not run_initializer(parameters.initializer, parameters.initargs):
            context.state = ERROR
            return

    for task in get_next_task(context, parameters.max_tasks):
        execute_next_task(task)
        queue.task_done()
615872
def startup
      enable_gc_optimizations
      if Thread.current == Thread.main
        register_signal_handlers
        prune_dead_workers
      end
      run_hook :before_first_fork
      register_worker

      # Fix buffering so we can `rake resque:work > resque.log` and
      # get output from the child in there.
      $stdout.sync = true
    end
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
182793
public static <V> V doWorkInPoolNicely(final Pool<Jedis> pool, final PoolWork<Jedis, V> work) {
        final V result;
        try {
            result = doWorkInPool(pool, work);
        } catch (RuntimeException re) {
            throw re;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return result;
    }
278971
final void signalWork() {
        for (;;) {
            long c; int sp, i; WorkQueue v; WorkQueue[] ws;
            if ((c = ctl) >= 0L)                      // enough workers
                break;
            else if ((sp = (int)c) == 0) {            // no idle workers
                if ((c & ADD_WORKER) != 0L)           // too few workers
                    tryAddWorker(c);
                break;
            }
            else if ((ws = workQueues) == null)
                break;                                // unstarted/terminated
            else if (ws.length <= (i = sp & SMASK))
                break;                                // terminated
            else if ((v = ws[i]) == null)
                break;                                // terminating
            else {
                int ns = sp & ~UNSIGNALLED;
                int vs = v.scanState;
                long nc = (v.stackPred & SP_MASK) | (UC_MASK & (c + AC_UNIT));
                if (sp == vs && U.compareAndSwapLong(this, CTL, c, nc)) {
                    v.scanState = ns;
                    LockSupport.unpark(v.parker);
                    break;
                }
            }
        }
    }
775790
def do_threaded_dispatch(participant, msg)

      msg = Rufus::Json.dup(msg)
        #
        # the thread gets its own copy of the message
        # (especially important if the main thread does something with
        # the message 'during' the dispatch)

      # Maybe at some point a limit on the number of dispatch threads
      # would be OK.
      # Or maybe it's the job of an extension / subclass

      Thread.new { do_dispatch(participant, msg) }
    end
87688
private void showPools(PrintWriter out, boolean advancedView,
      Set<String> poolFilterSet) {
    ResourceReporter reporter = jobTracker.getResourceReporter();
    synchronized (jobTracker) {
      synchronized(scheduler) {
        PoolManager poolManager = scheduler.getPoolManager();
        out.print("<h2>Active Pools</h2>\n");
        out.print("<table border=\"2\" cellpadding=\"5\" cellspacing=\"2\" " +
                  "class=\"tablesorter\">\n");
        out.print("<thead><tr><th>Pool</th><th>Running Jobs</th>" +
            "<th>Preparing Jobs</th>" +
            "<th>Min Maps</th><th>Min Reduces</th>" +
            "<th>Max Maps</th><th>Max Reduces</th>" +
            "<th>Initialized Tasks</th>" +
            "<th>Max Initialized Tasks</th>" +
            "<th>Running/Waiting Maps</th><th>Running/Waiting Reduces</th>" +
            (reporter != null ? "<th>CPU</th><th>Memory</th>" : "") +
            "<th>Map Avg Wait Seccond</th>" +
            "<th>Reduce Avg Wait Second</th>" +
            "</tr></thead><tbody>\n");
        List<Pool> pools = new ArrayList<Pool>(poolManager.getPools());
        Collections.sort(pools, new Comparator<Pool>() {
          public int compare(Pool p1, Pool p2) {
            if (p1.isDefaultPool())
              return 1;
            else if (p2.isDefaultPool())
              return -1;
            else return p1.getName().compareTo(p2.getName());
          }});
        int numActivePools = 0;
        int totalInitedTasks = 0;
        int totalMaxInitedTasks = 0;
        int totalRunningMaps = 0;
        int totalWaitingMaps = 0;
        int totalRunningReduces = 0;
        int totalWaitingReduces = 0;
        int totalMinReduces = 0;
        int totalMaxReduces = 0;
        int totalMinMaps = 0;
        int totalMaxMaps = 0;
        int totalRunningJobs = 0;
        long totalMapWaitTime = 0;
        long totalReduceWaitTime = 0;
        long totalNonConfiguredFirstMapWaitTime = 0;
        long totalNonConfiguredFirstReduceWaitTime = 0;
        long totalJobsInNonConfiguredPools = 0;
        int totalReduceTasks = 0;
        int totalMapTasks = 0;
        int totalPrepareJobs = 0;
        double totalCpu = 0;
        double totalMemory = 0;
        for (Pool pool: pools) {
          String poolName = pool.getName();
          if ((poolFilterSet != null) && !poolFilterSet.contains(poolName)) {
            continue;
          }
          int initedTasks = 0;
          int runningMaps = 0;
          int waitingMaps = 0;
          int runningReduces = 0;
          int waitingReduces = 0;
          long poolMapWaitTime = 0;
          long poolReduceWaitTime = 0;
          int poolMapTasks = 0;
          int poolReduceTasks = 0;
          int poolPrepareJobs = 0;
          long poolFirstMapWaitTime = 0;
          long poolFirstReduceWaitTime = 0;
          boolean isConfiguredPool = pool.isConfiguredPool();
          for (JobInProgress job: pool.getJobs()) {
            if (job.getStatus().getRunState() == JobStatus.PREP) {
              poolPrepareJobs += 1;
            }
            if (isConfiguredPool) {
              totalJobsInNonConfiguredPools++;
              totalNonConfiguredFirstMapWaitTime += job.getFirstMapWaitTime();
              totalNonConfiguredFirstReduceWaitTime += job.getFirstReduceWaitTime();
            }
            JobInfo info = scheduler.infos.get(job);
            if (info != null) {
              initedTasks += info.totalInitedTasks;
              runningMaps += info.runningMaps;
              runningReduces += info.runningReduces;
              waitingMaps += info.neededMaps;
              waitingReduces += info.neededReduces;
              poolMapWaitTime += job.getTotalMapWaitTime();
              poolReduceWaitTime += job.getTotalReduceWaitTime();
              poolMapTasks += job.desiredMaps();
              poolReduceTasks += job.desiredReduces();
            }
          }
          double poolMapAverageWaitTime = 0;
          double poolReduceAverageWaitTime = 0;
          if (poolMapTasks != 0) {
            poolMapAverageWaitTime = (double)poolMapWaitTime / poolMapTasks;
            totalMapWaitTime += poolMapWaitTime;
            totalMapTasks += poolMapTasks;
          }
          if (poolReduceTasks != 0) {
            poolReduceAverageWaitTime = (double)poolReduceWaitTime / poolReduceTasks;
            totalReduceWaitTime += poolReduceWaitTime;
            totalReduceTasks += poolReduceTasks;
          }
          int runningJobs = pool.getJobs().size();
          int minMaps = poolManager.getMinSlots(poolName, TaskType.MAP);
          int minReduces = poolManager.getMinSlots(poolName, TaskType.REDUCE);
          int maxMaps = poolManager.getMaxSlots(poolName, TaskType.MAP);
          int maxReduces = poolManager.getMaxSlots(poolName, TaskType.REDUCE);
          int maxInitedTasks = poolManager.getPoolMaxInitedTasks(poolName);
          totalRunningJobs += runningJobs;
          totalInitedTasks += initedTasks;
          totalRunningMaps += runningMaps;
          totalWaitingMaps += waitingMaps;
          totalRunningReduces += runningReduces;
          totalWaitingReduces += waitingReduces;
          totalMinMaps += minMaps;
          totalMinReduces += minReduces;
          if (runningJobs == 0 && minMaps == 0 && minReduces == 0 &&
              maxMaps == Integer.MAX_VALUE && maxReduces == Integer.MAX_VALUE &&
              initedTasks == 0 && runningMaps == 0 && runningReduces == 0) {
            continue;
          }
          numActivePools++;
          out.print("<tr>\n");
          out.printf("<td>%s</td>\n", poolName);
          out.printf("<td>%s</td>\n", runningJobs);
          out.printf("<td>%s</td>\n", poolPrepareJobs);
          out.printf("<td>%s</td>\n", minMaps);
          out.printf("<td>%s</td>\n", minReduces);
          if (maxMaps == Integer.MAX_VALUE) {
            out.printf("<td>-</td>\n");
          } else {
            out.printf("<td>%s</td>\n", maxMaps);
            totalMaxMaps += maxMaps;
          }
          if (maxReduces == Integer.MAX_VALUE) {
            out.printf("<td>-</td>\n");
          } else {
            out.printf("<td>%s</td>\n", maxReduces);
            totalMaxReduces += maxReduces;
          }
          out.printf("<td>%s</td>\n", initedTasks);
          if (maxInitedTasks == Integer.MAX_VALUE) {
            out.printf("<td>-</td>\n");
          } else {
            out.printf("<td>%s</td>\n", maxInitedTasks);
            totalMaxInitedTasks += maxInitedTasks;
          }
          out.printf("<td>%s/%s</td>\n", runningMaps, waitingMaps);
          out.printf("<td>%s/%s</td>\n", runningReduces, waitingReduces);

          // Compute the CPU and memory usage
          double cpuUsage = 0; // in percentage
          double memoryUsage = 0; // in percentage
          if (reporter != null) {
            for (JobInProgress job : pool.getJobs()) {
              double cpu =
                reporter.getJobCpuPercentageOnCluster(job.getJobID());
              double memory =
                reporter.getJobMemPercentageOnCluster(job.getJobID());
              cpuUsage += cpu != ResourceReporter.UNAVAILABLE ? cpu : 0;
              memoryUsage += memory != ResourceReporter.UNAVAILABLE ?
                             memory : 0;
            }
            out.printf("<td>%.1f%%</td>\n", cpuUsage);
            out.printf("<td>%.1f%%</td>\n", memoryUsage);
          }
          totalCpu += cpuUsage;
          totalMemory += memoryUsage;
          totalPrepareJobs += poolPrepareJobs;
          out.printf("<td>%.1f</td>\n", poolMapAverageWaitTime / 1000D);
          out.printf("<td>%.1f</td>\n", poolReduceAverageWaitTime / 1000D);
          out.print("</tr>\n");
        }
        out.print("<tr>\n");
        out.printf("<td>Total</td>\n");
        out.printf("<td>%s</td>\n", totalRunningJobs);
        out.printf("<td>%s</td>\n", totalPrepareJobs);
        out.printf("<td>%s</td>\n", totalMinMaps);
        out.printf("<td>%s</td>\n", totalMinReduces);
        if (totalMaxMaps == 0) {
          out.printf("<td>-</td>\n");
        } else {
          out.printf("<td>%s</td>\n", totalMaxMaps);
        }
        if (totalMaxReduces == 0) {
          out.printf("<td>-</td>\n");
        } else {
          out.printf("<td>%s</td>\n", totalMaxReduces);
        }
        out.printf("<td>%s</td>\n", totalInitedTasks);
        out.printf("<td>%s</td>\n", totalMaxInitedTasks);
        out.printf("<td>%s/%s</td>\n", totalRunningMaps, totalWaitingMaps);
        out.printf("<td>%s/%s</td>\n",
                   totalRunningReduces, totalWaitingReduces);

        if (reporter != null) {
          out.printf("<td>%.1f%%</td>\n", totalCpu);
          out.printf("<td>%.1f%%</td>\n", totalMemory);
        }
        double mapAverageWaitTime = totalMapTasks == 0 ? 0 :
            (double)totalMapWaitTime / totalMapTasks;
        double reduceAverageWaitTime = totalReduceTasks == 0 ? 0 :
            (double)totalReduceWaitTime / totalReduceTasks;
        out.printf("<td>%.1f</td>\n", mapAverageWaitTime);
        out.printf("<td>%.1f</td>\n", reduceAverageWaitTime);
        out.print("</tr>\n");

        out.print("</tbody></table>\n");
        out.printf("<p>Number of active/total pools : %d/%d</p>",
                   numActivePools, pools.size());
        double nonConfiguredAverageFirstMapWaitTime = totalJobsInNonConfiguredPools == 0 ? 0:
            (double)totalNonConfiguredFirstMapWaitTime / totalJobsInNonConfiguredPools;
        double nonConfiguredAverageFirstReduceWaitTime = totalJobsInNonConfiguredPools == 0 ? 0:
            (double)totalNonConfiguredFirstReduceWaitTime / totalJobsInNonConfiguredPools;
        // Non-configured == ad-hoc.
        out.printf("<p>Average first map wait time in ad-hoc pools: %f</p>", nonConfiguredAverageFirstMapWaitTime);
        out.printf("<p>Average first reduce wait time in ad-hoc pools: %f</p>", nonConfiguredAverageFirstReduceWaitTime);
      }
    }
  }
1268236
function dispatchQueuedRequests()
  {
    me.requestQueue.forEach(me.worker.send.bind(me.worker));
    me.requestQueue = [];
  }