internal/sansio/top_level_algo.hpp

0001 //
0002 // Copyright (c) 2019-2025 Ruben Perez Hidalgo (rubenperez038 at gmail dot com)
0003 //
0004 // Distributed under the Boost Software License, Version 1.0. (See accompanying
0005 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0006 //
0007
0008 #ifndef BOOST_MYSQL_IMPL_INTERNAL_SANSIO_TOP_LEVEL_ALGO_HPP
0009 #define BOOST_MYSQL_IMPL_INTERNAL_SANSIO_TOP_LEVEL_ALGO_HPP
0010
0011 #include <boost/mysql/client_errc.hpp>
0012 #include <boost/mysql/diagnostics.hpp>
0013 #include <boost/mysql/error_code.hpp>
0014
0015 #include <boost/mysql/detail/next_action.hpp>
0016
0017 #include <boost/mysql/impl/internal/coroutine.hpp>
0018 #include <boost/mysql/impl/internal/sansio/connection_state_data.hpp>
0019 #include <boost/mysql/impl/internal/sansio/run_pipeline.hpp>
0020
0021 #include <boost/core/span.hpp>
0022
0023 #include <cstddef>
0024 #include <cstdint>
0025
0026 #ifdef BOOST_USE_VALGRIND
0027 #include <valgrind/memcheck.h>
0028 #endif
0029
0030 namespace boost {
0031 namespace mysql {
0032 namespace detail {
0033
0034 // Valgrind
0035 #ifdef BOOST_USE_VALGRIND
0036 inline void valgrind_make_mem_defined(const void* data, std::size_t size)
0037 {
0038     VALGRIND_MAKE_MEM_DEFINED(data, size);
0039 }
0040 #else
0041 inline void valgrind_make_mem_defined(const void*, std::size_t) noexcept {}
0042 #endif
0043
0044 // InnerAlgo should have
0045 //   Constructible from the args forwarded by the ctor.
0046 //   next_action resume(connection_state_data&, error_code);
0047 //   AlgoParams::result_type result(const connection_state_data&) const; // if AlgoParams::result_type != void
0048 template <class InnerAlgo>
0049 class top_level_algo
0050 {
0051     int resume_point_{0};
0052     connection_state_data* st_;
0053     diagnostics* diag_;
0054     InnerAlgo algo_;
0055     span<const std::uint8_t> bytes_to_write_;
0056
0057 public:
0058     template <class... Args>
0059     top_level_algo(connection_state_data& st, diagnostics& diag, Args&&... args)
0060         : st_(&st), diag_(&diag), algo_(std::forward<Args>(args)...)
0061     {
0062     }
0063
0064     const InnerAlgo& inner_algo() const { return algo_; }
0065
0066     next_action resume(error_code ec, std::size_t bytes_transferred)
0067     {
0068         next_action act;
0069
0070         switch (resume_point_)
0071         {
0072         case 0:
0073             // We shouldn't be running another operation if we get here
0074             // (caught by setup).
0075             // Record that we're running an operation
0076             BOOST_ASSERT(!st_->op_in_progress);
0077             st_->op_in_progress = true;
0078
0079             // Run until completion
0080             while (true)
0081             {
0082                 // Run the op
0083                 act = algo_.resume(*st_, *diag_, ec);
0084
0085                 // Check next action
0086                 if (act.is_done())
0087                 {
0088                     // Record that we're no longer running an operation
0089                     st_->op_in_progress = false;
0090
0091                     // Done
0092                     return act;
0093                 }
0094                 else if (act.type() == next_action_type::read)
0095                 {
0096                     // Read until a complete message is received
0097                     // (may be zero times if cached)
0098                     while (!st_->reader.done() && !ec)
0099                     {
0100                         ec = st_->reader.prepare_buffer();
0101                         if (ec)
0102                             break;
0103                         BOOST_MYSQL_YIELD(
0104                             resume_point_,
0105                             1,
0106                             next_action::read({st_->reader.buffer(), st_->tls_active})
0107                         )
0108                         valgrind_make_mem_defined(st_->reader.buffer().data(), bytes_transferred);
0109                         st_->reader.resume(bytes_transferred);
0110                     }
0111
0112                     // Check for errors
0113                     if (!ec)
0114                         ec = st_->reader.error();
0115
0116                     // We've got a message, continue
0117                 }
0118                 else if (act.type() == next_action_type::write)
0119                 {
0120                     // Write until a complete message was written
0121                     bytes_to_write_ = act.write_args().buffer;
0122
0123                     while (!bytes_to_write_.empty() && !ec)
0124                     {
0125                         BOOST_MYSQL_YIELD(
0126                             resume_point_,
0127                             2,
0128                             next_action::write({bytes_to_write_, st_->tls_active})
0129                         )
0130                         bytes_to_write_ = bytes_to_write_.subspan(bytes_transferred);
0131                     }
0132
0133                     // We fully wrote a message, continue
0134                 }
0135                 else
0136                 {
0137                     // Other ops always require I/O
0138                     BOOST_MYSQL_YIELD(resume_point_, 3, act)
0139                 }
0140             }
0141         }
0142
0143         return next_action();
0144     }
0145 };
0146
0147 }  // namespace detail
0148 }  // namespace mysql
0149 }  // namespace boost
0150
0151 #endif