feat: perf benchmark example

examples/benchmark/src/index.ts

@@ -0,0 +1,133 @@
+/* eslint-disable no-console */
+
+import { tcp } from '@libp2p/tcp'
+import { yamux } from '@chainsafe/libp2p-yamux'
+import { mplex } from '@libp2p/mplex'
+import { noise } from '@chainsafe/libp2p-noise'
+import { createLibp2p, type Libp2pOptions, type Libp2pNode } from 'libp2p'
+import { PerfService } from 'libp2p/perf'
+import { plaintext } from 'libp2p/insecure'
+
+async function newNode (options: Libp2pOptions): Promise<Libp2pNode> {
+  return await createLibp2p({
+    ...options
+  })
+}
+
+async function benchmarkWithOptions (serverOptions: Libp2pOptions, clientOptions: Libp2pOptions) {
+  const server = await newNode(serverOptions)
+  const client = await newNode(clientOptions)
+
+  const serverPerf = new PerfService(server.components)
+  await serverPerf.start()
+
+  const clientPerf = new PerfService(client.components)
+  await clientPerf.start()
+  await client.dial(server.getMultiaddrs()[0])
+
+  // Warmup
+  await clientPerf.measureDownloadBandwidth(server.peerId, 10n << 20n)
+  await clientPerf.measureUploadBandwidth(server.peerId, 10n << 20n)
+
+  const downloadBandwidth = await clientPerf.measureDownloadBandwidth(server.peerId, 100n << 20n)
+  console.log('Download bandwidth is (mbits/s)', BigInt(downloadBandwidth) >> 20n)
+  const uploadBandwidth = await clientPerf.measureDownloadBandwidth(server.peerId, 50n << 20n)
+  console.log('Upload bandwidth is (mbits/s)', BigInt(uploadBandwidth) >> 20n)
+
+  await clientPerf.stop()
+  await serverPerf.stop()
+
+  await server.stop()
+  await client.stop()
+}
+
+async function run () {
+  const testcases = [
+    {
+      name: 'TCP+mplex+noise',
+      baseOptions: {
+        transports: [
+          tcp()
+        ],
+        streamMuxers: [
+          mplex()
+        ],
+        connectionEncryption: [
+          noise()
+        ]
+      },
+      serverOptions: {
+        addresses: {
+          listen: ['/ip4/0.0.0.0/tcp/0']
+        }
+      }
+    },
+    {
+      name: 'TCP+yamux+noise',
+      baseOptions: {
+        transports: [
+          tcp()
+        ],
+        streamMuxers: [
+          yamux()
+        ],
+        connectionEncryption: [
+          noise()
+        ]
+      },
+      serverOptions: {
+        addresses: {
+          listen: ['/ip4/0.0.0.0/tcp/0']
+        }
+      }
+    },
+    {
+      name: 'TCP+yamux+plaintext',
+      baseOptions: {
+        transports: [
+          tcp()
+        ],
+        streamMuxers: [
+          yamux()
+        ],
+        connectionEncryption: [
+          plaintext()
+        ]
+      },
+      serverOptions: {
+        addresses: {
+          listen: ['/ip4/0.0.0.0/tcp/0']
+        }
+      }
+    },
+    {
+      name: 'TCP+mplex+plaintext',
+      baseOptions: {
+        transports: [
+          tcp()
+        ],
+        streamMuxers: [
+          mplex()
+        ],
+        connectionEncryption: [
+          plaintext()
+        ]
+      },
+      serverOptions: {
+        addresses: {
+          listen: ['/ip4/0.0.0.0/tcp/0']
+        }
+      }
+    }
+  ]
+
+  for (const testcase of testcases) {
+    console.log(testcase.name)
+    await benchmarkWithOptions({
+      ...testcase.baseOptions,
+      ...testcase.serverOptions
+    }, testcase.baseOptions)
+  }
+}
+
+void run()

examples/tsconfig.json

@@ -0,0 +1,10 @@
+{
+  "extends": "aegir/src/config/tsconfig.aegir.json",
+  "compilerOptions": {
+    "outDir": "dist"
+  },
+  "include": [
+    "benchmark",
+  ],
+  "exclude": []
+}

package.json

@@ -52,6 +52,10 @@
       "types": "./src/index.d.ts",
       "import": "./dist/src/index.js"
     },
+    "./perf": {
+      "types": "./src/perf/index.d.ts",
+      "import": "./dist/src/perf/index.js"
+    },
     "./insecure": {
       "types": "./dist/src/insecure/index.d.ts",
       "import": "./dist/src/insecure/index.js"
@@ -203,4 +207,4 @@
   "browser": {
     "nat-api": false
   }
-}
+}

src/index.ts

@@ -14,7 +14,7 @@
  * ```
  */
 
-import { createLibp2pNode } from './libp2p.js'
+import { createLibp2pNode, type Libp2pNode } from './libp2p.js'
 import type { RecursivePartial } from '@libp2p/interfaces'
 import type { TransportManagerInit } from './transport-manager.js'
 import type { IdentifyServiceInit } from './identify/index.js'
@@ -168,7 +168,7 @@ export interface Libp2pEvents {
   'peer:discovery': CustomEvent<PeerInfo>
 }
 
-export type { Libp2p }
+export type { Libp2p, Libp2pNode }
 
 export type Libp2pOptions = RecursivePartial<Libp2pInit> & { start?: boolean }
 
@@ -197,7 +197,7 @@ export type Libp2pOptions = RecursivePartial<Libp2pInit> & { start?: boolean }
  * const libp2p = await createLibp2p(options)
  * ```
  */
