| 112 | } |
| 113 | |
| 114 | std::vector<std::unique_ptr<SerializedPage>> ExchangeQueue::dequeueLocked( |
| 115 | uint32_t maxBytes, |
| 116 | bool* atEnd, |
| 117 | ContinueFuture* future) { |
| 118 | BOLT_CHECK_NOT_NULL(future); |
| 119 | TestValue::adjust( |
| 120 | "bytedance::bolt::exec::ExchangeQueue::dequeueLocked", this); |
| 121 | if (!error_.empty()) { |
| 122 | *atEnd = true; |
| 123 | BOLT_FAIL(error_); |
| 124 | } |
| 125 | |
| 126 | *atEnd = false; |
| 127 | |
| 128 | std::vector<std::unique_ptr<SerializedPage>> pages; |
| 129 | uint32_t pageBytes = 0; |
| 130 | for (;;) { |
| 131 | if (queue_.empty()) { |
| 132 | if (atEnd_) { |
| 133 | *atEnd = true; |
| 134 | } else { |
| 135 | promises_.emplace_back("ExchangeQueue::dequeue"); |
| 136 | *future = promises_.back().getSemiFuture(); |
| 137 | } |
| 138 | return pages; |
| 139 | } |
| 140 | |
| 141 | if (pageBytes > 0 && pageBytes + queue_.front()->size() > maxBytes) { |
| 142 | return pages; |
| 143 | } |
| 144 | |
| 145 | pages.emplace_back(std::move(queue_.front())); |
| 146 | queue_.pop_front(); |
| 147 | pageBytes += pages.back()->size(); |
| 148 | totalBytes_ -= pages.back()->size(); |
| 149 | } |
| 150 | |
| 151 | BOLT_UNREACHABLE(); |
| 152 | } |
| 153 | |
| 154 | void ExchangeQueue::setError(const std::string& error) { |
| 155 | std::vector<ContinuePromise> promises; |