如何在WSL终端向运行中的Python脚本发送信号实现任务控制

方案一：基于POSIX信号的控制 #

WSL完全支持POSIX信号机制，可通过自定义信号实现任务的暂停/恢复、停止新任务调度：

• 定义全局状态变量跟踪运行状态：is_paused（控制子进程暂停/恢复）、stop_scheduling（控制是否停止提交新任务）
• 为SIGUSR1和SIGUSR2信号注册处理函数，修改状态变量
• 在任务执行逻辑中检查状态，对运行中的子进程发送SIGSTOP/SIGCONT实现暂停/恢复，在任务提交环节检查是否停止调度

示例代码 #

import signal
import subprocess
from concurrent.futures import ThreadPoolExecutor
import time

is_paused = False
stop_scheduling = False

def toggle_pause(signum, frame):
    global is_paused
    is_paused = not is_paused
    print(f"\n任务已{'暂停' if is_paused else '恢复'}")

def stop_new_tasks(signum, frame):
    global stop_scheduling
    stop_scheduling = True
    print("\n已停止调度新任务，将完成当前运行的所有任务")

# 绑定信号处理函数
signal.signal(signal.SIGUSR1, toggle_pause)
signal.signal(signal.SIGUSR2, stop_new_tasks)

def execute_task(cmd):
    global is_paused
    proc = subprocess.Popen(cmd, shell=True)
    while proc.poll() is None:
        if is_paused:
            # 暂停子进程
            proc.send_signal(signal.SIGSTOP)
            # 等待恢复信号
            while is_paused:
                time.sleep(0.5)
            # 恢复子进程
            proc.send_signal(signal.SIGCONT)
        time.sleep(0.1)
    return proc.returncode

if __name__ == "__main__":
    task_list = ["sleep 5", "sleep 6", "sleep 7", "sleep 8"]
    with ThreadPoolExecutor(max_workers=2) as executor:
        pending_futures = []
        for task in task_list:
            if stop_scheduling:
                print(f"跳过未启动任务: {task}")
                continue
            pending_futures.append(executor.submit(execute_task, task))
            time.sleep(0.5)  # 延迟提交，方便测试停止调度功能
        # 等待所有已提交任务完成
        for future in pending_futures:
            future.result()
    print("所有任务处理完毕")

使用方法 #

1. 启动脚本后，在另一个WSL终端中用ps aux | grep your_script.py找到脚本的进程ID（PID）
2. 发送暂停/恢复信号：kill -USR1 <脚本PID>
3. 发送停止新任务调度信号：kill -USR2 <脚本PID>

方案二：非阻塞读取终端输入 #

通过修改标准输入的属性，实现非阻塞读取终端按键，无需回车即可触发控制指令，避免input()的阻塞问题：

• 使用fcntl模块将stdin设置为非阻塞模式
• 定期检查输入字符，绑定p键切换暂停/恢复，s键停止调度新任务
• 在任务执行和提交逻辑中嵌入输入检查，实时响应指令

示例代码 #

import fcntl
import os
import signal
import subprocess
from concurrent.futures import ThreadPoolExecutor
import time

is_paused = False
stop_scheduling = False

def set_stdin_non_blocking():
    fd = os.stdin.fileno()
    flags = fcntl.fcntl(fd, fcntl.F_GETFL)
    fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)

def check_terminal_input():
    global is_paused, stop_scheduling
    try:
        # 读取一个字符
        char = os.read(0, 1).decode().lower()
        if char == 'p':
            is_paused = not is_paused
            print(f"\n任务已{'暂停' if is_paused else '恢复'}")
        elif char == 's':
            stop_scheduling = True
            print("\n已停止调度新任务，将完成当前运行的所有任务")
    except BlockingIOError:
        # 无输入时抛出该异常，直接忽略
        pass

def execute_task(cmd):
    global is_paused
    proc = subprocess.Popen(cmd, shell=True)
    while proc.poll() is None:
        check_terminal_input()
        if is_paused:
            proc.send_signal(signal.SIGSTOP)
            while is_paused:
                check_terminal_input()
                time.sleep(0.5)
            proc.send_signal(signal.SIGCONT)
        time.sleep(0.1)
    return proc.returncode

if __name__ == "__main__":
    set_stdin_non_blocking()
    task_list = ["sleep 5", "sleep 6", "sleep 7", "sleep 8"]
    with ThreadPoolExecutor(max_workers=2) as executor:
        pending_futures = []
        for task in task_list:
            check_terminal_input()
            if stop_scheduling:
                print(f"跳过未启动任务: {task}")
                continue
            pending_futures.append(executor.submit(execute_task, task))
            time.sleep(0.5)
        for future in pending_futures:
            future.result()
    print("所有任务处理完毕")

使用方法 #

1. 启动脚本前，先执行stty -icanon禁用终端行缓冲（确保按键立即被读取）
2. 运行脚本后，直接在当前终端按p键切换暂停/恢复，按s键停止调度新任务
3. 脚本结束后，执行stty icanon恢复终端默认设置

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