| 435 | } |
| 436 | |
| 437 | Status KrpcDataStreamSender::Channel::WaitForRpcLocked(std::unique_lock<SpinLock>* lock) { |
| 438 | DCHECK(lock != nullptr); |
| 439 | DCHECK(lock->owns_lock()); |
| 440 | |
| 441 | ScopedTimer<MonotonicStopWatch> timer(parent_->profile()->inactive_timer(), |
| 442 | parent_->state_->total_network_send_timer()); |
| 443 | |
| 444 | // Wait for in-flight RPCs to complete unless the parent sender is closed or cancelled. |
| 445 | while(rpc_in_flight_ && !ShouldTerminate()) { |
| 446 | rpc_done_cv_.wait_for(*lock, std::chrono::milliseconds(50)); |
| 447 | } |
| 448 | int64_t elapsed_time_ns = timer.ElapsedTime(); |
| 449 | if (IsSlowRpc(elapsed_time_ns)) { |
| 450 | LOG(INFO) << "Long delay waiting for RPC to " << address_ |
| 451 | << " (fragment_instance_id=" << PrintId(fragment_instance_id_) << "): " |
| 452 | << "took " << PrettyPrinter::Print(elapsed_time_ns, TUnit::TIME_NS); |
| 453 | } |
| 454 | |
| 455 | if (UNLIKELY(ShouldTerminate())) { |
| 456 | // DSS is single-threaded so it's impossible for shutdown_ to be true here. |
| 457 | DCHECK(!shutdown_); |
| 458 | return Status::CANCELLED; |
| 459 | } |
| 460 | |
| 461 | DCHECK(!rpc_in_flight_); |
| 462 | if (UNLIKELY(!rpc_status_.ok())) { |
| 463 | LOG(ERROR) << "channel send to " << address_ << " failed: (fragment_instance_id=" |
| 464 | << PrintId(fragment_instance_id_) |
| 465 | << "): " << rpc_status_.GetDetail(); |
| 466 | return rpc_status_; |
| 467 | } |
| 468 | return Status::OK(); |
| 469 | } |
| 470 | |
| 471 | void KrpcDataStreamSender::Channel::RetryCb( |
| 472 | const DoRpcFn& rpc_fn, const kudu::Status& cb_status) { |
nothing calls this directly
no test coverage detected