CDRIVER-721 test reconnection

Also implement mock_server_quit.

Author: ajdavis

tests/mock-server.c

@@ -37,10 +37,12 @@ struct _mock_server_t
 
    mongoc_thread_t        main_thread;
    mongoc_cond_t          cond;
+   mongoc_cond_t          stopped_condition;
    mongoc_mutex_t         mutex;
    bool                   using_main_thread;
 
    const char            *address;
+   mongoc_array_t         streams;  /* mongoc_stream_t pointers */
 
    uint16_t               port;
    mongoc_socket_t       *sock;
@@ -55,6 +57,9 @@ struct _mock_server_t
 
    const char                  *setName;
    const mongoc_host_list_t    *hosts;
+
+   bool stopping;
+   bool stopped;
 };
 
 
@@ -223,6 +228,7 @@ handle_command (mock_server_t   *server,
 static void *
 mock_server_worker (void *data)
 {
+   size_t streams_buffer_pos;
    mongoc_buffer_t buffer;
    mongoc_stream_t *stream;
    mock_server_t *server;
@@ -238,6 +244,11 @@ mock_server_worker (void *data)
    server = closure[0];
    stream = closure[1];
 
+   mongoc_mutex_lock (&server->mutex);
+   streams_buffer_pos = server->streams.len;
+   _mongoc_array_append_val (&server->streams, stream);
+   mongoc_mutex_unlock (&server->mutex);
+
    _mongoc_buffer_init(&buffer, NULL, 0, NULL, NULL);
 
 again:
@@ -284,6 +295,15 @@ mock_server_worker (void *data)
    GOTO (again);
 
 failure:
+   mongoc_mutex_lock (&server->mutex);
+   if (!server->stopping) {
+      _mongoc_array_index (&server->streams,
+                           mongoc_stream_t *,
+                           streams_buffer_pos) = NULL;
+   }
+
+   mongoc_mutex_unlock (&server->mutex);
+
    mongoc_stream_close (stream);
    mongoc_stream_destroy (stream);
    bson_free(closure);
@@ -324,6 +344,7 @@ mock_server_new_rs (const char                  *address,
    server->handler = handler ? handler : dummy_handler;
    server->handler_data = handler_data;
    server->sock = NULL;
+   _mongoc_array_init (&server->streams, sizeof (mongoc_stream_t *));
    server->address = address;
    server->port = port;
 
@@ -338,6 +359,10 @@ mock_server_new_rs (const char                  *address,
 
    mongoc_mutex_init (&server->mutex);
    mongoc_cond_init (&server->cond);
+   mongoc_cond_init (&server->stopped_condition);
+
+   server->stopping = false;
+   server->stopped = false;
 
    return server;
 }
@@ -403,22 +428,31 @@ mock_server_run (mock_server_t *server)
    mongoc_mutex_unlock (&server->mutex);
 
    for (;;) {
-      csock = mongoc_socket_accept (server->sock, -1);
-      if (!csock) {
-         perror ("Failed to accept client socket");
+      mongoc_mutex_lock (&server->mutex);
+      if (server->stopping) {
+         mongoc_mutex_unlock (&server->mutex);
          break;
       }
+      mongoc_mutex_unlock (&server->mutex);
 
-      stream = mongoc_stream_socket_new (csock);
-      closure = bson_malloc0 (sizeof(void*) * 2);
-      closure[0] = server;
-      closure[1] = stream;
+      csock = mongoc_socket_accept (server->sock,
+                                    bson_get_monotonic_time () + 1000 * 1000);
+      if (csock) {
+         stream = mongoc_stream_socket_new (csock);
+         closure = bson_malloc0 (sizeof (void *) * 2);
+         closure[0] = server;
+         closure[1] = stream;
 
-      mongoc_thread_create (&thread, mock_server_worker, closure);
+         mongoc_thread_create (&thread, mock_server_worker, closure);
+      }
    }
 
+   mongoc_mutex_lock (&server->mutex);
    mongoc_socket_close (server->sock);
    server->sock = NULL;
+   server->stopped = true;
+   mongoc_cond_signal (&server->stopped_condition);
+   mongoc_mutex_unlock (&server->mutex);
 
    return 0;
 }
@@ -450,22 +484,36 @@ mock_server_run_in_thread (mock_server_t *server)
 
 
 void
-mock_server_quit (mock_server_t *server,
-                  int            code)
+mock_server_quit (mock_server_t *server)
 {
+   size_t i;
+   mongoc_stream_t *stream;
+
    bson_return_if_fail(server);
 
-   /*
-    * TODO: Exit server loop.
-    */
+   mongoc_mutex_lock (&server->mutex);
+   assert (!server->stopping);
+   assert (!server->stopped);
+   server->stopping = true;
+   assert (0 == mongoc_socket_close (server->sock));  /* success */
+   for (i = 0; i < server->streams.len; i++) {
+      stream = _mongoc_array_index (&server->streams, mongoc_stream_t *, i);
+      if (stream) {
+         assert (0 == mongoc_stream_close (stream));  /* success */
+      }
+   }
+   mongoc_cond_wait (&server->stopped_condition, &server->mutex);
+   mongoc_mutex_unlock (&server->mutex);
 }
 
 
 void
 mock_server_destroy (mock_server_t *server)
 {
    if (server) {
+      _mongoc_array_destroy (&server->streams);
       mongoc_cond_destroy (&server->cond);
+      mongoc_cond_destroy (&server->stopped_condition);
       mongoc_mutex_destroy (&server->mutex);
       bson_free(server);
    }

tests/mock-server.h

@@ -58,8 +58,7 @@ void           mock_server_reply_simple     (mock_server_t        *server,
                                              const bson_t         *doc);
 int            mock_server_run              (mock_server_t         *server);
 void           mock_server_run_in_thread    (mock_server_t         *server);
-void           mock_server_quit             (mock_server_t         *server,
-                                             int                    code);
+void           mock_server_quit             (mock_server_t         *server);
 void           mock_server_destroy          (mock_server_t         *server);
 
 

tests/test-mongoc-client.c

@@ -231,7 +231,7 @@ test_wire_version (void)
 
    mongoc_cursor_destroy (cursor);
    mongoc_collection_destroy (collection);
-   mock_server_quit (server, 0);
+   mock_server_quit (server);
    mongoc_client_destroy (client);
    bson_free (uristr);
 }
@@ -373,7 +373,7 @@ test_mongoc_client_read_prefs (void)
    mongoc_cursor_destroy (cursor);
    mongoc_collection_destroy (collection);
    mongoc_client_destroy (client);
-   mock_server_quit (server, 0);
+   mock_server_quit (server);
    bson_destroy (&b);
    bson_free (uristr);
 }
@@ -504,10 +504,17 @@ test_unavailable_seeds(void)
 }
 
 
+typedef enum {
+   NO_CONNECT,
+   CONNECT,
+   RECONNECT
+} connection_option_t;
+
+
 /* CDRIVER-721 catch errors in _mongoc_cluster_destroy */
 static void 
 test_seed_list (bool rs,
-                bool connect)
+                connection_option_t connection_option)
 {
    uint16_t port;
    char *uri_str;
@@ -531,7 +538,7 @@ test_seed_list (bool rs,
 
    mock_server_run_in_thread (server);
 
-   for (i = 0; i < 10; i++) {
+   for (i = 0; i < 12; i++) {
       suppress_one_message ();
    }
 
@@ -542,12 +549,25 @@ test_seed_list (bool rs,
       assert (client->cluster.nodes[i].valid);
    }
 
-   if (connect) {
+   expected_nodes_len = rs ? 1 : 4;
+
+   if (connection_option != NO_CONNECT) {
       /* only localhost:port responds, nodes_len is set to 1 */
       assert (_mongoc_client_warm_up (client, &error));
 
       /* a mongos load-balanced connection never removes down nodes */
-      expected_nodes_len = rs ? 1 : 4;
+      ASSERT_CMPINT (expected_nodes_len, ==, client->cluster.nodes_len);
+      for (i = 0; i < expected_nodes_len; i++) {
+         assert (client->cluster.nodes[i].valid);
+      }
+   }
+
+   if (connection_option == RECONNECT) {
+      for (i = 0; i < 12; i++) {
+         suppress_one_message ();
+      }
+
+      assert (_mongoc_cluster_reconnect (&client->cluster, &error));
       ASSERT_CMPINT (expected_nodes_len, ==, client->cluster.nodes_len);
       for (i = 0; i < expected_nodes_len; i++) {
          assert (client->cluster.nodes[i].valid);
@@ -557,7 +577,7 @@ test_seed_list (bool rs,
    /* testing for crashes like CDRIVER-721 */
    mongoc_client_destroy (client);
 
-   mock_server_quit (server, 0);
+   mock_server_quit (server);
    mock_server_destroy (server);
    mongoc_uri_destroy (uri);
    bson_free (uri_str);
@@ -567,28 +587,42 @@ test_seed_list (bool rs,
 static void
 test_rs_seeds_no_connect (void)
 {
-   test_seed_list (true, false);
+   test_seed_list (true, NO_CONNECT);
 }
 
 
 static void
 test_rs_seeds_connect (void)
 {
-   test_seed_list (true, true);
+   test_seed_list (true, CONNECT);
+}
+
+
+static void
+test_rs_seeds_reconnect (void)
+{
+   test_seed_list (true, RECONNECT);
 }
 
 
 static void
 test_mongos_seeds_no_connect (void)
 {
-   test_seed_list (false, false);
+   test_seed_list (false, NO_CONNECT);
 }
 
 
 static void
 test_mongos_seeds_connect (void)
 {
-   test_seed_list (false, true);
+   test_seed_list (false, CONNECT);
+}
+
+
+static void
+test_mongos_seeds_reconnect (void)
+{
+   test_seed_list (false, RECONNECT);
 }
 
 
@@ -833,8 +867,10 @@ test_client_install (TestSuite *suite)
    TestSuite_Add (suite, "/Client/unavailable_seeds", test_unavailable_seeds);
    TestSuite_Add (suite, "/Client/rs_seeds_no_connect", test_rs_seeds_no_connect);
    TestSuite_Add (suite, "/Client/rs_seeds_connect", test_rs_seeds_connect);
+   TestSuite_Add (suite, "/Client/rs_seeds_reconnect", test_rs_seeds_reconnect);
    TestSuite_Add (suite, "/Client/mongos_seeds_no_connect", test_mongos_seeds_no_connect);
    TestSuite_Add (suite, "/Client/mongos_seeds_connect", test_mongos_seeds_connect);
+   TestSuite_Add (suite, "/Client/mongos_seeds_reconnect", test_mongos_seeds_reconnect);
    TestSuite_Add (suite, "/Client/exhaust_cursor", test_exhaust_cursor);
    TestSuite_Add (suite, "/Client/server_status", test_server_status);
 }

tests/test-mongoc-database.c

@@ -441,7 +441,7 @@ test_get_collection_names_error (void)
 
    mongoc_database_destroy (database);
    mongoc_client_destroy (client);
-   mock_server_quit (server, 0);
+   mock_server_quit (server);
    bson_destroy (&b);
    bson_free (uristr);
 }