chore: update

Author: CyanChanges

plugins/scheduler/state/mod.ts

@@ -35,7 +35,7 @@ export namespace $ScheduleState {
   }
 
   export const Config: z<Config> = z.object({
-    mode: z.union(['work-steal', 'semaphore']).default('semaphore'),
+    mode: z.union(['work-steal', 'semaphore']).default('work-steal'),
     cap: z.number().min(1).max(99999),
     period: z.number().min(10).default(1000),
   })

plugins/scheduler/state/semaphore/mod.ts

@@ -1,7 +1,7 @@
 import util from 'node:util'
 import type { Context } from '@cordisjs/core'
 import { symbols } from '@cordisjs/core'
-import type { Awaitable, Promisify } from 'cosmokit'
+import type { Promisify } from 'cosmokit'
 import z from 'schemastery'
 import { Tracker } from '../../tracker.ts'
 import { after } from '../../utils.ts'

plugins/scheduler/state/worksteal/fate.ts

@@ -8,6 +8,8 @@ export class WorkerFate {
   #decision: Promise<symbol>
   #resolve: (decision: symbol) => void
 
+  #reflection: WorkerFate | null = null
+
   static decided(fate: WorkerFate | symbol) {
     if (typeof fate === 'symbol') return new WorkerFate(fate)
     return fate
@@ -21,17 +23,22 @@ export class WorkerFate {
   }
 
   async found(): Promise<symbol> {
+    const self = this.#reflection || this
+    if (self) return await self.#decision
     return await this.#decision
   }
 
   decide(decision: symbol, next?: Promise<WorkerFate>) {
-    this.#resolve(decision)
-    this.#nexted = next ?? Promise.resolve(this)
+    const self = this.#reflection || this
+    self.#resolve(decision)
+    if (next) self.#nexted = next
   }
 
   async next(): Promise<WorkerFate> {
-    await this.#decision
-    return this.#nexted || Promise.resolve(this)
+    const self = this.#reflection || this
+    await self.#decision
+    if (self.#nexted) this.#reflection = await self.#nexted
+    return self.#nexted || Promise.resolve(self)
   }
 }
 

plugins/scheduler/state/worksteal/queue.ts

@@ -8,14 +8,14 @@ export class WorkQueue extends List<WorkTask> {
     tag: string | undefined,
     fn: (...args: A[]) => Awaitable<R>,
     ...args: A[]
-  )
+  ): WorkTask
 
   put<R, A extends any[]>(
     tag: string | undefined,
     promise: Promise<unknown>,
     fn: (...args: A[]) => Awaitable<R>,
     ...args: A[]
-  )
+  ): WorkTask
 
   put(
     tag: string | undefined,
@@ -57,6 +57,9 @@ export class WorkTask<R = any, A extends any[] = any[]> extends Item {
   args: A[]
   promise?: Promise<unknown>
 
+  hasError = false
+  error?: unknown
+
   constructor(
     public tag: string | undefined,
     fn: (...args: A[]) => Awaitable<R>,
@@ -72,6 +75,16 @@ export class WorkTask<R = any, A extends any[] = any[]> extends Item {
     return this
   }
 
+  markError(reason?: unknown) {
+    this.hasError = true
+    this.error = reason
+  }
+
+  throwIfError() {
+    if (!this.hasError) return
+    throw this.error
+  }
+
   call(): Promisify<R> {
     const resolved = Tracker.promise(
       Promise.try(this.fn, ...this.args),

plugins/scheduler/state/worksteal/worker.ts

@@ -12,20 +12,20 @@ export async function worker(
   let { value: newTask } = Atomics.waitAsync(shared, TASK_NOTIFY, shared[0])
 
   while (await newTask) {
-    const task = queue.get()
-    await Tracker.promise(
-      task.call(),
-      'worksteal::worker::[email protected]',
-      `worker-${id}`,
-    )
     switch (await decideMy(fate)) {
       case WorkerFate.KILL:
         return
       case WorkerFate.CONTINUE:
         break
     }
+    const task = queue.get()
+    await Tracker.promise(
+      task.call(),
+      'worksteal::worker::[email protected]',
+      `worker-${id}`,
+    ).catch(reason => task.markError(reason))
     if (queue.has()) continue
     ;({ value: newTask } = Atomics.waitAsync(shared, TASK_NOTIFY, shared[0]))
   }
-  await fate
+  await decideMy(fate)
 }