Removes coalesce property of the requests.

It doesn't make any sense after the implementation of full-duplex
communication.

Author: mzimbres

CMakeLists.txt

@@ -224,14 +224,6 @@ if (MSVC)
     target_compile_definitions(test_conn_quit PRIVATE _WIN32_WINNT=0x0601)
 endif()
 
-add_executable(test_conn_quit_coalesce tests/conn_quit_coalesce.cpp)
-add_test(test_conn_quit_coalesce test_conn_quit_coalesce)
-target_compile_features(test_conn_quit_coalesce PUBLIC cxx_std_17)
-if (MSVC)
-    target_compile_options(test_conn_quit_coalesce PRIVATE /bigobj)
-    target_compile_definitions(test_conn_quit_coalesce PRIVATE _WIN32_WINNT=0x0601)
-endif()
-
 add_executable(test_conn_reconnect tests/conn_reconnect.cpp)
 target_compile_features(test_conn_reconnect PUBLIC cxx_std_20)
 target_link_libraries(test_conn_reconnect common)

README.md

@@ -261,10 +261,8 @@ co_await (conn.async_exec(...) || time.async_wait(...))
 co_await (conn.async_exec(...) || conn.async_exec(...) || ... || conn.async_exec(...))
 ```
 
-* This works but is unnecessary. Unless the user has set
-  `boost::redis::request::config::coalesce` to `false`, and he
-  usually shouldn't, the connection will automatically merge the
-  individual requests into a single payload.
+* This works but is unnecessary, the connection will automatically
+  merge the individual requests into a single payload.
 
 <a name="requests"></a>
 ## Requests
@@ -874,11 +872,16 @@ Acknowledgement to people that helped shape Boost.Redis
   including possible resp3 errors without losing the error diagnostic
   part. Basicaly instead of accessing values as `std::get<N>(resp)`
   users have to type `std::get<N>(resp).value()`
+* Implements full-duplex communication. Before these changes the
+  connection would wait for a response to arrive before sending the
+  next. Now requests are continuously coalesced and written to the
+  socket. This made the request::coalesce unnecessary and threfore it
+  was removed.
 
 ### v1.4.0-1
 
 * Renames `retry_on_connection_lost` to `cancel_if_unresponded`.  (v1.4.1)
-* Removes dependency on Boost.Hana, boost::string_view, Boost.Variant2 and Boost.Spirit.
+* Removes dependency on Boost.Hana, `boost::string_view`, Boost.Variant2 and Boost.Spirit.
 * Fixes build and setup CI on windows.
 
 ### v1.3.0-1

include/boost/redis/detail/connection_base.hpp

@@ -193,6 +193,7 @@ class connection_base {
       // There is small optimization possible here: traverse only the
       // partition of unwritten requests instead of them all.
       std::for_each(std::begin(reqs_), std::end(reqs_), [](auto const& ptr) {
+         BOOST_ASSERT_MSG(ptr != nullptr, "Expects non-null pointer.");
          if (ptr->is_staged())
             ptr->mark_written();
       });
@@ -364,35 +365,21 @@ class connection_base {
          >(detail::exec_read_op<Derived, Adapter>{&derived(), adapter, cmds}, token, writer_timer_);
    }
 
-   void stage_request(req_info& ri)
-   {
-      write_buffer_ += ri.get_request().payload();
-      ri.mark_staged();
-   }
-
    [[nodiscard]] bool coalesce_requests()
    {
       // Coalesces the requests and marks them staged. After a
       // successful write staged requests will be marked as written.
-      std::size_t pos = 0;
-      for (; pos < std::size(reqs_); ++pos)
-         if (reqs_.at(pos)->is_waiting_write())
-            break;
-
-      if (pos == std::size(reqs_))
-         return false;
-
-      stage_request(*reqs_.at(pos));
+      auto const point = std::partition_point(std::cbegin(reqs_), std::cend(reqs_), [](auto const& ri) {
+            return !ri->is_waiting_write();
+      });
 
-      for (std::size_t i = pos + 1; i < std::size(reqs_); ++i) {
-         if (!reqs_.at(i - 1)->get_request().get_config().coalesce ||
-             !reqs_.at(i - 0)->get_request().get_config().coalesce) {
-            break;
-         }
-         stage_request(*reqs_.at(i));
-      }
+      std::for_each(point, std::cend(reqs_), [this](auto const& ri) {
+         // Stage the request.
+         write_buffer_ += ri->get_request().payload();
+         ri->mark_staged();
+      });
 
-      return true;
+      return point != std::cend(reqs_);
    }
 
    bool is_waiting_response() const noexcept

include/boost/redis/request.hpp

@@ -55,12 +55,6 @@ class request {
        */
       bool cancel_on_connection_lost = true;
 
-      /** \brief If `true` the request will be coalesced with other
-       * requests, see https://redis.io/topics/pipelining. Otherwise
-       * the request is sent individually.
-       */
-      bool coalesce = true;
-
       /** \brief If `true` the request will complete with
        * boost::redis::error::not_connected if `async_exec` is called before
        * the connection with Redis was established.
@@ -88,7 +82,7 @@ class request {
     *  \param cfg Configuration options.
     */
     explicit
-    request(config cfg = config{true, true, false, true, true})
+    request(config cfg = config{true, false, true, true})
     : cfg_{cfg} {}
 
     //// Returns the number of commands contained in this request.

tests/conn_exec.cpp

@@ -30,18 +30,15 @@ using boost::redis::ignore_t;
 BOOST_AUTO_TEST_CASE(hello_priority)
 {
    request req1;
-   req1.get_config().coalesce = false;
    req1.push("PING", "req1");
 
    request req2;
-   req2.get_config().coalesce = false;
    req2.get_config().hello_with_priority = false;
    req2.push("HELLO", 3);
    req2.push("PING", "req2");
    req2.push("QUIT");
 
    request req3;
-   req3.get_config().coalesce = false;
    req3.get_config().hello_with_priority = true;
    req3.push("HELLO", 3);
    req3.push("PING", "req3");

tests/conn_exec_cancel.cpp

@@ -49,12 +49,10 @@ auto async_ignore_explicit_cancel_of_req_written() -> net::awaitable<void>
 
    // See NOTE1.
    request req0;
-   req0.get_config().coalesce = false;
    req0.push("HELLO", 3);
    co_await conn->async_exec(req0, gresp, net::use_awaitable);
 
    request req1;
-   req1.get_config().coalesce = false;
    req1.push("BLPOP", "any", 3);
 
    // Should not be canceled.
@@ -97,82 +95,79 @@ auto ignore_implicit_cancel_of_req_written() -> net::awaitable<void>
 
    // See NOTE1.
    request req0;
-   req0.get_config().coalesce = false;
    req0.push("HELLO", 3);
-   std::ignore = co_await conn->async_exec(req0, ignore, net::use_awaitable);
+   co_await conn->async_exec(req0, ignore, net::use_awaitable);
 
    // Will be cancelled after it has been written but before the
    // response arrives.
    request req1;
-   req1.get_config().coalesce = false;
    req1.push("BLPOP", "any", 3);
 
-   // Will be cancelled before it is written.
-   request req2;
-   req2.get_config().coalesce = false;
-   req2.get_config().cancel_on_connection_lost = true;
-   req2.push("PING");
-
    net::steady_timer st{ex};
    st.expires_after(std::chrono::seconds{1});
 
-   boost::system::error_code ec1, ec2, ec3;
+   boost::system::error_code ec1, ec2;
    co_await (
       conn->async_exec(req1, ignore, redir(ec1)) ||
-      conn->async_exec(req2, ignore, redir(ec2)) ||
-      st.async_wait(redir(ec3))
+      st.async_wait(redir(ec2))
    );
 
    BOOST_CHECK_EQUAL(ec1, net::error::basic_errors::operation_aborted);
-   BOOST_CHECK_EQUAL(ec2, net::error::basic_errors::operation_aborted);
-   BOOST_TEST(!ec3);
+   BOOST_TEST(!ec2);
+}
+
+BOOST_AUTO_TEST_CASE(test_ignore_explicit_cancel_of_req_written)
+{
+   run(async_ignore_explicit_cancel_of_req_written());
+}
+
+BOOST_AUTO_TEST_CASE(test_ignore_implicit_cancel_of_req_written)
+{
+   run(ignore_implicit_cancel_of_req_written());
 }
 
-auto cancel_of_req_written_on_run_canceled() -> net::awaitable<void>
+BOOST_AUTO_TEST_CASE(test_cancel_of_req_written_on_run_canceled)
 {
+   net::io_context ioc;
+   auto const endpoints = resolve();
+   connection conn{ioc};
+   net::connect(conn.next_layer(), endpoints);
+
    request req0;
-   req0.get_config().coalesce = false;
    req0.push("HELLO", 3);
 
+   // Sends a request that will be blocked forever, so we can test
+   // canceling it while waiting for a response.
    request req1;
    req1.get_config().cancel_on_connection_lost = true;
    req1.get_config().cancel_if_unresponded = true;
    req1.push("BLPOP", "any", 0);
 
-   auto ex = co_await net::this_coro::executor;
-   auto conn = std::make_shared<connection>(ex);
-   co_await connect(conn, "127.0.0.1", "6379");
-
-   net::steady_timer st{ex};
-   st.expires_after(std::chrono::seconds{1});
+   auto c1 = [&](auto ec, auto)
+   {
+      BOOST_CHECK_EQUAL(ec, net::error::basic_errors::operation_aborted);
+   };
 
-   boost::system::error_code ec0, ec1, ec2, ec3;
-   co_await (
-      conn->async_exec(req0, ignore, redir(ec0)) &&
-      (conn->async_exec(req1, ignore, redir(ec1)) ||
-      conn->async_run(redir(ec2)) ||
-      st.async_wait(redir(ec3)))
-   );
+   auto c0 = [&](auto ec, auto)
+   {
+      BOOST_TEST(!ec);
+      conn.async_exec(req1, ignore, c1);
+   };
 
-   BOOST_TEST(!ec0);
-   BOOST_CHECK_EQUAL(ec1, net::error::basic_errors::operation_aborted);
-   BOOST_CHECK_EQUAL(ec2, net::error::basic_errors::operation_aborted);
-   BOOST_TEST(!ec3);
-}
+   conn.async_exec(req0, ignore, c0);
 
-BOOST_AUTO_TEST_CASE(test_ignore_explicit_cancel_of_req_written)
-{
-   run(async_ignore_explicit_cancel_of_req_written());
-}
+   conn.async_run([](auto ec){
+      BOOST_CHECK_EQUAL(ec, net::error::basic_errors::operation_aborted);
+   });
 
-BOOST_AUTO_TEST_CASE(test_ignore_implicit_cancel_of_req_written)
-{
-   run(ignore_implicit_cancel_of_req_written());
-}
+   net::steady_timer st{ioc};
+   st.expires_after(std::chrono::seconds{1});
+   st.async_wait([&](auto ec){
+      BOOST_TEST(!ec);
+      conn.cancel(operation::run);
+   });
 
-BOOST_AUTO_TEST_CASE(test_cancel_of_req_written_on_run_canceled)
-{
-   run(cancel_of_req_written_on_run_canceled());
+   ioc.run();
 }
 
 #else