TLA  Line data    Source code

       1                 : //
       2                 : // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
       3                 : // Copyright (c) 2026 Michael Vandeberg
       4                 : //
       5                 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
       6                 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
       7                 : //
       8                 : // Official repository: https://github.com/cppalliance/capy
       9                 : //
      10                 : 
      11                 : #ifndef BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP
      12                 : #define BOOST_CAPY_SRC_EX_DETAIL_STRAND_QUEUE_HPP
      13                 : 
      14                 : #include <boost/capy/continuation.hpp>
      15                 : #include <boost/capy/detail/config.hpp>
      16                 : #include <boost/capy/ex/frame_allocator.hpp>
      17                 : 
      18                 : namespace boost {
      19                 : namespace capy {
      20                 : namespace detail {
      21                 : 
      22                 : /** Single-threaded intrusive FIFO of pending continuations.
      23                 : 
      24                 :     Links continuations directly through `continuation::reserved` (a
      25                 :     typed `continuation*` round-tripped through the `void*` slot), so
      26                 :     push() carries no per-item allocation.
      27                 : 
      28                 :     @par Thread Safety
      29                 :     Not thread-safe. Caller must externally synchronize push() and
      30                 :     take_all(). dispatch_batch() does not touch queue state and may
      31                 :     run unlocked once the batch has been taken.
      32                 : */
      33                 : class strand_queue
      34                 : {
      35                 :     continuation* head_ = nullptr;
      36                 :     continuation* tail_ = nullptr;
      37                 : 
      38                 : public:
      39 HIT       11443 :     strand_queue() = default;
      40                 :     strand_queue(strand_queue const&) = delete;
      41                 :     strand_queue& operator=(strand_queue const&) = delete;
      42                 : 
      43                 :     /** Returns true if there are no pending continuations. */
      44                 :     bool
      45           20432 :     empty() const noexcept
      46                 :     {
      47           20432 :         return head_ == nullptr;
      48                 :     }
      49                 : 
      50                 :     /** Push a continuation to the queue.
      51                 : 
      52                 :         @param c The continuation to enqueue; see `continuation`
      53                 :             for lifetime and aliasing requirements.
      54                 :     */
      55                 :     void
      56           30340 :     push(continuation& c) noexcept
      57                 :     {
      58           30340 :         c.reserved = nullptr;
      59           30340 :         if(tail_)
      60            9908 :             tail_->reserved = &c;
      61                 :         else
      62           20432 :             head_ = &c;
      63           30340 :         tail_ = &c;
      64           30340 :     }
      65                 : 
      66                 :     /** Batch of taken items for thread-safe dispatch. */
      67                 :     struct taken_batch
      68                 :     {
      69                 :         continuation* head = nullptr;
      70                 :         continuation* tail = nullptr;
      71                 :     };
      72                 : 
      73                 :     /** Take all pending items atomically.
      74                 : 
      75                 :         Removes all items from the queue and returns them as a
      76                 :         batch. The queue is left empty.
      77                 : 
      78                 :         @return The batch of taken items.
      79                 :     */
      80                 :     taken_batch
      81           20432 :     take_all() noexcept
      82                 :     {
      83           20432 :         taken_batch batch{head_, tail_};
      84           20432 :         head_ = tail_ = nullptr;
      85           20432 :         return batch;
      86                 :     }
      87                 : 
      88                 :     /** Resume each continuation in a taken batch.
      89                 : 
      90                 :         Advances past each node before resuming, since the
      91                 :         resumed coroutine may destroy the awaitable (and thus
      92                 :         the continuation) before control returns here.
      93                 : 
      94                 :         @param batch The batch to dispatch.
      95                 : 
      96                 :         @note Thread-safe with respect to push() because the queue
      97                 :             itself is not touched.
      98                 :     */
      99                 :     static
     100                 :     void
     101           20432 :     dispatch_batch(taken_batch& batch)
     102                 :     {
     103           50772 :         while(batch.head)
     104                 :         {
     105           30340 :             continuation* c = batch.head;
     106           30340 :             batch.head = static_cast<continuation*>(c->reserved);
     107           30340 :             safe_resume(c->h);
     108                 :         }
     109           20432 :         batch.tail = nullptr;
     110           20432 :     }
     111                 : };
     112                 : 
     113                 : } // namespace detail
     114                 : } // namespace capy
     115                 : } // namespace boost
     116                 : 
     117                 : #endif