Async URL Downloader

A concurrent URL downloading system built in Python using asyncio and aiohttp. This is a rewrite of a threaded implementation that uses cooperative multitasking instead of OS threads to achieve concurrency.

Architecture

The system runs entirely on a single thread using an async event loop:

asyncio.gather()
    ├── process_job(job_1)  ──┐
    ├── process_job(job_2)    ├── asyncio.Semaphore (limits concurrent downloads)
    ├── process_job(job_3)    │
    └── process_job(job_N)  ──┘

aiohttp.ClientSession (shared across all coroutines)
JobRegistry (lifecycle tracking)
MetricsCollector (aggregate statistics)

Event Loop runs all job coroutines concurrently. Each coroutine suspends at await points — network calls, backoff sleeps — and yields control to the event loop, which runs other ready coroutines in the meantime.

Semaphore limits how many coroutines are actively downloading at once. Without this, 1000 URLs would spawn 1000 simultaneous connections, overwhelming the network and the target servers.

Shared Session is created once and reused across all coroutines. aiohttp.ClientSession manages connection pooling internally — creating one per request would defeat this and is explicitly discouraged by aiohttp.

Retry logic lives inside each coroutine as a loop. When a job fails and needs a retry, the coroutine calls await asyncio.sleep(backoff) and tries again. No separate scheduler thread is needed — the coroutine suspends itself and the event loop runs other work during the delay.

Key Differences from the Threaded Version

	Threaded	Async
Concurrency model	OS threads	Cooperative coroutines
Worker pool	4 threads pulling from a queue	N coroutines via asyncio.gather()
Task queue	queue.Queue (thread-safe handoff)	Not needed — coroutines called directly
Retry scheduling	Separate scheduler thread + priority queue	await asyncio.sleep() inside each coroutine
Completion tracking	threading.Semaphore	asyncio.gather() blocks until all coroutines return
Shared state safety	threading.Lock on all shared state	No locks needed — only one coroutine runs at a time
Shutdown sequence	6-phase shutdown with poison pills	gather() returns when all coroutines complete

Features

Networking

• Async streaming HTTP downloads via aiohttp
• Content-Length-aware download validation — handles both declared-size and chunked transfer encoding responses
• Configurable session-level timeout

Reliability

• Exponential backoff with full jitter to prevent retry storms
• Response classification: 2xx success, 4xx terminal failure (no retry), 5xx and network errors are retried
• Configurable max retries per job
• Unhandled exceptions in coroutines are caught via return_exceptions=True and logged without killing the entire run

Concurrency

• asyncio.Semaphore limits concurrent downloads to avoid overwhelming network or target servers
• Single shared aiohttp.ClientSession for connection pooling across all coroutines

Lifecycle Management

• JobStatus state machine: PENDING → RUNNING → SUCCESS | FAILED | RETRY_SCHEDULED → RUNNING
• JobRegistry tracks every job's state, attempt count, timestamps, bytes downloaded, and last error
• Transition validation — illegal state transitions raise immediately
• No locks required on registry or metrics — cooperative multitasking guarantees only one coroutine modifies shared state at a time

Observability

• Per-job registry queryable at any point during execution
• Final metrics and registry summary logged at shutdown

Project Structure

async_url_downloader/
├── main.py               # Entry point, argument parsing, asyncio.run()
├── process_jobs.py       # Coroutines, download pipeline, gather orchestration
├── input_parser.py       # URL file parsing and job creation
├── shared/
│   ├── models.py         # DownloadJob, JobResult, JobStatus, JobRecord
│   ├── metrics.py        # MetricsCollector (no locks)
│   └── job_registry.py   # JobRegistry with transition validation (no locks)
├── downloads/            # Output directory for downloaded files
└── test_urls.txt         # Sample URL list

Usage

python main.py test_urls.txt
python main.py test_urls.txt --max_workers 20
python main.py test_urls.txt --timeout_in_seconds 15
python main.py test_urls.txt --verbose

Arguments:

• file_path — path to a file containing one URL per line
• -w, --max_workers — max concurrent downloads via semaphore (default: 10)
• -t, --timeout_in_seconds — session-level timeout in seconds (default: 10)
• -v, --verbose — enable debug logging

Requirements

aiohttp
validators

Key Design Decisions

No task queue — in the threaded version, a bounded queue was necessary because the producer and workers ran on different threads and needed a thread-safe handoff point. In asyncio, everything runs on one thread. The producer simply creates coroutines and passes them to gather() directly.

No locks — a coroutine can only be interrupted at an await point. Between two await calls, a coroutine runs atomically — no other coroutine can interleave. This means incrementing a counter or updating a dict is always safe without a lock, as long as the operation contains no await.

Retry as a loop, not a separate thread — in the threaded version, a dedicated scheduler thread managed a priority queue of delayed retries. In asyncio, await asyncio.sleep(backoff) suspends the coroutine for the backoff duration while the event loop runs other work. The retry scheduler collapses into three lines inside the job coroutine.

return_exceptions=True in gather — by default, asyncio.gather() cancels all coroutines and raises the first exception it encounters. With return_exceptions=True, all coroutines run to completion and exceptions are returned as values in the results list. This prevents one bad job from killing all other in-flight downloads.