* Acquires a slot for CPU-bound work. * * If and as long as the lock-free TryAcquireSlot() doesn't succeed, * subscribes to the slow path by waiting on a condition variable. * It is woken up by Done() which is called by the destructor. * * @param yc Needed to asynchronously wait for the condition variable. * @param strand Where to post the wake-up of the condition variable. */
| 28 | * @param strand Where to post the wake-up of the condition variable. |
| 29 | */ |
| 30 | CpuBoundWork::CpuBoundWork(boost::asio::yield_context yc, boost::asio::io_context::strand& strand) |
| 31 | : m_Done(false) |
| 32 | { |
| 33 | VERIFY(IoEngine::IsStrandRunningOnThisThread(strand)); |
| 34 | |
| 35 | auto& ie (IoEngine::Get()); |
| 36 | Shared<AsioConditionVariable>::Ptr cv; |
| 37 | |
| 38 | while (!TryAcquireSlot()) { |
| 39 | if (!cv) { |
| 40 | cv = Shared<AsioConditionVariable>::Make(ie.GetIoContext()); |
| 41 | } |
| 42 | |
| 43 | { |
| 44 | std::unique_lock lock (ie.m_CpuBoundWaitingMutex); |
| 45 | |
| 46 | // The above lines may take a little bit, so let's optimistically re-check. |
| 47 | // Also mitigate lost wake-ups by re-checking during the lock: |
| 48 | // |
| 49 | // During our lock, Done() can't retrieve the subscribers to wake up, |
| 50 | // so any ongoing wake-up is either done at this point or has not started yet. |
| 51 | // If such a wake-up is done, it's a lost wake-up to us unless we re-check here |
| 52 | // whether the slot being freed (just before the wake-up) is still available. |
| 53 | if (TryAcquireSlot()) { |
| 54 | break; |
| 55 | } |
| 56 | |
| 57 | // If the (hypothetical) slot mentioned above was taken by another coroutine, |
| 58 | // there are no free slots again, just as if no wake-ups happened just now. |
| 59 | ie.m_CpuBoundWaiting.emplace_back(strand, cv); |
| 60 | } |
| 61 | |
| 62 | cv->Wait(yc); |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | /** |
| 67 | * Tries to acquire a slot for CPU-bound work. |