Check for closed sockets before attempting a rpc

Ensure that the socket we've selected has been read from within a
second.  If it hasn't, check if it's closed by polling for readability
and if that works, MSG_PEEK'ing a single byte.  If we get eof, or an
error, we can assume it's broken.  This lets us avoid a large class of
transient network failures.

Closes #136

Author: hanumantmk

src/mongoc/mongoc-cluster-private.h

@@ -78,6 +78,7 @@ typedef struct
    int32_t             max_wire_version;
    int32_t             max_write_batch_size;
    char               *replSet;
+   int64_t             last_read_msec;
 } mongoc_cluster_node_t;
 
 

src/mongoc/mongoc-cluster.c

@@ -63,6 +63,8 @@
 #define MIN_WIRE_VERSION 0
 #define MAX_WIRE_VERSION 3
 
+#define CHECK_CLOSED_DURATION_MSEC 1000
+
 
 #ifndef DEFAULT_SOCKET_TIMEOUT_MSEC
 /*
@@ -2656,20 +2658,34 @@ _mongoc_cluster_sendv (mongoc_cluster_t             *cluster,
       }
    }
 
-   /*
-    * Try to find a node to deliver to. Since we are allowed to block in this
-    * version of sendv, we try to reconnect if we cannot select a node.
-    */
-   while (!(node = _mongoc_cluster_select (cluster, rpcs, rpcs_len, hint,
-                                           write_concern, read_prefs,
-                                           error))) {
-      if ((retry_count++ == MAX_RETRY_COUNT) ||
-          !_mongoc_cluster_reconnect (cluster, error)) {
-         RETURN (false);
+   for (;;) {
+      /*
+       * Try to find a node to deliver to. Since we are allowed to block in this
+       * version of sendv, we try to reconnect if we cannot select a node.
+       */
+      while (!(node = _mongoc_cluster_select (cluster, rpcs, rpcs_len, hint,
+                                              write_concern, read_prefs,
+                                              error))) {
+         if ((retry_count++ == MAX_RETRY_COUNT) ||
+             !_mongoc_cluster_reconnect (cluster, error)) {
+            RETURN (false);
+         }
       }
-   }
 
-   BSON_ASSERT(node->stream);
+      BSON_ASSERT(node->stream);
+
+      if (node->last_read_msec + CHECK_CLOSED_DURATION_MSEC < now) {
+         if (mongoc_stream_check_closed (node->stream)) {
+            _mongoc_cluster_disconnect_node (cluster, node);
+            _mongoc_cluster_reconnect (cluster, NULL);
+         } else {
+            node->last_read_msec = now;
+            break;
+         }
+      } else {
+         break;
+      }
+   }
 
    _mongoc_array_clear (&cluster->iov);
 
@@ -3005,6 +3021,8 @@ _mongoc_cluster_try_recv (mongoc_cluster_t *cluster,
       RETURN (false);
    }
 
+   node->last_read_msec = bson_get_monotonic_time ();
+
    DUMP_BYTES (buffer, buffer->data + buffer->off, buffer->len);
 
    _mongoc_rpc_swab_from_le (rpc);

src/mongoc/mongoc-socket.c

@@ -1069,3 +1069,28 @@ mongoc_socket_getnameinfo (mongoc_socket_t *sock) /* IN */
 
    RETURN (NULL);
 }
+
+
+bool
+mongoc_socket_check_closed (mongoc_socket_t *sock) /* IN */
+{
+   bool closed = false;
+   char buf [1];
+   ssize_t r;
+
+   if (_mongoc_socket_wait (sock->sd, POLLIN, 0)) {
+      sock->errno_ = 0;
+
+      r = recv (sock->sd, buf, 1, MSG_PEEK);
+
+      if (r < 0) {
+         _mongoc_socket_capture_errno (sock);
+      }
+
+      if (r < 1) {
+         closed = true;
+      }
+   }
+
+   return closed;
+}

src/mongoc/mongoc-socket.h

@@ -86,6 +86,7 @@ ssize_t          mongoc_socket_sendv      (mongoc_socket_t       *sock,
                                            mongoc_iovec_t        *iov,
                                            size_t                 iovcnt,
                                            int64_t                expire_at);
+bool             mongoc_socket_check_closed (mongoc_socket_t       *sock);
 
 
 BSON_END_DECLS

src/mongoc/mongoc-stream-buffered.c

@@ -244,6 +244,15 @@ _mongoc_stream_buffered_get_base_stream (mongoc_stream_t *stream) /* IN */
 }
 
 
+static bool
+_mongoc_stream_buffered_check_closed (mongoc_stream_t *stream) /* IN */
+{
+   mongoc_stream_buffered_t *buffered = (mongoc_stream_buffered_t *)stream;
+   bson_return_val_if_fail(stream, -1);
+   return mongoc_stream_check_closed (buffered->base_stream);
+}
+
+
 /*
  *--------------------------------------------------------------------------
  *
@@ -283,6 +292,7 @@ mongoc_stream_buffered_new (mongoc_stream_t *base_stream, /* IN */
    stream->stream.writev = mongoc_stream_buffered_writev;
    stream->stream.readv = mongoc_stream_buffered_readv;
    stream->stream.get_base_stream = _mongoc_stream_buffered_get_base_stream;
