From 6afff1f1708c2e807dcc56e37cb2954f81460822 Mon Sep 17 00:00:00 2001
From: Nir Soffer <nsoffer@redhat.com>
Date: Fri, 7 Aug 2020 04:08:25 +0300
Subject: [PATCH nbdkit] vddk connection pool

Modified patch from Richard.
---
 plugins/vddk/nbdkit-vddk-plugin.pod |   7 +
 plugins/vddk/vddk.c                 | 265 ++++++++++++++++++++--------
 2 files changed, 203 insertions(+), 69 deletions(-)

diff --git a/plugins/vddk/nbdkit-vddk-plugin.pod b/plugins/vddk/nbdkit-vddk-plugin.pod
index e5539da9..288aed4b 100644
--- a/plugins/vddk/nbdkit-vddk-plugin.pod
+++ b/plugins/vddk/nbdkit-vddk-plugin.pod
@@ -175,6 +175,13 @@ Read the password from file descriptor number C<FD>, inherited from
 the parent process when nbdkit starts up.  This is also a secure
 method to supply a password.
 
+=item B<pool-max=>N
+
+To improve performance, for read-only connections (see I<-r> option)
+the plugin will open a pool of VixDiskLibHandle disk handles.  You can
+use this option to control the maximum size of the pool.  The default
+is 8.  To disable this feature, set it to 0 or 1.
+
 =item B<port=>PORT
 
 The port on the VCenter/ESXi host.  Defaults to 443.
diff --git a/plugins/vddk/vddk.c b/plugins/vddk/vddk.c
index 5926e181..b2a045dc 100644
--- a/plugins/vddk/vddk.c
+++ b/plugins/vddk/vddk.c
@@ -52,6 +52,7 @@
 #include "isaligned.h"
 #include "minmax.h"
 #include "rounding.h"
+#include "vector.h"
 
 #include "vddk.h"
 #include "vddk-structs.h"
@@ -85,6 +86,7 @@ char *libdir;                              /* libdir */
 static uint16_t nfc_host_port;             /* nfchostport */
 char *password;                            /* password */
 static uint16_t port;                      /* port */
+static unsigned pool_max = 8;              /* pool-max */
 static const char *server_name;            /* server */
 static bool single_link;                   /* single-link */
 static const char *snapshot_moref;         /* snapshot */
@@ -233,6 +235,12 @@ vddk_config (const char *key, const char *value)
     if (nbdkit_parse_uint16_t ("port", value, &port) == -1)
       return -1;
   }
