Networks and I/O for Data Engineers: Throughput, Latency, Retries, Timeouts

In plain terms

Very little in a data platform sits still. Records are pulled from a database in another building, files are fetched from a vendor on another continent, messages are pushed to a queue in another data centre. A small local test can hide delays and failures that appear when services are farther apart or under load.

A working model of the network: a postal service that is usually fast, occasionally very slow, and that will sometimes deliver a letter while reporting that it failed. Retrying an uncertain write can create a duplicate charge; treating the error as proof that nothing happened can leave records unreconciled. The caller needs evidence of the outcome or a way to repeat the operation safely.

We will distinguish capacity from achieved throughput, follow a request through its possible failures, and decide when a retry is safe. The standard-library labs use a loopback HTTP server to delay responses, return an error after a write, and send an incomplete file. They illustrate mechanisms rather than reproduce a production outage.

Bandwidth, throughput, and latency

Three related quantities answer different questions:

• Bandwidth is the link’s nominal capacity, usually measured in bits per second (Mbps or Gbps).
• Throughput is the measured transfer or completion rate. State whether you count wire bytes, useful payload bytes, records, or successful requests; retries are not new useful work.
• Latency is a delay. Here round-trip time (RTT) means travel to the other end and back; application response latency also includes server work and queueing. Define whether measurement ends at the first byte or the complete response.

A 100 Mbps link has a nominal ceiling of 12.5 MB/s when using decimal units: eight bits make one byte. Useful throughput is lower when protocol overhead, loss, server CPU, disk I/O, or rate limits become bottlenecks. More bandwidth does not remove propagation delay, though it can reduce transmission or queueing delay. The RTT values below are illustrative assumptions, not current route measurements:

When fixed waiting dominates the transfer and processing of each small request, sequential calls are latency-bound. A slow server or storage device can still be the dominant bottleneck. The arithmetic is worth doing once:

round_trip = 0.180
one_at_a_time = 100_000 * round_trip
in_pages_of_1000 = 100 * round_trip
print(f"{one_at_a_time / 3600:.1f} hours, {in_pages_of_1000:.0f} seconds")
# 5.0 hours, 18 seconds

Under a fixed 180 ms wait per request, 100,000 sequential calls accumulate five hours of waiting. A hundred pages accumulate eighteen seconds. Neither figure includes connection setup, server processing, payload transfer, parsing, or retries. Batching reduces fixed request overhead; it does not remove the bytes that must move.

Batching reduces request count; bounded concurrency overlaps waiting. With one outstanding request, a 0.180-second response time limits completion to about 5.6 requests/s. Ten outstanding requests could approach 55.6 requests/s if response time stays fixed and the service, network, and client have spare capacity. Beyond saturation, more concurrency can increase queueing and reduce useful throughput. Measure payload bytes/s, successful records/s, latency percentiles, and retry counts together.

I/O connects the network to storage

I/O means input and output: reading bytes from a socket or disk and writing them elsewhere. A download may receive data into memory, parse or decompress it, then write it to storage. A faster network cannot make the whole transfer faster than a slower destination can sustain. Compare rates at the same boundary: compressed network MB/s and uncompressed disk MB/s count different byte volumes.

For an illustrative 1,000 MB file with unchanged bytes, suppose receiving sustains 100 MB/s and writing sustains 80 MB/s. Fully sequential receiving and writing take at least 10 + 12.5 = 22.5 seconds. An ideally overlapped stream has a lower bound of 12.5 seconds, set by the slower stage; setup, buffering, computation, and finalization can add time. Measure the actual overlap before adding stage durations. A successful buffered write also need not mean the bytes are durable on storage.

What one request costs

A fresh HTTPS request over TCP can involve several stages. Caching, protocol version, and connection reuse change which stages actually run:

1. DNS lookup: turn the hostname into an address. Usually cached; occasionally the thing that is broken.
2. TCP handshake: one round trip to open the connection.
3. TLS handshake: additional exchanges to authenticate the peer and establish encryption; the cost depends on the TLS version and resumption.
4. The request and response.

