update boost

linx64/include/boost/mysql/impl/internal/channel/channel.hpp

@@ -0,0 +1,150 @@
+//
+// Copyright (c) 2019-2023 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_MYSQL_IMPL_INTERNAL_CHANNEL_CHANNEL_HPP
+#define BOOST_MYSQL_IMPL_INTERNAL_CHANNEL_CHANNEL_HPP
+
+#include <boost/mysql/diagnostics.hpp>
+#include <boost/mysql/error_code.hpp>
+#include <boost/mysql/field_view.hpp>
+#include <boost/mysql/metadata_mode.hpp>
+
+#include <boost/mysql/detail/any_stream.hpp>
+
+#include <boost/mysql/impl/internal/channel/message_reader.hpp>
+#include <boost/mysql/impl/internal/channel/message_writer.hpp>
+#include <boost/mysql/impl/internal/channel/write_message.hpp>
+#include <boost/mysql/impl/internal/protocol/capabilities.hpp>
+#include <boost/mysql/impl/internal/protocol/db_flavor.hpp>
+
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/async_result.hpp>
+
+#include <cstddef>
+#include <memory>
+#include <utility>
+#include <vector>
+
+namespace boost {
+namespace mysql {
+namespace detail {
+
+// Implements the message layer of the MySQL protocol
+class channel
+{
+    db_flavor flavor_{db_flavor::mysql};
+    capabilities current_caps_;
+    std::uint8_t shared_sequence_number_{};
+    diagnostics shared_diag_;  // for async ops
+    std::vector<field_view> shared_fields_;
+    metadata_mode meta_mode_{metadata_mode::minimal};
+    message_reader reader_;
+    message_writer writer_;
+    std::unique_ptr<any_stream> stream_;
+
+public:
+    channel(std::size_t read_buffer_size, std::unique_ptr<any_stream> stream)
+        : reader_(read_buffer_size), stream_(std::move(stream))
+    {
+    }
+
+    // Executor
+    using executor_type = asio::any_io_executor;
+    executor_type get_executor() { return stream_->get_executor(); }
+
+    // Reading
+    bool has_read_messages() const noexcept { return reader_.has_message(); }
+
+    span<const std::uint8_t> next_read_message(std::uint8_t& seqnum, error_code& err) noexcept
+    {
+        return reader_.get_next_message(seqnum, err);
+    }
+
+    void read_some(error_code& code) { read_some_messages(*stream_, reader_, code); }
+
+    template <BOOST_ASIO_COMPLETION_TOKEN_FOR(void(error_code)) CompletionToken>
+    BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void(error_code))
+    async_read_some(CompletionToken&& token)
+    {
+        return async_read_some_messages(*stream_, reader_, std::forward<CompletionToken>(token));
+    }
+
+    span<const std::uint8_t> read_one(std::uint8_t& seqnum, error_code& ec)
+    {
+        return read_one_message(*stream_, reader_, seqnum, ec);
+    }
+
+    template <BOOST_ASIO_COMPLETION_TOKEN_FOR(void(error_code, span<const std::uint8_t>)) CompletionToken>
+    BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void(error_code, span<const std::uint8_t>))
+    async_read_one(std::uint8_t& seqnum, CompletionToken&& token)
+    {
+        return async_read_one_message(*stream_, reader_, seqnum, std::forward<CompletionToken>(token));
+    }
+
+    // Exposed for the sake of testing
+    std::size_t read_buffer_size() const noexcept { return reader_.buffer().size(); }
+
+    // Writing. serialize() gets all the required data into the write buffers so it can be written
+    template <class Serializable>
+    void serialize(const Serializable& message, std::uint8_t& sequence_number)
+    {
+        std::size_t size = message.get_size();
+        auto buff = writer_.prepare_buffer(size, sequence_number);
+        message.serialize(buff);
+    }
+
+    // Writes what has been set up by serialize()
+    void write(error_code& code) { write_message(*stream_, writer_, code); }
+
+    template <BOOST_ASIO_COMPLETION_TOKEN_FOR(void(error_code)) CompletionToken>
+    BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void(error_code))
+    async_write(CompletionToken&& token)
+    {
+        return async_write_message(*stream_, writer_, std::forward<CompletionToken>(token));
+    }
+
+    // Capabilities
+    capabilities current_capabilities() const noexcept { return current_caps_; }
+    void set_current_capabilities(capabilities value) noexcept { current_caps_ = value; }
+
+    // DB flavor
+    db_flavor flavor() const noexcept { return flavor_; }
+    void set_flavor(db_flavor v) noexcept { flavor_ = v; }
+
+    void reset()
+    {
+        flavor_ = db_flavor::mysql;
+        current_caps_ = capabilities();
+        reset_sequence_number();
+        stream_->reset_ssl_active();
+        // Metadata mode does not get reset on handshake
+    }
+
+    // Internal buffer, diagnostics and sequence_number to help async ops
+    diagnostics& shared_diag() noexcept { return shared_diag_; }
+    std::uint8_t& shared_sequence_number() noexcept { return shared_sequence_number_; }
+    std::uint8_t& reset_sequence_number() noexcept { return shared_sequence_number_ = 0; }
+    std::vector<field_view>& shared_fields() noexcept { return shared_fields_; }
+    const std::vector<field_view>& shared_fields() const noexcept { return shared_fields_; }
+
+    // Metadata mode
+    metadata_mode meta_mode() const noexcept { return meta_mode_; }
+    void set_meta_mode(metadata_mode v) noexcept { meta_mode_ = v; }
+
+    // SSL
+    bool ssl_active() const noexcept { return stream_->ssl_active(); }
+
+    // Getting the underlying stream
+    any_stream& stream() noexcept { return *stream_; }
+    const any_stream& stream() const noexcept { return *stream_; }
+};
+
+}  // namespace detail
+}  // namespace mysql
+}  // namespace boost
+
+#endif