+  else if (strcmp (key, "pool-max") == 0) {
+    if (nbdkit_parse_unsigned ("pool-max", value, &pool_max) == -1)
+      return -1;
+    if (pool_max < 1)
+      pool_max = 1;
+  }
   else if (strcmp (key, "reexeced_") == 0) {
     /* Special name because it is only for internal use. */
     reexeced = (char *)value;
@@ -482,20 +490,38 @@ vddk_dump_plugin (void)
  * https://code.vmware.com/docs/11750/virtual-disk-development-kit-programming-guide/GUID-6BE903E8-DC70-46D9-98E4-E34A2002C2AD.html
  *
  * Before nbdkit 1.22 we used SERIALIZE_ALL_REQUESTS.  Since nbdkit
- * 1.22 we changed this to SERIALIZE_REQUESTS and added a mutex around
- * calls to VixDiskLib_Open and VixDiskLib_Close.  This is not quite
- * within the letter of the rules, but is within the spirit.
+ * 1.22 we changed this to PARALLEL, added a mutex around calls to
+ * VixDiskLib_ConnectEx, VixDiskLib_Open and VixDiskLib_Close, and use
+ * a pool of disk handles.  This is not quite within the letter of the
+ * rules, but is within the spirit.
  */
-#define THREAD_MODEL NBDKIT_THREAD_MODEL_SERIALIZE_REQUESTS
+#define THREAD_MODEL NBDKIT_THREAD_MODEL_PARALLEL
 
 /* Lock protecting open/close calls - see above. */
 static pthread_mutex_t open_close_lock = PTHREAD_MUTEX_INITIALIZER;
 
+struct handle;
+struct entry {
+  VixDiskLibConnection conn;
+  VixDiskLibHandle disk;
+  bool in_use;               /* true if in use by any thread */
+  struct handle *h;          /* points back to owning nbdkit handle */
+};
+DEFINE_VECTOR_TYPE(pool, struct entry)
+
 /* The per-connection handle. */
-struct vddk_handle {
+struct handle {
   VixDiskLibConnectParams *params; /* connection parameters */
-  VixDiskLibConnection connection; /* connection */
-  VixDiskLibHandle handle;         /* disk handle */
+  int readonly;                    /* readonly flag for this connection */
+  uint32_t flags;                  /* open flags */
+
+  /* Pool of VDDK connections and disk handles.  Do not access this
+   * directly, use GET_DISK_FOR_CURRENT_SCOPE macro to get an unused
+   * handle.
+   */
+  pthread_mutex_t pool_lock;
+  pthread_cond_t pool_cond;
+  pool pool;
 };
 
 static inline VixDiskLibConnectParams *
@@ -527,21 +553,95 @@ free_connect_params (VixDiskLibConnectParams *params)
     free (params);
 }
 
+/* Get a VDDK connection and disk handle on demand. */
+static int
+open_disk (struct handle *h,
+           VixDiskLibConnection *conn_rtn, VixDiskLibHandle *disk_rtn)
+{
+  ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&open_close_lock);
+  VixError err;
+
+  /* XXX Some documentation suggests we should call
+   * VixDiskLib_PrepareForAccess here.  It may be required for
+   * Advanced Transport modes, but I could not make it work with
+   * either ESXi or vCenter servers.
+   */
+
+  DEBUG_CALL ("VixDiskLib_ConnectEx",
+              "h->params, %d, %s, %s, &connection",
+              h->readonly,
+              snapshot_moref ? : "NULL",
+              transport_modes ? : "NULL");
+  err = VixDiskLib_ConnectEx (h->params,
+                              h->readonly,
+                              snapshot_moref,
+                              transport_modes,
+                              conn_rtn);
+  if (err != VIX_OK) {
+    VDDK_ERROR (err, "VixDiskLib_ConnectEx");
+    *conn_rtn = NULL;
+    return -1;
+  }
+
+  DEBUG_CALL ("VixDiskLib_Open",
+              "connection, %s, %d, &handle", filename, h->flags);
+  err = VixDiskLib_Open (*conn_rtn, filename, h->flags, disk_rtn);
+  if (err != VIX_OK) {
+    VDDK_ERROR (err, "VixDiskLib_Open: %s", filename);
+    *disk_rtn = NULL;
+    return -1;
+  }
+
+  nbdkit_debug ("transport mode: %s", VixDiskLib_GetTransportMode (*disk_rtn));
+  return 0;
+}
+
+/* Populate the pool of VDDK connections and disk handles. */
+static int
+populate_pool(struct handle *h)
+{
+  ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&h->pool_lock);
+  int i;
+  const unsigned max = h->readonly ? pool_max : 1;
+
+  /* We have to reserve this vector to ensure that it is not
+   * reallocated, as that would make the pointer returned by get_disk in
+   * another thread invalid.
+   */
+  if (pool_reserve (&h->pool, max) == -1) {
+    nbdkit_error ("pool_reserve: %m");
+    return -1;
+  }
+
+  for (i = 0; i < max; ++i) {
+    VixDiskLibConnection conn = {0};
+    VixDiskLibHandle disk = {0};
+
+    if (open_disk (h, &conn, &disk) == -1)
+      return -1;
+
+    pool_append (&h->pool, (struct entry){ conn, disk, true, h });
+  }
+
+  return 0;
+}
+
 /* Create the per-connection handle. */
 static void *
 vddk_open (int readonly)
 {
-  ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&open_close_lock);
-  struct vddk_handle *h;
-  VixError err;
-  uint32_t flags;
+  struct handle *h;
 