As an example, a TCP handshake plus a full one-round-trip TLS handshake adds roughly two RTTs before the application request, excluding DNS and computation. Reusing a live connection avoids repeating that setup. The TLS 1.3 overview shows the handshake exchanges. HTTP/3 uses QUIC rather than this TCP sequence, so do not apply this count to every protocol.

In Python, Requests sessions provide connection pooling. Consume or close responses so connections can be released appropriately, and close the session when finished. The standard-library lab compares fresh urlopen calls with a reused HTTPConnection. Reuse removes setup work, but its measured speedup depends on the workload and transport behavior.

The outcomes a network call can have

An exception tells the caller that it did not obtain a normal result; it does not always tell it whether a remote side effect occurred. Separate evidence about delivery from evidence about application state:

A response timeout after sending a write leaves an uncertain outcome. If safe replay is unavailable, query status using a stable operation ID or reconcile the destination before submitting again. HTTP’s idempotency rules distinguish repeatable intended effects from an identical response. A retry can return a different status while leaving the same intended state.

For example, retrying a batch after only some rows committed can duplicate those rows. Lab 4 uses a smaller case: the server stores one row, returns 500, and the client repeats the write. It demonstrates duplicate effects after a reported failure, not a real crash or a multi-row transaction.

Idempotency: making retries safe

An operation has idempotency when repeating the same logical operation has the same intended effect as applying it once. “Set the customer’s city to Seoul” is idempotent; “add 100 points to the balance” is not, and a retry of it is a second deposit. The standard repair is an idempotency key: the caller generates a unique ID for the operation and sends it with every attempt, and the server records completed keys and, on seeing one again, returns the original result instead of doing the work twice. Setting the city again does not add a second change, but a stale retry could overwrite a later update unless a version condition protects it. The following single-threaded in-memory sketch shows replay only; durable and concurrent handling needs more:

balance = {"alice": 0}
completed = {}

def add_points(customer, points, key=None):
    if key in completed:
        return completed[key]
    balance[customer] += points
    result = {"balance": balance[customer]}
    if key is not None:
        completed[key] = result
    return result

for attempt in range(2):
    add_points("alice", 100)
print(balance["alice"])
# 200
balance["alice"] = 0
for attempt in range(2):
    add_points("alice", 100, key="order-8813")
print(balance["alice"])
# 100

The caller reuses one key for one logical operation, not a new key per retry. A production service must scope keys to the caller and operation, reject the same key with a different payload, and coordinate concurrent requests. It must commit the effect and deduplication record atomically in durable storage, retain that record across the retry window, and define behavior while a request is still running. A dictionary disappears on restart and cannot establish those guarantees.

• Deterministic destination. Write each run’s output to a path derived from its inputs, such as /raw/orders/date=2026-03-14/, and overwrite it. A retry replaces the file instead of adding a second one. The same input version must produce the same output, and publication must prevent partial or concurrent overwrites.
• Merge instead of insert. Load into a staging table, then merge on a business key. Use unique business keys and deterministic updates; an update that adds to a balance on each run is still not idempotent.

Retry safety prevents duplicate business effects; it does not prevent excess load. Bound retries even for idempotent operations. If a key expires or a dependent side effect is outside the atomic commit, repeating the request can become unsafe again.

Retrying without making things worse

Suppose a service slows down under load. Two hundred workers time out at once and retry immediately. The service now receives double its normal traffic while already struggling, fails harder, and more clients retry. This is a retry storm, and clients cause outages this way regularly.

The standard defence has three parts:

• Exponential backoff. Wait 1 s, then 2 s, 4 s, 8 s, 16 s, giving the other side room to recover.
• Jitter. Draw each wait at random from between zero and the backoff value. Without it, all 200 workers wait exactly 4 seconds and retry at the same instant, which rebuilds the stampede on a schedule. Jitter reduces synchronization; it does not add server capacity.
• A cap and a budget. Set a per-delay cap, maximum attempts including the first, and an overall deadline. Report exhaustion and retain unresolved operations for investigation or controlled replay. Unlimited retries convert a visible failure into an invisible hang.

The schedules three workers would follow, with and without jitter:

import random

