CDRIVER-1571 slow initiator cancels topology scan

Async ismaster commands started before a slow call to a blocking stream
initiator had the initiator's duration subtracted from their timeouts.
The initiator and the async commands use the same initial timeout value,
so once an initiator times out, all async commands started beforehand
now have 0 seconds remaining. They're canceled before they run.

This change relies on the overall timeout applied to mongoc_async_run
instead of per-command timeouts.

Author: ajdavis

src/mongoc/mongoc-async-cmd-private.h

@@ -55,7 +55,7 @@ typedef struct _mongoc_async_cmd
    void                    *data;
    bson_error_t             error;
    int64_t                  start_time;
-   int64_t                  expire_at;
+   int64_t                  timeout_msec;
    bson_t                   cmd;
    mongoc_buffer_t          buffer;
    mongoc_array_t           array;

src/mongoc/mongoc-async-cmd.c

@@ -170,16 +170,14 @@ mongoc_async_cmd_new (mongoc_async_t           *async,
                       int32_t                   timeout_msec)
 {
    mongoc_async_cmd_t *acmd;
-   mongoc_async_cmd_t *tmp;
-   bool found = false;
 
    BSON_ASSERT (cmd);
    BSON_ASSERT (dbname);
    BSON_ASSERT (stream);
 
    acmd = (mongoc_async_cmd_t *)bson_malloc0 (sizeof (*acmd));
    acmd->async = async;
-   acmd->expire_at = bson_get_monotonic_time () + (int64_t) timeout_msec * 1000;
+   acmd->timeout_msec = timeout_msec;
    acmd->stream = stream;
    acmd->setup = setup;
    acmd->setup_ctx = setup_ctx;
@@ -194,22 +192,8 @@ mongoc_async_cmd_new (mongoc_async_t           *async,
 
    _mongoc_async_cmd_state_start (acmd);
 
-   /* slot the cmd into the right place in the expiration list */
-   {
-      async->ncmds++;
-      DL_FOREACH (async->cmds, tmp)
-      {
-         if (tmp->expire_at >= acmd->expire_at) {
-            DL_PREPEND_ELEM (async->cmds, tmp, acmd);
-            found = true;
-            break;
-         }
-      }
-
-      if (! found) {
-         DL_APPEND (async->cmds, acmd);
-      }
-   }
+   async->ncmds++;
+   DL_APPEND (async->cmds, acmd);
 
    return acmd;
 }
@@ -238,16 +222,12 @@ mongoc_async_cmd_destroy (mongoc_async_cmd_t *acmd)
 mongoc_async_cmd_result_t
 _mongoc_async_cmd_phase_setup (mongoc_async_cmd_t *acmd)
 {
-   int64_t now;
-   int64_t timeout_msec;
-
-   now = bson_get_monotonic_time ();
-   timeout_msec = (acmd->expire_at - now) / 1000;
-
-   BSON_ASSERT (timeout_msec < INT32_MAX);
+   int retval;
 
-   switch (acmd->setup (acmd->stream, &acmd->events, acmd->setup_ctx,
-                        (int32_t) timeout_msec, &acmd->error)) {
+   BSON_ASSERT (acmd->timeout_msec < INT32_MAX);
+   retval = acmd->setup (acmd->stream, &acmd->events, acmd->setup_ctx,
+                         (int32_t) acmd->timeout_msec, &acmd->error);
+   switch (retval) {
       case -1:
          return MONGOC_ASYNC_CMD_ERROR;
          break;

src/mongoc/mongoc-async.c

@@ -67,42 +67,21 @@ mongoc_async_run (mongoc_async_t *async,
    mongoc_async_cmd_t *acmd, *tmp;
    mongoc_stream_poll_t *poller = NULL;
    int i;
-   ssize_t nactive = 0;
+   ssize_t nactive;
    int64_t now;
-   int64_t expire_at = 0;
+   int64_t expire_at;
+   int64_t poll_timeout_msec;
+   size_t poll_size;
 
-   size_t poll_size = 0;
+   BSON_ASSERT (timeout_msec > 0);
 
-   for (;;) {
-      now = bson_get_monotonic_time ();
-
-      if (expire_at == 0) {
-         if (timeout_msec >= 0) {
-            expire_at = now + ((int64_t) timeout_msec * 1000);
-         } else {
-            expire_at = -1;
-         }
-      } else if (timeout_msec >= 0) {
-         timeout_msec = (expire_at - now) / 1000;
-      }
-
-      if (now > expire_at) {
-         break;
-      }
-
-      DL_FOREACH_SAFE (async->cmds, acmd, tmp)
-      {
-         /* async commands are sorted by expire_at */
-         if (now > acmd->expire_at) {
-            acmd->cb (MONGOC_ASYNC_CMD_TIMEOUT, NULL, (now - acmd->start_time), acmd->data,
-                      &acmd->error);
-            mongoc_async_cmd_destroy (acmd);
-         } else {
-            break;
-         }
-      }
+   now = bson_get_monotonic_time ();
+   expire_at = now + ((int64_t) timeout_msec * 1000);
+   poll_size = 0;
 
+   for (;;) {
       if (!async->ncmds) {
+         /* work complete */
          break;
       }
 
@@ -120,13 +99,10 @@ mongoc_async_run (mongoc_async_t *async,
          i++;
       }
 
-      if (timeout_msec >= 0) {
-         timeout_msec = BSON_MIN (timeout_msec, (async->cmds->expire_at - now) / 1000);
-      } else {
-         timeout_msec = (async->cmds->expire_at - now) / 1000;
-      }
-
-      nactive = mongoc_stream_poll (poller, async->ncmds, timeout_msec);
+      poll_timeout_msec = (expire_at - now) / 1000;
+      BSON_ASSERT (poll_timeout_msec < INT32_MAX);
+      nactive = mongoc_stream_poll (poller, async->ncmds,
+                                    (int32_t) poll_timeout_msec);
 
       if (nactive) {
          i = 0;
@@ -151,11 +127,25 @@ mongoc_async_run (mongoc_async_t *async,
             i++;
          }
       }
+
+      now = bson_get_monotonic_time ();
+      if (now > expire_at) {
+         break;
+      }
    }
 
    if (poll_size) {
       bson_free (poller);
    }
 
-   return async->ncmds;
+   /* commands that succeeded or failed already have been removed from the
+    * list and freed. therefore, all remaining commands have timed out. */
+   DL_FOREACH_SAFE (async->cmds, acmd, tmp) {
+      acmd->cb (MONGOC_ASYNC_CMD_TIMEOUT, NULL, (now - acmd->start_time),
+                acmd->data, &acmd->error);
+      mongoc_async_cmd_destroy (acmd);
+   }
+
+   /* cancel the loop in the caller, mongoc_topology_scanner_work() */
+   return false;
 }

tests/test-mongoc-topology-scanner.c

@@ -6,6 +6,7 @@
 #include "mongoc-tests.h"
 #include "TestSuite.h"
 #include "mock_server/mock-server.h"
+#include "mock_server/mock-rs.h"
 #include "mock_server/future.h"
 #include "mock_server/future-functions.h"
 
@@ -296,6 +297,126 @@ test_topology_scanner_oscillate ()
 }
 
 
+/* skip on Windows: https://daniel.haxx.se/blog/2012/10/10/wsapoll-is-broken/ */
+#ifndef _WIN32
+void
+test_topology_scanner_connection_error (void)
+{
+   mongoc_client_t *client;
+   bson_error_t error;
+
+   /* assuming nothing is listening on this port */
+   client = mongoc_client_new (
+      "mongodb://localhost:9876/?connectTimeoutMS=10");
+
+   ASSERT (!mongoc_client_command_simple (client, "db", tmp_bson ("{'foo': 1}"),
+                                          NULL, NULL, &error));
+
+   ASSERT_ERROR_CONTAINS (error, MONGOC_ERROR_SERVER_SELECTION,
+                          MONGOC_ERROR_SERVER_SELECTION_FAILURE,
+                          "connection error calling ismaster on "
+                          "'localhost:9876'");
+
+   mongoc_client_destroy (client);
+}
+#endif
+
+
+void
+test_topology_scanner_connection_timeout (void)
+{
+   mock_server_t *server;
+   mongoc_client_t *client;
+   mongoc_uri_t *uri;
+   bson_error_t error;
+   char *expected_msg;
+
+   /* server does NOT automatically reply to ismaster */
+   server = mock_server_new ();
+   mock_server_run (server);
+
+   uri = mongoc_uri_copy (mock_server_get_uri (server));
+   mongoc_uri_set_option_as_int32 (uri, "connectTimeoutMS", 10);
+   client = mongoc_client_new_from_uri (uri);
+
+   ASSERT (!mongoc_client_command_simple (client, "db", tmp_bson ("{'foo': 1}"),
+                                          NULL, NULL, &error));
+
+   expected_msg = bson_strdup_printf (
+      "connection timeout calling ismaster on '%s'",
+      mongoc_uri_get_hosts (uri)->host_and_port);
+
+   ASSERT_ERROR_CONTAINS (error, MONGOC_ERROR_SERVER_SELECTION,
+                          MONGOC_ERROR_SERVER_SELECTION_FAILURE,
+                          expected_msg);
+
+   bson_free (expected_msg);
+   mongoc_client_destroy (client);
+   mongoc_uri_destroy (uri);
+   mock_server_destroy (server);
+}
+
+
+typedef struct
+{
+   uint16_t         slow_port;
+   mongoc_client_t *client;
+} initiator_data_t;
+
+
+static mongoc_stream_t *
+slow_initiator (const mongoc_uri_t       *uri,
+                const mongoc_host_list_t *host,
+                void                     *user_data,
+                bson_error_t             *err)
+{
+   initiator_data_t *data;
+
+   data = (initiator_data_t *) user_data;
+
+   if (host->port == data->slow_port) {
+      _mongoc_usleep (500 * 1000);  /* 500 ms is longer than connectTimeoutMS */
+   }
+
+   return mongoc_client_default_stream_initiator (uri, host, data->client, err);
+}
+
+
+static void
+test_topology_scanner_blocking_initiator (void)
+{
+   mock_rs_t *rs;
+   mongoc_uri_t *uri;
+   mongoc_client_t *client;
+   initiator_data_t data;
+   bson_error_t error;
+
+   rs = mock_rs_with_autoismaster (0,    /* wire version   */
+                                   true, /* has primary    */
+                                   1,    /* n_secondaries  */
+                                   0     /* n_arbiters     */);
+
+   mock_rs_run (rs);
+   uri = mongoc_uri_copy (mock_rs_get_uri (rs));
+   mongoc_uri_set_option_as_int32 (uri, "connectTimeoutMS", 100);
+   client = mongoc_client_new_from_uri (uri);
+
+   /* pretend last host in linked list is slow */
+   data.slow_port = mongoc_uri_get_hosts (uri)->next->port;
+   data.client = client;
+   mongoc_client_set_stream_initiator (client, slow_initiator, &data);
+
+   ASSERT_OR_PRINT (mongoc_client_command_simple (client, "admin",
+                                                  tmp_bson ("{'ismaster': 1}"),
+                                                  NULL,
+                                                  NULL, &error), error);
+
+   mongoc_client_destroy (client);
+   mongoc_uri_destroy (uri);
+   mock_rs_destroy (rs);
+}
+
+
 void
 test_topology_scanner_install (TestSuite *suite)
 {
@@ -307,4 +428,12 @@ test_topology_scanner_install (TestSuite *suite)
                   test_topology_scanner_discovery);
    TestSuite_Add (suite, "/TOPOLOGY/scanner_oscillate",
                   test_topology_scanner_oscillate);
+#ifndef _WIN32
+   TestSuite_Add (suite, "/TOPOLOGY/scanner_connection_error",
+                  test_topology_scanner_connection_error);
+#endif
+   TestSuite_Add (suite, "/TOPOLOGY/scanner_connection_timeout",
+                  test_topology_scanner_connection_timeout);
+   TestSuite_Add (suite, "/TOPOLOGY/blocking_initiator",
+                  test_topology_scanner_blocking_initiator);
 }