Pull chain: 5-minute idle watchdog per mirror
A dead mirror emitted zero bytes and the fixed 3600s attempt timeout let it stall the whole chain for an hour (third occurrence tonight: 15m49s frozen at 0 bytes). select() with a 300s idle timeout now kills the attempt and moves to the next mirror; a trickling source (1 line/s Downloading updates) resets the timer and stays alive. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
3ae1578325
commit
d58fdb4198
@ -115,6 +115,15 @@ def _pull_one(image: str, verbose: bool = True) -> None:
|
|||||||
proc = subprocess.Popen(['docker', 'pull', ref],
|
proc = subprocess.Popen(['docker', 'pull', ref],
|
||||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||||||
bufsize=0)
|
bufsize=0)
|
||||||
|
# idle watchdog: a dead mirror emits NOTHING (0 bytes/s for hours);
|
||||||
|
# the old fixed 3600s timeout let one dead source stall the whole
|
||||||
|
# chain. 5 minutes of silence = kill and try the next mirror.
|
||||||
|
import os as _os
|
||||||
|
import selectors as _sel
|
||||||
|
|
||||||
|
_selr = _sel.DefaultSelector()
|
||||||
|
_selr.register(proc.stdout, _sel.EVENT_READ)
|
||||||
|
_fd = proc.stdout.fileno()
|
||||||
last_line = ''
|
last_line = ''
|
||||||
last_show = 0.0
|
last_show = 0.0
|
||||||
layer_prog = {} # layer id -> (done_bytes, total_bytes): the raw
|
layer_prog = {} # layer id -> (done_bytes, total_bytes): the raw
|
||||||
@ -134,20 +143,27 @@ def _pull_one(image: str, verbose: bool = True) -> None:
|
|||||||
# \n -- readline() buffered them until a rare \n arrived, so the
|
# \n -- readline() buffered them until a rare \n arrived, so the
|
||||||
# Downloading snapshots (and the ⏳ summary fed by them) never
|
# Downloading snapshots (and the ⏳ summary fed by them) never
|
||||||
# surfaced. Split on BOTH terminators, chunked reads.
|
# surfaced. Split on BOTH terminators, chunked reads.
|
||||||
import io as _io
|
|
||||||
stream = _io.TextIOWrapper(proc.stdout, errors='replace')
|
|
||||||
buf = ''
|
buf = ''
|
||||||
_lines_iter = iter(lambda: None, 1) # placeholder, replaced below
|
_lines_iter = iter(lambda: None, 1) # placeholder, replaced below
|
||||||
|
|
||||||
|
_IDLE_S = 300 # no output for 5 minutes -> dead mirror
|
||||||
|
|
||||||
def _iter_lines():
|
def _iter_lines():
|
||||||
nonlocal buf
|
nonlocal buf
|
||||||
while True:
|
while True:
|
||||||
chunk = stream.read(512)
|
if not _selr.select(timeout=_IDLE_S):
|
||||||
|
proc.kill()
|
||||||
|
if verbose:
|
||||||
|
print(f'· no data for {_IDLE_S // 60}min -- dead '
|
||||||
|
'mirror, trying next source ...',
|
||||||
|
file=_sys.stderr, flush=True)
|
||||||
|
return
|
||||||
|
chunk = _os.read(_fd, 4096)
|
||||||
if not chunk:
|
if not chunk:
|
||||||
if buf:
|
if buf:
|
||||||
yield buf
|
yield buf
|
||||||
return
|
return
|
||||||
buf += chunk
|
buf += chunk.decode('utf-8', 'replace')
|
||||||
parts = buf.split('\r') if '\r' in buf else buf.split('\n')
|
parts = buf.split('\r') if '\r' in buf else buf.split('\n')
|
||||||
if len(parts) > 1:
|
if len(parts) > 1:
|
||||||
for p in parts[:-1]:
|
for p in parts[:-1]:
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user