| 201 | } |
| 202 | |
| 203 | size_t SCQueueSynchronizer::consumer_fetch(size_t max, size_t min) { |
| 204 | mgb_assert(max >= min && min >= 1); |
| 205 | size_t spin = 0, cur_finished = m_finished_task.load(std::memory_order_relaxed); |
| 206 | |
| 207 | // relaxed mem order suffices because acquire would be called for ret |
| 208 | while (m_tot_task.load(std::memory_order_relaxed) < cur_finished + min) { |
| 209 | ++spin; |
| 210 | if (spin >= m_max_spin) { |
| 211 | while (m_consumer_waiting.test_and_set(std::memory_order_relaxed)) |
| 212 | ; |
| 213 | SpinlockReleaser releaser(m_consumer_waiting); |
| 214 | |
| 215 | std::unique_lock<std::mutex> lock(m_mtx_more_task); |
| 216 | if (m_should_exit.load(std::memory_order_relaxed)) |
| 217 | return 0; |
| 218 | if (m_tot_task.load(std::memory_order_relaxed) >= cur_finished + min) |
| 219 | break; |
| 220 | m_cv_more_task.wait(lock); |
| 221 | } |
| 222 | if (m_should_exit.load(std::memory_order_relaxed)) |
| 223 | return 0; |
| 224 | } |
| 225 | auto ret = std::min(m_tot_task.load(std::memory_order_acquire) - cur_finished, max); |
| 226 | mgb_assert(ret >= min); |
| 227 | return ret; |
| 228 | } |
| 229 | |
| 230 | void SCQueueSynchronizer::consumer_commit(size_t nr) { |
| 231 | auto done = m_finished_task.fetch_add(nr, std::memory_order_relaxed) + nr; |