rng = random.Random(0)
for worker in range(3):
    plain = [2 ** n for n in range(5)]
    jittered = [round(rng.uniform(0, 2 ** n), 1) for n in range(5)]
    print(worker, plain, jittered)
# 0 [1, 2, 4, 8, 16] [0.8, 1.5, 1.7, 2.1, 8.2]
# 1 [1, 2, 4, 8, 16] [0.4, 1.6, 1.2, 3.8, 9.3]
# 2 [1, 2, 4, 8, 16] [0.9, 1.0, 1.1, 6.0, 9.9]

The values are per-retry waits, not absolute arrival times. Random waits spread attempts, but collisions remain possible. Lab 3 simulates 200 clients and a deliberately simplified overloaded server; it does not send 200 live HTTP requests.

A server can supply Retry-After as a number of seconds or an HTTP date. Respect it when applicable; if the wait exceeds the remaining deadline, stop or defer instead of retrying early. Use independent random state in real workers; identical test seeds would recreate synchronization. Also choose one layer to own retries: three attempts at each of three nested layers can produce up to 27 downstream calls. Cloud Storage’s retry guidance covers transient errors, idempotency, and retry anti-patterns.

Timeouts

Timeout defaults and meanings differ by client. Requests has no timeout unless supplied; HTTPConnection uses the global socket default unless overridden. Identify three separate budgets:

• Connect timeout: time allowed to establish connectivity. Check whether DNS resolution and TLS setup are included by your client.
• Read timeout: how long to wait for the next piece of data, based on the endpoint’s observed behaviour.
• Overall deadline: the whole operation, retries included. This is what protects the schedule.

Choose thresholds from the acceptable false-timeout rate, latency tails, cold connection behavior, and the business deadline. A read timeout usually bounds inactivity or an individual blocking read, not total download duration: a peer sending small chunks often enough can keep it alive. A true overall deadline must cover setup, reads, backoff, and retries using elapsed monotonic time and supported cancellation or deadline propagation. Setting the same socket timeout on each attempt does not do that. See HTTPConnection’s timeout contract.

For a 10-second overall budget, two attempts taking 3 seconds each plus 2 seconds of backoff leave only 2 seconds. A third attempt must fit that remainder or be deferred; giving it another full 3 seconds would exceed the budget. Propagate the remaining time to downstream work, and record attempt count, elapsed time, and whether the final outcome is known. Canceling local waiting does not roll back a remote commit.

When the producer is faster than the consumer

Data arrives at whatever rate the source produces it, and the system processes at whatever rate it can. During an imbalance, the system needs a policy for excess arrivals. Buffering, dropping, and backpressure can be combined; increasing capacity or reducing per-record work may also be needed for a sustained imbalance:

• Buffer: hold the excess in memory or on disk. Use a bounded buffer and define what happens when it fills: block, reject, spill, or drop. A process crash loses volatile memory; durable buffers have different recovery guarantees.
• Drop: discard what cannot be handled. Acceptable for some metrics, unacceptable for payments.
• Backpressure: tell the producer to slow down, or stop reading so that it blocks.

A queue buys time, not processing capacity. If arrivals stay at 1,200 records/s and consumption at 1,000 records/s, backlog grows by 200 records/s; 60,000 free slots last 300 seconds. A durable queue still needs capacity, retention, and recovery settings. Monitor queue depth and oldest-record age, and test the full-buffer policy of each collector rather than assuming it slows the source safely. Stopping one reader may only move the backlog into an upstream buffer. Recovery must process faster than ongoing arrivals to drain the queue.

Moving large files

Transferring a 50 GB file over a network that occasionally drops has its own small set of practices:

• Verify with a checksum. Obtain the expected SHA-256 and exact length for a fixed source version, then recompute them. Length catches truncation; the digest also catches same-length corruption. Authenticate the manifest source: a digest cannot detect an attacker replacing both file and manifest.
• Write to a temporary name, then rename. Use atomic publication: on a filesystem that supports atomic replacement, publish a validated temporary file on the same filesystem. Atomic visibility is not crash durability; object stores need their own commit or version protocol. Readers must watch only the published destination, not temporary files.
• Use multipart or resumable transfer. With supported SDK settings, multipart or resumable uploads can retry failed parts instead of restarting 50 GB after a final 200 MB failure. Resuming after a process restart also requires retained session or upload state. Confirm expiry, source-version rules, and cleanup of abandoned parts; not every client resumes automatically. The S3 multipart upload documentation describes upload IDs and completion or abort handling.
• Ask for a manifest. A list of expected filenames, row counts, and checksums turns “did we get everything?” into a check.

