Test improvements and bugfix in send-retry.

Author: mzimbres

CMakeLists.txt

@@ -217,6 +217,7 @@ 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)
 add_test(test_conn_reconnect test_conn_reconnect)
 if (MSVC)
     target_compile_options(test_conn_reconnect PRIVATE /bigobj)
@@ -250,6 +251,7 @@ endif()
 
 add_executable(test_conn_exec_cancel tests/conn_exec_cancel.cpp)
 target_compile_features(test_conn_exec_cancel PUBLIC cxx_std_20)
+target_link_libraries(test_conn_exec_cancel common)
 add_test(test_conn_exec_cancel test_conn_exec_cancel)
 if (MSVC)
     target_compile_options(test_conn_exec_cancel PRIVATE /bigobj)
@@ -258,6 +260,7 @@ endif()
 
 add_executable(test_conn_echo_stress tests/conn_echo_stress.cpp)
 target_compile_features(test_conn_echo_stress PUBLIC cxx_std_20)
+target_link_libraries(test_conn_echo_stress common)
 add_test(test_conn_echo_stress test_conn_echo_stress)
 if (MSVC)
     target_compile_options(test_conn_echo_stress PRIVATE /bigobj)

examples/common/common.cpp

@@ -72,4 +72,23 @@ connect(
       throw std::runtime_error("Connect timeout");
 }
 
+auto run(net::awaitable<void> op) -> int
+{
+   try {
+      net::io_context ioc;
+      net::co_spawn(ioc, std::move(op), [](std::exception_ptr p) {
+         if (p)
+            std::rethrow_exception(p);
+      });
+      ioc.run();
+
+      return 0;
+
+   } catch (std::exception const& e) {
+      std::cerr << "Error: " << e.what() << std::endl;
+   }
+
+   return 1;
+}
+
 #endif // defined(BOOST_ASIO_HAS_CO_AWAIT)

examples/common/common.hpp

@@ -28,5 +28,7 @@ connect(
 
 auto healthy_checker(std::shared_ptr<connection> conn) -> boost::asio::awaitable<void>;
 
+auto run(boost::asio::awaitable<void> op) -> int;
+
 #endif // defined(BOOST_ASIO_HAS_CO_AWAIT)
 #endif // AEDIS_EXAMPLES_COMMON_HPP

examples/common/main.cpp

@@ -4,30 +4,27 @@
  * accompanying file LICENSE.txt)
  */
 
-#include <iostream>
 #include <boost/asio.hpp>
+
 #if defined(BOOST_ASIO_HAS_CO_AWAIT)
 
-namespace net = boost::asio;
-extern net::awaitable<void> async_main();
+#include "common.hpp"
+
+extern boost::asio::awaitable<void> async_main();
 
-// The main function used in our examples.
 auto main() -> int
 {
-   try {
-      net::io_context ioc;
-      net::co_spawn(ioc, async_main(), net::detached);
-      net::co_spawn(ioc, async_main(), [](std::exception_ptr p) {
-         if (p)
-            std::rethrow_exception(p);
-      });
-      ioc.run();
-   } catch (std::exception const& e) {
-      std::cerr << "Error: " << e.what() << std::endl;
-      return 1;
-   }
+   return run(async_main());
 }
 
 #else // defined(BOOST_ASIO_HAS_CO_AWAIT)
-auto main() -> int {std::cout << "Requires coroutine support." << std::endl; return 0;}
+
+#include <iostream>
+
+auto main() -> int
+{
+   std::cout << "Requires coroutine support." << std::endl;
+   return 0;
+}
+
 #endif // defined(BOOST_ASIO_HAS_CO_AWAIT)

include/aedis/detail/connection_base.hpp

@@ -104,16 +104,19 @@ class connection_base {
       return ret;
    }
 
