javascript - JS/TS 中使用 async/await 的异步有界队列

Question

我正在努力思考 async/await，我有以下代码:

class AsyncQueue<T> {
    queue = Array<T>()
    maxSize = 1

    async enqueue(x: T) {
        if (this.queue.length > this.maxSize) {
            // Block until available
        }

        this.queue.unshift(x)
    }

    async dequeue() {
        if (this.queue.length == 0) {
            // Block until available
        }

        return this.queue.pop()!
    }
}

async function produce<T>(q: AsyncQueue, x: T) {
    await q.enqueue(x)
}

async function consume<T>(q: AsyncQueue): T {
    return await q.dequeue()
}

// Expecting 3 4 in the console
(async () => {
    const q = new AsyncQueue<number>()
    consume(q).then(console.log)
    consume(q).then(console.log)
    produce(q, 3)
    produce(q, 4)
    consume(q).then(console.log)
    consume(q).then(console.log)
})()

当然，我的问题出在代码的“阻止直到可用”部分。我期望能够“停止”执行直到发生某些事情(例如，出队停止直到存在入队，反之亦然，给定可用空间)。我觉得我可能需要为此使用协同程序，但我真的想确保我不会在这里遗漏任何 async/await 魔法。

Answer 1

17/04/2019 更新:长话短说，下面的 AsyncSemaphore 实现中存在一个错误，该错误是使用 property-based 捕获的测试。 You can read all about this "tale" here .这是固定版本:

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))
    }
}

---

最后，经过相当大的努力，并受到@Titian 答案的启发，我想我解决了这个问题。代码中充满了调试消息，但它可能有助于教学有关控制流的目的:

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
    }
}

---

更新:这是一个使用 AsyncSemaphore 的干净版本，它真正封装了通常使用并发原语完成的事情的方式，但适应了异步 CPS-single-带有 async/await 的 JavaScript 的 threaded-event-loop™ 风格。可以看到AsyncQueue的逻辑变得直观多了，通过Promises的双同步委托(delegate)给了两个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()!
    }
}

---

更新 2: 上面的代码中似乎隐藏了一个微妙的错误，在尝试使用大小为 0 的 AsyncQueue 时变得很明显。语义确实使意义:它是一个没有任何缓冲区的队列，发布者总是在其中等待消费者的存在。阻止它工作的线路是:

await this.waitingEnqueue.wait()
this.waitingDequeue.signal()

如果仔细观察，您会发现 dequeue() 与 enqueue() 并不完全对称。事实上，如果交换这两条指令的顺序:

this.waitingDequeue.signal()
await this.waitingEnqueue.wait()

然后一切又开始了；对我来说，在实际等待 enqueuing 发生之前，我们发出对 dequeuing() 感兴趣的信号似乎很直观。

如果没有进行广泛的测试，我仍然不确定这是否会重新引入细微的错误。我会把它作为一个挑战；)