| 1846 | } |
| 1847 | |
| 1848 | std::vector<Operator*> Task::findPeerOperators( |
| 1849 | int pipelineId, |
| 1850 | Operator* caller) { |
| 1851 | std::vector<Operator*> peers; |
| 1852 | const auto operatorId = caller->operatorId(); |
| 1853 | const auto& operatorType = caller->operatorType(); |
| 1854 | const auto splitGroupId = caller->splitGroupId(); |
| 1855 | std::lock_guard<std::timed_mutex> l(mutex_); |
| 1856 | for (auto& driver : drivers_) { |
| 1857 | if (driver == nullptr) { |
| 1858 | continue; |
| 1859 | } |
| 1860 | if (driver->driverCtx()->pipelineId != pipelineId) { |
| 1861 | continue; |
| 1862 | } |
| 1863 | if (driver->driverCtx()->splitGroupId != splitGroupId) { |
| 1864 | continue; |
| 1865 | } |
| 1866 | Operator* peer = driver->findOperator(operatorId); |
| 1867 | BOLT_CHECK_EQ(peer->operatorType(), operatorType); |
| 1868 | peers.push_back(peer); |
| 1869 | } |
| 1870 | return peers; |
| 1871 | } |
| 1872 | |
| 1873 | bool Task::allPeersFinished( |
| 1874 | const core::PlanNodeId& planNodeId, |