linx64/include/boost/mysql/impl/internal/channel/message_parser.hpp

@@ -0,0 +1,92 @@
+//
+// Copyright (c) 2019-2023 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_MYSQL_IMPL_INTERNAL_CHANNEL_MESSAGE_PARSER_HPP
+#define BOOST_MYSQL_IMPL_INTERNAL_CHANNEL_MESSAGE_PARSER_HPP
+
+#include <boost/mysql/detail/config.hpp>
+
+#include <boost/mysql/impl/internal/channel/read_buffer.hpp>
+#include <boost/mysql/impl/internal/protocol/constants.hpp>
+
+#include <cstddef>
+#include <cstdint>
+
+namespace boost {
+namespace mysql {
+namespace detail {
+
+class message_parser
+{
+    struct state_t
+    {
+        bool is_first_frame{true};
+        std::uint8_t seqnum_first{};
+        std::uint8_t seqnum_last{};
+        bool reading_header{true};
+        std::size_t remaining_bytes{0};
+        bool more_frames_follow{false};
+        bool has_seqnum_mismatch{false};
+    };
+
+    std::size_t max_frame_size_;
+    state_t state_;
+
+public:
+    struct result
+    {
+        // whether it has a message or not
+        bool has_message{false};
+
+        // if !has_message, number of bytes required to parse the current message
+        std::size_t required_size{0};
+
+        // if has_message, the actual parsed message
+        struct message_t
+        {
+            std::uint8_t seqnum_first;
+            std::uint8_t seqnum_last;
+            std::size_t size;
+            bool has_seqnum_mismatch;  // for multi-frame messages, set to true if an error mismatch
+                                       // happened
+        } message{};
+
+        void set_required_size(std::size_t size) noexcept
+        {
+            has_message = false;
+            required_size = size;
+        }
+
+        void set_message(const message_t& msg) noexcept
+        {
+            has_message = true;
+            message = msg;
+        }
+    };
+
+    // max_frame_size is configurable so tests run faster
+    message_parser(std::size_t max_frame_size = MAX_PACKET_SIZE) noexcept : max_frame_size_(max_frame_size){};
+
+    // Attempts to process a message from buff and puts it into msg.
+    // If a message is read, res.has_message == true, and res.message will be populated.
+    // Otherwise, res.required_size will contain
+    // the number of bytes needed to complete the message part we're parsing.
+    // Doesn't cause buffer reallocations, and doesn't change the contents
+    // of buff's reserved area.
+    BOOST_MYSQL_DECL
+    void parse_message(read_buffer& buff, result& res) noexcept;
+};
+
+}  // namespace detail
+}  // namespace mysql
+}  // namespace boost
+
+#ifdef BOOST_MYSQL_HEADER_ONLY
+#include <boost/mysql/impl/internal/channel/message_parser.ipp>
+#endif
+
+#endif

linx64/include/boost/mysql/impl/internal/channel/message_parser.ipp

