I have a test harness (written in Python) that needs to shut down the program under test (written in C) by sending it ^C. On Unix,
proc.send_signal(signal.SIGINT)
works perfectly. On Windows, that throws an error ("signal 2 is not supported" or something like that). I am using Python 2.7 for Windows, so I have the impression that I should be able to do instead
proc.send_signal(signal.CTRL_C_EVENT)
but this doesn't do anything at all. What do I have to do? This is the code that creates the subprocess:
# 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)
I write a simple script to check the subprocess module and I tested it on both Windows and Linux. The script works fine on Windows but not on Linux. The interpreter in python is used in 3 versions on both.
To start a new process, or in other words, a new subprocess in Python, you need to use the Popen function call. It is possible to pass two parameters in the function call. The first parameter is the program you want to start, and the second is the file argument.
Python method popen() opens a pipe to or from command. The return value is an open file object connected to the pipe, which can be read or written depending on whether mode is 'r' (default) or 'w'. The bufsize argument has the same meaning as in open() function.
The subprocess module defines one class, Popen and a few wrapper functions that use that class. The constructor for Popen takes arguments to set up the new process so the parent can communicate with it via pipes. It provides all of the functionality of the other modules and functions it replaces, and more.
There is a solution by using a wrapper (as described in the link Vinay provided) which is started in a new console window with the Windows start command.
Code of the wrapper:
#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)
Code of the program catching 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)
Launch the wrapper like e.g.:
PythonPrompt> import subprocess PythonPrompt> subprocess.Popen("start python wrapper.py", shell=True)
You need to add some IPC code which allows you to control the wrapper firing the os.kill(signal.CTRL_C_EVENT, 0) command. I used sockets for this purpose in my application.
Explanation:
Preinformation
send_signal(CTRL_C_EVENT)
does not work because CTRL_C_EVENT
is only for os.kill
. [REF1] os.kill(CTRL_C_EVENT)
sends the signal to all processes running in the current cmd window [REF2] Popen(..., creationflags=CREATE_NEW_PROCESS_GROUP)
does not work because CTRL_C_EVENT
is ignored for process groups. [REF2] This is a bug in the python documentation [REF3] Implemented solution
Helpful posts were:
I had to remove the http in front of the links because I'm a new user and are not allowed to post more than two links.
Update: IPC based CTRL-C Wrapper
Here you can find a selfwritten python module providing a CTRL-C wrapping including a socket based IPC. The syntax is quite similiar to the subprocess module.
Usage:
>>> 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()
Code
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)
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With