Lab

The labs use Python 3.9 or later and only the standard library. They require permission to open a loopback port on 127.0.0.1. Run the server block once in the same process as the solutions; they depend on its HOST, PORT, and state. This is a local teaching server, with no authentication, TLS, or durable storage. Lab 3 is a separate simulation. Close the server after finishing as described below.

import hashlib, json, threading, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

RECORDS = [{"id": i, "amount": i * 1.5} for i in range(1000)]
FILE_BYTES = b"".join(f"row-{i:06d}\n".encode() for i in range(50_000))
STATE = {"rows": [], "completed": {}, "fail_after_write": 0}
STATE_LOCK = threading.Lock()

class MisbehavingHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    disable_nagle_algorithm = True

    def log_message(self, *args):
        pass

    def reply(self, status, body):
        self.send_response(status)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        try:
            self.wfile.write(body)
        except (BrokenPipeError, ConnectionResetError):
            # The timeout lab deliberately closes a connection before the reply.
            self.close_connection = True

    def do_GET(self):
        path, _, query = self.path.partition("?")
        params = dict(p.split("=") for p in query.split("&") if p)
        if path == "/record":
            self.reply(200, json.dumps(RECORDS[int(params["id"])]).encode())
        elif path == "/records":
            start, size = int(params.get("start", 0)), int(params.get("size", 100))
            self.reply(200, json.dumps(RECORDS[start:start + size]).encode())
        elif path == "/slow":
            time.sleep(float(params.get("seconds", 3)))
            self.reply(200, b"finally")
        elif path == "/manifest":
            manifest = {"size": len(FILE_BYTES), "sha256": hashlib.sha256(FILE_BYTES).hexdigest()}
            self.reply(200, json.dumps(manifest).encode())
        elif path == "/file":
            cut_at = int(params.get("cut_at", len(FILE_BYTES)))
            self.send_response(200)
            self.send_header("Connection", "close")
            self.end_headers()
            self.wfile.write(FILE_BYTES[:cut_at])
            self.close_connection = True
        else:
            self.reply(404, b"no such path")

    def do_POST(self):
        if self.path != "/points":
            self.close_connection = True
            self.reply(404, b"no such path")
            return
        payload = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        key = self.headers.get("Idempotency-Key")
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        with STATE_LOCK:
            if key and key in STATE["completed"]:
                original, body = STATE["completed"][key]
                if canonical != original:
                    self.reply(409, b"key reused with a different payload")
                else:
                    self.reply(200, body)
                return
            STATE["rows"].append(payload)
            body = json.dumps({"stored": len(STATE["rows"])}).encode()
            if key:
                STATE["completed"][key] = (canonical, body)
            if STATE["fail_after_write"] > 0:
                STATE["fail_after_write"] -= 1
                self.reply(500, b"simulated error after writing")
                return
            self.reply(200, body)

server = ThreadingHTTPServer(("127.0.0.1", 0), MisbehavingHandler)
threading.Thread(target=server.serve_forever, daemon=True).start()
HOST, PORT = server.server_address
print("listening on", HOST)
# listening on 127.0.0.1

HTTP/1.1 permits reuse, and the handler disables Nagle’s small-write buffering to reduce one source of local timing artifacts; this is not a production tuning recommendation. The POST handler stores a row before returning a simulated 500. A lock coordinates its in-memory key check and write, and a different payload under the same key returns 409. This protects the running demo from concurrent duplicate keys but does not survive restart or model a database commit.

1. Measure the round-trip cost. Fetch 1,000 records one per request with a new connection each time, then again over one reused connection, then in pages of 100, then in a single request. Record the four durations.

import http.client, json, time, urllib.request

