Bcrypt worker threads

extra/admin-api/Tests/Spacebar.Tests/Tests/AuthenticationTests.cs

@@ -31,6 +31,7 @@ public class AuthenticationTests(ITestOutputHelper testOutputHelper, TestFixture
     [Fact]
     public async Task RegisterUsersConcurrent() {
         testOutputHelper.WriteLine($"Registering {_config.RegisterConcurrentCount} users concurrently...");
+        int i = 0;
         var tasks = Enumerable.Range(0, _config.RegisterConcurrentCount).Select(async _ => {
             var sw = Stopwatch.StartNew();
             var rr = new RegisterRequest() {
@@ -42,7 +43,7 @@ public class AuthenticationTests(ITestOutputHelper testOutputHelper, TestFixture
             };
             
             var result = await Assert.SuccessfullyHttpPostAsJsonAsync($"{_config.TestInstance}/api/v9/auth/register", rr);
-            testOutputHelper.WriteLine($"Registered {rr.Email} in {sw.Elapsed}...");
+            testOutputHelper.WriteLine($"[{i++}] Registered {rr.Email} in {sw.Elapsed}...");
             return (rr, result);
         }).ToList();
         await Task.WhenAll(tasks);

extra/admin-api/Tests/Spacebar.Tests/appsettings.json

@@ -1,6 +1,6 @@
 {
   "Configuration": {
     "TestInstance": "http://localhost:3001",
-    "RegisterConcurrentCount": 15
+    "RegisterConcurrentCount": 150
   }
 }

src/api/Server.ts

@@ -27,6 +27,7 @@ import { initInstance } from "./util/handlers/Instance";
 import { route } from "./util";
 import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
 import { Monitoring } from "../util/monitoring/Monitoring";
+import { BcryptWorkerPool } from "../util/util/workers/bcrypt/BcryptWorkerPool";
 
 const ASSETS_FOLDER = path.join(__dirname, "..", "..", "assets");
 const PUBLIC_ASSETS_FOLDER = path.join(ASSETS_FOLDER, "public");
@@ -61,6 +62,7 @@ export class SpacebarServer extends Server {
         await ConnectionConfig.init();
         await initInstance();
         WebAuthn.init();
+        await BcryptWorkerPool.Init(4); // TODO: make configurable
 
         const logRequests = process.env["LOG_REQUESTS"] != undefined;
         if (logRequests) {

src/api/routes/auth/register.ts

@@ -23,6 +23,7 @@ import { Request, Response, Router } from "express";
 import { HTTPError } from "lambert-server";
 import { MoreThan } from "typeorm";
 import { RegisterSchema } from "@spacebar/schemas";
+import { BcryptWorkerPool } from "../../../util/util/workers/bcrypt/BcryptWorkerPool";
 
 const router: Router = Router({ mergeParams: true });
 
@@ -305,7 +306,8 @@ router.post(
                 });
             }
             // the salt is saved in the password refer to bcrypt docs
-            body.password = await bcrypt.hash(body.password, 12);
+            body.password = await BcryptWorkerPool.GetBcryptWorker().hashPassword(body.password, 12);
+            // body.password = await bcrypt.hash(body.password, 12);
         } else if (register.password.required) {
             throw FieldErrors({
                 password: {

src/util/util/workers/bcrypt/BcryptWorkerPool.ts

@@ -0,0 +1,113 @@
+import { isMainThread, parentPort, Worker } from "node:worker_threads";
+import { ProcessLifecycle } from "../../ProcessLifecycle";
+import { Stopwatch } from "../../Stopwatch";
+import bcrypt from "bcrypt";
+
+export class BcryptWorkerPool {
+    private static _workers: BcryptWorker[] = [];
+    private static _idx: number = 0;
+
+    public static async Init(count: number = 2) {
+        for (let i = 0; i < count; i++) {
+            const sw = Stopwatch.startNew();
+            const w = new BcryptWorker();
+            await w.Init();
+            this._workers.push(w);
+            console.log("[BcryptWorkerPool] Started worker", i, "of", count, "in", sw.elapsed().toString());
+        }
+
+        ProcessLifecycle.eventEmitter.on("stopped", async () => {
+            for (const woKey of this._workers) {
+                await woKey.terminate();
+                console.log("[BcryptWorkerPool] Terminated worker", this._workers.indexOf(woKey), "of", this._workers.length);
+            }
+        });
+    }
+
+    public static GetBcryptWorker() {
+        return this._workers[this._idx++ % this._workers.length];
+    }
+}
+
+class BcryptWorker {
+    public worker: Worker;
+
+    public async Init() {
+        this.worker = await new Promise((res, rej) => {
+            const w = new Worker(__filename, {});
+
+            w.once("online", () => {
+                console.log("[BcryptWorker] Worker", w.threadId, "is online!");
+                res(w);
+            });
+            w.on("error", rej);
+            w.on("exit", (exitCode) => console.log("[BcryptWorker] Worker", w.threadId, "exited with code", exitCode));
+        });
+    }
+
+    public async terminate() {
+        await this.worker.terminate();
+    }
+
+    public async hashPassword(password: string, rounds: number): Promise<string> {
+        const requestId = Math.random().toString(36).substring(2);
+
+        return new Promise((res) => {
+            const sw = Stopwatch.startNew();
+            const handler = (msg: BcryptWorkerMessage) => {
+                if (msg.type == "hash" && msg.requestId === requestId) {
+                    res((msg as BcryptHashMessage).password);
+                    this.worker.off("message", handler);
+                    console.log("[BcryptWorker] Got response to hashPassword in", sw.elapsed().toString());
+                }
+            };
+            this.worker.on("message", handler);
+            this.worker.postMessage({ type: "hash", password, rounds, requestId });
+        });
+    }
+}
+
+interface BcryptWorkerMessage {
+    type: "hash" | "verify";
+    requestId: string;
+}
+
+interface BcryptHashMessage extends BcryptWorkerMessage {
+    type: "hash";
+    password: string;
+    rounds: number;
+}
+
+interface BcryptHashVerify extends BcryptWorkerMessage {
+    type: "verify";
+    password: string;
+}
+
+//region Worker implementation
+if (!isMainThread) {
+    parentPort!.on("message", async (msg: BcryptWorkerMessage) => {
+        console.log("[BcryptWorker] Received", msg.type, "message");
+        switch (msg.type) {
+            case "hash":
+                parentPort?.postMessage({
+                    type: "hash",
+                    requestId: msg.requestId,
+                    rounds: (<BcryptHashMessage>msg).rounds,
+                    password: await bcrypt.hash((<BcryptHashMessage>msg).password, (<BcryptHashMessage>msg).rounds),
+                } satisfies BcryptHashMessage);
+                break;
+            case "verify":
+                parentPort?.postMessage({
+                    type: "hash",
+                    requestId: msg.requestId,
+                    rounds: (<BcryptHashMessage>msg).rounds,
+                    password: await bcrypt.hash((<BcryptHashMessage>msg).password, (<BcryptHashMessage>msg).rounds),
+                } satisfies BcryptHashMessage);
+                break;
+            default:
+                console.error("[BcryptWorker] Unknown message type:", msg.type);
+        }
+    });
+}
+
+//endregion