| 967 | } |
| 968 | |
| 969 | void QueryState::MonitorFInstances() { |
| 970 | // Wait for all fragment instances to finish preparing. |
| 971 | discard_result(WaitForPrepare()); |
| 972 | UpdateBackendExecState(); |
| 973 | if (IsTerminalState()) goto done; |
| 974 | |
| 975 | // Once all fragment instances finished preparing successfully, start periodic |
| 976 | // reporting back to the coordinator. |
| 977 | DCHECK(backend_exec_state_ == BackendExecState::EXECUTING) |
| 978 | << BackendExecStateToString(backend_exec_state_); |
| 979 | if (query_ctx().status_report_interval_ms > 0) { |
| 980 | while (!WaitForFinishOrTimeout(GetReportWaitTimeMs())) { |
| 981 | ReportExecStatus(); |
| 982 | } |
| 983 | } else { |
| 984 | WaitForFinish(); |
| 985 | } |
| 986 | UpdateBackendExecState(); |
| 987 | DCHECK(IsTerminalState()); |
| 988 | |
| 989 | done: |
| 990 | if (backend_exec_state_ == BackendExecState::FINISHED) { |
| 991 | for (const auto& entry : fis_map_) { |
| 992 | DCHECK(entry.second->IsDone()); |
| 993 | } |
| 994 | } else { |
| 995 | // If the query execution hit an error, when the final status report is sent, the |
| 996 | // coordinator's response will instruct the QueryState to cancel itself, so Cancel() |
| 997 | // should have always been called by this point. |
| 998 | DCHECK_EQ(is_cancelled_.Load(), 1); |
| 999 | } |
| 1000 | } |
| 1001 | |
| 1002 | void QueryState::AcquireBackendResourceRefcount() { |
| 1003 | DCHECK(!released_backend_resources_); |
no test coverage detected