fix: fix worker node cross tasks stealing

[poolifier.git] / src / pools / worker.ts

diff --git a/src/pools/worker.ts b/src/pools/worker.ts
index 03cecce849a2cb9d71ad04a566795d3588238509..9c8b77fc623b68ed2a25aed883531dfa5e25cf70 100644 (file)
--- a/src/pools/worker.ts
+++ b/src/pools/worker.ts
@@ -1,14 +1,19 @@
-import type { MessageChannel } from 'node:worker_threads'
+import type { MessageChannel, WorkerOptions } from 'node:worker_threads'
+import type { EventEmitter } from 'node:events'
 import type { CircularArray } from '../circular-array'
 import type { Task } from '../utility-types'
 
 /**
  * Callback invoked when the worker has started successfully.
+ *
+ * @typeParam Worker - Type of worker.
  */
 export type OnlineHandler<Worker extends IWorker> = (this: Worker) => void
 
 /**
  * Callback invoked if the worker has received a message.
+ *
+ * @typeParam Worker - Type of worker.
  */
 export type MessageHandler<Worker extends IWorker> = (
   this: Worker,
@@ -17,6 +22,8 @@ export type MessageHandler<Worker extends IWorker> = (
 
 /**
  * Callback invoked if the worker raised an error.
+ *
+ * @typeParam Worker - Type of worker.
  */
 export type ErrorHandler<Worker extends IWorker> = (
   this: Worker,
@@ -25,6 +32,8 @@ export type ErrorHandler<Worker extends IWorker> = (
 
 /**
  * Callback invoked when the worker exits successfully.
+ *
+ * @typeParam Worker - Type of worker.
  */
 export type ExitHandler<Worker extends IWorker> = (
   this: Worker,
@@ -96,6 +105,10 @@ export interface TaskStatistics {
    * Maximum number of queued tasks.
    */
   readonly maxQueued?: number
+  /**
+   * Number of sequentially stolen tasks.
+   */
+  sequentiallyStolen: number
   /**
    * Number of stolen tasks.
    */
@@ -141,6 +154,11 @@ export interface WorkerInfo {
    * Ready flag.
    */
   ready: boolean
+  /**
+   * Stealing flag.
+   * This flag is set to `true` when worker node is stealing tasks from another worker node.
+   */
+  stealing: boolean
   /**
    * Task function names.
    */
@@ -185,36 +203,65 @@ export interface StrategyData {
  */
 export interface IWorker {
   /**
-   * Worker id.
+   * Cluster worker id.
    */
   readonly id?: number
+  /**
+   * Worker thread worker id.
+   */
   readonly threadId?: number
   /**
-   * Registers an event listener.
+   * Registers an event handler.
    *
    * @param event - The event.
    * @param handler - The event handler.
    */
-  readonly on: ((event: 'online', handler: OnlineHandler<this>) => void) &
-  ((event: 'message', handler: MessageHandler<this>) => void) &
-  ((event: 'error', handler: ErrorHandler<this>) => void) &
-  ((event: 'exit', handler: ExitHandler<this>) => void)
+  readonly on: (
+    event: string,
+    handler:
+    | OnlineHandler<this>
+    | MessageHandler<this>
+    | ErrorHandler<this>
+    | ExitHandler<this>
+  ) => void
   /**
-   * Registers a listener to the exit event that will only be performed once.
+   * Registers once an event handler.
    *
-   * @param event - The `'exit'` event.
-   * @param handler - The exit handler.
+   * @param event - The event.
+   * @param handler - The event handler.
+   */
+  readonly once: (
+    event: string,
+    handler:
+    | OnlineHandler<this>
+    | MessageHandler<this>
+    | ErrorHandler<this>
+    | ExitHandler<this>
+  ) => void
+  /**
+   * Stop all JavaScript execution in the worker thread as soon as possible.
+   * Returns a Promise for the exit code that is fulfilled when the `'exit' event` is emitted.
+   */
+  readonly terminate?: () => Promise<number>
+  /**
+   * Cluster worker disconnect.
+   */
+  readonly disconnect?: () => void
+  /**
+   * Cluster worker kill.
    */
-  readonly once: (event: 'exit', handler: ExitHandler<this>) => void
+  readonly kill?: (signal?: string) => void
 }
 
 /**
- * Worker node event detail.
+ * Worker node options.
  *
  * @internal
  */
-export interface WorkerNodeEventDetail {
-  workerId: number
+export interface WorkerNodeOptions {
+  workerOptions?: WorkerOptions
+  env?: Record<string, unknown>
+  tasksQueueBackPressureSize: number
 }
 
 /**
@@ -225,7 +272,7 @@ export interface WorkerNodeEventDetail {
  * @internal
  */
 export interface IWorkerNode<Worker extends IWorker, Data = unknown>
-  extends EventTarget {
+  extends EventEmitter {
   /**
    * Worker.
    */
@@ -244,7 +291,7 @@ export interface IWorkerNode<Worker extends IWorker, Data = unknown>
    */
   strategyData?: StrategyData
   /**
-   * Message channel (worker_threads only).
+   * Message channel (worker thread only).
    */
   readonly messageChannel?: MessageChannel
   /**
@@ -299,9 +346,37 @@ export interface IWorkerNode<Worker extends IWorker, Data = unknown>
    */
   readonly resetUsage: () => void
   /**
-   * Closes communication channel.
+   * Terminates the worker node.
+   */
+  readonly terminate: () => Promise<void>
+  /**
+   * Registers a worker event handler.
+   *
+   * @param event - The event.
+   * @param handler - The event handler.
    */
-  readonly closeChannel: () => void
+  readonly registerWorkerEventHandler: (
+    event: string,
+    handler:
+    | OnlineHandler<Worker>
+    | MessageHandler<Worker>
+    | ErrorHandler<Worker>
+    | ExitHandler<Worker>
+  ) => void
+  /**
+   * Registers once a worker event handler.
+   *
+   * @param event - The event.
+   * @param handler - The event handler.
+   */
+  readonly registerOnceWorkerEventHandler: (
+    event: string,
+    handler:
+    | OnlineHandler<Worker>
+    | MessageHandler<Worker>
+    | ErrorHandler<Worker>
+    | ExitHandler<Worker>
+  ) => void
   /**
    * Gets task function worker usage statistics.
    *
@@ -317,3 +392,13 @@ export interface IWorkerNode<Worker extends IWorker, Data = unknown>
    */
   readonly deleteTaskFunctionWorkerUsage: (name: string) => boolean
 }
+
+/**
+ * Worker node event detail.
+ *
+ * @internal
+ */
+export interface WorkerNodeEventDetail {
+  workerId: number
+  workerNodeKey?: number
+}