| 144 | /// inserted into the queue. |
| 145 | template <typename V> |
| 146 | bool BlockingPut(V&& val) { |
| 147 | MonotonicStopWatch timer; |
| 148 | int64_t val_bytes = ElemBytesFn()(val); |
| 149 | DCHECK_GE(val_bytes, 0); |
| 150 | std::unique_lock<std::mutex> write_lock(put_lock_); |
| 151 | while (!HasCapacityInternal(write_lock, val_bytes) && !shutdown_) { |
| 152 | if (put_wait_timer_ != nullptr) timer.Start(); |
| 153 | put_cv_.Wait(write_lock); |
| 154 | if (put_wait_timer_ != nullptr) timer.Stop(); |
| 155 | } |
| 156 | if (put_wait_timer_ != nullptr) put_wait_timer_->Add(timer.ElapsedTime()); |
| 157 | if (UNLIKELY(shutdown_)) return false; |
| 158 | |
| 159 | DCHECK_LT(put_list_.size(), max_elements_); |
| 160 | put_bytes_enqueued_ += val_bytes; |
| 161 | Put(std::forward<V>(val)); |
| 162 | write_lock.unlock(); |
| 163 | get_cv_.NotifyOne(); |
| 164 | return true; |
| 165 | } |
| 166 | |
| 167 | /// Puts an element into the queue, waiting until 'timeout_micros' elapses, if there is |
| 168 | /// no space. If the queue is shut down, or if the timeout elapsed without being able to |