@@ -0,0 +1,110 @@
+//
+// Copyright (c) 2019-2023 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_MYSQL_IMPL_INTERNAL_CHANNEL_MESSAGE_PARSER_IPP
+#define BOOST_MYSQL_IMPL_INTERNAL_CHANNEL_MESSAGE_PARSER_IPP
+
+#pragma once
+
+#include <boost/mysql/impl/internal/channel/message_parser.hpp>
+#include <boost/mysql/impl/internal/protocol/protocol.hpp>
+
+void boost::mysql::detail::message_parser::parse_message(read_buffer& buff, result& res) noexcept
+{
+    while (true)
+    {
+        if (state_.reading_header)
+        {
+            // If there are not enough bytes to process a header, request more
+            if (buff.pending_size() < HEADER_SIZE)
+            {
+                res.set_required_size(HEADER_SIZE - buff.pending_size());
+                return;
+            }
+
+            // Mark the header as belonging to the current message
+            buff.move_to_current_message(HEADER_SIZE);
+
+            // Deserialize the header
+            auto header = deserialize_frame_header(
+                span<const std::uint8_t, frame_header_size>(buff.pending_first() - HEADER_SIZE, HEADER_SIZE)
+            );
+
+            // Process the sequence number
+            if (state_.is_first_frame)
+            {
+                state_.seqnum_first = header.sequence_number;
+                state_.seqnum_last = header.sequence_number;
+            }
+            else
+            {
+                std::uint8_t expected_seqnum = state_.seqnum_last + 1;
+                if (header.sequence_number != expected_seqnum)
+                {
+                    state_.has_seqnum_mismatch = true;
+                }
+                state_.seqnum_last = expected_seqnum;
+            }
+
+            // Process the packet size
+            state_.remaining_bytes = header.size;
+            state_.more_frames_follow = (state_.remaining_bytes == max_frame_size_);
+
+            // We are done with the header
+            if (state_.is_first_frame)
+            {
+                // If it's the 1st frame, we can just move the header bytes to the reserved
+                // area, avoiding a big memmove
+                buff.move_to_reserved(HEADER_SIZE);
+            }
+            else
+            {
+                buff.remove_current_message_last(HEADER_SIZE);
+            }
+            state_.is_first_frame = false;
+            state_.reading_header = false;
+        }
+
+        if (!state_.reading_header)
+        {
+            // Get the number of bytes belonging to this message
+            std::size_t new_bytes = (std::min)(buff.pending_size(), state_.remaining_bytes);
+
+            // Mark them as belonging to the current message in the buffer
+            buff.move_to_current_message(new_bytes);
+
+            // Update remaining bytes
+            state_.remaining_bytes -= new_bytes;
+            if (state_.remaining_bytes == 0)
+            {
+                state_.reading_header = true;
+            }
+            else
+            {
+                res.set_required_size(state_.remaining_bytes);
+                return;
+            }
+
+            // If we've fully read a message, we're done
+            if (!state_.remaining_bytes && !state_.more_frames_follow)
+            {
+                std::size_t message_size = buff.current_message_size();
+                buff.move_to_reserved(message_size);
+                res.set_message({
+                    state_.seqnum_first,
+                    state_.seqnum_last,
+                    message_size,
+                    state_.has_seqnum_mismatch,
+                });
+                state_ = state_t();
+                return;
+            }
+        }
+    }
+}
+
+#endif

linx64/include/boost/mysql/impl/internal/channel/message_reader.hpp

