<commit_before>/**
 * @license
 * Copyright Google Inc. All Rights Reserved.
 *
 * Use of this source code is governed by an MIT-style license that can be
 * found in the LICENSE file at https://angular.io/license
 */

/// <reference types="node" />

import * as cluster from 'cluster';

import {resolve} from '../../../../src/ngtsc/file_system';
import {Logger} from '../../logging/logger';
import {PackageJsonUpdater} from '../../writing/package_json_updater';
import {AnalyzeEntryPointsFn} from '../api';
import {CreateTaskCompletedCallback, Task, TaskCompletedCallback, TaskQueue} from '../tasks/api';
import {stringifyTask} from '../tasks/utils';

import {MessageFromWorker, TaskCompletedMessage, UpdatePackageJsonMessage} from './api';
import {Deferred, sendMessageToWorker} from './utils';


/**
 * The cluster master is responsible for analyzing all entry-points, planning the work that needs to
 * be done, distributing it to worker-processes and collecting/post-processing the results.
 */
export class ClusterMaster {
  private finishedDeferred = new Deferred<void>();
  private processingStartTime: number = -1;
  private taskAssignments = new Map<number, Task|null>();
  private taskQueue: TaskQueue;
  private onTaskCompleted: TaskCompletedCallback;

  constructor(
      private workerCount: number, private logger: Logger,
      private pkgJsonUpdater: PackageJsonUpdater, analyzeEntryPoints: AnalyzeEntryPointsFn,
      createTaskCompletedCallback: CreateTaskCompletedCallback) {
    if (!cluster.isMaster) {
      throw new Error('Tried to instantiate `ClusterMaster` on a worker process.');
    }

    this.taskQueue = analyzeEntryPoints();
    this.onTaskCompleted = createTaskCompletedCallback(this.taskQueue);
  }

  run(): Promise<void> {
    if (this.taskQueue.allTasksCompleted) {
      return Promise.resolve();
    }

    // Set up listeners for worker events (emitted on `cluster`).
    cluster.on('online', this.wrapEventHandler(worker => this.onWorkerOnline(worker.id)));

    cluster.on(
        'message', this.wrapEventHandler((worker, msg) => this.onWorkerMessage(worker.id, msg)));

    cluster.on(
        'exit',
        this.wrapEventHandler((worker, code, signal) => this.onWorkerExit(worker, code, signal)));

    // Since we have pending tasks at the very minimum we need a single worker.
    cluster.fork();

    return this.finishedDeferred.promise.then(() => this.stopWorkers(), err => {
      this.stopWorkers();
      return Promise.reject(err);
    });
  }

