<commit_before>import * as _cp from "child_process"
import * as _os from "os"
import * as _p from "process"

import {
  BatchClusterOptions,
  BatchProcessOptions,
  logger
} from "./BatchCluster"
import { Deferred } from "./Deferred"
import { delay } from "./Delay"
import { Task } from "./Task"

export interface BatchProcessObserver {
  onIdle(): void
  onStartError(error: any): void
  retryTask(task: Task<any>, error: any): void
}

export interface InternalBatchProcessOptions
  extends BatchProcessOptions,
    BatchClusterOptions {
  readonly passRE: RegExp
  readonly failRE: RegExp
}

/**
 * BatchProcess manages the care and feeding of a single child process.
 */
export class BatchProcess {
  readonly start = Date.now()
  private _taskCount = -1 // don't count the startupTask
  /**
   * true if `this.end()` has been called, or this process is no longer in the
   * process table.
   */
  private _ended: boolean = false
  /**
   * Supports non-polling notification of process shutdown
   */
  private _exited = new Deferred<void>()
  /**
   * Data from stdout, to be given to _currentTask
   */
  private buff = ""
  /**
   * Should be undefined if this instance is not currently processing a task.
   */
  private currentTask: Task<any> | undefined
  private currentTaskTimeout: NodeJS.Timer | undefined
  private readonly startupTask: Task<any>

  constructor(
    readonly proc: _cp.ChildProcess,
    readonly opts: InternalBatchProcessOptions,
    readonly observer: BatchProcessObserver
  ) {
    // don't let node count the child processes as a reason to stay alive
    this.proc.unref()

    // forking or plumbing issues are not the task's fault, so retry:
    this.proc.on("error", err => this.onError("proc", err))
    this.proc.on("close", () => this.onExit())
    this.proc.on("exit", () => this.onExit())
    this.proc.on("disconnect", () => this.onExit())

    this.proc.stdin.on("error", err => this.onError("stdin", err))

    this.proc.stdout.on("error", err => this.onError("stdout.error", err))
    this.proc.stdout.on("data", d => this.onData(d))

    this.proc.stderr.on("error", err => this.onError("stderr", err))
    this.proc.stderr.on("data", err =>
      this.onError("stderr.data", err.toString())
    )

    this.startupTask = new Task(opts.versionCommand, ea => ea)

    // Prevent unhandled startup task rejections from killing node:
    this.startupTask.promise.catch(() => {
      //
    })

    this.execTask(this.startupTask)
  }

  get pid(): number {
    return this.proc.pid
  }

  get taskCount(): number {
    return this._taskCount
  }

  get exitedPromise(): Promise<void> {
    return this._exited.promise
  }

  get idle(): boolean {
    return !this.ended && this.currentTask == null
  }

  get busy(): boolean {
    return !this.ended && this.currentTask != null
  }

  /**
   * @return true if the child process is in the process table
   */
  get running(): boolean {
    return running(this.pid)
  }

  /**
   * @return {boolean} true if `this.end()` has been requested or the child
   * process has exited.
   */
  get ended(): boolean {
    return this._ended || !this.running
  }

  execTask(task: Task<any>): boolean {
    if (!this.idle) {
      return false
    }
    this._taskCount++
    this.currentTask = task
    const cmd = ensureSuffix(task.command, "\n")
    const timeout =
      task === this.startupTask
        ? this.opts.spawnTimeoutMillis
        : this.opts.taskTimeoutMillis
    if (timeout > 0) {
      this.currentTaskTimeout = setTimeout(() => this.onTimeout(task), timeout)
    }
    this.proc.stdin.write(cmd)
    return true
  }

  async end(gracefully: boolean = true): Promise<void> {
    const tasks = [this.exitedPromise]

    if (!this._ended) {
      this._ended = true
      const cmd = this.opts.exitCommand
        ? ensureSuffix(this.opts.exitCommand, "\n")
        : undefined
      this.proc.stdin.end(cmd)
      if (gracefully) {
        if (this.opts.endGracefulWaitTimeMillis > 0) {
          tasks.push(delay(this.opts.endGracefulWaitTimeMillis))
        }
      }
    }

    if (this.currentTask != null && this.currentTask !== this.startupTask) {
      this.observer.retryTask(this.currentTask, "process end")
    }
    this.clearCurrentTask()

    try {
      this.proc.stdin.end()
      this.proc.stdout.destroy()
      this.proc.stderr.destroy()
      this.proc.disconnect()
    } catch (_) {
      // don't care
    }

    if (this.running) {
      logger().debug(
        `end(${gracefully}): PID ${
          this.pid
        } still running. Waiting for PID to end.`
      )
      await Promise.race(tasks)
    }
    if (this.running) {
      logger().info("end(" + gracefully + "): killing PID " + this.pid)
      kill(this.proc.pid, !gracefully)
    }
    return this.exitedPromise
  }

