Compress faster by running brotli in parallel

build-scripts/brotli.mjs

@@ -0,0 +1,46 @@
+// Gulp transform that brotli-compresses files, several at a time.
+//
+// Drop-in replacement for gulp-brotli. zlib already does the work off the main
+// thread, but that plugin wraps it in through2, which waits for each file
+// before starting the next, so only one compression is ever in flight. The
+// compressed bytes are unchanged; only how many run at once differs.
+//
+// The real ceiling is libuv's threadpool, which zlib runs on. It sizes itself
+// from UV_THREADPOOL_SIZE before any JavaScript runs, so it can only be raised
+// from the environment, never from inside the build.
+
+import { availableParallelism } from "node:os";
+import { buffer as readStream } from "node:stream/consumers";
+import { promisify } from "node:util";
+import { brotliCompress } from "node:zlib";
+import { ParallelTransform } from "./parallel-transform.mjs";
+
+const EXTENSION = ".br";
+
+const compress = promisify(brotliCompress);
+
+/**
+ * @param {object} [options]
+ * @param {boolean} [options.skipLarger] Drop files that compression grows.
+ * @param {object} [options.params] Brotli parameters, passed to zlib as-is.
+ */
+export default ({ skipLarger = false, params } = {}) =>
+  new ParallelTransform(availableParallelism(), async (file) => {
+    if (file.isNull()) {
+      return file;
+    }
+    if (file.isStream()) {
+      file.contents = await readStream(file.contents);
+    }
+
+    const compressed = await compress(file.contents, { params });
+    if (skipLarger && compressed.length >= file.contents.length) {
+      // Dropped rather than passed through, as gulp-brotli did: the
+      // uncompressed file is already in the output directory.
+      return undefined;
+    }
+
+    file.contents = compressed;
+    file.path += EXTENSION;
+    return file;
+  });

build-scripts/gulp/compress.js

@@ -2,7 +2,7 @@
 
 import { constants } from "node:zlib";
 import gulp from "gulp";
-import brotli from "gulp-brotli";
+import brotli from "../brotli.mjs";
 import paths from "../paths.cjs";
 import zopfli from "../zopfli.mjs";
 

build-scripts/parallel-transform.mjs

@@ -0,0 +1,64 @@
+// Object-mode transform that keeps several files in flight at once.
+//
+// through2 and node's Transform both wait for the previous callback before
+// handling the next file, which serialises asynchronous work down to one file
+// at a time. This keeps `limit` files in flight and applies backpressure beyond
+// that. Files are emitted in completion order rather than input order.
+
+import { Transform } from "node:stream";
+
+export class ParallelTransform extends Transform {
+  #limit;
+
+  #handle;
+
+  #inFlight = 0;
+
+  #resume;
+
+  #finish;
+
+  constructor(limit, handle) {
+    super({ objectMode: true, highWaterMark: limit });
+    this.#limit = limit;
+    this.#handle = handle;
+  }
+
+  // eslint-disable-next-line @typescript-eslint/naming-convention -- node's Transform API
+  _transform(file, _encoding, callback) {
+    this.#inFlight += 1;
+    this.#handle(file)
+      .then((result) => {
+        if (result) {
+          this.push(result);
+        }
+      })
+      .catch((error) => this.destroy(error))
+      .finally(() => {
+        this.#inFlight -= 1;
+        const resume = this.#resume;
+        this.#resume = undefined;
+        resume?.();
+        if (this.#inFlight === 0) {
+          const finish = this.#finish;
+          this.#finish = undefined;
+          finish?.();
+        }
+      });
+
+    if (this.#inFlight < this.#limit) {
+      callback();
+    } else {
+      this.#resume = callback;
+    }
+  }
+
+  // eslint-disable-next-line @typescript-eslint/naming-convention -- node's Transform API
+  _flush(callback) {
+    if (this.#inFlight === 0) {
+      callback();
+    } else {
+      this.#finish = callback;
+    }
+  }
+}

build-scripts/zopfli.mjs

@@ -6,9 +6,9 @@
 // instance per worker parallelises it; the output bytes are unchanged.
 
 import { availableParallelism } from "node:os";
-import { Transform } from "node:stream";
 import { buffer as readStream } from "node:stream/consumers";
 import { Worker } from "node:worker_threads";
+import { ParallelTransform } from "./parallel-transform.mjs";
 
 const WORKER_URL = new URL("./zopfli-worker.mjs", import.meta.url);
 const EXTENSION = ".gz";
@@ -112,65 +112,6 @@ const sharedPool = () => {
   return pool;
 };
 
