Commits

mattip committed fa73c4c

add os.kill

  • Participants
  • Parent commits d919340
  • Branches win32-kill

Comments (0)

Files changed (4)

pypy/rlib/rwin32.py

 from pypy.rlib.rarithmetic import intmask
 from pypy.rlib.rposix import validate_fd
 from pypy.rlib import jit
-import os, sys, errno
+import os, sys, errno, signal
 
 # This module can be imported on any platform,
 # but most symbols are not usable...
         return handle
     TerminateProcess = winexternal(
         'TerminateProcess', [HANDLE, rffi.UINT], BOOL)
+    GenerateConsoleCtrlEvent = winexternal(
+        'GenerateConsoleCtrlEvent', [DWORD, DWORD], BOOL)
     _GetCurrentProcessId = winexternal(
         'GetCurrentProcessId', [], DWORD)
     def GetCurrentProcessId():
         return rffi.cast(lltype.Signed, _GetCurrentProcessId())
+    def os_kill(pid, sig):
+        if sig == signal.CTRL_C_EVENT or sig == signal.CTRL_BREAK_EVENT:
+            if 0 == GenerateConsoleCtrlEvent(sig, pid):
+                raise lastWindowsError('os_kill failed generating event')
+            return 0
+        handle = OpenProcess(PROCESS_ALL_ACCESS, False, pid)
+        if handle == NULL_HANDLE:
+            raise lastWindowsError('os_kill failed opening process')
+        return TerminateProcess(handle, sig)
+        

pypy/rlib/test/test_rwin32.py

     assert pid != 0
     handle = rwin32.OpenProcess(rwin32.PROCESS_QUERY_INFORMATION, False, pid)
     rwin32.CloseHandle(handle)
+    raises(WindowsError, rwin32.OpenProcess, rwin32.PROCESS_TERMINATE, False, 0)
+
+def test_terminate_process():
+    import subprocess, signal, sys
+    proc = subprocess.Popen([sys.executable, "-c",
+                         "import time;"
+                         "time.sleep(10)",
+                         ],
+                        ) 
+    print proc.pid
+    handle = rwin32.OpenProcess(rwin32.PROCESS_ALL_ACCESS, False, proc.pid)
+    assert rwin32.TerminateProcess(handle, signal.SIGTERM) == 0
+    rwin32.CloseHandle(handle)
+    assert proc.wait() == signal.SIGTERM
+ 

pypy/rpython/module/ll_os.py

 from pypy.rpython.lltypesystem import lltype
 from pypy.rpython.tool import rffi_platform as platform
 from pypy.rlib import rposix
+from pypy.rlib import rwin32
 from pypy.tool.udir import udir
 from pypy.translator.tool.cbuild import ExternalCompilationInfo
 from pypy.rpython.lltypesystem.rstr import mallocstr
         return extdef([int], int, llimpl=umask_llimpl,
                       export_name="ll_os.ll_os_umask")
 
-    @registering_if(os, 'kill', sys.platform != 'win32')
+    @registering(os.kill)
     def register_os_kill(self):
-        os_kill = self.llexternal('kill', [rffi.PID_T, rffi.INT],
+        if sys.platform == 'win32':
+            kill_llimpl = rwin32.os_kill
+        else:
+            os_kill = self.llexternal('kill', [rffi.PID_T, rffi.INT],
                                   rffi.INT)
-
-        def kill_llimpl(pid, sig):
-            res = rffi.cast(lltype.Signed, os_kill(rffi.cast(rffi.PID_T, pid),
-                                                   rffi.cast(rffi.INT, sig)))
-            if res < 0:
-                raise OSError(rposix.get_errno(), "os_kill failed")
-
+            def kill_llimpl(pid, sig):
+                res = rffi.cast(lltype.Signed, 
+                             os_kill(rffi.cast(rffi.PID_T, pid),
+                                               rffi.cast(rffi.INT, sig)))
+                if res < 0:
+                    raise OSError(rposix.get_errno(), "os_kill failed")
         return extdef([int, int], s_None, llimpl=kill_llimpl,
                       export_name="ll_os.ll_os_kill")
 

pypy/rpython/module/test/test_ll_os.py

 
 
 def test_os_kill():
-    try:
-        f = getllimpl(os.kill)
-    except:
-        skip('No kill in os')
+    f = getllimpl(os.kill)
     import subprocess
     import signal
     proc = subprocess.Popen([sys.executable, "-c",