Skip to content

Commit 87077d0

Browse files
saripovdenissaripovdenis
authored
Improve performance (#244)
1 parent 5384861 commit 87077d0

3 files changed

Lines changed: 262 additions & 45 deletions

File tree

‎bench.ts‎

Lines changed: 77 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -1,61 +1,99 @@
1+
import process from 'node:process';
12
import Benchmark, {type Deferred, type Event} from 'benchmark';
23
import PQueue from './source/index.js';
4+
import PriorityQueue from './source/priority-queue.js';
35

46
const suite = new Benchmark.Suite() as Benchmark.Suite;
7+
const smallTaskCount = 100;
8+
const largeBacklogTaskCount = 10_000;
9+
const selectedBenchmarks = process.argv.slice(2);
510

611
// Benchmark typings aren't up to date, let's help out manually
712
type Resolvable = Deferred & {resolve: () => void};
813

9-
(suite as any)
10-
.add('baseline', {
11-
defer: true,
14+
const noop = () => undefined;
1215

13-
async fn(deferred: Resolvable) {
14-
const queue = new PQueue();
16+
const shouldRunBenchmark = (name: string) => selectedBenchmarks.length === 0
17+
|| selectedBenchmarks.some(selectedBenchmark => name.includes(selectedBenchmark));
1518

16-
for (let i = 0; i < 100; i++) {
17-
// eslint-disable-next-line @typescript-eslint/no-empty-function
18-
queue.add(async () => {});
19-
}
19+
const addBenchmark = (name: string, fn: () => Promise<void>) => {
20+
if (!shouldRunBenchmark(name)) {
21+
return;
22+
}
2023

21-
await queue.onEmpty();
22-
deferred.resolve();
23-
},
24-
})
25-
.add('operation with random priority', {
24+
(suite as any).add(name, {
2625
defer: true,
2726

2827
async fn(deferred: Resolvable) {
29-
const queue = new PQueue();
30-
31-
for (let i = 0; i < 100; i++) {
32-
// eslint-disable-next-line @typescript-eslint/no-empty-function
33-
queue.add(async () => {}, {
34-
priority: Math.trunc(Math.random() * 100),
35-
});
36-
}
37-
38-
await queue.onEmpty();
28+
await fn();
3929
deferred.resolve();
4030
},
41-
})
42-
.add('operation with increasing priority', {
43-
defer: true,
31+
});
32+
};
4433

45-
async fn(deferred: Resolvable) {
46-
const queue = new PQueue();
34+
addBenchmark('baseline', async () => {
35+
const queue = new PQueue();
4736

48-
for (let i = 0; i < 100; i++) {
49-
// eslint-disable-next-line @typescript-eslint/no-empty-function
50-
queue.add(async () => {}, {
51-
priority: i,
52-
});
53-
}
37+
for (let i = 0; i < smallTaskCount; i++) {
38+
queue.add(noop);
39+
}
5440

55-
await queue.onEmpty();
56-
deferred.resolve();
57-
},
58-
})
41+
await queue.onEmpty();
42+
});
43+
44+
addBenchmark('operation with random priority', async () => {
45+
const queue = new PQueue();
46+
47+
for (let i = 0; i < smallTaskCount; i++) {
48+
queue.add(noop, {
49+
priority: Math.trunc(Math.random() * smallTaskCount),
50+
});
51+
}
52+
53+
await queue.onEmpty();
54+
});
55+
56+
addBenchmark('operation with increasing priority', async () => {
57+
const queue = new PQueue();
58+
59+
for (let i = 0; i < smallTaskCount; i++) {
60+
queue.add(noop, {
61+
priority: i,
62+
});
63+
}
64+
65+
await queue.onEmpty();
66+
});
67+
68+
addBenchmark('large-fifo-backlog', async () => {
69+
const queue = new PQueue({autoStart: false});
70+
71+
for (let i = 0; i < largeBacklogTaskCount; i++) {
72+
queue.add(noop);
73+
}
74+
75+
queue.start();
76+
await queue.onIdle();
77+
});
78+
79+
addBenchmark('priority-queue-dequeue', async () => {
80+
const queue = new PriorityQueue();
81+
82+
for (let i = 0; i < largeBacklogTaskCount; i++) {
83+
queue.enqueue(noop);
84+
}
85+
86+
let dequeued = 0;
87+
while (queue.dequeue() !== undefined) {
88+
dequeued++;
89+
}
90+
91+
if (dequeued !== largeBacklogTaskCount) {
92+
throw new Error('Dequeued task count mismatch');
93+
}
94+
});
95+
96+
suite
5997
.on('cycle', (event: Event) => {
6098
console.log(String(event.target as any));
6199
})

‎source/priority-queue.ts‎

Lines changed: 55 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,36 +2,50 @@ import {type Queue, type RunFunction} from './queue.js';
22
import lowerBound from './lower-bound.js';
33
import {type QueueAddOptions} from './options.js';
44

5+
const compactionThreshold = 100;
6+
57
export type PriorityQueueOptions = {
68
priority?: number;
79
} & QueueAddOptions;
810

911
export default class PriorityQueue implements Queue<RunFunction, PriorityQueueOptions> {
1012
readonly #queue: Array<PriorityQueueOptions & {run: RunFunction}> = [];
1113

14+
// Index of the next item to dequeue. Old items are compacted lazily so dequeue stays O(1).
15+
#head = 0;
16+
1217
enqueue(run: RunFunction, options?: Partial<PriorityQueueOptions>): void {
1318
const {
1419
priority = 0,
1520
id,
1621
} = options ?? {};
1722

23+
const {size} = this;
1824
const element = {
1925
priority,
2026
id,
2127
run,
2228
};
2329

24-
if (this.size === 0 || this.#queue[this.size - 1]!.priority! >= priority) {
30+
if (size === 0) {
31+
this.#queue.length = 0;
32+
this.#head = 0;
33+
this.#queue.push(element);
34+
return;
35+
}
36+
37+
if (this.#queue.at(-1)!.priority! >= priority) {
2538
this.#queue.push(element);
2639
return;
2740
}
2841

42+
this.#compact();
2943
const index = lowerBound(this.#queue, element, (a: Readonly<PriorityQueueOptions>, b: Readonly<PriorityQueueOptions>) => b.priority! - a.priority!);
3044
this.#queue.splice(index, 0, element);
3145
}
3246

3347
setPriority(id: string, priority: number) {
34-
const index: number = this.#queue.findIndex((element: Readonly<PriorityQueueOptions>) => element.id === id);
48+
const index = this.#queue.findIndex((element: Readonly<PriorityQueueOptions>, index) => index >= this.#head && element.id === id);
3549
if (index === -1) {
3650
throw new ReferenceError(`No promise function with the id "${id}" exists in the queue.`);
3751
}
@@ -43,7 +57,11 @@ export default class PriorityQueue implements Queue<RunFunction, PriorityQueueOp
4357
remove(id: string): void;
4458
remove(run: RunFunction): void;
4559
remove(idOrRun: string | RunFunction): void {
46-
const index = this.#queue.findIndex((element: Readonly<PriorityQueueOptions & {run: RunFunction}>) => {
60+
const index = this.#queue.findIndex((element: Readonly<PriorityQueueOptions & {run: RunFunction}>, index) => {
61+
if (index < this.#head) {
62+
return false;
63+
}
64+
4765
if (typeof idOrRun === 'string') {
4866
return element.id === idOrRun;
4967
}
@@ -57,15 +75,46 @@ export default class PriorityQueue implements Queue<RunFunction, PriorityQueueOp
5775
}
5876

5977
dequeue(): RunFunction | undefined {
60-
const item = this.#queue.shift();
78+
if (this.#head === this.#queue.length) {
79+
return undefined;
80+
}
81+
82+
const item = this.#queue[this.#head];
83+
this.#head++;
84+
85+
if (this.#head === this.#queue.length) {
86+
this.#queue.length = 0;
87+
this.#head = 0;
88+
} else if (this.#head > compactionThreshold && this.#head > this.#queue.length / 2) {
89+
this.#compact();
90+
}
91+
6192
return item?.run;
6293
}
6394

6495
filter(options: Readonly<Partial<PriorityQueueOptions>>): RunFunction[] {
65-
return this.#queue.filter((element: Readonly<PriorityQueueOptions>) => element.priority === options.priority).map((element: Readonly<{run: RunFunction}>) => element.run);
96+
const result: RunFunction[] = [];
97+
98+
for (let index = this.#head; index < this.#queue.length; index++) {
99+
const element = this.#queue[index]!;
100+
if (element.priority === options.priority) {
101+
result.push(element.run);
102+
}
103+
}
104+
105+
return result;
66106
}
67107

68108
get size(): number {
69-
return this.#queue.length;
109+
return this.#queue.length - this.#head;
110+
}
111+
112+
#compact(): void {
113+
if (this.#head === 0) {
114+
return;
115+
}
116+
117+
this.#queue.splice(0, this.#head);
118+
this.#head = 0;
70119
}
71120
}

‎test/priority-queue.ts‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
import {test} from 'node:test';
2+
import assert from 'node:assert/strict';
3+
import PriorityQueue from '../source/priority-queue.js';
4+
5+
const createRun = (value: string) => async () => value;
6+
7+
test('PriorityQueue ignores dequeued items in queue operations', () => {
8+
const queue = new PriorityQueue();
9+
const first = createRun('first');
10+
const second = createRun('second');
11+
const third = createRun('third');
12+
13+
queue.enqueue(first, {id: 'same', priority: 0});
14+
queue.enqueue(second, {id: 'same', priority: 0});
15+
queue.enqueue(third, {id: 'third', priority: 0});
16+
17+
assert.equal(queue.dequeue(), first);
18+
assert.deepEqual(queue.filter({priority: 0}), [second, third]);
19+
20+
queue.setPriority('same', 2);
21+
queue.remove(third);
22+
23+
assert.equal(queue.size, 1);
24+
assert.equal(queue.dequeue(), second);
25+
assert.equal(queue.dequeue(), undefined);
26+
});
27+
28+
test('PriorityQueue inserts high-priority items correctly after partial dequeue', () => {
29+
const queue = new PriorityQueue();
30+
const first = createRun('first');
31+
const second = createRun('second');
32+
const third = createRun('third');
33+
const urgent = createRun('urgent');
34+
35+
queue.enqueue(first, {priority: 0});
36+
queue.enqueue(second, {priority: 0});
37+
queue.enqueue(third, {priority: -1});
38+
39+
assert.equal(queue.dequeue(), first);
40+
41+
queue.enqueue(urgent, {priority: 1});
42+
43+
assert.equal(queue.size, 3);
44+
assert.equal(queue.dequeue(), urgent);
45+
assert.equal(queue.dequeue(), second);
46+
assert.equal(queue.dequeue(), third);
47+
assert.equal(queue.dequeue(), undefined);
48+
});
49+
50+
test('PriorityQueue removes the live duplicate id after partial dequeue', () => {
51+
const queue = new PriorityQueue();
52+
const first = createRun('first');
53+
const second = createRun('second');
54+
const third = createRun('third');
55+
56+
queue.enqueue(first, {id: 'same'});
57+
queue.enqueue(second, {id: 'same'});
58+
queue.enqueue(third, {id: 'third'});
59+
60+
assert.equal(queue.dequeue(), first);
61+
62+
queue.remove('same');
63+
64+
assert.equal(queue.size, 1);
65+
assert.equal(queue.dequeue(), third);
66+
assert.equal(queue.dequeue(), undefined);
67+
});
68+
69+
test('PriorityQueue removes by run after partial dequeue', () => {
70+
const queue = new PriorityQueue();
71+
const first = createRun('first');
72+
const second = createRun('second');
73+
const third = createRun('third');
74+
75+
queue.enqueue(first);
76+
queue.enqueue(second);
77+
queue.enqueue(third);
78+
79+
assert.equal(queue.dequeue(), first);
80+
81+
queue.remove(second);
82+
83+
assert.equal(queue.size, 1);
84+
assert.equal(queue.dequeue(), third);
85+
assert.equal(queue.dequeue(), undefined);
86+
});
87+
88+
test('PriorityQueue stays ordered after cursor compaction', () => {
89+
const queue = new PriorityQueue();
90+
const urgent = createRun('urgent');
91+
92+
for (let index = 0; index < 150; index++) {
93+
queue.enqueue(createRun(`task-${index}`));
94+
}
95+
96+
// Cross the compaction threshold while leaving queued items behind.
97+
for (let index = 0; index < 120; index++) {
98+
assert.notEqual(queue.dequeue(), undefined);
99+
}
100+
101+
queue.enqueue(urgent, {priority: 1});
102+
103+
assert.equal(queue.size, 31);
104+
assert.equal(queue.dequeue(), urgent);
105+
106+
let remaining = 0;
107+
while (queue.dequeue() !== undefined) {
108+
remaining++;
109+
}
110+
111+
assert.equal(remaining, 30);
112+
assert.equal(queue.size, 0);
113+
});
114+
115+
test('PriorityQueue setPriority works when only one live item remains', () => {
116+
const queue = new PriorityQueue();
117+
const first = createRun('first');
118+
const second = createRun('second');
119+
120+
queue.enqueue(first, {id: 'first'});
121+
queue.enqueue(second, {id: 'second'});
122+
123+
assert.equal(queue.dequeue(), first);
124+
125+
queue.setPriority('second', 1);
126+
127+
assert.equal(queue.size, 1);
128+
assert.equal(queue.dequeue(), second);
129+
assert.equal(queue.dequeue(), undefined);
130+
});

0 commit comments

Comments
 (0)