Fixes issue 50 and 44.

Author: mzimbres

CMakeLists.txt

@@ -299,6 +299,15 @@ if (MSVC)
     target_compile_definitions(test_request PRIVATE _WIN32_WINNT=0x0601)
 endif()
 
+add_executable(test_issue_50 tests/issue_50.cpp)
+target_compile_features(test_issue_50 PUBLIC cxx_std_20)
+target_link_libraries(test_issue_50 common)
+add_test(test_issue_50 test_issue_50)
+if (MSVC)
+    target_compile_options(test_issue_50 PRIVATE /bigobj)
+    target_compile_definitions(test_issue_50 PRIVATE _WIN32_WINNT=0x0601)
+endif()
+
 # Install
 #=======================================================================
 

README.md

@@ -48,7 +48,7 @@ connection to read Redis
 ```cpp
 auto co_main() -> net::awaitable<void>
 {
-   auto conn = std::make_shared<connection>(co_await net::this_coro::executor);
+   connection conn{co_await net::this_coro::executor};
 
    // From examples/common.hpp to avoid vebosity
    co_await connect(conn, "127.0.0.1", "6379");
@@ -65,7 +65,7 @@ auto co_main() -> net::awaitable<void>
    std::tuple<ignore, std::map<std::string, std::string>, ignore> resp;
 
    // Executes the request. See below why we are using operator ||.
-   co_await (conn->async_run() || conn->async_exec(req, adapt(resp)));
+   co_await (conn.async_run() || conn.async_exec(req, adapt(resp)));
    // Use the map from std::get<1>(resp) ...
 }
 ```

include/aedis/adapt.hpp

@@ -40,22 +40,13 @@ namespace detail
 
 class ignore_adapter {
 public:
-   explicit ignore_adapter(std::size_t max_read_size) : max_read_size_{max_read_size} {}
-
    void
    operator()(
       std::size_t, resp3::node<std::string_view> const&, boost::system::error_code&) { }
 
    [[nodiscard]]
    auto get_supported_response_size() const noexcept
       { return static_cast<std::size_t>(-1);}
-
-   [[nodiscard]]
-   auto get_max_read_size(std::size_t) const noexcept
-      { return max_read_size_;}
-
-private:
-   std::size_t max_read_size_;
 };
 
 template <class Tuple>
