Merge pull request #6 from ivancorrea/main

Refactor: Optimization: Single `fastHash` call per execution

Author: samsantosb

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "bee-threads",
-  "version": "4.0.5",
+  "version": "4.0.9",
   "description": "Handle threading as promises. Run CPU-intensive code in worker threads with a simple fluent API.",
   "main": "dist/index.js",
   "module": "dist/index.js",

package.json

@@ -171,22 +171,24 @@ export function clearNonDeterministicCache(): void {
  * @param fnString - Function source code
  * @param args - Function arguments
  * @param context - Execution context (closures)
+ * @param fnHash - Pre-computed function hash (optional, avoids recomputation)
  * @returns Unique request key
  */
 function createRequestKey(
   fnString: string,
   args: unknown[],
-  context: Record<string, unknown> | null | undefined
+  context: Record<string, unknown> | null | undefined,
+  fnHash?: string
 ): string {
   // Hash the function (avoids huge keys for large functions)
-  const fnHash = fastHash(fnString);
+  const hash = fnHash ?? fastHash(fnString);
 
   // Use createContextKey for args and context (faster than JSON.stringify)
   // Note: createContextKey handles empty arrays/objects efficiently
   const argsKey = createContextKey(args);
   const contextKey = context ? createContextKey(context) : '';
 
-  return `${fnHash}:${argsKey}:${contextKey}`;
+  return `${hash}:${argsKey}:${contextKey}`;
 }
 
 // ============================================================================
@@ -209,14 +211,16 @@ function createRequestKey(
  * @param context - Execution context
  * @param factory - Factory function that creates the actual promise
  * @param skipCoalescing - Force skip coalescing for this request
+ * @param fnHash - Pre-computed function hash (optional, avoids recomputation)
  * @returns The (possibly shared) promise
  */
 export function coalesce<T>(
   fnString: string,
   args: unknown[],
   context: Record<string, unknown> | null | undefined,
   factory: () => Promise<T>,
-  skipCoalescing: boolean = false
+  skipCoalescing: boolean = false,
+  fnHash?: string
 ): Promise<T> {
   // Skip coalescing if:
   // 1. Globally disabled
@@ -226,7 +230,7 @@ export function coalesce<T>(
     return factory();
   }
 
-  const key = createRequestKey(fnString, args, context);
+  const key = createRequestKey(fnString, args, context, fnHash);
 
   // Check if there's already an in-flight promise for this request
   const existing = inFlightPromises.get(key);

src/coalescing.ts

@@ -56,11 +56,17 @@ import type {
 
 /**
  * Executes a function once in a worker thread (no retry).
+ * 
+ * @param fn - Function to execute
+ * @param args - Arguments to pass to the function
+ * @param options - Execution options
+ * @param precomputedHash - Pre-computed function hash (optional, avoids recomputation)
  */
 export async function executeOnce<T = unknown>(
   fn: Function | { toString(): string },
   args: unknown[],
-  options: ExecutionOptions = {}
+  options: ExecutionOptions = {},
+  precomputedHash?: string
 ): Promise<T | SafeResult<T>> {
   const {
     safe = false,
@@ -75,8 +81,8 @@ export async function executeOnce<T = unknown>(
   const startTime = Date.now();
   const fnString = fn.toString();
 
-  // Compute hash for worker affinity
-  const fnHash = fastHash(fnString);
+  // Use pre-computed hash or compute now
+  const fnHash = precomputedHash ?? fastHash(fnString);
 
   // Pre-execution checks
   if (signal?.aborted) {
@@ -303,14 +309,23 @@ export async function executeOnce<T = unknown>(
 
 /**
  * Executes a function with optional retry logic and exponential backoff.
+ * 
+ * @param fn - Function to execute
+ * @param args - Arguments to pass to the function
+ * @param options - Execution options
+ * @param precomputedHash - Pre-computed function hash (optional, avoids recomputation)
  */
 export async function execute<T = unknown>(
   fn: Function | { toString(): string },
   args: unknown[],
-  options: ExecutionOptions & { retry?: RetryConfig } = {}
+  options: ExecutionOptions & { retry?: RetryConfig } = {},
+  precomputedHash?: string
 ): Promise<T | SafeResult<T>> {
   const { retry: retryOpts = config.retry, safe = false, context = null, skipCoalescing = false } = options;
   const fnString = fn.toString();
+  
+  // Use pre-computed hash or compute now (fallback for direct execute() calls)
+  const fnHash = precomputedHash ?? fastHash(fnString);
 
   // Wrap execution in coalescing to deduplicate identical concurrent requests
   // Note: Coalescing is skipped for requests with AbortSignal (each needs its own lifecycle)
@@ -319,15 +334,15 @@ export async function execute<T = unknown>(
   const executeWithRetry = async (): Promise<T | SafeResult<T>> => {
     // No retry enabled - execute once
     if (!retryOpts?.enabled) {
-      return executeOnce<T>(fn, args, options);
+      return executeOnce<T>(fn, args, options, fnHash);
     }
 
     const { maxAttempts, baseDelay, maxDelay, backoffFactor } = retryOpts;
     let lastError: Error | undefined;
 
     for (let attempt = 0, len = maxAttempts; attempt < len; attempt++) {
       try {
-        const result = await executeOnce<T>(fn, args, { ...options, safe: false });
+        const result = await executeOnce<T>(fn, args, { ...options, safe: false }, fnHash);
         return safe ? { status: 'fulfilled', value: result as T } : (result as T);
       } catch (err) {
         lastError = err as Error;
@@ -356,7 +371,8 @@ export async function execute<T = unknown>(
       args,
       context,
       executeWithRetry,
-      skipCoalescing
+      skipCoalescing,
+      fnHash
     );
   }
 

src/execution.ts

@@ -40,6 +40,7 @@
 
 import { config } from './config';
 import { execute } from './execution';
+import { fastHash } from './pool';
 import { validateFunction, validateFunctionSize, validateContextSecurity } from './validation';
 import type { Priority, ExecutionOptions, RetryOptions } from './types';
 
@@ -50,6 +51,7 @@ import type { Priority, ExecutionOptions, RetryOptions } from './types';
 /** Internal state for an executor instance */
 interface ExecutorState {
   fnString: string;
+  fnHash: string;
   options: ExecutionOptions;
   args: unknown[];
 }
@@ -76,7 +78,7 @@ export interface Executor<T = unknown> {
  * Creates an immutable, chainable executor.
  */
 export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
-  const { fnString, options, args } = state;
+  const { fnString, fnHash, options, args } = state;
 
   const executor: Executor<T> = {
     /**
@@ -85,6 +87,7 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
     usingParams(...params: unknown[]): Executor<T> {
       return createExecutor<T>({
         fnString,
+        fnHash,
         options,
         args: args.length > 0 ? args.concat(params) : params
       });
@@ -132,6 +135,7 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
       }
       return createExecutor<T>({
         fnString,
+        fnHash,
         options: { ...options, context: serializedContext },
         args
       });
@@ -143,6 +147,7 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
     signal(abortSignal: AbortSignal): Executor<T> {
       return createExecutor<T>({
         fnString,
+        fnHash,
         options: { ...options, signal: abortSignal },
         args
       });
@@ -154,6 +159,7 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
     transfer(list: ArrayBufferLike[]): Executor<T> {
       return createExecutor<T>({
         fnString,
+        fnHash,
         options: { ...options, transfer: list },
         args
       });
@@ -172,6 +178,7 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
       };
       return createExecutor<T>({
         fnString,
+        fnHash,
         options: {
           ...options,
           retry: mergedRetry
@@ -190,6 +197,7 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
       }
       return createExecutor<T>({
         fnString,
+        fnHash,
         options: { ...options, priority: level },
         args
       });
@@ -208,6 +216,7 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
     noCoalesce(): Executor<T> {
       return createExecutor<T>({
         fnString,
+        fnHash,
         options: { ...options, skipCoalescing: true },
         args
       });
@@ -231,6 +240,7 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
     reconstructBuffers(): Executor<T> {
       return createExecutor<T>({
         fnString,
+        fnHash,
         options: { ...options, reconstructBuffers: true },
         args
       });
@@ -243,7 +253,8 @@ export function createExecutor<T = unknown>(state: ExecutorState): Executor<T> {
       return execute<T>(
         { toString: () => fnString },
         args,
-        { ...options, poolType: 'normal' }
+        { ...options, poolType: 'normal' },
+        fnHash
       ) as Promise<T>;
     }
   };
@@ -273,8 +284,12 @@ export function createCurriedRunner(
     // Security: Validate function size (DoS prevention)
     validateFunctionSize(fnString, config.security.maxFunctionSize);
 
+    // Compute hash once at executor creation (not at execute time)
+    const fnHash = fastHash(fnString);
+    
     return createExecutor<ReturnType<T>>({
       fnString,
+      fnHash,
       options: baseOptions,
       args: []
     });

src/executor.ts

@@ -37,8 +37,8 @@
 
 import { config, pools, poolCounters, queues, metrics, RUNTIME, IS_BUN } from './config';
 import { createCurriedRunner, Executor } from './executor';
-import { warmupPool, getQueueLength } from './pool';
-import { validateTimeout, validatePoolSize, validateContextSecurity } from './validation';
+import { warmupPool, getQueueLength, fastHash } from './pool';
+import { validateTimeout, validatePoolSize, validateContextSecurity, validateFunctionSize } from './validation';
 import { deepFreeze } from './utils';
 import { createFileWorker, terminateFileWorkers } from './file-worker';
 import type { FileWorkerExecutor } from './file-worker';
@@ -148,14 +148,12 @@ export function bee<T extends (...args: any[]) => any>(fn: T): CurriedFunction<R
   }
 
   const fnString = fn.toString();
-  
+
   // Security: Validate function size (DoS prevention)
-  const fnSize = Buffer.byteLength(fnString, 'utf8');
-  if (fnSize > config.security.maxFunctionSize) {
-    throw new RangeError(
-      `Function source exceeds maximum size (${fnSize} bytes > ${config.security.maxFunctionSize} bytes limit)`
-    );
-  }
+  validateFunctionSize(fnString, config.security.maxFunctionSize);
+
+  // Compute hash once at executor creation (not at execute time)
+  const fnHash = fastHash(fnString);
 
   type R = ReturnType<T>;
 
@@ -198,12 +196,12 @@ export function bee<T extends (...args: any[]) => any>(fn: T): CurriedFunction<R
         const allArgs = accumulatedArgs.length > 0 
           ? accumulatedArgs.concat(params)
           : params;
-        return execute<R>(fnString, allArgs, { context: serializedClosures }) as unknown as CurriedFunction<R>;
+        return execute<R>(fnString, allArgs, { context: serializedClosures }, fnHash) as unknown as CurriedFunction<R>;
       }
 
       if (callArgs.length === 0) {
         // Empty call () - execute with accumulated args (returns Promise, which is PromiseLike)
-        return execute<R>(fnString, accumulatedArgs, {}) as unknown as CurriedFunction<R>;
+        return execute<R>(fnString, accumulatedArgs, {}, fnHash) as unknown as CurriedFunction<R>;
       }
 
       // Accumulate args and return new curry (thenable for await)
@@ -221,17 +219,17 @@ export function bee<T extends (...args: any[]) => any>(fn: T): CurriedFunction<R
       onFulfilled?: ((value: R) => TResult1 | PromiseLike<TResult1>) | null,
       onRejected?: ((reason: unknown) => TResult2 | PromiseLike<TResult2>) | null
     ): Promise<TResult1 | TResult2> => {
-      return (execute<R>(fnString, accumulatedArgs, {}) as Promise<R>).then(onFulfilled, onRejected);
+      return (execute<R>(fnString, accumulatedArgs, {}, fnHash) as Promise<R>).then(onFulfilled, onRejected);
     };
 
     curry.catch = <TResult = never>(
       onRejected?: ((reason: unknown) => TResult | PromiseLike<TResult>) | null
     ): Promise<R | TResult> => {
-      return (execute<R>(fnString, accumulatedArgs, {}) as Promise<R>).catch(onRejected);
+      return (execute<R>(fnString, accumulatedArgs, {}, fnHash) as Promise<R>).catch(onRejected);
     };
 
     curry.finally = (onFinally?: (() => void) | null): Promise<R> => {
-      return (execute<R>(fnString, accumulatedArgs, {}) as Promise<R>).finally(onFinally);
+      return (execute<R>(fnString, accumulatedArgs, {}, fnHash) as Promise<R>).finally(onFinally);
     };
 
     // Symbol.toStringTag for full Promise compatibility (enables Promise.all, etc.)