  /** Try to find available (idle) workers and assign them available (non-blocked) tasks. */
  private maybeDistributeWork(): void {
    let isWorkerAvailable = false;

    // First, check whether all tasks have been completed.
    if (this.taskQueue.allTasksCompleted) {
      const duration = Math.round((Date.now() - this.processingStartTime) / 100) / 10;
      this.logger.debug(`Processed tasks in ${duration}s.`);

      return this.finishedDeferred.resolve();
    }

    // Look for available workers and available tasks to assign to them.
    for (const [workerId, assignedTask] of Array.from(this.taskAssignments)) {
      if (assignedTask !== null) {
        // This worker already has a job; check other workers.
        continue;
      } else {
        // This worker is available.
        isWorkerAvailable = true;
      }

      // This worker needs a job. See if any are available.
      const task = this.taskQueue.getNextTask();
      if (task === null) {
        // No suitable work available right now.
        break;
      }

      // Process the next task on the worker.
      this.taskAssignments.set(workerId, task);
      sendMessageToWorker(workerId, {type: 'process-task', task});

      isWorkerAvailable = false;
    }

    if (!isWorkerAvailable) {
      const spawnedWorkerCount = Object.keys(cluster.workers).length;
      if (spawnedWorkerCount < this.workerCount) {
        this.logger.debug('Spawning another worker process as there is more work to be done.');
        cluster.fork();
      } else {
        // If there are no available workers or no available tasks, log (for debugging purposes).
        this.logger.debug(
            `All ${spawnedWorkerCount} workers are currently busy and cannot take on more work.`);
      }
    } else {
      const busyWorkers = Array.from(this.taskAssignments)
                              .filter(([_workerId, task]) => task !== null)
                              .map(([workerId]) => workerId);
      const totalWorkerCount = this.taskAssignments.size;
      const idleWorkerCount = totalWorkerCount - busyWorkers.length;

      this.logger.debug(
          `No assignments for ${idleWorkerCount} idle (out of ${totalWorkerCount} total) ` +
          `workers. Busy workers: ${busyWorkers.join(', ')}`);

      if (busyWorkers.length === 0) {
        // This is a bug:
        // All workers are idle (meaning no tasks are in progress) and `taskQueue.allTasksCompleted`<commit_msg>refactor(ngcc): rename `workerCount` to `maxWorkerCount` (#36298)<commit_after>/**
 * @license
 * Copyright Google Inc. All Rights Reserved.
 *
 * Use of this source code is governed by an MIT-style license that can be
 * found in the LICENSE file at https://angular.io/license
 */

/// <reference types="node" />

import * as cluster from 'cluster';

import {resolve} from '../../../../src/ngtsc/file_system';
import {Logger} from '../../logging/logger';
import {PackageJsonUpdater} from '../../writing/package_json_updater';
import {AnalyzeEntryPointsFn} from '../api';
import {CreateTaskCompletedCallback, Task, TaskCompletedCallback, TaskQueue} from '../tasks/api';
import {stringifyTask} from '../tasks/utils';

import {MessageFromWorker, TaskCompletedMessage, UpdatePackageJsonMessage} from './api';
import {Deferred, sendMessageToWorker} from './utils';


/**
 * The cluster master is responsible for analyzing all entry-points, planning the work that needs to
 * be done, distributing it to worker-processes and collecting/post-processing the results.
 */
export class ClusterMaster {
  private finishedDeferred = new Deferred<void>();
  private processingStartTime: number = -1;
  private taskAssignments = new Map<number, Task|null>();
  private taskQueue: TaskQueue;
  private onTaskCompleted: TaskCompletedCallback;

  constructor(
      private maxWorkerCount: number, private logger: Logger,
      private pkgJsonUpdater: PackageJsonUpdater, analyzeEntryPoints: AnalyzeEntryPointsFn,
      createTaskCompletedCallback: CreateTaskCompletedCallback) {
    if (!cluster.isMaster) {
      throw new Error('Tried to instantiate `ClusterMaster` on a worker process.');
    }

    this.taskQueue = analyzeEntryPoints();
    this.onTaskCompleted = createTaskCompletedCallback(this.taskQueue);
  }

  run(): Promise<void> {
    if (this.taskQueue.allTasksCompleted) {
      return Promise.resolve();
    }

    // Set up listeners for worker events (emitted on `cluster`).
    cluster.on('online', this.wrapEventHandler(worker => this.onWorkerOnline(worker.id)));

    cluster.on(
        'message', this.wrapEventHandler((worker, msg) => this.onWorkerMessage(worker.id, msg)));

    cluster.on(
        'exit',
        this.wrapEventHandler((worker, code, signal) => this.onWorkerExit(worker, code, signal)));

    // Since we have pending tasks at the very minimum we need a single worker.
    cluster.fork();

    return this.finishedDeferred.promise.then(() => this.stopWorkers(), err => {
      this.stopWorkers();
      return Promise.reject(err);
    });
  }

  /** Try to find available (idle) workers and assign them available (non-blocked) tasks. */
  private maybeDistributeWork(): void {
    let isWorkerAvailable = false;

    // First, check whether all tasks have been completed.
    if (this.taskQueue.allTasksCompleted) {
      const duration = Math.round((Date.now() - this.processingStartTime) / 100) / 10;
      this.logger.debug(`Processed tasks in ${duration}s.`);

      return this.finishedDeferred.resolve();
    }

    // Look for available workers and available tasks to assign to them.
    for (const [workerId, assignedTask] of Array.from(this.taskAssignments)) {
      if (assignedTask !== null) {
        // This worker already has a job; check other workers.
        continue;
      } else {
        // This worker is available.
        isWorkerAvailable = true;
      }

      // This worker needs a job. See if any are available.
      const task = this.taskQueue.getNextTask();
      if (task === null) {
        // No suitable work available right now.
        break;
      }

      // Process the next task on the worker.
      this.taskAssignments.set(workerId, task);
      sendMessageToWorker(workerId, {type: 'process-task', task});

      isWorkerAvailable = false;
    }

    if (!isWorkerAvailable) {
      const spawnedWorkerCount = Object.keys(cluster.workers).length;
      if (spawnedWorkerCount < this.maxWorkerCount) {
        this.logger.debug('Spawning another worker process as there is more work to be done.');
        cluster.fork();
      } else {
        // If there are no available workers or no available tasks, log (for debugging purposes).
        this.logger.debug(
            `All ${spawnedWorkerCount} workers are currently busy and cannot take on more work.`);
      }
    } else {
      const busyWorkers = Array.from(this.taskAssignments)
                              .filter(([_workerId, task]) => task !== null)
                              .map(([workerId]) => workerId);
      const totalWorkerCount = this.taskAssignments.size;
      const idleWorkerCount = totalWorkerCount - busyWorkers.length;

      this.logger.debug(
          `No assignments for ${idleWorkerCount} idle (out of ${totalWorkerCount} total) ` +
          `workers. Busy workers: ${busyWorkers.join(', ')}`);

      if (busyWorkers.length === 0) {
        // This is a bug:
        // All workers are idle (meaning no tasks are in progress) and `taskQueue.allTasksCompleted`
{'fraction_non_alphanumeric': 0.116693354683747, 'fraction_numerical': 0.0019015212169735788, 'mean_word_length': 3.082107843137255, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 25, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5524665', 'n_tokens_mistral': 2714, 'n_tokens_neox': 2667, 'n_words': 844}