lib/message_passing: make sync first connection an option

GIT_ORIGIN_SPP_REV_ID: 1d2daddd4bac1f53306f56451876efec19818d3a

Author: kitsnet

score/message_passing/client_connection.cpp

@@ -334,7 +334,20 @@ void ClientConnection::DoRestart() noexcept
               << std::endl;
 
     connect_retry_ms_ = kConnectRetryMsStart;
-    TryConnect();
+    if (client_config_.sync_first_connect)
+    {
+        TryConnect();
+    }
+    else
+    {
+        engine_->EnqueueCommand(
+            connection_timer_,
+            ISharedResourceEngine::TimePoint{},
+            [this](auto) noexcept {
+                TryConnect();
+            },
+            this);
+    }
 }
 
 void ClientConnection::TryConnect() noexcept

score/message_passing/client_connection_test.cpp

@@ -65,15 +65,22 @@ class ClientConnectionTest : public ::testing::Test
         engine_.reset();
     }
 
-    void CatchImmediateConnectEndpointRegistrationCommand()
+    void CatchImmediateConnectCommand()
     {
         EXPECT_CALL(*engine_, EnqueueCommand(_, ISharedResourceEngine::TimePoint{}, _, _))
             .WillOnce([&](auto&, auto, ISharedResourceEngine::CommandCallback callback, auto) {
-                // reusing the connect callback in case ow queued endpoint registration
                 connect_command_callback_ = std::move(callback);
             });
     }
 
+    void CatchEndpointRegistrationCommand()
+    {
+        EXPECT_CALL(*engine_, EnqueueCommand(_, ISharedResourceEngine::TimePoint{}, _, _))
+            .WillOnce([&](auto&, auto, ISharedResourceEngine::CommandCallback callback, auto) {
+                endpoint_command_callback_ = std::move(callback);
+            });
+    }
+
     void CatchTimedConnectCommand()
     {
         EXPECT_CALL(*engine_, EnqueueCommand(_, Gt(ISharedResourceEngine::Clock::now()), _, _))
@@ -126,6 +133,17 @@ class ClientConnectionTest : public ::testing::Test
         on_callback_thread_ = false;
     }
 
