private ScheduleResult scheduleTask(TaskInfo taskInfo, Pair<Resource, Map<String, List<NodeInfo>>> availabilityPair,
      Ref<TaskInfo> downgradedTask) {
    Preconditions.checkNotNull(availabilityPair.getLeft(), "totalResource can not be null");
    // If there's no memory available at the cluster, fail
    if (availabilityPair.getLeft().getMemory() <= 0) {
      return SELECT_HOST_RESULT_INADEQUATE_TOTAL_CAPACITY.scheduleResult;
    }
    SelectHostResult selectHostResult = selectHost(taskInfo, availabilityPair.getRight());
    if (selectHostResult.scheduleResult != ScheduleResult.SCHEDULED) {
      return selectHostResult.scheduleResult;
    }
    boolean isGuaranteed = false;
    if (unusedGuaranteed > 0) {
      boolean wasGuaranteed;
      synchronized (taskInfo) {
        assert !taskInfo.isPendingUpdate; // No updates before it's running.
        wasGuaranteed = taskInfo.isGuaranteed;
        taskInfo.isGuaranteed = true;
        isGuaranteed = true;
      }
      if (wasGuaranteed) {
        // This should never happen - we only schedule one attempt once.
        WM_LOG.error("The task had guaranteed flag set before scheduling: " + taskInfo);
      } else {
        int result = --unusedGuaranteed;
        if (metrics != null) {
          metrics.setWmUnusedGuaranteed(result);
        }
        WM_LOG.info("Using an unused duck for " + taskInfo.attemptId
            + "; unused is now " + result);
      }
    } else {
      // We could be scheduling a guaranteed task when a higher priority task cannot be
      // scheduled. Try to take a duck away from a lower priority task here.
      if (findGuaranteedToReallocate(taskInfo, downgradedTask)) {
        // We are revoking another duck; don't wait. We could also give the duck
        // to this task in the callback instead.
        synchronized (taskInfo) {
          assert !taskInfo.isPendingUpdate; // No updates before it's running.
          taskInfo.isGuaranteed = true;
          isGuaranteed = true;
        }
        // Note: after this, the caller MUST send the downgrade message to downgradedTask
        //       (outside of the writeLock, preferably), before exiting.
      }
    }

    // in LLAP, the containers are used exactly once for exactly one attempt
    // the container ids are entirely arbitrary and have 64 bit space per application
    // this is why the isGuaranteed (& any other initial information) can be encoded into the 
    // bits indicating container id
    NodeInfo nodeInfo = selectHostResult.nodeInfo;
    Container container =
        containerFactory.createContainer(nodeInfo.getResourcePerExecutor(), taskInfo.priority,
            nodeInfo.getHost(),
            nodeInfo.getRpcPort(),
            nodeInfo.getServiceAddress(),
            isGuaranteed);
    writeLock.lock(); // While updating local structures
    // Note: this is actually called under the epic writeLock in schedulePendingTasks
    try {
      // The canAccept part of this log message does not account for this allocation.
      assignedTaskCounter.incrementAndGet();
      LOG.info("Assigned #{}, task={} on node={}, to container={}",
          assignedTaskCounter.get(),
          taskInfo, nodeInfo.toShortString(), container.getId());
      dagStats.registerTaskAllocated(taskInfo.requestedHosts, taskInfo.requestedRacks,
          nodeInfo.getHost());
      taskInfo.setAssignmentInfo(nodeInfo, container.getId(), clock.getTime());
      registerRunningTask(taskInfo);
      nodeInfo.registerTaskScheduled();
      // if no more resources on Node -> remove
      if (!nodeInfo.canAcceptTask()) {
        availabilityPair.getRight().get(nodeInfo.getHost()).remove(nodeInfo);
      }
    } finally {
      writeLock.unlock();
    }
    getContext().taskAllocated(taskInfo.task, taskInfo.clientCookie, container);
    return selectHostResult.scheduleResult;
  }
Should cycle when:
 1. There are available Resources OR
 2. There is a pending task with higher pri than running ones
@param availableHostMap
@return
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java