@@ -0,0 +1,282 @@
+//
+// Copyright (c) 2019-2023 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_MYSQL_IMPL_INTERNAL_CHANNEL_MESSAGE_READER_HPP
+#define BOOST_MYSQL_IMPL_INTERNAL_CHANNEL_MESSAGE_READER_HPP
+
+#include <boost/mysql/client_errc.hpp>
+#include <boost/mysql/error_code.hpp>
+
+#include <boost/mysql/detail/any_stream.hpp>
+
+#include <boost/mysql/impl/internal/channel/message_parser.hpp>
+#include <boost/mysql/impl/internal/channel/read_buffer.hpp>
+#include <boost/mysql/impl/internal/channel/valgrind.hpp>
+#include <boost/mysql/impl/internal/protocol/constants.hpp>
+
+#include <boost/asio/async_result.hpp>
+#include <boost/asio/buffer.hpp>
+#include <boost/asio/compose.hpp>
+#include <boost/asio/coroutine.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/assert.hpp>
+
+#include <cstddef>
+#include <cstdint>
+
+namespace boost {
+namespace mysql {
+namespace detail {
+
+class message_reader
+{
+public:
+    message_reader(std::size_t initial_buffer_size, std::size_t max_frame_size = MAX_PACKET_SIZE)
+        : buffer_(initial_buffer_size), parser_(max_frame_size)
+    {
+    }
+
+    bool has_message() const noexcept { return result_.has_message; }
+
+    span<const std::uint8_t> get_next_message(std::uint8_t& seqnum, error_code& ec) noexcept
+    {
+        {
+            BOOST_ASSERT(has_message());
+            if (result_.message.has_seqnum_mismatch || seqnum != result_.message.seqnum_first)
+            {
+                ec = make_error_code(client_errc::sequence_number_mismatch);
+                return {};
+            }
+            seqnum = result_.message.seqnum_last + 1;
+            span<const std::uint8_t> res(
+                buffer_.current_message_first() - result_.message.size,
+                result_.message.size
+            );
+            parse_message();
+            ec = error_code();
+            return res;
+        }
+    }
+
+    // Reads some messages from stream, until there is at least one
+    // or an error happens. On success, has_message() returns true
+    // and get_next_message() returns the parsed message.
+    // May relocate the buffer, modifying buffer_first().
+    // The reserved area bytes will be removed before the actual read.
+    void read_some(any_stream& stream, error_code& ec)
+    {
+        // If we already have a message, complete immediately
+        if (has_message())
+        {
+            ec = error_code();
+            return;
+        }
+
+        // Remove processed messages
+        buffer_.remove_reserved();
+
+        while (!has_message())
+        {
+            // If any previous process_message indicated that we need more
+            // buffer space, resize the buffer now
+            maybe_resize_buffer();
+
+            // Actually read bytes
+            std::size_t bytes_read = stream.read_some(free_area(), ec);
+            if (ec)
+                break;
+            valgrind_make_mem_defined(buffer_.free_first(), bytes_read);
+
+            // Process them
+            on_read_bytes(bytes_read);
+        }
+    }
+
+    template <BOOST_ASIO_COMPLETION_TOKEN_FOR(void(::boost::mysql::error_code)) CompletionToken>
+    BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void(error_code))
+    async_read_some(any_stream& stream, CompletionToken&& token);
+
+    // Exposed for the sake of testing
+    read_buffer& buffer() noexcept { return buffer_; }
+    const read_buffer& buffer() const noexcept { return buffer_; }
+
+private:
+    struct read_some_op;
+
+    read_buffer buffer_;
+    message_parser parser_;
+    message_parser::result result_;
+
+    void parse_message() { parser_.parse_message(buffer_, result_); }
+
+    void maybe_resize_buffer()
+    {
+        if (!result_.has_message)
+        {
+            buffer_.grow_to_fit(result_.required_size);
+        }
+    }
+
+    void on_read_bytes(size_t num_bytes)
+    {
+        buffer_.move_to_pending(num_bytes);
+        parse_message();
+    }
+
+    asio::mutable_buffer free_area() noexcept
+    {
+        auto res = buffer_.free_area();
+        return asio::mutable_buffer(res.data(), res.size());
+    }
+};
+
+struct boost::mysql::detail::message_reader::read_some_op : boost::asio::coroutine
+{
+    message_reader& reader_;
+    any_stream& stream_;
+
+    read_some_op(message_reader& reader, any_stream& stream) noexcept : reader_(reader), stream_(stream) {}
+
+    template <class Self>
+    void operator()(Self& self, error_code ec = {}, std::size_t bytes_read = 0)
+    {
+        // Error handling
+        if (ec)
+        {
+            self.complete(ec);
+            return;
+        }
+
+        // Non-error path
+        BOOST_ASIO_CORO_REENTER(*this)
+        {
+            // If we already have a message, complete immediately
+            if (reader_.has_message())
+            {
+                BOOST_ASIO_CORO_YIELD boost::asio::post(stream_.get_executor(), std::move(self));
+                self.complete(error_code());
+                BOOST_ASIO_CORO_YIELD break;
+            }
+
+            // Remove processed messages
+            reader_.buffer_.remove_reserved();
+
+            while (!reader_.has_message())
+            {
+                // If any previous process_message indicated that we need more
+                // buffer space, resize the buffer now
+                reader_.maybe_resize_buffer();
+
+                // Actually read bytes
+                BOOST_ASIO_CORO_YIELD stream_.async_read_some(reader_.free_area(), std::move(self));
+                valgrind_make_mem_defined(reader_.buffer_.free_first(), bytes_read);
+
+                // Process them
+                reader_.on_read_bytes(bytes_read);
+            }
+
+            self.complete(error_code());
+        }
+    }
+};
+
+// Public interface
+inline void read_some_messages(any_stream& stream, message_reader& reader, error_code& ec)
+{
+    return reader.read_some(stream, ec);
+}
+
+template <class CompletionToken>
+BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void(error_code))
+async_read_some_messages(any_stream& stream, message_reader& reader, CompletionToken&& token)
+{
+    return reader.async_read_some(stream, std::forward<CompletionToken>(token));
+}
+
+// Equivalent to read_some + get_next_message
+inline span<const std::uint8_t> read_one_message(
+    any_stream& stream,
+    message_reader& reader,
+    std::uint8_t& seqnum,
+    error_code& ec
+)
+{
+    read_some_messages(stream, reader, ec);
+    if (ec)
+        return {};
+    else
+        return reader.get_next_message(seqnum, ec);
+}
+
+struct read_one_message_op : boost::asio::coroutine
+{
+    message_reader& reader_;
+    any_stream& stream_;
+    std::uint8_t& seqnum_;
+
+    read_one_message_op(message_reader& reader, any_stream& stream, std::uint8_t& seqnum)
+        : reader_(reader), stream_(stream), seqnum_(seqnum)
+    {
+    }
+
+    template <class Self>
+    void operator()(Self& self, error_code code = {})
+    {
+        // Error handling
+        if (code)
+        {
+            self.complete(code, span<const std::uint8_t>());
+            return;
+        }
+
+        // Non-error path
+        BOOST_ASIO_CORO_REENTER(*this)
+        {
+            BOOST_ASIO_CORO_YIELD reader_.async_read_some(stream_, std::move(self));
+            {
+                auto b = reader_.get_next_message(seqnum_, code);
+                self.complete(code, b);
+            }
+        }
+    }
+};
+
+template <class CompletionToken>
+BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(
+    CompletionToken,
+    void(boost::mysql::error_code, ::boost::span<const std::uint8_t>)
+)
+async_read_one_message(
+    any_stream& stream,
+    message_reader& reader,
+    std::uint8_t& seqnum,
+    CompletionToken&& token
+)
+{
+    return boost::asio::async_compose<CompletionToken, void(error_code, span<const std::uint8_t>)>(
+        read_one_message_op(reader, stream, seqnum),
+        token,
+        stream
+    );
+}
+
+}  // namespace detail
+}  // namespace mysql
+}  // namespace boost
+
+template <BOOST_ASIO_COMPLETION_TOKEN_FOR(void(::boost::mysql::error_code)) CompletionToken>
+BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void(::boost::mysql::error_code))
+boost::mysql::detail::message_reader::async_read_some(any_stream& stream, CompletionToken&& token)
+{
+    return boost::asio::async_compose<CompletionToken, void(error_code)>(
+        read_some_op{*this, stream},
+        token,
+        stream
+    );
+}
+
+#endif /* INCLUDE_BOOST_MYSQL_DETAIL_AUXILIAR_STATIC_STRING_HPP_ */