+   stream->stream.check_closed = _mongoc_stream_buffered_check_closed;
 
    stream->base_stream = base_stream;
 

src/mongoc/mongoc-stream-file.c

@@ -167,6 +167,13 @@ _mongoc_stream_file_writev (mongoc_stream_t *stream,       /* IN */
 }
 
 
+static bool
+_mongoc_stream_file_check_closed (mongoc_stream_t *stream) /* IN */
+{
+   return false;
+}
+
+
 mongoc_stream_t *
 mongoc_stream_file_new (int fd) /* IN */
 {
@@ -181,6 +188,7 @@ mongoc_stream_file_new (int fd) /* IN */
    stream->vtable.flush = _mongoc_stream_file_flush;
    stream->vtable.readv = _mongoc_stream_file_readv;
    stream->vtable.writev = _mongoc_stream_file_writev;
+   stream->vtable.check_closed = _mongoc_stream_file_check_closed;
    stream->fd = fd;
 
    return (mongoc_stream_t *)stream;

src/mongoc/mongoc-stream-gridfs.c

@@ -139,6 +139,13 @@ _mongoc_stream_gridfs_writev (mongoc_stream_t *stream,
 }
 
 
+static bool
+_mongoc_stream_gridfs_check_closed (mongoc_stream_t *stream) /* IN */
+{
+   return false;
+}
+
+
 mongoc_stream_t *
 mongoc_stream_gridfs_new (mongoc_gridfs_file_t *file)
 {
@@ -156,6 +163,7 @@ mongoc_stream_gridfs_new (mongoc_gridfs_file_t *file)
    stream->stream.flush = _mongoc_stream_gridfs_flush;
    stream->stream.writev = _mongoc_stream_gridfs_writev;
    stream->stream.readv = _mongoc_stream_gridfs_readv;
+   stream->stream.check_closed = _mongoc_stream_gridfs_check_closed;
 
    mongoc_counter_streams_active_inc ();
 

src/mongoc/mongoc-stream-socket.c

@@ -210,6 +210,23 @@ mongoc_stream_socket_get_socket (mongoc_stream_socket_t *stream) /* IN */
 }
 
 
+static bool
+_mongoc_stream_socket_check_closed (mongoc_stream_t *stream) /* IN */
+{
+   mongoc_stream_socket_t *ss = (mongoc_stream_socket_t *)stream;
+
+   ENTRY;
+
+   bson_return_val_if_fail (stream, true);
+
+   if (ss->sock) {
+      RETURN (mongoc_socket_check_closed (ss->sock));
+   }
+
+   RETURN (true);
+}
+
+
 /*
  *--------------------------------------------------------------------------
  *
@@ -242,6 +259,7 @@ mongoc_stream_socket_new (mongoc_socket_t *sock) /* IN */
    stream->vtable.readv = _mongoc_stream_socket_readv;
    stream->vtable.writev = _mongoc_stream_socket_writev;
    stream->vtable.setsockopt = _mongoc_stream_socket_setsockopt;
+   stream->vtable.check_closed = _mongoc_stream_socket_check_closed;
    stream->sock = sock;
 
    return (mongoc_stream_t *)stream;

src/mongoc/mongoc-stream-tls.c

@@ -721,6 +721,15 @@ _mongoc_stream_tls_get_base_stream (mongoc_stream_t *stream)
 }
 
 
+static bool
+_mongoc_stream_tls_check_closed (mongoc_stream_t *stream) /* IN */
+{
+   mongoc_stream_tls_t *tls = (mongoc_stream_tls_t *)stream;
+   bson_return_val_if_fail(stream, -1);
+   return mongoc_stream_check_closed (tls->base_stream);
+}
+
+
 /*
  *--------------------------------------------------------------------------
  *
@@ -779,6 +788,7 @@ mongoc_stream_tls_new (mongoc_stream_t  *base_stream,
    tls->parent.readv = _mongoc_stream_tls_readv;
    tls->parent.setsockopt = _mongoc_stream_tls_setsockopt;
    tls->parent.get_base_stream = _mongoc_stream_tls_get_base_stream;
+   tls->parent.check_closed = _mongoc_stream_tls_check_closed;
    tls->weak_cert_validation = opt->weak_cert_validation;
    tls->bio = bio_ssl;
    tls->ctx = ssl_ctx;

src/mongoc/mongoc-stream.c

@@ -239,3 +239,18 @@ mongoc_stream_get_base_stream (mongoc_stream_t *stream) /* IN */
 
    return NULL;
 }
+
+
+bool
+mongoc_stream_check_closed (mongoc_stream_t *stream)
+{
+   int ret;
+
+   ENTRY;
+
+   bson_return_val_if_fail(stream, -1);
+
+   ret = stream->check_closed(stream);
+
+   RETURN (ret);
+}