Thread-safe pool for server session objects

Author: vector-of-bool

src/libmongoc/CMakeLists.txt

@@ -558,6 +558,7 @@ set (SOURCES ${SOURCES}
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-topology-description.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-topology-description-apm.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-topology-scanner.c
+   ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-ts-pool.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-uri.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-util.c
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-version-functions.c
@@ -615,6 +616,7 @@ set (HEADERS
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-stream-gridfs.h
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-stream-socket.h
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-topology-description.h
+   ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-ts-pool.h
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-uri.h
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-version-functions.h
    ${PROJECT_SOURCE_DIR}/src/mongoc/mongoc-write-concern.h

src/libmongoc/src/mongoc/mongoc-client-pool.c

@@ -185,7 +185,7 @@ mongoc_client_pool_destroy (mongoc_client_pool_t *pool)
       EXIT;
    }
 
-   if (pool->topology->session_pool) {
+   if (!mongoc_ts_pool_is_empty (pool->topology->session_pool)) {
       client = mongoc_client_pool_pop (pool);
       _mongoc_client_end_sessions (client);
       mongoc_client_pool_push (pool, client);

src/libmongoc/src/mongoc/mongoc-client-session-private.h

@@ -102,16 +102,17 @@ _mongoc_client_session_handle_reply (mongoc_client_session_t *session,
                                      const char *cmd_name,
                                      const bson_t *reply);
 
-mongoc_server_session_t *
-_mongoc_server_session_new (bson_error_t *error);
+extern bool
+_mongoc_server_session_init (mongoc_server_session_t *session,
+                             bson_error_t *error);
+
+extern void
+_mongoc_server_session_dtor (mongoc_server_session_t *session);
 
 bool
 _mongoc_server_session_timed_out (const mongoc_server_session_t *server_session,
                                   int64_t session_timeout_minutes);
 
-void
-_mongoc_server_session_destroy (mongoc_server_session_t *server_session);
-
 mongoc_client_session_t *
 _mongoc_client_session_new (mongoc_client_t *client,
                             mongoc_server_session_t *server_session,

src/libmongoc/src/mongoc/mongoc-client-session.c

@@ -709,34 +709,28 @@ _mongoc_client_session_handle_reply (mongoc_client_session_t *session,
    }
 }
 
-
-mongoc_server_session_t *
-_mongoc_server_session_new (bson_error_t *error)
+bool
+_mongoc_server_session_init (mongoc_server_session_t *self, bson_error_t *error)
 {
    uint8_t uuid_data[16];
-   mongoc_server_session_t *s;
-
    ENTRY;
 
-   if (!_mongoc_server_session_uuid (uuid_data, error)) {
-      RETURN (NULL);
-   }
-
-   s = bson_malloc0 (sizeof (mongoc_server_session_t));
-   s->last_used_usec = SESSION_NEVER_USED;
-   s->prev = NULL;
-   s->next = NULL;
-   bson_init (&s->lsid);
-   bson_append_binary (
-      &s->lsid, "id", 2, BSON_SUBTYPE_UUID, uuid_data, sizeof uuid_data);
-
+   self->last_used_usec = SESSION_NEVER_USED;
+   self->prev = NULL;
+   self->next = NULL;
+   bson_init (&self->lsid);
    /* transaction number is a positive integer and will be incremented before
     * each use, so ensure it is initialized to zero. */
-   s->txn_number = 0;
+   self->txn_number = 0;
 
-   RETURN (s);
-}
+   if (!_mongoc_server_session_uuid (uuid_data, error)) {
+      RETURN (false);
+   }
+   BSON_APPEND_BINARY (
+      &self->lsid, "id", BSON_SUBTYPE_UUID, uuid_data, sizeof uuid_data);
 
+   RETURN (true);
+}
 
 bool
 _mongoc_server_session_timed_out (const mongoc_server_session_t *server_session,
@@ -766,14 +760,9 @@ _mongoc_server_session_timed_out (const mongoc_server_session_t *server_session,
 
 
 void
-_mongoc_server_session_destroy (mongoc_server_session_t *server_session)
+_mongoc_server_session_dtor (mongoc_server_session_t *self)
 {
-   ENTRY;
-
-   bson_destroy (&server_session->lsid);
-   bson_free (server_session);
-
-   EXIT;
+   bson_destroy (&self->lsid);
 }
 
 
@@ -806,7 +795,8 @@ _mongoc_client_session_new (mongoc_client_t *client,
                  DEFAULT_MAX_COMMIT_TIME_MS);
 
    if (opts) {
-      mongoc_optional_copy (&opts->causal_consistency, &session->opts.causal_consistency);
+      mongoc_optional_copy (&opts->causal_consistency,
+                            &session->opts.causal_consistency);
       mongoc_optional_copy (&opts->snapshot, &session->opts.snapshot);
       txn_opts_set (&session->opts.default_txn_opts,
                     opts->default_txn_opts.read_concern,
@@ -1686,7 +1676,8 @@ mongoc_client_session_destroy (mongoc_client_session_t *session)
    } else {
       /* If the client has been reset, destroy the server session instead of
     pushing it back into the topology's pool. */
-      _mongoc_server_session_destroy (session->server_session);
+      mongoc_ts_pool_free_item (session->client->topology->session_pool,
+                                session->server_session);
    }
 
    txn_opts_cleanup (&session->opts.default_txn_opts);

src/libmongoc/src/mongoc/mongoc-client.c

@@ -3013,7 +3013,7 @@ _mongoc_client_end_sessions (mongoc_client_t *client)
    mongoc_cluster_t *cluster = &client->cluster;
    bool r;
 
-   if (t->session_pool) {
+   while (!mongoc_ts_pool_is_empty (t->session_pool)) {
       prefs = mongoc_read_prefs_new (MONGOC_READ_PRIMARY_PREFERRED);
       server_id =
          mongoc_topology_select_server_id (t, MONGOC_SS_READ, prefs, &error);
@@ -3087,7 +3087,7 @@ mongoc_client_reset (mongoc_client_t *client)
    client->client_sessions = mongoc_set_new (8, NULL, NULL);
 
    /* Server sessions are owned by us, so we clear the pool on reset. */
-   _mongoc_topology_clear_session_pool (client->topology);
+   mongoc_ts_pool_clear (client->topology->session_pool);
 }
 
 mongoc_change_stream_t *

src/libmongoc/src/mongoc/mongoc-topology-private.h

@@ -27,6 +27,7 @@
 #include "mongoc-uri.h"
 #include "mongoc-client-session-private.h"
 #include "mongoc-crypt-private.h"
+#include "mongoc-ts-pool-private.h"
 
 #define MONGOC_TOPOLOGY_MIN_HEARTBEAT_FREQUENCY_MS 500
 #define MONGOC_TOPOLOGY_SOCKET_CHECK_INTERVAL_MS 5000
@@ -102,7 +103,7 @@ typedef struct _mongoc_topology_t {
    bool single_threaded;
    bool stale;
 
-   mongoc_server_session_t *session_pool;
+   mongoc_ts_pool session_pool;
 
    /* Is client side encryption enabled? */
    bool cse_enabled;
@@ -255,7 +256,7 @@ _mongoc_topology_handle_app_error (mongoc_topology_t *topology,
 void
 _mongoc_topology_clear_connection_pool (mongoc_topology_t *topology,
                                         uint32_t server_id,
-                                        const bson_oid_t* service_id);
+                                        const bson_oid_t *service_id);
 
 void
 mongoc_topology_rescan_srv (mongoc_topology_t *topology);

src/libmongoc/src/mongoc/mongoc-topology.c

@@ -260,7 +260,9 @@ mongoc_topology_new (const mongoc_uri_t *uri, bool single_threaded)
 #endif
 
    topology = (mongoc_topology_t *) bson_malloc0 (sizeof *topology);
-   topology->session_pool = NULL;
+   topology->session_pool =
+      mongoc_ts_pool_new (sizeof (mongoc_server_session_t),
+                          (void (*) (void *)) _mongoc_server_session_dtor);
    heartbeat_default =
       single_threaded ? MONGOC_TOPOLOGY_HEARTBEAT_FREQUENCY_MS_SINGLE_THREADED
                       : MONGOC_TOPOLOGY_HEARTBEAT_FREQUENCY_MS_MULTI_THREADED;
@@ -569,47 +571,14 @@ mongoc_topology_destroy (mongoc_topology_t *topology)
    mongoc_uri_destroy (topology->uri);
    mongoc_topology_description_destroy (&topology->description);
    mongoc_topology_scanner_destroy (topology->scanner);
-
-   /* If we are single-threaded, the client will try to call
-      _mongoc_topology_end_sessions_cmd when it dies. This removes
-      sessions from the pool as it calls endSessions on them. In
-      case this does not succeed, we clear the pool again here. */
-   _mongoc_topology_clear_session_pool (topology);
+   mongoc_ts_pool_free (topology->session_pool);
 
    mongoc_cond_destroy (&topology->cond_client);
    bson_mutex_destroy (&topology->mutex);
 
    bson_free (topology);
 }
 
-/*
- *--------------------------------------------------------------------------
- *
- * _mongoc_topology_clear_session_pool --
- *
- *       Clears the pool of server sessions without sending endSessions.
- *
- * Returns:
- *       Nothing.
- *
- * Side effects:
- *       Server session pool will be emptied.
- *
- *--------------------------------------------------------------------------
- */
-
-void
-_mongoc_topology_clear_session_pool (mongoc_topology_t *topology)
-{
-   mongoc_server_session_t *ss, *tmp1, *tmp2;
-
-   CDL_FOREACH_SAFE (topology->session_pool, ss, tmp1, tmp2)
-   {
-      _mongoc_server_session_destroy (ss);
-   }
-   topology->session_pool = NULL;
-}
-
 /* Returns false if none of the hosts were valid. */
 bool
 mongoc_topology_apply_scanned_srv_hosts (mongoc_uri_t *uri,
@@ -1570,45 +1539,49 @@ _mongoc_topology_pop_server_session (mongoc_topology_t *topology,
          bson_mutex_unlock (&topology->mutex);
          if (!mongoc_topology_select_server_id (
                 topology, MONGOC_SS_READ, NULL, error)) {
-            RETURN (NULL);
+            ss = NULL;
+            goto done;
          }
 
          bson_mutex_lock (&topology->mutex);
          timeout = td->session_timeout_minutes;
       }
 
       if (timeout == MONGOC_NO_SESSIONS) {
-         bson_mutex_unlock (&topology->mutex);
          bson_set_error (error,
                          MONGOC_ERROR_CLIENT,
                          MONGOC_ERROR_CLIENT_SESSION_FAILURE,
                          "Server does not support sessions");
-         RETURN (NULL);
+         ss = NULL;
+         bson_mutex_unlock (&topology->mutex);
+         goto done;
       }
    }
+   bson_mutex_unlock (&topology->mutex);
 
-   while (topology->session_pool) {
-      ss = topology->session_pool;
-      CDL_DELETE (topology->session_pool, ss);
-      /* Sessions do not expire when the topology type is load balanced. */
-      if (loadbalanced) {
+   while (true) {
+      bool got_session;
+      ss = mongoc_ts_pool_try_pop (topology->session_pool, &got_session);
+      if (!got_session) {
+         /* The pool is empty */
+         ss = mongoc_ts_pool_alloc_item (topology->session_pool);
+         if (!_mongoc_server_session_init (ss, error)) {
+            mongoc_ts_pool_free_item (topology->session_pool, ss);
+            ss = NULL;
+            goto done;
+         }
          break;
       }
 
       if (_mongoc_server_session_timed_out (ss, timeout)) {
-         _mongoc_server_session_destroy (ss);
+         mongoc_ts_pool_free_item (topology->session_pool, ss);
          ss = NULL;
-      } else {
-         break;
+         continue;
       }
+         break;
    }
 
-   bson_mutex_unlock (&topology->mutex);
-
-   if (!ss) {
-      ss = _mongoc_server_session_new (error);
-   }
-
+done:
    RETURN (ss);
 }
 
@@ -1633,46 +1606,25 @@ _mongoc_topology_push_server_session (mongoc_topology_t *topology,
    ENTRY;
 
    bson_mutex_lock (&topology->mutex);
-
    timeout = topology->description.session_timeout_minutes;
    loadbalanced = topology->description.type == MONGOC_TOPOLOGY_LOAD_BALANCED;
+   bson_mutex_unlock (&topology->mutex);
 
-   /* start at back of queue and reap timed-out sessions */
-   while (topology->session_pool && topology->session_pool->prev) {
-      ss = topology->session_pool->prev;
-      /* Sessions do not expire when the topology type is load balanced. */
-      if (!loadbalanced && _mongoc_server_session_timed_out (ss, timeout)) {
-         BSON_ASSERT (ss->next); /* silences clang scan-build */
-         CDL_DELETE (topology->session_pool, ss);
-         _mongoc_server_session_destroy (ss);
-      } else {
-         /* if ss is not timed out, sessions in front of it are ok too */
-         break;
-      }
-   }
-
-   /* If session is expiring or "dirty" (a network error occurred on it), do not
-    * return it to the pool. Sessions do not expire when the topology type is
-    * load balanced. */
    if ((!loadbalanced &&
         _mongoc_server_session_timed_out (server_session, timeout)) ||
        server_session->dirty) {
-      _mongoc_server_session_destroy (server_session);
-   } else {
-      /* silences clang scan-build */
-      BSON_ASSERT (!topology->session_pool || (topology->session_pool->next &&
-                                               topology->session_pool->prev));
-
+      /* If session is expiring or "dirty" (a network error occurred on it), do
+       * not return it to the pool. Sessions do not expire when the topology
+       * type is load balanced. */
+      mongoc_ts_pool_free_item (topology->session_pool, server_session);
+   } else if (server_session->last_used_usec == SESSION_NEVER_USED) {
       /* add server session (lsid) to session pool to be reused only if the
        * server session has been used (the server is aware of the session) */
-      if (server_session->last_used_usec == SESSION_NEVER_USED) {
-         _mongoc_server_session_destroy (server_session);
+      mongoc_ts_pool_free_item (topology->session_pool, server_session);
       } else {
-         CDL_PREPEND (topology->session_pool, server_session);
-      }
+      mongoc_ts_pool_push (topology->session_pool, server_session);
    }
 
-   bson_mutex_unlock (&topology->mutex);
 
    EXIT;
 }
@@ -1701,26 +1653,24 @@ _mongoc_topology_push_server_session (mongoc_topology_t *topology,
 bool
 _mongoc_topology_end_sessions_cmd (mongoc_topology_t *topology, bson_t *cmd)
 {
-   mongoc_server_session_t *ss, *tmp1, *tmp2;
-   char buf[16];
-   const char *key;
-   uint32_t i;
    bson_t ar;
+   int i = 0;
 
    bson_init (cmd);
    BSON_APPEND_ARRAY_BEGIN (cmd, "endSessions", &ar);
 
-   i = 0;
-   CDL_FOREACH_SAFE (topology->session_pool, ss, tmp1, tmp2)
-   {
-      bson_uint32_to_string (i, &key, buf, sizeof buf);
-      BSON_APPEND_DOCUMENT (&ar, key, &ss->lsid);
-      CDL_DELETE (topology->session_pool, ss);
-      _mongoc_server_session_destroy (ss);
-
-      if (++i == 10000) {
+   for (; i < 10000; ++i) {
+      char buf[16];
+      const char *key;
+      bool did_pop;
+      mongoc_server_session_t *ss =
+         mongoc_ts_pool_try_pop (topology->session_pool, &did_pop);
+      if (!did_pop) {
          break;
       }
+      bson_uint32_to_string (i, &key, buf, sizeof buf);
+      BSON_APPEND_DOCUMENT (&ar, key, &ss->lsid);
+      mongoc_ts_pool_free_item (topology->session_pool, ss);
    }
 
    bson_append_array_end (cmd, &ar);