diff --git a/unshackle/core/downloaders/requests.py b/unshackle/core/downloaders/requests.py index 56b51335..cc2c84aa 100644 --- a/unshackle/core/downloaders/requests.py +++ b/unshackle/core/downloaders/requests.py @@ -29,6 +29,9 @@ READ_TIMEOUT = 30 HEDGE_FACTOR = 3 HEDGE_MIN_WAIT = 5.0 +# racers read per network arrival (read1) so superseded hedge losers exit promptly +RACER_READ1 = True + # Adaptive chunk sizing — benchmarked optimal range MIN_CHUNK = 524_288 # 512KB MAX_CHUNK = 4_194_304 # 4MB @@ -326,7 +329,11 @@ def download( if use_rnet: chunks = stream.stream() elif use_raw: - chunks = iter(lambda: stream.raw.read(chunk_size), b"") + _read1 = getattr(stream.raw, "read1", None) if claimed is not None and RACER_READ1 else None + if _read1 is not None: + chunks = iter(lambda: _read1(chunk_size), b"") + else: + chunks = iter(lambda: stream.raw.read(chunk_size), b"") else: chunks = stream.iter_content(chunk_size=chunk_size) @@ -336,6 +343,13 @@ def download( for chunk in chunks: if DOWNLOAD_CANCELLED.is_set(): break + if claimed is not None and claimed(): + # close the handle or Windows can't delete the stray .!dev at merge (WinError 32) + try: + stream.close() + except Exception: + pass + return _write(chunk) download_size = len(chunk) written += download_size @@ -387,6 +401,9 @@ def download( stream.close() except Exception: pass + if claimed is not None and claimed(): + # a superseded loser's error must not retry or kill the batch + return if DOWNLOAD_CANCELLED.is_set() or attempts == MAX_ATTEMPTS: if part_mode and not DOWNLOAD_CANCELLED.is_set(): raise @@ -702,7 +719,9 @@ def requests( yield dict(downloaded="[yellow]CANCELLED") raise finally: - pool.shutdown(wait=False, cancel_futures=True) + # losers must close their handles before merge sweeps *.!dev (WinError 32); + # no wait on cancel/fail: merge never runs and workers may be blocked in reads + pool.shutdown(wait=not DOWNLOAD_CANCELLED.is_set(), cancel_futures=True) # Drain remaining events while True: