diff --git a/packages/shared/src/queue.test.ts b/packages/shared/src/queue.test.ts index 1a5f550..70177ce 100644 --- a/packages/shared/src/queue.test.ts +++ b/packages/shared/src/queue.test.ts @@ -22,6 +22,23 @@ describe('queue', () => { expect(await queue.pull()).toBe('a') }) + it('keeps order across internal compaction, including undefined items', async () => { + const queue = new Queue() + const value = (i: number) => i % 3 === 0 ? undefined : i + + // Splices the pulled half at 1024, then fully drains the remaining 1024. + for (let i = 0; i < 2048; i++) { + queue.push(value(i)) + } + + for (let i = 0; i < 2048; i++) { + expect(await queue.pull()).toBe(value(i)) + } + + queue.push(2048) + expect(await queue.pull()).toBe(2048) + }) + it('resolves a pending pull on push', async () => { const queue = new Queue() diff --git a/packages/shared/src/queue.ts b/packages/shared/src/queue.ts index b4db816..59a187e 100644 --- a/packages/shared/src/queue.ts +++ b/packages/shared/src/queue.ts @@ -1,7 +1,11 @@ import { AbortError } from './error' +const COMPACT_THRESHOLD = 1024 + export class Queue { - private readonly items: T[] = [] + /** Items before `head` have already been pulled. */ + private readonly items: (T | undefined)[] = [] + private head = 0 private readonly pendingPulls: (readonly [resolve: (item: T) => void, reject: (err: unknown) => void])[] = [] private closed: undefined | { reason: unknown } @@ -30,8 +34,23 @@ export class Queue { * @throws when the queue is closed or aborted. Note that buffered items can still be pulled after close until the buffer is drained. */ async pull(): Promise { - if (this.items.length > 0) { - return this.items.shift() as T + if (this.head < this.items.length) { + const item = this.items[this.head] as T + this.items[this.head++] = undefined // release the reference so pulled items can be GC'd + + // Compact once at least half the array is consumed, so the O(n) splice is amortized O(1) per pull. + if (this.head >= COMPACT_THRESHOLD && this.head * 2 >= this.items.length) { + if (this.head === this.items.length) { + this.items.length = 0 + } + else { + this.items.splice(0, this.head) + } + + this.head = 0 + } + + return item } if (this.closed) { @@ -66,6 +85,7 @@ export class Queue { abort(reason?: unknown): void { reason ??= new AbortError('Queue was aborted.') this.items.length = 0 + this.head = 0 this.close(reason) } }