improve performances

@xen-orchestra/disk-transform/src/consumer/VhdDirectory.mts

@@ -21,7 +21,6 @@ export class VhdDirectoryRemote extends BaseVhd {
     super(source)
     this.#target = target
   }
-
   async write() {
     const { handler, path, compression, flags, validator, concurrency } = this.#target
     const dataPath = `${dirname(path)}/data/${uuidv4()}.vhd`
@@ -30,9 +29,12 @@ export class VhdDirectoryRemote extends BaseVhd {
       const vhd = new VhdDirectory(handler, dataPath, { flags, compression })
       vhd.footer = unpackFooter(this.computeVhdFooter())
       vhd.header = unpackHeader(this.computeVhdHeader())
+
       await asyncEach(
         this.source.diskBlocks(),
-        ({ index, data }) => vhd.writeEntireBlock({ id: index, buffer: Buffer.concat([FULL_BLOCK_BITMAP, data]) }),
+        async ({ index, data }) => {
+          await vhd.writeEntireBlock({ id: index, buffer: Buffer.concat([FULL_BLOCK_BITMAP, data]) })
+        },
         { concurrency }
       )
       await Promise.all([vhd.writeFooter(), vhd.writeHeader(), vhd.writeBlockAllocationTable()])

@xen-orchestra/disk-transform/src/consumer/VhdFileRemote.mts

@@ -1,4 +1,4 @@
-import type { PortableDifferencingDisk } from '../PortableDisk.mts'
+import type { DiskBlockData, PortableDifferencingDisk, RandomAccessPartableDisk } from '../PortableDisk.mts'
 import { VhdStream } from './VhdStream.mts'
 
 type VhdRemoteTarget = {

@xen-orchestra/disk-transform/src/consumer/VhdStream.mts

@@ -13,16 +13,19 @@ export class VhdStream extends BaseVhd {
     const { bat, fileSize } = this.computeVhdBatAndFileSize() // the bat contains the calculated position of the futures blocks
     const blocks = this.source.diskBlocks()
     const FULL_BLOCK_BITMAP = Buffer.alloc(SECTOR_SIZE, 255)
+
     async function* generator(): AsyncGenerator<Buffer> {
+      let start = Date.now()
       yield footer
       yield header
       yield bat
       for await (const { data } of blocks) {
         yield Buffer.concat([FULL_BLOCK_BITMAP, data])
       }
       yield footer
+      console.log(Math.round(Date.now() - start) / 1000)
     }
-    const stream = Readable.from(generator()) as VhdStream
+    const stream = Readable.from(generator(), { objectMode: false, highWaterMark: 10 * 1024 * 1024 }) as VhdStream
     stream.length = fileSize
     return stream
   }

@xen-orchestra/disk-transform/src/producer/XapiVhdStreamNbd.mts

@@ -6,7 +6,7 @@ export class XapiVhdStreamNbdSource extends XapiVhdStreamSource {
   #nbdClient: any
   #nbdConcurrency: number
 
-  constructor({ vdiRef, baseRef, xapi, nbdConcurrency }) {
+  constructor({ vdiRef, baseRef, xapi, nbdConcurrency = 4 }) {
     super({ vdiRef, baseRef, xapi })
     this.#nbdConcurrency = nbdConcurrency
   }

@xen-orchestra/disk-transform/src/producer/XapiVhdStreamSource.mts

@@ -3,7 +3,7 @@ import { readChunkStrict, skipStrict } from '@vates/read-chunk'
 
 import { unpackFooter, unpackHeader } from 'vhd-lib/Vhd/_utils.js'
 import { BLOCK_UNUSED, FOOTER_SIZE, HEADER_SIZE, SECTOR_SIZE } from 'vhd-lib/_constants.js'
-import { type Readable } from 'node:stream'
+import { Readable } from 'node:stream'
 import assert from 'node:assert'
 
 export class XapiVhdStreamSource extends PortableDifferencingDisk {
@@ -62,13 +62,12 @@ export class XapiVhdStreamSource extends PortableDifferencingDisk {
 
   async #getExportStream(): Promise<Readable> {
     // get the stream with nbd or any magic
-    console.log({ baseRef: this.#baseRef, format: 'vhd', preferNbd: false })
     const stream = await this.#xapi.VDI_exportContent(this.#ref, {
       baseRef: this.#baseRef,
       format: 'vhd',
       preferNbd: false,
     })
-    return stream
+    return Readable.from(stream, { highWaterMark: 5 * 1024 * 1024, objectMode: false })
   }
 
   async init(): Promise<void> {
@@ -110,7 +109,6 @@ export class XapiVhdStreamSource extends PortableDifferencingDisk {
     assert.strictEqual(this.#initDone, true, 'init must be done to call buildDiskGenerator')
     const blockIndexes = this.#blocks
     for (const { offset, index } of blockIndexes) {
-      // if this fails, that means an error on source
       await this.#skip(offset - this.#streamOffset) // this will skip the bitmap
       const data = await this.#read(2 * 1024 * 1024)
 
@@ -120,5 +118,6 @@ export class XapiVhdStreamSource extends PortableDifferencingDisk {
         data,
       }
     }
+    // @todo read the end ond check the end footer
   }
 }