feat(xo-server/Xapi): make built-in concurrency limits configurable (#4743)

See xoa-support#2075

CHANGELOG.unreleased.md

@@ -10,6 +10,11 @@
 - [Snapshot] Fallback to normal snapshot if quiesce is not available [#4735](https://github.com/vatesfr/xen-orchestra/issues/4735) (PR [#4736](https://github.com/vatesfr/xen-orchestra/pull/4736)) \
   Fixes compatibility with **Citrix Hypervisor 8.1**.
 - [Uncompressed full backup] Quick healthcheck of downloaded XVAs in case there was an undetected issue (PR [#4741](https://github.com/vatesfr/xen-orchestra/pull/4741))
+- [Backup] Make built-in concurrency limits configurable (PR [#4743](https://github.com/vatesfr/xen-orchestra/pull/4743)) \
+  Via the following entries in `xo-server`'s configuration file:
+  - `xapiOptions.vdiExportConcurrency`
+  - `xapiOptions.vmExportConcurrency`
+  - `xapiOptions.vmSnapshotConcurrency`
 
 ### Bug fixes
 

packages/xo-server/config.toml

@@ -121,3 +121,6 @@ timeout = 600e3
 
 [xapiOptions]
 maxUncoalescedVdis = 1
+vdiExportConcurrency = 12
+vmExportConcurrency = 2
+vmSnapshotConcurrency = 2

packages/xo-server/src/xapi/index.js

@@ -94,12 +94,31 @@ export const IPV6_CONFIG_MODES = ['None', 'DHCP', 'Static', 'Autoconf']
 
 @mixin(mapToArray(mixins))
 export default class Xapi extends XapiBase {
-  constructor({ guessVhdSizeOnImport, maxUncoalescedVdis, ...opts }) {
+  constructor({
+    guessVhdSizeOnImport,
+    maxUncoalescedVdis,
+    vdiExportConcurrency,
+    vmExportConcurrency,
+    vmSnapshotConcurrency,
+    ...opts
+  }) {
     super(opts)
 
     this._guessVhdSizeOnImport = guessVhdSizeOnImport
     this._maxUncoalescedVdis = maxUncoalescedVdis
 
+    const waitStreamEnd = async stream => fromEvent(await stream, 'end')
+    this._exportVdi = concurrency(
+      vdiExportConcurrency,
+      waitStreamEnd
+    )(this._exportVdi)
+    this.exportVm = concurrency(
+      vmExportConcurrency,
+      waitStreamEnd
+    )(this.exportVm)
+
+    this._snapshotVm = concurrency(vmSnapshotConcurrency)(this._snapshotVm)
+
     // Patch getObject to resolve _xapiId property.
     this.getObject = (getObject => (...args) => {
       let tmp
@@ -689,7 +708,6 @@ export default class Xapi extends XapiBase {
   }
 
   // Returns a stream to the exported VM.
-  @concurrency(2, stream => stream.then(stream => fromEvent(stream, 'end')))
   @cancelable
   async exportVm($cancelToken, vmId, { compress = false } = {}) {
     const vm = this.getObject(vmId)
@@ -1459,7 +1477,6 @@ export default class Xapi extends XapiBase {
     }
   }
 
-  @concurrency(2)
   @cancelable
   async _snapshotVm($cancelToken, { $ref: vmRef }, nameLabel) {
     const vm = await this.getRecord('VM', vmRef)
@@ -1916,7 +1933,6 @@ export default class Xapi extends XapiBase {
     return snap
   }
 
-  @concurrency(12, stream => stream.then(stream => fromEvent(stream, 'end')))
   @cancelable
   _exportVdi($cancelToken, vdi, base, format = VDI_FORMAT_VHD) {
     const query = {