-// through2 and node's Transform both wait for the previous callback before
-// handling the next file, which would serialise the pool down to one worker.
-// This keeps `limit` files in flight and applies backpressure beyond that.
-class ParallelTransform extends Transform {
-  #limit;
-
-  #handle;
-
-  #inFlight = 0;
-
-  #resume;
-
-  #finish;
-
-  constructor(limit, handle) {
-    super({ objectMode: true, highWaterMark: limit });
-    this.#limit = limit;
-    this.#handle = handle;
-  }
-
-  // eslint-disable-next-line @typescript-eslint/naming-convention -- node's Transform API
-  _transform(file, _encoding, callback) {
-    this.#inFlight += 1;
-    this.#handle(file)
-      .then((result) => {
-        if (result) {
-          this.push(result);
-        }
-      })
-      .catch((error) => this.destroy(error))
-      .finally(() => {
-        this.#inFlight -= 1;
-        const resume = this.#resume;
-        this.#resume = undefined;
-        resume?.();
-        if (this.#inFlight === 0) {
-          const finish = this.#finish;
-          this.#finish = undefined;
-          finish?.();
-        }
-      });
-
-    if (this.#inFlight < this.#limit) {
-      callback();
-    } else {
-      this.#resume = callback;
-    }
-  }
-
-  // eslint-disable-next-line @typescript-eslint/naming-convention -- node's Transform API
-  _flush(callback) {
-    if (this.#inFlight === 0) {
-      callback();
-    } else {
-      this.#finish = callback;
-    }
-  }
-}
-
 /**
  * @param {object} [options]
  * @param {number} [options.threshold] Skip files smaller than this many bytes.

package.json

@@ -188,7 +188,6 @@
     "glob": "13.0.6",
     "globals": "17.8.0",
     "gulp": "5.0.1",
-    "gulp-brotli": "3.0.0",
     "gulp-json-transform": "0.5.0",
     "gulp-rename": "2.1.0",
     "html-minifier-terser": "7.2.0",

test/build-scripts/brotli.test.js

@@ -0,0 +1,96 @@
+/**
+ * @vitest-environment node
+ */
+
+import { Buffer } from "node:buffer";
+import { Readable } from "node:stream";
+import { promisify } from "node:util";
+import { brotliCompress, constants } from "node:zlib";
+import { describe, expect, it } from "vitest";
+import brotli from "../../build-scripts/brotli.mjs";
+import { file, filler, run as runStream } from "./vinyl-stub.js";
+
+// What build-scripts/gulp/compress.js asks for.
+const PARAMS = {
+  [constants.BROTLI_PARAM_QUALITY]: constants.BROTLI_MAX_QUALITY,
+};
+
+const run = (files, options = { skipLarger: true, params: PARAMS }) =>
+  runStream(brotli(options), files);
+
+const compressInProcess = promisify(brotliCompress);
+
+describe("brotli", () => {
+  it("appends .br and matches zlib byte for byte", async () => {
+    const contents = filler(4096);
+    const [result] = await run([file("/out/app.js", contents)]);
+
+    expect(result.path).toBe("/out/app.js.br");
+    expect(result.contents).toEqual(
+      await compressInProcess(contents, { params: PARAMS })
+    );
+  });
+
+  it("returns contents as a Buffer, which vinyl requires", async () => {
+    const [result] = await run([file("/out/app.js", filler(1024))]);
+
+    expect(Buffer.isBuffer(result.contents)).toBe(true);
+  });
+
+  it("appends to the path rather than replacing the extension", async () => {
+    const [result] = await run([file("/out/nested/chunk.min.js", filler(500))]);
+
+    expect(result.path).toBe("/out/nested/chunk.min.js.br");
+  });
+
+  it("drops files that compression grows when skipLarger is set", async () => {
+    expect(await run([file("/out/tiny.js", filler(1))])).toEqual([]);
+  });
+
+  it("keeps files that compression grows when skipLarger is not set", async () => {
+    const [result] = await run([file("/out/tiny.js", filler(1))], {
+      params: PARAMS,
+    });
+
+    expect(result.path).toBe("/out/tiny.js.br");
+  });
+
+  it("passes null files through", async () => {
+    const [result] = await run([file("/out/adirectory", null)]);
+
+    expect(result.path).toBe("/out/adirectory");
+    expect(result.contents).toBeNull();
+  });
+
+  it("buffers stream-mode contents", async () => {
+    const contents = filler(600);
+    const [result] = await run([
+      file("/out/streamed.js", Readable.from([contents])),
+    ]);
+
+    expect(result.path).toBe("/out/streamed.js.br");
+    expect(result.contents).toEqual(
+      await compressInProcess(contents, { params: PARAMS })
+    );
+  });
+
+  it("handles more files than it keeps in flight", async () => {
+    const count = 40;
+    const files = Array.from({ length: count }, (_, index) =>
+      file(`/out/chunk${index}.js`, filler(200 + index))
+    );
+
+    const results = await run(files);
+
+    expect(results).toHaveLength(count);
+    expect(new Set(results.map((result) => result.path)).size).toBe(count);
+    await Promise.all(
+      results.map(async (result) => {
+        const index = Number(result.path.match(/chunk(\d+)/)[1]);
+        expect(result.contents).toEqual(
+          await compressInProcess(filler(200 + index), { params: PARAMS })
+        );
+      })
+    );
+  });
+});