+    void InvokeEndpointRegistrationCommand()
+    {
+        on_callback_thread_ = true;
+        {
+            auto callback = std::move(endpoint_command_callback_);
+            callback(ISharedResourceEngine::Clock::now());
+            // the destructor for callback capture, if any, is called now
+        }
+        on_callback_thread_ = false;
+    }
+
     void InvokeDisconnectCommand()
     {
         on_callback_thread_ = true;
@@ -200,13 +218,26 @@ class ClientConnectionTest : public ::testing::Test
                                   IClientConnection::StateCallback state_callback = {},
                                   IClientConnection::NotifyCallback notify_callback = {})
     {
-        AtTryOpenCall_Return(kValidFd);
-        CatchImmediateConnectEndpointRegistrationCommand();
+        if (client_config_.sync_first_connect)
+        {
+            AtTryOpenCall_Return(kValidFd);
+            CatchEndpointRegistrationCommand();
 
-        connection.Start(std::move(state_callback), std::move(notify_callback));
+            connection.Start(std::move(state_callback), std::move(notify_callback));
 
-        CatchPosixEndpointRegistration();
-        InvokeConnectCommand();
+            CatchPosixEndpointRegistration();
+            InvokeEndpointRegistrationCommand();
+        }
+        else
+        {
+            CatchImmediateConnectCommand();
+            connection.Start(std::move(state_callback), std::move(notify_callback));
+            EXPECT_EQ(connection.GetState(), State::kStarting);
+
+            AtTryOpenCall_Return(kValidFd);
+            CatchPosixEndpointRegistration();
+            InvokeConnectCommand();
+        }
 
         EXPECT_EQ(connection.GetState(), State::kReady);
         EXPECT_EQ(posix_endpoint_->fd, kValidFd);
@@ -216,12 +247,20 @@ class ClientConnectionTest : public ::testing::Test
 
     void StopConnectionInProgress(detail::ClientConnection& connection)
     {
-        CatchDisconnectCommand();
-        connection.Stop();
-        EXPECT_EQ(connection.GetState(), State::kStopping);
-        EXPECT_EQ(connection.GetStopReason(), StopReason::kUserRequested);
+        if (client_config_.sync_first_connect)
+        {
+            // for the immediate stop on the user thread
+            CatchDisconnectCommand();
+            connection.Stop();
+            EXPECT_EQ(connection.GetState(), State::kStopping);
+            EXPECT_EQ(connection.GetStopReason(), StopReason::kUserRequested);
 
-        InvokeDisconnectCommand();
+            InvokeDisconnectCommand();
+        }
+        else
+        {
+            connection.Stop();
+        }
         EXPECT_EQ(connection.GetState(), State::kStopped);
         EXPECT_EQ(connection.GetStopReason(), StopReason::kUserRequested);
     }
@@ -259,8 +298,9 @@ class ClientConnectionTest : public ::testing::Test
     std::shared_ptr<StrictMock<SharedResourceEngineMock>> engine_{};
     const std::string service_identifier_{"test_identifier"};
     const ServiceProtocolConfig protocol_config_{service_identifier_, kMaxSendSize, kMaxReplySize, kMaxNotifySize};
-    IClientFactory::ClientConfig client_config_{0, 1, false, false};
+    IClientFactory::ClientConfig client_config_{0, 1, false, false, false};
     ISharedResourceEngine::CommandCallback connect_command_callback_{};
+    ISharedResourceEngine::CommandCallback endpoint_command_callback_{};
     ISharedResourceEngine::CommandCallback disconnect_command_callback_{};
     ISharedResourceEngine::CommandCallback send_queue_command_callback_{};
     ISharedResourceEngine::PosixEndpointEntry* posix_endpoint_{};
@@ -287,6 +327,27 @@ TEST_F(ClientConnectionTest, TryingToConnectOnceStoppingWhileConnecting)
 {
     detail::ClientConnection connection(engine_, protocol_config_, client_config_);
 
+    CatchImmediateConnectCommand();
+    connection.Start(IClientConnection::StateCallback(), IClientConnection::NotifyCallback());
+    EXPECT_EQ(connection.GetState(), State::kStarting);
+
+    EXPECT_CALL(*engine_, TryOpenClientConnection(std::string_view{service_identifier_}))
+        .WillOnce([this, &connection](auto&&) {
+            StopConnectionInProgress(connection);
+            return score::cpp::make_unexpected(score::os::Error::createFromErrno(EIO));  // this error code would be ignored
+        });
+    ExpectCleanUpOwner(connection);
+    InvokeConnectCommand();
+
+    EXPECT_EQ(connection.GetState(), State::kStopped);
+    EXPECT_EQ(connection.GetStopReason(), StopReason::kUserRequested);
+}
+
+TEST_F(ClientConnectionTest, TryingToSyncConnectOnceStoppingWhileConnecting)
+{
+    client_config_.sync_first_connect = true;
+    detail::ClientConnection connection(engine_, protocol_config_, client_config_);
+
     EXPECT_CALL(*engine_, TryOpenClientConnection(std::string_view{service_identifier_}))
         .WillOnce([this, &connection](auto&&) {
             StopConnectionInProgress(connection);
@@ -301,6 +362,30 @@ TEST_F(ClientConnectionTest, TryingToConnectOnceStoppingWhileConnecting)
 
 TEST_F(ClientConnectionTest, TryingToConnectOnceStoppingOnHardError)
 {
+    const auto error_list_to_test = {std::make_pair(EACCES, StopReason::kPermission),
+                                     std::make_pair(EPIPE, StopReason::kIoError)};
+
+    for (auto entry : error_list_to_test)
+    {
+        detail::ClientConnection connection(engine_, protocol_config_, client_config_);
+
+        CatchImmediateConnectCommand();
+        connection.Start(IClientConnection::StateCallback(), IClientConnection::NotifyCallback());
+        EXPECT_EQ(connection.GetState(), State::kStarting);
+
+        AtTryOpenCall_Return(score::cpp::make_unexpected(score::os::Error::createFromErrno(entry.first)));
+        ExpectCleanUpOwner(connection);
+        InvokeConnectCommand();
+
+        EXPECT_EQ(connection.GetState(), State::kStopped);
+        EXPECT_EQ(connection.GetStopReason(), entry.second);
+    }
+}
+
+TEST_F(ClientConnectionTest, TryingToSyncConnectOnceStoppingOnHardError)
+{
+    client_config_.sync_first_connect = true;
+
     const auto error_list_to_test = {std::make_pair(EACCES, StopReason::kPermission),
                                      std::make_pair(EPIPE, StopReason::kIoError)};
 
@@ -323,6 +408,38 @@ TEST_F(ClientConnectionTest, TryingToConnectMultipleTimesStoppingOnPermissionErr
 
     detail::ClientConnection connection(engine_, protocol_config_, client_config_);
 
+    for (auto entry : error_list_to_test)
+    {
+        if (entry == *error_list_to_test.begin())
+        {
+            CatchImmediateConnectCommand();
+            connection.Start(IClientConnection::StateCallback(), IClientConnection::NotifyCallback());
+        }
+        else
+        {
+            AtTryOpenCall_Return(score::cpp::make_unexpected(score::os::Error::createFromErrno(entry)));
+            CatchTimedConnectCommand();
+            InvokeConnectCommand();
+        }
+        EXPECT_EQ(connection.GetState(), State::kStarting);
+    }
+
+    AtTryOpenCall_Return(score::cpp::make_unexpected(score::os::Error::createFromErrno(EACCES)));
+    ExpectCleanUpOwner(connection);
+    InvokeConnectCommand();
+
+    EXPECT_EQ(connection.GetState(), State::kStopped);
+    EXPECT_EQ(connection.GetStopReason(), StopReason::kPermission);
+}
+
+TEST_F(ClientConnectionTest, TryingToSyncConnectMultipleTimesStoppingOnPermissionError)
+{
+    client_config_.sync_first_connect = true;
+
+    const auto error_list_to_test = {EAGAIN, ECONNREFUSED, ENOENT};
+
+    detail::ClientConnection connection(engine_, protocol_config_, client_config_);
+
     for (auto entry : error_list_to_test)
     {
         AtTryOpenCall_Return(score::cpp::make_unexpected(score::os::Error::createFromErrno(entry)));
@@ -350,8 +467,7 @@ TEST_F(ClientConnectionTest, TryingToConnectMultipleTimesFinallyConnectingAndImp
 {
     detail::ClientConnection connection(engine_, protocol_config_, client_config_);
 
-    AtTryOpenCall_Return(score::cpp::make_unexpected(score::os::Error::createFromErrno(ENOENT)));
-    CatchTimedConnectCommand();
+    CatchImmediateConnectCommand();
     connection.Start(IClientConnection::StateCallback(), IClientConnection::NotifyCallback());
     EXPECT_EQ(connection.GetState(), State::kStarting);
 
@@ -383,6 +499,15 @@ TEST_F(ClientConnectionTest, SuccessfullyConnectingAtFirstAttemptThenExplicitlyS
     StopCurrentConnection(connection);
 }
 
+TEST_F(ClientConnectionTest, SuccessfullySyncConnectingAtFirstAttemptThenExplicitlyStopping)
+{
+    client_config_.sync_first_connect = true;
+    detail::ClientConnection connection(engine_, protocol_config_, client_config_);
+    MakeSuccessfulConnection(connection);
+
+    StopCurrentConnection(connection);
+}
+
 TEST_F(ClientConnectionTest, SuccessfullyConnectingAtFirstAttemptThenExplicitlyStoppingAsIfFromCallback)
 {
     detail::ClientConnection connection(engine_, protocol_config_, client_config_);
@@ -391,6 +516,15 @@ TEST_F(ClientConnectionTest, SuccessfullyConnectingAtFirstAttemptThenExplicitlyS
     StopCurrentConnectionAsIfInCallback(connection);
 }
 
+TEST_F(ClientConnectionTest, SuccessfullySyncConnectingAtFirstAttemptThenExplicitlyStoppingAsIfFromCallback)
+{
+    client_config_.sync_first_connect = true;
+    detail::ClientConnection connection(engine_, protocol_config_, client_config_);
+    MakeSuccessfulConnection(connection);
+
+    StopCurrentConnectionAsIfInCallback(connection);
+}
+
 TEST_F(ClientConnectionTest, SuccessfullyConnectingAtFirstAttemptThenFirstReadDisconnected)
 {
     detail::ClientConnection connection(engine_, protocol_config_, client_config_);

score/message_passing/i_client_factory.h

@@ -52,6 +52,9 @@ class IClientFactory
         // coverity[autosar_cpp14_a9_6_1_violation : FALSE]
         bool truly_async;  ///< true if Send and SendWithCallback calls always use background thread
                            ///< for IPC (requires nonzero max_queued_sends)
+        // coverity[autosar_cpp14_a9_6_1_violation : FALSE]
+        bool sync_first_connect;  ///< true if the first connection attempt uses the thread on which Start() is called
+                                  ///< (can lead to deadlocks if the connection is established from within a callback)
     };
 
     /// \brief Creates an implementation instance of IClientConnection.

score/message_passing/qnx_dispatch_server_to_client_test.cpp

@@ -50,7 +50,7 @@ class ServerToClientQnxFixture : public ::testing::Test, public testing::WithPar
         test_prefix += std::to_string(::getpid()) + "_";
         service_identifier_ = test_prefix + "1";
         protocol_config_ = ServiceProtocolConfig{service_identifier_, 6U, 6U, 6U};
-        client_config_ = IClientFactory::ClientConfig{1U, 1U, false, true};
+        client_config_ = IClientFactory::ClientConfig{1U, 1U, false, true, false};
         server_config_ = IServerFactory::ServerConfig{0U, 0U, 1U};
 
         server_connections_started_ = 0U;

score/message_passing/unix_domain_server_to_client_test.cpp

@@ -50,7 +50,7 @@ class ServerToClientTestFixtureUnix : public ::testing::Test, public testing::Wi
         test_prefix += std::to_string(::getpid()) + "_";
         service_identifier_ = test_prefix + "1";
         protocol_config_ = ServiceProtocolConfig{service_identifier_, 1024, 1024, 1024};
-        client_config_ = IClientFactory::ClientConfig{1, 1, false, true};
+        client_config_ = IClientFactory::ClientConfig{1, 1, false, true, false};
 
         server_connections_started_ = 0;
         server_connections_finished_ = 0;

score/mw/com/impl/bindings/lola/messaging/message_passing_client_cache.cpp

@@ -77,7 +77,7 @@ std::shared_ptr<score::message_passing::IClientConnection> MessagePassingClientC
 
     const bool fully_async = asil_level_ == ClientQualityType::kASIL_QMfromB;
     const score::message_passing::ServiceProtocolConfig protocol_config{service_identifier, kMaxSendSize, 0U, 0U};
-    const score::message_passing::IClientFactory::ClientConfig client_config{0U, 20U, false, fully_async};
+    const score::message_passing::IClientFactory::ClientConfig client_config{0U, 20U, false, fully_async, true};
 
     auto new_sender_unique_p = client_factory_.Create(protocol_config, client_config);