t = time.perf_counter()
fresh = []
for i in range(1000):
    with urllib.request.urlopen(f"http://{HOST}:{PORT}/record?id={i}", timeout=5) as response:
        fresh.append(json.loads(response.read()))
fresh_seconds = time.perf_counter() - t

conn = http.client.HTTPConnection(HOST, PORT, timeout=5)
t = time.perf_counter()
reused = []
for i in range(1000):
    conn.request("GET", f"/record?id={i}")
    reused.append(json.loads(conn.getresponse().read()))
reused_seconds = time.perf_counter() - t

t = time.perf_counter()
paged = []
for start in range(0, 1000, 100):
    conn.request("GET", f"/records?start={start}&size=100")
    paged.extend(json.loads(conn.getresponse().read()))
paged_seconds = time.perf_counter() - t

t = time.perf_counter()
conn.request("GET", "/records?start=0&size=1000")
bulk = json.loads(conn.getresponse().read())
bulk_seconds = time.perf_counter() - t
conn.close()

print(len(fresh), len(reused), len(paged), len(bulk), fresh == reused == paged == bulk)
# 1000 1000 1000 1000 True
print(f"new connection each time {fresh_seconds:.2f} s | one connection {reused_seconds:.2f} s"
      f" | 10 pages {paged_seconds:.3f} s | 1 request {bulk_seconds:.3f} s")
# new connection each time 0.53 s | one connection 0.20 s | 10 pages 0.004 s | 1 request 0.001 s # varies by machine

2. Hang a request. Ask the server to sleep before answering. Call it with no timeout and watch the client wait the full time. Set a read timeout shorter than the sleep and watch it give up.

import http.client, time

conn = http.client.HTTPConnection(HOST, PORT, timeout=None)
t = time.perf_counter()
conn.request("GET", "/slow?seconds=2")
print(conn.getresponse().read(), f"after {time.perf_counter() - t:.1f} s")
# b'finally' after 2.0 s   # varies by machine
conn.close()

conn = http.client.HTTPConnection(HOST, PORT, timeout=0.5)
t = time.perf_counter()
try:
    conn.request("GET", "/slow?seconds=2")
    conn.getresponse().read()
except TimeoutError as exc:
    print(type(exc).__name__)
    # TimeoutError
    print(f"gave up after {time.perf_counter() - t:.1f} s")
finally:
    conn.close()
# gave up after 0.5 s # varies by machine

3. Build a retry storm. Simulate 200 clients against a server that handles 100 requests per second when healthy and only 10 when overloaded, because an overloaded server spends its time rejecting. Run the crowd with immediate retries, with exponential backoff, and with backoff plus jitter, and count the requests the server receives each second.

import random

def simulate(policy, clients=200, capacity=100, seed=0, limit=120):
    rng = random.Random(seed)
    next_try = {c: 0 for c in range(clients)}
    failures = {c: 0 for c in range(clients)}
    load = []
    while next_try and len(load) < limit:
        t = len(load)
        due = [c for c, when in next_try.items() if when <= t]
        load.append(len(due))
        served = len(due) if len(due) <= capacity else capacity // 10
        rng.shuffle(due)
        for c in due[:served]:
            del next_try[c]
        for c in due[served:]:
            failures[c] += 1
            base = min(2 ** failures[c], 32)
            if policy == "immediate":
                wait = 1
            elif policy == "backoff":
                wait = base
            else:
                wait = rng.uniform(1, base)
            next_try[c] = t + wait
    return load, not next_try

for policy in ("immediate", "backoff", "jitter"):
    load, finished = simulate(policy)
    print(f"{policy:9s} finished={finished} seconds={len(load):3d} requests={sum(load):4d}"
          f" first 12 s: {load[:12]}")
# immediate finished=True seconds= 11 requests=1650 first 12 s: [200, 190, 180, 170, 160, 150, 140, 130, 120, 110, 100]
# backoff   finished=False seconds=120 requests=1190 first 12 s: [200, 0, 190, 0, 0, 0, 180, 0, 0, 0, 0, 0]
# jitter    finished=True seconds=  7 requests= 570 first 12 s: [200, 0, 190, 0, 66, 59, 55]

