Worker threads/signature verification pool tests

packages/vm/src/worker/signatureWorker.ts

@@ -0,0 +1,25 @@
+import { ecrecover } from '@ethereumjs/util'
+import { parentPort } from 'worker_threads'
+
+interface SignatureTask {
+  msgHash: Uint8Array
+  v: bigint
+  r: Uint8Array
+  s: Uint8Array
+  chainId?: bigint
+}
+
+interface SignatureResult {
+  publicKey: Uint8Array
+}
+
+function processSignature(task: SignatureTask): SignatureResult {
+  const publicKey = ecrecover(task.msgHash, task.v, task.r, task.s, task.chainId)
+  return { publicKey }
+}
+
+// Listen for messages from the main thread
+parentPort?.on('message', (tasks: SignatureTask[]) => {
+  const results = tasks.map(processSignature)
+  parentPort?.postMessage(results)
+})

packages/vm/src/worker/signatureWorkerPool.ts

@@ -0,0 +1,86 @@
+import { Worker } from 'worker_threads'
+
+interface SignatureTask {
+  msgHash: Uint8Array
+  v: bigint
+  r: Uint8Array
+  s: Uint8Array
+  chainId?: bigint
+}
+
+interface SignatureResult {
+  publicKey: Uint8Array
+}
+
+export class SignatureWorkerPool {
+  private workers: Worker[] = []
+  private results: Map<number, SignatureResult> = new Map()
+  private nextTaskId = 0
+  private pendingResults = 0
+
+  constructor(numWorkers: number = 4) {
+    const workerCode = `
+            import { parentPort } from 'worker_threads'
+            import { ecrecover } from '@ethereumjs/util'
+
+            parentPort.on('message', (data) => {
+                const { tasks, taskId } = data
+                const results = tasks.map(task => {
+                    const publicKey = ecrecover(task.msgHash, task.v, task.r, task.s, task.chainId)
+                    return { publicKey }
+                })
+                parentPort.postMessage({ results, taskId })
+            })
+        `
+
+    // Initialize workers
+    for (let i = 0; i < numWorkers; i++) {
+      const worker = new Worker(workerCode, { eval: true })
+
+      worker.on('message', (data: { results: SignatureResult[]; taskId: number }) => {
+        const { results, taskId } = data
+        results.forEach((result, index) => {
+          this.results.set(taskId + index, result)
+        })
+        this.pendingResults--
+      })
+
+      this.workers.push(worker)
+    }
+  }
+
+  async processBatch(tasks: SignatureTask[]): Promise<Map<number, SignatureResult>> {
+    // Clear previous results
+    this.results.clear()
+    this.nextTaskId = 0
+    this.pendingResults = 0
+
+    // Calculate batch size per worker
+    const batchSize = Math.ceil(tasks.length / this.workers.length)
+
+    // Process tasks in parallel
+    for (let i = 0; i < this.workers.length; i++) {
+      const start = i * batchSize
+      const end = Math.min(start + batchSize, tasks.length)
+      if (start >= tasks.length) break
+
+      const batch = tasks.slice(start, end)
+      this.pendingResults++
+      this.workers[i].postMessage({ tasks: batch, taskId: this.nextTaskId })
+      this.nextTaskId += batch.length
+    }
+
+    // Wait for all results
+    while (this.pendingResults > 0) {
+      await new Promise((resolve) => setTimeout(resolve, 10))
+    }
+
+    return this.results
+  }
+
+  terminate() {
+    for (const worker of this.workers) {
+      worker.terminate()
+    }
+  }
+}

packages/vm/test/benchmark/signatureVerification.ts

@@ -0,0 +1,112 @@
+import { LegacyTx } from '@ethereumjs/tx'
+import {
+  bigIntToUnpaddedBytes,
+  bytesToHex,
+  ecrecover,
+  equalsBytes,
+  randomBytes,
+} from '@ethereumjs/util'
+import { SignatureWorkerPool } from '../../src/worker/signatureWorkerPool.ts'
+
+async function runBenchmark() {
+  // Create 10 test transactions
+  const transactions = Array.from({ length: 2000 }, (_, i) => {
+    return new LegacyTx({
+      nonce: i,
+      gasPrice: 1000000000n,
+      gasLimit: 21000n,
+      to: '0x0000000000000000000000000000000000000000',
+      value: 1000000000000000000n,
+      data: '0x',
+    })
+  })
+
+  // Sign all transactions
+  const signedTxs = transactions.map((tx) => tx.sign(randomBytes(32)))
+
+  // Precompute all hashes and signature values
+  const msgHashes = signedTxs.map((tx) => tx.getMessageToVerifySignature())
+  const vs = signedTxs.map((tx) => tx.v!)
+  const rs = signedTxs.map((tx) => bigIntToUnpaddedBytes(tx.r!))
+  const ss = signedTxs.map((tx) => bigIntToUnpaddedBytes(tx.s!))
+  const chainIds = signedTxs.map((tx) => tx.common.chainId())
+
+  console.log('Running benchmark...\n')
+
+  // Sequential verification
+  console.log('Sequential verification:')
+  const startSeq = performance.now()
+
+  // Store sequential results
+  const sequentialResults = new Map()
+  for (let i = 0; i < signedTxs.length; i++) {
+    const publicKey = ecrecover(msgHashes[i], vs[i], rs[i], ss[i], 1n)
+    sequentialResults.set(i, publicKey)
+  }
+
+  const endSeq = performance.now()
+  console.log(`Time taken: ${endSeq - startSeq}ms\n`)
+
+  // Parallel verification using worker pool
+  console.log('Parallel verification using worker pool:')
+  const startPar = performance.now()
+
+  const signatureTasks = msgHashes.map((msgHash, i) => {
+    return {
+      msgHash,
+      v: vs[i],
+      r: rs[i],
+      s: ss[i],
+      chainId: chainIds[i],
+    }
+  })
+
+  const workerPool = new SignatureWorkerPool(4) // Use 4 workers
+  const parallelResults = await workerPool.processBatch(signatureTasks)
+  await workerPool.terminate()
+
+  const endPar = performance.now()
+  console.log(`Time taken: ${endPar - startPar}ms\n`)
+
+  // Print speedup
+  const speedup = (endSeq - startSeq) / (endPar - startPar)
+  console.log(`Speedup: ${speedup.toFixed(2)}x`)
+
+  // Print result counts
+  console.log(
+    `\nResults count - Sequential: ${sequentialResults.size}, Parallel: ${parallelResults.size}`,
+  )
+
+  // Verify results match by comparing the stored sequential results with parallel results
+  console.log('\nVerifying results...')
+  let allMatch = true
+  let missingResults = 0
+  let mismatchedResults = 0
+
+  for (let i = 0; i < signedTxs.length; i++) {
+    const parallelResult = parallelResults.get(i)
+    const sequentialResult = sequentialResults.get(i)
+
+    if (!parallelResult) {
+      console.log(`Missing parallel result at index ${i}`)
+      missingResults++
+      allMatch = false
+      continue
+    }
+
+    if (!equalsBytes(parallelResult.publicKey, sequentialResult)) {
+      console.log(`Mismatch at index ${i}`)
+      console.log(`Sequential: ${sequentialResult}`)
+      console.log(`Parallel: ${parallelResult.publicKey}`)
+      mismatchedResults++
+      allMatch = false
+    }
+  }
+
+  console.log(`\nVerification Summary:`)
+  console.log(`- All results match: ${allMatch}`)
+  console.log(`- Missing results: ${missingResults}`)
+  console.log(`- Mismatched results: ${mismatchedResults}`)
+}
+
+runBenchmark().catch(console.error)