Python中子进程交互最佳实践：实现父子进程通信或子进程订阅系统事件以触发数据持久化

Great question! Since you already have a watchdog parent process monitoring system events, trying to have the child process listen to those same events is redundant and (as you found) prone to issues like infinite loops. The cleanest approach here is to set up parent-child inter-process communication (IPC)—letting your watchdog notify the child directly when an error occurs, so the child can trigger its state persistence.

Below are the best Python-based solutions for your Windows environment (given you're using pyautoit and winevt), ranked by suitability:

1. Subprocess Pipes (Optimal for Your Current Workflow) #

This is the best fit because you’re already using subprocess.Popen to launch your job script. We’ll use a dedicated pipe (the child’s stdin) for the parent to send error notifications, and the child will run a background thread to listen for these messages without blocking its main data collection flow.

Parent Process (Watchdog) Code:

import subprocess
from winevt import EventLog

child_process = None
job_script = "your_job_script.py"
src_path = "/path/to/your/data"

def handle_system_error(event):
    print(f"Detected critical system event: {event}")
    # Notify child process to persist state
    try:
        # Send a plaintext trigger message to the child's stdin
        child_process.stdin.write("ERROR_TRIGGERED\n")
        child_process.stdin.flush()  # Ensure message isn't stuck in buffer
        # Wait briefly for the child to finish persistence (adjust timeout as needed)
        child_process.wait(timeout=20)
    except Exception as e:
        print(f"Failed to notify child: {str(e)}")
    # Terminate and restart the child
    child_process.kill()
    restart_child()

def restart_child():
    global child_process
    # Launch child with stdin pipe open for communication
    child_process = subprocess.Popen(
        ["python", job_script, src_path],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        stdin=subprocess.PIPE,
        shell=True,
        text=True  # Use text mode to avoid byte-handling overhead
    )
    print(f"Restarted job process (PID: {child_process.pid})")

# Initialize child process
restart_child()

# Subscribe to system error events
EventLog.Subscribe('System', 'Event/System[Level<=2]', handle_system_error)

Child Process (Job Script) Code:

import pickle
import sys
import threading
import time

# Your global state-tracking object
global_task_state = {
    "current_step": 0,
    "collected_data": []
}

def persist_state():
    """Save the global state to disk using pickle"""
    print("Persisting task state due to system error...")
    with open("task_state.pkl", "wb") as f:
        pickle.dump(global_task_state, f)
    print("State saved successfully")
    # Exit after persistence so parent can restart fresh
    sys.exit(0)

def listen_for_parent_notifications():
    """Background thread to monitor messages from the parent"""
    while True:
        message = sys.stdin.readline().strip()
        if not message:
            # Pipe closed (parent exited), terminate listener
            break
        if message == "ERROR_TRIGGERED":
            persist_state()

def main_data_processing():
    global global_task_state
    # Load existing state if it exists (for resume functionality)
    try:
        with open("task_state.pkl", "rb") as f:
            global_task_state = pickle.load(f)
            print(f"Resumed from saved state at step {global_task_state['current_step']}")
    except FileNotFoundError:
        print("No saved state found, starting fresh")

    # Simulate your data collection workflow
    while True:
        global_task_state["current_step"] += 1
        global_task_state["collected_data"].append(f"batch_{global_task_state['current_step']}_data")
        print(f"Processed step {global_task_state['current_step']}")
        time.sleep(2)

if __name__ == "__main__":
    # Start the notification listener as a daemon thread
    listener_thread = threading.Thread(target=listen_for_parent_notifications, daemon=True)
    listener_thread.start()
    # Run main processing workflow
    main_data_processing()

2. Multiprocessing IPC (Alternative for Python-Only Workflows) #

If you’re open to replacing subprocess with Python’s multiprocessing module (for more Pythonic IPC), you can use a Queue or Pipe to send messages between the watchdog parent and task child. This avoids dealing with raw pipes and is easier to extend for complex messages.

Key benefits:

• Built-in serialization (no need to handle text/byte conversion)
• More robust for Python-to-Python communication
• Supports sending structured data (e.g., error details, event IDs)

The core idea is to pass a Queue object to the child process when starting it, then have the parent put a "persist" message into the queue when an error is detected. The child runs a background thread to listen for queue messages.

Critical Notes for Both Approaches #

• Always flush pipes/queues after sending messages to avoid delays.
• Set a reasonable timeout when waiting for the child to persist state—don’t let the watchdog hang indefinitely.
• Mark listener threads as daemon=True so they exit automatically if the child’s main process terminates.
• Store your pickle file in a dedicated, writable directory to avoid permission issues.

内容的提问来源于stack exchange，提问作者Mahakala108