CDRIVER-4666 remove locks during single-threaded scanning (#1367)

* add a failing test for CDRIVER-4666

* assert single_threaded

* remove locks during single-threaded scanning

Fixes a possible recursive mutex lock in some error scenarios. Refer: CDRIVER-4666

Author: kevinAlbs

src/libmongoc/src/mongoc/mongoc-topology.c

@@ -73,6 +73,7 @@ mongoc_topology_reconcile (const mongoc_topology_t *topology,
    mongoc_server_description_t *sd;
    mongoc_topology_scanner_node_t *ele, *tmp;
 
+   BSON_ASSERT (topology->single_threaded);
    servers = mc_tpld_servers (td);
    /* Add newly discovered nodes */
    for (size_t i = 0u; i < servers->items_len; i++) {
@@ -124,15 +125,16 @@ _mongoc_topology_scanner_setup_err_cb (uint32_t id,
 {
    mongoc_topology_t *topology = BSON_ASSERT_PTR_INLINE (data);
 
+   BSON_ASSERT (topology->single_threaded);
    if (_mongoc_topology_get_type (topology) == MONGOC_TOPOLOGY_LOAD_BALANCED) {
       /* In load balanced mode, scanning is only for connection establishment.
        * It must not modify the topology description. */
    } else {
-      /* We need to update the topology description */
-      mc_tpld_modification mod = mc_tpld_modify_begin (topology);
+      // Use `mc_tpld_unsafe_get_mutable` to get a mutable topology description
+      // without locking. This function only applies to single-threaded clients.
+      mongoc_topology_description_t *td = mc_tpld_unsafe_get_mutable (topology);
       mongoc_topology_description_handle_hello (
-         mod.new_td, id, NULL /* hello reply */, -1 /* rtt_msec */, error);
-      mc_tpld_modify_commit (mod);
+         td, id, NULL /* hello reply */, -1 /* rtt_msec */, error);
    }
 }
 
@@ -159,51 +161,47 @@ _mongoc_topology_scanner_cb (uint32_t id,
 {
    mongoc_topology_t *const topology = BSON_ASSERT_PTR_INLINE (data);
    mongoc_server_description_t *sd;
-   mc_tpld_modification tdmod;
+   mongoc_topology_description_t *td;
 
+   BSON_ASSERT (topology->single_threaded);
    if (_mongoc_topology_get_type (topology) == MONGOC_TOPOLOGY_LOAD_BALANCED) {
       /* In load balanced mode, scanning is only for connection establishment.
        * It must not modify the topology description. */
       return;
    }
 
-   tdmod = mc_tpld_modify_begin (topology);
+   // Use `mc_tpld_unsafe_get_mutable` to get a mutable topology description
+   // without locking. This function only applies to single-threaded clients.
+   td = mc_tpld_unsafe_get_mutable (topology);
 
-   sd = mongoc_topology_description_server_by_id (tdmod.new_td, id, NULL);
+   sd = mongoc_topology_description_server_by_id (td, id, NULL);
 
    if (!hello_response) {
       /* Server monitoring: When a server check fails due to a network error
        * (including a network timeout), the client MUST clear its connection
        * pool for the server */
       _mongoc_topology_description_clear_connection_pool (
-         tdmod.new_td, id, &kZeroServiceId);
+         td, id, &kZeroServiceId);
    }
 
    /* Server Discovery and Monitoring Spec: "Once a server is connected, the
     * client MUST change its type to Unknown only after it has retried the
     * server once." */
    if (!hello_response && sd && sd->type != MONGOC_SERVER_UNKNOWN) {
-      _mongoc_topology_update_no_lock (
-         id, hello_response, rtt_msec, tdmod.new_td, error);
+      _mongoc_topology_update_no_lock (id, hello_response, rtt_msec, td, error);
 
       /* add another hello call to the current scan - the scan continues
        * until all commands are done */
       mongoc_topology_scanner_scan (topology->scanner, sd->id);
    } else {
-      _mongoc_topology_update_no_lock (
-         id, hello_response, rtt_msec, tdmod.new_td, error);
+      _mongoc_topology_update_no_lock (id, hello_response, rtt_msec, td, error);
 
       /* The processing of the hello results above may have added, changed, or
        * removed server descriptions. We need to reconcile that with our
        * monitoring agents
        */
-      mongoc_topology_reconcile (topology, tdmod.new_td);
-
-      mongoc_cond_broadcast (&topology->cond_client);
+      mongoc_topology_reconcile (topology, td);
    }
-
-
-   mc_tpld_modify_commit (tdmod);
 }
 
 static void
@@ -881,7 +879,8 @@ mongoc_topology_rescan_srv (mongoc_topology_t *topology)
 static void
 mongoc_topology_scan_once (mongoc_topology_t *topology, bool obey_cooldown)
 {
-   mc_tpld_modification tdmod;
+   mongoc_topology_description_t *td;
+   BSON_ASSERT (topology->single_threaded);
    if (mongoc_topology_should_rescan_srv (topology)) {
       /* Prior to scanning hosts, update the list of SRV hosts, if applicable.
        */
@@ -892,9 +891,10 @@ mongoc_topology_scan_once (mongoc_topology_t *topology, bool obey_cooldown)
     * description based on hello responses in connection handshakes, see
     * _mongoc_topology_update_from_handshake. retire scanner nodes for removed
     * members and create scanner nodes for new ones. */
-   tdmod = mc_tpld_modify_begin (topology);
-   mongoc_topology_reconcile (topology, tdmod.new_td);
-   mc_tpld_modify_commit (tdmod);
+   // Use `mc_tpld_unsafe_get_mutable` to get a mutable topology description
+   // without locking. This function only applies to single-threaded clients.
+   td = mc_tpld_unsafe_get_mutable (topology);
+   mongoc_topology_reconcile (topology, td);
 
    mongoc_topology_scanner_start (topology->scanner, obey_cooldown);
    mongoc_topology_scanner_work (topology->scanner);
@@ -920,6 +920,7 @@ void
 _mongoc_topology_do_blocking_scan (mongoc_topology_t *topology,
                                    bson_error_t *error)
 {
+   BSON_ASSERT (topology->single_threaded);
    _mongoc_handshake_freeze ();
 
    mongoc_topology_scan_once (topology, true /* obey cooldown */);
@@ -1162,9 +1163,11 @@ mongoc_topology_select_server_id (mongoc_topology_t *topology,
 
    if (topology->single_threaded) {
       if (!td.ptr->opened) {
-         mc_tpld_modification tdmod = mc_tpld_modify_begin (topology);
-         _mongoc_topology_description_monitor_opening (tdmod.new_td);
-         mc_tpld_modify_commit (tdmod);
+         // Use `mc_tpld_unsafe_get_mutable` to get a mutable topology
+         // description without locking. This block only applies to
+         // single-threaded clients.
+         _mongoc_topology_description_monitor_opening (
+            mc_tpld_unsafe_get_mutable (topology));
          mc_tpld_renew_ref (&td, topology);
       }
 

src/libmongoc/tests/test-mongoc-topology.c

@@ -2573,6 +2573,108 @@ test_hello_ok_pooled (void)
    _test_hello_ok (true);
 }
 
+// initiator_fail is a stream initiator that always fails.
+static mongoc_stream_t *
+initiator_fail (const mongoc_uri_t *uri,
+                const mongoc_host_list_t *host,
+                void *user_data,
+                bson_error_t *error)
+{
+   bson_set_error (error,
+                   MONGOC_ERROR_STREAM,
+                   MONGOC_ERROR_STREAM_CONNECT,
+                   "failing in initiator");
+   printf ("failing in initiator\n");
+   return false;
+}
+
+// Test failure in `mongoc_topology_scanner_node_setup` during retry of scanning
+// a known server. This is a regression test of CDRIVER-4666.
+static void
+test_failure_to_setup_after_retry (void)
+{
+   mock_server_t *server;
+   mongoc_uri_t *uri;
+   mongoc_client_t *client;
+   future_t *future;
+   bson_error_t error;
+
+   server = mock_server_new ();
+   mock_server_run (server);
+   uri = mongoc_uri_copy (mock_server_get_uri (server));
+   client = mongoc_client_new_from_uri_with_error (uri, &error);
+   ASSERT_OR_PRINT (client, error);
+
+   // Override the heartbeatFrequencyMS (default 60 seconds) and
+   // minHeartbeatFrequencyMS (default 500ms) to speed up the test.
+   const int64_t overridden_heartbeat_ms = 1;
+   {
+      mc_tpld_modification tdmod = mc_tpld_modify_begin (client->topology);
+      tdmod.new_td->heartbeat_msec = overridden_heartbeat_ms;
+      mc_tpld_modify_commit (tdmod);
+      client->topology->min_heartbeat_frequency_msec = overridden_heartbeat_ms;
+   }
+
+   future = future_client_command_simple (
+      client, "test", tmp_bson ("{'ping': 1}"), NULL, NULL, &error);
+
+   // The first command starts the first topology scan.
+   // Expect legacy hello with handshake.
+   {
+      request_t *request = mock_server_receives_legacy_hello (server, "{}");
+      char *reply = bson_strdup_printf ("{'ok': 1,"
+                                        " 'minWireVersion': %d,"
+                                        " 'maxWireVersion': %d }",
+                                        WIRE_VERSION_MIN,
+                                        WIRE_VERSION_MAX);
+
+      reply_to_request_simple (request, reply);
+      bson_free (reply);
+      request_destroy (request);
+   }
+
+   // Expect "ping" command.
+   {
+      request_t *request = mock_server_receives_msg (
+         server, MONGOC_MSG_NONE, tmp_bson ("{'ping': 1}"));
+      reply_to_request_with_ok_and_destroy (request);
+      ASSERT_OR_PRINT (future_get_bool (future), error);
+      future_destroy (future);
+   }
+
+   // Wait until ready for next topology scan.
+   _mongoc_usleep (overridden_heartbeat_ms * 1000);
+
+   // Send another command.
+   future = future_client_command_simple (
+      client, "test", tmp_bson ("{'ping': 1}"), NULL, NULL, &error);
+
+   // Expect legacy hello with handshake.
+   {
+      request_t *request = mock_server_receives_legacy_hello (server, "{}");
+      // Set the initiator to fail.
+      mongoc_client_set_stream_initiator (client, initiator_fail, NULL);
+      // A network error on a previously known server triggers the retry.
+      reply_to_request_with_hang_up (request);
+      request_destroy (request);
+   }
+
+   // The initiator fails in `mongoc_topology_scanner_node_setup`. Causes a
+   // deadlock similar to that observed in CDRIVER-4666.
+   // Test fails on macOS due to deadlock with "future_get_bool timed out".
+
+   ASSERT (!future_get_bool (future));
+   future_destroy (future);
+   ASSERT_ERROR_CONTAINS (error,
+                          MONGOC_ERROR_SERVER_SELECTION,
+                          MONGOC_ERROR_SERVER_SELECTION_FAILURE,
+                          "No suitable servers found");
+
+   mongoc_client_destroy (client);
+   mongoc_uri_destroy (uri);
+   mock_server_destroy (server);
+}
+
 void
 test_topology_install (TestSuite *suite)
 {
@@ -2737,4 +2839,7 @@ test_topology_install (TestSuite *suite)
       suite, "/Topology/hello_ok/single", test_hello_ok_single);
    TestSuite_AddMockServerTest (
       suite, "/Topology/hello_ok/pooled", test_hello_ok_pooled);
+   TestSuite_AddMockServerTest (suite,
+                                "/Topology/failure_to_setup_after_retry",
+                                test_failure_to_setup_after_retry);
 }