-
-
Notifications
You must be signed in to change notification settings - Fork 6.5k
/
index.ts
158 lines (136 loc) · 4.88 KB
/
index.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
/**
* Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
*
* This source code is licensed under the MIT license found in the
* LICENSE file in the root directory of this source tree.
*/
import {cpus} from 'os';
import Farm from './Farm';
import WorkerPool from './WorkerPool';
import type {
FarmOptions,
PoolExitResult,
PromiseWithCustomMessage,
TaskQueue,
WorkerPoolInterface,
WorkerPoolOptions,
} from './types';
export {default as PriorityQueue} from './PriorityQueue';
export {default as FifoQueue} from './FifoQueue';
export {default as messageParent} from './workers/messageParent';
export type {PromiseWithCustomMessage, TaskQueue};
function getExposedMethods(
workerPath: string,
options: FarmOptions,
): ReadonlyArray<string> {
let exposedMethods = options.exposedMethods;
// If no methods list is given, try getting it by auto-requiring the module.
if (!exposedMethods) {
const module: Record<string, unknown> = require(workerPath);
exposedMethods = Object.keys(module).filter(
name => typeof module[name] === 'function',
);
if (typeof module === 'function') {
exposedMethods = [...exposedMethods, 'default'];
}
}
return exposedMethods;
}
/**
* The Jest farm (publicly called "Worker") is a class that allows you to queue
* methods across multiple child processes, in order to parallelize work. This
* is done by providing an absolute path to a module that will be loaded on each
* of the child processes, and bridged to the main process.
*
* Bridged methods are specified by using the "exposedMethods" property of the
* "options" object. This is an array of strings, where each of them corresponds
* to the exported name in the loaded module.
*
* You can also control the amount of workers by using the "numWorkers" property
* of the "options" object, and the settings passed to fork the process through
* the "forkOptions" property. The amount of workers defaults to the amount of
* CPUS minus one.
*
* Queueing calls can be done in two ways:
* - Standard method: calls will be redirected to the first available worker,
* so they will get executed as soon as they can.
*
* - Sticky method: if a "computeWorkerKey" method is provided within the
* config, the resulting string of this method will be used as a key.
* Every time this key is returned, it is guaranteed that your job will be
* processed by the same worker. This is specially useful if your workers
* are caching results.
*/
export class Worker {
private _ending: boolean;
private _farm: Farm;
private _options: FarmOptions;
private _workerPool: WorkerPoolInterface;
constructor(workerPath: string, options?: FarmOptions) {
this._options = {...options};
this._ending = false;
const workerPoolOptions: WorkerPoolOptions = {
enableWorkerThreads: this._options.enableWorkerThreads ?? false,
forkOptions: this._options.forkOptions ?? {},
maxRetries: this._options.maxRetries ?? 3,
numWorkers: this._options.numWorkers ?? Math.max(cpus().length - 1, 1),
resourceLimits: this._options.resourceLimits ?? {},
setupArgs: this._options.setupArgs ?? [],
};
if (this._options.WorkerPool) {
this._workerPool = new this._options.WorkerPool(
workerPath,
workerPoolOptions,
);
} else {
this._workerPool = new WorkerPool(workerPath, workerPoolOptions);
}
this._farm = new Farm(
workerPoolOptions.numWorkers,
this._workerPool.send.bind(this._workerPool),
{
computeWorkerKey: this._options.computeWorkerKey,
taskQueue: this._options.taskQueue,
workerSchedulingPolicy: this._options.workerSchedulingPolicy,
},
);
this._bindExposedWorkerMethods(workerPath, this._options);
}
private _bindExposedWorkerMethods(
workerPath: string,
options: FarmOptions,
): void {
getExposedMethods(workerPath, options).forEach(name => {
if (name.startsWith('_')) {
return;
}
if (this.constructor.prototype.hasOwnProperty(name)) {
throw new TypeError('Cannot define a method called ' + name);
}
// @ts-expect-error: dynamic extension of the class instance is expected.
this[name] = this._callFunctionWithArgs.bind(this, name);
});
}
private _callFunctionWithArgs(
method: string,
...args: Array<unknown>
): Promise<unknown> {
if (this._ending) {
throw new Error('Farm is ended, no more calls can be done to it');
}
return this._farm.doWork(method, ...args);
}
getStderr(): NodeJS.ReadableStream {
return this._workerPool.getStderr();
}
getStdout(): NodeJS.ReadableStream {
return this._workerPool.getStdout();
}
async end(): Promise<PoolExitResult> {
if (this._ending) {
throw new Error('Farm is ended, no more calls can be done to it');
}
this._ending = true;
return this._workerPool.end();
}
}