Unable to send the `CTRL_BREAK_EVENT` signal to the process in no console mode on windows (pythonw / pyinstaller)

[siva-kranthi]

Bug report

Bug description:

We are starting a tool process using the subprocess.popen. To stop the tool we are sending ctrl+c signal so that the tool catches this & does the required clean up & report generation. It is working fine with python / pyinstaller with console.

But not working with pythonw / pyinstaller no console. It is failing with below exception

OSError: [WinError 6] The handle is invalid

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "E:\workspace\git\ABATA-Agent\t1.py", line 59, in <module>
    out, err = process.communicate()
  File "C:\Program Files\Python311\Lib\subprocess.py", line 1661, in send_signal
    os.kill(self.pid, signal.CTRL_BREAK_EVENT)
SystemError: <built-in function kill> returned a result with an exception set

Below is the standalone script to reproduce this issue with pythonw

import logging
import shlex
import signal
import subprocess
import time

import psutil

logging.basicConfig(
    filename="pyw_log.txt",
    filemode="a",
    format="%(asctime)s,%(msecs)03d %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s",
    datefmt="%Y-%m-%d:%H:%M:%S",
    level=logging.DEBUG,
)
logger = logging.getLogger(__name__)

logger.debug(f"===================================================================")

def process_tree(parent_pid):
    parent_process = psutil.Process(parent_pid)
    for i, child_process in enumerate(parent_process.children(recursive=True)):
        logger.debug(f"Child PID: {child_process.pid}, cmd: {child_process.cmdline()}")
        print(f"Child PID: {child_process.pid}, cmd: {child_process.cmdline()}")

try:
    startup_info = subprocess.STARTUPINFO()
    startup_info.dwFlags |= subprocess.STARTF_USESHOWWINDOW
    creation_flags = subprocess.CREATE_NEW_PROCESS_GROUP

    cmd_line = "netstat"
    print(cmd_line)
    logger.info(cmd_line)

    cmd_line = shlex.split(cmd_line)
    logger.debug(cmd_line)

    with open("process_log.txt", mode="w", encoding="UTF-8") as file:
        process = subprocess.Popen(
            cmd_line,
            stdout=file,
            stderr=file,
            stdin=subprocess.DEVNULL,
            # shell=True,  # no effect
            encoding="UTF-8",
            startupinfo=startup_info,
            creationflags=creation_flags,
        )
        print(process.pid)
        logger.info(process.pid)

        time.sleep(2)

        process_tree(process.pid)
        process.send_signal(signal.CTRL_BREAK_EVENT)

        out, err = process.communicate()
        logger.debug(f"Ret code: {process.returncode}")
        logger.debug(f"Out: {out}")
        logger.debug(f"Error: {err}")
        print(f"Out: {out}")
except Exception:
    logging.exception("")
    # raise
finally:
    process.kill()  # This one doesn't raise any exception when the process already died
    logger.debug("Process killed/died automatically")

@eryksun

CPython versions tested on:

3.11

Operating systems tested on:

Windows

[eryksun]

To send the console control event CTRL_BREAK_EVENT to a process group, the current process has to be in the console session of the process group. By default, "pythonw.exe" is not attached to any console session because it's intended for either GUI or background processes. You can use one of the following workarounds:

• Allocate and attach to a windowless console session that each child process will inherit automatically.
• Or temporarily attach to each child's console session via WinAPI AttachConsole() and FreeConsole().

Since a process can only attach to one console session at a time, the latter approach is more difficult to manage reliably if the current process has multiple threads that spawn console processes. I prefer to simply allocate a windowless console. If the current process has no console -- e.g. WinAPI GetConsoleCP() fails with ERROR_INVALID_HANDLE (6) -- then allocate and attach to a windowless console via the following steps:

1. Spawn "cmd.exe" with the creation flag CREATE_NO_WINDOW.
2. Attach to the console session via WinAPI AttachConsole(pid), where pid is the ID of the spawned "cmd.exe" process.
3. Terminate the spawned "cmd.exe" process.

