redis/impl/reader_fsm.ipp

0001 /* Copyright (c) 2018-2025 Marcelo Zimbres Silva (mzimbres@gmail.com)
0002  *
0003  * Distributed under the Boost Software License, Version 1.0. (See
0004  * accompanying file LICENSE.txt)
0005  */
0006
0007 #include <boost/redis/detail/connection_state.hpp>
0008 #include <boost/redis/detail/coroutine.hpp>
0009 #include <boost/redis/detail/multiplexer.hpp>
0010 #include <boost/redis/detail/reader_fsm.hpp>
0011 #include <boost/redis/impl/is_terminal_cancel.hpp>
0012 #include <boost/redis/impl/log_utils.hpp>
0013
0014 #include <boost/asio/cancellation_type.hpp>
0015 #include <boost/asio/error.hpp>
0016
0017 namespace boost::redis::detail {
0018
0019 reader_fsm::action reader_fsm::resume(
0020    connection_state& st,
0021    std::size_t bytes_read,
0022    system::error_code ec,
0023    asio::cancellation_type_t cancel_state)
0024 {
0025    switch (resume_point_) {
0026       BOOST_REDIS_CORO_INITIAL
0027
0028       for (;;) {
0029          // Prepare the buffer for the read operation
0030          ec = st.mpx.prepare_read();
0031          if (ec) {
0032             log_debug(st.logger, "Reader task: error in prepare_read: ", ec);
0033             return {ec};
0034          }
0035
0036          // Read. The connection might spend health_check_interval without writing data.
0037          // Give it another health_check_interval for the response to arrive.
0038          // If we don't get anything in this time, consider the connection as dead
0039          log_debug(st.logger, "Reader task: issuing read");
0040          BOOST_REDIS_YIELD(resume_point_, 1, action::read_some(2 * st.cfg.health_check_interval))
0041
0042          // Check for cancellations
0043          if (is_terminal_cancel(cancel_state)) {
0044             log_debug(st.logger, "Reader task: cancelled (1)");
0045             return system::error_code(asio::error::operation_aborted);
0046          }
0047
0048          // Translate timeout errors caused by operation_aborted to more legible ones.
0049          // A timeout here means that we didn't receive data in time.
0050          // Note that cancellation is already handled by the above statement.
0051          if (ec == asio::error::operation_aborted) {
0052             ec = error::pong_timeout;
0053          }
0054
0055          // Log what we read
0056          if (ec) {
0057             log_debug(st.logger, "Reader task: ", bytes_read, " bytes read, error: ", ec);
0058          } else {
0059             log_debug(st.logger, "Reader task: ", bytes_read, " bytes read");
0060          }
0061
0062          // Process the bytes read, even if there was an error
0063          st.mpx.commit_read(bytes_read);
0064
0065          // Check for read errors
0066          if (ec) {
0067             // TODO: If an error occurred but data was read (i.e.
0068             // bytes_read != 0) we should try to process that data and
0069             // deliver it to the user before calling cancel_run.
0070             return ec;
0071          }
0072
0073          // Process the data that we've read
0074          while (st.mpx.get_read_buffer_size() != 0) {
0075             res_ = st.mpx.consume(ec);
0076
0077             if (ec) {
0078                // TODO: Perhaps log what has not been consumed to aid
0079                // debugging.
0080                log_debug(st.logger, "Reader task: error processing message: ", ec);
0081                return ec;
0082             }
0083
0084             if (res_.first == consume_result::needs_more) {
0085                log_debug(st.logger, "Reader task: incomplete message received");
0086                break;
0087             }
0088
0089             if (res_.first == consume_result::got_push) {
0090                BOOST_REDIS_YIELD(resume_point_, 2, action::notify_push_receiver(res_.second))
0091                // Check for cancellations
0092                if (is_terminal_cancel(cancel_state)) {
0093                   log_debug(st.logger, "Reader task: cancelled (2)");
0094                   return system::error_code(asio::error::operation_aborted);
0095                }
0096
0097                // Check for other errors
0098                if (ec) {
0099                   log_debug(st.logger, "Reader task: error notifying push receiver: ", ec);
0100                   return ec;
0101                }
0102             } else {
0103                // TODO: Here we should notify the exec operation that
0104                // it can be completed. This will improve log clarity
0105                // and will make this code symmetrical in how it
0106                // handles pushes and other messages. The new action
0107                // type can be named notify_exec. To do that we need to
0108                // refactor the multiplexer.
0109             }
0110          }
0111       }
0112    }
0113
0114    BOOST_ASSERT(false);
0115    return system::error_code();
0116 }
0117
0118 }  // namespace boost::redis::detail