Method enterArbitration

bolt/exec/Operator.cpp:1119–1149

}

void Operator::MemoryReclaimer::enterArbitration() {
  DriverThreadContext* driverThreadCtx = driverThreadContext();
  if (FOLLY_UNLIKELY(driverThreadCtx == nullptr)) {
    // Skips the driver suspension handling if this memory arbitration request
    // is not issued from a driver thread. For example, async streaming shuffle
    // and table scan prefetch execution path might initiate memory arbitration
    // request from non-driver thread.
    return;
  }

  Driver* const runningDriver = driverThreadCtx->driverCtx()->driver;
  if (!FLAGS_bolt_memory_pool_capacity_transfer_across_tasks) {
    if (auto opDriver = ensureDriver()) {
      // NOTE: the current running driver might not be the driver of the
      // operator that requests memory arbitration. The reason is that an
      // operator might extend the buffer allocated from the other operator
      // either from the same or different drivers. But they must be from the
      // same task.
      BOLT_CHECK_EQ(
          runningDriver->task()->taskId(),
          opDriver->task()->taskId(),
          "The current running driver and the request driver must be from the same task");
    }
  }
  if (runningDriver->task()->enterSuspended(runningDriver->state()) !=
      StopReason::kNone) {
    // There is no need for arbitration if the associated task has already
    // terminated.
    BOLT_FAIL("Terminate detected when entering suspension");
  }
}

void Operator::MemoryReclaimer::leaveArbitration() noexcept {
  DriverThreadContext* driverThreadCtx = driverThreadContext();