@@ -67,11 +58,9 @@ class static_adapter {
    using adapters_array_type = std::array<variant_type, size>;
 
    adapters_array_type adapters_;
-   std::size_t max_read_size_;
 
 public:
-   explicit static_adapter(Tuple& r, std::size_t max_read_size)
-   : max_read_size_{max_read_size}
+   explicit static_adapter(Tuple& r)
    {
       adapter::detail::assigner<size - 1>::assign(adapters_, r);
    }
@@ -80,10 +69,6 @@ class static_adapter {
    auto get_supported_response_size() const noexcept
       { return size;}
 
-   [[nodiscard]]
-   auto get_max_read_size(std::size_t) const noexcept
-      { return max_read_size_;}
-
    void
    operator()(
       std::size_t i,
@@ -102,23 +87,17 @@ class vector_adapter {
 private:
    using adapter_type = typename adapter::detail::response_traits<Vector>::adapter_type;
    adapter_type adapter_;
-   std::size_t max_read_size_;
 
 public:
-   explicit vector_adapter(Vector& v, std::size_t max_read_size)
+   explicit vector_adapter(Vector& v)
    : adapter_{adapter::adapt2(v)}
-   , max_read_size_{max_read_size}
    { }
 
    [[nodiscard]]
    auto
    get_supported_response_size() const noexcept
       { return static_cast<std::size_t>(-1);}
 
-   [[nodiscard]]
-   auto get_max_read_size(std::size_t) const noexcept
-      { return max_read_size_;}
-
    void
    operator()(
       std::size_t,
@@ -137,26 +116,26 @@ struct response_traits<void> {
    using response_type = void;
    using adapter_type = detail::ignore_adapter;
 
-   static auto adapt(std::size_t max_read_size) noexcept
-      { return detail::ignore_adapter{max_read_size}; }
+   static auto adapt() noexcept
+      { return detail::ignore_adapter{}; }
 };
 
 template <class String, class Allocator>
 struct response_traits<std::vector<resp3::node<String>, Allocator>> {
    using response_type = std::vector<resp3::node<String>, Allocator>;
    using adapter_type = vector_adapter<response_type>;
 
-   static auto adapt(response_type& v, std::size_t max_read_size) noexcept
-      { return adapter_type{v, max_read_size}; }
+   static auto adapt(response_type& v) noexcept
+      { return adapter_type{v}; }
 };
 
 template <class ...Ts>
 struct response_traits<std::tuple<Ts...>> {
    using response_type = std::tuple<Ts...>;
    using adapter_type = static_adapter<response_type>;
 
-   static auto adapt(response_type& r, std::size_t max_read_size) noexcept
-      { return adapter_type{r, max_read_size}; }
+   static auto adapt(response_type& r) noexcept
+      { return adapter_type{r}; }
 };
 
 template <class Adapter>
@@ -171,10 +150,6 @@ class wrapper {
    auto get_supported_response_size() const noexcept
       { return adapter_.get_supported_response_size();}
 
-   [[nodiscard]]
-   auto get_max_read_size(std::size_t) const noexcept
-      { return adapter_.get_max_read_size(0); }
-
 private:
    Adapter adapter_;
 };
@@ -192,13 +167,10 @@ auto make_adapter_wrapper(Adapter adapter)
  *
  *  This function can be used to create adapters that ignores
  *  responses.
- *
- *  @param max_read_size Specifies the maximum size of the read
- *  buffer.
  */
-inline auto adapt(std::size_t max_read_size = (std::numeric_limits<std::size_t>::max)()) noexcept
+inline auto adapt() noexcept
 {
-   return detail::response_traits<void>::adapt(max_read_size);
+   return detail::response_traits<void>::adapt();
 }
 
 /** @brief Adapts a type to be used as a response.
@@ -213,13 +185,11 @@ inline auto adapt(std::size_t max_read_size = (std::numeric_limits<std::size_t>:
  *  and `std::string`.
  *
  *  @param t Tuple containing the responses.
- *  @param max_read_size Specifies the maximum size of the read
- *  buffer.
  */
 template<class T>
-auto adapt(T& t, std::size_t max_read_size = (std::numeric_limits<std::size_t>::max)()) noexcept
+auto adapt(T& t) noexcept
 {
-   return detail::response_traits<T>::adapt(t, max_read_size);
+   return detail::response_traits<T>::adapt(t);
 }
 
 } // aedis

include/aedis/connection.hpp

@@ -187,15 +187,21 @@ class basic_connection :
    auto cancel(operation op) -> std::size_t
       { return base_type::cancel(op); }
 
+   /// Sets the maximum size of the read buffer.
+   void set_max_buffer_read_size(std::size_t max_read_size) noexcept
+      { base_type::set_max_buffer_read_size(max_read_size); }
+
 private:
    using this_type = basic_connection<next_layer_type>;
 
    template <class, class> friend class detail::connection_base;
    template <class, class> friend struct detail::exec_read_op;
    template <class, class> friend struct detail::exec_op;
+   template <class, class> friend struct detail::receive_op;
    template <class> friend struct detail::reader_op;
    template <class> friend struct detail::writer_op;
    template <class> friend struct detail::run_op;
+   template <class> friend struct detail::wait_receive_op;
 
    void close() { stream_.close(); }
    auto is_open() const noexcept { return stream_.is_open(); }

include/aedis/detail/connection_base.hpp

@@ -47,7 +47,7 @@ class connection_base {
    connection_base(executor_type ex, std::pmr::memory_resource* resource)
    : writer_timer_{ex}
    , read_timer_{ex}
-   , guarded_op_{ex}
+   , channel_{ex}
    , read_buffer_{resource}
    , write_buffer_{resource}
    , reqs_{resource}
@@ -76,7 +76,7 @@ class connection_base {
          }
          case operation::receive:
          {
-            guarded_op_.cancel();
+            channel_.cancel();
             return 1U;
          }
          default: BOOST_ASSERT(false); return 0;
@@ -136,7 +136,7 @@ class connection_base {
    template <class Adapter, class CompletionToken>
    auto async_exec(resp3::request const& req, Adapter adapter, CompletionToken token)
    {
-      BOOST_ASSERT_MSG(req.size() <= adapter.get_supported_response_size(), "Request and adapter have incompatible sizes.");
+      BOOST_ASSERT_MSG(req.size() <= adapter.get_supported_response_size(), "Request and response have incompatible sizes.");
 
       return boost::asio::async_compose
          < CompletionToken
@@ -149,9 +149,10 @@ class connection_base {
    {
       auto f = detail::make_adapter_wrapper(adapter);
 
-      return guarded_op_.async_wait(
-         resp3::async_read(derived().next_layer(), make_dynamic_buffer(adapter.get_max_read_size(0)), f, boost::asio::deferred),
-         std::move(token));
+      return boost::asio::async_compose
+         < CompletionToken
+         , void(boost::system::error_code, std::size_t)
+         >(detail::receive_op<Derived, decltype(f)>{&derived(), f}, token, channel_);
    }
 
    template <class CompletionToken>
@@ -163,10 +164,14 @@ class connection_base {
          >(detail::run_op<Derived>{&derived()}, token, writer_timer_);
    }
 
+   void set_max_buffer_read_size(std::size_t max_read_size) noexcept
+      {max_read_size_ = max_read_size;}
+
 private:
    using clock_type = std::chrono::steady_clock;
    using clock_traits_type = boost::asio::wait_traits<clock_type>;
    using timer_type = boost::asio::basic_waitable_timer<clock_type, clock_traits_type, executor_type>;
+   using channel_type = boost::asio::experimental::channel<executor_type, void(boost::system::error_code, std::size_t)>;
 
    auto derived() -> Derived& { return static_cast<Derived&>(*this); }
 
@@ -272,7 +277,17 @@ class connection_base {
    template <class> friend struct detail::run_op;
    template <class, class> friend struct detail::exec_op;
    template <class, class> friend struct detail::exec_read_op;
-   template <class> friend struct detail::send_receive_op;
+   template <class, class> friend struct detail::receive_op;
+   template <class> friend struct detail::wait_receive_op;
+
+   template <class CompletionToken>
+   auto async_wait_receive(CompletionToken token)
+   {
+      return boost::asio::async_compose
+         < CompletionToken
+         , void(boost::system::error_code)
+         >(wait_receive_op<Derived>{&derived()}, token, channel_);
+   }
 
    void cancel_push_requests()
    {
@@ -299,12 +314,12 @@ class connection_base {
          std::rotate(std::rbegin(reqs_), std::rbegin(reqs_) + 1, rend);
       }
 
-      if (derived().is_open() && cmds_ == 0 && write_buffer_.empty())
+      if (derived().is_open() && !is_waiting_response() && write_buffer_.empty())
          writer_timer_.cancel();
    }
 
-   auto make_dynamic_buffer(std::size_t max_read_size = 512)
-      { return boost::asio::dynamic_buffer(read_buffer_, max_read_size); }
+   auto make_dynamic_buffer()
+      { return boost::asio::dynamic_buffer(read_buffer_, max_read_size_); }
 
    template <class CompletionToken>
    auto reader(CompletionToken&& token)
@@ -336,7 +351,6 @@ class connection_base {
    void stage_request(req_info& ri)
    {
       write_buffer_ += ri.get_request().payload();
-      cmds_ += ri.get_request().size();
       ri.mark_staged();
    }
 
@@ -358,17 +372,22 @@ class connection_base {
       }
    }
 
+   bool is_waiting_response() const noexcept
+   {
+      return !std::empty(reqs_) && reqs_.front()->is_written();
+   }
+
    // Notice we use a timer to simulate a condition-variable. It is
    // also more suitable than a channel and the notify operation does
    // not suspend.
    timer_type writer_timer_;
    timer_type read_timer_;
-   detail::guarded_operation<executor_type> guarded_op_;
+   channel_type channel_;
 
    std::pmr::string read_buffer_;
    std::pmr::string write_buffer_;
-   std::size_t cmds_ = 0;
    reqs_type reqs_;
+   std::size_t max_read_size_ = (std::numeric_limits<std::size_t>::max)();
 };
 
 } // aedis