redis/impl/writer_fsm.ipp

0001 //
0002 // Copyright (c) 2025 Marcelo Zimbres Silva (mzimbres@gmail.com),
0003 // Ruben Perez Hidalgo (rubenperez038 at gmail dot com)
0004 //
0005 // Distributed under the Boost Software License, Version 1.0. (See accompanying
0006 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0007 //
0008
0009 #ifndef BOOST_REDIS_WRITER_FSM_IPP
0010 #define BOOST_REDIS_WRITER_FSM_IPP
0011
0012 #include <boost/redis/adapter/any_adapter.hpp>
0013 #include <boost/redis/detail/connection_state.hpp>
0014 #include <boost/redis/detail/coroutine.hpp>
0015 #include <boost/redis/detail/multiplexer.hpp>
0016 #include <boost/redis/detail/writer_fsm.hpp>
0017 #include <boost/redis/impl/is_terminal_cancel.hpp>
0018 #include <boost/redis/impl/log_utils.hpp>
0019 #include <boost/redis/logger.hpp>
0020
0021 #include <boost/asio/cancellation_type.hpp>
0022 #include <boost/asio/error.hpp>
0023 #include <boost/assert.hpp>
0024 #include <boost/system/error_code.hpp>
0025
0026 #include <cstddef>
0027
0028 namespace boost::redis::detail {
0029
0030 inline void process_ping_node(
0031    buffered_logger& lgr,
0032    resp3::basic_node<std::string_view> const& nd,
0033    system::error_code& ec)
0034 {
0035    switch (nd.data_type) {
0036       case resp3::type::simple_error: ec = redis::error::resp3_simple_error; break;
0037       case resp3::type::blob_error:   ec = redis::error::resp3_blob_error; break;
0038       default:                        ;
0039    }
0040
0041    if (ec) {
0042       log_info(lgr, "Health checker: server answered ping with an error: ", nd.value);
0043    }
0044 }
0045
0046 inline any_adapter make_ping_adapter(buffered_logger& lgr)
0047 {
0048    return any_adapter{
0049       [&lgr](any_adapter::parse_event evt, resp3::node_view const& nd, system::error_code& ec) {
0050          if (evt == any_adapter::parse_event::node)
0051             process_ping_node(lgr, nd, ec);
0052       }};
0053 }
0054
0055 writer_action writer_fsm::resume(
0056    connection_state& st,
0057    system::error_code ec,
0058    std::size_t bytes_written,
0059    asio::cancellation_type_t cancel_state)
0060 {
0061    switch (resume_point_) {
0062       BOOST_REDIS_CORO_INITIAL
0063
0064       for (;;) {
0065          // Attempt to write while we have requests ready to send
0066          while (st.mpx.prepare_write() != 0u) {
0067             // Write an entire message. We can't use asio::async_write because we want
0068             // to apply timeouts to individual write operations
0069             for (;;) {
0070                // Write what we can. If nothing has been written for the health check
0071                // interval, we consider the connection as failed
0072                BOOST_REDIS_YIELD(
0073                   resume_point_,
0074                   1,
0075                   writer_action::write_some(st.cfg.health_check_interval))
0076
0077                // Commit the received bytes. This accounts for partial success
0078                bool finished = st.mpx.commit_write(bytes_written);
0079                log_debug(st.logger, "Writer task: ", bytes_written, " bytes written.");
0080
0081                // Check for cancellations and translate error codes
0082                if (is_terminal_cancel(cancel_state))
0083                   ec = asio::error::operation_aborted;
0084                else if (ec == asio::error::operation_aborted)
0085                   ec = error::write_timeout;
0086
0087                // Check for errors
0088                if (ec) {
0089                   if (ec == asio::error::operation_aborted) {
0090                      log_debug(st.logger, "Writer task: cancelled (1).");
0091                   } else {
0092                      log_debug(st.logger, "Writer task error: ", ec);
0093                   }
0094                   return ec;
0095                }
0096
0097                // Are we done yet?
0098                if (finished)
0099                   break;
0100             }
0101          }
0102
0103          // No more requests ready to be written. Wait for more, or until we need to send a PING
0104          BOOST_REDIS_YIELD(resume_point_, 2, writer_action::wait(st.cfg.health_check_interval))
0105
0106          // Check for cancellations
0107          if (is_terminal_cancel(cancel_state)) {
0108             log_debug(st.logger, "Writer task: cancelled (2).");
0109             return system::error_code(asio::error::operation_aborted);
0110          }
0111
0112          // If we weren't notified, it's because there is no data and we should send a health check
0113          if (!ec) {
0114             auto elem = make_elem(st.ping_req, make_ping_adapter(st.logger));
0115             elem->set_done_callback([] { });
0116             st.mpx.add(elem);
0117          }
0118       }
0119    }
0120
0121    // We should never reach here
0122    BOOST_ASSERT(false);
0123    return system::error_code();
0124 }
0125
0126 }  // namespace boost::redis::detail
0127
0128 #endif