diff --git a/main.py b/main.py index 4f5024f..aeb261c 100644 --- a/main.py +++ b/main.py @@ -558,7 +558,7 @@ def clear(self): class AutomationCore: def __init__(self, log_callback, app_instance): self.running = False; self.paused = False; self.stop_event = threading.Event(); self.pause_event = threading.Event() - self.log = log_callback; self.app = app_instance; self.project = None; self.runtime_memory = {}; self.io_lock = threading.Lock() + self.log = log_callback; self.app = app_instance; self.project = None; self.runtime_memory = {}; self.memory_lock = threading.RLock(); self.stats_lock = threading.Lock(); self.io_lock = threading.Lock(); self.run_generation = 0 self.active_threads = 0; self.thread_lock = threading.Lock(); self.scaling_ratio = 1.0; self.breakpoints = set() self.max_threads = 50 self.pending_nodes = [] @@ -597,18 +597,21 @@ def start(self, start_node_id=None): if self.running or not self.project: return self.running = True + self.run_generation += 1 + generation = self.run_generation self.paused = False self.stop_event.clear() self.pause_event.set() self.active_threads = 0 self.pending_nodes.clear() - self.runtime_memory = {} - self.context = {'window_rect': None, 'window_handle': 0, 'window_offset': (0, 0)} - self.performance_stats = {'nodes_executed': 0, 'errors': 0, 'start_time': time.time()} - self.log("🚀 引擎启动", "exec") - self.app.iconify() try: - threading.Thread(target=self._run_flow_engine, args=(start_node_id,), daemon=True).start() + with self.memory_lock: + self.runtime_memory.clear() + self.context = {'window_rect': None, 'window_handle': 0, 'window_offset': (0, 0)} + self.performance_stats = {'nodes_executed': 0, 'errors': 0, 'start_time': time.time()} + self.log("🚀 引擎启动", "exec") + self.app.iconify() + threading.Thread(target=self._run_flow_engine, args=(start_node_id, generation), daemon=True).start() except Exception as e: with self.thread_lock: self.running = False @@ -656,7 +659,7 @@ def _check_pause(self, node_id=None): def _get_next_links(self, node_id, port_name='out'): return [l['target'] for l in self.project['links'] if l.get('source') == node_id and l.get('source_port', 'out') == port_name and l.get('target') in self.project['nodes']] - def _run_flow_engine(self, start_node_id=None): + def _run_flow_engine(self, start_node_id=None, generation=None): try: start_nodes = [start_node_id] if start_node_id else [nid for nid, n in self.project['nodes'].items() if n['type'] == 'start'] if not start_nodes: self.log("未找到开始节点", "error"); return @@ -687,14 +690,30 @@ def _run_flow_engine(self, start_node_id=None): if active_threads <= 0: break time.sleep(0.05) - self.running = False - if self.performance_stats['start_time']: - elapsed = time.time() - self.performance_stats['start_time'] - self.log(f"📊 执行统计: {self.performance_stats['nodes_executed']}个节点, {self.performance_stats['errors']}个错误, 耗时{elapsed:.2f}秒", "info") - self.log("🏁 流程结束", "info"); - self.app.highlight_node_safe(None); - self.app.after(0, self.app.deiconify); - self.app.after(100, self.app.reset_ui_state) + with self.thread_lock: + if generation is None: + generation = self.run_generation + current_run = generation == self.run_generation + stats = self._stats_snapshot() if current_run else None + if current_run: + self.running = False + if stats is not None: + if stats['start_time']: + elapsed = time.time() - stats['start_time'] + self.log(f"📊 执行统计: {stats['nodes_executed']}个节点, {stats['errors']}个错误, 耗时{elapsed:.2f}秒", "info") + self.log("🏁 流程结束", "info") + self.app.after(0, lambda: self._finish_run_ui(generation, reset=False)) + self.app.after(100, lambda: self._finish_run_ui(generation, reset=True)) + + def _finish_run_ui(self, generation, reset=False): + with self.thread_lock: + if generation != self.run_generation or self.running: + return + if reset: + self.app.reset_ui_state() + else: + self.app.highlight_node_safe(None) + self.app.deiconify() def _fork_node(self, node_id): with self.thread_lock: @@ -719,11 +738,11 @@ def _process_node_thread(self, node_id): self.app.highlight_node_safe(node_id, 'running'); self.app.select_node_safe(node_id) try: out_port = self._execute_node(node) - self.performance_stats['nodes_executed'] += 1 + self._increment_stat('nodes_executed') except Exception as e: self.log(f"💥 节点[{node_id}]错误: {e}", "error"); traceback.print_exc(); - self.performance_stats['errors'] += 1 + self._increment_stat('errors') out_port = 'fail' if out_port == '__STOP__' or self.stop_event.is_set(): return @@ -755,15 +774,25 @@ def _process_node_thread(self, node_id): self.pause_event.set() self.log(f"无法启动排队节点[{next_id}],已停止剩余任务: {e}", "error") + def _memory_snapshot(self): + with self.memory_lock: + return dict(self.runtime_memory) + + def _increment_stat(self, key): + with self.stats_lock: + self.performance_stats[key] += 1 + + def _stats_snapshot(self): + with self.stats_lock: + return dict(self.performance_stats) + def _replace_variables(self, text): if not isinstance(text, str): return str(text) - try: - for k, v in self.runtime_memory.items(): - val_str = str(v) if v is not None else "" - text = text.replace(f'${{{k}}}', val_str) - text = text.replace(f'${k}', val_str) - text = text.replace(f'{{{k}}}', val_str) - except Exception: pass + for k, v in self._memory_snapshot().items(): + val_str = str(v) if v is not None else "" + text = text.replace(f'${{{k}}}', val_str) + text = text.replace(f'${k}', val_str) + text = text.replace(f'{{{k}}}', val_str) return text def _update_context_rect(self): @@ -816,7 +845,8 @@ def _execute_node(self, node): msg = data.get('msg', '执行到此节点') if "{ALL_VARS}" in msg.upper() or "{所有变量}" in msg or "${ALL_VARS}" in msg.upper() or "${所有变量}" in msg: - vars_list_str = "\n".join([f"• {k} = {v}" for k, v in self.runtime_memory.items()]) if self.runtime_memory else "(目前无任何内存变量)" + memory = self._memory_snapshot() + vars_list_str = "\n".join([f"• {k} = {v}" for k, v in memory.items()]) if memory else "(目前无任何内存变量)" msg = msg.replace("${ALL_VARS}", vars_list_str).replace("${all_vars}", vars_list_str).replace("${所有变量}", vars_list_str).replace("{ALL_VARS}", vars_list_str).replace("{all_vars}", vars_list_str).replace("{所有变量}", vars_list_str) use_sound = bool(data.get('use_sound', False)) @@ -867,15 +897,19 @@ def _execute_node(self, node): return 'fail' if ntype == 'set_var': - if 'batch_vars' in data: [self.runtime_memory.update({i['name']:i['value']}) for i in data['batch_vars'] if i.get('name')] - if data.get('var_name'): self.runtime_memory[data['var_name']] = data.get('var_value', '') + with self.memory_lock: + if 'batch_vars' in data: + self.runtime_memory.update({i['name']: i['value'] for i in data['batch_vars'] if i.get('name')}) + if data.get('var_name'): + self.runtime_memory[data['var_name']] = data.get('var_value', '') return 'out' if ntype == 'clipboard': mode = data.get('clip_mode', 'read') if mode == 'read': if HAS_PYPERCLIP: - self.runtime_memory[data.get('var_name', 'clipboard_data')] = pyperclip.paste() + with self.memory_lock: + self.runtime_memory[data.get('var_name', 'clipboard_data')] = pyperclip.paste() else: self.log("⚠️ 未安装 pyperclip,无法读取剪贴板", "warning") elif mode == 'write': @@ -886,10 +920,12 @@ def _execute_node(self, node): return 'out' if ntype == 'var_switch': - val = str(self.runtime_memory.get(data.get('var_name',''), '')) + with self.memory_lock: + val = str(self.runtime_memory.get(data.get('var_name',''), '')) op, target = data.get('operator', '='), data.get('var_value', '') if data.get('var_name'): return 'yes' if ((val==target) if op=='=' else (val!=target)) else 'no' - vals = [str(self.runtime_memory.get(vn.strip(), '')) for vn in data.get('var_list', '').split(',') if vn.strip()] + with self.memory_lock: + vals = [str(self.runtime_memory.get(vn.strip(), '')) for vn in data.get('var_list', '').split(',') if vn.strip()] for case in data.get('cases', []): if all(v == case.get('value', '') for v in vals): return case.get('id', 'else') return 'else' @@ -1102,11 +1138,11 @@ def _execute_node(self, node): if ntype == 'web': webbrowser.open(data.get('url')); self._smart_wait(2); return 'out' if ntype == 'loop': if data.get('infinite', True): return 'loop' - with self.io_lock: + with self.memory_lock: k = f"loop_{node['id']}"; c = self.runtime_memory.get(k, 0) if c < safe_int(data.get('count', 3)): self.runtime_memory[k] = c + 1; return 'loop' - else: - if k in self.runtime_memory: del self.runtime_memory[k] + else: + self.runtime_memory.pop(k, None) return 'exit' if ntype == 'if_img': self._ensure_window_focus() diff --git a/tests/test_regressions.py b/tests/test_regressions.py index e750885..d6eb280 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -71,6 +71,18 @@ def setUpClass(cls): "AutomationCore", "start", {"threading": threading, "time": time} ) + cls.MemorySnapshotProbe = extract_class_with_method( + "AutomationCore", "_memory_snapshot", + {"threading": threading} + ) + cls.IncrementStatProbe = extract_class_with_method( + "AutomationCore", "_increment_stat", + {"threading": threading} + ) + cls.FinishUiProbe = extract_class_with_method( + "AutomationCore", "_finish_run_ui", + {"threading": threading} + ) cls.ForkProbe = extract_class_with_method( "AutomationCore", "_fork_node", {"threading": threading} @@ -123,6 +135,7 @@ def reset_ui_state(self): engine.log = lambda *args: None engine.thread_lock = threading.Lock() engine.running = False + engine.run_generation = 0 engine.project = {"nodes": {"start": {"type": "start"}}} engine.paused = False engine.stop_event = Event() @@ -130,6 +143,7 @@ def reset_ui_state(self): engine.active_threads = 0 engine.pending_nodes = [] engine.runtime_memory = {} + engine.memory_lock = threading.RLock() engine.context = {} engine.performance_stats = {} engine.app = App() @@ -147,6 +161,109 @@ def reset_ui_state(self): self.assertTrue(engine.running) self.assertEqual(FakeThread.starts, 1) + def test_start_initialization_failure_resets_running_state(self): + import threading + + class Event: + def __init__(self, set_value=False): + self.value = set_value + def clear(self): + self.value = False + def set(self): + self.value = True + def is_set(self): + return self.value + + class App: + def iconify(self): + raise RuntimeError("window unavailable") + def after(self, delay, callback): + callback() + + engine = self.StartProbe() + engine.log = lambda *args: None + engine.thread_lock = threading.Lock() + engine.memory_lock = threading.RLock() + engine.running = False + engine.run_generation = 0 + engine.project = {"nodes": {"start": {"type": "start"}}} + engine.paused = False + engine.stop_event = Event() + engine.pause_event = Event(True) + engine.active_threads = 0 + engine.pending_nodes = [] + engine.runtime_memory = {"old": "value"} + engine.app = App() + engine.app.reset_ui_state = lambda: None + + with self.assertRaisesRegex(RuntimeError, "window unavailable"): + engine.start() + + self.assertFalse(engine.running) + self.assertTrue(engine.stop_event.is_set()) + self.assertTrue(engine.pause_event.is_set()) + self.assertEqual(engine.pending_nodes, []) + + def test_parallel_stat_updates_do_not_lose_counts(self): + import threading + + core = self.IncrementStatProbe() + core.stats_lock = threading.Lock() + core.performance_stats = {"nodes_executed": 0} + workers = 8 + increments = 5000 + threads = [ + threading.Thread( + target=lambda: [ + core._increment_stat("nodes_executed") + for _ in range(increments) + ] + ) + for _ in range(workers) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual(core.performance_stats["nodes_executed"], workers * increments) + + def test_stale_run_cannot_reset_new_run_ui(self): + import threading + + class App: + def __init__(self): + self.calls = [] + def reset_ui_state(self): + self.calls.append("reset") + def highlight_node_safe(self, node_id): + self.calls.append(("highlight", node_id)) + def deiconify(self): + self.calls.append("show") + + core = self.FinishUiProbe() + core.thread_lock = threading.Lock() + core.run_generation = 2 + core.running = False + core.app = App() + + core._finish_run_ui(1, reset=True) + self.assertEqual(core.app.calls, []) + + core._finish_run_ui(2, reset=True) + self.assertEqual(core.app.calls, ["reset"]) + + def test_memory_snapshot_is_consistent_and_detached(self): + import threading + + core = self.MemorySnapshotProbe() + core.memory_lock = threading.RLock() + core.runtime_memory = {"a": 1} + snapshot = core._memory_snapshot() + core.runtime_memory["b"] = 2 + self.assertEqual(snapshot, {"a": 1}) + self.assertEqual(core._memory_snapshot(), {"a": 1, "b": 2}) + def test_fork_queues_node_when_worker_limit_is_reached(self): import threading