startCheckers

tcmts.cli.checkers.startCheckers · #2hye3szf · term · runs on the server

uses node:child_process

size checker processes running command internal-check. While one works on a request it keeps this process running; idle, it doesn't (and goes when this process does). One that ends fails what it was asked, and is started again for the next.

import type { CheckerPool, CheckRequest, CheckResult } from "#codebase/tcmts/cli/checkers";
import { spawn } from "node:child_process";
import type { ChildProcess } from "node:child_process";

export function startCheckers(command: string[], size: number): CheckerPool {
  type Slot = {
    child?: ChildProcess;
    /** what's come of the reply being read, a chunk at a time (a reply is tens of megabytes) */
    buf: string[];
    job?: { resolve: (r: CheckResult) => void; reject: (e: Error) => void };
  };
  const slots: Slot[] = Array.from({ length: size }, () => ({ buf: [] }));
  const waiting: (() => void)[] = [];
  const idle = (s: Slot) => {
    s.child?.unref();
    (s.child?.stdout as { unref?: () => void } | null)?.unref?.();
    (s.child?.stdin as { unref?: () => void } | null)?.unref?.();
    waiting.shift()?.();
  };
  const start = (s: Slot) => {
    const child = spawn(command[0]!, [...command.slice(1), "internal-check"], {
      stdio: ["pipe", "pipe", "inherit"],
      env: process.env,
    });
    s.child = child;
    s.buf = [];
    child.stdout!.setEncoding("utf8").on("data", (d: string) => {
      // (a line ends in what just came, if anywhere: what came before isn't searched again)
      for (let rest = d, nl = rest.indexOf("\n"); ; nl = rest.indexOf("\n")) {
        if (nl < 0) {
          if (rest) s.buf.push(rest);
          break;
        }
        const line = s.buf.join("") + rest.slice(0, nl);
        s.buf = [];
        rest = rest.slice(nl + 1);
        if (!line.startsWith("\0check ")) continue;
        const job = s.job;
        s.job = undefined;
        const r = JSON.parse(line.slice(7)) as CheckResult | { error: string };
        if ("error" in r) job?.reject(new Error(r.error));
        else job?.resolve(r);
        idle(s);
      }
    });
    child.on("exit", (code) => {
      if (s.child !== child) return;
      s.child = undefined;
      const job = s.job;
      s.job = undefined;
      job?.reject(new Error(`a checker process ended (${code})`));
      idle(s);
    });
    child.on("error", () => {
      // (said by its exit)
    });
  };
  process.on("exit", () => {
    for (const s of slots) s.child?.kill();
  });
  const check = (req: CheckRequest): Promise<CheckResult> => {
    const s = slots.find((x) => !x.job);
    if (!s) return new Promise<void>((r) => waiting.push(r)).then(() => check(req));
    if (!s.child) start(s);
    const child = s.child!;
    child.ref();
    (child.stdout as { ref?: () => void } | null)?.ref?.();
    return new Promise<CheckResult>((resolve, reject) => {
      s.job = { resolve, reject };
      child.stdin!.write(`${JSON.stringify(req)}\n`);
    });
  };
  return { size, check };
}

uses

used by