Отправка ^C в объекты подпроцесса Python в Windows
У меня есть тестовый набор (написанный на Python), который должен завершить тестируемую программу (написанный на C), отправив его ^ C. В Unix,
proc.send_signal(signal.SIGINT)
работает отлично. В Windows это выдает ошибку ("сигнал 2 не поддерживается" или что-то в этом роде). Я использую Python 2.7 для Windows, поэтому у меня сложилось впечатление, что я должен иметь возможность сделать вместо
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 ответов
Существует решение с использованием оболочки (как описано в ссылке, предоставленной Vinay), которая запускается в новом окне консоли с помощью команды запуска Windows.
Код обёртки:
#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). Я использовал для этого сокеты в своем приложении.
Объяснение:
Preinformation
send_signal(CTRL_C_EVENT)
не работает, потому чтоCTRL_C_EVENT
только дляos.kill
, [REF1]os.kill(CTRL_C_EVENT)
отправляет сигнал всем процессам, запущенным в текущем окне cmd [REF2]Popen(..., creationflags=CREATE_NEW_PROCESS_GROUP)
не работает, потому чтоCTRL_C_EVENT
игнорируется для групп процессов. [REF2] Это ошибка в документации по питону [REF3]
Внедренное решение
- Пусть ваша программа запускается в другом окне cmd с помощью команды оболочки Windows.
- Добавьте оболочку запроса CTRL-C между вашим управляющим приложением и приложением, которое должно получить сигнал CTRL-C. Оболочка будет работать в том же окне cmd, что и приложение, которое должно получить сигнал CTRL-C.
- Оболочка сама закроет и программу, которая должна получить сигнал CTRL-C, отправив всем процессам в окне cmd CTRL_C_EVENT.
- Управляющая программа должна иметь возможность запрашивать оболочку для отправки сигнала CTRL-C. Это может быть реализовано через средства IPC, например, сокеты.
Полезные посты были:
Мне пришлось удалить http перед ссылками, потому что я новый пользователь и не могу публиковать более двух ссылок.
- http://social.msdn.microsoft.com/Forums/en-US/windowsgeneraldevelopmentissues/thread/dc9586ab-1ee8-41aa-a775-cf4828ac1239/
- Могу ли я отправить Ctrl-C (SIGINT) в приложение в Windows?
- Отправка SIGINT в подпроцесс python
- http://bugs.python.org/issue9524
- http://ss64.com/nt/start.html
- http://objectmix.com/python/387639-sending-cntrl-c.html
Обновление: оболочка CTRL-C на основе IPC
Здесь вы можете найти самописный модуль python, обеспечивающий упаковку CTRL-C, включая IPC на основе сокетов. Синтаксис очень похож на модуль подпроцесса.
Использование:
>>> 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)
Мое решение также включает сценарий оболочки, но он не требует IPC, поэтому его гораздо проще использовать.
Сценарий оболочки сначала отсоединяется от любой существующей консоли, затем присоединяется к целевой консоли, а затем сохраняет событие 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
.
Попробуйте позвонить GenerateConsoleCtrlEvent
использовать функцию ctypes
, Когда вы создаете новую группу процессов, идентификатор группы процессов должен совпадать с идентификатором pid. Итак, что-то вроде
import ctypes
ctypes.windll.kernel32.GenerateConsoleCtrlEvent(0, proc.pid) # 0 => Ctrl-C
должно сработать.
Обновление: Вы правы, я пропустил эту часть детали. Вот пост, который предлагает возможное решение, хотя это немного глупо. Более подробно в этом ответе.
Вот полностью рабочий пример, который не требует каких-либо изменений в целевом скрипте.
Это отменяет sitecustomize
модуль, поэтому он может не подойти для каждого сценария. Однако в этом случае вы можете использовать файл *.pth в пакетах сайта для выполнения кода при запуске подпроцесса (см. https://nedbatchelder.com/blog/201001/running_code_at_python_startup.html).
Редактировать Это работает только из коробки для подпроцессов в Python. Другие процессы должны вызывать вручную 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()
заказ на сайте \ sitecustomize.py
import ctypes
try:
ctypes.windll.kernel32.SetConsoleCtrlHandler(None, False)
except Exception as e:
print('SetConsoleCtrlHandler Error %s' % e)
else:
print('SetConsoleCtrlHandler OK')
sub.py
import atexit
import time
def cleanup():
print ('cleanup')
atexit.register(cleanup)
while True:
time.sleep(1)
Для тех, кто заинтересован в «быстром исправлении», я сделал пакет console-ctrl на основе ответа Сиюань Рена, чтобы упростить его использование.
Просто беги
pip install console-ctrl
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)
У меня есть однофайловое решение со следующими преимуществами: - Отсутствие внешних библиотек. (Кроме ctypes) - не требует, чтобы процесс открывался определенным образом.
Решение адаптировано из этого сообщения о переполнении стека, но я думаю, что в 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)
(Предполагалось, что это будет комментарий под ответом Сиюань Рена, но у меня недостаточно представителей, поэтому вот немного более длинная версия.)
Если вы не хотите создавать какие-либо вспомогательные сценарии, вы можете использовать:
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
Но это не сработает, если вы используете PyInstaller -
sys.executable
указывает на ваш исполняемый файл, а не на интерпретатор Python. Чтобы решить эту проблему, я создал крошечную утилиту для Windows: https://github.com/anadius/ctrlc .
Теперь вы можете отправить событие Ctrl+C с помощью:
subprocess.run(["ctrlc", str(p.pid)])
Я пытался это, но по какой-то причине Ctrl+ Break работает, а Ctrl+ C нет. Итак, используя
os.kill(signal.CTRL_C_EVENT, 0)
не удается, но делаетos.kill(signal.CTRL_C_EVENT, 1)
работает. Мне сказали, что это как-то связано с тем, что владелец процесса создания является единственным, кто может передать ctrl c? Имеет ли это смысл?
Для пояснения: при запуске fio вручную в командном окне он работает, как и ожидалось. Использование разрывов CTRL + BREAK без сохранения журнала, как ожидается, и CTRL + C завершает запись в файл также, как и ожидалось. Проблема, кажется, в сигнале для CTRL_C_EVENT.
Похоже, что это ошибка в Python, но, скорее, ошибка в Windows. Также еще одна вещь, у меня была запущена версия cygwin и там тоже работала отправка ctrl+ c в python, но с другой стороны, мы не запускаем собственные окна там.
пример:
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)