Sending ^C to Python subprocess objects on Windows

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

1. Let your program run in a different cmd window with the Windows shell command start.
2. Add a CTRL-C request wrapper between your control application and the application which should get the CTRL-C signal. The wrapper will run in the same cmd window as the application which should get the CTRL-C signal.
3. The wrapper will shutdown itself and the program which should get the CTRL-C signal by sending all processes in the cmd window the CTRL_C_EVENT.
4. The control program should be able to request the wrapper to send the CTRL-C signal. This might be implemnted trough IPC means, e.g. sockets.

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.

• http://social.msdn.microsoft.com/Forums/en-US/windowsgeneraldevelopmentissues/thread/dc9586ab-1ee8-41aa-a775-cf4828ac1239/#6589714f-12a7-447e-b214-27372f31ca11
• Can I send a ctrl-C (SIGINT) to an application on Windows?
• Sending SIGINT to a subprocess of python
• http://bugs.python.org/issue9524
• http://ss64.com/nt/start.html
• http://objectmix.com/python/387639-sending-cntrl-c.html#post1443948

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)

New answer:

When you create the process, use the flag CREATE_NEW_PROCESS_GROUP. And then you can send CTRL_BREAK to the child process. The default behavior is the same as CTRL_C, except that it won't affect the calling process.

Old answer:

My solution also involves a wrapper script, but it does not need IPC, so it is far simpler to use.

The wrapper script first detaches itself from any existing console, then attach to the target console, then files the Ctrl-C event.

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)

The initial process must be launched in a separate console so that the Ctrl-C event will not leak. Example

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

where I named the wrapper script as ctrl_c.py.

Try calling the GenerateConsoleCtrlEvent function using ctypes. As you are creating a new process group, the process group ID should be the same as the pid. So, something like

import ctypes

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

should work.

Update: You're right, I missed that part of the detail. Here's a post which suggests a possible solution, though it's a bit kludgy. More details are in this answer.

Here is a fully working example which doesn't need any modification in the target script.

This overrides the sitecustomize module so it might no be suitable for every scenario. However, in this case you could use a *.pth file in site-packages to execute code at the subprocess startup (see https://nedbatchelder.com/blog/201001/running_code_at_python_startup.html).

Edit This works only out of the box for subprocesses in Python. Other processes have to manually call SetConsoleCtrlHandler(NULL, FALSE).

main.py

import os
import signal
import subprocess
import sys
import time


def main():
    env = os.environ.copy()
    env['PYTHONPATH'] = '%s%s%s' % ('custom-site', os.pathsep,
                                    env.get('PYTHONPATH', ''))
    proc = subprocess.Popen(
        [sys.executable, 'sub.py'],
        env=env,
        creationflags=subprocess.CREATE_NEW_PROCESS_GROUP,
        )
    time.sleep(1)
    proc.send_signal(signal.CTRL_C_EVENT)
    proc.wait()


if __name__ == '__main__':
    main()

custom-site\sitecustomize.py

import ctypes
import sys
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)

if not kernel32.SetConsoleCtrlHandler(None, False):
    print('SetConsoleCtrlHandler Error: ', ctypes.get_last_error(),
          file=sys.stderr)

sub.py

import atexit
import time


def cleanup():
    print ('cleanup')

atexit.register(cleanup)


while True:
    time.sleep(1)

I have a single file solution with the following advantages: - No external libraries. (Other than ctypes) - Doesn't require the process to be opened in a specific way.

The solution is adapted from this stack overflow post, but I think it's much more elegant in python.

import os
import signal
import subprocess
import sys
import time

# Terminates a Windows console app sending Ctrl-C
def terminateConsole(processId: int, timeout: int = None) -> bool:
    currentFilePath = os.path.abspath(__file__)
    # Call the below code in a separate process. This is necessary due to the FreeConsole call.
    try:
        code = subprocess.call('{} {} {}'.format(sys.executable, currentFilePath, processId), timeout=timeout)
        if code == 0: return True
    except subprocess.TimeoutExpired:
        pass

    # Backup plan
    subprocess.call('taskkill /F /PID {}'.format(processId))


if __name__ == '__main__':
    pid = int(sys.argv[1])

    import ctypes
    kernel = ctypes.windll.kernel32

    r = kernel.FreeConsole()
    if r == 0: exit(-1)
    r = kernel.AttachConsole(pid)
    if r == 0: exit(-1)
    r = kernel.SetConsoleCtrlHandler(None, True)
    if r == 0: exit(-1)
    r = kernel.GenerateConsoleCtrlEvent(0, 0)
    if r == 0: exit(-1)
    r = kernel.FreeConsole()
    if r == 0: exit(-1)

    # use tasklist to wait while the process is still alive.
    while True:
        time.sleep(1)
        # We pass in stdin as PIPE because there currently is no Console, and stdin is currently invalid.
        searchOutput: bytes = subprocess.check_output('tasklist /FI "PID eq {}"'.format(pid), stdin=subprocess.PIPE)
        if str(pid) not in searchOutput.decode(): break;

    # The following two commands are not needed since we're about to close this script.
    # You can leave them here if you want to do more console operations.
    r = kernel.SetConsoleCtrlHandler(None, False)
    if r == 0: exit(-1)
    r = kernel.AllocConsole()
    if r == 0: exit(-1)

    exit(0)

For those interested in a "quick fix", I've made a console-ctrl package based on Siyuan Ren's answer to make it even easier to use.

Simply run pip install console-ctrl, and in your code:

import console_ctrl
import subprocess

# Start some command IN A SEPARATE CONSOLE
p = subprocess.Popen(['some_command'], creationflags=subprocess.CREATE_NEW_CONSOLE)
# ...

