[pypy-svn] r46387 - in pypy/dist/pypy/tool: . test

arigo at codespeak.net arigo at codespeak.net
Fri Sep 7 11:53:46 CEST 2007


Author: arigo
Date: Fri Sep  7 11:53:45 2007
New Revision: 46387

Added:
   pypy/dist/pypy/tool/killsubprocess.py   (contents, props changed)
   pypy/dist/pypy/tool/test/test_killsubprocess.py   (contents, props changed)
Log:
A way to kill subprocess on Unix and Windows, from Python cookbook
entries.  I think this method is reliable against race conditions like
the subprocess dying just at the wrong moment and another process
reusing the pid.


Added: pypy/dist/pypy/tool/killsubprocess.py
==============================================================================
--- (empty file)
+++ pypy/dist/pypy/tool/killsubprocess.py	Fri Sep  7 11:53:45 2007
@@ -0,0 +1,24 @@
+"""Kill a subprocess created by subprocess.Popen().
+The two Windows versions come from the Python cookbook.
+"""
+import sys, os
+
+if sys.platform != "win32":
+    import signal
+    assert hasattr(os, 'kill')
+
+    def killsubprocess(process):
+        if process.poll() is None:
+            os.kill(process.pid, signal.SIGTERM)
+
+else:
+    # on Windows, we need either win32api or ctypes
+    try:
+        import ctypes
+        TerminateProcess = ctypes.windll.kernel32.TerminateProcess
+    except ImportError:
+        from win32api import TerminateProcess
+
+    def killsubprocess(process):
+        if process.poll() is None:
+            TerminateProcess(int(process._handle), -1)

Added: pypy/dist/pypy/tool/test/test_killsubprocess.py
==============================================================================
--- (empty file)
+++ pypy/dist/pypy/tool/test/test_killsubprocess.py	Fri Sep  7 11:53:45 2007
@@ -0,0 +1,34 @@
+import sys, time
+from py.compat import subprocess
+from pypy.tool.killsubprocess import killsubprocess
+
+def waitdead(process):
+    for i in range(50):
+        time.sleep(0.1)
+        if process.poll() is not None:
+            break       # ok
+    else:
+        raise AssertionError("the subprocess did not die within 5 seconds")
+
+def test_killsubprocess():
+    popen = subprocess.Popen([sys.executable, '-c', 'raw_input()'],
+                             stdin=subprocess.PIPE)
+    time.sleep(0.9)
+    assert popen.poll() is None
+    assert popen.poll() is None
+    killsubprocess(popen)
+    waitdead(popen)
+
+def test_already_dead_but_no_poll():
+    popen = subprocess.Popen([sys.executable, '-c', 'pass'],
+                             stdin=subprocess.PIPE)
+    time.sleep(3)    # a safe margin to be sure the subprocess is already dead
+    killsubprocess(popen)
+    assert popen.poll() is not None
+
+def test_already_dead_and_polled():
+    popen = subprocess.Popen([sys.executable, '-c', 'pass'],
+                             stdin=subprocess.PIPE)
+    waitdead(popen)
+    killsubprocess(popen)
+    assert popen.poll() is not None



More information about the Pypy-commit mailing list