From 05232d1742e5c52456dc44ded9d26a26457acbb1 Mon Sep 17 00:00:00 2001 From: vivek1504 Date: Wed, 17 Jun 2026 15:54:56 +0530 Subject: [PATCH] reimplemented sechduler --- src/routes/invoke.ts | 1 + src/runtime/cleanup.test.ts | 14 +- src/runtime/cleanup.ts | 8 +- src/runtime/deque.ts | 51 +++++ src/runtime/scheduler.test.ts | 8 +- src/runtime/scheduler.ts | 369 +++++++++++++++++++++++++++++----- src/runtime/vm-manager.ts | 1 + src/types/types.ts | 10 +- src/utils/metrics.ts | 50 +++++ 9 files changed, 456 insertions(+), 56 deletions(-) create mode 100644 src/runtime/deque.ts diff --git a/src/routes/invoke.ts b/src/routes/invoke.ts index 0b76b12..b8eddba 100644 --- a/src/routes/invoke.ts +++ b/src/routes/invoke.ts @@ -23,6 +23,7 @@ invokeRouter.use("/:functionId", async (req, res) => { subPath, resolve, reject, + enqueuedAt: performance.now(), }); }); diff --git a/src/runtime/cleanup.test.ts b/src/runtime/cleanup.test.ts index aa8ff25..b0686ff 100644 --- a/src/runtime/cleanup.test.ts +++ b/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(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(); }); }); - diff --git a/src/runtime/cleanup.ts b/src/runtime/cleanup.ts index ddcb58b..a2bcd1b 100644 --- a/src/runtime/cleanup.ts +++ b/src/runtime/cleanup.ts @@ -1,6 +1,7 @@ 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"; @@ -8,7 +9,10 @@ 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 }, diff --git a/src/runtime/deque.ts b/src/runtime/deque.ts new file mode 100644 index 0000000..1b12bae --- /dev/null +++ b/src/runtime/deque.ts @@ -0,0 +1,51 @@ +export class Deque { + 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; + } +} diff --git a/src/runtime/scheduler.test.ts b/src/runtime/scheduler.test.ts index 47df60b..4acf7a8 100644 --- a/src/runtime/scheduler.test.ts +++ b/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 } { const promise = new Promise((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; } diff --git a/src/runtime/scheduler.ts b/src/runtime/scheduler.ts index 15dba4b..28550f8 100644 --- a/src/runtime/scheduler.ts +++ b/src/runtime/scheduler.ts @@ -2,85 +2,358 @@ import { runtimeStore } from "./store.js"; import { createVm } from "./vm-manager.js"; import { sendRequest } from "./transport.js"; import { schedulerLogger } from "../utils/logger.js"; -import { invocationQueueDepth } from "../utils/metrics.js"; +import { + invocationQueueDepth, + schedulerEnqueueDuration, + schedulerDrainCycleDuration, + schedulerDispatchDuration, + schedulerVmLookupDuration, + schedulerVmProvisionDuration, + schedulerDrainIterations, + schedulerQueueWaitTime, +} from "../utils/metrics.js"; +import { Deque } from "./deque.js"; -import type { RequestTask, RuntimeFunction } from "../types/types.js"; +import type { RequestTask, RuntimeFunction, Vm } from "../types/types.js"; -const MAX_VMS = 100; +const MAX_GLOBAL_VMS = 200; +const MAX_PENDING_PER_FN = 4; +const QUANTUM = 1; + +const DISPATCH_BATCH_SIZE = 512; + +const perfLog = schedulerLogger; + +let globalProcessing = false; +let rescheduleNeeded = false; +let totalVmCount = 0; +let pendingVmCreations = 0; + +const activeFunctions = new Set(); + +export function enqueueRequest(functionId: string, task: RequestTask) { + const t0 = performance.now(); -export async function enqueueRequest(functionId: string, task: RequestTask) { let fn = runtimeStore.functions.get(functionId); if (!fn) { fn = { functionId, - queue: [], + queue: new Deque(), vms: [], - processing: false, + readyVms: new Set(), + weight: 1, + inflightCount: 0, + deficit: 0, + pendingCreations: 0, }; runtimeStore.functions.set(functionId, fn); schedulerLogger.info({ functionId }, "new runtime function registered"); } fn.queue.push(task); + activeFunctions.add(fn); invocationQueueDepth.inc({ function_id: functionId }); - schedulerLogger.debug( - { functionId, queueDepth: fn.queue.length }, - "request enqueued", + + const enqueueDurationMs = performance.now() - t0; + schedulerEnqueueDuration.observe(enqueueDurationMs / 1000); + perfLog.debug( + { + functionId, + queueDepth: fn.queue.length, + enqueueDurationMs: round(enqueueDurationMs), + }, + "perf:enqueue", ); - processQueue(fn); + scheduleGlobal(); } -async function processQueue(fn: RuntimeFunction) { - if (fn.processing) return; - fn.processing = true; +function scheduleGlobal() { + if (globalProcessing) { + rescheduleNeeded = true; + return; + } + globalProcessing = true; try { - while (fn.queue.length > 0) { - let vm = fn.vms.find((v) => v.state === "ready"); + drainBatch(); + } finally { + globalProcessing = false; + } - if (!vm && fn.vms.length < MAX_VMS) { - schedulerLogger.info( - { functionId: fn.functionId, currentVms: fn.vms.length }, - "creating new VM", - ); - vm = await createVm(fn.functionId, fn); - } + if (rescheduleNeeded) { + rescheduleNeeded = false; + scheduleGlobal(); + } +} - if (!vm) { - schedulerLogger.warn( - { functionId: fn.functionId, vmCount: fn.vms.length }, - "no VM available, max reached", - ); - return; +function drainBatch() { + const drainStart = performance.now(); + let dispatched: boolean; + let iterations = 0; + + do { + dispatched = false; + rescheduleNeeded = false; + + for (const fn of activeFunctions) { + if (fn.queue.length === 0) { + fn.deficit = 0; + activeFunctions.delete(fn); + continue; } - const task = fn.queue.shift(); - if (!task) return; + fn.deficit += fn.weight * QUANTUM; - invocationQueueDepth.dec({ function_id: fn.functionId }); - vm.state = "busy"; - schedulerLogger.debug( - { functionId: fn.functionId, vmId: vm.id, subPath: task.subPath }, - "dispatching request to VM", - ); + while (fn.deficit > 0 && fn.queue.length > 0) { + const lookupStart = performance.now(); + const vm = pickReadyVm(fn); + const lookupDurationMs = performance.now() - lookupStart; + schedulerVmLookupDuration.observe(lookupDurationMs / 1000); + + if (!vm) { + perfLog.debug( + { + functionId: fn.functionId, + readyVms: fn.readyVms.size, + totalVms: fn.vms.length, + lookupDurationMs: round(lookupDurationMs), + }, + "perf:vm_lookup_miss", + ); + maybeProvisionVm(fn); + break; + } + + const dispatchStart = performance.now(); + + const task = fn.queue.shift()!; + fn.deficit -= 1; + markVmBusy(fn, vm); + fn.inflightCount += 1; + invocationQueueDepth.dec({ function_id: fn.functionId }); + dispatched = true; + + const dispatchOverheadMs = performance.now() - dispatchStart; + schedulerDispatchDuration.observe(dispatchOverheadMs / 1000); - try { - await sendRequest(task.subPath, task.req, task.res, vm); - task.resolve(); - } catch (err) { - schedulerLogger.error( - { functionId: fn.functionId, vmId: vm.id, err }, - "request handling failed", + const queueWaitMs = performance.now() - task.enqueuedAt; + schedulerQueueWaitTime.observe( + { function_id: fn.functionId }, + queueWaitMs / 1000, ); - task.reject(err); - } finally { - vm.state = "ready"; - vm.idleTime = Date.now(); + + perfLog.debug( + { + functionId: fn.functionId, + vmId: vm.id, + subPath: task.subPath, + queueWaitMs: round(queueWaitMs), + lookupDurationMs: round(lookupDurationMs), + dispatchOverheadMs: round(dispatchOverheadMs), + queueRemaining: fn.queue.length, + readyVmsRemaining: fn.readyVms.size, + }, + "perf:dispatch", + ); + + dispatchTask(fn, vm, task); + + iterations++; + if (iterations >= DISPATCH_BATCH_SIZE) { + const drainDurationMs = performance.now() - drainStart; + schedulerDrainCycleDuration.observe(drainDurationMs / 1000); + schedulerDrainIterations.observe(iterations); + perfLog.debug( + { + iterations, + drainDurationMs: round(drainDurationMs), + yielding: true, + }, + "perf:drain_batch_yield", + ); + + if (hasMoreWork()) { + setImmediate(() => { + globalProcessing = true; + try { + drainBatch(); + } finally { + globalProcessing = false; + } + if (rescheduleNeeded) { + rescheduleNeeded = false; + scheduleGlobal(); + } + }); + } + return; + } + } + + if (fn.queue.length === 0) { + fn.deficit = 0; + activeFunctions.delete(fn); } } + } while (dispatched || rescheduleNeeded); + + const drainDurationMs = performance.now() - drainStart; + schedulerDrainCycleDuration.observe(drainDurationMs / 1000); + schedulerDrainIterations.observe(iterations); + + if (iterations > 0) { + perfLog.debug( + { + iterations, + drainDurationMs: round(drainDurationMs), + activeFunctions: activeFunctions.size, + yielding: false, + }, + "perf:drain_batch_complete", + ); + } +} + +function pickReadyVm(fn: RuntimeFunction): Vm | undefined { + for (const vm of fn.readyVms) { + return vm; + } + return undefined; +} + +function markVmBusy(fn: RuntimeFunction, vm: Vm) { + vm.state = "busy"; + fn.readyVms.delete(vm); +} + +function markVmReady(fn: RuntimeFunction, vm: Vm) { + vm.state = "ready"; + vm.idleTime = Date.now(); + fn.readyVms.add(vm); +} + +function hasMoreWork(): boolean { + for (const fn of activeFunctions) { + if (fn.queue.length > 0) return true; + } + return false; +} + +async function dispatchTask(fn: RuntimeFunction, vm: Vm, task: RequestTask) { + const taskStart = performance.now(); + + try { + const sendStart = performance.now(); + await sendRequest(task.subPath, task.req, task.res, vm); + const sendDurationMs = performance.now() - sendStart; + + task.resolve(); + + const totalMs = performance.now() - taskStart; + perfLog.debug( + { + functionId: fn.functionId, + vmId: vm.id, + subPath: task.subPath, + sendRequestMs: round(sendDurationMs), + totalTaskMs: round(totalMs), + }, + "perf:task_complete", + ); + } catch (err) { + const totalMs = performance.now() - taskStart; + schedulerLogger.error( + { + functionId: fn.functionId, + vmId: vm.id, + err, + totalTaskMs: round(totalMs), + }, + "request handling failed", + ); + task.reject(err); } finally { - fn.processing = false; + markVmReady(fn, vm); + fn.inflightCount -= 1; + scheduleGlobal(); + } +} + +function maybeProvisionVm(fn: RuntimeFunction) { + if (totalVmCount + pendingVmCreations >= MAX_GLOBAL_VMS) { + schedulerLogger.warn( + { functionId: fn.functionId, totalVmCount, pendingVmCreations }, + "global VM limit reached, cannot provision", + ); + return; } + + if (fn.pendingCreations >= MAX_PENDING_PER_FN) return; + + fn.pendingCreations += 1; + pendingVmCreations += 1; + + const provisionStart = performance.now(); + schedulerLogger.info( + { functionId: fn.functionId, totalVmCount, pendingVmCreations }, + "provisioning new VM asynchronously", + ); + + createVm(fn.functionId, fn) + .then(() => { + totalVmCount += 1; + const newVm = fn.vms[fn.vms.length - 1]; + if (newVm && newVm.state === "ready") { + fn.readyVms.add(newVm); + } + + const provisionDurationMs = performance.now() - provisionStart; + schedulerVmProvisionDuration.observe(provisionDurationMs / 1000); + + perfLog.info( + { + functionId: fn.functionId, + totalVmCount, + provisionDurationMs: round(provisionDurationMs), + }, + "perf:vm_provision_complete", + ); + }) + .catch((err) => { + const provisionDurationMs = performance.now() - provisionStart; + schedulerVmProvisionDuration.observe(provisionDurationMs / 1000); + + schedulerLogger.error( + { + functionId: fn.functionId, + err, + provisionDurationMs: round(provisionDurationMs), + }, + "async VM provisioning failed", + ); + }) + .finally(() => { + fn.pendingCreations -= 1; + pendingVmCreations -= 1; + scheduleGlobal(); + }); +} + +export function notifyVmDestroyed() { + totalVmCount = Math.max(0, totalVmCount - 1); + scheduleGlobal(); +} + +export function resetSchedulerState() { + globalProcessing = false; + rescheduleNeeded = false; + totalVmCount = 0; + pendingVmCreations = 0; + activeFunctions.clear(); +} + +function round(ms: number): number { + return Math.round(ms * 1000) / 1000; } diff --git a/src/runtime/vm-manager.ts b/src/runtime/vm-manager.ts index 09934c5..172b3e7 100644 --- a/src/runtime/vm-manager.ts +++ b/src/runtime/vm-manager.ts @@ -50,6 +50,7 @@ export async function createVm( }; fn.vms.push(vm); + fn.readyVms.add(vm); const durationSec = (performance.now() - start) / 1000; vmCreationTime.observe(durationSec); diff --git a/src/types/types.ts b/src/types/types.ts index f185ba2..a5ce129 100644 --- a/src/types/types.ts +++ b/src/types/types.ts @@ -1,5 +1,6 @@ import type { ChildProcessWithoutNullStreams } from "child_process"; import type { Socket } from "net"; +import type { Deque } from "../runtime/deque.js"; export type VmState = "creating" | "restoring" | "ready" | "busy" | "dead"; @@ -20,11 +21,16 @@ export interface RequestTask { subPath: string; resolve: () => void; reject: (err: any) => void; + enqueuedAt: number; } export interface RuntimeFunction { functionId: string; - queue: RequestTask[]; + weight: number; + deficit: number; + inflightCount: number; + pendingCreations: number; + queue: Deque; vms: Vm[]; - processing: boolean; + readyVms: Set; } diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts index 08bb719..05af03e 100644 --- a/src/utils/metrics.ts +++ b/src/utils/metrics.ts @@ -115,3 +115,53 @@ export const vsockErrors = new Counter({ labelNames: ["error_type"], registers: [register], }); + +export const schedulerEnqueueDuration = new Histogram({ + name: "scheduler_enqueue_duration_seconds", + help: "Time spent in enqueueRequest (registration + queue push)", + buckets: [0.00001, 0.00005, 0.0001, 0.0005, 0.001, 0.005, 0.01], + registers: [register], +}); + +export const schedulerDrainCycleDuration = new Histogram({ + name: "scheduler_drain_cycle_duration_seconds", + help: "Wall-clock time of a single drainBatch() call", + buckets: [0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5], + registers: [register], +}); + +export const schedulerDispatchDuration = new Histogram({ + name: "scheduler_dispatch_overhead_seconds", + help: "Overhead per dispatch (dequeue + state transition, excludes sendRequest)", + buckets: [0.000001, 0.000005, 0.00001, 0.00005, 0.0001, 0.0005], + registers: [register], +}); + +export const schedulerVmLookupDuration = new Histogram({ + name: "scheduler_vm_lookup_duration_seconds", + help: "Time to find a ready VM from the readyVms index", + buckets: [0.000001, 0.000005, 0.00001, 0.00005, 0.0001], + registers: [register], +}); + +export const schedulerVmProvisionDuration = new Histogram({ + name: "scheduler_vm_provision_duration_seconds", + help: "End-to-end time for async VM provisioning (createVm call)", + buckets: [0.1, 0.25, 0.5, 1, 2, 5, 10], + registers: [register], +}); + +export const schedulerDrainIterations = new Histogram({ + name: "scheduler_drain_iterations", + help: "Number of dispatch iterations per drainBatch() call", + buckets: [1, 5, 10, 25, 50, 100, 250, 512], + registers: [register], +}); + +export const schedulerQueueWaitTime = new Histogram({ + name: "scheduler_queue_wait_time_seconds", + help: "Time a request spends waiting in the scheduler queue before dispatch", + labelNames: ["function_id"], + buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10], + registers: [register], +});