diff --git a/scripts/ci/xdist_watchdog.py b/scripts/ci/xdist_watchdog.py index 93e8082756..5a7385bbc5 100644 --- a/scripts/ci/xdist_watchdog.py +++ b/scripts/ci/xdist_watchdog.py @@ -24,6 +24,27 @@ d'acceptance de l'issue : un run bloque doit echouer vite, en disant pourquoi. +Mode 2 -- le garde lui-meme tuait un run sain (run 35276661841, PR #16240, +attempt 2, 2026-09-17 23:27Z) : en fin de parcours sous ``-q``, pytest +ecrit ses points de test SANS saut de ligne tant que la ligne de ~72 +caracteres n'est pas pleine. Le fil de lecture, base sur ``readline``, +restait bloque sur le fragment sans ``\\n`` pendant que des octets vivants +traversaient le tube : verdict « silence 480 s ... ``[99%]`` ... workers +morts : gw2 », kill d'un run a 99 % sans echec -- et le flush d'EOF du +kill a laisse la preuve sur le log : une ligne partielle de 43 resultats +emis PENDANT la fenetre dite muette (23:19:06 -> 23:27:07). Le master +n'etait pas bloque, il finissait la queue (tests de queue lents, puis +re-execution du lot du worker remplace). Correctif : la fraicheur se +mesure desormais par OCTET lu (``os.read`` par chunks, lignes +reconstituees en interne), pas par ligne complete. Un run qui emet ne +peut plus etre tue ; un blocage reel (zero octet, signature originelle) +l'est toujours. + +Une tolerance au pourcentage de fin de parcours (« [95%+] => grace ») a +ete ecarte a dessin : le blocage originel #16288 s'est AUSSI produit a +``[99%]`` -- le pourcentage ne discrimine rien, seul le flux d'octets le +fait. + Pistes rejetees, pour memoire (detail dans #16288) : - ``pytest-timeout`` + ``--timeout`` : borne un TEST qui bloque ; ici @@ -92,6 +113,26 @@ def __init__(self) -> None: self.last_progress_line: str | None = None self.dead_workers: list[str] = [] self.line_count = 0 + # Forensique mode 2 : des octets sans aucune ligne complete (points + # de fin de parcours sans \n) sont le signe d'un run VIVANT -- le + # verdict doit pouvoir les citer apres coup. + self.byte_count = 0 + + def record_bytes(self, nbytes: int) -> None: + """Fraicheur par OCTET, pas par ligne complete (mode 2). + + Sous ``-q``, les points de test n'emportent pas de ``\\n`` avant + que la ligne de ~72 caracteres soit pleine : un fil base sur + ``readline`` ne verrait rien pendant que le run finit sa queue + (run 35276661841 : 43 resultats emis pendant une fenetre que le + garde croyait muette). La boucle de surveillance appelle donc + ceci sur CHAQUE chunk lu du tube. + """ + if nbytes <= 0: + return + with self.lock: + self.last_output = time.monotonic() + self.byte_count += nbytes def record(self, line: str) -> None: now = time.monotonic() @@ -112,13 +153,45 @@ def idle_for(self) -> float: return time.monotonic() - self.last_output +# Taille d'un chunk de lecture du tube. Un seul os.read = un seul appel +# systeme : il retourne des le PREMIER octet disponible (jamais apres un +# remplissage complet ni un \n) -- c'est la propriete qui repare le mode 2. +PIPE_CHUNK_BYTES = 65536 + + def _pump(stream, state: _StreamState, echo) -> None: - """Lit les lignes d'un pipe du fils, les horodate et les recopie.""" - # Sentinelle en BYTES : le pipe est binaire (Popen sans text=True), - # et b"" == "" est faux -- un sentinelle str ferait boucler le fil - # a l'infini apres l'EOF, inondant l'echo de lignes vides. - for raw in iter(stream.readline, b""): - line = raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + """Recopie la sortie du fils en mesurant l'activite par OCTET. + + On lit le tube par chunks bruts (``os.read`` sur le fd) et on + reconstitue les lignes en interne : la fraicheur (``record_bytes``) + est mise a jour pour CHAQUE chunk, le decoupage en lignes ne sert + qu'au bookkeeping (progression, workers morts) et a l'echo. + + Pourquoi pas ``readline`` : en fin de parcours ``-q``, pytest ecrit + ses points SANS saut de ligne -- un readline resterait bloque sur le + fragment pendant que le run finit sa queue, et le garde tuerait un + vivant (mode 2, run 35276661841). Un fragment final sans ``\\n`` + n'est recopie qu'a l'EOF (ou au kill), exactement comme la ligne + partielle de 43 resultats revelee par le flush d'EOF du run cite. + """ + fd = stream.fileno() + pending = b"" + while True: + try: + chunk = os.read(fd, PIPE_CHUNK_BYTES) + except OSError: + break # tube ferme sous nos pieds : equivaut a l'EOF + if not chunk: + break + state.record_bytes(len(chunk)) + pending += chunk + while b"\n" in pending: + raw, pending = pending.split(b"\n", 1) + line = raw.decode("utf-8", errors="replace") + "\n" + state.record(line) + echo(line) + if pending: + line = pending.decode("utf-8", errors="replace") state.record(line) echo(line) try: @@ -210,11 +283,13 @@ def _verdict_blocked(state: _StreamState, idle: float, idle_limit: float, emit(f"##[error]{VERDICT_PREFIX}: BLOQUE -- silence de sortie depuis " f"{idle:.0f} s (limite {idle_limit:.0f} s), mur du job non atteint") emit(f"##[error]{VERDICT_PREFIX}: derniere progression pytest : " - f"\"{progress}\" ; {state.line_count} lignes emises au total ; " - f"wall du wrapper {wall:.0f} s") + f"\"{progress}\" ; {state.line_count} lignes ({state.byte_count} " + f"octets) emises au total ; wall du wrapper {wall:.0f} s") emit(f"##[error]{VERDICT_PREFIX}: workers morts : {workers}") - emit(f"##[error]{VERDICT_PREFIX}: le master etait vivant mais n'attendait " - f"pas du travail -- signature #16288 ; kill du groupe de processus") + emit(f"##[error]{VERDICT_PREFIX}: zero octet emis pendant la fenetre " + f"(ni ligne ni fragment) -- le master etait vivant mais " + f"n'attendait pas du travail, signature #16288 ; kill du groupe " + f"de processus") def main(argv: list[str] | None = None) -> int: diff --git a/scripts/tests/test_xdist_watchdog.py b/scripts/tests/test_xdist_watchdog.py index ffcebd28a9..19ebafb391 100644 --- a/scripts/tests/test_xdist_watchdog.py +++ b/scripts/tests/test_xdist_watchdog.py @@ -16,6 +16,15 @@ mur, et c'est ce qui fait lire un blocage comme un depassement. 4. **Arme des le demarrage** -- un enfant muet depuis sa naissance (hang de collection) est aussi tue : l'armement ne depend pas d'une premiere ligne. +5. **La fraicheur se mesure par OCTET, pas par ligne** (mode 2, run + 35276661841 / PR #16240) : en fin de parcours ``-q``, pytest emet ses + points SANS saut de ligne tant que la ligne de ~72 caracteres n'est pas + pleine. Un garde qui n'ecouterait que les lignes completes croit a un + silence de 8 min et tue un run a ``[99%]`` en train de finir -- le + flush d'EOF du kill avait laisse sur le log une ligne partielle de 43 + resultats emis PENDANT la fenetre dite muette. Des octets vivants sans + ``\\n`` doivent donc maintenir la fraicheur ; et un vrai blocage (zero + octet) doit toujours mourir. Les enfants sont des ``python -c`` mono-processus : aucun xdist requis (l'issue note le defaut propre a la classe de runner ; le garde doit etre @@ -113,6 +122,72 @@ def test_bloque_apres_progression_tue_et_nomme_le_worker(): assert "XDIST-WATCHDOG" not in out # le verdict ne pollue pas la sortie pilote +def test_fin_de_parcours_points_partiels_non_tue(): + # Mode 2, run 35276661841 : des octets vivants SANS \n (points de fin + # de parcours -q) doivent maintenir la fraicheur. Un garde qui + # n'ecouterait que les lignes completes croirait a un silence et + # tuerait ce run a 1,0 s de limite -- c'est exactement le faux + # positif qui a bloque la PR #16240 a [99%]. + code, out, verdict = _run_watchdog( + _child(""" + import sys, time + for i in range(16): + sys.stdout.write(".") + sys.stdout.flush() + time.sleep(0.15) + raise SystemExit(0) + """), + idle_limit=1.0, + ) + assert code == 0 + assert "XDIST-WATCHDOG" not in verdict + assert "." in out + + +def test_fragment_final_sans_saut_de_ligne_recopie(): + # Pass-through du fragment final : le dernier chunk sans \n doit etre + # recopie a l'EOF. La ligne partielle de 43 resultats du run + # 35276661841 n'aurait jamais du etre invisible jusqu'au kill. + code, out, verdict = _run_watchdog( + _child(""" + import sys + print(".... [ 99%]", flush=True) + sys.stdout.write("...............s...........................") + sys.stdout.flush() + raise SystemExit(0) + """), + idle_limit=10.0, + ) + assert code == 0 + assert "[ 99%]" in out + assert "...............s" in out + assert verdict == "" + + +def test_bloque_apres_fragment_partiel_tue_quand_meme(): + # Garde-fou anti-regression : la fraicheur par octet ne doit pas + # epargner les vrais blocages. Signature [99%] + worker mort, un + # dernier fragment partiel, puis ZERO octet : le kill doit partir + # (fenetre comptee depuis le DERNIER OCTET, pas la derniere ligne) + # et le verdict doit citer les octets pour la forensique. + code, out, verdict = _run_watchdog( + _child(""" + import sys, time + print("............................ [ 99%]", flush=True) + print("[gw2] node down: Not properly terminated", flush=True) + sys.stdout.write("..") + sys.stdout.flush() + time.sleep(300) + """), + idle_limit=1.0, + ) + assert code == wd.EXIT_BLOCKED + assert "BLOQUE" in verdict + assert "gw2" in verdict + assert "99%" in verdict + assert "octets" in verdict + + def test_bloque_muet_des_la_naissance_tue_aussi(): # Hang de collection : aucune ligne jamais emise. Le garde doit etre # arme des le demarrage, pas apres une premiere ligne.