Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- proc.send_signal(signal.SIGINT)
- 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)
- import ctypes
- ctypes.windll.kernel32.GenerateConsoleCtrlEvent(0, proc.pid) # 0 => Ctrl-C
- #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)
- #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)
- >>> 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)
- 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)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement