From 3e323d1f945fb29f9e8884cada27ffe3f9ab0c35 Mon Sep 17 00:00:00 2001 From: chrg6 Date: Thu, 12 Nov 2026 22:33:10 -0500 Subject: [PATCH] Change and read a job's steps under one lock. The worker, the thread that reads the build's output and the window all change a job while the window draws it. A step could be read with a new state and the value before it, the output's lines could change while they were copied, and the change counter was incremented without a lock although the window learns from it that the job ended. Every change and the counter now take the job's lock, and the job screen and the command line copy the steps under it. Co-Authored-By: Claude Fable 5.2 --- launcher/cli.py | 13 +++--- launcher/jobs.py | 91 +++++++++++++++++++++++++++--------------- launcher/ui/app.py | 20 +++++----- tests/launcher_test.py | 65 ++++++++++++++++++++++++++++++ 4 files changed, 141 insertions(+), 48 deletions(-) diff --git a/launcher/cli.py b/launcher/cli.py index 9f444542..91117b96 100644 --- a/launcher/cli.py +++ b/launcher/cli.py @@ -31,12 +31,13 @@ def follow(job): seen = {} job.start() while job.result is None: - for s in job.steps: - key = (s.state, s.value if s.state != jobs.RUNNING else int(s.fraction * 20)) - if seen.get(s.id) != key: - seen[s.id] = key - print(f'{s.label}: {s.value}' + (f' ({s.detail})' if s.detail and s.state == jobs.RUNNING else ''), - flush=True) + with job.locked(): # each step as one change of the worker left it + steps = [(s.id, s.label, s.state, s.value, s.fraction, s.detail) for s in job.steps] + for sid, label, state, value, fraction, detail in steps: + key = (state, value if state != jobs.RUNNING else int(fraction * 20)) + if seen.get(sid) != key: + seen[sid] = key + print(f'{label}: {value}' + (f' ({detail})' if detail and state == jobs.RUNNING else ''), flush=True) time.sleep(0.2) for s in job.steps: if seen.get(s.id, (None,))[0] != s.state: diff --git a/launcher/jobs.py b/launcher/jobs.py index f0f72c4f..bb1fc45a 100644 --- a/launcher/jobs.py +++ b/launcher/jobs.py @@ -147,7 +147,10 @@ class Job: self.finished = None self.after = None # what the window does next ('restart', ...) self.note = '' # a sentence for the home screen after the job (note_scan) - self._lock = threading.Lock() + # Held for every change of a step and of rev, and by a reader that wants the steps as one change left them + # (locked). The worker, the thread that reads the build's output and the window's thread (cancel) all change + # the job. Reentrant: progress() begins a step, and every change ends in _touch(). + self._lock = threading.RLock() # ---- state the window reads def step(self, sid): @@ -200,47 +203,68 @@ class Job: def elapsed(self): return (self.finished or time.time()) - self.started + def locked(self): + """The lock to read the steps under (`with job.locked():`): they are then as one change left them, a step's + state with the value and fraction that belong to it. Held only to copy what is read, never across a + wait.""" + return self._lock + # ---- changes, from the worker def _touch(self): - self.rev += 1 + # Under the lock: "rev += 1" from two threads at once can write an older count back, and a reader that had + # seen the newer one then takes the next change for one it has drawn already. The job's last change is its + # result, so the window stayed on the job screen of a job that had ended. + with self._lock: + self.rev += 1 def log(self, line): - self.lines.append(line) - self.ctx.log(line) + with self._lock: + self.lines.append(line) + self.ctx.log(line) # outside the lock: it waits for the disk self._touch() + def last_lines(self, n): + """The last n lines of the job's output, copied under the lock: the thread that reads the build's output + appends while the window draws them, and a deque changed while it is read raises.""" + with self._lock: + return list(self.lines)[-n:] + def begin(self, sid, detail=''): - s = self.step(sid) - if s and s.state != RUNNING: - s.state, s.fraction, s.detail, s.t0, s.value = RUNNING, 0.0, detail, time.time(), '0%' - s.pace.clear() - self._touch() + with self._lock: + s = self.step(sid) + if s and s.state != RUNNING: + s.state, s.fraction, s.detail, s.t0, s.value = RUNNING, 0.0, detail, time.time(), '0%' + s.pace.clear() + self._touch() def progress(self, sid, fraction, detail=''): - s = self.step(sid) - if s: - if s.state == PENDING: - self.begin(sid) - s.fraction = max(0.0, min(1.0, fraction)) - s.pace.append((time.time(), s.fraction)) - s.value = f'{int(s.fraction * 100)}%' - if detail: - s.detail = detail - self._touch() + with self._lock: + s = self.step(sid) + if s: + if s.state == PENDING: + self.begin(sid) + s.fraction = max(0.0, min(1.0, fraction)) + s.pace.append((time.time(), s.fraction)) + s.value = f'{int(s.fraction * 100)}%' + if detail: + s.detail = detail + self._touch() def finish(self, sid, value=None): - s = self.step(sid) - if s: - s.seconds = time.time() - s.t0 if s.t0 else 0.0 - s.state, s.fraction = DONE, 1.0 - s.value = value or f'Done · {mmss(s.seconds)}' - self._touch() + with self._lock: + s = self.step(sid) + if s: + s.seconds = time.time() - s.t0 if s.t0 else 0.0 + s.state, s.fraction = DONE, 1.0 + s.value = value or f'Done · {mmss(s.seconds)}' + self._touch() def skip(self, sid, value='Unchanged'): - s = self.step(sid) - if s: - s.state, s.value = SKIPPED, value - self._touch() + with self._lock: + s = self.step(sid) + if s: + s.state, s.value = SKIPPED, value + self._touch() def cancelled(self): return self.cancel_flag.is_set() @@ -313,10 +337,11 @@ class Job: self.result, self.error = 'failed', str(e) or type(e).__name__ self.ctx.log(failure_line(e) if isinstance(e, PLAIN_FAILURES) else traceback.format_exc()) finally: - for s in self.steps: - if s.state == RUNNING: - s.state = FAILED if self.result == 'failed' else PENDING - s.value = 'Stopped' if self.result == 'failed' else 'Cancelled' + with self._lock: + for s in self.steps: + if s.state == RUNNING: + s.state = FAILED if self.result == 'failed' else PENDING + s.value = 'Stopped' if self.result == 'failed' else 'Cancelled' if marked: try: (self.ctx.layout.data / MARKER).unlink(missing_ok=True) diff --git a/launcher/ui/app.py b/launcher/ui/app.py index 977d4278..4d447099 100644 --- a/launcher/ui/app.py +++ b/launcher/ui/app.py @@ -2074,15 +2074,17 @@ class App: def rows_job(self): j = self.job rows = [] - for st in j.steps: - if st.state == jobs.RUNNING: - rows.append(Row('progress', st.label, fraction=st.fraction, value=f'{int(st.fraction * 100)}%', + with j.locked(): # each step as one change of the worker left it, not a state with the value before it + steps = [(st.id, st.label, st.state, st.value, st.fraction) for st in j.steps] + for sid, label, state, value, fraction in steps: + if state == jobs.RUNNING: + rows.append(Row('progress', label, fraction=fraction, value=f'{int(fraction * 100)}%', focus_value=True, desc=None)) else: - style = {jobs.DONE: 'normal', jobs.SKIPPED: 'dim', jobs.PENDING: 'dim', jobs.FAILED: 'salmon'}[st.state] - rows.append(Row('value', st.label, st.value, style, check=st.state == jobs.DONE, desc=None, - focus_value=st.state == jobs.FAILED)) - if st.id in ('romfs', 'verify') and st is not j.steps[-1]: + style = {jobs.DONE: 'normal', jobs.SKIPPED: 'dim', jobs.PENDING: 'dim', jobs.FAILED: 'salmon'}[state] + rows.append(Row('value', label, value, style, check=state == jobs.DONE, desc=None, + focus_value=state == jobs.FAILED)) + if sid in ('romfs', 'verify') and sid != steps[-1][0]: rows.append(Row('divider')) if j.result is None: label = {'install': 'Cancel Installation', 'update': 'Cancel Update'}.get(j.kind, 'Cancel') @@ -2105,7 +2107,7 @@ class App: def job_desc(self): j = self.job cur = j.current() - lines = [ln for ln in j.lines if ln.strip() and not ln.startswith('@@')][-6:] + lines = [ln for ln in j.last_lines(200) if ln.strip() and not ln.startswith('@@')][-6:] if j.result == 'failed': return {'title': 'The Build Stopped' if j.kind != 'update' or cur.id not in ('download', 'verify') else 'The Update Stopped', @@ -2165,7 +2167,7 @@ class App: j = self.job report = [f'NX2navi {self.layout.version} ({"checkout" if self.layout.dev else "release"}), {sys.platform}', f'Job: {j.kind}, step: {j.current().label}', f'Error: {j.error}', '', 'Last lines:'] - report += [ln for ln in list(j.lines)[-60:]] + report += j.last_lines(60) self.root.clipboard_clear() self.root.clipboard_append('\n'.join(report)) self.message = 'Report copied.' diff --git a/tests/launcher_test.py b/tests/launcher_test.py index d1dbc8e2..d1dd99d1 100644 --- a/tests/launcher_test.py +++ b/tests/launcher_test.py @@ -3833,6 +3833,71 @@ class JobTest(Tmp): (game / 'romfs' / 'romfs.bin').unlink() # the image is gone: only the dump brings it back self.assertEqual(jobs.InstallJob(ctx, self.tmp / 'g.nup', game).args()[0], '--nup') + def test_a_reader_sees_each_step_as_one_change_left_it(self): + """The worker, the build's reader thread and the window change a job while the window reads it. Under + job.locked() a step's state comes with the value and fraction that belong to it, every change counts in + rev (the window draws again when rev changed, and learns from it that the job ended), and the output's + last lines can be copied while lines arrive.""" + ctx = self.make_ctx('ok') + self.addCleanup(ctx._file.close) + j = jobs.Job(ctx, ['lift', 'compile']) + stop = threading.Event() + wrong, errors = [], [] + + def work(): + while not stop.is_set(): + for sid in ('lift', 'compile'): + j.begin(sid, 'starting') + j.progress(sid, 0.5, 'half') + j.finish(sid) + j.log(f'working on {sid}') + with j.locked(): + j.step(sid).state, j.step(sid).value, j.step(sid).fraction = jobs.PENDING, 'Waiting', 0.0 + + def read(): + try: + while not stop.is_set(): + with j.locked(): + seen = [(s.state, s.value, s.fraction) for s in j.steps] + for state, value, fraction in seen: + ok = {jobs.PENDING: value == 'Waiting' and fraction == 0.0, + jobs.RUNNING: value.endswith('%') and value == f'{int(fraction * 100)}%', + jobs.DONE: value.startswith('Done') and fraction == 1.0}[state] + if not ok: + wrong.append((state, value, fraction)) + j.last_lines(6) + except Exception as e: # noqa: BLE001 - whatever a reader meets is the failure + errors.append(e) + + threads = [threading.Thread(target=work), threading.Thread(target=read), threading.Thread(target=read)] + for t in threads: + t.start() + time.sleep(0.5) + stop.set() + for t in threads: + t.join(30) + self.assertEqual((wrong[:3], errors), ([], [])) + + counted = jobs.Job(ctx, []) + touchers = [threading.Thread(target=lambda: [counted._touch() for _ in range(20000)]) for _ in range(4)] + for t in touchers: + t.start() + for t in touchers: + t.join(60) + self.assertEqual(counted.rev, 80000) # no change was lost + try: + from launcher.ui.app import App + except ImportError: + return + a = mock.MagicMock() # the job screen's rows, from a real job + a.job = jobs.InstallJob(ctx, self.tmp / 'g.nup', self.tmp / 'game') + a.job.begin('tools') + a.job.progress('tools', 0.25) + rows = App.rows_job(a) + self.assertEqual((rows[0].kind, rows[0].value, rows[0].fraction), ('progress', '25%', 0.25)) + self.assertEqual([r.kind for r in rows].count('divider'), 1) # after the game data, as before + self.assertEqual(rows[-1].kind, 'action') # Cancel + # ---------------------------------------------------------------- one launcher, one job