在Windows上向Python子进程发送^C

32 投票
8 回答
35519 浏览
提问于 2025-04-16 23:44

我有一个测试工具(用Python写的),需要通过发送^C来关闭正在测试的程序(用C写的)。在Unix系统上,

proc.send_signal(signal.SIGINT)

这个方法非常有效。但是在Windows上,就会出现错误(提示“信号2不支持”之类的)。我在Windows上使用的是Python 2.7,所以我觉得我应该可以用下面的方式来实现

proc.send_signal(signal.CTRL_C_EVENT)

但这样完全没有效果。我该怎么做呢?这是创建子进程的代码:

# Windows needs an extra argument passed to subprocess.Popen,
# but the constant isn't defined on Unix.
try: kwargs['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP
except AttributeError: pass
proc = subprocess.Popen(argv,
                        stdin=open(os.path.devnull, "r"),
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE,
                        **kwargs)

8 个回答

8

试着用 GenerateConsoleCtrlEvent 这个函数,配合 ctypes 来调用。因为你在创建一个新的进程组,所以进程组的ID应该和进程的ID(pid)是一样的。所以,像下面这样写

import ctypes

ctypes.windll.kernel32.GenerateConsoleCtrlEvent(0, proc.pid) # 0 => Ctrl-C

应该是可以的。

更新:你说得对,我之前漏掉了这个细节。这儿有一篇 帖子,里面提到了一种可能的解决办法,虽然有点笨拙。更多的细节可以在 这个回答里找到。

11

新答案:

当你创建一个进程时,记得使用 CREATE_NEW_PROCESS_GROUP 这个标志。这样你就可以给子进程发送 CTRL_BREAK 信号。默认情况下,这个信号的效果和 CTRL_C 是一样的,只不过它不会影响到调用这个进程的主进程。


旧答案:

我的解决方案也涉及到一个包装脚本,但它不需要进程间通信,所以使用起来简单多了。

这个包装脚本首先会和任何现有的控制台断开连接,然后再连接到目标控制台,接着就会发送 Ctrl-C 事件。

import ctypes
import sys

kernel = ctypes.windll.kernel32

pid = int(sys.argv[1])
kernel.FreeConsole()
kernel.AttachConsole(pid)
kernel.SetConsoleCtrlHandler(None, 1)
kernel.GenerateConsoleCtrlEvent(0, 0)
sys.exit(0)

最初的进程必须在一个单独的控制台中启动,这样 Ctrl-C 事件才不会泄露。示例:

p = subprocess.Popen(['some_command'], creationflags=subprocess.CREATE_NEW_CONSOLE)

# Do something else

subprocess.check_call([sys.executable, 'ctrl_c.py', str(p.pid)]) # Send Ctrl-C

我把这个包装脚本命名为 ctrl_c.py

19

这里有一个解决方案,可以通过使用一个“包装器”(就像Vinay提供的链接中描述的那样),它会在新的控制台窗口中启动,使用Windows的start命令。

包装器的代码:

#wrapper.py
import subprocess, time, signal, sys, os

def signal_handler(signal, frame):
  time.sleep(1)
  print 'Ctrl+C received in wrapper.py'

signal.signal(signal.SIGINT, signal_handler)
print "wrapper.py started"
subprocess.Popen("python demo.py")
time.sleep(3) #Replace with your IPC code here, which waits on a fire CTRL-C request
os.kill(signal.CTRL_C_EVENT, 0)

捕获CTRL-C的程序代码:

#demo.py

import signal, sys, time

def signal_handler(signal, frame):
  print 'Ctrl+C received in demo.py'
  time.sleep(1)
  sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)
print 'demo.py started'
#signal.pause() # does not work under Windows
while(True):
  time.sleep(1)

像这样启动包装器:

PythonPrompt> import subprocess
PythonPrompt> subprocess.Popen("start python wrapper.py", shell=True)

