javascript - คิวจำกัดแบบอะซิงโครนัสใน JS/TS โดยใช้ async/await

ฉันกำลังพยายามทำความเข้าใจ 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)
})()

ปัญหาของฉัน แน่นอน อยู่ที่ส่วนของโค้ดที่ "Block until available". ฉันคาดหวังที่จะสามารถ "หยุด" การทำงานจนกว่าจะเกิดบางสิ่งบางอย่าง (เช่น dequeue จะหยุดจนกว่า enqueue จะมีอยู่ และในทางกลับกันถ้ามีพื้นที่ว่าง) ฉันมีความรู้สึกว่าฉันอาจต้องใช้ coroutines สำหรับเรื่องนี้ แต่ฉันอยากมั่นใจว่าฉันไม่ได้พลาดเวทมนตร์ async/await อะไรอยู่

คำตอบ #1

17/04/2019 อัปเดต: พูดให้ง่าย ๆ ก็คือ มีข้อบกพร่องในโค้ด AsyncSemaphore ที่ใช้ด้านล่าง ซึ่งถูกจับได้โดยการทดสอบโดยอิงคุณสมบัติ คุณสามารถอ่านเรื่องราวทั้งหมดเกี่ยวกับ "นิทาน" นี้ได้ที่นี่ นี่คือเวอร์ชันที่แก้ไขแล้ว:

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 ซึ่งแสดงถึงวิธีที่สิ่งต่าง ๆ มักทำโดยใช้ primitives ของการแข่งขัน แต่ปรับให้เข้ากับสไตล์ asynchronous-CPS-single-threaded-event-loop™ ของ JavaScript ด้วย async/await คุณจะเห็นว่า logic ของ AsyncQueue กลายเป็นเรื่องที่เข้าใจได้มากขึ้น และการซิงโครไนซ์ซ้ำสองผ่าน Promises จะถูกมอบหมายให้กับสอง semaphore:

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: ดูเหมือนว่าจะมีข้อบกพร่องเล็ก ๆ ซ่อนอยู่ในโค้ดข้างต้น ซึ่งปรากฏชัดเมื่อพยายามใช้ AsyncQueue ขนาด 0 ความหมายของมันก็สมเหตุสมผล: มันคือคิวที่ไม่มีบัฟเฟอร์ โดยที่ผู้เผยแพร่จะต้องรอให้ผู้บริโภคมีอยู่เสมอ บรรทัดที่ทำให้มันไม่ทำงานคือ:

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

ถ้าคุณสังเกตอย่างใกล้ชิด คุณจะเห็นว่า dequeue() ไม่สมมาตรอย่างสมบูรณ์กับ enqueue() แท้จริงแล้ว หากมีการสลับลำดับของคำสั่งทั้งสองนี้:

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

จากนั้นทุกอย่างก็ทำงานได้อีกครั้ง; มันดูเป็นเรื่องสัญชาตญาณสำหรับฉันที่เราส่งสัญญาณว่า มีบางสิ่งสนใจใน dequeuing() ก่อนที่จะรอให้เกิด enqueuing จริง ๆ

ฉันยังไม่แน่ใจว่านี่จะไม่กลับไปสร้างบั๊กเล็ก ๆ โดยไม่ได้ทดสอบอย่างละเอียด ฉันจะปล่อยให้เป็นความท้าทาย ;)