private int preemptOneSession(int maxToPreemt, long maxRunningTime) {
    Queue<PoolGroupSchedulable> poolGroupQueue =
        poolGroupManager.getPreemptQueue();
    while (!poolGroupQueue.isEmpty()) {
      PoolGroupSchedulable poolGroup = poolGroupQueue.poll();
      poolGroup.distributeShare();
      Queue<PoolSchedulable> poolQueue = poolGroup.getPreemptQueue();
      while (!poolQueue.isEmpty()) {
        PoolSchedulable pool = poolQueue.poll();
        pool.distributeShare();
        if (!pool.isPreemptable()) {
          continue;
        }
        Queue<SessionSchedulable> sessionQueue = pool.getPreemptQueue();
        while (!sessionQueue.isEmpty()) {
          SessionSchedulable schedulable = sessionQueue.poll();
          try {
            int overScheduled =
                (int) (schedulable.getGranted() - schedulable.getShare());
            if (overScheduled <= 0) {
              continue;
            }
            maxToPreemt = Math.min(maxToPreemt, overScheduled);
            LOG.info("Trying to preempt " + maxToPreemt + " " + type +
                " from " + schedulable.getSession().getHandle());
            int preempted = preemptSession(
                schedulable, maxToPreemt, maxRunningTime);
            poolGroup.incGranted(-1 * preempted);
            pool.incGranted(-1 * preempted);
            schedulable.incGranted(-1 * preempted);
            return preempted;
          } catch (InvalidSessionHandle e) {
            LOG.warn("Invalid session handle:" +
                schedulable.getSession().getHandle() +
                " Session may be removed");
          } finally {
            // Add back the queue so it can be further preempt for other
            // sessions.
            poolGroupQueue.add(poolGroup);
            poolQueue.add(pool);
          }
        }
      }
    }
    return 0;
  }
private int preemptOneSession ( int maxToPreemt , long maxRunningTime ) { Queue < PoolGroupSchedulable > poolGroupQueue = poolGroupManager . getPreemptQueue ( ) ; while ( ! poolGroupQueue . isEmpty ( ) ) { PoolGroupSchedulable poolGroup = poolGroupQueue . poll ( ) ; poolGroup . distributeShare ( ) ; Queue < PoolSchedulable > poolQueue = poolGroup . getPreemptQueue ( ) ; while ( ! poolQueue . isEmpty ( ) ) { PoolSchedulable pool = poolQueue . poll ( ) ; pool . distributeShare ( ) ; if ( ! pool . isPreemptable ( ) ) { continue ; } Queue < SessionSchedulable > sessionQueue = pool . getPreemptQueue ( ) ; while ( ! sessionQueue . isEmpty ( ) ) { SessionSchedulable schedulable = sessionQueue . poll ( ) ; try { int overScheduled = ( int ) ( schedulable . getGranted ( ) - schedulable . getShare ( ) ) ; if ( overScheduled <= 0 ) { continue ; } maxToPreemt = Math . min ( maxToPreemt , overScheduled ) ; LOG . info ( "Trying to preempt " + maxToPreemt + " " + type + " from " + schedulable . getSession ( ) . getHandle ( ) ) ; int preempted = preemptSession ( schedulable , maxToPreemt , maxRunningTime ) ; poolGroup . incGranted ( - 1 * preempted ) ; pool . incGranted ( - 1 * preempted ) ; schedulable . incGranted ( - 1 * preempted ) ; return preempted ; } catch ( InvalidSessionHandle e ) { LOG . warn ( "Invalid session handle:" + schedulable . getSession ( ) . getHandle ( ) + " Session may be removed" ) ; } finally { poolGroupQueue . add ( poolGroup ) ; poolQueue . add ( pool ) ; } } } } return 0 ; }
Find the most over-scheduled session in the most over-scheduled pool.
Kill tasks from this session.
@param maxToPreemt The maximum number of tasks to kill
@param maxRunningTime The killed task cannot be older than this time
@return The number of tasks actually killed