CDRIVER-830 do blocking reconnect if single-threaded

The new mongoc_cluster_node_reconnect initiates a non-blocking reconnect
but doesn't wait for connection to complete. If the driver is in single-
threaded mode (so the scanner and the application share a connection)
and immediately uses that server, connection hasn't completed:

cursor error: Failure during socket delivery: Socket not connected (57)

With this change, apply connectTimeoutMS to the reconnection in single-
threaded mode.

Author: ajdavis

src/mongoc/mongoc-cluster.c

@@ -1241,6 +1241,8 @@ mongoc_cluster_node_reconnect (mongoc_cluster_t *cluster, uint32_t server_id, bs
    mongoc_topology_t *topology = cluster->client->topology;
    mongoc_cluster_node_t *cluster_node;
    mongoc_server_description_t *sd = NULL;
+   int64_t expire_at;
+
    ENTRY;
 
    bson_return_val_if_fail(cluster, false);
@@ -1259,7 +1261,24 @@ mongoc_cluster_node_reconnect (mongoc_cluster_t *cluster, uint32_t server_id, bs
 
       mongoc_stream_failed (scanner_node->stream);
       scanner_node->stream = NULL;
-      mongoc_topology_scanner_node_setup (scanner_node, error);
+
+      /* begin a non-blocking connect, then await connection */
+      if (!mongoc_topology_scanner_node_setup (scanner_node, error)) {
+         RETURN(false);
+      }
+
+      expire_at = bson_get_monotonic_time ()
+                  + topology->connect_timeout_msec * 1000;
+
+      if (!mongoc_stream_wait (scanner_node->stream, expire_at)) {
+         bson_set_error (error,
+                         MONGOC_ERROR_STREAM,
+                         MONGOC_ERROR_STREAM_NOT_ESTABLISHED,
+                         "Could not reconnect to %s",
+                         scanner_node->host.host_and_port);
+         RETURN(false);
+      }
+
       if (scanner_node->stream && cluster->requires_auth) {
          sd = mongoc_topology_server_by_id (topology, server_id);
          if (!sd) {

src/mongoc/mongoc-stream.c

@@ -20,6 +20,7 @@
 #include "mongoc-array-private.h"
 #include "mongoc-buffer-private.h"
 #include "mongoc-error.h"
+#include "mongoc-errno-private.h"
 #include "mongoc-flags.h"
 #include "mongoc-log.h"
 #include "mongoc-opcode.h"
@@ -353,6 +354,87 @@ 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)
 {

src/mongoc/mongoc-stream.h

@@ -94,6 +94,8 @@ int              mongoc_stream_setsockopt      (mongoc_stream_t       *stream,
                                                 int                    optname,
                                                 void                  *optval,
                                                 socklen_t              optlen);
+bool             mongoc_stream_wait            (mongoc_stream_t       *stream,
+                                                int64_t                expire_at);
 bool             mongoc_stream_check_closed    (mongoc_stream_t       *stream);
 ssize_t          mongoc_stream_poll            (mongoc_stream_poll_t  *streams,
                                                 size_t                 nstreams,