EIC code displayed by LXR master/​include/​boost/​mqtt5/​detail/​async_mutex.hpp	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2025-09-17 08:38:19

 //
 // Copyright (c) 2023-2025 Ivica Siladic, Bruno Iljazovic, Korina Simicevic
 //
 // Distributed under the Boost Software License, Version 1.0.
 // (See accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt)
 //
 
 #ifndef BOOST_MQTT5_ASYNC_MUTEX_HPP
 #define BOOST_MQTT5_ASYNC_MUTEX_HPP
 
 #include <boost/mqtt5/detail/async_traits.hpp>
 
 #include <boost/asio/any_completion_handler.hpp>
 #include <boost/asio/any_io_executor.hpp>
 #include <boost/asio/associated_allocator.hpp>
 #include <boost/asio/associated_cancellation_slot.hpp>
 #include <boost/asio/async_result.hpp>
 #include <boost/asio/bind_cancellation_slot.hpp>
 #include <boost/asio/execution.hpp>
 #include <boost/asio/require.hpp>
 #include <boost/system/error_code.hpp>
 
 #include <deque>
 
 namespace boost::mqtt5::detail {
 
 namespace asio = boost::asio;
 using error_code = boost::system::error_code;
 
 class async_mutex {
 public:
     using executor_type = asio::any_io_executor;
 private:
     using queued_op_t = asio::any_completion_handler<
         void (error_code)
     >;
     using queue_t = std::deque<queued_op_t>;
 
     // Handler with assigned tracking executor.
     // Objects of this type are type-erased by any_completion_handler
     // and stored in the waiting queue.
     template <typename Handler, typename Executor>
     class tracked_op {
         tracking_type<Handler, Executor> _executor;
         Handler _handler;
     public:
         tracked_op(Handler&& h, const Executor& ex) :
             _executor(tracking_executor(h, ex)), _handler(std::move(h))
         {}
 
         tracked_op(tracked_op&&) = default;
         tracked_op(const tracked_op&) = delete;
 
         tracked_op& operator=(tracked_op&&) = default;
         tracked_op& operator=(const tracked_op&) = delete;
 
         using allocator_type = asio::associated_allocator_t<Handler>;
         allocator_type get_allocator() const noexcept {
             return asio::get_associated_allocator(_handler);
         }
 
         using cancellation_slot_type =
             asio::associated_cancellation_slot_t<Handler>;
         cancellation_slot_type get_cancellation_slot() const noexcept {
             return asio::get_associated_cancellation_slot(_handler);
         }
 
         using executor_type = tracking_type<Handler, Executor>;
         executor_type get_executor() const noexcept {
             return _executor;
         }
 
         void operator()(error_code ec) {
             std::move(_handler)(ec);
         }
     };
 
     // Per-operation cancellation helper.
     // It is safe to emit the cancellation signal from any thread
     // provided there are no other concurrent calls to the async_mutex.
     // The helper stores queue iterator to operation since the iterator
     // would not be invalidated by other queue operations.
     class cancel_waiting_op {
         queue_t::iterator _ihandler;
     public:
         explicit cancel_waiting_op(queue_t::iterator ih) : _ihandler(ih) {}
 
         void operator()(asio::cancellation_type_t type) {
             if (type == asio::cancellation_type_t::none)
                 return;
             if (*_ihandler) {
                 auto h = std::move(*_ihandler);
                 auto ex = asio::get_associated_executor(h);
                 asio::require(ex, asio::execution::blocking.possibly)
                     .execute([h = std::move(h)]() mutable {
                         std::move(h)(asio::error::operation_aborted);
                     });
             }
         }
     };
 
     bool _locked { false };
     queue_t _waiting;
     executor_type _ex;
 
 public:
     template <typename Executor>
     explicit async_mutex(Executor&& ex) : _ex(std::forward<Executor>(ex)) {}
 
     async_mutex(const async_mutex&) = delete;
     async_mutex& operator=(const async_mutex&) = delete;
 
     ~async_mutex() {
         cancel();
     }
 
     const executor_type& get_executor() const noexcept {
         return _ex;
     }
 
     bool is_locked() const noexcept {
         return _locked;
     }
 
     // Schedules mutex for lock operation and return immediately.
     // Calls given completion handler when mutex is locked.
     // It's the responsibility of the completion handler to unlock the mutex.
     template <typename CompletionToken>
     decltype(auto) lock(CompletionToken&& token) noexcept {
         using Signature = void (error_code);
 
         auto initiation = [] (auto handler, async_mutex& self) {
             self.execute_or_queue(std::move(handler));
         };
 
         return asio::async_initiate<CompletionToken, Signature>(
             initiation, token, std::ref(*this)
         );
     }
 
     // Unlocks the mutex. The mutex must be in locked state.
     // Next queued operation, if any, will be executed in a manner
     // equivalent to asio::post.
     void unlock() {
         while (!_waiting.empty()) {
             auto op = std::move(_waiting.front());
             _waiting.pop_front();
             if (!op) continue;
             op.get_cancellation_slot().clear();
             execute_op(std::move(op));
             return;
         }
         _locked = false;
     }
 
     // Cancels all outstanding operations waiting on the mutex.
     void cancel() {
         while (!_waiting.empty()) {
             auto op = std::move(_waiting.front());
             _waiting.pop_front();
             if (!op) continue;
             op.get_cancellation_slot().clear();
             asio::require(_ex, asio::execution::blocking.never)
                 .execute([ex = _ex, op = std::move(op)]() mutable {
                     auto opex = asio::get_associated_executor(op, ex);
                     opex.execute(
                         [op = std::move(op)]() mutable {
                             op(asio::error::operation_aborted);
                         }
                     );
                 });
         }
     }
 
 private:
 
     // Schedule operation to `opex` executor using `_ex` executor.
     // The operation is equivalent to asio::post(_ex, op) but
     // for some reason this form of execution is much faster.
     void execute_op(queued_op_t op) {
         asio::require(_ex, asio::execution::blocking.never)
             .execute([ex = _ex, op = std::move(op)]() mutable {
                 auto opex = asio::get_associated_executor(op, ex);
                 opex.execute(
                     [op = std::move(op)]() mutable {
                         op(error_code {});
                     }
                 );
             });
     }
 
     // Executes operation immediately if mutex is not locked
     // or queues it for later execution otherwise. In both cases
     // the operation will be executed in a manner equivalent
     // to asio::post to avoid recursion.
     template <typename Handler>
     void execute_or_queue(Handler&& handler) noexcept {
         tracked_op h { std::move(handler), _ex };
         if (_locked) {
             _waiting.emplace_back(std::move(h));
             auto slot = _waiting.back().get_cancellation_slot();
             if (slot.is_connected())
                 slot.template emplace<cancel_waiting_op>(
                     _waiting.end() - 1
                 );
         }
         else {
             _locked = true;
             execute_op(queued_op_t { std::move(h) });
         }
     }
 };
 
 } // end namespace boost::mqtt5::detail
 
 #endif // !BOOST_MQTT5_ASYNC_MUTEX_HPP

This page was automatically generated by the 2.3.7 LXR engine. The LXR team