test/build-scripts/parallel-transform.test.js

@@ -0,0 +1,104 @@
+/**
+ * @vitest-environment node
+ */
+
+import { setImmediate } from "node:timers/promises";
+import { describe, expect, it } from "vitest";
+import { ParallelTransform } from "../../build-scripts/parallel-transform.mjs";
+import { run } from "./vinyl-stub.js";
+
+const defer = () => {
+  const handle = {};
+  handle.promise = new Promise((resolve, reject) => {
+    handle.resolve = resolve;
+    handle.reject = reject;
+  });
+  return handle;
+};
+
+describe("ParallelTransform", () => {
+  it("keeps `limit` handlers in flight and no more", async () => {
+    const pending = [];
+    let active = 0;
+    let peak = 0;
+
+    const stream = new ParallelTransform(3, () => {
+      active += 1;
+      peak = Math.max(peak, active);
+      const handle = defer();
+      pending.push(handle);
+      return handle.promise.then((result) => {
+        active -= 1;
+        return result;
+      });
+    });
+
+    const items = Array.from({ length: 10 }, (_, index) => `item${index}`);
+    const done = run(stream, items);
+
+    await setImmediate();
+    expect(pending).toHaveLength(3);
+
+    for (let index = 0; index < items.length; index += 1) {
+      pending[index].resolve(items[index]);
+      // eslint-disable-next-line no-await-in-loop -- releasing one job at a time is the point
+      await setImmediate();
+    }
+
+    expect(await done).toEqual(items);
+    expect(peak).toBe(3);
+  });
+
+  it("emits in completion order rather than input order", async () => {
+    const pending = new Map();
+    const stream = new ParallelTransform(3, (item) => {
+      const handle = defer();
+      pending.set(item, handle);
+      return handle.promise;
+    });
+
+    const done = run(stream, ["a", "b", "c"]);
+    await setImmediate();
+
+    ["c", "a", "b"].forEach((item) => pending.get(item).resolve(item));
+
+    expect(await done).toEqual(["c", "a", "b"]);
+  });
+
+  it("drops results the handler resolves to nothing for", async () => {
+    const stream = new ParallelTransform(2, async (item) =>
+      item === "skip" ? undefined : item
+    );
+
+    expect(await run(stream, ["a", "skip", "b"])).toEqual(["a", "b"]);
+  });
+
+  it("fails the stream when a handler rejects", async () => {
+    const stream = new ParallelTransform(2, async (item) => {
+      if (item === "bad") {
+        throw new Error("boom");
+      }
+      return item;
+    });
+
+    await expect(run(stream, ["a", "bad", "b"])).rejects.toThrow("boom");
+  });
+
+  it("does not finish until in-flight work completes", async () => {
+    const handle = defer();
+    let ended = false;
+
+    const stream = new ParallelTransform(2, () => handle.promise);
+    const done = run(stream, ["a"]).then((results) => {
+      ended = true;
+      return results;
+    });
+
+    await setImmediate();
+    expect(ended).toBe(false);
+
+    handle.resolve("a");
+
+    expect(await done).toEqual(["a"]);
+  });
+});

test/build-scripts/vinyl-stub.js

@@ -0,0 +1,26 @@
+import { Buffer } from "node:buffer";
+import { Readable } from "node:stream";
+
+// Stand-in for the vinyl files gulp.src yields in buffer mode.
+export const file = (path, contents) => ({
+  path,
+  contents,
+  isNull() {
+    return this.contents === null;
+  },
+  isStream() {
+    return this.contents instanceof Readable;
+  },
+});
+
+export const filler = (length) => Buffer.alloc(length, "a");
+
+export const run = (stream, files) =>
+  new Promise((resolve, reject) => {
+    const out = [];
+    stream.on("data", (result) => out.push(result));
+    stream.on("error", reject);
+    stream.on("end", () => resolve(out));
+    files.forEach((entry) => stream.write(entry));
+    stream.end();
+  });

test/build-scripts/zopfli.test.js

@@ -7,6 +7,7 @@ import process from "node:process";
 import { Readable } from "node:stream";
 import gfxZopfli from "@gfx/zopfli";
 import { describe, expect, it } from "vitest";
+import { file, filler, run as runStream } from "./vinyl-stub.js";
 
 // Keep the pool small; it is created once, on the first factory call.
 process.env.ZOPFLI_WORKERS = "2";
