perf(CI-05): bound concurrent browser conformance workers

docs/testing.md

@@ -50,6 +50,15 @@ when another corpus changes. Standalone `--reference-only` checks always execute
 CPython again, so nightly/manual/release validation independently verifies the
 recorded expectations. Remove the directory to benchmark a cold run.
 
+Browser differential requests use two concurrent fresh workers, with the same
+10-second per-case deadline and results retained in input order. A failure
+terminates the active pool before the browser closes. Set
+`POCKET_PYTHON_BROWSER_WORKERS=1` for a serial comparison (accepted range: 1–8).
+`scripts/benchmark_browser_workers.py --repeats 3` compares 1/2/4/8 workers on
+the complete exceptions corpus and requires identical ordered results, including
+resource/teardown observations. Use `--corpus PATH` for another corpus. It writes
+timings and host/corpus identities to `build/browser-worker-benchmark.json`.
+
 ## Fast development and merge policy
 
 Run the smallest meaningful local checks for the behavior being changed. A

runtime/browser-pool.mjs

@@ -0,0 +1,50 @@
+// Test transport: each request owns a fresh worker, even when requests overlap.
+export async function runWorkerRequests(workerPath, requests, {
+  concurrency = 2, timeout = 10000, WorkerClass = globalThis.Worker,
+} = {}) {
+  if (!Number.isInteger(concurrency) || concurrency < 1 || concurrency > 8) {
+    throw new RangeError('Browser concurrency must be an integer from 1 to 8.');
+  }
+  if (!Number.isFinite(timeout) || timeout <= 0) throw new RangeError('Invalid worker timeout.');
+  const results = new Array(requests.length);
+  const active = new Set();
+  let next = 0, failed = false;
+  const run = request => new Promise((resolve, reject) => {
+    let worker, timer;
+    const cleanup = () => {
+      clearTimeout(timer);
+      active.delete(cleanup);
+      if (worker) {
+        worker.onmessage = worker.onerror = worker.onmessageerror = null;
+        worker.terminate();
+      }
+    };
+    const fail = error => {
+      failed = true;
+      cleanup();
+      for (const close of active) close();
+      reject(new Error(`Conformance worker ${request.id}: ${error.message}`, {cause: error}));
+    };
+    try {
+      worker = new WorkerClass(workerPath, {type: 'module'});
+      active.add(cleanup);
+      timer = setTimeout(() => fail(new Error(`timed out after ${timeout}ms`)), timeout);
+      worker.onmessage = event => {
+        cleanup();
+        resolve({...event.data, id: request.id});
+      };
+      worker.onerror = event => fail(new Error(event.message));
+      worker.onmessageerror = () => fail(new Error('could not deserialize response'));
+      worker.postMessage(request);
+    } catch (error) {
+      fail(error);
+    }
+  });
+  await Promise.all(Array.from({length: Math.min(concurrency, requests.length)}, async () => {
+    while (!failed && next < requests.length) {
+      const index = next++;
+      results[index] = await run(requests[index]);
+    }
+  }));
+  return results;
+}

runtime/browser-pool.test.mjs

@@ -0,0 +1,69 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {runWorkerRequests} from './browser-pool.mjs';
+
+function workers() {
+  const instances = [];
+  let live = 0, peak = 0;
+  class FakeWorker {
+    constructor(path, options) {
+      assert.equal(path, '/fixture.mjs');
+      assert.deepEqual(options, {type: 'module'});
+      instances.push(this);
+      peak = Math.max(peak, ++live);
+    }
+    postMessage(request) {
+      this.request = request;
+      if (request.throw) throw new Error('post failed');
+      if (request.hang) return;
+      this.timer = setTimeout(() => {
+        if (request.error) this.onerror?.({message: 'worker failed'});
+        else if (request.messageerror) this.onmessageerror?.();
+        else this.onmessage?.({data: {ok: true, value: request.id}});
+      }, request.delay ?? 1);
+    }
+    terminate() {
+      assert.equal(this.closed, undefined, 'worker terminated once');
+      this.closed = true;
+      clearTimeout(this.timer);
+      live--;
+    }
+  }
+  return {WorkerClass: FakeWorker, instances, live: () => live, peak: () => peak};
+}
+
+test('bounded fresh workers preserve input order despite out-of-order completion', async () => {
+  for (const concurrency of [1, 2, 4, 8]) {
+    const fixture = workers();
+    const requests = Array.from({length: 12}, (_, id) => ({id, delay: id === 0 ? 20 : 1}));
+    const results = await runWorkerRequests('/fixture.mjs', requests, {...fixture, concurrency});
+    assert.deepEqual(results, requests.map(({id}) => ({id, ok: true, value: id})));
+    assert.equal(fixture.instances.length, requests.length);
+    assert.equal(fixture.peak(), concurrency);
+    assert.equal(fixture.live(), 0);
+  }
+});
+
+test('errors, serialization failures and per-case deadlines close all workers and stop scheduling', async () => {
+  for (const failure of [{error: true}, {messageerror: true}, {throw: true}, {hang: true}]) {
+    const fixture = workers();
+    const requests = [{id: 'blocked', hang: true}, {id: 'fails', ...failure}, {id: 'never'}];
+    await assert.rejects(runWorkerRequests('/fixture.mjs', requests,
+      {...fixture, concurrency: 2, timeout: 25}), /Conformance worker (fails|blocked):/);
+    assert.equal(fixture.live(), 0);
+    assert.equal(fixture.instances.length, 2);
+  }
+});
+
+test('worker construction failure is reported and invalid bounds are rejected', async () => {
+  await assert.rejects(runWorkerRequests('/fixture.mjs', [{id: 'one'}], {
+    WorkerClass: class {constructor() {throw new Error('construction failed');}},
+  }), /one: construction failed/);
+  for (const concurrency of [0, 9, 1.5, NaN]) {
+    await assert.rejects(runWorkerRequests('', [], {concurrency}), RangeError);
+  }
+  for (const timeout of [0, -1, Infinity]) {
+    await assert.rejects(runWorkerRequests('', [], {timeout}), RangeError);
+  }
+  assert.deepEqual(await runWorkerRequests('', []), []);
+});

