| 95 | } |
| 96 | |
| 97 | Status JoinBuilder::WaitForInitialBuild(RuntimeState* join_node_state) { |
| 98 | DCHECK(is_separate_build_); |
| 99 | join_node_state->AddCancellationCV(&separate_build_lock_, &probe_wakeup_cv_); |
| 100 | VLOG(2) << "JoinBuilder (id=" << join_node_id_ << ")" |
| 101 | << " WaitForInitialBuild() called by finstance " |
| 102 | << PrintId(join_node_state->fragment_instance_id()); |
| 103 | unique_lock<mutex> l(separate_build_lock_); |
| 104 | // Wait until either the build is ready to use or this finstance has been cancelled. |
| 105 | // We can't safely pick up the build side if the build side was cancelled - instead we |
| 106 | // need to wait for this finstance to be cancelled. |
| 107 | while (!ready_to_probe_ && !join_node_state->is_cancelled()) { |
| 108 | probe_wakeup_cv_.Wait(l); |
| 109 | } |
| 110 | if (join_node_state->is_cancelled()) { |
| 111 | VLOG(2) << "Finstance " << PrintId(join_node_state->fragment_instance_id()) |
| 112 | << " cancelled while waiting for JoinBuilder (id=" << join_node_id_ << ")"; |
| 113 | return Status::CANCELLED; |
| 114 | } |
| 115 | ++probe_refcount_; |
| 116 | --outstanding_probes_; |
| 117 | VLOG(2) << "JoinBuilder (id=" << join_node_id_ << ")" |
| 118 | << " initial build handoff to finstance " |
| 119 | << PrintId(join_node_state->fragment_instance_id()) |
| 120 | << " probe_refcount_=" << probe_refcount_ |
| 121 | << " outstanding_probes_=" << outstanding_probes_; |
| 122 | DCHECK_GE(outstanding_probes_, 0); |
| 123 | return Status::OK(); |
| 124 | } |
| 125 | |
| 126 | void JoinBuilder::HandoffToProbesAndWait(RuntimeState* build_side_state) { |
| 127 | DCHECK(is_separate_build_) << "Doesn't make sense for embedded builder."; |
no test coverage detected