feat(nbd-client,xapi): implement multiple connections NBD

@vates/nbd-client/multi.mjs

@@ -0,0 +1,64 @@
+import { asyncEach } from '@vates/async-each'
+import { NBD_DEFAULT_BLOCK_SIZE } from './constants.mjs'
+import NbdClient from './index.mjs'
+
+export default class MultiNbdClient {
+  #clients = []
+  #readAhead
+
+  constructor(settings, { nbdConcurrency = 8, readAhead = 16, ...options } = {}) {
+    this.#readAhead = readAhead
+    if (!Array.isArray(settings)) {
+      settings = [settings]
+    }
+    for (let i = 0; i < nbdConcurrency; i++) {
+      this.#clients.push(
+        new NbdClient(settings[i % settings.length], { ...options, readAhead: Math.ceil(readAhead / nbdConcurrency) })
+      )
+    }
+  }
+
+  async connect() {
+    for (const client of this.#clients) {
+      await client.connect()
+    }
+  }
+
+  async disconnect() {
+    asyncEach(this.#clients, 
+      client => client.disconnect(),
+      {
+        stopOnError: false
+      }) 
+  }
+
+  async readBlock(index, size = NBD_DEFAULT_BLOCK_SIZE) {
+    const clientId = index % this.#clients.length
+    return this.#clients[clientId].readBlock(index, size)
+  }
+
+  async *readBlocks(indexGenerator) {
+    // default : read all blocks
+    const readAhead = []
+    const makeReadBlockPromise = (index, size) => {
+      const promise = this.readBlock(index, size)
+      // error is handled during unshift
+      promise.catch(() => {})
+      return promise
+    }
+
+    // read all blocks, but try to keep readAheadMaxLength promise waiting ahead
+    for (const { index, size } of indexGenerator()) {
+      // stack readAheadMaxLength promises before starting to handle the results
+      if (readAhead.length === this.#readAhead) {
+        // any error will stop reading blocks
+        yield readAhead.shift()
+      }
+
+      readAhead.push(makeReadBlockPromise(index, size))
+    }
+    while (readAhead.length > 0) {
+      yield readAhead.shift()
+    }
+  }
+}

@xen-orchestra/xapi/vdi.mjs

@@ -6,14 +6,13 @@ import { decorateClass } from '@vates/decorate-with'
 import { strict as assert } from 'node:assert'
 
 import extractOpaqueRef from './_extractOpaqueRef.mjs'
-import NbdClient from '@vates/nbd-client'
-import { createNbdRawStream, createNbdVhdStream } from 'vhd-lib/createStreamNbd.js'
+import MultiNbdClient from '@vates/nbd-client/multi.mjs'
+import { createNbdVhdStream, createNbdRawStream } from 'vhd-lib/createStreamNbd.js'
 import { VDI_FORMAT_RAW, VDI_FORMAT_VHD } from './index.mjs'
 
 const { warn } = createLogger('xo:xapi:vdi')
 
 const noop = Function.prototype
-
 class Vdi {
   async clone(vdiRef) {
     return extractOpaqueRef(await this.callAsync('VDI.clone', vdiRef))
@@ -64,13 +63,13 @@ class Vdi {
     })
   }
 
-  async _getNbdClient(ref) {
+  async _getNbdClient(ref, { nbdConcurrency = 1 } = {}) {
     const nbdInfos = await this.call('VDI.get_nbd_info', ref)
     if (nbdInfos.length > 0) {
       // a little bit of randomization to spread the load
       const nbdInfo = nbdInfos[Math.floor(Math.random() * nbdInfos.length)]
       try {
-        const nbdClient = new NbdClient(nbdInfo, this._nbdOptions)
+        const nbdClient = new MultiNbdClient(nbdInfos, { ...this._nbdOptions, nbdConcurrency })
         await nbdClient.connect()
         return nbdClient
       } catch (err) {
@@ -83,7 +82,10 @@ class Vdi {
     }
   }
 
-  async exportContent(ref, { baseRef, cancelToken = CancelToken.none, format, preferNbd = this._preferNbd }) {
+  async exportContent(
+    ref,
+    { baseRef, cancelToken = CancelToken.none, format, nbdConcurrency = 1, preferNbd = this._preferNbd }
+  ) {
     const query = {
       format,
       vdi: ref,
@@ -97,7 +99,7 @@ class Vdi {
     let nbdClient, stream
     try {
       if (preferNbd) {
-        nbdClient = await this._getNbdClient(ref)
+        nbdClient = await this._getNbdClient(ref, { nbdConcurrency })
       }
       // the raw nbd export does not need to peek ath the vhd source
       if (nbdClient !== undefined && format === VDI_FORMAT_RAW) {

CHANGELOG.unreleased.md

@@ -42,7 +42,8 @@
 
 <!--packages-start-->
 
-- @xen-orchestra/xapi patch
+- @vates/nbd-client minor
+- @xen-orchestra/xapi minor
 - vhd-lib patch
 - xo-server minor
 - xo-server-auth-saml minor