runtime/check_conformance.py

@@ -5,6 +5,7 @@
 import hashlib
 from http.server import ThreadingHTTPServer
 import json
+import os
 from pathlib import Path
 import platform
 import random
@@ -54,10 +55,13 @@ def run_node(engine, requests):
     return response["host"], response["results"]
 
 
-def run_browser(engine, requests, channel):
+def run_browser(engine, requests, channel, concurrency=None):
     from browser_smoke import Handler
     from playwright.sync_api import sync_playwright
 
+    concurrency = int(os.environ.get('POCKET_PYTHON_BROWSER_WORKERS', '2')) if concurrency is None else concurrency
+    if not 1 <= concurrency <= 8:
+        raise ValueError('Browser concurrency must be between 1 and 8.')
     server = ThreadingHTTPServer(("127.0.0.1", 0), partial(Handler, directory=str(ROOT)))
     thread = Thread(target=server.serve_forever, daemon=True)
     thread.start()
@@ -68,26 +72,11 @@ def run_browser(engine, requests, channel):
             try:
                 page = browser.new_page()
                 page.goto(origin)
-                responses = page.evaluate("""async ({workerPath, requests}) => {
-                  const results = [];
-                  for (const request of requests) {
-                    const result = await new Promise((resolve, reject) => {
-                      const worker = new Worker(workerPath, {type: 'module'});
-                      const timer = setTimeout(() => {
-                        worker.terminate(); reject(new Error('Conformance worker timed out.'));
-                      }, 10000);
-                      worker.onmessage = event => {
-                        clearTimeout(timer); worker.terminate(); resolve(event.data);
-                      };
-                      worker.onerror = event => {
-                        clearTimeout(timer); worker.terminate(); reject(new Error(event.message));
-                      };
-                      worker.postMessage(request);
-                    });
-                    results.push({id: request.id, ...result});
-                  }
-                  return results;
-                }""", {"workerPath": "/" + engine["worker"], "requests": requests})
+                responses = page.evaluate("""async ({workerPath, requests, concurrency}) => {
+                  const {runWorkerRequests} = await import('/browser-pool.mjs');
+                  return runWorkerRequests(workerPath, requests, {concurrency});
+                }""", {"workerPath": "/" + engine["worker"], "requests": requests,
+                       "concurrency": concurrency})
                 return browser.version, responses
             finally:
                 browser.close()

scripts/benchmark_browser_workers.py

@@ -0,0 +1,60 @@
+"""Compare isolated browser worker pool sizes on the same differential corpus."""
+
+import argparse
+import json
+import os
+from pathlib import Path
+import platform
+import statistics
+import sys
+import time
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / 'runtime'))
+from check_classes import check
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--corpus', type=Path, default=ROOT / 'tests/fixtures/interpreter/exceptions.json')
+    parser.add_argument('--repeats', type=int, default=3)
+    parser.add_argument('--channel')
+    parser.add_argument('--report', type=Path, default=ROOT / 'build/browser-worker-benchmark.json')
+    args = parser.parse_args()
+    if args.repeats < 1:
+        parser.error('--repeats must be positive')
+    baseline, measurements = None, []
+    original = os.environ.get('POCKET_PYTHON_BROWSER_WORKERS')
+    try:
+        for concurrency in (1, 2, 4, 8):
+            durations = []
+            os.environ['POCKET_PYTHON_BROWSER_WORKERS'] = str(concurrency)
+            for _ in range(args.repeats):
+                start = time.monotonic()
+                report = check('browser', args.channel, args.corpus)
+                durations.append(round(time.monotonic() - start, 3))
+                if report['summary']['fail']:
+                    raise AssertionError(report['summary'])
+                observed = [(case['id'], case['status'], case['actual']) for case in report['cases']]
+                if baseline is None:
+                    baseline = observed
+                if observed != baseline:
+                    raise AssertionError('Pool size changed the ordered differential results')
+            row = {'workers': concurrency, 'seconds': durations,
+                   'median_seconds': statistics.median(durations), 'cases': len(baseline)}
+            measurements.append(row)
+            print(json.dumps(row), flush=True)
+    finally:
+        if original is None:
+            os.environ.pop('POCKET_PYTHON_BROWSER_WORKERS', None)
+        else:
+            os.environ['POCKET_PYTHON_BROWSER_WORKERS'] = original
+    args.report.parent.mkdir(parents=True, exist_ok=True)
+    args.report.write_text(json.dumps({'schema_version': 1, 'python': platform.python_version(),
+        'host': platform.platform(), 'browser': report['host_version'],
+        'corpus_sha256': report['corpus_sha256'], 'repeats': args.repeats,
+        'measurements': measurements}, indent=2) + '\n', encoding='utf-8')
+
+
+if __name__ == '__main__':
+    main()