# Stop the target process
console_ctrl.send_ctrl_c(p.pid)

I have been trying this but for some reason ctrl+break works, and ctrl+c does not. So using os.kill(signal.CTRL_C_EVENT, 0) fails, but doing os.kill(signal.CTRL_C_EVENT, 1) works. I am told this has something to do with the create process owner being the only one that can pass a ctrl c? Does that make sense?

To clarify, while running fio manually in a command window it appears to be running as expected. Using the CTRL + BREAK breaks without storing the log as expected and CTRL + C finishes writing to the file also as expected. The problem appears to be in the signal for the CTRL_C_EVENT.

It almost appears to be a bug in Python but may rather be a bug in Windows. Also one other thing, I had a cygwin version running and sending the ctrl+c in python there worked as well, but then again we aren't really running native windows there.

example:

import subprocess, time, signal, sys, os
command = '"C:\\Program Files\\fio\\fio.exe" --rw=randrw --bs=1M --numjobs=8 --iodepth=64 --direct=1 ' \
    '--sync=0 --ioengine=windowsaio --name=test --loops=10000 ' \
    '--size=99901800 --rwmixwrite=100 --do_verify=0 --filename=I\\:\\test ' \
    '--thread --output=C:\\output.txt'
def signal_handler(signal, frame):
  time.sleep(1)
  print 'Ctrl+C received in wrapper.py'

signal.signal(signal.SIGINT, signal_handler)
print 'command Starting'
subprocess.Popen(command)
print 'command started'
time.sleep(15) 
print 'Timeout Completed'
os.kill(signal.CTRL_C_EVENT, 0)

(This was supposed to be a comment under Siyuan Ren's answer but I don't have enough rep so here's a slightly longer version.)

If you don't want to create any helper scripts you can use:

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

# Do something else

subprocess.run([
    sys.executable,
    "-c",
    "import ctypes, 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)",
    str(p.pid)
]) # Send Ctrl-C

But it won't work if you use PyInstaller - sys.executable points to your executable, not the Python interpreter. To solve that issue I've created a tiny utility for Windows: https://github.com/anadius/ctrlc

Now you can send the Ctrl+C event with:

subprocess.run(["ctrlc", str(p.pid)])

In summary:

1. parent code: launch the subprocess with creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
2. child code: enable Ctrl+C with ctypes.WinDLL('kernel32').SetConsoleCtrlHandler(None, False)
3. terminate child process gracefully using proc.send_signal(signal.CTRL_C_EVENT)

Now here is a trick: if you can't call SetConsoleCtrlHandler in the child process code, you can proxy the child process by running another python process on top of it.

import signal
import subprocess
import sys
import time


def launch_subprocess_with_proxy(cmd):
    # create a new process group that can receive Ctrl+C event
    proc = subprocess.Popen([sys.executable, __file__] + list(cmd), creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)

    # do something

    # stop process
    proc.send_signal(signal.CTRL_C_EVENT)


def launch_subprocess():
    import ctypes

    # enable Ctrl+C handling in this process group
    ctypes.WinDLL('kernel32').SetConsoleCtrlHandler(None, False)

    # this process and the child process will all receive the Ctrl+C event
    proc = subprocess.Popen(sys.argv[1:])
    try:
        proc.wait()
    except KeyboardInterrupt:
        # wait for child process to stop or terminate it after 5 seconds
        for _ in range(5):
            if proc.poll() is not None:
                break
            time.sleep(1)
        else:
            proc.terminate()


if __name__ == '__main__':
    launch_subprocess()

It can't be done easily, see this fantastic answer for why: https://stackoverflow.com/a/7980368/30618301

So we trick the app into thinking it runs under a new terminal via winpty and send the CTRL+C as a string '\x03':

import re
import threading
import time
from winpty import PtyProcess

ANSI_ESCAPE_REGEX = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
CTRL_C_STRING = '\x03'
proc:PtyProcess = None

def read_stream():
    global proc
    print(f"READER: Starting.", flush=True)
    while True:
        try:
            line = proc.readline()
            if line is not None and len(line) > 0:
                line_cleaned = ANSI_ESCAPE_REGEX.sub('', line)
                print(f"PING: {line_cleaned.strip()}", flush=True)
        except EOFError:
            print(f"PING EOF. GOOD.")
            break
        except Exception as e:
            # baad
            print(f"PING baaad ending: {e}.")
            break

    print(f"READER: Ended.", flush=True)

proc = PtyProcess.spawn(["ping.exe", "-t", "8.8.8.8"], dimensions=(25, 80))
threading.Thread(target=read_stream, daemon=True).start()

print(f"main: subprocess started, waiting 2s")
time.sleep(2)
print(f"main: sending ctrl+c to subprocess")

proc.write(CTRL_C_STRING)

print(f"main: did it work?")
time.sleep(2)
if proc is not None and proc.isalive():
    print(f"main: NOPE! still running... force terminate")
    proc.terminate(force=True)
else:
    print(f"main: YESSS!")

This works for me under Windows 10:

READER: Starting.
main: subprocess started, waiting 2s
PING: 0;C:\Windows\system32\ping.exe
PING: Pinging 8.8.8.8 with 32 bytes of data:
PING: Reply from 8.8.8.8: bytes=32 time=17ms TTL=117
PING: Reply from 8.8.8.8: bytes=32 time=17ms TTL=117
main: sending ctrl+c to subprocess
main: did it work?
PING:
PING: Ping statistics for 8.8.8.8:
PING: Packets: Sent = 2, Received = 2, Lost = 0 (0% loss),
PING: Approximate round trip times in milli-seconds:
PING: Minimum = 17ms, Maximum = 17ms, Average = 17ms
PING: Control-C
main: YESSS!