linx64/include/boost/mysql/impl/internal/channel/message_writer.hpp

@@ -0,0 +1,142 @@
+//
+// Copyright (c) 2019-2023 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_MYSQL_IMPL_INTERNAL_CHANNEL_MESSAGE_WRITER_HPP
+#define BOOST_MYSQL_IMPL_INTERNAL_CHANNEL_MESSAGE_WRITER_HPP
+
+#include <boost/mysql/impl/internal/protocol/constants.hpp>
+#include <boost/mysql/impl/internal/protocol/protocol.hpp>
+
+#include <array>
+#include <cstddef>
+#include <cstdint>
+
+namespace boost {
+namespace mysql {
+namespace detail {
+
+class chunk_processor
+{
+    std::size_t first_{};
+    std::size_t last_{};
+
+    std::size_t remaining() const noexcept { return last_ - first_; }
+
+public:
+    chunk_processor() = default;
+    void reset() noexcept { reset(0, 0); }
+    void reset(std::size_t first, std::size_t last) noexcept
+    {
+        BOOST_ASSERT(last >= first);
+        first_ = first;
+        last_ = last;
+    }
+    void on_bytes_written(std::size_t n) noexcept
+    {
+        BOOST_ASSERT(remaining() >= n);
+        first_ += n;
+    }
+    bool done() const noexcept { return first_ == last_; }
+    span<const std::uint8_t> get_chunk(const std::vector<std::uint8_t>& buff) const noexcept
+    {
+        BOOST_ASSERT(buff.size() >= last_);
+        return {buff.data() + first_, remaining()};
+    }
+};
+
+class message_writer
+{
+    std::vector<std::uint8_t> buffer_;
+    std::size_t max_frame_size_;
+    std::uint8_t* seqnum_{nullptr};
+
+    chunk_processor chunk_;
+    std::size_t total_bytes_{};
+    std::size_t total_bytes_written_{};
+    bool should_send_empty_frame_{};
+
+    void process_header_write(std::uint32_t size_to_write, std::uint8_t seqnum, std::size_t buff_offset)
+    {
+        serialize_frame_header(
+            frame_header{size_to_write, seqnum},
+            span<std::uint8_t, frame_header_size>(buffer_.data() + buff_offset, frame_header_size)
+        );
+    }
+
+    std::uint8_t next_seqnum() noexcept { return (*seqnum_)++; }
+
+    void prepare_next_chunk()
+    {
+        if (should_send_empty_frame_)
+        {
+            process_header_write(0, next_seqnum(), 0);
+            chunk_.reset(0, HEADER_SIZE);
+            should_send_empty_frame_ = false;
+        }
+        else if (total_bytes_written_ < total_bytes_)
+        {
+            std::size_t offset = total_bytes_written_;
+            std::size_t remaining = total_bytes_ - total_bytes_written_;
+            std::size_t size = (std::min)(max_frame_size_, remaining);
+            process_header_write(static_cast<std::uint32_t>(size), next_seqnum(), offset);
+            chunk_.reset(offset, offset + size + HEADER_SIZE);
+            if (remaining == max_frame_size_)
+            {
+                should_send_empty_frame_ = true;
+            }
+            total_bytes_written_ += size;
+        }
+        else
+        {
+            // We're done
+            chunk_.reset();
+        }
+    }
+
+public:
+    message_writer(std::size_t max_frame_size = MAX_PACKET_SIZE) noexcept : max_frame_size_(max_frame_size) {}
+
+    span<std::uint8_t> prepare_buffer(std::size_t msg_size, std::uint8_t& seqnum)
+    {
+        buffer_.resize(msg_size + HEADER_SIZE);
+        total_bytes_ = msg_size;
+        total_bytes_written_ = 0;
+        should_send_empty_frame_ = msg_size == 0;
+        seqnum_ = &seqnum;
+        prepare_next_chunk();
+        return {buffer_.data() + HEADER_SIZE, msg_size};
+    }
+
+    bool done() const noexcept { return chunk_.done(); }
+
+    // This function returns an empty buffer to signal that we're done
+    span<const std::uint8_t> next_chunk() const
+    {
+        BOOST_ASSERT(!done());
+        return chunk_.get_chunk(buffer_);
+    }
+
+    void on_bytes_written(std::size_t n)
+    {
+        BOOST_ASSERT(!done());
+
+        // Acknowledge the written bytes
+        chunk_.on_bytes_written(n);
+
+        // Prepare the next chunk, if required
+        if (chunk_.done())
+        {
+            prepare_next_chunk();
+        }
+    }
+};
+
+}  // namespace detail
+}  // namespace mysql
+}  // namespace boost
+
+#endif

