Fix connect_timeout request_timeout not work + tsan in RPCServerTestSuite (#790)

Fixes #791

src/backend/cassandra/SettingsProvider.cpp

@@ -124,6 +124,14 @@ SettingsProvider::parseSettings() const
         "max_concurrent_requests_threshold",
         (settings.maxReadRequestsOutstanding + settings.maxWriteRequestsOutstanding) / settings.coreConnectionsPerHost);
 
+    auto const connectTimeoutSecond = config_.maybeValue<uint32_t>("connect_timeout");
+    if (connectTimeoutSecond)
+        settings.connectionTimeout = std::chrono::milliseconds{*connectTimeoutSecond * 1000};
+
+    auto const requestTimeoutSecond = config_.maybeValue<uint32_t>("request_timeout");
+    if (requestTimeoutSecond)
+        settings.requestTimeout = std::chrono::milliseconds{*requestTimeoutSecond * 1000};
+
     settings.certificate = parseOptionalCertificate();
     settings.username = config_.maybeValue<std::string>("username");
     settings.password = config_.maybeValue<std::string>("password");

src/backend/cassandra/impl/ExecutionStrategy.h

@@ -77,7 +77,7 @@ public:
     using ResultType = typename HandleType::ResultType;
     using CompletionTokenType = boost::asio::yield_context;
 
-    DefaultExecutionStrategy(Settings settings, HandleType const& handle)
+    DefaultExecutionStrategy(Settings const& settings, HandleType const& handle)
         : maxWriteRequestsOutstanding_{settings.maxWriteRequestsOutstanding}
         , maxReadRequestsOutstanding_{settings.maxReadRequestsOutstanding}
         , work_{ioc_}

unittests/util/MockRPCEngine.h

@@ -28,26 +28,18 @@
 
 struct MockAsyncRPCEngine
 {
-public:
-    MockAsyncRPCEngine()
-    {
-        work_.emplace(ioc_);  // make sure ctx does not stop on its own
-        runner_.emplace([this] { ioc_.run(); });
-    }
-
-    ~MockAsyncRPCEngine()
-    {
-        work_.reset();
-        ioc_.stop();
-        if (runner_->joinable())
-            runner_->join();
-    }
-
     template <typename Fn>
     bool
-    post(Fn&& func, std::string const& ip)
+    post(Fn&& func, [[maybe_unused]] std::string const& ip = "")
     {
-        boost::asio::spawn(ioc_, [handler = std::move(func)](auto yield) mutable { handler(yield); });
+        using namespace boost::asio;
+        io_context ioc;
+
+        spawn(ioc, [handler = std::forward<Fn>(func), _ = make_work_guard(ioc.get_executor())](auto yield) mutable {
+            handler(yield);
+        });
+
+        ioc.run();
         return true;
     }
 
@@ -62,16 +54,10 @@ public:
     MOCK_METHOD(void, notifyUnknownCommand, (), ());
     MOCK_METHOD(void, notifyInternalError, (), ());
     MOCK_METHOD(RPC::Result, buildResponse, (Web::Context const&), ());
-
-private:
-    boost::asio::io_context ioc_;
-    std::optional<boost::asio::io_service::work> work_;
-    std::optional<std::thread> runner_;
 };
 
 struct MockRPCEngine
 {
-public:
     MOCK_METHOD(bool, post, (std::function<void(boost::asio::yield_context)>&&, std::string const&), ());
     MOCK_METHOD(void, notifyComplete, (std::string const&, std::chrono::microseconds const&), ());
     MOCK_METHOD(void, notifyErrored, (std::string const&), ());

unittests/webserver/RPCServerHandlerTest.cpp

@@ -114,7 +114,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPDefaultPath)
     EXPECT_CALL(*etl, lastCloseAgeSeconds()).WillOnce(testing::Return(45));
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -149,7 +148,6 @@ TEST_F(WebRPCServerHandlerTest, WsNormalPath)
     EXPECT_CALL(*etl, lastCloseAgeSeconds()).WillOnce(testing::Return(45));
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -189,7 +187,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPForwardedPath)
     EXPECT_CALL(*etl, lastCloseAgeSeconds()).WillOnce(testing::Return(45));
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -232,7 +229,6 @@ TEST_F(WebRPCServerHandlerTest, WsForwardedPath)
     EXPECT_CALL(*etl, lastCloseAgeSeconds()).WillOnce(testing::Return(45));
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -279,7 +275,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPErrorPath)
     EXPECT_CALL(*etl, lastCloseAgeSeconds()).WillOnce(testing::Return(45));
 
     (*handler)(std::move(requestJSON), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -320,7 +315,6 @@ TEST_F(WebRPCServerHandlerTest, WsErrorPath)
     EXPECT_CALL(*etl, lastCloseAgeSeconds()).WillOnce(testing::Return(45));
 
     (*handler)(std::move(requestJSON), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -348,7 +342,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPNotReady)
     EXPECT_CALL(*rpcEngine, notifyNotReady).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -377,7 +370,6 @@ TEST_F(WebRPCServerHandlerTest, WsNotReady)
     EXPECT_CALL(*rpcEngine, notifyNotReady).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -398,7 +390,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPInvalidAPIVersion)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(session->message, response);
     EXPECT_EQ(session->lastStatus, boost::beast::http::status::bad_request);
 }
