CDRIVER-894: Race condition in connecting to node

Author: bjori

src/mongoc/mongoc-cluster.c

@@ -1236,8 +1236,8 @@ _mongoc_cluster_add_node (mongoc_cluster_t *cluster,
                           bson_error_t *error /* OUT */)
 {
    mongoc_cluster_node_t *cluster_node;
-
    mongoc_stream_t *stream;
+   int64_t expire_at;
 
    ENTRY;
 
@@ -1252,6 +1252,18 @@ _mongoc_cluster_add_node (mongoc_cluster_t *cluster,
       RETURN (NULL);
    }
 
+   expire_at = bson_get_monotonic_time() + cluster->client->topology->connect_timeout_msec * 1000;
+   if (!mongoc_stream_wait (stream, expire_at)) {
+      bson_set_error (error,
+                      MONGOC_ERROR_STREAM,
+                      MONGOC_ERROR_STREAM_CONNECT,
+                      "Failed to connect to target host: '%s'",
+                      sd->host.host_and_port);
+      memcpy (&sd->error, error, sizeof sd->error);
+      mongoc_stream_failed (stream);
+      return NULL;
+   }
+
    /* take critical fields from a fresh ismaster */
    cluster_node = _mongoc_cluster_node_new (stream);
    if (!_mongoc_cluster_run_ismaster (cluster, cluster_node)) {
@@ -1396,6 +1408,7 @@ mongoc_cluster_fetch_stream_single (mongoc_cluster_t *cluster,
 {
    mongoc_stream_t *stream = NULL;
    mongoc_topology_scanner_node_t *scanner_node;
+   int64_t expire_at;
    bson_t reply;
 
    scanner_node = mongoc_topology_scanner_get_node (topology->scanner, sd->id);
@@ -1411,8 +1424,20 @@ mongoc_cluster_fetch_stream_single (mongoc_cluster_t *cluster,
       if (!mongoc_topology_scanner_node_setup (scanner_node, error)) {
          return NULL;
       }
-
       stream = scanner_node->stream;
+
+      expire_at = bson_get_monotonic_time() + topology->connect_timeout_msec * 1000;
+      if (!mongoc_stream_wait (stream, expire_at)) {
+         bson_set_error (error,
+                         MONGOC_ERROR_STREAM,
+                         MONGOC_ERROR_STREAM_CONNECT,
+                         "Failed to connect to target host: '%s'",
+                         sd->host.host_and_port);
+         memcpy (&sd->error, error, sizeof sd->error);
+         mongoc_topology_scanner_node_disconnect (scanner_node, true);
+         return NULL;
+      }
+
       if (!_mongoc_stream_run_ismaster (cluster, stream, &reply, error)) {
          mongoc_topology_scanner_node_disconnect (scanner_node, true);
          return NULL;

src/mongoc/mongoc-stream-private.h

@@ -34,6 +34,10 @@ BSON_BEGIN_DECLS
 #define MONGOC_STREAM_GRIDFS   4
 #define MONGOC_STREAM_TLS      5
 
+bool
+mongoc_stream_wait (mongoc_stream_t *stream,
+                    int64_t expire_at);
+
 bool
 _mongoc_stream_writev_full (mongoc_stream_t *stream,
                             mongoc_iovec_t  *iov,

src/mongoc/mongoc-stream.c

@@ -357,6 +357,86 @@ mongoc_stream_poll (mongoc_stream_poll_t *streams,
 }
 
 
+/*
+ *--------------------------------------------------------------------------
+ *
+ * mongoc_stream_wait --
+ *
+ *       Internal helper, poll a single stream until it connects.
+ *
+ *       For now, only the POLLOUT (connected) event is supported.
+ *
+ *       @expire_at should be an absolute time at which to expire using
+ *       the monotonic clock (bson_get_monotonic_time(), which is in
+ *       microseconds). expire_at of 0 or -1 is prohibited.
+ *
+ * Returns:
+ *       true if an event matched. otherwise false.
+ *       a timeout will return false.
+ *
+ * Side effects:
+ *       None.
+ *
+ *--------------------------------------------------------------------------
+ */
+
+bool
+mongoc_stream_wait (mongoc_stream_t *stream,
+                    int64_t expire_at)
+{
+   mongoc_stream_poll_t poller;
+   int64_t now;
+   int32_t timeout_msec;
+   ssize_t ret;
+
+   ENTRY;
+
+   BSON_ASSERT (stream);
+   BSON_ASSERT (expire_at > 0);
+
+   poller.stream = stream;
+   poller.events = POLLOUT;
+   poller.revents = 0;
+
+   now = bson_get_monotonic_time();
+
+   for (;;) {
+      /* TODO CDRIVER-804 use int64_t for timeouts consistently */
+      timeout_msec = (int32_t) BSON_MIN ((expire_at - now) / 1000L, INT32_MAX);
+      if (timeout_msec < 0) {
+         timeout_msec = 0;
+      }
+
+      ret = mongoc_stream_poll (&poller, 1, timeout_msec);
+
+      if (ret > 0) {
+         /* an event happened, return true if POLLOUT else false */
+         RETURN (0 != (poller.revents & POLLOUT));
+      } else if (ret < 0) {
+         /* poll itself failed */
+
+         TRACE("errno is: %d", errno);
+         if (MONGOC_ERRNO_IS_AGAIN(errno)) {
+            now = bson_get_monotonic_time();
+
+            if (expire_at < now) {
+               RETURN (false);
+            } else {
+               continue;
+            }
+         } else {
+            /* poll failed for some non-transient reason */
+            RETURN (false);
+         }
+      } else {
+         /* poll timed out */
+         RETURN (false);
+      }
+   }
+
+   return true;
+}
+
 bool
 mongoc_stream_check_closed (mongoc_stream_t *stream)
 {