linx64/include/boost/mysql/impl/internal/channel/read_buffer.hpp

@@ -0,0 +1,133 @@
+//
+// Copyright (c) 2019-2023 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_MYSQL_IMPL_INTERNAL_CHANNEL_READ_BUFFER_HPP
+#define BOOST_MYSQL_IMPL_INTERNAL_CHANNEL_READ_BUFFER_HPP
+
+#include <boost/assert.hpp>
+#include <boost/core/span.hpp>
+
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <vector>
+
+namespace boost {
+namespace mysql {
+namespace detail {
+
+// Custom buffer type optimized for read operations performed in the MySQL protocol.
+// The buffer is a single, resizable chunk of memory with four areas:
+//   - Reserved area: messages that have already been read but are kept alive,
+//     either because we need them or because we haven't cleaned them yet.
+//   - Current message area: delimits the message we are currently parsing.
+//   - Pending bytes area: bytes we've read but haven't been parsed into a message yet.
+//   - Free area: free space for more bytes to be read.
+class read_buffer
+{
+    std::vector<std::uint8_t> buffer_;
+    std::size_t current_message_offset_{0};
+    std::size_t pending_offset_{0};
+    std::size_t free_offset_{0};
+
+public:
+    read_buffer(std::size_t size) : buffer_(size, std::uint8_t(0)) { buffer_.resize(buffer_.capacity()); }
+
+    // Whole buffer accessors
+    const std::uint8_t* first() const noexcept { return buffer_.data(); }
+    std::size_t size() const noexcept { return buffer_.size(); }
+
+    // Area accessors
+    std::uint8_t* reserved_first() noexcept { return buffer_.data(); }
+    const std::uint8_t* reserved_first() const noexcept { return buffer_.data(); }
+    std::uint8_t* current_message_first() noexcept { return buffer_.data() + current_message_offset_; }
+    const std::uint8_t* current_message_first() const noexcept
+    {
+        return buffer_.data() + current_message_offset_;
+    }
+    std::uint8_t* pending_first() noexcept { return buffer_.data() + pending_offset_; }
+    const std::uint8_t* pending_first() const noexcept { return buffer_.data() + pending_offset_; }
+    std::uint8_t* free_first() noexcept { return buffer_.data() + free_offset_; }
+    const std::uint8_t* free_first() const noexcept { return buffer_.data() + free_offset_; }
+
+    std::size_t reserved_size() const noexcept { return current_message_offset_; }
+    std::size_t current_message_size() const noexcept { return pending_offset_ - current_message_offset_; }
+    std::size_t pending_size() const noexcept { return free_offset_ - pending_offset_; }
+    std::size_t free_size() const noexcept { return buffer_.size() - free_offset_; }
+
+    span<const std::uint8_t> reserved_area() const noexcept { return {reserved_first(), reserved_size()}; }
+    span<const std::uint8_t> current_message() const noexcept
+    {
+        return {current_message_first(), current_message_size()};
+    }
+    span<const std::uint8_t> pending_area() const noexcept { return {pending_first(), pending_size()}; }
+    span<std::uint8_t> free_area() noexcept { return {free_first(), free_size()}; }
+
+    // Moves n bytes from the free to the processing area (e.g. they've been read)
+    void move_to_pending(std::size_t length) noexcept
+    {
+        BOOST_ASSERT(length <= free_size());
+        free_offset_ += length;
+    }
+
+    // Moves n bytes from the processing to the current message area
+    void move_to_current_message(std::size_t length) noexcept
+    {
+        BOOST_ASSERT(length <= pending_size());
+        pending_offset_ += length;
+    }
+
+    // Removes the last length bytes from the current message area,
+    // effectively memmove'ing all subsequent bytes backwards.
+    // Used to remove intermediate headers. length must be > 0
+    void remove_current_message_last(std::size_t length) noexcept
+    {
+        BOOST_ASSERT(length <= current_message_size());
+        BOOST_ASSERT(length > 0);
+        std::memmove(pending_first() - length, pending_first(), pending_size());
+        pending_offset_ -= length;
+        free_offset_ -= length;
+    }
+
+    // Moves length bytes from the current message area to the reserved area
+    // Used to move entire parsed messages or message headers
+    void move_to_reserved(std::size_t length) noexcept
+    {
+        BOOST_ASSERT(length <= current_message_size());
+        current_message_offset_ += length;
+    }
+
+    // Removes the reserved area, effectively memmove'ing evth backwards
+    void remove_reserved() noexcept
+    {
+        if (reserved_size() > 0)
+        {
+            std::size_t currmsg_size = current_message_size();
+            std::size_t pend_size = pending_size();
+            std::memmove(buffer_.data(), current_message_first(), currmsg_size + pend_size);
+            current_message_offset_ = 0;
+            pending_offset_ = currmsg_size;
+            free_offset_ = currmsg_size + pend_size;
+        }
+    }
+
+    // Makes sure the free size is at least n bytes long; resizes the buffer if required
+    void grow_to_fit(std::size_t n)
+    {
+        if (free_size() < n)
+        {
+            buffer_.resize(buffer_.size() + n - free_size());
+            buffer_.resize(buffer_.capacity());
+        }
+    }
+};
+
+}  // namespace detail
+}  // namespace mysql
+}  // namespace boost
+
+#endif /* INCLUDE_BOOST_MYSQL_DETAIL_AUXILIAR_STATIC_STRING_HPP_ */