4. Create duplicates, then prevent them. Tell the server to write a row and then fail. Retry and confirm the row exists twice. Add an idempotency key and confirm the second attempt returns the first result without writing again.

import http.client, json

def post_points(payload, key=None, attempts=3):
    conn = http.client.HTTPConnection(HOST, PORT, timeout=5)
    headers = {"Content-Type": "application/json"}
    if key:
        headers["Idempotency-Key"] = key
    try:
        for attempt in range(1, attempts + 1):
            conn.request("POST", "/points", body=json.dumps(payload), headers=headers)
            response = conn.getresponse()
            body = response.read()
            if response.status == 200:
                return attempt, json.loads(body)
            if response.status != 500:
                raise RuntimeError(f"HTTP {response.status}: {body.decode()}")
        raise RuntimeError("demo attempt limit exhausted")
    finally:
        conn.close()

STATE["rows"].clear(); STATE["completed"].clear()
STATE["fail_after_write"] = 1
print(post_points({"customer": "alice", "points": 100}))
# (2, {'stored': 2})
print(STATE["rows"])
# [{'customer': 'alice', 'points': 100}, {'customer': 'alice', 'points': 100}]

STATE["rows"].clear(); STATE["completed"].clear()
STATE["fail_after_write"] = 1
print(post_points({"customer": "alice", "points": 100}, key="order-8813"))
# (2, {'stored': 1})
print(STATE["rows"])
# [{'customer': 'alice', 'points': 100}]
try:
    post_points({"customer": "alice", "points": 200}, key="order-8813")
except RuntimeError as exc:
    print(str(exc))
# HTTP 409: key reused with a different payload
print(len(STATE["rows"]))
# 1

5. Catch a truncated file. Serve a file but cut the connection halfway. Confirm a plain download accepts the partial file. Add a manifest check and a write-then-rename step, and confirm it now refuses the half file and accepts the whole one.

import hashlib, json, os, tempfile, urllib.request

def download(cut_at):
    with urllib.request.urlopen(f"http://{HOST}:{PORT}/file?cut_at={cut_at}", timeout=5) as response:
        return response.read()

half = download(cut_at=len(FILE_BYTES) // 2)
print(len(half), len(FILE_BYTES))
# 275000 550000

with urllib.request.urlopen(f"http://{HOST}:{PORT}/manifest", timeout=5) as response:
    manifest = json.loads(response.read())

def download_verified(cut_at, dest):
    digest = hashlib.sha256()
    received = 0
    fd, partial = tempfile.mkstemp(dir=os.path.dirname(dest), suffix=".part")
    try:
        with os.fdopen(fd, "wb") as output:
            url = f"http://{HOST}:{PORT}/file?cut_at={cut_at}"
            with urllib.request.urlopen(url, timeout=5) as response:
                while True:
                    chunk = response.read(64 * 1024)
                    if not chunk:
                        break
                    received += len(chunk)
                    if received > manifest["size"]:
                        return False, os.path.exists(dest)
                    digest.update(chunk)
                    output.write(chunk)
        ok = (received == manifest["size"]
              and digest.hexdigest() == manifest["sha256"])
        if ok:
            os.replace(partial, dest)
        return ok, os.path.exists(dest)
    finally:
        if os.path.exists(partial):
            os.remove(partial)

work = tempfile.TemporaryDirectory()
dest = os.path.join(work.name, "orders.txt")
print(download_verified(len(FILE_BYTES) // 2, dest))
# (False, False)
print(download_verified(len(FILE_BYTES), dest))
# (True, True)
print(download_verified(len(FILE_BYTES) // 2, dest))
# (False, True)
with open(dest, "rb") as saved:
    print(hashlib.sha256(saved.read()).hexdigest() == manifest["sha256"])
# True
work.cleanup()

Python’s os.replace describes atomic replacement and cross-filesystem constraints. This lab uses a fixed local source and one writer. Real transfers also need authenticated transport, a version-bound manifest, an overall deadline, and a policy for concurrent publishers and crash durability. After all labs, call server.shutdown() and then server.server_close() from the main thread.