reimplemented sechduler

src/routes/invoke.ts

@@ -23,6 +23,7 @@ invokeRouter.use("/:functionId", async (req, res) => {
         subPath,
         resolve,
         reject,
+        enqueuedAt: performance.now(),
       });
     });
 

src/runtime/cleanup.test.ts

@@ -2,6 +2,7 @@ import { describe, it, expect, vi, beforeEach } from "vitest";
 import { cleanupVm } from "./cleanup.js";
 import fs from "fs";
 import type { RuntimeFunction, Vm } from "../types/types.js";
+import { Deque } from "./deque.js";
 
 function makeVm(overrides = {}): Vm {
   return {
@@ -16,7 +17,17 @@ function makeVm(overrides = {}): Vm {
 }
 
 function makeFn(vms: Vm[]): RuntimeFunction {
-  return { functionId: "fn1", queue: [], vms, processing: false };
+  const readyVms = new Set<Vm>(vms.filter((v) => v.state === "ready"));
+  return {
+    functionId: "fn1",
+    queue: new Deque(),
+    vms,
+    readyVms,
+    weight: 0,
+    deficit: 0,
+    inflightCount: 0,
+    pendingCreations: 0,
+  };
 }
 
 describe("cleanupVm", () => {
@@ -43,4 +54,3 @@ describe("cleanupVm", () => {
     expect(vm.firecrackerProcess.kill).not.toHaveBeenCalled();
   });
 });
-

src/runtime/cleanup.ts

@@ -1,14 +1,18 @@
 import fs from "fs";
 import { cleanupLogger } from "../utils/logger.js";
 import { vmCleanupTotal, vmCount } from "../utils/metrics.js";
+import { notifyVmDestroyed } from "./scheduler.js";
 
 import type { RuntimeFunction, Vm } from "../types/types.js";
 
 export async function cleanupVm(fn: RuntimeFunction, vm: Vm) {
   if (vm.cleaned) return;
 
   vm.cleaned = true;
-  cleanupLogger.info({ functionId: fn.functionId, vmId: vm.id }, "cleaning up VM");
+  cleanupLogger.info(
+    { functionId: fn.functionId, vmId: vm.id },
+    "cleaning up VM",
+  );
 
   try {
     vm.firecrackerProcess.kill();
@@ -28,8 +32,10 @@ export async function cleanupVm(fn: RuntimeFunction, vm: Vm) {
   } catch {}
 
   fn.vms = fn.vms.filter((v) => v !== vm);
+  fn.readyVms.delete(vm);
   vmCleanupTotal.inc();
   vmCount.dec({ function_id: fn.functionId, state: "ready" });
+  notifyVmDestroyed();
 
   cleanupLogger.info(
     { functionId: fn.functionId, vmId: vm.id, remainingVms: fn.vms.length },

src/runtime/deque.ts

@@ -0,0 +1,51 @@
+export class Deque<T> {
+  private buf: (T | undefined)[];
+  private head = 0;
+  private tail = 0;
+  private count = 0;
+
+  constructor(initialCapacity = 16) {
+    this.buf = new Array(initialCapacity);
+  }
+
+  get length(): number {
+    return this.count;
+  }
+
+  push(item: T): void {
+    if (this.count === this.buf.length) {
+      this.grow();
+    }
+    this.buf[this.tail] = item;
+    this.tail = (this.tail + 1) % this.buf.length;
+    this.count++;
+  }
+
+  shift(): T | undefined {
+    if (this.count === 0) return undefined;
+    const item = this.buf[this.head];
+    this.buf[this.head] = undefined;
+    this.head = (this.head + 1) % this.buf.length;
+    this.count--;
+    return item;
+  }
+
+  toArray(): T[] {
+    const result: T[] = new Array(this.count);
+    for (let i = 0; i < this.count; i++) {
+      result[i] = this.buf[(this.head + i) % this.buf.length] as T;
+    }
+    return result;
+  }
+
+  private grow(): void {
+    const newCap = this.buf.length * 2;
+    const newBuf: (T | undefined)[] = new Array(newCap);
+    for (let i = 0; i < this.count; i++) {
+      newBuf[i] = this.buf[(this.head + i) % this.buf.length];
+    }
+    this.buf = newBuf;
+    this.head = 0;
+    this.tail = this.count;
+  }
+}

src/runtime/scheduler.test.ts

@@ -4,8 +4,9 @@ import type { RequestTask } from "../types/types.js";
 
 vi.mock("./vm-manager.js", () => ({
   createVm: vi.fn(async (fid, fn) => {
-    const vm = { id: "mock", state: "ready", idleTime: Date.now() };
+    const vm = { id: "mock", state: "ready" as const, idleTime: Date.now() };
     fn.vms.push(vm);
+    fn.readyVms.add(vm);
     return vm;
   }),
 }));
@@ -14,12 +15,13 @@ vi.mock("./transport.js", () => ({
   sendRequest: vi.fn(async () => {}),
 }));
 
-import { enqueueRequest } from "./scheduler.js";
+import { enqueueRequest, resetSchedulerState } from "./scheduler.js";
 import { sendRequest } from "./transport.js";
 
 describe("scheduler", () => {
   beforeEach(() => {
     runtimeStore.functions.clear();
+    resetSchedulerState();
     vi.clearAllMocks();
   });
 
@@ -51,6 +53,6 @@ function makeTask(subPath = "/"): RequestTask & { promise: Promise<void> } {
   const promise = new Promise<void>((res, rej) => { resolve = res; reject = rej; });
   const req = { method: "GET", headers: {}, query: {}, body: {} };
   const res = { status: vi.fn().mockReturnThis(), json: vi.fn(), send: vi.fn() };
-  return { req, res, subPath, resolve, reject, promise } as any;
+  return { req, res, subPath, resolve, reject, enqueuedAt: performance.now(), promise } as any;
 }