linx64/include/boost/mysql/impl/internal/channel/valgrind.hpp

@@ -0,0 +1,38 @@
+//
+// Copyright (c) 2019-2023 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_MYSQL_IMPL_INTERNAL_CHANNEL_VALGRIND_HPP
+#define BOOST_MYSQL_IMPL_INTERNAL_CHANNEL_VALGRIND_HPP
+
+#include <cstddef>
+
+#ifdef BOOST_MYSQL_VALGRIND_TESTS
+#include <valgrind/memcheck.h>
+#endif
+
+namespace boost {
+namespace mysql {
+namespace detail {
+
+#ifdef BOOST_MYSQL_VALGRIND_TESTS
+
+inline void valgrind_make_mem_defined(const void* data, std::size_t size)
+{
+    VALGRIND_MAKE_MEM_DEFINED(data, size);
+}
+
+#else
+
+inline void valgrind_make_mem_defined(const void*, std::size_t) noexcept {}
+
+#endif
+
+}  // namespace detail
+}  // namespace mysql
+}  // namespace boost
+
+#endif /* INCLUDE_BOOST_MYSQL_DETAIL_AUXILIAR_VALGRIND_HPP_ */

linx64/include/boost/mysql/impl/internal/channel/write_message.hpp

@@ -0,0 +1,97 @@
+//
+// Copyright (c) 2019-2023 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_MYSQL_IMPL_INTERNAL_CHANNEL_WRITE_MESSAGE_HPP
+#define BOOST_MYSQL_IMPL_INTERNAL_CHANNEL_WRITE_MESSAGE_HPP
+
+#include <boost/mysql/error_code.hpp>
+
+#include <boost/mysql/detail/any_stream.hpp>
+
+#include <boost/mysql/impl/internal/channel/message_writer.hpp>
+#include <boost/mysql/impl/internal/protocol/constants.hpp>
+
+#include <boost/asio/async_result.hpp>
+#include <boost/asio/buffer.hpp>
+#include <boost/asio/compose.hpp>
+#include <boost/asio/coroutine.hpp>
+
+#include <cstddef>
+#include <cstdint>
+
+namespace boost {
+namespace mysql {
+namespace detail {
+
+// Writes an entire message to stream; partitions the message into
+// chunks and adds the required headers
+inline void write_message(any_stream& stream, message_writer& processor, error_code& ec)
+{
+    while (!processor.done())
+    {
+        std::size_t bytes_written = stream.write_some(asio::buffer(processor.next_chunk()), ec);
+        if (ec)
+            break;
+        processor.on_bytes_written(bytes_written);
+    }
+}
+
+struct write_message_op : boost::asio::coroutine
+{
+    any_stream& stream_;
+    message_writer& processor_;
+
+    write_message_op(any_stream& stream, message_writer& processor) noexcept
+        : stream_(stream), processor_(processor)
+    {
+    }
+
+    template <class Self>
+    void operator()(Self& self, error_code ec = {}, std::size_t bytes_written = 0)
+    {
+        // Error handling
+        if (ec)
+        {
+            self.complete(ec);
+            return;
+        }
+
+        // Non-error path
+        BOOST_ASIO_CORO_REENTER(*this)
+        {
+            // done() never returns false after a call to prepare_buffer(), so no post() needed
+            BOOST_ASSERT(!processor_.done());
+            while (!processor_.done())
+            {
+                BOOST_ASIO_CORO_YIELD stream_.async_write_some(
+                    asio::buffer(processor_.next_chunk()),
+                    std::move(self)
+                );
+                processor_.on_bytes_written(bytes_written);
+            };
+
+            self.complete(error_code());
+        }
+    }
+};
+
+template <BOOST_ASIO_COMPLETION_TOKEN_FOR(void(::boost::mysql::error_code)) CompletionToken>
+BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void(boost::mysql::error_code))
+async_write_message(any_stream& stream, message_writer& processor, CompletionToken&& token)
+{
+    return boost::asio::async_compose<CompletionToken, void(error_code)>(
+        write_message_op(stream, processor),
+        token,
+        stream
+    );
+}
+
+}  // namespace detail
+}  // namespace mysql
+}  // namespace boost
+
+#endif