-  h = malloc (sizeof *h);
+  h = calloc (1, sizeof *h);
   if (h == NULL) {
-    nbdkit_error ("malloc: %m");
+    nbdkit_error ("calloc: %m");
     return NULL;
   }
 
+  h->readonly = readonly;
+  pthread_mutex_init (&h->pool_lock, NULL);
+  pthread_cond_init (&h->pool_cond, NULL);
+
   h->params = allocate_connect_params ();
   if (h->params == NULL) {
     nbdkit_error ("allocate VixDiskLibConnectParams: %m");
@@ -568,70 +668,91 @@ vddk_open (int readonly)
     h->params->specType = VIXDISKLIB_SPEC_VMX;
   }
 
-  /* XXX Some documentation suggests we should call
-   * VixDiskLib_PrepareForAccess here.  It may be required for
-   * Advanced Transport modes, but I could not make it work with
-   * either ESXi or vCenter servers.
-   */
-
-  DEBUG_CALL ("VixDiskLib_ConnectEx",
-              "h->params, %d, %s, %s, &connection",
-              readonly,
-              snapshot_moref ? : "NULL",
-              transport_modes ? : "NULL");
-  err = VixDiskLib_ConnectEx (h->params,
-                              readonly,
-                              snapshot_moref,
-                              transport_modes,
-                              &h->connection);
-  if (err != VIX_OK) {
-    VDDK_ERROR (err, "VixDiskLib_ConnectEx");
-    goto err1;
-  }
-
-  flags = 0;
+  h->flags = 0;
   if (readonly)
-    flags |= VIXDISKLIB_FLAG_OPEN_READ_ONLY;
+    h->flags |= VIXDISKLIB_FLAG_OPEN_READ_ONLY;
   if (single_link)
-    flags |= VIXDISKLIB_FLAG_OPEN_SINGLE_LINK;
+    h->flags |= VIXDISKLIB_FLAG_OPEN_SINGLE_LINK;
   if (unbuffered)
-    flags |= VIXDISKLIB_FLAG_OPEN_UNBUFFERED;
+    h->flags |= VIXDISKLIB_FLAG_OPEN_UNBUFFERED;
 
-  DEBUG_CALL ("VixDiskLib_Open",
-              "connection, %s, %d, &handle", filename, flags);
-  err = VixDiskLib_Open (h->connection, filename, flags, &h->handle);
-  if (err != VIX_OK) {
-    VDDK_ERROR (err, "VixDiskLib_Open: %s", filename);
-    goto err2;
+  /* Populating the pool here ensures that all disks are opened in the same
+   * thread, required in the documentation. */
+  if (populate_pool(h) == -1) {
+    nbdkit_error ("populate_pool: %m");
+    goto err0;
   }
 
-  nbdkit_debug ("transport mode: %s",
-                VixDiskLib_GetTransportMode (h->handle));
-
   return h;
 
- err2:
-  DEBUG_CALL ("VixDiskLib_Disconnect", "connection");
-  VixDiskLib_Disconnect (h->connection);
- err1:
-  free_connect_params (h->params);
  err0:
+  free (h->pool.ptr);
+  pthread_cond_destroy (&h->pool_cond);
+  pthread_mutex_destroy (&h->pool_lock);
   free (h);
   return NULL;
 }
 
