Last active
February 22, 2025 23:45
-
-
Save ushkinaz/1274038 to your computer and use it in GitHub Desktop.
Tests HTTP mirrors of Cygwin by measuring download times.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
""" | |
Script to test HTTP mirrors of Cygwin by measuring download times. | |
Originally written in 2011, modernized in 2025. | |
""" | |
from math import floor | |
from urllib.request import urlopen | |
import argparse | |
import sys | |
import random | |
import time | |
import ssl | |
import certifi | |
import concurrent.futures | |
import multiprocessing | |
__author__ = 'Dmitry Sidorenko' | |
__maintainer__ = 'Dmitry Sidorenko (@ushkinaz on GitHub)' | |
__last_updated__ = '2025-02-22' | |
MIRRORS_URL = "https://cygwin.com/mirrors.lst" | |
TEST_FILE = "/x86_64/setup.ini" | |
BLOCK_SZ = 8096 | |
default_threads: int = floor(multiprocessing.cpu_count()/2) | |
parser = argparse.ArgumentParser(description="Test Cygwin HTTP(S) mirrors for speed.") | |
parser.add_argument("--limit", type=int, default=9999, | |
help="Limit the number of mirrors to test (default: all available)") | |
parser.add_argument("--timeout", type=int, default=5, | |
help="Timeout for each request in seconds (default: 5)") | |
parser.add_argument("--top", type=int, default=5, | |
help="Number of fastest mirrors to show (default: 5)") | |
parser.add_argument("--parallel", type=int, default=default_threads, | |
help=f"Number of threads to use. Values higher than number of CPU cores may negatively affect download times. (default: {default_threads})") | |
parser.add_argument("--include-noshow", action="store_true", | |
help="Include redundant legacy mirrors marked as 'noshow' (default: False)") | |
parser.add_argument("-s", "--silent", action="store_true", | |
help="Suppress output (only show results)") | |
args = parser.parse_args() | |
max_hosts_to_try: int = args.limit | |
request_timeout: int = args.timeout | |
max_top_mirrors: int = args.top | |
include_noshow: bool = args.include_noshow | |
parallel: int = args.parallel | |
verbose_mode: bool = not args.silent | |
mirrors = [ | |
# {"host": [], | |
# "time": 1.0 | |
# } | |
] | |
def test_mirror(host_entry): | |
"""Test a single mirror and measure its response time.""" | |
host = host_entry["host"] | |
url = f"{host[0]}{TEST_FILE}" | |
start_time = time.time() | |
try: | |
with urlopen(url, timeout=request_timeout) as response: | |
response.read(BLOCK_SZ) # Read a small block to trigger connection | |
elapsed_time = time.time() - start_time | |
except Exception: | |
elapsed_time = float("inf") # If failure, set to infinity | |
host_entry["time"] = elapsed_time | |
if verbose_mode: | |
elapsed_str = "timeout" if elapsed_time == float("inf") else f"{elapsed_time:.2f}s" | |
print(f"{host[1]:>{max_host_len}} - {elapsed_str}") | |
return host_entry # Return result | |
if verbose_mode: | |
print("Retrieving mirror list...", end='') | |
sys.stdout.flush() | |
max_host_len = 0 | |
context = ssl.create_default_context(cafile=certifi.where()) | |
response = urlopen(MIRRORS_URL, context=context).read().decode("utf-8") | |
for line in response.splitlines(): | |
host = line.split(";") | |
# Only test http(s) | |
if host[0].startswith("http") and (include_noshow or len(host) < 5 or host[4] != "noshow"): | |
mirrors.append({"host": host, "time": 9999.0}) | |
if verbose_mode: | |
print(f"done, {len(mirrors)} entries.") | |
if max_hosts_to_try == 9999: | |
max_hosts_to_try = len(mirrors) | |
if verbose_mode: | |
print(f"Checking {max_hosts_to_try} mirrors...\n") | |
mirrors_to_test = random.sample(mirrors, min(max_hosts_to_try, len(mirrors))) | |
max_host_len = max(len(host["host"][1]) for host in mirrors_to_test) + 1 | |
with concurrent.futures.ThreadPoolExecutor(max_workers=parallel) as executor: | |
mirrors = list(executor.map(test_mirror, mirrors_to_test)) | |
mirrors = sorted(mirrors, key=lambda entry: entry["time"]) | |
actual_top_mirrors = min(max_hosts_to_try, max_top_mirrors) | |
if verbose_mode: | |
print(f"\nTop {actual_top_mirrors} fastest mirrors:\n", file=sys.stderr) | |
for i in range(actual_top_mirrors): | |
mirror = mirrors[i] | |
host_info = mirror["host"] | |
if mirror["time"] < 9999: | |
print("%.3f, %14s, %15s, %s" % (mirror["time"], host_info[2], host_info[3], host_info[0]), file=sys.stderr) |
Very nice tool, thanks!
Nice, thanks for this!
works great thank you!
That is amazing! thank you so much!
Very nice, thanks!
awesome!
Cool script. Saved my time.
That's so great, thanks : )
if you got an error ..timeout
replace test_file = "/setup.ini"
to test_file = "x86/setup.ini"
Thanks !
...So is this testing ping or speed?
...So is this testing ping or speed?
It's testing speed by downloading single file from each mirror
Works Great! Thank you!
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
works like a charm, thanks