在Windows上向Python子进程发送^C
我有一个测试工具(用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 个回答
试着用 GenerateConsoleCtrlEvent
这个函数,配合 ctypes
来调用。因为你在创建一个新的进程组,所以进程组的ID应该和进程的ID(pid)是一样的。所以,像下面这样写
import ctypes
ctypes.windll.kernel32.GenerateConsoleCtrlEvent(0, proc.pid) # 0 => Ctrl-C
应该是可以的。
更新:你说得对,我之前漏掉了这个细节。这儿有一篇 帖子,里面提到了一种可能的解决办法,虽然有点笨拙。更多的细节可以在 这个回答里找到。
新答案:
当你创建一个进程时,记得使用 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
。
这里有一个解决方案,可以通过使用一个“包装器”(就像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]
实现的解决方案
- 让你的程序在一个不同的命令窗口中运行,使用Windows的shell命令start。
- 在你的控制应用和应该接收CTRL-C信号的应用之间添加一个CTRL-C请求的包装器。这个包装器将在与应该接收CTRL-C信号的应用相同的命令窗口中运行。
- 包装器会通过向命令窗口中的所有进程发送CTRL_C_EVENT来关闭自己和应该接收CTRL-C信号的程序。
- 控制程序应该能够请求包装器发送CTRL-C信号。这可以通过IPC手段实现,例如套接字。
有帮助的帖子包括:
我必须去掉链接前面的http,因为我是新用户,不允许发布超过两个链接。
- http://social.msdn.microsoft.com/Forums/en-US/windowsgeneraldevelopmentissues/thread/dc9586ab-1ee8-41aa-a775-cf4828ac1239/#6589714f-12a7-447e-b214-27372f31ca11
- 我可以向Windows上的应用发送CTRL-C(SIGINT)吗?
- 向Python的子进程发送SIGINT
- http://bugs.python.org/issue9524
- http://ss64.com/nt/start.html
- http://objectmix.com/python/387639-sending-cntrl-c.html#post1443948
更新:基于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)