+static struct entry *
+get_disk (struct handle *h)
+{
+  ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&h->pool_lock);
+  size_t i;
+
+  do {
+    /* See if there's a handle in the pool which is not in use. */
+    for (i = 0; i < h->pool.size; ++i) {
+      if (!h->pool.ptr[i].in_use) {
+        h->pool.ptr[i].in_use = true;
+        return &h->pool.ptr[i];
+      }
+    }
+
+    /* If the pool is full we have to wait for another thread to
+     * finish using its handle and try again.
+     */
+    pthread_cond_wait (&h->pool_cond, &h->pool_lock);
+  } while (1);
+}
+
+static void
+put_disk (struct entry **p)
+{
+  if (*p) {
+    ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&(*p)->h->pool_lock);
+    assert ((*p)->in_use);
+    (*p)->in_use = false;
+
+    /* Wake up one thread waiting in get_disk() for a free handle. */
+    pthread_cond_signal (&(*p)->h->pool_cond);
+  }
+}
+
+/* Wrap get/put_disk in nicer macro. */
+#define GET_DISK_FOR_CURRENT_SCOPE(h, name)             \
+  __attribute__((cleanup (put_disk)))                   \
+  struct entry *name##_h = get_disk (h);                \
+  if (name##_h == NULL) return -1;                      \
+  VixDiskLibHandle name = name##_h->disk
+
 /* Free up the per-connection handle. */
 static void
 vddk_close (void *handle)
 {
   ACQUIRE_LOCK_FOR_CURRENT_SCOPE (&open_close_lock);
-  struct vddk_handle *h = handle;
+  struct handle *h = handle;
+  size_t i;
 
-  DEBUG_CALL ("VixDiskLib_Close", "handle");
-  VixDiskLib_Close (h->handle);
-  DEBUG_CALL ("VixDiskLib_Disconnect", "connection");
-  VixDiskLib_Disconnect (h->connection);
+  for (i = 0; i < h->pool.size; ++i) {
+    DEBUG_CALL ("VixDiskLib_Close", "handle");
+    VixDiskLib_Close (h->pool.ptr[i].disk);
+    DEBUG_CALL ("VixDiskLib_Disconnect", "connection");
+    VixDiskLib_Disconnect (h->pool.ptr[i].conn);
+  }
+  free (h->pool.ptr);
   free_connect_params (h->params);
+  pthread_cond_destroy (&h->pool_cond);
+  pthread_mutex_destroy (&h->pool_lock);
   free (h);
 }
 
@@ -639,13 +760,14 @@ vddk_close (void *handle)
 static int64_t
 vddk_get_size (void *handle)
 {
-  struct vddk_handle *h = handle;
+  struct handle *h = handle;
+  GET_DISK_FOR_CURRENT_SCOPE (h, disk);
   VixDiskLibInfo *info;
   VixError err;
   uint64_t size;
 
   DEBUG_CALL ("VixDiskLib_GetInfo", "handle, &info");
-  err = VixDiskLib_GetInfo (h->handle, &info);
+  err = VixDiskLib_GetInfo (disk, &info);
   if (err != VIX_OK) {
     VDDK_ERROR (err, "VixDiskLib_GetInfo");
     return -1;
@@ -687,7 +809,8 @@ static int
 vddk_pread (void *handle, void *buf, uint32_t count, uint64_t offset,
             uint32_t flags)
 {
-  struct vddk_handle *h = handle;
+  struct handle *h = handle;
+  GET_DISK_FOR_CURRENT_SCOPE (h, disk);
   VixError err;
 
   /* Align to sectors. */
@@ -706,7 +829,7 @@ vddk_pread (void *handle, void *buf, uint32_t count, uint64_t offset,
                        "handle, %" PRIu64 " sectors, "
                        "%" PRIu32 " sectors, buffer",
                        offset, count);
-  err = VixDiskLib_Read (h->handle, offset, count, buf);
+  err = VixDiskLib_Read (disk, offset, count, buf);
   if (err != VIX_OK) {
     VDDK_ERROR (err, "VixDiskLib_Read");
     return -1;
@@ -726,7 +849,8 @@ vddk_pwrite (void *handle, const void *buf, uint32_t count, uint64_t offset,
              uint32_t flags)
 {
   const bool fua = flags & NBDKIT_FLAG_FUA;
-  struct vddk_handle *h = handle;
+  struct handle *h = handle;
+  GET_DISK_FOR_CURRENT_SCOPE (h, disk);
   VixError err;
 
   /* Align to sectors. */
@@ -745,7 +869,7 @@ vddk_pwrite (void *handle, const void *buf, uint32_t count, uint64_t offset,
                        "handle, %" PRIu64 " sectors, "
                        "%" PRIu32 " sectors, buffer",
                        offset, count);
-  err = VixDiskLib_Write (h->handle, offset, count, buf);
+  err = VixDiskLib_Write (disk, offset, count, buf);
   if (err != VIX_OK) {
     VDDK_ERROR (err, "VixDiskLib_Write");
     return -1;
@@ -761,7 +885,8 @@ vddk_pwrite (void *handle, const void *buf, uint32_t count, uint64_t offset,
 static int
 vddk_flush (void *handle, uint32_t flags)
 {
-  struct vddk_handle *h = handle;
+  struct handle *h = handle;
+  GET_DISK_FOR_CURRENT_SCOPE (h, disk);
   VixError err;
 
   /* The Flush call was not available in VDDK < 6.0 so this is simply
@@ -771,7 +896,7 @@ vddk_flush (void *handle, uint32_t flags)
     return 0;
 
   DEBUG_CALL ("VixDiskLib_Flush", "handle");
-  err = VixDiskLib_Flush (h->handle);
+  err = VixDiskLib_Flush (disk);
   if (err != VIX_OK) {
     VDDK_ERROR (err, "VixDiskLib_Flush");
     return -1;
@@ -783,7 +908,8 @@ vddk_flush (void *handle, uint32_t flags)
 static int
 vddk_can_extents (void *handle)
 {
-  struct vddk_handle *h = handle;
+  struct handle *h = handle;
+  GET_DISK_FOR_CURRENT_SCOPE (h, disk);
   VixError err;
   VixDiskLibBlockList *block_list;
 
@@ -808,7 +934,7 @@ vddk_can_extents (void *handle)
   DEBUG_CALL ("VixDiskLib_QueryAllocatedBlocks",
               "handle, 0, %d sectors, %d sectors",
               VIXDISKLIB_MIN_CHUNK_SIZE, VIXDISKLIB_MIN_CHUNK_SIZE);
-  err = VixDiskLib_QueryAllocatedBlocks (h->handle,
+  err = VixDiskLib_QueryAllocatedBlocks (disk,
                                          0, VIXDISKLIB_MIN_CHUNK_SIZE,
                                          VIXDISKLIB_MIN_CHUNK_SIZE,
                                          &block_list);
@@ -864,7 +990,8 @@ static int
 vddk_extents (void *handle, uint32_t count, uint64_t offset, uint32_t flags,
               struct nbdkit_extents *extents)
 {
-  struct vddk_handle *h = handle;
+  struct handle *h = handle;
+  GET_DISK_FOR_CURRENT_SCOPE (h, disk);
   bool req_one = flags & NBDKIT_FLAG_REQ_ONE;
   uint64_t position, end, start_sector;
 
@@ -896,7 +1023,7 @@ vddk_extents (void *handle, uint32_t count, uint64_t offset, uint32_t flags,
                 "handle, %" PRIu64 " sectors, %" PRIu64 " sectors, "
                 "%d sectors",
                 start_sector, nr_sectors, VIXDISKLIB_MIN_CHUNK_SIZE);
-    err = VixDiskLib_QueryAllocatedBlocks (h->handle,
+    err = VixDiskLib_QueryAllocatedBlocks (disk,
                                            start_sector, nr_sectors,
                                            VIXDISKLIB_MIN_CHUNK_SIZE,
                                            &block_list);
-- 
2.25.4

