- protected workerListener (): (message: MessageValue<Response>) => void {
- return message => {
- if (message.workerId != null && message.started != null) {
- // Worker started message received
- this.workerNodes[
- this.getWorkerNodeKey(this.getWorkerById(message.workerId) as Worker)
- ].info.started = message.started
- } else if (message.id != null) {
- // Task execution response received
- const promiseResponse = this.promiseResponseMap.get(message.id)
- if (promiseResponse != null) {
- if (message.taskError != null) {
- if (this.emitter != null) {
- this.emitter.emit(PoolEvents.taskError, message.taskError)
- }
- promiseResponse.reject(message.taskError.message)
- } else {
- promiseResponse.resolve(message.data as Response)
- }
- this.afterTaskExecutionHook(promiseResponse.worker, message)
- this.promiseResponseMap.delete(message.id)
- const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
+ protected abstract registerWorkerMessageListener<
+ Message extends Data | Response
+ >(
+ workerNodeKey: number,
+ listener: (message: MessageValue<Message>) => void
+ ): void
+
+ /**
+ * Method hooked up after a worker node has been newly created.
+ * Can be overridden.
+ *
+ * @param workerNodeKey - The newly created worker node key.
+ */
+ protected afterWorkerNodeSetup (workerNodeKey: number): void {
+ // Listen to worker messages.
+ this.registerWorkerMessageListener(workerNodeKey, this.workerListener())
+ // Send the startup message to worker.
+ this.sendStartupMessageToWorker(workerNodeKey)
+ // Send the worker statistics message to worker.
+ this.sendWorkerStatisticsMessageToWorker(workerNodeKey)
+ }
+
+ /**
+ * Sends the startup message to worker given its worker node key.
+ *
+ * @param workerNodeKey - The worker node key.
+ */
+ protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
+
+ /**
+ * Sends the worker statistics message to worker given its worker node key.
+ *
+ * @param workerNodeKey - The worker node key.
+ */
+ private sendWorkerStatisticsMessageToWorker (workerNodeKey: number): void {
+ this.sendToWorker(workerNodeKey, {
+ statistics: {
+ runTime:
+ this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ .runTime.aggregate,
+ elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ .elu.aggregate
+ },
+ workerId: this.getWorkerInfo(workerNodeKey).id as number
+ })
+ }
+
+ private redistributeQueuedTasks (workerNodeKey: number): void {
+ while (this.tasksQueueSize(workerNodeKey) > 0) {
+ let targetWorkerNodeKey: number = workerNodeKey
+ let minQueuedTasks = Infinity
+ let executeTask = false
+ for (const [workerNodeId, workerNode] of this.workerNodes.entries()) {
+ const workerInfo = this.getWorkerInfo(workerNodeId)
+ if (
+ workerNodeId !== workerNodeKey &&
+ workerInfo.ready &&
+ workerNode.usage.tasks.queued === 0
+ ) {