| 61 | } |
| 62 | |
| 63 | void Coordinator::BackendResourceState::MarkBackendFinished( |
| 64 | BackendState* backend_state, vector<BackendState*>* releasable_backend_states) { |
| 65 | lock_guard<SpinLock> lock(lock_); |
| 66 | if (!closed_ |
| 67 | && backend_resource_states_.at(backend_state) == ResourceState::IN_USE) { |
| 68 | // Transition the BackendState to PENDING and update any related counters. |
| 69 | backend_resource_states_.at(backend_state) = ResourceState::PENDING; |
| 70 | ++num_pending_; |
| 71 | --num_in_use_; |
| 72 | |
| 73 | // If the coordinator backend has not been released, but all other have, then the only |
| 74 | // running Backend must be the coordinator. The Coordinator fragment should buffer |
| 75 | // enough rows to allow all other fragments to be released (if result spooling is |
| 76 | // enabled this is especially true, but even without spooling many queries have a |
| 77 | // coordinator fragment that buffers multiple RowBatches). If the client does not |
| 78 | // fetch all rows immediately, then the Coordinator Backend will be long lived |
| 79 | // compared to the rest of the Backends. |
| 80 | bool is_coordinator_the_last_unreleased_backend = |
| 81 | !released_coordinator_ && num_in_use_ == 1; |
| 82 | |
| 83 | // True if the 'Timed Release' heuristic should be triggered. |
| 84 | bool release_backends_timeout_expired = |
| 85 | released_timer_.ElapsedTime() > release_backend_states_delay_ns_; |
| 86 | |
| 87 | // True if the 'Batched Release' heuristic should be triggered. |
| 88 | bool unreleased_backend_threshold_reached = num_pending_ |
| 89 | >= std::max(floor(num_backends_ / batched_release_decay_value_), 1.0); |
| 90 | |
| 91 | // If no Backends are running or if only the Coordinator Backend is running or if both |
| 92 | // the 'Timed Release' and 'Batched Release' heuristic are true, then transition all |
| 93 | // PENDING BackendStates to RELEASABLE and update any state necessary for the |
| 94 | // heuristics. |
| 95 | if (is_coordinator_the_last_unreleased_backend |
| 96 | || (release_backends_timeout_expired && unreleased_backend_threshold_reached)) { |
| 97 | released_timer_.Reset(); |
| 98 | batched_release_decay_value_ *= FLAGS_batched_release_decay_factor; |
| 99 | for (auto backend_resource_state : backend_resource_states_) { |
| 100 | if (backend_resource_state.second == ResourceState::PENDING) { |
| 101 | releasable_backend_states->push_back(backend_resource_state.first); |
| 102 | backend_resource_states_[backend_resource_state.first] = |
| 103 | ResourceState::RELEASABLE; |
| 104 | --num_pending_; |
| 105 | } |
| 106 | } |
| 107 | DCHECK_GE(num_pending_, 0); |
| 108 | } |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | void Coordinator::BackendResourceState::BackendsReleased( |
| 113 | const vector<BackendState*>& released_backend_states) { |