{"version":3,"sources":["../../src/ipc/proc_pool.ts"],"sourcesContent":["// SPDX-FileCopyrightText: 2024 LiveKit, Inc.\n//\n// SPDX-License-Identifier: Apache-2.0\nimport { MultiMutex, Mutex } from '@livekit/mutex';\nimport { type Throws, ThrowsPromise } from '@livekit/throws-transformer/throws';\nimport type { RunningJobInfo } from '../job.js';\nimport { Queue } from '../utils.js';\nimport type { InferenceExecutor } from './inference_executor.js';\nimport type { JobExecutor } from './job_executor.js';\nimport { JobProcExecutor } from './job_proc_executor.js';\n\ntype ProcPoolOptions = {\n  agent: string;\n  numIdleProcesses: number;\n  initializeTimeout: number;\n  closeTimeout: number;\n  sessionEndTimeout: number;\n  inferenceExecutor?: InferenceExecutor;\n  memoryWarnMB: number;\n  memoryLimitMB: number;\n};\n\nexport class ProcPool {\n  agent: string;\n  initializeTimeout: number;\n  closeTimeout: number;\n  sessionEndTimeout: number;\n  executors: JobExecutor[] = [];\n  tasks: Promise<void>[] = [];\n  started = false;\n  closed = false;\n  controller = new AbortController();\n  initMutex = new Mutex();\n  procMutex?: MultiMutex;\n  // Keep each lock token paired with its warmed process so MultiMutex slots are always released correctly.\n  warmedProcQueue = new Queue<{ proc: JobExecutor; unlock: () => void }>();\n  inferenceExecutor?: InferenceExecutor;\n  memoryWarnMB: number;\n  memoryLimitMB: number;\n\n  constructor({\n    agent,\n    numIdleProcesses,\n    initializeTimeout,\n    closeTimeout,\n    sessionEndTimeout,\n    inferenceExecutor,\n    memoryWarnMB,\n    memoryLimitMB,\n  }: ProcPoolOptions) {\n    this.agent = agent;\n    if (numIdleProcesses > 0) {\n      this.procMutex = new MultiMutex(numIdleProcesses);\n    }\n    this.initializeTimeout = initializeTimeout;\n    this.closeTimeout = closeTimeout;\n    this.sessionEndTimeout = sessionEndTimeout;\n    this.inferenceExecutor = inferenceExecutor;\n    this.memoryWarnMB = memoryWarnMB;\n    this.memoryLimitMB = memoryLimitMB;\n  }\n\n  get processes(): JobExecutor[] {\n    return this.executors;\n  }\n\n  getByJobId(id: string): JobExecutor | null {\n    return this.executors.find((x) => x.runningJob && x.runningJob.job.id === id) || null;\n  }\n\n  async launchJob(info: RunningJobInfo): Promise<Throws<void, Error>> {\n    let proc: JobExecutor;\n    if (this.procMutex) {\n      const entry = await this.warmedProcQueue.get();\n      proc = entry.proc;\n      // Release exactly the slot that produced this warmed process.\n      entry.unlock();\n    } else {\n      proc = new JobProcExecutor({\n        agent: this.agent,\n        inferenceExecutor: this.inferenceExecutor,\n        initializeTimeout: this.initializeTimeout,\n        closeTimeout: this.closeTimeout,\n        memoryWarnMB: this.memoryWarnMB,\n        memoryLimitMB: this.memoryLimitMB,\n        pingInterval: 2500,\n        pingTimeout: 60000,\n        highPingThreshold: 500,\n      });\n      this.executors.push(proc);\n      await proc.start();\n      await proc.initialize({ sessionEndTimeout: this.sessionEndTimeout });\n    }\n    await proc.launchJob(info);\n  }\n\n  async procWatchTask(procUnlock: () => void) {\n    const proc = new JobProcExecutor({\n      agent: this.agent,\n      inferenceExecutor: this.inferenceExecutor,\n      initializeTimeout: this.initializeTimeout,\n      closeTimeout: this.closeTimeout,\n      memoryWarnMB: this.memoryWarnMB,\n      memoryLimitMB: this.memoryLimitMB,\n      pingInterval: 2500,\n      pingTimeout: 60000,\n      highPingThreshold: 500,\n    });\n\n    try {\n      this.executors.push(proc);\n\n      const unlock = await this.initMutex.lock();\n      let initReleased = false;\n      let procUnlockTransferred = false;\n      try {\n        if (this.closed) {\n          return;\n        }\n\n        await proc.start();\n        try {\n          await proc.initialize({ sessionEndTimeout: this.sessionEndTimeout });\n          await this.warmedProcQueue.put({ proc, unlock: procUnlock });\n          procUnlockTransferred = true;\n          // Release initMutex after enqueue — holding it through join() serialises\n          // the pool to concurrency 1 since child procs are one-shot.\n          unlock();\n          initReleased = true;\n        } catch {\n          // Initialization failed before enqueue, so release the acquired slot immediately.\n        }\n\n        await proc.join();\n      } finally {\n        if (!initReleased) {\n          unlock();\n        }\n        if (!procUnlockTransferred) {\n          procUnlock();\n        }\n      }\n    } finally {\n      const procIndex = this.executors.indexOf(proc);\n      if (procIndex !== -1) {\n        this.executors.splice(procIndex, 1);\n      } else {\n        throw new Error(`proc ${proc} not found in executors`);\n      }\n    }\n  }\n\n  start() {\n    if (this.started) {\n      return;\n    }\n\n    this.started = true;\n    this.run(this.controller.signal);\n  }\n\n  async run(signal: AbortSignal) {\n    if (this.procMutex) {\n      while (!signal.aborted) {\n        const procUnlock = await this.procMutex.lock();\n        const task = this.procWatchTask(procUnlock);\n        this.tasks.push(task);\n        task.finally(() => {\n          const taskIndex = this.tasks.indexOf(task);\n          if (taskIndex !== -1) {\n            this.tasks.splice(taskIndex, 1);\n          } else {\n            throw new Error(`task ${task} not found in tasks`);\n          }\n        });\n      }\n    }\n  }\n\n  async close() {\n    if (!this.started) {\n      return;\n    }\n    this.closed = true;\n    this.controller.abort();\n    const executors = new Set(this.executors);\n    this.warmedProcQueue.items.forEach((e) => {\n      e.unlock();\n      executors.add(e.proc);\n    });\n    const closeTasks = [...executors].map((executor) => executor.close());\n    await ThrowsPromise.allSettled([...closeTasks, ...this.tasks]);\n  }\n}\n"],"mappings":";;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AAGA,mBAAkC;AAClC,oBAA2C;AAE3C,mBAAsB;AAGtB,+BAAgC;AAazB,MAAM,SAAS;AAAA,EACpB;AAAA,EACA;AAAA,EACA;AAAA,EACA;AAAA,EACA,YAA2B,CAAC;AAAA,EAC5B,QAAyB,CAAC;AAAA,EAC1B,UAAU;AAAA,EACV,SAAS;AAAA,EACT,aAAa,IAAI,gBAAgB;AAAA,EACjC,YAAY,IAAI,mBAAM;AAAA,EACtB;AAAA;AAAA,EAEA,kBAAkB,IAAI,mBAAiD;AAAA,EACvE;AAAA,EACA;AAAA,EACA;AAAA,EAEA,YAAY;AAAA,IACV;AAAA,IACA;AAAA,IACA;AAAA,IACA;AAAA,IACA;AAAA,IACA;AAAA,IACA;AAAA,IACA;AAAA,EACF,GAAoB;AAClB,SAAK,QAAQ;AACb,QAAI,mBAAmB,GAAG;AACxB,WAAK,YAAY,IAAI,wBAAW,gBAAgB;AAAA,IAClD;AACA,SAAK,oBAAoB;AACzB,SAAK,eAAe;AACpB,SAAK,oBAAoB;AACzB,SAAK,oBAAoB;AACzB,SAAK,eAAe;AACpB,SAAK,gBAAgB;AAAA,EACvB;AAAA,EAEA,IAAI,YAA2B;AAC7B,WAAO,KAAK;AAAA,EACd;AAAA,EAEA,WAAW,IAAgC;AACzC,WAAO,KAAK,UAAU,KAAK,CAAC,MAAM,EAAE,cAAc,EAAE,WAAW,IAAI,OAAO,EAAE,KAAK;AAAA,EACnF;AAAA,EAEA,MAAM,UAAU,MAAoD;AAClE,QAAI;AACJ,QAAI,KAAK,WAAW;AAClB,YAAM,QAAQ,MAAM,KAAK,gBAAgB,IAAI;AAC7C,aAAO,MAAM;AAEb,YAAM,OAAO;AAAA,IACf,OAAO;AACL,aAAO,IAAI,yCAAgB;AAAA,QACzB,OAAO,KAAK;AAAA,QACZ,mBAAmB,KAAK;AAAA,QACxB,mBAAmB,KAAK;AAAA,QACxB,cAAc,KAAK;AAAA,QACnB,cAAc,KAAK;AAAA,QACnB,eAAe,KAAK;AAAA,QACpB,cAAc;AAAA,QACd,aAAa;AAAA,QACb,mBAAmB;AAAA,MACrB,CAAC;AACD,WAAK,UAAU,KAAK,IAAI;AACxB,YAAM,KAAK,MAAM;AACjB,YAAM,KAAK,WAAW,EAAE,mBAAmB,KAAK,kBAAkB,CAAC;AAAA,IACrE;AACA,UAAM,KAAK,UAAU,IAAI;AAAA,EAC3B;AAAA,EAEA,MAAM,cAAc,YAAwB;AAC1C,UAAM,OAAO,IAAI,yCAAgB;AAAA,MAC/B,OAAO,KAAK;AAAA,MACZ,mBAAmB,KAAK;AAAA,MACxB,mBAAmB,KAAK;AAAA,MACxB,cAAc,KAAK;AAAA,MACnB,cAAc,KAAK;AAAA,MACnB,eAAe,KAAK;AAAA,MACpB,cAAc;AAAA,MACd,aAAa;AAAA,MACb,mBAAmB;AAAA,IACrB,CAAC;AAED,QAAI;AACF,WAAK,UAAU,KAAK,IAAI;AAExB,YAAM,SAAS,MAAM,KAAK,UAAU,KAAK;AACzC,UAAI,eAAe;AACnB,UAAI,wBAAwB;AAC5B,UAAI;AACF,YAAI,KAAK,QAAQ;AACf;AAAA,QACF;AAEA,cAAM,KAAK,MAAM;AACjB,YAAI;AACF,gBAAM,KAAK,WAAW,EAAE,mBAAmB,KAAK,kBAAkB,CAAC;AACnE,gBAAM,KAAK,gBAAgB,IAAI,EAAE,MAAM,QAAQ,WAAW,CAAC;AAC3D,kCAAwB;AAGxB,iBAAO;AACP,yBAAe;AAAA,QACjB,QAAQ;AAAA,QAER;AAEA,cAAM,KAAK,KAAK;AAAA,MAClB,UAAE;AACA,YAAI,CAAC,cAAc;AACjB,iBAAO;AAAA,QACT;AACA,YAAI,CAAC,uBAAuB;AAC1B,qBAAW;AAAA,QACb;AAAA,MACF;AAAA,IACF,UAAE;AACA,YAAM,YAAY,KAAK,UAAU,QAAQ,IAAI;AAC7C,UAAI,cAAc,IAAI;AACpB,aAAK,UAAU,OAAO,WAAW,CAAC;AAAA,MACpC,OAAO;AACL,cAAM,IAAI,MAAM,QAAQ,IAAI,yBAAyB;AAAA,MACvD;AAAA,IACF;AAAA,EACF;AAAA,EAEA,QAAQ;AACN,QAAI,KAAK,SAAS;AAChB;AAAA,IACF;AAEA,SAAK,UAAU;AACf,SAAK,IAAI,KAAK,WAAW,MAAM;AAAA,EACjC;AAAA,EAEA,MAAM,IAAI,QAAqB;AAC7B,QAAI,KAAK,WAAW;AAClB,aAAO,CAAC,OAAO,SAAS;AACtB,cAAM,aAAa,MAAM,KAAK,UAAU,KAAK;AAC7C,cAAM,OAAO,KAAK,cAAc,UAAU;AAC1C,aAAK,MAAM,KAAK,IAAI;AACpB,aAAK,QAAQ,MAAM;AACjB,gBAAM,YAAY,KAAK,MAAM,QAAQ,IAAI;AACzC,cAAI,cAAc,IAAI;AACpB,iBAAK,MAAM,OAAO,WAAW,CAAC;AAAA,UAChC,OAAO;AACL,kBAAM,IAAI,MAAM,QAAQ,IAAI,qBAAqB;AAAA,UACnD;AAAA,QACF,CAAC;AAAA,MACH;AAAA,IACF;AAAA,EACF;AAAA,EAEA,MAAM,QAAQ;AACZ,QAAI,CAAC,KAAK,SAAS;AACjB;AAAA,IACF;AACA,SAAK,SAAS;AACd,SAAK,WAAW,MAAM;AACtB,UAAM,YAAY,IAAI,IAAI,KAAK,SAAS;AACxC,SAAK,gBAAgB,MAAM,QAAQ,CAAC,MAAM;AACxC,QAAE,OAAO;AACT,gBAAU,IAAI,EAAE,IAAI;AAAA,IACtB,CAAC;AACD,UAAM,aAAa,CAAC,GAAG,SAAS,EAAE,IAAI,CAAC,aAAa,SAAS,MAAM,CAAC;AACpE,UAAM,4BAAc,WAAW,CAAC,GAAG,YAAY,GAAG,KAAK,KAAK,CAAC;AAAA,EAC/D;AACF;","names":[]}