diff --git a/extra/admin-api/Tests/Spacebar.Tests/Tests/AuthenticationTests.cs b/extra/admin-api/Tests/Spacebar.Tests/Tests/AuthenticationTests.cs index ca63222..f57916a 100644 --- a/extra/admin-api/Tests/Spacebar.Tests/Tests/AuthenticationTests.cs +++ b/extra/admin-api/Tests/Spacebar.Tests/Tests/AuthenticationTests.cs @@ -31,6 +31,7 @@ [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 @@ }; 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); diff --git a/extra/admin-api/Tests/Spacebar.Tests/appsettings.json b/extra/admin-api/Tests/Spacebar.Tests/appsettings.json index fa03afc..6f444ea 100644 --- a/extra/admin-api/Tests/Spacebar.Tests/appsettings.json +++ b/extra/admin-api/Tests/Spacebar.Tests/appsettings.json @@ -1,6 +1,6 @@ { "Configuration": { "TestInstance": "http://localhost:3001", - "RegisterConcurrentCount": 15 + "RegisterConcurrentCount": 150 } } \ No newline at end of file diff --git a/src/api/Server.ts b/src/api/Server.ts index f9f1c70..08a1e87 100644 --- a/src/api/Server.ts +++ b/src/api/Server.ts @@ -27,6 +27,7 @@ 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 @@ await ConnectionConfig.init(); await initInstance(); WebAuthn.init(); + await BcryptWorkerPool.Init(4); // TODO: make configurable const logRequests = process.env["LOG_REQUESTS"] != undefined; if (logRequests) { diff --git a/src/api/routes/auth/register.ts b/src/api/routes/auth/register.ts index 1427b3c..d831313 100644 --- a/src/api/routes/auth/register.ts +++ b/src/api/routes/auth/register.ts @@ -23,6 +23,7 @@ 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 @@ }); } // 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: { diff --git a/src/util/util/workers/bcrypt/BcryptWorkerPool.ts b/src/util/util/workers/bcrypt/BcryptWorkerPool.ts new file mode 100644 index 0000000..bda5a57 --- /dev/null +++ b/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 { + 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: (msg).rounds, + password: await bcrypt.hash((msg).password, (msg).rounds), + } satisfies BcryptHashMessage); + break; + case "verify": + parentPort?.postMessage({ + type: "hash", + requestId: msg.requestId, + rounds: (msg).rounds, + password: await bcrypt.hash((msg).password, (msg).rounds), + } satisfies BcryptHashMessage); + break; + default: + console.error("[BcryptWorker] Unknown message type:", msg.type); + } + }); +} + +//endregion