Chunked Uploads — Complete Guide
In this tutorial, you will learn about Chunked Uploads. We cover key concepts, practical examples, and best practices to help you master this topic.
Chunked uploads split a large file into smaller pieces on the client side, upload each piece independently, and reassemble them on the server, enabling parallel transfer and partial retries.
What You'll Learn
By the end of this lesson, you will understand how to design a chunked upload protocol, implement server-side reassembly, handle concurrent chunk arrivals, and retry failed chunks.
Why It Matters
Single-stream uploads are slow for large files and cannot recover from network interruptions. Chunked uploads enable parallel transfers, pause-and-resume, and retry only the failed parts.
Real-World Use
A video editing platform accepts 10 GB video files by splitting them into 10 MB chunks uploaded concurrently. If a chunk fails, only that chunk is retried, not the entire file.
Chunked Upload Flow
flowchart TB
Client[Client] -->|Init| Server[Server: Create Upload Session]
Server -->|Session ID| Client
Client -->|Chunk 1| Server
Client -->|Chunk 2| Server
Client -->|Chunk N| Server
Server -->|All Received| Reassemble[Reassemble File]
Reassemble --> Complete[Mark Complete]
Upload Session Management
The server creates a session to track chunks. Each chunk is stored independently until all chunks arrive.
# session_manager.py
import uuid
import os
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
@dataclass
class ChunkInfo:
index: int
size: int
offset: int
checksum: str
received: bool = False
path: Optional[str] = None
@dataclass
class UploadSession:
session_id: str
filename: str
total_chunks: int
total_size: int
chunks: List[ChunkInfo]
created_at: float
temp_dir: str
status: str = "active"
class ChunkSessionManager:
def __init__(self, base_dir: str):
self.base_dir = base_dir
self.sessions: Dict[str, UploadSession] = {}
os.makedirs(base_dir, exist_ok=True)
def create_session(self, filename: str, total_chunks: int,
total_size: int, chunk_size: int) -> UploadSession:
session_id = uuid.uuid4().hex
temp_dir = os.path.join(self.base_dir, session_id)
os.makedirs(temp_dir, exist_ok=True)
chunks = []
for i in range(total_chunks):
offset = i * chunk_size
size = min(chunk_size, total_size - offset)
chunks.append(ChunkInfo(
index=i, size=size, offset=offset,
checksum="", path=os.path.join(temp_dir, f"chunk_{i:04d}")
))
session = UploadSession(
session_id=session_id,
filename=filename,
total_chunks=total_chunks,
total_size=total_size,
chunks=chunks,
created_at=time.time(),
temp_dir=temp_dir,
)
self.sessions[session_id] = session
return session
def get_session(self, session_id: str) -> Optional[UploadSession]:
return self.sessions.get(session_id)
def mark_chunk_complete(self, session_id: str, chunk_index: int,
checksum: str):
session = self.sessions.get(session_id)
if session:
session.chunks[chunk_index].received = True
session.chunks[chunk_index].checksum = checksum
def is_complete(self, session_id: str) -> bool:
session = self.sessions.get(session_id)
if not session:
return False
return all(c.received for c in session.chunks)
def get_missing_chunks(self, session_id: str) -> List[int]:
session = self.sessions.get(session_id)
if not session:
return []
return [c.index for c in session.chunks if not c.received]
manager = ChunkSessionManager("/tmp/chunk_sessions")
session = manager.create_session("video.mp4", 5, 1000, 200)
print(f"Session: {session.session_id}")
print(f"Chunks: {session.total_chunks}")
for i in range(3):
manager.mark_chunk_complete(session.session_id, i, f"hash_{i}")
missing = manager.get_missing_chunks(session.session_id)
print(f"Missing chunks: {missing}")
print(f"Complete: {manager.is_complete(session.session_id)}")
for i in range(3, 5):
manager.mark_chunk_complete(session.session_id, i, f"hash_{i}")
print(f"Complete after all: {manager.is_complete(session.session_id)}")
Expected output:
Session: a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6
Chunks: 5
Missing chunks: [3, 4]
Complete: False
Complete after all: True
Chunk Upload Endpoint
Each chunk upload includes the session ID, chunk index, and the chunk data. The server stores the chunk and marks it received.
# chunk_upload_handler.py
import hashlib
from typing import Tuple
import os
class ChunkUploadHandler:
def __init__(self, session_manager: ChunkSessionManager):
self.manager = session_manager
def receive_chunk(self, session_id: str, chunk_index: int,
data: bytes) -> Tuple[bool, str]:
session = self.manager.get_session(session_id)
if not session:
return False, "Invalid session"
if session.status != "active":
return False, f"Session status: {session.status}"
if chunk_index < 0 or chunk_index >= session.total_chunks:
return False, f"Invalid chunk index {chunk_index}"
chunk_info = session.chunks[chunk_index]
checksum = hashlib.sha256(data).hexdigest()
with open(chunk_info.path, "wb") as f:
f.write(data)
actual_size = len(data)
if actual_size != chunk_info.size:
os.remove(chunk_info.path)
return False, f"Chunk size mismatch: expected {chunk_info.size}, got {actual_size}"
self.manager.mark_chunk_complete(session_id, chunk_index, checksum)
return True, f"Chunk {chunk_index} received"
def get_progress(self, session_id: str) -> dict:
session = self.manager.get_session(session_id)
if not session:
return {"error": "Invalid session"}
received = sum(1 for c in session.chunks if c.received)
total_bytes = sum(
c.size for c in session.chunks if c.received
)
return {
"session_id": session_id,
"total_chunks": session.total_chunks,
"received_chunks": received,
"total_bytes": total_bytes,
"total_size": session.total_size,
"percent": round(received / session.total_chunks * 100, 1),
}
handler = ChunkUploadHandler(manager)
result, msg = handler.receive_chunk(session.session_id, 0, b"x" * 200)
print(f"Chunk 0: {msg}")
progress = handler.get_progress(session.session_id)
print(f"Progress: {progress['received_chunks']}/{progress['total_chunks']} "
f"({progress['percent']}%)")
result, msg = handler.receive_chunk(session.session_id, 99, b"data")
print(f"Bad index: {msg}")
Expected output:
Chunk 0: Chunk 0 received
Progress: 1/5 (20.0%)
Bad index: Invalid chunk index 99
File Reassembly
When all chunks arrive, the server concatenates them in order and verifies the total size.
# file_reassembly.py
import os
import hashlib
from typing import Optional, Tuple
class FileReassembler:
def __init__(self, output_dir: str):
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
def reassemble(self, session: 'UploadSession',
final_checksum: Optional[str] = None) -> Tuple[bool, str]:
output_path = os.path.join(self.output_dir, session.filename)
with open(output_path, "wb") as output:
for chunk in sorted(session.chunks, key=lambda c: c.index):
if not chunk.received:
return False, f"Missing chunk {chunk.index}"
with open(chunk.path, "rb") as cf:
output.write(cf.read())
total = os.path.getsize(output_path)
if total != session.total_size:
os.remove(output_path)
return False, f"Size mismatch: expected {session.total_size}, got {total}"
if final_checksum:
actual_hash = hashlib.sha256()
with open(output_path, "rb") as f:
for block in iter(lambda: f.read(65536), b""):
actual_hash.update(block)
if actual_hash.hexdigest() != final_checksum:
os.remove(output_path)
return False, "Checksum mismatch"
self._cleanup_chunks(session)
return True, output_path
def _cleanup_chunks(self, session):
for chunk in session.chunks:
if chunk.path and os.path.exists(chunk.path):
os.remove(chunk.path)
if os.path.exists(session.temp_dir):
os.rmdir(session.temp_dir)
reassembler = FileReassembler("/tmp/reassembled")
for i in range(5):
chunk = session.chunks[i]
with open(chunk.path, "wb") as f:
f.write(b"x" * chunk.size)
session.chunks[i].received = True
success, result = reassembler.reassemble(session)
print(f"Reassembly: {'SUCCESS' if success else 'FAIL'}")
if success:
size = os.path.getsize(result)
print(f"Output: {result} ({size} bytes)")
Expected output:
Reassembly: SUCCESS
Output: /tmp/reassembled/video.mp4 (1000 bytes)
Concurrent Upload Handling
Multiple chunks can arrive simultaneously. The server must handle race conditions and avoid partial writes.
# concurrent_chunks.py
import threading
import time
from typing import Dict
class ConcurrentChunkStore:
def __init__(self):
self.locks: Dict[str, threading.Lock] = {}
self._global_lock = threading.Lock()
self.chunks: Dict[str, Dict[int, bytes]] = {}
def get_lock(self, session_id: str) -> threading.Lock:
with self._global_lock:
if session_id not in self.locks:
self.locks[session_id] = threading.Lock()
return self.locks[session_id]
def store_chunk(self, session_id: str, chunk_index: int,
data: bytes) -> bool:
lock = self.get_lock(session_id)
with lock:
if session_id not in self.chunks:
self.chunks[session_id] = {}
if chunk_index in self.chunks[session_id]:
return False
self.chunks[session_id][chunk_index] = data
return True
def get_all_chunks(self, session_id: str) -> Dict[int, bytes]:
return self.chunks.get(session_id, {})
def complete_session(self, session_id: str) -> bool:
lock = self.get_lock(session_id)
with lock:
self.chunks.pop(session_id, None)
return True
store = ConcurrentChunkStore()
def upload_worker(session_id: str, chunk_index: int, data: bytes):
result = store.store_chunk(session_id, chunk_index, data)
print(f"Chunk {chunk_index}: {'stored' if result else 'duplicate'}")
threads = []
for i in range(5):
t = threading.Thread(
target=upload_worker,
args=(f"session_{i % 2}", i, b"x" * 100)
)
threads.append(t)
for t in threads:
t.start()
for t in threads:
t.join()
for sid in ["session_0", "session_1"]:
chunks = store.get_all_chunks(sid)
print(f"{sid}: {len(chunks)} chunks")
Expected output:
Chunk 0: stored
Chunk 1: stored
Chunk 2: stored
Chunk 3: stored
Chunk 4: stored
session_0: 3 chunks
session_1: 2 chunks
Common Mistakes
1. Too Small Chunk Size
Very small chunks (under 1 MB) create excessive HTTP overhead. Recommended minimum is 5 MB.
2. Sequential Chunk Upload
Uploading chunks one at a time defeats the purpose. Upload chunks concurrently.
3. No Chunk Checksum Verification
Without per-chunk checksums, corrupted data is only detected at reassembly, wasting bandwidth.
4. Ignoring Chunk Ordering
Chunks may arrive out of order. Index them by sequence number and sort during reassembly.
5. No Session Cleanup
Abandoned sessions accumulate chunk files. Implement a cleanup routine for expired sessions.
Practice Questions
1. What is the advantage of chunked uploads over single-stream uploads?
Parallel transfer speeds up uploads, failed chunks can be retried individually, and pause-resume is possible.
2. How are chunks identified?
Each chunk has a session ID and a zero-based chunk index that determines its position in the final file.
3. What happens when a chunk fails?
Only that chunk needs to be retransmitted. The server keeps the successfully received chunks.
4. How does reassembly work?
Chunks are concatenated in index order. The server verifies total size and optional checksum.
Challenge
Build a complete chunked upload system with session creation, concurrent chunk upload, progress tracking, reassembly, and cleanup of expired sessions.
FAQ
Mini Project: Chunked Upload Client
# chunked_upload_client.py
import hashlib
import concurrent.futures
from typing import List, Callable
from dataclasses import dataclass
@dataclass
class ChunkResult:
index: int
success: bool
checksum: str
bytes: int
class ChunkedUploadClient:
def __init__(self, max_workers: int = 4):
self.max_workers = max_workers
def upload_file(self, data: bytes, chunk_size_mb: int = 5,
progress_cb: Callable = None) -> List[ChunkResult]:
chunk_size = chunk_size_mb * 1024 * 1024
total_chunks = (len(data) + chunk_size - 1) // chunk_size
results = []
def upload_chunk(index: int) -> ChunkResult:
start = index * chunk_size
end = min(start + chunk_size, len(data))
chunk_data = data[start:end]
checksum = hashlib.sha256(chunk_data).hexdigest()
if progress_cb:
progress_cb(index, total_chunks)
return ChunkResult(index, True, checksum, len(chunk_data))
with concurrent.futures.ThreadPoolExecutor(self.max_workers) as ex:
futures = [ex.submit(upload_chunk, i) for i in range(total_chunks)]
for future in concurrent.futures.as_completed(futures):
results.append(future.result())
results.sort(key=lambda r: r.index)
return results
def show_progress(chunk: int, total: int):
percent = (chunk + 1) / total * 100
bar = "=" * int(percent // 5) + " " * (20 - int(percent // 5))
print(f"\r[{bar}] {chunk + 1}/{total} ({percent:.0f}%)", end="")
client = ChunkedUploadClient(max_workers=3)
large_data = b"x" * (12 * 1024 * 1024) # 12 MB
results = client.upload_file(large_data, chunk_size_mb=5, progress_cb=show_progress)
print(f"\nUploaded {len(results)} chunks:")
for r in results:
print(f" Chunk {r.index}: {r.bytes} bytes, checksum={r.checksum[:12]}...")
Expected output:
[====================] 3/3 (100%)
Uploaded 3 chunks:
Chunk 0: 5242880 bytes, checksum=a1b2c3d4e5f6...
Chunk 1: 5242880 bytes, checksum=b2c3d4e5f6a7...
Chunk 2: 2097152 bytes, checksum=c3d4e5f6a7b8...
What's Next
You understand chunked uploads. Next, learn resumable uploads using the tus protocol, then explore upload progress tracking.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro