Method runInternal

bolt/exec/Driver.cpp:546–866

}

StopReason Driver::runInternal(
    std::shared_ptr<Driver>& self,
    std::shared_ptr<BlockingState>& blockingState,
    RowVectorPtr& result) {
  const auto now = getCurrentTimeMicro();
  const auto queuedTime = (now - queueTimeStartMicros_) * 1'000;
  // Update the next operator's queueTime.
  StopReason stop =
      closed_ ? StopReason::kTerminate : task()->enter(state_, now);
  if (stop != StopReason::kNone) {
    if (stop == StopReason::kTerminate) {
      // ctx_ still has a reference to the Task. 'this' is not on
      // thread from the Task's viewpoint, hence no need to call
      // close().
      ctx_->task->setError(std::make_exception_ptr(BoltRuntimeError(
          __FILE__,
          __LINE__,
          __FUNCTION__,
          "",
          "Cancelled",
          error_source::kErrorSourceRuntime,
          error_code::kInvalidState,
          false)));
    }
    return stop;
  }

  // Update the queued time after entering the Task to ensure the stats have not
  // been deleted.
  if (curOperatorId_ < operators_.size()) {
    operators_[curOperatorId_]->addRuntimeStat(
        "queuedWallNanos",
        RuntimeCounter(queuedTime, RuntimeCounter::Unit::kNanos));
  }

  CancelGuard guard(task().get(), &state_, [&](StopReason reason) {
    // This is run on error or cancel exit.
    if (reason == StopReason::kTerminate) {
      task()->setError(std::make_exception_ptr(BoltRuntimeError(
          __FILE__,
          __LINE__,
          __FUNCTION__,
          "",
          "Cancelled",
          error_source::kErrorSourceRuntime,
          error_code::kInvalidState,
          false)));
    }
    close();
  });

  try {
    // Invoked to initialize the operators once before driver starts execution.
    initializeOperators();

    BOLT_TEST_ADJUST("bytedance::bolt::exec::Driver::runInternal", this);

    const int32_t numOperators = operators_.size();