diff --git a/main.py b/main.py index aeb261c..ea19041 100644 --- a/main.py +++ b/main.py @@ -623,25 +623,27 @@ def start(self, start_node_id=None): raise def stop(self): - if not self.running: return - self.stop_event.set() - self.pause_event.set() with self.thread_lock: + if not self.running: return + self.stop_event.set() + self.pause_event.set() self.pending_nodes.clear() self.log("🛑 正在停止...", "warning") # The engine waits for active workers before allowing another run. def pause(self): - if not self.running or self.stop_event.is_set() or self.paused: return - self.paused = True - self.pause_event.clear() + with self.thread_lock: + if not self.running or self.stop_event.is_set() or self.paused: return + self.paused = True + self.pause_event.clear() self.log("⏸️ 流程暂停", "paused") self.app.after(0, lambda: self.app.update_debug_btn_state(True)) def resume(self): - if not self.running or not self.paused: return - self.paused = False - self.pause_event.set() + with self.thread_lock: + if not self.running or not self.paused: return + self.paused = False + self.pause_event.set() self.log("▶️ 流程继续", "info") self.app.after(0, lambda: self.app.update_debug_btn_state(False)) diff --git a/tests/test_regressions.py b/tests/test_regressions.py index d6eb280..ddab150 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -71,6 +71,15 @@ def setUpClass(cls): "AutomationCore", "start", {"threading": threading, "time": time} ) + cls.StopProbe = extract_class_with_method( + "AutomationCore", "stop" + ) + cls.PauseProbe = extract_class_with_method( + "AutomationCore", "pause" + ) + cls.ResumeProbe = extract_class_with_method( + "AutomationCore", "resume" + ) cls.MemorySnapshotProbe = extract_class_with_method( "AutomationCore", "_memory_snapshot", {"threading": threading} @@ -204,6 +213,75 @@ def after(self, delay, callback): self.assertTrue(engine.pause_event.is_set()) self.assertEqual(engine.pending_nodes, []) + def test_stop_atomically_signals_workers_and_discards_queued_nodes(self): + import threading + + class Event: + def __init__(self, value=False): + self.value = value + def set(self): + self.value = True + def clear(self): + self.value = False + def is_set(self): + return self.value + + core = self.StopProbe() + core.thread_lock = threading.Lock() + core.running = True + core.stop_event = Event() + core.pause_event = Event(False) + core.pending_nodes = ["queued-a", "queued-b"] + logs = [] + core.log = lambda *args: logs.append(args) + + core.stop() + + self.assertTrue(core.stop_event.is_set()) + self.assertTrue(core.pause_event.is_set()) + self.assertEqual(core.pending_nodes, []) + self.assertEqual(len(logs), 1) + + def test_pause_and_resume_update_state_under_lock(self): + import threading + + class Event: + def __init__(self, value=False): + self.value = value + def set(self): + self.value = True + def clear(self): + self.value = False + def is_set(self): + return self.value + + class App: + def __init__(self): + self.calls = [] + def after(self, delay, callback): + callback() + def update_debug_btn_state(self, paused): + self.calls.append(paused) + + core = self.PauseProbe() + core.thread_lock = threading.Lock() + core.running = True + core.paused = False + core.stop_event = Event() + core.pause_event = Event(True) + core.log = lambda *args: None + core.app = App() + core.resume = self.ResumeProbe.resume.__get__(core, type(core)) + + core.pause() + self.assertTrue(core.paused) + self.assertFalse(core.pause_event.is_set()) + + core.resume() + self.assertFalse(core.paused) + self.assertTrue(core.pause_event.is_set()) + self.assertEqual(core.app.calls, [True, False]) + def test_parallel_stat_updates_do_not_lose_counts(self): import threading