你需要添加一些进程间通信(IPC)的代码,这样你就可以通过执行os.kill(signal.CTRL_C_EVENT, 0)命令来控制包装器。我在我的应用中使用了套接字来实现这个功能。

解释:

前提信息

  • send_signal(CTRL_C_EVENT)不起作用,因为CTRL_C_EVENT只能用于os.kill[参考1]
  • os.kill(CTRL_C_EVENT)会将信号发送到当前命令窗口中运行的所有进程。[参考2]
  • Popen(..., creationflags=CREATE_NEW_PROCESS_GROUP)不起作用,因为CTRL_C_EVENT会被进程组忽略。[参考2]这是Python文档中的一个错误。[参考3]

实现的解决方案

  1. 让你的程序在一个不同的命令窗口中运行,使用Windows的shell命令start
  2. 在你的控制应用和应该接收CTRL-C信号的应用之间添加一个CTRL-C请求的包装器。这个包装器将在与应该接收CTRL-C信号的应用相同的命令窗口中运行。
  3. 包装器会通过向命令窗口中的所有进程发送CTRL_C_EVENT来关闭自己和应该接收CTRL-C信号的程序。
  4. 控制程序应该能够请求包装器发送CTRL-C信号。这可以通过IPC手段实现,例如套接字。

有帮助的帖子包括:

我必须去掉链接前面的http,因为我是新用户,不允许发布超过两个链接。

更新:基于IPC的CTRL-C包装器

这里你可以找到一个自写的Python模块,提供了一个包含基于套接字的IPC的CTRL-C包装。语法与子进程模块非常相似。

用法:

>>> import winctrlc
>>> p1 = winctrlc.Popen("python demo.py")
>>> p2 = winctrlc.Popen("python demo.py")
>>> p3 = winctrlc.Popen("python demo.py")
>>> p2.send_ctrl_c()
>>> p1.send_ctrl_c()
>>> p3.send_ctrl_c()

代码

import socket
import subprocess
import time
import random
import signal, os, sys


class Popen:
  _port = random.randint(10000, 50000)
  _connection = ''

  def _start_ctrl_c_wrapper(self, cmd):
    cmd_str = "start \"\" python winctrlc.py "+"\""+cmd+"\""+" "+str(self._port)
    subprocess.Popen(cmd_str, shell=True)

  def _create_connection(self):
    self._connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self._connection.connect(('localhost', self._port))

  def send_ctrl_c(self):
    self._connection.send(Wrapper.TERMINATION_REQ)
    self._connection.close()

  def __init__(self, cmd):
    self._start_ctrl_c_wrapper(cmd)
    self._create_connection()


class Wrapper:
  TERMINATION_REQ = "Terminate with CTRL-C"

  def _create_connection(self, port):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('localhost', port))
    s.listen(1)
    conn, addr = s.accept()
    return conn

  def _wait_on_ctrl_c_request(self, conn):
    while True:
      data = conn.recv(1024)
      if data == self.TERMINATION_REQ:
        ctrl_c_received = True
        break
      else:
        ctrl_c_received = False
    return ctrl_c_received

  def _cleanup_and_fire_ctrl_c(self, conn):
    conn.close()
    os.kill(signal.CTRL_C_EVENT, 0)

  def _signal_handler(self, signal, frame):
    time.sleep(1)
    sys.exit(0)

  def __init__(self, cmd, port):
    signal.signal(signal.SIGINT, self._signal_handler)
    subprocess.Popen(cmd)
    conn = self._create_connection(port)
    ctrl_c_req_received = self._wait_on_ctrl_c_request(conn)
    if ctrl_c_req_received:
      self._cleanup_and_fire_ctrl_c(conn)
    else:
      sys.exit(0)


if __name__ == "__main__":
  command_string = sys.argv[1]
  port_no = int(sys.argv[2])
  Wrapper(command_string, port_no)

撰写回答