+   // Remove requests that have the flag cancel_if_not_sent_when_connection_lost set
    auto cancel_on_conn_lost() -> std::size_t
    {
+      // Must return false if the request should be removed.
       auto cond = [](auto const& ptr)
       {
          BOOST_ASSERT(ptr != nullptr);
 
-         if (ptr->get_request().get_config().cancel_on_connection_lost)
-            return false;
-
-         return !(!ptr->get_request().get_config().retry && ptr->is_written());
+         if (ptr->is_written()) {
+            return ptr->get_request().get_config().retry_on_connection_lost;
+         } else {
+            return !ptr->get_request().get_config().cancel_on_connection_lost;
+         }
       };
 
       auto point = std::stable_partition(std::begin(reqs_), std::end(reqs_), cond);
@@ -128,6 +131,7 @@ class connection_base {
       std::for_each(std::begin(reqs_), std::end(reqs_), [](auto const& ptr) {
          return ptr->reset_status();
       });
+
       return ret;
    }
 

include/aedis/detail/connection_ops.hpp

@@ -161,14 +161,7 @@ struct exec_op {
          BOOST_ASSERT(conn->cmds_ != 0);
          yield
          conn->async_exec_read(adapter, conn->reqs_.front()->get_number_of_commands(), std::move(self));
-         if (is_cancelled(self)) {
-            conn->remove_request(info);
-            return self.complete(boost::asio::error::operation_aborted, {});
-         }
-
-         if (ec) {
-            return self.complete(ec, {});
-         }
+         AEDIS_CHECK_OP1(;);
 
          read_size = n;
 

include/aedis/resp3/request.hpp

@@ -169,29 +169,30 @@ class request {
 public:
    /// Request configuration options.
    struct config {
-      /** \brief If true the request will complete with error if the
-       * connection is lost while the request is pending. The default
-       * behaviour is not to close requests.
+      /** \brief Setting it to true will cause
+       * `aedis::connection::async_exec` to complete with error if the
+       * connection is lost. Affects only requests that haven't been
+       * sent yet.
        */
       bool cancel_on_connection_lost = false;
 
-      /** \brief If true the request will be coalesced with other requests,
-       *  see https://redis.io/topics/pipelining. Otherwise the
-       *  request is sent individually.
+      /** \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 error if the
-       * call happens before the connection with Redis was stablished.
+       * call happens before the connection with Redis was established.
        */
       bool cancel_if_not_connected = false;
 
-      /** \brief If true, the implementation will resend this
-       * request if it remains unresponded when
-       * `aedis::connection::async_run` completes. Has effect only if
-       * cancel_on_connection_lost is true.
+      /** \brief If true `aedis::connection::async_exec` will not
+       * cancel this request if the connection is lost. Affects only
+       * requests that have been written to the socket but remained
+       * unresponded when `aedis::connection::async_run` completed.
        */
-      bool retry = true;
+      bool retry_on_connection_lost = true;
 
       /** \brief If this request has a HELLO command and this flag is
        * true, the `aedis::connection` will move it to the front of

tests/common.hpp

@@ -15,3 +15,9 @@ resolve(
    net::ip::tcp::resolver resv{ioc};
    return resv.resolve(host, port);
 }
+
+#ifdef BOOST_ASIO_HAS_CO_AWAIT
+inline
+auto redir(boost::system::error_code& ec)
+   { return net::redirect_error(net::use_awaitable, ec); }
+#endif // BOOST_ASIO_HAS_CO_AWAIT

tests/conn_echo_stress.cpp

@@ -8,27 +8,20 @@
 #include <boost/asio.hpp>
 #ifdef BOOST_ASIO_HAS_CO_AWAIT
 #include <boost/system/errc.hpp>
-
 #define BOOST_TEST_MODULE low level
 #include <boost/test/included/unit_test.hpp>
-
 #include <aedis.hpp>
 #include <aedis/src.hpp>
-
 #include "common.hpp"
+#include "../examples/common/common.hpp"
 
 namespace net = boost::asio;
+namespace resp3 = aedis::resp3;
 using error_code = boost::system::error_code;
-
-using aedis::resp3::request;
 using aedis::operation;
 using aedis::adapt;
-using connection = net::use_awaitable_t<>::as_default_on_t<aedis::connection>;
-
-#include <boost/asio/experimental/awaitable_operators.hpp>
-using namespace net::experimental::awaitable_operators;
 
-net::awaitable<void> push_consumer(std::shared_ptr<connection> conn, int expected)
+auto push_consumer(std::shared_ptr<connection> conn, int expected) -> net::awaitable<void>
 {
    int c = 0;
    for (;;) {
@@ -37,7 +30,7 @@ net::awaitable<void> push_consumer(std::shared_ptr<connection> conn, int expecte
          break;
    }
 
-   request req;
+   resp3::request req;
    req.push("HELLO", 3);
    req.push("QUIT");
    co_await conn->async_exec(req, adapt());
@@ -47,7 +40,7 @@ auto echo_session(std::shared_ptr<connection> conn, std::string id, int n) -> ne
 {
    auto ex = co_await net::this_coro::executor;
 
-   request req;
+   resp3::request req;
    std::tuple<aedis::ignore, std::string> resp;
 
    for (auto i = 0; i < n; ++i) {
@@ -57,7 +50,7 @@ auto echo_session(std::shared_ptr<connection> conn, std::string id, int n) -> ne
       req.push("PING", msg);
       req.push("SUBSCRIBE", "channel");
       boost::system::error_code ec;
-      co_await conn->async_exec(req, adapt(resp), net::redirect_error(net::use_awaitable, ec));
+      co_await conn->async_exec(req, adapt(resp), redir(ec));
       BOOST_TEST(!ec);
       BOOST_CHECK_EQUAL(msg, std::get<1>(resp));
       req.clear();
@@ -79,16 +72,13 @@ auto async_echo_stress() -> net::awaitable<void>
    for (int i = 0; i < sessions; ++i) 
       net::co_spawn(ex, echo_session(conn, std::to_string(i), msgs), net::detached);
 
-   auto const addrs = resolve();
-   co_await net::async_connect(conn->next_layer(), addrs);
+   co_await connect(conn, "127.0.0.1", "6379");
    co_await conn->async_run();
 }
 
 BOOST_AUTO_TEST_CASE(echo_stress)
 {
-   net::io_context ioc;
-   net::co_spawn(ioc, async_echo_stress(), net::detached);
-   ioc.run();
+   run(async_echo_stress());
 }
 
 #else

tests/conn_exec.cpp

@@ -123,16 +123,23 @@ BOOST_AUTO_TEST_CASE(cancel_request_if_not_connected)
    ioc.run();
 }
 
-BOOST_AUTO_TEST_CASE(request_retry)
+// TODO: This test is broken.
+BOOST_AUTO_TEST_CASE(request_retry_false)
 {
+   resp3::request req0;
+   req0.get_config().coalesce = false;
+   req0.get_config().cancel_on_connection_lost = true;
+   req0.push("HELLO", 3);
+
    resp3::request req1;
+   req1.get_config().coalesce = true;
    req1.get_config().cancel_on_connection_lost = true;
-   req1.push("HELLO", 3);
-   req1.push("CLIENT", "PAUSE", 7000);
+   req1.push("BLPOP", "any", 0);
 
    resp3::request req2;
+   req2.get_config().coalesce = true;
    req2.get_config().cancel_on_connection_lost = false;
-   req2.get_config().retry = false;
+   req2.get_config().retry_on_connection_lost = false;
    req2.push("PING");
 
    net::io_context ioc;
@@ -150,10 +157,14 @@ BOOST_AUTO_TEST_CASE(request_retry)
    auto const endpoints = resolve();
    net::connect(conn.next_layer(), endpoints);
 
-   conn.async_exec(req1, adapt(), [](auto ec, auto){
+   conn.async_exec(req0, adapt(), [](auto ec, auto){
       BOOST_TEST(!ec);
    });
 
+   conn.async_exec(req1, adapt(), [](auto ec, auto){
+      BOOST_CHECK_EQUAL(ec, boost::system::errc::errc_t::operation_canceled);
+   });
+
    conn.async_exec(req2, adapt(), [](auto ec, auto){
       BOOST_CHECK_EQUAL(ec, boost::system::errc::errc_t::operation_canceled);
    });
@@ -164,3 +175,70 @@ BOOST_AUTO_TEST_CASE(request_retry)
 
    ioc.run();
 }
+
+BOOST_AUTO_TEST_CASE(request_retry_true)
+{
+   resp3::request req0;
+   req0.get_config().coalesce = false;
+   req0.get_config().cancel_on_connection_lost = true;
+   req0.push("HELLO", 3);
+
+   resp3::request req1;
+   req1.get_config().coalesce = true;
+   req1.get_config().cancel_on_connection_lost = true;
+   req1.push("BLPOP", "any", 0);
+
+   resp3::request req2;
+   req2.get_config().coalesce = true;
+   req2.get_config().cancel_on_connection_lost = false;
+   req2.get_config().retry_on_connection_lost = true;
+   req2.push("PING");
+
+   resp3::request req3;
+   req3.get_config().coalesce = true;
+   req3.get_config().cancel_on_connection_lost = true;
+   req3.get_config().retry_on_connection_lost = false;
+   req3.push("QUIT");
+
+   net::io_context ioc;
+   connection conn{ioc};
+
+   net::steady_timer st{ioc};
+   st.expires_after(std::chrono::seconds{1});
+   st.async_wait([&](auto){
+      // Cancels the request before receiving the response. This
+      // should cause the second request to complete with error
+      // although it has cancel_on_connection_lost = false.
+      conn.cancel(aedis::operation::run);
+   });
+
+   auto const endpoints = resolve();
+   net::connect(conn.next_layer(), endpoints);
+
+   conn.async_exec(req0, adapt(), [](auto ec, auto){
+      BOOST_TEST(!ec);
+   });
+
+   conn.async_exec(req1, adapt(), [](auto ec, auto){
+      BOOST_CHECK_EQUAL(ec, boost::system::errc::errc_t::operation_canceled);
+   });
+
+   conn.async_exec(req2, adapt(), [&](auto ec, auto){
+      BOOST_TEST(!ec);
+      conn.async_exec(req3, adapt(), [&](auto ec, auto){
+         BOOST_TEST(!ec);
+      });
+   });
+
+   conn.async_run([&](auto ec){
+      BOOST_CHECK_EQUAL(ec, boost::system::errc::errc_t::operation_canceled);
+      conn.reset_stream();
+      net::connect(conn.next_layer(), endpoints);
+      conn.async_run([&](auto ec){
+         std::cout << ec.message() << std::endl;
+         BOOST_TEST(!ec);
+      });
+   });
+
+   ioc.run();
+}