L2 · Working engineerAdvanced L2~12 min · 3 tests#82

Producer-Consumer With a Queue

Implement the producer-consumer pattern with asyncio.Queue and a sentinel, consuming items in order with backpressure. The threading version is explained.

The problem

Write async def run_pipeline(items, maxsize=2):

a producer coroutine puts every item onto an asyncio.Queue(maxsize=maxsize), then a sentinel (None) to signal the end
a consumer coroutine takes items until it sees the sentinel, recording each one
run both concurrently with asyncio.gather and return the consumed items, in order

Running in your browser: Python here runs on WebAssembly, which has no OS threads or processes. The standard APIs still work — threading, concurrent.futures, multiprocessing, asyncio — but they run on a deterministic simulator: threads run to completion when started, pools run tasks in order, and asyncio uses a virtual clock (await asyncio.sleep(0.2) advances time by 0.2 s instantly). Write exactly the code you would write in the interview.

Examples

Example 1

Input

asyncio.run(run_pipeline([1, 2, 3, 4, 5]))

Expected output

[1, 2, 3, 4, 5]

+ 2 hidden tests on Submit — tiny buffer still works.

Edge cases to ask about

Empty input
maxsize = 1
Multiple consumers need multiple sentinels

How the tests call your code

These helpers run before your code. The test inputs above call them.

import asyncio

Hints

0/3

How an interviewer scores this

0/9

1. Understand the problemRestate it, ask about edge cases and input size.
2. Write a brute-force solutionCorrect first, clever later.
3. Get the correct outputautoAll visible examples pass.
4. Analyse time complexityautoCommit to a Big-O in the Complexity Lab.
5. Analyse space complexityautoExtra memory, not the input.
6. Identify the bottleneckautoMeasure your code — where does it bend?
7. OptimiseUsually: a set or dict instead of a scan.
8. Explain why it worksSay it out loud to Solvi.
9. Test the edge casesautoSubmit passes every hidden test.

Python 3.13 · run_pipeline

import asyncio

async def run_pipeline(items, maxsize=2):
    queue = asyncio.Queue(maxsize=maxsize)
    consumed = []
    # write producer() and consumer(), then gather them
    return consumed

⌘/Ctrl + Enter runs the examples

Your code runs in real CPython inside your browser — nothing is sent anywhere. The first run downloads the interpreter (about 6 MB, once). Your code is saved on this device as you type.

Complexity Lab

What does this cost as n grows?

Interviewers score the analysis as much as the code. Commit to an answer first — then check it, and read why.

Time complexity of the optimal solution

Space complexity (extra memory)

Pick both to reveal the answer.

From brute force to optimal

The progression an interviewer wants to hear, one step at a time.

Approach	Time	Space	Idea
asyncio.Queue + sentinel	O(n)	O(maxsize)	put() waits when full: that is backpressure.
bestthreading + queue.Queue	O(n)	O(maxsize)	Same shape with threads; queue.Queue does the locking.

Walkthrough of the optimal approach (try it yourself first)

A bounded queue decouples a fast producer from a slow consumer: put waits when full (backpressure), get waits when empty. A sentinel (None) tells the consumer to stop — with several consumers, send one sentinel per consumer.

The threading version has the same shape: queue.Queue(maxsize), a producer thread, a consumer thread, None as the sentinel, and join() both.

Complexity: O(n) time, O(maxsize) space. Each item is put and taken once; the bounded queue never holds more than maxsize items.

Reveal the reference solution

import asyncio

async def run_pipeline(items, maxsize=2):
    queue = asyncio.Queue(maxsize=maxsize)
    consumed = []

    async def producer():
        for item in items:
            await queue.put(item)          # waits while the queue is full
        await queue.put(None)              # sentinel: no more work

    async def consumer():
        while True:
            item = await queue.get()
            if item is None:
                break
            consumed.append(item)

    await asyncio.gather(producer(), consumer())
    return consumed

Follow-ups interviewers ask

Two consumers sharing one queue.
Write it with threads and queue.Queue.

Frequently asked interview questions

Core interview concepts, complexities, and follow-ups scored by hiring teams.

What is the time complexity of Producer-Consumer With a Queue in Python?

The optimal solution runs in O(n) time and O(maxsize) auxiliary space. Each item is put and taken once; the bounded queue never holds more than maxsize items.

What is the brute-force approach, and how do you optimise it?

asyncio.Queue + sentinel: O(n) time, O(maxsize) space. put() waits when full: that is backpressure. threading + queue.Queue: O(n) time, O(maxsize) space. Same shape with threads; queue.Queue does the locking.

What follow-up questions do interviewers ask about Producer-Consumer With a Queue?

Two consumers sharing one queue. Write it with threads and queue.Queue.