-export async function createLibp2p (options: Libp2pOptions): Promise<Libp2p> {
+export async function createLibp2p (options: Libp2pOptions): Promise<Libp2pNode> {
   const node = await createLibp2pNode(options)
 
   if (options.start !== false) {

src/perf/index.ts

@@ -0,0 +1,133 @@
+import { logger } from '@libp2p/logger'
+import type { IncomingStreamData, Registrar } from '@libp2p/interface-registrar'
+import type { PeerId } from '@libp2p/interface-peer-id'
+import type { Startable } from '@libp2p/interfaces/startable'
+import type { AbortOptions } from '@libp2p/interfaces'
+import type { ConnectionManager } from '@libp2p/interface-connection-manager'
+
+export const PROTOCOL = '/perf/1.0.0'
+
+const log = logger('libp2p:perf')
+
+const writeBlockSize = BigInt(64 << 10)
+const maxStreams = 1 << 10
+
+export interface PerfComponents {
+  registrar: Registrar
+  connectionManager: ConnectionManager
+}
+
+export class PerfService implements Startable {
+  public readonly protocol: string
+  private readonly components: PerfComponents
+  private started: boolean
+  private readonly databuf: ArrayBuffer
+
+  constructor (components: PerfComponents) {
+    this.components = components
+    this.started = false
+    this.protocol = PROTOCOL
+    this.databuf = new ArrayBuffer(Number(writeBlockSize))
+  }
+
+  async start () {
+    await this.components.registrar.handle(this.protocol, (data: IncomingStreamData) => { void this.handleMessage(data) }, {
+      maxInboundStreams: maxStreams,
+      maxOutboundStreams: maxStreams
+    })
+    this.started = true
+  }
+
+  async stop () {
+    await this.components.registrar.unhandle(this.protocol)
+    this.started = false
+  }
+
+  isStarted () {
+    return this.started
+  }
+
+  async handleMessage (data: IncomingStreamData) {
+    const { stream } = data
+
+    let bytesToSendBack: bigint | null = null
+    for await (const buf of stream.source) {
+      if (bytesToSendBack === null) {
+        bytesToSendBack = BigInt(buf.getBigUint64(0, false))
+      }
+      // Ingest all the bufs and wait for the read side to close
+    }
+
+    const uint8Buf = new Uint8Array(this.databuf)
+
+    if (bytesToSendBack === null) {
+      throw new Error('bytesToSendBack was null')
+    }
+    await stream.sink(async function * () {
+      while (bytesToSendBack > 0n) {
+        let toSend: bigint = writeBlockSize
+        if (toSend > bytesToSendBack) {
+          toSend = bytesToSendBack
+        }
+        bytesToSendBack = bytesToSendBack - toSend
+        yield uint8Buf.subarray(0, Number(toSend))
+      }
+    }())
+  }
+
+  async startPerfOnStream (peer: PeerId, sendBytes: bigint, recvBytes: bigint, options: AbortOptions = {}): Promise<void> {
+    log('dialing %s to %p', this.protocol, peer)
+
+    const uint8Buf = new Uint8Array(this.databuf)
+
+    const connection = await this.components.connectionManager.openConnection(peer, options)
+    const signal = options.signal
+    const stream = await connection.newStream([this.protocol], {
+      signal
+    })
+
+    // Convert sendBytes to uint64 big endian buffer
+    const view = new DataView(this.databuf)
+    view.setBigInt64(0, recvBytes, false)
+
+    await stream.sink((async function * () {
+      // Send the number of bytes to receive
+      yield uint8Buf.subarray(0, 8)
+      // Send the number of bytes to send
+      while (sendBytes > 0n) {
+        let toSend: bigint = writeBlockSize
+        if (toSend > sendBytes) {
+          toSend = sendBytes
+        }
+        sendBytes = sendBytes - toSend
+        yield uint8Buf.subarray(0, Number(toSend))
+      }
+    })())
+
+    // Read the received bytes
+    let actualRecvdBytes = BigInt(0)
+    for await (const buf of stream.source) {
+      actualRecvdBytes += BigInt(buf.length)
+    }
+
+    if (actualRecvdBytes !== recvBytes) {
+      throw new Error(`Expected to receive ${recvBytes} bytes, but received ${actualRecvdBytes}`)
+    }
+
+    stream.close()
+  }
+
+  // measureDownloadBandwidth returns the measured bandwidth in bits per second
+  async measureDownloadBandwidth (peer: PeerId, size: bigint) {
+    const now = Date.now()
+    await this.startPerfOnStream(peer, 0n, size)
+    return Number((8000n * size) / BigInt(Date.now() - now))
+  }
+
+  // measureUploadBandwidth returns the measured bandwidth in bit per second
+  async measureUploadBandwidth (peer: PeerId, size: bigint) {
+    const now = Date.now()
+    await this.startPerfOnStream(peer, size, 0n)
+    return Number((8000n * size) / BigInt(Date.now() - now))
+  }
+}