| 3006 | } |
| 3007 | |
| 3008 | void Task::createExchangeClientLocked( |
| 3009 | int32_t pipelineId, |
| 3010 | const core::PlanNodeId& planNodeId) { |
| 3011 | BOLT_CHECK_NULL( |
| 3012 | getExchangeClientLocked(pipelineId), |
| 3013 | "Exchange client has been created at pipeline: {} for planNode: {}", |
| 3014 | pipelineId, |
| 3015 | planNodeId); |
| 3016 | BOLT_CHECK_NULL( |
| 3017 | getExchangeClientLocked(planNodeId), |
| 3018 | "Exchange client has been created for planNode: {}", |
| 3019 | planNodeId); |
| 3020 | // Low-water mark for filling the exchange queue is 1/2 of the per worker |
| 3021 | // buffer size of the producers. |
| 3022 | exchangeClients_[pipelineId] = std::make_shared<ExchangeClient>( |
| 3023 | taskId_, |
| 3024 | destination_, |
| 3025 | queryCtx()->queryConfig().maxExchangeBufferSize(), |
| 3026 | addExchangeClientPool(planNodeId, pipelineId), |
| 3027 | queryCtx()->executor()); |
| 3028 | exchangeClientByPlanNode_.emplace(planNodeId, exchangeClients_[pipelineId]); |
| 3029 | } |
| 3030 | |
| 3031 | std::shared_ptr<ExchangeClient> Task::getExchangeClientLocked( |
| 3032 | const core::PlanNodeId& planNodeId) const { |
nothing calls this directly
no test coverage detected