Lazy workers are instances that are running and reachable but failed to 
    register with the cldb to join the mapr cluster. This trys to find these 
    missing workers and add them to the cluster.
325096
def ConfigureLazyWorkers(self):
    
    lazy_worker_instances = self.__GetMissingWorkers()
    if not lazy_worker_instances:
      return
    reachable_states = self.__AreInstancesReachable(lazy_worker_instances)
    reachable_instances = [t[0] for t in zip(lazy_worker_instances, 
                                             reachable_states) if t[1]]
    print 'reachable_instances: %s' % reachable_instances 
    self.__ConfigureWorkers(reachable_instances)
    return
799839
function MappedLock() {
  if (!(this instanceof MappedLock))
    return MappedLock.create();

  this.jobs = new Map();
  this.busy = new Set();
  this.destroyed = false;
}
931415
protected MultiThreadedWorker getMultiThreadedWorker(SelectionKey key, long threadIdfWQM) {

        MultiThreadedWorker worker = null;
        synchronized (multiThreadedObjectPool) {
            worker = (MultiThreadedWorker) multiThreadedObjectPool.get();
        }

        if (worker == null) {
            worker = new MultiThreadedWorker(this);
        }

        worker.set(key);
        return worker;
    }
57139
def prune_dead_workers
      all_workers = self.class.all
      return if all_workers.empty?
      known_workers = JRUBY ? worker_thread_ids : []
      pids = nil, hostname = self.hostname
      all_workers.each do |worker|
        host, pid, thread, queues = self.class.split_id(worker.id)
        next if host != hostname
        next if known_workers.include?(thread) && pid == self.pid.to_s
        # NOTE: allow flexibility of running workers :
        # 1. worker might run in another JVM instance
        # 2. worker might run as a process (with MRI)
        next if (pids ||= system_pids).include?(pid)
        log! "Pruning dead worker: #{worker}"
        if worker.respond_to?(:unregister_worker)
          worker.unregister_worker
        else # Resque 2.x
          Registry.for(worker).unregister
        end
      end
    end
747359
def result_worker(ctx, result_cls, get_object=False):
    
    g = ctx.obj
    ResultWorker = load_cls(None, None, result_cls)

    result_worker = ResultWorker(resultdb=g.resultdb, inqueue=g.processor2result)

    g.instances.append(result_worker)
    if g.get('testing_mode') or get_object:
        return result_worker

    result_worker.run()
908353
def ready(worker)
      if @work_queue.empty?
        @workers.insert(0, worker) unless @workers.include?(worker)
      else
        job = @work_queue.pop()
        @add_thread.wakeup() unless @add_thread.nil?
        ask_worker(worker, job)
      end
    end
1374043
private void lazyInitWriting() {
        if (!writeInitialized) {
            this.writeInitialized = true;
            this.bulkProcessor = new BulkProcessor(client, resources.getResourceWrite(), settings);
            this.trivialBytesRef = new BytesRef();
            this.bulkEntryWriter = new BulkEntryWriter(settings, BulkCommands.create(settings, metaExtractor, client.clusterInfo.getMajorVersion()));
        }
    }
979740
@SuppressWarnings("unused")
    private Map<Integer, LocalAssignment> readMyTasks(StormClusterState stormClusterState, String topologyId,
                                                      String supervisorId, Assignment assignmentInfo) throws Exception {
        Map<Integer, LocalAssignment> portTasks = new HashMap<>();

        Set<ResourceWorkerSlot> workers = assignmentInfo.getWorkers();
        if (workers == null) {
            LOG.error("No worker found for assignment {}!", assignmentInfo);
            return portTasks;
        }

        for (ResourceWorkerSlot worker : workers) {
            if (!supervisorId.equals(worker.getNodeId())) {
                continue;
            }
            portTasks.put(worker.getPort(), new LocalAssignment(
                    topologyId, worker.getTasks(), Common.topologyIdToName(topologyId), worker.getMemSize(),
                    worker.getCpu(), worker.getJvm(), assignmentInfo.getTimeStamp()));
        }

        return portTasks;
    }
240673
def __AddWorkers(self, num_to_add):
                    
    # Check preconditions
    assert(self.__IsWebUiReady())
    zone_to_ips = self.__GetZoneToWorkerIpsTable()    
    zone_old_new = []
    for zone, ips in zone_to_ips.iteritems():
      num_nodes_in_zone = len(ips) 
      num_nodes_to_add = 0
      zone_old_new.append((zone, num_nodes_in_zone, num_nodes_to_add))    
    print 'num_to_add %s' % num_to_add
    for _ in range(num_to_add):
      zone_old_new.sort(key= lambda z : z[1]+z[2])
      zt = zone_old_new[0]
      zone_old_new[0] = (zt[0], zt[1], zt[2]+1)
    #print zone_old_new
    zone_plan = [(zt[2], zt[0]) for zt in zone_old_new]
    print 'resize plan'
    if self.config.workers_on_spot_instances:
      new_worker_instances = self.__LaunchSpotWorkerInstances(zone_plan)
    else:  
      new_worker_instances = self.__LaunchOnDemandWorkerInstances(zone_plan)
    self.__WaitForInstancesReachable(new_worker_instances)
    self.__ConfigureWorkers(new_worker_instances)
    return