  private onTimeout(task: Task<any>): void {
    if (task === this.currentTask && task.pending) {
      this.onError(
        "timeout",<commit_msg>more aggressive PID cleanup and logging.<commit_after>import * as _cp from "child_process"
import * as _os from "os"
import * as _p from "process"

import {
  BatchClusterOptions,
  BatchProcessOptions,
  logger
} from "./BatchCluster"
import { Deferred } from "./Deferred"
import { until } from "./Delay"
import { Task } from "./Task"

export interface BatchProcessObserver {
  onIdle(): void
  onStartError(error: any): void
  retryTask(task: Task<any>, error: any): void
}

export interface InternalBatchProcessOptions
  extends BatchProcessOptions,
    BatchClusterOptions {
  readonly passRE: RegExp
  readonly failRE: RegExp
}

/**
 * BatchProcess manages the care and feeding of a single child process.
 */
export class BatchProcess {
  readonly start = Date.now()
  private _taskCount = -1 // don't count the startupTask
  /**
   * true if `this.end()` has been called, or this process is no longer in the
   * process table.
   */
  private _ended: boolean = false
  /**
   * Supports non-polling notification of process shutdown
   */
  private _exited = new Deferred<void>()
  /**
   * Data from stdout, to be given to _currentTask
   */
  private buff = ""
  /**
   * Should be undefined if this instance is not currently processing a task.
   */
  private currentTask: Task<any> | undefined
  private currentTaskTimeout: NodeJS.Timer | undefined
  private readonly startupTask: Task<any>

  constructor(
    readonly proc: _cp.ChildProcess,
    readonly opts: InternalBatchProcessOptions,
    readonly observer: BatchProcessObserver
  ) {
    // don't let node count the child processes as a reason to stay alive
    this.proc.unref()

    // forking or plumbing issues are not the task's fault, so retry:
    this.proc.on("error", err => this.onError("proc", err))
    this.proc.on("close", () => this.onExit())
    this.proc.on("exit", () => this.onExit())
    this.proc.on("disconnect", () => this.onExit())

    this.proc.stdin.on("error", err => this.onError("stdin", err))

    this.proc.stdout.on("error", err => this.onError("stdout.error", err))
    this.proc.stdout.on("data", d => this.onData(d))

    this.proc.stderr.on("error", err => this.onError("stderr", err))
    this.proc.stderr.on("data", err =>
      this.onError("stderr.data", err.toString())
    )

    this.startupTask = new Task(opts.versionCommand, ea => ea)

    // Prevent unhandled startup task rejections from killing node:
    this.startupTask.promise.catch(() => {
      //
    })

    this.execTask(this.startupTask)
  }

  get pid(): number {
    return this.proc.pid
  }

  get taskCount(): number {
    return this._taskCount
  }

  get exitedPromise(): Promise<void> {
    return this._exited.promise
  }

  get idle(): boolean {
    return !this.ended && this.currentTask == null
  }

  get busy(): boolean {
    return !this.ended && this.currentTask != null
  }

  /**
   * @return true if the child process is in the process table
   */
  get running(): boolean {
    return running(this.pid)
  }

  /**
   * @return {boolean} true if `this.end()` has been requested or the child
   * process has exited.
   */
  get ended(): boolean {
    return this._ended || !this.running
  }

  execTask(task: Task<any>): boolean {
    if (!this.idle) {
      return false
    }
    this._taskCount++
    this.currentTask = task
    const cmd = ensureSuffix(task.command, "\n")
    const timeout =
      task === this.startupTask
        ? this.opts.spawnTimeoutMillis
        : this.opts.taskTimeoutMillis
    if (timeout > 0) {
      this.currentTaskTimeout = setTimeout(() => this.onTimeout(task), timeout)
    }
    this.proc.stdin.write(cmd)
    return true
  }

  async end(gracefully: boolean = true): Promise<void> {
    if (!this._ended) {
      this._ended = true
      const cmd = this.opts.exitCommand
        ? ensureSuffix(this.opts.exitCommand, "\n")
        : undefined
      this.proc.stdin.end(cmd)
    }

    if (this.currentTask != null && this.currentTask !== this.startupTask) {
      this.observer.retryTask(this.currentTask, "process end")
    }
    this.clearCurrentTask()

    try {
      this.proc.stdin.end()
      this.proc.stdout.destroy()
      this.proc.stderr.destroy()
      this.proc.disconnect()
    } catch (_) {
      // don't care
    }

    if (this.running) {
      if (gracefully && this.opts.endGracefulWaitTimeMillis > 0) {
        await this.awaitNotRunning(this.opts.endGracefulWaitTimeMillis)
      }
    }
    if (this.running) {
      logger().info("end(): killing PID " + this.pid)
      kill(this.proc.pid, true)
    }
    // The OS is srsly f@rked if `kill` doesn't respond within a couple ms. 5s
    // should be 100x longer than necessary.
    if (!await this.awaitNotRunning(5000)) {
      logger().error("end(): PID " + this.pid + " did not respond to kill.")
    }

    return this.exitedPromise
  }

  private awaitNotRunning(timeout: number) {
    return until(() => !this.running, timeout)
  }

  private onTimeout(task: Task<any>): void {
    if (task === this.currentTask && task.pending) {
      this.onError(
        "timeout",
{'fraction_non_alphanumeric': 0.13187250996015937, 'fraction_numerical': 0.0013944223107569722, 'mean_word_length': 2.828059473884865, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 48, 'https://': 0, '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': '3263714', 'n_tokens_mistral': 3175, 'n_tokens_neox': 3107, 'n_words': 821}