redis/impl/run_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 #include <boost/redis/adapter/any_adapter.hpp>
0010 #include <boost/redis/config.hpp>
0011 #include <boost/redis/detail/connection_state.hpp>
0012 #include <boost/redis/detail/coroutine.hpp>
0013 #include <boost/redis/detail/multiplexer.hpp>
0014 #include <boost/redis/detail/run_fsm.hpp>
0015 #include <boost/redis/impl/is_terminal_cancel.hpp>
0016 #include <boost/redis/impl/log_utils.hpp>
0017 #include <boost/redis/impl/setup_request_utils.hpp>
0018
0019 #include <boost/asio/cancellation_type.hpp>
0020 #include <boost/asio/error.hpp>
0021 #include <boost/asio/local/basic_endpoint.hpp>  // for BOOST_ASIO_HAS_LOCAL_SOCKETS
0022 #include <boost/system/error_code.hpp>
0023
0024 namespace boost::redis::detail {
0025
0026 inline system::error_code check_config(const config& cfg)
0027 {
0028    if (!cfg.unix_socket.empty()) {
0029       if (cfg.use_ssl)
0030          return error::unix_sockets_ssl_unsupported;
0031 #ifndef BOOST_ASIO_HAS_LOCAL_SOCKETS
0032       return error::unix_sockets_unsupported;
0033 #endif
0034    }
0035    return system::error_code{};
0036 }
0037
0038 inline void compose_ping_request(const config& cfg, request& to)
0039 {
0040    to.clear();
0041    to.push("PING", cfg.health_check_id);
0042 }
0043
0044 inline void process_setup_node(
0045    connection_state& st,
0046    resp3::basic_node<std::string_view> const& nd,
0047    system::error_code& ec)
0048 {
0049    switch (nd.data_type) {
0050       case resp3::type::simple_error:
0051       case resp3::type::blob_error:
0052       case resp3::type::null:
0053          ec = redis::error::resp3_hello;
0054          st.setup_diagnostic = nd.value;
0055          break;
0056       default:;
0057    }
0058 }
0059
0060 inline any_adapter make_setup_adapter(connection_state& st)
0061 {
0062    return any_adapter{
0063       [&st](any_adapter::parse_event evt, resp3::node_view const& nd, system::error_code& ec) {
0064          if (evt == any_adapter::parse_event::node)
0065             process_setup_node(st, nd, ec);
0066       }};
0067 }
0068
0069 inline void on_setup_done(const multiplexer::elem& elm, connection_state& st)
0070 {
0071    const auto ec = elm.get_error();
0072    if (ec) {
0073       if (st.setup_diagnostic.empty()) {
0074          log_info(st.logger, "Setup request execution: ", ec);
0075       } else {
0076          log_info(st.logger, "Setup request execution: ", ec, " (", st.setup_diagnostic, ")");
0077       }
0078    } else {
0079       log_info(st.logger, "Setup request execution: success");
0080    }
0081 }
0082
0083 run_action run_fsm::resume(
0084    connection_state& st,
0085    system::error_code ec,
0086    asio::cancellation_type_t cancel_state)
0087 {
0088    switch (resume_point_) {
0089       BOOST_REDIS_CORO_INITIAL
0090
0091       // Check config
0092       ec = check_config(st.cfg);
0093       if (ec) {
0094          log_err(st.logger, "Invalid configuration: ", ec);
0095          stored_ec_ = ec;
0096          BOOST_REDIS_YIELD(resume_point_, 1, run_action_type::immediate)
0097          return stored_ec_;
0098       }
0099
0100       // Compose the setup request. This only depends on the config, so it can be done just once
0101       compose_setup_request(st.cfg);
0102
0103       // Compose the PING request. Same as above
0104       compose_ping_request(st.cfg, st.ping_req);
0105
0106       for (;;) {
0107          // Try to connect
0108          BOOST_REDIS_YIELD(resume_point_, 2, run_action_type::connect)
0109
0110          // Check for cancellations
0111          if (is_terminal_cancel(cancel_state)) {
0112             log_debug(st.logger, "Run: cancelled (1)");
0113             return system::error_code(asio::error::operation_aborted);
0114          }
0115
0116          // If we were successful, run all the connection tasks
0117          if (!ec) {
0118             // Initialization
0119             st.mpx.reset();
0120             st.setup_diagnostic.clear();
0121
0122             // Add the setup request to the multiplexer
0123             if (st.cfg.setup.get_commands() != 0u) {
0124                auto elm = make_elem(st.cfg.setup, make_setup_adapter(st));
0125                elm->set_done_callback([&elem_ref = *elm, &st] {
0126                   on_setup_done(elem_ref, st);
0127                });
0128                st.mpx.add(elm);
0129             }
0130
0131             // Run the tasks
0132             BOOST_REDIS_YIELD(resume_point_, 3, run_action_type::parallel_group)
0133
0134             // Store any error yielded by the tasks for later
0135             stored_ec_ = ec;
0136
0137             // We've lost connection or otherwise been cancelled.
0138             // Remove from the multiplexer the required requests.
0139             st.mpx.cancel_on_conn_lost();
0140
0141             // The receive operation must be cancelled because channel
0142             // subscription does not survive a reconnection but requires
0143             // re-subscription.
0144             BOOST_REDIS_YIELD(resume_point_, 4, run_action_type::cancel_receive)
0145
0146             // Restore the error
0147             ec = stored_ec_;
0148          }
0149
0150          // Check for cancellations
0151          if (is_terminal_cancel(cancel_state)) {
0152             log_debug(st.logger, "Run: cancelled (2)");
0153             return system::error_code(asio::error::operation_aborted);
0154          }
0155
0156          // If we are not going to try again, we're done
0157          if (st.cfg.reconnect_wait_interval.count() == 0) {
0158             return ec;
0159          }
0160
0161          // Wait for the reconnection interval
0162          BOOST_REDIS_YIELD(resume_point_, 5, run_action_type::wait_for_reconnection)
0163
0164          // Check for cancellations
0165          if (is_terminal_cancel(cancel_state)) {
0166             log_debug(st.logger, "Run: cancelled (3)");
0167             return system::error_code(asio::error::operation_aborted);
0168          }
0169       }
0170    }
0171
0172    // We should never get here
0173    BOOST_ASSERT(false);
0174    return system::error_code();
0175 }
0176
0177 }  // namespace boost::redis::detail