Repository: celery/celery
Pull request: Terminate job implementation for eventlet concurrency backend
 # Terminate job implementation for eventlet concurrency backend

## Description

Current implementation of **TaskPool** which is based on **eventlet** library does not support terminating jobs. 
Revoking with `terminate=True` or terminating task raises `NotImplementedError: <class 'celery.concurrency.eventlet.TaskPool'> does not implement kill_job` exception while task continues to run. 

Fixes #5209 #3429 
Fixes only eventlet backend for this issue #4019 


Problem statement:
I can‘t stop a task by its task_id
[2018-12-02 23:53:58,955: INFO/MainProcess] Received task: tasks.add[bb1fe102-c1f9-4361-9370-1129900c0d52]
[2018-12-02 23:54:02,479: INFO/MainProcess] Terminating bb1fe102-c1f9-4361-9370-1129900c0d52 (Signals.SIGTERM)
[2018-12-02 23:54:02,490: ERROR/MainProcess] pidbox command error: NotImplementedError("<class 'celery.concurrency.eventlet.TaskPool'> does not implement kill_job",)
Traceback (most recent call last):
  File "d:\envs\aidcs\lib\site-packages\kombu\pidbox.py", line 101, in dispatch
    reply = handle(method, arguments)
  File "d:\envs\aidcs\lib\site-packages\kombu\pidbox.py", line 122, in handle_cast
    return self.handle(method, arguments)
  File "d:\envs\aidcs\lib\site-packages\kombu\pidbox.py", line 116, in handle
    return self.handlers[method](self.state, **arguments)
  File "d:\envs\aidcs\lib\site-packages\celery\worker\control.py", line 163, in revoke
    request.terminate(state.consumer.pool, signal=signum)
  File "d:\envs\aidcs\lib\site-packages\celery\worker\request.py", line 249, in terminate
    pool.terminate_job(self.worker_pid, signal)
  File "d:\envs\aidcs\lib\site-packages\celery\concurrency\base.py", line 115, in terminate_job
    '{0} does not implement kill_job'.format(type(self)))
NotImplementedError: <class 'celery.concurrency.eventlet.TaskPool'> does not implement kill_job
[2018-12-02 23:55:38,956: INFO/MainProcess] Task tasks.add[bb1fe102-c1f9-4361-9370-1129900c0d52] succeeded in 100.0s: 8




this is my main code:

from celery.app.control import Control
from tasks import add, app

myControl=Control(app)
myControl.revoke(task_id="b11729b0-6272-4527-af9d-dc24c0ad492d", terminate=True)



finally，if i want to look at the state of the task only by task_id (just like above), how .

Hints from issue discussion:
Even though I use different concurrency (prefork, using Redis as broker), I've seen the same problem - sometimes worker can't kill some tasks...
I used the  `signal='SIGKILL'` param and got `Task handler raised error: Terminated(9,)`.

Does that mean the task termination succeed?

```python
celery_app.control.revoke(celery_task_id, terminate=True, signal='SIGKILL')
```

celery log:

```
[2018-12-05 16:07:32,872: DEBUG/MainProcess] pidbox received method revoke(signal='SIGKILL', task_id='970f8054-b4ee-4f70-9814-4899852342bd', terminate=True) [reply_to:None ticket:None]
[2018-12-05 16:07:32,873: INFO/MainProcess] Terminating 970f8054-b4ee-4f70-9814-4899852342bd (Signals.SIGKILL)
[2018-12-05 16:07:33,068: ERROR/MainProcess] Task handler raised error: Terminated(9,)
```
The trackback is showing that the taskpool component you used in celery is implement based on eventlet which has not implemented terminate_job function yet. But the default setting is prefork taskpool which  and is the only one who has implemented that function.

And if you want to look at the state of the task only by task_id, you can use : 
status = AsyncResult( id).status.The query is blocking which sent a request to your backend line database to get the status.

