| 359 | } |
| 360 | |
| 361 | void KrpcDataStreamMgr::Cancel(const TUniqueId& query_id) { |
| 362 | VLOG_QUERY << "cancelling active streams for query_id=" << PrintId(query_id); |
| 363 | vector<shared_ptr<KrpcDataStreamRecvr>> receivers_to_cancel; |
| 364 | { |
| 365 | lock_guard<mutex> l(lock_); |
| 366 | // Fragment instance IDs are the query ID with the lower bits set to the instance |
| 367 | // index. Therefore all finstances for a query are clustered together, starting |
| 368 | // after the position in the map where the query_id would be. |
| 369 | FragmentRecvrSet::iterator iter = |
| 370 | fragment_recvr_set_.lower_bound(make_pair(query_id, 0)); |
| 371 | while (iter != fragment_recvr_set_.end() && |
| 372 | GetQueryId(iter->first) == query_id) { |
| 373 | bool unused; |
| 374 | shared_ptr<KrpcDataStreamRecvr> recvr = |
| 375 | FindRecvr(iter->first, iter->second, &unused); |
| 376 | if (recvr != nullptr) { |
| 377 | receivers_to_cancel.push_back(recvr); |
| 378 | } else { |
| 379 | // keep going but at least log it |
| 380 | LOG(ERROR) << Substitute( |
| 381 | "Cancel(): missing in stream_map: fragment_instance_id=$0 node=$1", |
| 382 | PrintId(iter->first), iter->second); |
| 383 | } |
| 384 | ++iter; |
| 385 | } |
| 386 | } |
| 387 | // Call CancelStream() after releasing lock_. |
| 388 | for (shared_ptr<KrpcDataStreamRecvr>& recvr: receivers_to_cancel) { |
| 389 | DCHECK(recvr != nullptr); |
| 390 | recvr->CancelStream(); |
| 391 | } |
| 392 | } |
| 393 | |
| 394 | template<typename ContextType, typename RequestPBType> |
| 395 | void KrpcDataStreamMgr::RespondToTimedOutSender(const std::unique_ptr<ContextType>& ctx) { |
nothing calls this directly
no test coverage detected