Scheduler optimisation

.gitignore

@@ -0,0 +1 @@
+.idea

dxlink-javascript/dxlink-core/package.json

@@ -23,8 +23,8 @@
     "prepublishOnly": "npm run build && npm run test",
     "build": "microbundle -f esm,cjs",
     "lint": "eslint src",
-    "test": "tsx src/index.test.ts",
-    "test:watch": "tsx watch src/index.test.ts"
+    "test": "tsx src/index.test.ts && tsx src/scheduler.test.ts",
+    "test:watch": "tsx watch src/index.test.ts src/scheduler.test.ts"
   },
   "dependencies": {},
   "author": "Dmitry Petrov <dmitry.petrov@devexperts.com>",

dxlink-javascript/dxlink-core/src/scheduler.test.ts

@@ -0,0 +1,163 @@
+import { test } from 'uvu'
+import * as assert from 'uvu/assert'
+
+import { Scheduler } from './scheduler'
+
+test('schedule invokes callback after timeout', async () => {
+  const scheduler = new Scheduler()
+  let called = false
+  scheduler.schedule(
+    () => {
+      called = true
+    },
+    10,
+    'key'
+  )
+  await new Promise((r) => setTimeout(r, 50))
+  assert.is(called, true)
+  scheduler.clear()
+})
+
+test('cancel prevents callback from running', async () => {
+  const scheduler = new Scheduler()
+  let called = false
+  scheduler.schedule(
+    () => {
+      called = true
+    },
+    50,
+    'key'
+  )
+  scheduler.cancel('key')
+  await new Promise((r) => setTimeout(r, 100))
+  assert.is(called, false)
+})
+
+test('schedule with same key replaces previous', async () => {
+  const scheduler = new Scheduler()
+  let lastCall = 0
+  scheduler.schedule(
+    () => {
+      lastCall = 1
+    },
+    30,
+    'key'
+  )
+  scheduler.schedule(
+    () => {
+      lastCall = 2
+    },
+    30,
+    'key'
+  )
+  await new Promise((r) => setTimeout(r, 80))
+  assert.is(lastCall, 2)
+  scheduler.clear()
+})
+
+test('has returns true when scheduled, false otherwise', () => {
+  const scheduler = new Scheduler()
+  assert.is(scheduler.has('key'), false)
+  scheduler.schedule(() => {}, 1000, 'key')
+  assert.is(scheduler.has('key'), true)
+  scheduler.cancel('key')
+  assert.is(scheduler.has('key'), false)
+})
+
+test('clear cancels all scheduled tasks', async () => {
+  const scheduler = new Scheduler()
+  let a = false
+  let b = false
+  scheduler.schedule(
+    () => {
+      a = true
+    },
+    20,
+    'a'
+  )
+  scheduler.schedule(
+    () => {
+      b = true
+    },
+    20,
+    'b'
+  )
+  scheduler.clear()
+  await new Promise((r) => setTimeout(r, 50))
+  assert.is(a, false)
+  assert.is(b, false)
+})
+
+test('cancel called from within a batch prevents the cancelled callback in same batch from running', async () => {
+  const scheduler = new Scheduler()
+  let bCalled = false
+  scheduler.schedule(
+    () => {
+      scheduler.cancel('B')
+    },
+    50,
+    'A'
+  )
+  scheduler.schedule(
+    () => {
+      bCalled = true
+    },
+    50,
+    'B'
+  )
+  await new Promise((r) => setTimeout(r, 100))
+  assert.is(bCalled, false)
+  scheduler.clear()
+})
+
+test('when a callback in a batch throws, keys of not-run callbacks are cleared from scheduler state', async () => {
+  const scheduler = new Scheduler()
+  const expectedMessage = 'callback error'
+  let caught: Error | undefined
+  const onUncaught = (err: Error) => {
+    caught = err
+    process.off('uncaughtException', onUncaught)
+  }
+  process.on('uncaughtException', onUncaught)
+  try {
+    scheduler.schedule(
+      () => {
+        throw new Error(expectedMessage)
+      },
+      50,
+      'A'
+    )
+    scheduler.schedule(() => {}, 50, 'B')
+    await new Promise((r) => setTimeout(r, 100))
+    assert.is(scheduler.has('B'), false)
+    assert.is(caught?.message, expectedMessage)
+  } finally {
+    process.off('uncaughtException', onUncaught)
+  }
+  scheduler.clear()
+})
+
+test('many schedule calls with same key create only one timer', () => {
+  let setTimeoutCalls = 0
+  const originalSetTimeout = globalThis.setTimeout
+  const wrapper = (...args: Parameters<typeof setTimeout>): ReturnType<typeof setTimeout> => {
+    setTimeoutCalls++
+    return originalSetTimeout.apply(globalThis, args)
+  }
+  globalThis.setTimeout = wrapper as typeof setTimeout
+  try {
+    const scheduler = new Scheduler()
+    for (let i = 0; i < 100; i++) {
+      scheduler.schedule(() => {}, 100, 'sameKey')
+    }
+    assert.is(
+      setTimeoutCalls,
+      1,
+      'expected exactly one setTimeout for many schedule() calls with same key'
+    )
+  } finally {
+    globalThis.setTimeout = originalSetTimeout
+  }
+})
+
+test.run()

