detail/impl/strand_service.ipp

0001 //
0002 // detail/impl/strand_service.ipp
0003 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
0004 //
0005 // Copyright (c) 2003-2025 Christopher M. Kohlhoff (chris at kohlhoff dot com)
0006 //
0007 // Distributed under the Boost Software License, Version 1.0. (See accompanying
0008 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0009 //
0010
0011 #ifndef BOOST_ASIO_DETAIL_IMPL_STRAND_SERVICE_IPP
0012 #define BOOST_ASIO_DETAIL_IMPL_STRAND_SERVICE_IPP
0013
0014 #if defined(_MSC_VER) && (_MSC_VER >= 1200)
0015 # pragma once
0016 #endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
0017
0018 #include <boost/asio/detail/config.hpp>
0019 #include <boost/asio/detail/call_stack.hpp>
0020 #include <boost/asio/detail/strand_service.hpp>
0021
0022 #include <boost/asio/detail/push_options.hpp>
0023
0024 namespace boost {
0025 namespace asio {
0026 namespace detail {
0027
0028 struct strand_service::on_do_complete_exit
0029 {
0030   io_context_impl* owner_;
0031   strand_impl* impl_;
0032
0033   ~on_do_complete_exit()
0034   {
0035     impl_->mutex_.lock();
0036     impl_->ready_queue_.push(impl_->waiting_queue_);
0037     bool more_handlers = impl_->locked_ = !impl_->ready_queue_.empty();
0038     impl_->mutex_.unlock();
0039
0040     if (more_handlers)
0041       owner_->post_immediate_completion(impl_, true);
0042   }
0043 };
0044
0045 strand_service::strand_service(boost::asio::io_context& io_context)
0046   : boost::asio::detail::service_base<strand_service>(io_context),
0047     io_context_(io_context),
0048     io_context_impl_(boost::asio::use_service<io_context_impl>(io_context)),
0049     mutex_(),
0050     salt_(0)
0051 {
0052 }
0053
0054 void strand_service::shutdown()
0055 {
0056   op_queue<operation> ops;
0057
0058   boost::asio::detail::mutex::scoped_lock lock(mutex_);
0059
0060   for (std::size_t i = 0; i < num_implementations; ++i)
0061   {
0062     if (strand_impl* impl = implementations_[i].get())
0063     {
0064       ops.push(impl->waiting_queue_);
0065       ops.push(impl->ready_queue_);
0066     }
0067   }
0068 }
0069
0070 void strand_service::construct(strand_service::implementation_type& impl)
0071 {
0072   boost::asio::detail::mutex::scoped_lock lock(mutex_);
0073
0074   std::size_t salt = salt_++;
0075 #if defined(BOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION)
0076   std::size_t index = salt;
0077 #else // defined(BOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION)
0078   std::size_t index = reinterpret_cast<std::size_t>(&impl);
0079   index += (reinterpret_cast<std::size_t>(&impl) >> 3);
0080   index ^= salt + 0x9e3779b9 + (index << 6) + (index >> 2);
0081 #endif // defined(BOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION)
0082   index = index % num_implementations;
0083
0084   if (!implementations_[index])
0085   {
0086     execution_context::allocator<void> alloc(context());
0087     implementations_[index] = allocate_shared<strand_impl>(alloc);
0088   }
0089   impl = implementations_[index].get();
0090 }
0091
0092 bool strand_service::running_in_this_thread(
0093     const implementation_type& impl) const
0094 {
0095   return call_stack<strand_impl>::contains(impl) != 0;
0096 }
0097
0098 struct strand_service::on_dispatch_exit
0099 {
0100   io_context_impl* io_context_impl_;
0101   strand_impl* impl_;
0102
0103   ~on_dispatch_exit()
0104   {
0105     impl_->mutex_.lock();
0106     impl_->ready_queue_.push(impl_->waiting_queue_);
0107     bool more_handlers = impl_->locked_ = !impl_->ready_queue_.empty();
0108     impl_->mutex_.unlock();
0109
0110     if (more_handlers)
0111       io_context_impl_->post_immediate_completion(impl_, false);
0112   }
0113 };
0114
0115 void strand_service::do_dispatch(implementation_type& impl, operation* op)
0116 {
0117   // If we are running inside the io_context, and no other handler already
0118   // holds the strand lock, then the handler can run immediately.
0119   bool can_dispatch = io_context_impl_.can_dispatch();
0120   impl->mutex_.lock();
0121   if (can_dispatch && !impl->locked_)
0122   {
0123     // Immediate invocation is allowed.
0124     impl->locked_ = true;
0125     impl->mutex_.unlock();
0126
0127     // Indicate that this strand is executing on the current thread.
0128     call_stack<strand_impl>::context ctx(impl);
0129
0130     // Ensure the next handler, if any, is scheduled on block exit.
0131     on_dispatch_exit on_exit = { &io_context_impl_, impl };
0132     (void)on_exit;
0133
0134     op->complete(&io_context_impl_, boost::system::error_code(), 0);
0135     return;
0136   }
0137
0138   if (impl->locked_)
0139   {
0140     // Some other handler already holds the strand lock. Enqueue for later.
0141     impl->waiting_queue_.push(op);
0142     impl->mutex_.unlock();
0143   }
0144   else
0145   {
0146     // The handler is acquiring the strand lock and so is responsible for
0147     // scheduling the strand.
0148     impl->locked_ = true;
0149     impl->mutex_.unlock();
0150     impl->ready_queue_.push(op);
0151     io_context_impl_.post_immediate_completion(impl, false);
0152   }
0153 }
0154
0155 void strand_service::do_post(implementation_type& impl,
0156     operation* op, bool is_continuation)
0157 {
0158   impl->mutex_.lock();
0159   if (impl->locked_)
0160   {
0161     // Some other handler already holds the strand lock. Enqueue for later.
0162     impl->waiting_queue_.push(op);
0163     impl->mutex_.unlock();
0164   }
0165   else
0166   {
0167     // The handler is acquiring the strand lock and so is responsible for
0168     // scheduling the strand.
0169     impl->locked_ = true;
0170     impl->mutex_.unlock();
0171     impl->ready_queue_.push(op);
0172     io_context_impl_.post_immediate_completion(impl, is_continuation);
0173   }
0174 }
0175
0176 void strand_service::do_complete(void* owner, operation* base,
0177     const boost::system::error_code& ec, std::size_t /*bytes_transferred*/)
0178 {
0179   if (owner)
0180   {
0181     strand_impl* impl = static_cast<strand_impl*>(base);
0182
0183     // Indicate that this strand is executing on the current thread.
0184     call_stack<strand_impl>::context ctx(impl);
0185
0186     // Ensure the next handler, if any, is scheduled on block exit.
0187     on_do_complete_exit on_exit;
0188     on_exit.owner_ = static_cast<io_context_impl*>(owner);
0189     on_exit.impl_ = impl;
0190
0191     // Run all ready handlers. No lock is required since the ready queue is
0192     // accessed only within the strand.
0193     while (operation* o = impl->ready_queue_.front())
0194     {
0195       impl->ready_queue_.pop();
0196       o->complete(owner, ec, 0);
0197     }
0198   }
0199 }
0200
0201 } // namespace detail
0202 } // namespace asio
0203 } // namespace boost
0204
0205 #include <boost/asio/detail/pop_options.hpp>
0206
0207 #endif // BOOST_ASIO_DETAIL_IMPL_STRAND_SERVICE_IPP