check(host) (provided) takes about 20 ms of blocking I/O and either returns a latency in ms or raises
an exception. Check all hosts using a thread pool of workers threads and return
{host: {"ok": True, "ms": ...}} or {host: {"ok": False, "error": "<ExceptionType>"}} for every host.
The harness checks that the work actually ran in parallel.
Implement check_all. Input: hosts: list[str], workers: int. Returns: dict.
This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.
import threading
import time
_lock = threading.Lock()
_state = {"active": 0, "peak": 0}
def check(host):
with _lock:
_state["active"] += 1
_state["peak"] = max(_state["peak"], _state["active"])
try:
time.sleep(0.02)
if host.startswith("bad"):
raise ConnectionError(host)
if host.startswith("slow"):
raise TimeoutError(host)
return len(host)
finally:
with _lock:
_state["active"] -= 1
def __harness__(fn, hosts, workers):
_state.update(active=0, peak=0)
result = fn(hosts, workers)
return {"result": result, "parallel": _state["peak"] > 1 or workers == 1 or len(hosts) < 2, "within_limit": _state["peak"] <= workers}
def check_all(hosts: list[str], workers: int) -> dict:
return {h: {"ok": True, "ms": check(h)} for h in hosts}
Run executes the public tests. Submit grades against public and hidden tests.
check(host) (provided) takes about 20 ms of blocking I/O and either returns a latency in ms or raises
an exception. Check all hosts using a thread pool of workers threads and return
{host: {"ok": True, "ms": ...}} or {host: {"ok": False, "error": "<ExceptionType>"}} for every host.
The harness checks that the work actually ran in parallel.
Implement check_all. Input: hosts: list[str], workers: int. Returns: dict.
This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.
import threading
import time
_lock = threading.Lock()
_state = {"active": 0, "peak": 0}
def check(host):
with _lock:
_state["active"] += 1
_state["peak"] = max(_state["peak"], _state["active"])
try:
time.sleep(0.02)
if host.startswith("bad"):
raise ConnectionError(host)
if host.startswith("slow"):
raise TimeoutError(host)
return len(host)
finally:
with _lock:
_state["active"] -= 1
def __harness__(fn, hosts, workers):
_state.update(active=0, peak=0)
result = fn(hosts, workers)
return {"result": result, "parallel": _state["peak"] > 1 or workers == 1 or len(hosts) < 2, "within_limit": _state["peak"] <= workers}
def check_all(hosts: list[str], workers: int) -> dict:
return {h: {"ok": True, "ms": check(h)} for h in hosts}
Run executes the public tests. Submit grades against public and hidden tests.