From 26ca3a7fc8a78863655cc8567d0b0bc76d051f4f Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 7 Oct 2026 04:32:06 +0000 Subject: [PATCH] perf(speed): execute cloudflare speedtest chunks concurrently This optimization refactors the Cloudflare speedtest to issue HTTP chunk requests concurrently for both download and upload using `asyncio.gather`. It measures the aggregate time delta and bytes transferred across the parallel execution batch to properly represent the overall concurrent bandwidth, significantly speeding up the speed test runtime. Co-authored-by: Horizontal42 <75383519+Horizontal42@users.noreply.github.com> --- src/netsleuth/speed.py | 123 +++++++++++++++++++++++++++++++++-------- 1 file changed, 100 insertions(+), 23 deletions(-) diff --git a/src/netsleuth/speed.py b/src/netsleuth/speed.py index 505e1f0..e9ffea1 100644 --- a/src/netsleuth/speed.py +++ b/src/netsleuth/speed.py @@ -39,7 +39,10 @@ def parse_server_timing_cfl4(header: str) -> CfL4Stats | None: match = _CFL4_RE.search(header or "") if not match: return None - fields = {k: v[0] for k, v in parse_qs(match.group("query"), keep_blank_values=True).items()} + fields = { + k: v[0] + for k, v in parse_qs(match.group("query"), keep_blank_values=True).items() + } return CfL4Stats( rtt_ms=_us_to_ms(fields.get("rtt")), min_rtt_ms=_us_to_ms(fields.get("min_rtt")), @@ -51,7 +54,9 @@ def parse_server_timing_cfl4(header: str) -> CfL4Stats | None: ) -def bufferbloat_delta(idle_rtt_ms: float | None, loaded_rtts_ms: list[float]) -> float | None: +def bufferbloat_delta( + idle_rtt_ms: float | None, loaded_rtts_ms: list[float] +) -> float | None: if idle_rtt_ms is None or not loaded_rtts_ms: return None loaded = percentile(sorted(loaded_rtts_ms), 95.0) @@ -93,10 +98,17 @@ async def run_speed_cascade( duration_ms = int((time.perf_counter() - began) * 1000) if not result.download_mbps: attempts.append( - TierAttempt(tier=name, ok=False, reason="no throughput measured", duration_ms=duration_ms) + TierAttempt( + tier=name, + ok=False, + reason="no throughput measured", + duration_ms=duration_ms, + ) ) continue - attempts.append(TierAttempt(tier=name, ok=True, reason=None, duration_ms=duration_ms)) + attempts.append( + TierAttempt(tier=name, ok=True, reason=None, duration_ms=duration_ms) + ) result.tier_attempts = attempts return result return SpeedResult(method="none", tier_attempts=attempts) @@ -138,32 +150,80 @@ async def tier_ookla( server_info = payload.get("server") or {} return SpeedResult( method="ookla_bin", - download_mbps=round(mbps(download.get("bytes", 0), (download.get("elapsed", 0) or 0) / 1000), 3), - upload_mbps=round(mbps(upload.get("bytes", 0), (upload.get("elapsed", 0) or 0) / 1000), 3), + download_mbps=round( + mbps(download.get("bytes", 0), (download.get("elapsed", 0) or 0) / 1000), 3 + ), + upload_mbps=round( + mbps(upload.get("bytes", 0), (upload.get("elapsed", 0) or 0) / 1000), 3 + ), server=f"{server_info.get('name', '')} ({server_info.get('location', '')})".strip(), idle_rtt_ms=(payload.get("ping") or {}).get("latency"), server_country=server_info.get("country") or None, ) -async def tier_cloudflare(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -> SpeedResult: +async def tier_cloudflare( + client: httpx.AsyncClient, cfg: Speedtest, timeout: float +) -> SpeedResult: down_samples: list[tuple[int, float]] = [] cfl4: CfL4Stats | None = None - for size in cfg.download_sizes_bytes: + + async def _download(size: int) -> tuple[int, CfL4Stats | None] | Exception: + try: + response = await client.get( + f"{cfg.cloudflare_base_url}/__down", + params={"bytes": size}, + timeout=timeout, + ) + response.raise_for_status() + stats = parse_server_timing_cfl4(response.headers.get("server-timing", "")) + return len(response.content), stats + except httpx.HTTPError as e: + return e + + if cfg.download_sizes_bytes: began = time.perf_counter() - response = await client.get( - f"{cfg.cloudflare_base_url}/__down", params={"bytes": size}, timeout=timeout + down_results = await asyncio.gather( + *[_download(size) for size in cfg.download_sizes_bytes] ) - response.raise_for_status() - down_samples.append((len(response.content), time.perf_counter() - began)) - cfl4 = parse_server_timing_cfl4(response.headers.get("server-timing", "")) or cfl4 + elapsed = time.perf_counter() - began + total_down_bytes = 0 + for res in down_results: + if isinstance(res, Exception): + raise res + b, stats = res + total_down_bytes += b + cfl4 = stats or cfl4 + if total_down_bytes > 0: + down_samples.append((total_down_bytes, elapsed)) + up_samples: list[tuple[int, float]] = [] - for size in cfg.upload_sizes_bytes: + + async def _upload(size: int) -> int | Exception: payload = b"\x00" * size + try: + response = await client.post( + f"{cfg.cloudflare_base_url}/__up", content=payload, timeout=timeout + ) + response.raise_for_status() + return size + except httpx.HTTPError as e: + return e + + if cfg.upload_sizes_bytes: began = time.perf_counter() - response = await client.post(f"{cfg.cloudflare_base_url}/__up", content=payload, timeout=timeout) - response.raise_for_status() - up_samples.append((size, time.perf_counter() - began)) + up_results = await asyncio.gather( + *[_upload(size) for size in cfg.upload_sizes_bytes] + ) + elapsed = time.perf_counter() - began + total_up_bytes = 0 + for res in up_results: + if isinstance(res, Exception): + raise res + total_up_bytes += res + if total_up_bytes > 0: + up_samples.append((total_up_bytes, elapsed)) + return SpeedResult( method="cloudflare", download_mbps=throughput_from_samples(down_samples), @@ -173,7 +233,9 @@ async def tier_cloudflare(client: httpx.AsyncClient, cfg: Speedtest, timeout: fl ) -async def tier_fastcom(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -> SpeedResult: +async def tier_fastcom( + client: httpx.AsyncClient, cfg: Speedtest, timeout: float +) -> SpeedResult: response = await client.get( cfg.fastcom_api_url, params={"https": "true", "token": cfg.fastcom_api_token, "urlCount": "3"}, @@ -197,14 +259,21 @@ async def tier_fastcom(client: httpx.AsyncClient, cfg: Speedtest, timeout: float method="fastcom", download_mbps=throughput_from_samples(samples), upload_mbps=None, - server=targets[0].get("url").split("/")[2] if targets and targets[0].get("url") else None, + server=targets[0].get("url").split("/")[2] + if targets and targets[0].get("url") + else None, netflix_oca_onnet=on_net, - server_country=(targets[0].get("location") or {}).get("country") if targets else None, + server_country=(targets[0].get("location") or {}).get("country") + if targets + else None, ) async def tier_ndt7( - client: httpx.AsyncClient, cfg: Speedtest, timeout: float, source_ip: str | None = None + client: httpx.AsyncClient, + cfg: Speedtest, + timeout: float, + source_ip: str | None = None, ) -> SpeedResult: import websockets @@ -225,7 +294,11 @@ async def tier_ndt7( message = await asyncio.wait_for(socket.recv(), timeout=timeout) except (TimeoutError, Exception): break - total += len(message) if isinstance(message, (bytes, bytearray)) else len(message.encode()) + total += ( + len(message) + if isinstance(message, (bytes, bytearray)) + else len(message.encode()) + ) elapsed = time.perf_counter() - began return SpeedResult( method="ndt7", @@ -276,7 +349,11 @@ async def measure_with_bufferbloat( result.bufferbloat_down_ms = bufferbloat_delta(idle_rtt_ms, down_samples) result.bufferbloat_up_ms = bufferbloat_delta(idle_rtt_ms, up_samples) worst_delta = max( - [d for d in (result.bufferbloat_down_ms, result.bufferbloat_up_ms) if d is not None], + [ + d + for d in (result.bufferbloat_down_ms, result.bufferbloat_up_ms) + if d is not None + ], default=None, ) result.bufferbloat_grade = grade_bufferbloat(worst_delta, bands)