okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauffix(python): release obsolete job output watchers6c003b5

main
branchesmainall refs →
3.9 KiB107 linesraw
1"""Watcher ownership and completion/read races need GC and busy-host control that daemon E2E cannot reliably provide."""
2
3from pathlib import Path
4import sys
5import textwrap
6import unittest
7
8sys.path.insert(0, str(Path(__file__).resolve().parent))
9from job_wake_test import Owner  # noqa: E402
10
11
12class JobWatchersTest(unittest.TestCase):
13    def setUp(self):
14        self.owner = Owner(["run", "files"])
15        self.addCleanup(self.owner.close)
16        self.owner.wait_for(lambda frame: frame.get("type") == "ready")
17        self.cells = 0
18        self.cell(
19            "import __main__ as kernel, asyncio, gc, weakref\n"
20            "from albedo_plugins import run as runner, files as file_plugin"
21        )
22
23    def cell(self, code):
24        self.cells += 1
25        cell_id = f"c{self.cells}"
26        self.owner.send(
27            {"type": "execute", "id": cell_id, "code": textwrap.dedent(code)}
28        )
29        done = self.owner.wait_for(
30            lambda frame: frame.get("type") == "done" and frame.get("id") == cell_id
31        )
32        self.assertEqual(done["status"], "ok", done)
33        return done
34
35    def test_an_early_read_does_not_consume_the_completion_watcher(self):
36        self.cell("""
37            job = run('echo', 'result')
38            output.read(job.id)
39            await asyncio.shield(job.task)
40            assert not job._read and not job._awaited
41            assert output.read(job.id).strip() == 'result'
42            assert job._read
43        """)
44
45    def test_forgotten_jobs_and_internal_file_commands_release_watchers(self):
46        self.cell("""
47            job = run('true')
48            await job
49            reference = weakref.ref(job)
50            runner.forget(job)
51            del job
52            await file_plugin._run('printf internal')
53            await asyncio.sleep(0)
54            gc.collect()
55            assert reference() is None
56            assert not jobs and not kernel.READ_WATCHERS
57            assert not [c for c in kernel.ARCHIVES.values() if c.kind == 'job']
58        """)
59
60    def test_eviction_releases_jobs_but_preserves_held_handles(self):
61        self.cell("""
62            held = run('echo', 'held output')
63            await held
64            disposable = run('true')
65            await disposable
66            reference = weakref.ref(disposable)
67            capture = disposable.capture
68            del disposable
69            for index in range(runner.RETAINED_LIMIT):
70                # Keep this output past job eviction to catch strong watcher ownership.
71                kernel.remember(capture)
72                await run('true')
73            await asyncio.sleep(0)
74            gc.collect()
75            assert reference() is None and capture.id in kernel.ARCHIVES
76            assert capture.id not in kernel.READ_WATCHERS
77            assert held.id not in jobs and held.id not in kernel.ARCHIVES
78            assert held.id not in kernel.READ_WATCHERS
79            assert held.tail().strip() == 'held output' and await held is held
80        """)
81
82    def test_other_callables_and_replacement_watchers_remain_usable(self):
83        self.cell("""
84            capture = kernel.background_capture('callables')
85            reads = []
86            kernel.watch_output(capture.id, lambda: reads.append('function'))
87            gc.collect()
88            output.read(capture.id)
89            class Callback:
90                def __call__(self):
91                    reads.append('callable')
92            callback = Callback()
93            kernel.watch_output(capture.id, callback)
94            output.read(capture.id)
95            kernel.watch_output(capture.id, callback.__call__)
96            kernel.watch_output(capture.id, lambda: reads.append('replacement'))
97            del callback
98            gc.collect()
99            output.read(capture.id)
100            assert reads == ['function', 'callable', 'replacement']
101            kernel.forget_output(capture.id)
102            assert capture.id not in kernel.READ_WATCHERS
103        """)
104
105
106if __name__ == "__main__":
107    unittest.main()