-
Notifications
You must be signed in to change notification settings - Fork 130
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Replace async-await-queue with mutex based queue (#694)
* selectively use supported ts libs for supported browser targets * reenable babel * update readme * fix grammar * add compat action to check es compatibility of build artificats * replace async-await-queue with mutex based queue * rename * Create rude-snails-yell.md * revert clearing queued requests * add snapshot * rename
- Loading branch information
Showing
6 changed files
with
164 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,5 @@ | ||
--- | ||
"livekit-client": patch | ||
--- | ||
|
||
Replace async-await-queue with mutex based queue |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,99 @@ | ||
import { AsyncQueue } from './AsyncQueue'; | ||
import { sleep } from './room/utils'; | ||
|
||
describe('asyncQueue', () => { | ||
it('runs multiple tasks in order', async () => { | ||
const queue = new AsyncQueue(); | ||
const tasksExecuted: number[] = []; | ||
|
||
for (let i = 0; i < 5; i++) { | ||
queue.run(async () => { | ||
await sleep(50); | ||
tasksExecuted.push(i); | ||
}); | ||
} | ||
await queue.flush(); | ||
expect(tasksExecuted).toMatchObject([0, 1, 2, 3, 4]); | ||
}); | ||
it('runs tasks sequentially and not in parallel', async () => { | ||
const queue = new AsyncQueue(); | ||
const results: number[] = []; | ||
for (let i = 0; i < 5; i++) { | ||
queue.run(async () => { | ||
results.push(i); | ||
await sleep(10); | ||
results.push(i); | ||
}); | ||
} | ||
await queue.flush(); | ||
expect(results).toMatchObject([0, 0, 1, 1, 2, 2, 3, 3, 4, 4]); | ||
}); | ||
it('continues executing tasks if one task throws an error', async () => { | ||
const queue = new AsyncQueue(); | ||
|
||
let task1threw = false; | ||
let task2Executed = false; | ||
|
||
queue | ||
.run(async () => { | ||
await sleep(100); | ||
throw Error('task 1 throws'); | ||
}) | ||
.catch(() => { | ||
task1threw = true; | ||
}); | ||
|
||
await queue | ||
.run(async () => { | ||
task2Executed = true; | ||
}) | ||
.catch(() => { | ||
fail('task 2 should not have thrown'); | ||
}); | ||
|
||
expect(task1threw).toBeTruthy(); | ||
expect(task2Executed).toBeTruthy(); | ||
}); | ||
it('returns the result of the task', async () => { | ||
const queue = new AsyncQueue(); | ||
|
||
const result = await queue.run(async () => { | ||
await sleep(10); | ||
return 'result'; | ||
}); | ||
|
||
expect(result).toBe('result'); | ||
}); | ||
it('returns only when the enqueued task and all previous tasks have completed', async () => { | ||
const queue = new AsyncQueue(); | ||
const tasksExecuted: number[] = []; | ||
for (let i = 0; i < 10; i += 1) { | ||
queue.run(async () => { | ||
await sleep(10); | ||
tasksExecuted.push(i); | ||
return i; | ||
}); | ||
} | ||
|
||
const result = await queue.run(async () => { | ||
await sleep(10); | ||
tasksExecuted.push(999); | ||
return 'result'; | ||
}); | ||
|
||
expect(result).toBe('result'); | ||
expect(tasksExecuted).toMatchObject([...new Array(10).fill(0).map((_, idx) => idx), 999]); | ||
}); | ||
it('can handle queue sizes of up to 10_000 tasks', async () => { | ||
const queue = new AsyncQueue(); | ||
const tasksExecuted: number[] = []; | ||
|
||
for (let i = 0; i < 10_000; i++) { | ||
queue.run(async () => { | ||
tasksExecuted.push(i); | ||
}); | ||
} | ||
await queue.flush(); | ||
expect(tasksExecuted).toMatchObject(new Array(10_000).fill(0).map((_, idx) => idx)); | ||
}); | ||
}); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,57 @@ | ||
import { Mutex } from './room/utils'; | ||
|
||
type QueueTask<T> = () => PromiseLike<T>; | ||
|
||
enum QueueTaskStatus { | ||
'WAITING', | ||
'RUNNING', | ||
'COMPLETED', | ||
} | ||
|
||
type QueueTaskInfo = { | ||
id: number; | ||
enqueuedAt: number; | ||
executedAt?: number; | ||
status: QueueTaskStatus; | ||
}; | ||
|
||
export class AsyncQueue { | ||
private pendingTasks: Map<number, QueueTaskInfo>; | ||
|
||
private taskMutex: Mutex; | ||
|
||
private nextTaskIndex: number; | ||
|
||
constructor() { | ||
this.pendingTasks = new Map(); | ||
this.taskMutex = new Mutex(); | ||
this.nextTaskIndex = 0; | ||
} | ||
|
||
async run<T>(task: QueueTask<T>) { | ||
const taskInfo: QueueTaskInfo = { | ||
id: this.nextTaskIndex++, | ||
enqueuedAt: Date.now(), | ||
status: QueueTaskStatus.WAITING, | ||
}; | ||
this.pendingTasks.set(taskInfo.id, taskInfo); | ||
const unlock = await this.taskMutex.lock(); | ||
try { | ||
taskInfo.executedAt = Date.now(); | ||
taskInfo.status = QueueTaskStatus.RUNNING; | ||
return await task(); | ||
} finally { | ||
taskInfo.status = QueueTaskStatus.COMPLETED; | ||
this.pendingTasks.delete(taskInfo.id); | ||
unlock(); | ||
} | ||
} | ||
|
||
async flush() { | ||
return this.run(async () => {}); | ||
} | ||
|
||
snapshot() { | ||
return Array.from(this.pendingTasks.values()); | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters