Moves terminate_job to billiard

celery/concurrency/processes/__init__.py

@@ -22,14 +22,6 @@ from celery.concurrency.base import BasePool
 from celery.task import trace
 from billiard.pool import Pool, RUN, CLOSE
 
-if platform.system() == 'Windows':  # pragma: no cover
-    # On Windows os.kill calls TerminateProcess which cannot be
-    # handled by # any process, so this is needed to terminate the task
-    # *and its children* (if any).
-    from ._win import kill_processtree as _kill  # noqa
-else:
-    from os import kill as _kill                 # noqa
-
 #: List of signals to reset when a child process starts.
 WORKER_SIGRESET = frozenset(['SIGTERM',
                              'SIGHUP',
@@ -109,7 +101,7 @@ class TaskPool(BasePool):
             self._pool.close()
 
     def terminate_job(self, pid, signal=None):
-        _kill(pid, signal or _signal.SIGTERM)
+        return self._pool.terminate_job(pid, signal)
 
     def grow(self, n=1):
         return self._pool.grow(n)

celery/concurrency/processes/_win.py

@@ -1,116 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-    celery.concurrency.processes._win
-    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-    Windows utilities to terminate process groups.
-
-"""
-from __future__ import absolute_import
-
-import os
-
-# psutil is painfully slow in win32. So to avoid adding big
-# dependencies like pywin32 a ctypes based solution is preferred
-
-# Code based on the winappdbg project http://winappdbg.sourceforge.net/
-# (BSD License)
-from ctypes import (
-    byref, sizeof, windll,
-    Structure, WinError, POINTER,
-    c_size_t, c_char, c_void_p,
-)
-from ctypes.wintypes import DWORD, LONG
-
-ERROR_NO_MORE_FILES = 18
-INVALID_HANDLE_VALUE = c_void_p(-1).value
-
-
-class PROCESSENTRY32(Structure):
-    _fields_ = [
-        ('dwSize',              DWORD),
-        ('cntUsage',            DWORD),
-        ('th32ProcessID',       DWORD),
-        ('th32DefaultHeapID',   c_size_t),
-        ('th32ModuleID',        DWORD),
-        ('cntThreads',          DWORD),
-        ('th32ParentProcessID', DWORD),
-        ('pcPriClassBase',      LONG),
-        ('dwFlags',             DWORD),
-        ('szExeFile',           c_char * 260),
-    ]
-LPPROCESSENTRY32 = POINTER(PROCESSENTRY32)
-
-
-def CreateToolhelp32Snapshot(dwFlags=2, th32ProcessID=0):
-    hSnapshot = windll.kernel32.CreateToolhelp32Snapshot(dwFlags,
-                                                         th32ProcessID)
-    if hSnapshot == INVALID_HANDLE_VALUE:
-        raise WinError()
-    return hSnapshot
-
-
-def Process32First(hSnapshot, pe=None):
-    return _Process32n(windll.kernel32.Process32First, hSnapshot, pe)
-
-
-def Process32Next(hSnapshot, pe=None):
-    return _Process32n(windll.kernel32.Process32Next, hSnapshot, pe)
-
-
-def _Process32n(fun, hSnapshot, pe=None):
-    if pe is None:
-        pe = PROCESSENTRY32()
-    pe.dwSize = sizeof(PROCESSENTRY32)
-    success = fun(hSnapshot, byref(pe))
-    if not success:
-        if windll.kernel32.GetLastError() == ERROR_NO_MORE_FILES:
-            return
-        raise WinError()
-    return pe
-
-
-def get_all_processes_pids():
-    """Return a dictionary with all processes pids as keys and their
-       parents as value. Ignore processes with no parents.
-    """
-    h = CreateToolhelp32Snapshot()
-    parents = {}
-    pe = Process32First(h)
-    while pe:
-        if pe.th32ParentProcessID:
-            parents[pe.th32ProcessID] = pe.th32ParentProcessID
-        pe = Process32Next(h, pe)
-
-    return parents
-
-
-def get_processtree_pids(pid, include_parent=True):
-    """Return a list with all the pids of a process tree"""
-    parents = get_all_processes_pids()
-    all_pids = parents.keys()
-    pids = set([pid])
-    while 1:
-        pids_new = pids.copy()
-
-        for _pid in all_pids:
-            if parents[_pid] in pids:
-                pids_new.add(_pid)
-
-        if pids_new == pids:
-            break
-
-        pids = pids_new.copy()
-
-    if not include_parent:
-        pids.remove(pid)
-
-    return list(pids)
-
-
-def kill_processtree(pid, signum):
-    """Kill a process and all its descendants"""
-    family_pids = get_processtree_pids(pid)
-
-    for _pid in family_pids:
-        os.kill(_pid, signum)

extra/release/doc4allmods

@@ -7,7 +7,6 @@ SKIP_FILES="celery.__compat__.rst
             celery.bin.rst
             celery.bin.celeryd_detach.rst
             celery.bin.celeryctl.rst
-            celery.concurrency.processes._win.rst
             celery.contrib.rst
             celery.contrib.bundles.rst
             celery.local.rst