Update on 17th April 2019: To cut the long story short, there appears to be a bug in the AsyncSemaphore implementation provided below, which was identified through property-based testing. If you're interested, you can read all about this "tale" here. Here is the corrected version:
class AsyncSemaphore {
private promises = Array<() => void>()
constructor(private permits: number) {}
signal() {
this.permits += 1
if (this.promises.length > 0) this.promises.pop()!()
}
async wait() {
this.permits -= 1
if (this.permits < 0 || this.promises.length > 0)
await new Promise(r => this.promises.unshift(r))
}
}
After much effort and with inspiration from @Titian's answer, I believe I have resolved the issue. Although the code contains debug messages, it could be useful for educational purposes in understanding the flow of control:
class AsyncQueue<T> {
waitingEnqueue = new Array<() => void>()
waitingDequeue = new Array<() => void>()
enqueuePointer = 0
dequeuePointer = 0
queue = Array<T>()
maxSize = 1
trace = 0
async enqueue(x: T) {
this.trace += 1
const localTrace = this.trace
if ((this.queue.length + 1) > this.maxSize || this.waitingDequeue.length > 0) {
console.debug(`[${localTrace}] Producer Waiting`)
this.dequeuePointer += 1
await new Promise(r => this.waitingDequeue.unshift(r))
this.waitingDequeue.pop()
console.debug(`[${localTrace}] Producer Ready`)
}
this.queue.unshift(x)
console.debug(`[${localTrace}] Enqueueing ${x} Queue is now [${this.queue.join(', ')}]`)
if (this.enqueuePointer > 0) {
console.debug(`[${localTrace}] Notify Consumer`)
this.waitingEnqueue[this.enqueuePointer-1]()
this.enqueuePointer -= 1
}
}
async dequeue() {
this.trace += 1
const localTrace = this.trace
console.debug(`[${localTrace}] Queue length before pop: ${this.queue.length}`)
if (this.queue.length == 0 || this.waitingEnqueue.length > 0) {
console.debug(`[${localTrace}] Consumer Waiting`)
this.enqueuePointer += 1
await new Promise(r => this.waitingEnqueue.unshift(r))
this.waitingEnqueue.pop()
console.debug(`[${localTrace}] Consumer Ready`)
}
const x = this.queue.pop()!
console.debug(`[${localTrace}] Queue length after pop: ${this.queue.length} Popping ${x}`)
if (this.dequeuePointer > 0) {
console.debug(`[${localTrace}] Notify Producer`)
this.waitingDequeue[this.dequeuePointer - 1]()
this.dequeuePointer -= 1
}
return x
}
}
Update: Here is a cleaner version utilizing an AsyncSemaphore, which encapsulates the common practice when working with concurrency primitives, adapted for asynchronous-CPS-single-threaded-event-loop™ style of JavaScript using async/await. With this updated logic in place, the functionality of AsyncQueue becomes more intuitive, and the dual synchronization through Promises is delegated to two separate semaphores:
class AsyncSemaphore {
private promises = Array<() => void>()
constructor(private permits: number) {}
signal() {
this.permits += 1
if (this.promises.length > 0) this.promises.pop()()
}
async wait() {
if (this.permits == 0 || this.promises.length > 0)
await new Promise(r => this.promises.unshift(r))
this.permits -= 1
}
}
class AsyncQueue<T> {
private queue = Array<T>()
private waitingEnqueue: AsyncSemaphore
private waitingDequeue: AsyncSemaphore
constructor(readonly maxSize: number) {
this.waitingEnqueue = new AsyncSemaphore(0)
this.waitingDequeue = new AsyncSemaphore(maxSize)
}
async enqueue(x: T) {
await this.waitingDequeue.wait()
this.queue.unshift(x)
this.waitingEnqueue.signal()
}
async dequeue() {
await this.waitingEnqueue.wait()
this.waitingDequeue.signal()
return this.queue.pop()!
}
}
Update 2: A subtle bug appeared hidden in the above code when attempting to use an AsyncQueue of size 0. The semantics do make sense: it is a queue without any buffer, where the publisher always awaits for a consumer to exist. The issue preventing it from functioning properly was located in these lines:
await this.waitingEnqueue.wait()
this.waitingDequeue.signal()
Upon closer inspection, one can see that dequeue() is not perfectly symmetrical to enqueue(). In fact, swapping the order of these two instructions:
this.waitingDequeue.signal()
await this.waitingEnqueue.wait()
Resolves the problem; intuitively, we should indicate that there is interest in dequeuing() before actually waiting for an enqueuing to occur.
I cannot guarantee this change won't reintroduce unforeseen bugs without comprehensive testing. Consider it a challenge ;)