dxlink-javascript/dxlink-core/src/scheduler.ts

@@ -1,34 +1,95 @@
+const BATCHING_SIZE = 0.01
+
+const getBatchTime = (timeoutMs: number): number => {
+  const intervalMs = Math.max(1, timeoutMs * BATCHING_SIZE)
+  return Math.floor((Date.now() + timeoutMs) / intervalMs) * intervalMs
+}
+
+type Batch = {
+  timeoutId: ReturnType<typeof setTimeout>
+  callbacks: Map<string, () => void>
+}
+
 /**
  * Scheduler for scheduling callbacks.
+ * Batches timers by quantized end time so repeated schedules in the same time window share one timer.
  */
 export class Scheduler {
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  private timeoutIds: Record<string, any> = {}
+  private batches = new Map<number, Batch>()
+  private keyToBatch = new Map<string, number>()
+
+  private removeKeyFromBatch(key: string, batchKey: number): void {
+    const batch = this.batches.get(batchKey)
+    if (batch === undefined) return
+    batch.callbacks.delete(key)
+    if (batch.callbacks.size === 0) {
+      clearTimeout(batch.timeoutId)
+      this.batches.delete(batchKey)
+    }
+  }
+
+  private runBatch(batchKey: number): void {
+    const batch = this.batches.get(batchKey)
+    if (batch === undefined) return
+    try {
+      for (const [k, cb] of batch.callbacks) {
+        this.keyToBatch.delete(k)
+        cb()
+      }
+    } finally {
+      for (const k of batch.callbacks.keys()) {
+        this.keyToBatch.delete(k)
+      }
+      this.batches.delete(batchKey)
+    }
+  }
 
   schedule = (callback: () => void, timeout: number, key: string) => {
-    this.cancel(key)
-    this.timeoutIds[key] = setTimeout(() => {
-      delete this.timeoutIds[key]
-      callback()
-    }, timeout)
+    const batchKey = getBatchTime(timeout)
+    const existingBatchKey = this.keyToBatch.get(key)
+
+    if (existingBatchKey !== undefined) {
+      if (existingBatchKey === batchKey) {
+        const batch = this.batches.get(batchKey)
+        if (batch !== undefined) {
+          batch.callbacks.set(key, callback)
+          return key
+        }
+      }
+      this.removeKeyFromBatch(key, existingBatchKey)
+    }
+
+    let batch = this.batches.get(batchKey)
+    if (batch === undefined) {
+      const delay = Math.max(0, batchKey - Date.now())
+      batch = {
+        timeoutId: setTimeout(() => this.runBatch(batchKey), delay),
+        callbacks: new Map(),
+      }
+      this.batches.set(batchKey, batch)
+    }
 
+    batch.callbacks.set(key, callback)
+    this.keyToBatch.set(key, batchKey)
     return key
   }
 
   cancel = (key: string) => {
-    if (this.timeoutIds[key] !== undefined) {
-      clearTimeout(this.timeoutIds[key])
-      delete this.timeoutIds[key]
-    }
+    const batchKey = this.keyToBatch.get(key)
+    if (batchKey === undefined) return
+    this.keyToBatch.delete(key)
+    this.removeKeyFromBatch(key, batchKey)
   }
 
   clear = () => {
-    for (const key of Object.keys(this.timeoutIds)) {
-      this.cancel(key)
+    for (const batch of this.batches.values()) {
+      clearTimeout(batch.timeoutId)
     }
+    this.batches.clear()
+    this.keyToBatch.clear()
   }
 
   has = (key: string) => {
-    return this.timeoutIds[key] !== undefined
+    return this.keyToBatch.has(key)
   }
 }