Gets an element from the queue, waiting indefinitely for one to become available. Returns false if we were shut down prior to getting the element, and there are no more elements available.
| 87 | /// Returns false if we were shut down prior to getting the element, and there |
| 88 | /// are no more elements available. |
| 89 | bool BlockingGet(T* out) { |
| 90 | std::unique_lock<std::mutex> read_lock(get_lock_); |
| 91 | |
| 92 | if (UNLIKELY(get_list_.empty())) { |
| 93 | MonotonicStopWatch timer; |
| 94 | // Block off writers while swapping 'get_list_' with 'put_list_'. |
| 95 | std::unique_lock<std::mutex> write_lock(put_lock_); |
| 96 | while (put_list_.empty()) { |
| 97 | DCHECK(get_list_.empty()); |
| 98 | if (UNLIKELY(shutdown_)) return false; |
| 99 | // Note that it's intentional to signal the writer while holding 'put_lock_' to |
| 100 | // avoid the race in which the writer may be signalled between when it checks |
| 101 | // the queue size and when it calls Wait() in BlockingGet(). NotifyAll() is not |
| 102 | // used here to avoid thundering herd which leads to contention (e.g. InitTuple() |
| 103 | // in scanner). |
| 104 | put_cv_.NotifyOne(); |
| 105 | // Sleep with 'get_lock_' held to block off other readers which cannot |
| 106 | // make progress anyway. |
| 107 | if (get_wait_timer_ != nullptr) timer.Start(); |
| 108 | get_cv_.Wait(write_lock); |
| 109 | if (get_wait_timer_ != nullptr) timer.Stop(); |
| 110 | } |
| 111 | DCHECK(!put_list_.empty()); |
| 112 | put_list_.swap(get_list_); |
| 113 | get_list_size_.Store(get_list_.size()); |
| 114 | write_lock.unlock(); |
| 115 | if (get_wait_timer_ != nullptr) get_wait_timer_->Add(timer.ElapsedTime()); |
| 116 | } |
| 117 | |
| 118 | DCHECK(!get_list_.empty()); |
| 119 | *out = std::move(get_list_.front()); |
| 120 | get_list_.pop_front(); |
| 121 | get_list_size_.Store(get_list_.size()); |
| 122 | read_lock.unlock(); |
| 123 | int64_t val_bytes = ElemBytesFn()(*out); |
| 124 | DCHECK_GE(val_bytes, 0); |
| 125 | get_bytes_dequeued_.Add(val_bytes); |
| 126 | // Note that there is a race with any writer if NotifyOne() is called between when |
| 127 | // a writer checks the queue size and when it calls put_cv_.Wait(). If this race |
| 128 | // occurs, a writer can stay blocked even if the queue is not full until the next |
| 129 | // BlockingGet(). The race is benign correctness wise as BlockingGet() will always |
| 130 | // notify a writer with 'put_lock_' held when both lists are empty. |
| 131 | // |
| 132 | // Relatedly, if multiple writers hit the bytes limit of the queue and queue elements |
| 133 | // vary in size, we may not immediately unblock all writers. E.g. if two writers are |
| 134 | // waiting to enqueue elements of N bytes and we dequeue an element of 2N bytes, we |
| 135 | // could wake up both writers but actually only wake up one. This is also benign |
| 136 | // correctness-wise because we will continue to make progress. |
| 137 | put_cv_.NotifyOne(); |
| 138 | return true; |
| 139 | } |
| 140 | |
| 141 | /// Puts an element into the queue, waiting indefinitely until there is space. Rvalues |
| 142 | /// are moved into the queue, lvalues are copied. If the queue is shut down, returns |