@@ -15,28 +16,8 @@ const { default: zopfli } = await import("../../build-scripts/zopfli.mjs");
 
 const THRESHOLD = 150;
 
-// Stand-in for the vinyl files gulp.src yields in buffer mode.
-const file = (path, contents) => ({
-  path,
-  contents,
-  isNull() {
-    return this.contents === null;
-  },
-  isStream() {
-    return this.contents instanceof Readable;
-  },
-});
-
 const run = (files, options = { threshold: THRESHOLD }) =>
-  new Promise((resolve, reject) => {
-    const stream = zopfli(options);
-    const out = [];
-    stream.on("data", (result) => out.push(result));
-    stream.on("error", reject);
-    stream.on("end", () => resolve(out));
-    files.forEach((entry) => stream.write(entry));
-    stream.end();
-  });
+  runStream(zopfli(options), files);
 
 // What the old in-process plugin did, for byte-for-byte comparison.
 const gzipInProcess = (contents) =>
@@ -46,8 +27,6 @@ const gzipInProcess = (contents) =>
     );
   });
 
-const filler = (length) => Buffer.alloc(length, "a");
-
 describe("zopfli worker pool", () => {
   it("appends .gz and matches in-process zopfli byte for byte", async () => {
     const contents = filler(4096);

yarn.lock

@@ -9714,16 +9714,6 @@ __metadata:
   languageName: node
   linkType: hard
 
-"gulp-brotli@npm:3.0.0":
-  version: 3.0.0
-  resolution: "gulp-brotli@npm:3.0.0"
-  dependencies:
-    plugin-error: "npm:^1.0.1"
-    through2: "npm:^3.0.1"
-  checksum: 10/0eea1fc60ae7f256184155b61a30a916007d21d37234698d8cdb299f64f71b4d68ca3182528e7da5d71290079c32c0228573578b76f5af7af7230c31537ef9d2
-  languageName: node
-  linkType: hard
-
 "gulp-cli@npm:^3.1.0":
   version: 3.1.0
   resolution: "gulp-cli@npm:3.1.0"
@@ -9979,7 +9969,6 @@ __metadata:
     glob: "npm:13.0.6"
     globals: "npm:17.8.0"
     gulp: "npm:5.0.1"
-    gulp-brotli: "npm:3.0.0"
     gulp-json-transform: "npm:0.5.0"
     gulp-rename: "npm:2.1.0"
     hls.js: "npm:1.6.16"
@@ -12977,17 +12966,6 @@ __metadata:
   languageName: node
   linkType: hard
 
-"readable-stream@npm:2 || 3, readable-stream@npm:^3.4.0":
-  version: 3.6.2
-  resolution: "readable-stream@npm:3.6.2"
-  dependencies:
-    inherits: "npm:^2.0.3"
-    string_decoder: "npm:^1.1.1"
-    util-deprecate: "npm:^1.0.1"
-  checksum: 10/d9e3e53193adcdb79d8f10f2a1f6989bd4389f5936c6f8b870e77570853561c362bee69feca2bbb7b32368ce96a85504aa4cedf7cf80f36e6a9de30d64244048
-  languageName: node
-  linkType: hard
-
 "readable-stream@npm:^2.3.5, readable-stream@npm:~2.3.6":
   version: 2.3.8
   resolution: "readable-stream@npm:2.3.8"
@@ -13003,6 +12981,17 @@ __metadata:
   languageName: node
   linkType: hard
 
+"readable-stream@npm:^3.4.0":
+  version: 3.6.2
+  resolution: "readable-stream@npm:3.6.2"
+  dependencies:
+    inherits: "npm:^2.0.3"
+    string_decoder: "npm:^1.1.1"
+    util-deprecate: "npm:^1.0.1"
+  checksum: 10/d9e3e53193adcdb79d8f10f2a1f6989bd4389f5936c6f8b870e77570853561c362bee69feca2bbb7b32368ce96a85504aa4cedf7cf80f36e6a9de30d64244048
+  languageName: node
+  linkType: hard
+
 "readable-stream@npm:^4.7.0":
   version: 4.7.0
   resolution: "readable-stream@npm:4.7.0"
@@ -14561,16 +14550,6 @@ __metadata:
   languageName: node
   linkType: hard
 
-"through2@npm:^3.0.1":
-  version: 3.0.2
-  resolution: "through2@npm:3.0.2"
-  dependencies:
-    inherits: "npm:^2.0.4"
-    readable-stream: "npm:2 || 3"
-  checksum: 10/98bdffba8e877fd8beb2154adc4eb0d52fad281130f56f6e5d18f85d1e1aa528a7b27317b302eb5443f6636ab045d3c272e6dffc61d984775db284823b90532d
-  languageName: node
-  linkType: hard
-
 "time-stamp@npm:^1.0.0":
   version: 1.1.0
   resolution: "time-stamp@npm:1.1.0"