Container allocation is asynchronous. Requests all containers in the input set serially
to ensure allocated resources match the required resources.
343804
void scheduleHeronWorkers(Set<ContainerPlan> containers) throws ContainerAllocationException {
    for (ContainerPlan containerPlan : containers) {
      int id = containerPlan.getId();
      if (containerPlans.containsKey(containerPlan.getId())) {
        throw new ContainerAllocationException("Received duplicate allocation request for " + id);
      }
      Resource reqResource = containerPlan.getRequiredResource();
      containerPlans.put(id, containerPlan);
      requestContainerForWorker(id, new HeronWorker(id, reqResource));
    }
  }
862990
public void requestAndLaunchContainers() throws Exception {

    if (masterExists()) {
      InetAddress address = InetAddress.getByName(mMasterAddress);
      mMasterContainerNetAddress = address.getHostAddress();
      LOG.info("Found master already running on " + mMasterAddress);
    } else {
      LOG.info("Configuring master container request.");
      Resource masterResource = Records.newRecord(Resource.class);
      masterResource.setMemory(mMasterMemInMB);
      masterResource.setVirtualCores(mMasterCpu);
      mContainerAllocator = new ContainerAllocator("master", 1, 1, masterResource, mYarnClient,
              mRMClient, mMasterAddress);
      List<Container> masterContainers = mContainerAllocator.allocateContainers();
      launchMasterContainer(Iterables.getOnlyElement(masterContainers));
    }

    Resource workerResource = Records.newRecord(Resource.class);
    workerResource.setMemory(mWorkerMemInMB + mRamdiskMemInMB);
    workerResource.setVirtualCores(mWorkerCpu);
    mContainerAllocator = new ContainerAllocator("worker", mNumWorkers, mMaxWorkersPerHost,
        workerResource, mYarnClient, mRMClient);
    List<Container> workerContainers = mContainerAllocator.allocateContainers();
    for (Container container : workerContainers) {
      launchWorkerContainer(container);
    }
    LOG.info("Master and workers are launched");
  }
219872
function (next) {

        async.each(input, function (path, done) {
            fs.stat(path, function (err, stats) {
                if (err) return done(err);
                if (stats.isDirectory())
                    return done("Cannot have a directory as input");
                done();
            });
        }, next);

    }
210587
def worker_allocator(self, async_loop, to_do, **kwargs):
        
        d = kwargs
        threading.Thread(
            target=self._asyncio_thread, args=(async_loop, to_do, d)
        ).start()
65028
function initServices(container) {
        var calls = [];
        _.each(resources, function (resource) {
            if (!resource || !resource.adapters) { return; }

            var adapter = resource.adapters[container];

            if (adapter && adapter !== 'apiclient') {
                var service = pancakes.getService(resource.name);
                if (service.init) {
                    calls.push(service.init.bind(service));
                }
            }
        });

        return chainPromises(calls, null);
    }
396753
private synchronized void initializeListOfPreviousContainers() {
    if (this.previousContainers == null) {
      final List<Container> yarnPrevContainers =
          this.registration.getRegistration().getContainersFromPreviousAttempts();

      // If it's still null, create an empty list to indicate that it's not a restart.
      if (yarnPrevContainers == null) {
        this.previousContainers = Collections.unmodifiableSet(new HashSet<Container>());
      } else {
        this.previousContainers = Collections.unmodifiableSet(new HashSet<>(yarnPrevContainers));
      }

      yarnContainerManager.onContainersRecovered(this.previousContainers);
    }
  }
304878
public void run() {
        synchronized (this) {
            started = true;
        }
        finished = false;
        finish = false;

        try {
            while (true) {
                waitForInput(is);

                if (finish || Thread.interrupted()) {
                    break;
                }

                int availableBytes = is.available();
                byte[] buf = new byte[availableBytes];
                int actualBytesRead = is.read(buf, 0, availableBytes);

                if (actualBytesRead <= 0 || finish || Thread.interrupted()) {
                    break;
                }
                os.write(buf, 0, actualBytesRead);
                if (autoflush) {
                    os.flush();
                }
            }
            os.flush();
        } catch (InterruptedException ie) {
            // likely PumpStreamHandler trying to stop us
        } catch (Exception e) {
            synchronized (this) {
                exception = e;
            }
        } finally {
            if (closeWhenExhausted) {
                FileUtils.close(os);
            }
            finished = true;
            synchronized (this) {
                notifyAll();
            }
        }
    }
381021
private void flushIfPossible(boolean flushAll) throws IOException {
    if (flushAll) {
      flushLock.lock();
    } else if (pendingRequests.isEmpty() || !flushLock.tryLock()) {
      return;
    }
    try {
      do {
        flushPendingRequests();
        if (flushAll) {
          awaitRequestsCompletion();
        }
      } while (flushAll && (!pendingRequests.isEmpty() || !responseFutures.isEmpty()));
    } finally {
      flushLock.unlock();
    }
  }
