"""Count waiting work, not performance. Python 3.11+, no network or packages.

Both designs process the same 10,000 integer jobs. An Event deliberately
holds workers still so counts can be inspected without timing guesses.
This is an authored concurrency demonstration, not a model benchmark.
"""

import asyncio
import sys

JOBS = 10_000
WORKERS = 10
QUEUE_SIZE = 20


async def eager_tasks():
    semaphore = asyncio.Semaphore(WORKERS)
    release = asyncio.Event()
    all_started = asyncio.Event()
    started = active = completed = peak_active = 0

    async def job(number):
        nonlocal started, active, completed, peak_active
        started += 1
        if started == JOBS:
            all_started.set()
        async with semaphore:
            active += 1
            peak_active = max(peak_active, active)
            try:
                await release.wait()
                completed += 1
            finally:
                active -= 1

    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(job(i)) for i in range(JOBS)]
        await all_started.wait()
        unfinished = sum(not task.done() for task in tasks)
        assert (unfinished, active) == (JOBS, WORKERS)
        print(f"eager: active={active}, waiting={unfinished - active}, "
              f"unfinished_tasks={unfinished}")
        release.set()
    assert completed == JOBS and peak_active == WORKERS


async def pull_with_workers():
    queue = asyncio.Queue(maxsize=QUEUE_SIZE)
    release = asyncio.Event()
    saturated = asyncio.Event()
    stop = object()
    active = completed = pulled = admitted = peak_active = peak_queue = 0

    async def producer():
        nonlocal pulled, admitted, peak_queue
        for number in range(JOBS):  # lazy source; no list of job contexts
            pulled += 1
            if queue.full() and active == WORKERS:
                saturated.set()
            await queue.put(number)  # this producer waits here
            admitted += 1
            peak_queue = max(peak_queue, queue.qsize())
        for _ in range(WORKERS):
            await queue.put(stop)

    async def worker():
        nonlocal active, completed, peak_active
        while True:
            number = await queue.get()
            try:
                if number is stop:
                    return
                active += 1
                peak_active = max(peak_active, active)
                try:
                    await release.wait()
                    completed += 1
                finally:
                    active -= 1
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as group:
        workers = [group.create_task(worker()) for _ in range(WORKERS)]
        producer_task = group.create_task(producer())
        await saturated.wait()
        assert (active, queue.qsize(), admitted, pulled) == (10, 20, 30, 31)
        assert not producer_task.done() and all(not w.done() for w in workers)
        print(f"bounded: active={active}, queued={queue.qsize()}, "
              f"held_by_producer={pulled - admitted}, pulled={pulled}")
        release.set()
    assert completed == JOBS and peak_active <= WORKERS
    assert peak_queue <= QUEUE_SIZE and queue.empty()
    await queue.join()


async def main():
    await eager_tasks()
    await pull_with_workers()
    print(f"Both completed {JOBS} jobs. All count assertions passed.")


if __name__ == "__main__":
    print(f"Python {sys.version.split()[0]}")
    asyncio.run(asyncio.wait_for(main(), timeout=10))