static
| 881 | |
| 882 | // static |
| 883 | void Driver::run(std::shared_ptr<Driver> self) { |
| 884 | #ifndef SPARK_COMPATIBLE |
| 885 | process::ThreadNameHolder holder(self->driverCtx()->task->taskId()); |
| 886 | #endif |
| 887 | process::TraceContext trace("Driver::run"); |
| 888 | bytedance::bolt::process::ScopedThreadDebugInfo scopedInfo( |
| 889 | self->driverCtx()->threadDebugInfo); |
| 890 | ScopedDriverThreadContext scopedDriverThreadContext(self->driverCtx()); |
| 891 | std::shared_ptr<BlockingState> blockingState; |
| 892 | RowVectorPtr nullResult; |
| 893 | auto reason = self->runInternal(self, blockingState, nullResult); |
| 894 | |
| 895 | // When Driver runs on an executor, the last operator (sink) must not produce |
| 896 | // any results. |
| 897 | BOLT_CHECK_NULL( |
| 898 | nullResult, |
| 899 | "The last operator (sink) must not produce any results. " |
| 900 | "Results need to be consumed by either a callback or another operator. ") |
| 901 | |
| 902 | // There can be a race between Task terminating and the Driver being on the |
| 903 | // thread and exiting the runInternal() in a blocked state. If this happens |
| 904 | // the Driver won't be closed, so we need to check the Task here and exit w/o |
| 905 | // going into the resume mode waiting on a promise. |
| 906 | if (reason == StopReason::kBlock && |
| 907 | self->task()->shouldStop() == StopReason::kTerminate) { |
| 908 | return; |
| 909 | } |
| 910 | |
| 911 | switch (reason) { |
| 912 | case StopReason::kBlock: |
| 913 | // Set the resume action outside the Task so that, if the |
| 914 | // future is already realized we do not have a second thread |
| 915 | // entering the same Driver. |
| 916 | BlockingState::setResume(blockingState); |
| 917 | return; |
| 918 | |
| 919 | case StopReason::kYield: |
| 920 | // Go to the end of the queue. |
| 921 | enqueue(self); |
| 922 | return; |
| 923 | |
| 924 | case StopReason::kPause: |
| 925 | case StopReason::kTerminate: |
| 926 | case StopReason::kAlreadyTerminated: |
| 927 | case StopReason::kAtEnd: |
| 928 | return; |
| 929 | default: |
| 930 | BOLT_FAIL("Unhandled stop reason"); |
| 931 | } |
| 932 | } |
| 933 | |
| 934 | void Driver::initializeOperatorStats(std::vector<OperatorStats>& stats) { |
| 935 | stats.resize( |