523804
def _thread_worker_fn(samples, batchify_fn, dataset):
    """"""
    if isinstance(samples[0], (list, tuple)):
        batch = [batchify_fn([dataset[i] for i in shard]) for shard in samples]
    else:
        batch = batchify_fn([dataset[i] for i in samples])
    return batch
682581
public static void main(String[] args) throws InterruptedException {
    Microservices ms1 =
        Microservices.builder()
            .discovery(ScalecubeServiceDiscovery::new)
            .transport(ServiceTransports::rsocketServiceTransport)
            .defaultErrorMapper(new ServiceAProviderErrorMapper()) // default mapper for whole node
            .services(
                ServiceInfo.fromServiceInstance(new ServiceAImpl())
                    .errorMapper(new ServiceAProviderErrorMapper()) // mapper per service instance
                    .build())
            .startAwait();

    System.err.println("ms1 started: " + ms1.serviceAddress());

    Microservices ms2 =
        Microservices.builder()
            .discovery(serviceEndpoint -> serviceDiscovery(serviceEndpoint, ms1))
            .transport(ServiceTransports::rsocketServiceTransport)
            .services(
                call -> {
                  ServiceA serviceA =
                      call.errorMapper(
                              new ServiceAClientErrorMapper()) // service client error mapper
                          .api(ServiceA.class);

                  ServiceB serviceB = new ServiceBImpl(serviceA);

                  return Collections.singleton(ServiceInfo.fromServiceInstance(serviceB).build());
                })
            .startAwait();

    System.err.println("ms2 started: " + ms2.serviceAddress());

    ms2.call()
        .api(ServiceB.class)
        .doAnotherStuff(0)
        .subscribe(
            System.out::println,
            th ->
                System.err.println(
                    "No service client mapper is defined for ServiceB, "
                        + "so default scalecube mapper is used! -> "
                        + th),
            () -> System.out.println("Completed!"));

    Thread.currentThread().join();
  }
801092
def join(self, timeout=None, raise_error=False):
        """"""
        self._workers.join(timeout, raise_error)
177676
public synchronized ListenableFuture<?> waitForMinimumWorkers()
    {
        if (currentCount >= executionMinCount) {
            return immediateFuture(null);
        }

        SettableFuture<?> future = SettableFuture.create();
        futures.add(future);

        // if future does not finish in wait period, complete with an exception
        ScheduledFuture<?> timeoutTask = executor.schedule(
                () -> {
                    synchronized (this) {
                        future.setException(new PrestoException(
                                GENERIC_INSUFFICIENT_RESOURCES,
                                format("Insufficient active worker nodes. Waited %s for at least %s workers, but only %s workers are active", executionMaxWait, executionMinCount, currentCount)));
                    }
                },
                executionMaxWait.toMillis(),
                MILLISECONDS);

        // remove future if finished (e.g., canceled, timed out)
        future.addListener(() -> {
            timeoutTask.cancel(true);
            removeFuture(future);
        }, executor);

        return future;
    }
948546
function DaisyChain(workers, concurrency) {
  var self = this
      , queues = this.queues = new Array(workers.length);

  // create queues and inject unidirectional linked list marker.
  var last;
  for (var i = 0; i < workers.length; i++) {
    var queue = queues[i] = async.queue(workers[i], concurrency);
    last = queue;
  }

  // add hooks for empty and drain
  last.empty = function() {
    if (self.empty) {
      async.all(queues, function(q, cb){ cb(q.length() === 0); }, function(yes) { if (yes) self.empty(); });
    }
  }

  last.drain = function() {
    if (self.drain) {
      async.every(queues, function(q, cb){ cb((q.length()+q.running()) === 0); }, function(yes) { if (yes) self.drain(); });
    }
  }
}
1181411
def _get_external_workers(worker):
    
    worker_that_blocked_task = collections.defaultdict(set)
    get_work_response_history = worker._get_work_response_history
    for get_work_response in get_work_response_history:
        if get_work_response['task_id'] is None:
            for running_task in get_work_response['running_tasks']:
                other_worker_id = running_task['worker']
                other_task_id = running_task['task_id']
                other_task = worker._scheduled_tasks.get(other_task_id)
                if other_worker_id == worker._id or not other_task:
                    continue
                worker_that_blocked_task[other_worker_id].add(other_task)
    return worker_that_blocked_task
273142
def runner(opts, utils=None, context=None, whitelist=None):
    '''
    
    '''
    if utils is None:
        utils = {}
    if context is None:
        context = {}
    ret = LazyLoader(
        _module_dirs(opts, 'runners', 'runner', ext_type_dirs='runner_dirs'),
        opts,
        tag='runners',
        pack={'__utils__': utils, '__context__': context},
        whitelist=whitelist,
    )
    # TODO: change from __salt__ to something else, we overload __salt__ too much
    ret.pack['__salt__'] = ret
    return ret
731089
public void setupExecutors() throws ExecutorManagerException {
    final ImmutableSet<Executor> newExecutors = loadExecutors();
    if (newExecutors.isEmpty()) {
      final String error = "No active executors found";
      logger.error(error);
      throw new ExecutorManagerException(error);
    } else {
      this.activeExecutors = newExecutors;
    }
  }