Commit message: [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Patch to review:
diff --git a/celery/concurrency/eventlet.py b/celery/concurrency/eventlet.py
index c6bb3415f69..90c39e94e2b 100644
--- a/celery/concurrency/eventlet.py
+++ b/celery/concurrency/eventlet.py
@@ -2,6 +2,7 @@
 import sys
 from time import monotonic
 
+from greenlet import GreenletExit
 from kombu.asynchronous import timer as _timer
 
 from celery import signals
@@ -93,6 +94,7 @@ class TaskPool(base.BasePool):
     is_green = True
     task_join_will_block = False
     _pool = None
+    _pool_map = None
     _quick_put = None
 
     def __init__(self, *args, **kwargs):
@@ -107,8 +109,9 @@ def __init__(self, *args, **kwargs):
 
     def on_start(self):
         self._pool = self.Pool(self.limit)
+        self._pool_map = dict()
         signals.eventlet_pool_started.send(sender=self)
-        self._quick_put = self._pool.spawn_n
+        self._quick_put = self._pool.spawn
         self._quick_apply_sig = signals.eventlet_pool_apply.send
 
     def on_stop(self):
@@ -119,12 +122,17 @@ def on_stop(self):
 
     def on_apply(self, target, args=None, kwargs=None, callback=None,
                  accept_callback=None, **_):
-        self._quick_apply_sig(
-            sender=self, target=target, args=args, kwargs=kwargs,
+        target = TaskPool._make_killable_target(target)
+        self._quick_apply_sig(sender=self, target=target, args=args, kwargs=kwargs,)
+        gt = self._quick_put(
+            apply_target,
+            target, args,
+            kwargs,
+            callback,
+            accept_callback,
+            self.getpid
         )
-        self._quick_put(apply_target, target, args, kwargs,
-                        callback, accept_callback,
-                        self.getpid)
+        self._add_to_pool_map(id(gt), gt)
 
     def grow(self, n=1):
         limit = self.limit + n
@@ -132,10 +140,16 @@ def grow(self, n=1):
         self.limit = limit
 
     def shrink(self, n=1):
-        limit = self.limit - n
+        limit = self.limit-n
         self._pool.resize(limit)
         self.limit = limit
 
+    def terminate_job(self, pid, signal=None):
+        if pid in self._pool_map.keys():
+            gt = self._pool_map[pid]
+            gt.kill()
+            gt.wait()
+
     def _get_info(self):
         info = super()._get_info()
         info.update({
@@ -144,3 +158,26 @@ def _get_info(self):
             'running-threads': self._pool.running(),
         })
         return info
+
+    @staticmethod
+    def _make_killable_target(target):
+        def killable_target(*args, **kwargs):
+            try:
+                return target(*args, **kwargs)
+            except GreenletExit:
+                return (False, None, None)
+            except Exception:
+                raise
+        return killable_target
+
+    def _add_to_pool_map(self, pid, green_thread):
+        self._pool_map[pid] = green_thread
+        green_thread.link(
+            TaskPool._cleanup_after_job_finish,
+            self._pool_map,
+            pid
+        )
+
+    @staticmethod
+    def _cleanup_after_job_finish(gt, pool_map, pid):
+        pool_map.pop(pid)
diff --git a/t/unit/concurrency/test_eventlet.py b/t/unit/concurrency/test_eventlet.py
index dcd803e5342..f8ce3ac0c44 100644
--- a/t/unit/concurrency/test_eventlet.py
+++ b/t/unit/concurrency/test_eventlet.py
@@ -2,6 +2,7 @@
 from unittest.mock import Mock, patch
 
 import pytest
+from greenlet import GreenletExit
 
 import t.skip
 from celery.concurrency.eventlet import TaskPool, Timer, apply_target
@@ -101,6 +102,7 @@ def test_pool(self):
         x.on_apply(Mock())
         x._pool = None
         x.on_stop()
+        assert len(x._pool_map.keys()) == 1
         assert x.getpid()
 
     @patch('celery.concurrency.eventlet.base')
@@ -130,3 +132,37 @@ def test_get_info(self):
             'free-threads': x._pool.free(),
             'running-threads': x._pool.running(),
         }
+
+    def test_terminate_job(self):
+        func = Mock()
+        pool = TaskPool(10)
+        pool.on_start()
+        pool.on_apply(func)
+
+        assert len(pool._pool_map.keys()) == 1
+        pid = list(pool._pool_map.keys())[0]
+        task = pool._pool_map[pid]
+
+        pool.terminate_job(pid)
+        task.link.assert_called_once()
+        task.kill.assert_called_once()
+
+    def test_make_killable_target(self):
+        def valid_target():
+            return "some result..."
+
+        def terminating_target():
+            raise GreenletExit()
+
+        def failing_target():
+            raise RuntimeError("some error")
+
+        assert TaskPool._make_killable_target(valid_target)() == "some result..."
+        assert TaskPool._make_killable_target(terminating_target)() == (False, None, None)
+        with pytest.raises(RuntimeError):
+            TaskPool._make_killable_target(failing_target)()
+
+    def test_cleanup_after_job_finish(self):
+        testMap = {'1': None}
+        TaskPool._cleanup_after_job_finish(None, testMap, '1')
+        assert len(testMap) == 0


Code review comments:
diff_hunk: @@ -107,8 +109,9 @@ def __init__(self, *args, **kwargs):
 
     def on_start(self):
         self._pool = self.Pool(self.limit)
+        self._pool_map = dict()
path: celery/concurrency/eventlet.py
text: @user1:
please try to use {} for dicts