slotPool

tcmts.cli.budget.slotPool · #4gfo5opb · term · runs on the server

uses node:fs

Slots shared by every process using dir: size files there (slot-<i>), each held by the process that made it (its pid in it), and a line of those waiting (line-<time>-…), served first come first served. A slot or a place in line whose process is gone is nobody's: taken over by the next to look. What a process holds is given back when it exits, too.

import { alive } from "#codebase/tcmts/cli/budget";
import type { SlotHolder, SlotPool } from "#codebase/tcmts/cli/budget";
import crypto from "node:crypto";
import fs from "node:fs";
import path from "node:path";

export function slotPool(dir: string, size: number): SlotPool {
  const held = new Map<string, string>();
  const read = (file: string): SlotHolder | undefined => {
    try {
      return JSON.parse(fs.readFileSync(file, "utf8")) as SlotHolder;
    } catch {
      return undefined;
    }
  };
  const ours = (file: string, token: string) => read(file)?.token === token;
  const onExit = () => {
    for (const [file, token] of held) if (ours(file, token)) fs.rmSync(file, { force: true });
  };
  const hold = (file: string, token: string) => {
    if (!held.size) process.on("exit", onExit);
    held.set(file, token);
  };
  const drop = (file: string, token: string) => {
    if (ours(file, token)) fs.rmSync(file, { force: true });
    held.delete(file);
    if (!held.size) process.off("exit", onExit);
  };
  // who holds `file` (a slot, or a place in line), if anyone: one whose process is gone holds nothing, and its file
  // goes; one just made, not written yet, is its maker's
  const holder = (file: string): SlotHolder | undefined => {
    const h = read(file);
    if (h && alive(h.pid)) return h;
    if (!h) {
      try {
        if (Date.now() - fs.statSync(file).mtimeMs < 2000)
          return { pid: 0, what: "a process starting", since: Date.now(), token: "" };
      } catch {
        return undefined;
      }
    }
    fs.rmSync(file, { force: true });
    return undefined;
  };
  let seq = 0;
  return {
    size,
    async take(what, onWait, count = 1) {
      const want = Math.max(1, Math.min(count, size));
      fs.mkdirSync(dir, { recursive: true });
      const token = crypto.randomBytes(8).toString("hex");
      const me = JSON.stringify({
        pid: process.pid,
        what,
        since: Date.now(),
        token,
        want,
      } satisfies SlotHolder);
      const place = `line-${String(Date.now()).padStart(15, "0")}-${String(process.pid).padStart(10, "0")}-${String(seq++).padStart(6, "0")}`;
      const line = path.join(dir, place);
      fs.writeFileSync(line, me);
      hold(line, token);
      const slots = Array.from({ length: size }, (_, i) => path.join(dir, `slot-${i}`));
      try {
        for (;;) {
          // (what those ahead wait for)
          let ahead = 0;
          let wanted = 0;
          for (const n of fs.readdirSync(dir))
            if (n.startsWith("line-") && n < place) {
              const h = holder(path.join(dir, n));
              if (h) {
                ahead++;
                wanted += h.want ?? 1;
              }
            }
          const holders = slots.map((s) => (fs.existsSync(s) ? holder(s) : undefined));
          const free = slots.filter((_, i) => !holders[i]);
          // (those ahead are served first)
          if (free.length - wanted >= want) {
            const got: string[] = [];
            for (const s of free) {
              if (got.length === want) break;
              try {
                fs.writeFileSync(s, me, { flag: "wx" });
                hold(s, token);
                got.push(s);
              } catch (e) {
                if ((e as NodeJS.ErrnoException).code !== "EEXIST") throw e;
              }
            }
            if (got.length === want)
              return () => {
                for (const s of got) drop(s, token);
              };
            // (another took one first: all or none, so two waiting for several don't each hold some)
            for (const s of got) drop(s, token);
          }
          onWait?.({
            holders: holders.filter((h): h is SlotHolder => !!h),
            ahead,
            want,
            size,
          });
          await new Promise((r) => setTimeout(r, 100 + Math.random() * 100));
        }
      } finally {
        drop(line, token);
      }
    },
  };
}

uses

used by