@@ -429,7 +420,6 @@ TEST_F(WebRPCServerHandlerTest, WSInvalidAPIVersion)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -457,7 +447,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPBadSyntaxWhenRequestSubscribe)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -473,7 +462,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPMissingCommand)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(session->message, response);
     EXPECT_EQ(session->lastStatus, boost::beast::http::status::bad_request);
 }
@@ -490,7 +478,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPCommandNotString)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(session->message, response);
     EXPECT_EQ(session->lastStatus, boost::beast::http::status::bad_request);
 }
@@ -507,7 +494,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPCommandIsEmpty)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(session->message, response);
     EXPECT_EQ(session->lastStatus, boost::beast::http::status::bad_request);
 }
@@ -539,7 +525,6 @@ TEST_F(WebRPCServerHandlerTest, WsMissingCommand)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -558,7 +543,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPParamsUnparseableNotArray)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(requestJSON), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(session->message, response);
     EXPECT_EQ(session->lastStatus, boost::beast::http::status::bad_request);
 }
@@ -578,7 +562,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPParamsUnparseableEmptyArray)
     EXPECT_CALL(*rpcEngine, notifyBadSyntax).Times(1);
 
     (*handler)(std::move(requestJSON), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(session->message, response);
     EXPECT_EQ(session->lastStatus, boost::beast::http::status::bad_request);
 }
@@ -611,7 +594,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPInternalError)
     EXPECT_CALL(*rpcEngine, buildResponse(testing::_)).Times(1).WillOnce(testing::Throw(std::runtime_error("MyError")));
 
     (*handler)(std::move(requestJSON), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -644,7 +626,6 @@ TEST_F(WebRPCServerHandlerTest, WsInternalError)
     EXPECT_CALL(*rpcEngine, buildResponse(testing::_)).Times(1).WillOnce(testing::Throw(std::runtime_error("MyError")));
 
     (*handler)(std::move(requestJSON), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -681,7 +662,6 @@ TEST_F(WebRPCServerHandlerTest, HTTPOutDated)
     EXPECT_CALL(*etl, lastCloseAgeSeconds()).WillOnce(testing::Return(61));
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }
 
@@ -721,7 +701,6 @@ TEST_F(WebRPCServerHandlerTest, WsOutdated)
     EXPECT_CALL(*etl, lastCloseAgeSeconds()).WillOnce(testing::Return(61));
 
     (*handler)(std::move(request), session);
-    std::this_thread::sleep_for(200ms);
     EXPECT_EQ(boost::json::parse(session->message), boost::json::parse(response));
 }