async_exec now uses a sans-io strategy #250
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| // | ||
| // Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]), | ||
| // Ruben Perez Hidalgo (rubenperez038 at gmail dot com) | ||
| // | ||
| // Distributed under the Boost Software License, Version 1.0. (See accompanying | ||
| // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) | ||
| // | ||
|
|
||
| #ifndef BOOST_REDIS_DETAIL_COROUTINE_HPP | ||
| #define BOOST_REDIS_DETAIL_COROUTINE_HPP | ||
|
|
||
| // asio::coroutine uses __COUNTER__ internally, which can trigger | ||
| // ODR violations if we use them in header-only code. These manifest as | ||
| // extremely hard-to-debug bugs only present in release builds. | ||
| // Use this instead when doing coroutines in non-template code. | ||
| // Adapted from Boost.MySQL. | ||
|
|
||
| // Coroutine state is represented as an integer (resume_point_var). | ||
| // Every yield gets assigned a unique value (resume_point_id). | ||
| // Yielding sets the next resume point, returns, and sets a case label for re-entering. | ||
| // Coroutines need to switch on resume_point_var to re-enter. | ||
|
|
||
| // Enclosing this in a scope allows placing the macro inside a brace-less for/while loop | ||
| // The empty scope after the case label is required because labels can't be at the end of a compound statement | ||
| #define BOOST_REDIS_YIELD(resume_point_var, resume_point_id, ...) \ | ||
| { \ | ||
| resume_point_var = resume_point_id; \ | ||
| return __VA_ARGS__; \ | ||
| case resume_point_id: \ | ||
| { \ | ||
| } \ | ||
| } | ||
|
|
||
| #define BOOST_REDIS_CORO_INITIAL case 0: | ||
|
|
||
| #endif |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,149 @@ | ||
| // | ||
| // Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]), | ||
| // Ruben Perez Hidalgo (rubenperez038 at gmail dot com) | ||
| // | ||
| // Distributed under the Boost Software License, Version 1.0. (See accompanying | ||
| // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) | ||
| // | ||
|
|
||
| #ifndef BOOST_REDIS_EXEC_FSM_HPP | ||
| #define BOOST_REDIS_EXEC_FSM_HPP | ||
|
|
||
| #include <boost/redis/detail/coroutine.hpp> | ||
| #include <boost/redis/detail/multiplexer.hpp> | ||
| #include <boost/redis/request.hpp> | ||
|
|
||
| #include <boost/asio/cancellation_type.hpp> | ||
| #include <boost/asio/error.hpp> | ||
| #include <boost/assert.hpp> | ||
| #include <boost/system/error_code.hpp> | ||
|
|
||
| #include <cstddef> | ||
| #include <memory> | ||
|
|
||
| // Sans-io algorithm for async_exec, as a finite state machine | ||
|
|
||
| namespace boost::redis::detail { | ||
|
|
||
| // What should we do next? | ||
| enum class exec_action_type | ||
| { | ||
| setup_cancellation, // Set up the cancellation types supported by the composed operation | ||
| immediate, // Invoke asio::async_immediate to avoid re-entrancy problems | ||
| done, // Call the final handler | ||
| write, // Notify the writer task | ||
| wait_for_response, // Wait to be notified | ||
| cancel_run, // Cancel the connection's run operation | ||
| }; | ||
|
|
||
| class exec_action { | ||
| exec_action_type type_; | ||
| system::error_code ec_; | ||
| std::size_t bytes_read_; | ||
|
|
||
| public: | ||
| exec_action(exec_action_type type) noexcept | ||
| : type_{type} | ||
| { } | ||
|
|
||
| exec_action(system::error_code ec, std::size_t bytes_read = 0u) noexcept | ||
| : type_{exec_action_type::done} | ||
| , ec_{ec} | ||
| , bytes_read_{bytes_read} | ||
| { } | ||
|
|
||
| exec_action_type type() const { return type_; } | ||
| system::error_code error() const { return ec_; } | ||
| std::size_t bytes_read() const { return bytes_read_; } | ||
| }; | ||
|
|
||
| class exec_fsm { | ||
| int resume_point_{0}; | ||
| multiplexer* mpx_{nullptr}; | ||
| std::shared_ptr<multiplexer::elem> elem_; | ||
|
|
||
| static bool is_cancellation(asio::cancellation_type_t type) | ||
| { | ||
| return !!( | ||
| type & (asio::cancellation_type_t::total | asio::cancellation_type_t::partial | | ||
| asio::cancellation_type_t::terminal)); | ||
| } | ||
|
|
||
| exec_action resume_impl(bool connection_is_open, asio::cancellation_type_t cancel_state) | ||
| { | ||
| switch (resume_point_) { | ||
| BOOST_REDIS_CORO_INITIAL | ||
|
|
||
| // Check whether the user wants to wait for the connection to | ||
| // be established. | ||
| if (elem_->get_request().get_config().cancel_if_not_connected && !connection_is_open) { | ||
| BOOST_REDIS_YIELD(resume_point_, 1, exec_action_type::immediate) | ||
| return system::error_code(error::not_connected); | ||
| } | ||
|
|
||
| // No more immediate errors. Set up the supported cancellation types. | ||
| // This is required to get partial and total cancellations. | ||
| // This is a potentially allocating operation, so do it as late as we can. | ||
| BOOST_REDIS_YIELD(resume_point_, 2, exec_action_type::setup_cancellation) | ||
|
|
||
| // Add the request to the multiplexer | ||
| mpx_->add(elem_); | ||
|
|
||
| // Notify the writer task that there is work to do. If the task is not | ||
| // listening (e.g. it's already writing or the connection is not healthy), | ||
| // this is a no-op. Since this is sync, no cancellation can happen here. | ||
| BOOST_REDIS_YIELD(resume_point_, 3, exec_action_type::write) | ||
|
|
||
| while (true) { | ||
| // Wait until we get notified. This will return once the request completes, | ||
| // or upon any kind of cancellation | ||
| BOOST_REDIS_YIELD(resume_point_, 4, exec_action_type::wait_for_response) | ||
|
|
||
| // If the request has completed (with error or not), we're done | ||
| if (elem_->is_done()) { | ||
| return exec_action{elem_->get_error(), elem_->get_read_size()}; | ||
| } | ||
|
|
||
| // If we're cancelled, try to remove the request from the queue. This will only | ||
| // succeed if the request is waiting (wasn't written yet) | ||
| if (is_cancellation(cancel_state) && mpx_->remove(elem_)) { | ||
| return exec_action{asio::error::operation_aborted}; | ||
| } | ||
|
|
||
| // If we hit a terminal cancellation, tear down the connection. | ||
| // Otherwise, go back to waiting. | ||
| // TODO: we could likely do better here and mark the request as cancelled, removing | ||
| // the done callback and the adapter. But this requires further exploration | ||
| if (!!(cancel_state & asio::cancellation_type_t::terminal)) { | ||
| BOOST_REDIS_YIELD(resume_point_, 5, exec_action_type::cancel_run) | ||
| return exec_action{asio::error::operation_aborted}; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // We should never get here | ||
| BOOST_ASSERT(false); | ||
| return exec_action{system::error_code()}; | ||
| } | ||
|
|
||
| public: | ||
| exec_fsm(multiplexer& mpx, std::shared_ptr<multiplexer::elem> elem) noexcept | ||
| : mpx_(&mpx) | ||
| , elem_(std::move(elem)) | ||
| { } | ||
|
|
||
| exec_action resume(bool connection_is_open, asio::cancellation_type_t cancel_state) | ||
| { | ||
| // When completing, we should deallocate any temporary storage we acquired | ||
| // for the operation before invoking the final handler. | ||
| // This intercepts the returned action to implement this. | ||
| auto act = resume_impl(connection_is_open, cancel_state); | ||
| if (act.type() == exec_action_type::done) | ||
| elem_.reset(); | ||
| return act; | ||
| } | ||
| }; | ||
|
|
||
| } // namespace boost::redis::detail | ||
|
|
||
| #endif // BOOST_REDIS_CONNECTOR_HPP | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think the naming here could be more descriptive
notify_writerfor example.