910017
@Synchronized
    private void checkAssignment() {
        long traceId = LoggerHelpers.traceEnter(log, "checkAssignment");
        try {
            Exceptions.checkNotClosed(closed.get(), this);

            // Fetch the list of containers that is supposed to be owned by this host.
            Set<Integer> desiredList = getDesiredContainerList();
            if (desiredList != null) {
                Collection<Integer> runningContainers = new HashSet<>(this.handles.keySet());
                Collection<Integer> containersPendingTasks = new HashSet<>(this.pendingTasks);

                // Filter out containers which have pending tasks so we don't initiate conflicting events on the same
                // containers. Events for these containers will be tried on subsequent runs of this executor.
                Collection<Integer> containersToBeStarted = CollectionHelpers.filterOut(desiredList, runningContainers);
                containersToBeStarted = CollectionHelpers.filterOut(containersToBeStarted, containersPendingTasks);

                Collection<Integer> containersToBeStopped = CollectionHelpers.filterOut(runningContainers, desiredList);
                containersToBeStopped = CollectionHelpers.filterOut(containersToBeStopped, containersPendingTasks);

                log.info("Container Changes: Desired = {}, Current = {}, PendingTasks = {}, ToStart = {}, ToStop = {}.",
                        desiredList, runningContainers, containersPendingTasks, containersToBeStarted,
                        containersToBeStopped);

                // Initiate the start and stop tasks asynchronously.
                containersToBeStarted.forEach(this::startContainer);
                containersToBeStopped.forEach(this::stopContainer);
            } else {
                log.warn("No segment container assignments found");
            }
        } catch (Throwable e) {
            // Need to catch all exceptions here since throwing any exception here will halt this scheduled job.
            log.warn("Failed to monitor the segmentcontainer assignment: ", e);
        } finally {
            LoggerHelpers.traceLeave(log, "checkAssignment", traceId);
        }
    }
637768
function _resyncList() {
  var listKey = this.listKey;
  var parentPath = this.parentPath;
  if (!listKey || !parentPath) return;

  var newContainer = parentPath.node[listKey];
  if (this.container === newContainer) return;

  // container is out of sync. this is likely the result of it being reassigned

  if (newContainer) {
    this.container = newContainer;
  } else {
    this.container = null;
  }
}
1052694
@Override
    public boolean flushPuts(long waitTime, TimeUnit timeUnit) throws InterruptedException
    {
        long    msWaitRemaining = TimeUnit.MILLISECONDS.convert(waitTime, timeUnit);
        synchronized(putCount)
        {
            while ( putCount.get() > 0 )
            {
                if ( msWaitRemaining <= 0 )
                {
                    return false;
                }

                long        startMs = System.currentTimeMillis();

                putCount.wait(msWaitRemaining);

                long        elapsedMs = System.currentTimeMillis() - startMs;
                msWaitRemaining -= elapsedMs;
            }
        }
        return true;
    }
901878
def wait(state)
      if !LXC::Shell.valid_state?(status.state)
        raise ArgumentError, "Invalid container state: #{state}"
      end

      run("wait", "-s", state)
    end
159418
public static <T> SortedSet<T> asSynchronized(SortedSet<T> self) {
        return Collections.synchronizedSortedSet(self);
    }
630774
function AsyncLoop(iterations, _fn, _successCallback, offset) {
            if (offset === void 0) { offset = 0; }
            this.iterations = iterations;
            this._fn = _fn;
            this._successCallback = _successCallback;
            this.index = offset - 1;
            this._done = false;
        }
1219427
@Override
    public AllocationPoint allocateMemory(DataBuffer buffer, AllocationShape requiredMemory, boolean initialize) {
        // by default we allocate on initial location
        AllocationPoint point = null;

        if (configuration.getMemoryModel() == Configuration.MemoryModel.IMMEDIATE) {
            point = allocateMemory(buffer, requiredMemory, memoryHandler.getInitialLocation(), initialize);
        } else if (configuration.getMemoryModel() == Configuration.MemoryModel.DELAYED) {
            // for DELAYED memory model we allocate only host memory, regardless of firstMemory configuration value
            point = allocateMemory(buffer, requiredMemory, AllocationStatus.HOST, initialize);
        }

        return point;
    }
254793
private void awaitRequestsCompletion() throws IOException {
    // Don't wait until all requests will be completed if enough requests are pending for full batch
    while (!responseFutures.isEmpty() && pendingRequests.size() < maxRequestsPerBatch) {
      try {
        responseFutures.remove().get();
      } catch (InterruptedException | ExecutionException e) {
        if (e.getCause() instanceof IOException) {
          throw (IOException) e.getCause();
        }
        throw new RuntimeException("Failed to execute batch", e);
      }
    }
  }
1203439
@Override
    public void synchronizeThreadDevice(Long threadId, Integer deviceId, AllocationPoint point) {
        // we synchronize only if this AllocationPoint was used within device context, so for multiple consequent syncs only first one will be issued
        flowController.synchronizeToHost(point);
    }