Since each child console process will automatically inherit a console session that has no window, you won't need STARTF_USESHOWWINDOW anymore.

[siva-kranthi]

Thanks for the guidance. I tried the way that you suggested, but AttachConsole(pid) is failing with ERROR_INVALID_HANDLE (6). Below is the sample code for the same (Executed same with pythonw). Kindly help me by posting the working sample code.

import ctypes
import logging
import subprocess

logging.basicConfig(
    filename="pyw_log.txt",
    filemode="a",
    format="%(asctime)s,%(msecs)03d %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s",
    datefmt="%Y-%m-%d:%H:%M:%S",
    level=logging.DEBUG,
)
logger = logging.getLogger(__name__)

logger.debug(f"===================================================================")

try:
    GetConsoleCP = ctypes.windll.kernel32.GetConsoleCP()
    logger.debug(f"GetConsoleCP return - {ctypes.windll.kernel32.GetConsoleCP()}")
    print(f"GetConsoleCP return - {ctypes.windll.kernel32.GetConsoleCP()}")

    logger.debug(
        f"Last Error ctypes.windll.kernel32.GetLastError() - {ctypes.windll.kernel32.GetLastError()}"
    )

    # No console attached to it
    if GetConsoleCP == 0:
        with open("cmd_log.txt", mode="w", encoding="UTF-8") as file:
            process_cmd = subprocess.Popen(
                ["cmd.exe"],
                stdout=file,
                stderr=file,
                encoding="UTF-8",
                creationflags=subprocess.CREATE_NO_WINDOW,
            )

            logger.info(f"cmd process ID - {process_cmd.pid}")
            print(f"cmd process ID - {process_cmd.pid}")

        AttachConsole = ctypes.windll.kernel32.AttachConsole(process_cmd.pid)
        logger.debug(f"AttachConsole return - {AttachConsole}")
        print(f"AttachConsole return - {AttachConsole}")

        if AttachConsole == 0:
            logger.debug(
                f"Failed to attach the console. ctypes.windll.kernel32.GetLastError() - {ctypes.windll.kernel32.GetLastError()}"
            )
except Exception:
    logging.exception("")

[eryksun]

It takes some time for the API initialization routine in the shell process to allocate the console session. If the parent process calls AttachConsole() before the console session has been allocated, the call will fail with ERROR_INVALID_HANDLE (6). To work around the initialization delay, simply have the shell write a line to stdout and wait (e.g. echo ready & pause). Once the line has been read, you know that the console session exists. For example:

import ctypes
import subprocess

ERROR_INVALID_HANDLE = 6

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

def have_console():
    if kernel32.GetConsoleCP():
        return True
    error = ctypes.get_last_error()
    if error != ERROR_INVALID_HANDLE:
        raise ctypes.WinError(error)
    return False

def allocate_console(window=False, visible=False):
    if window and visible:
        if not kernel32.AllocConsole():
            raise ctypes.WinError(ctypes.get_last_error())
        return
    flags = 0 if window else subprocess.CREATE_NO_WINDOW
    with subprocess.Popen('echo ready & pause',
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            shell=True, creationflags=flags) as p:
        try:
            p.stdout.readline()
            if not kernel32.AttachConsole(p.pid):
                raise ctypes.WinError(ctypes.get_last_error())
        finally:
            p.kill()

I parameterized the allocate_console() function to support creating a console with or without a window. Unless a window is required for interactive console I/O, it's more efficient to create a headless (i.e. windowless) console session, so that's the default. If window is true, by default a hidden console window is created (due to the use of the Popen argument shell=True), unless visible is also true.

Note that I explicitly instantiated ctypes.WinDLL with use_last_error=True. This enables capturing the thread's last WinAPI error at a low level in the _ctypes extension module, which is more reliable. The last captured error for the current thread is returned by ctypes.get_last_error(). For raising an exception, you can get an OSError exception for a WinAPI error code via ctypes.WinError().