Actually interrupt blocking activities

Contributes to #185

Signed-off-by: Sylvain Hellegouarch <sh@defuze.org>

Author: Lawouach

CHANGELOG.md

@@ -2,7 +2,22 @@
 
 ## [Unreleased][]
 
-[Unreleased]: https://github.com/chaostoolkit/chaostoolkit-lib/compare/1.14.1...HEAD
+[Unreleased]: https://github.com/chaostoolkit/chaostoolkit-lib/compare/1.14.2...HEAD
+
+### Added
+
+- Raise the `chaoslib.execeptions.InterruptExecution` on `SIGTERM`
+- New exception `chaoslib.execeptions.ExperimentExitedException` that can only
+  be injected into blocking background activities when we received the SIGUSR2
+  signal
+- We now inject `chaoslib.execeptions.ExperimentExitedException` into blocking
+  background activities when we received the SIGUSR2 signal. This is relying
+  on https://docs.python.org/3/c-api/init.html#c.PyThreadState_SetAsyncExc
+  There is much we can to interrupt blocking calls and the limit is now reached
+  because we have no control over any call that is out of the Python VM (just
+  calling `time.sleep()` will get you in that situation). This is a constraint
+  we have to live in and this extension authors must keep this in mind when
+  they create their blocking calls. [#185][185]
 
 ## [1.14.1][] - 2020-09-10
 

chaoslib/exceptions.py

@@ -44,3 +44,16 @@ class InterruptExecution(ChaosException):
 
 class InvalidControl(ChaosException):
     pass
+
+
+class ExperimentExitedException(ChaosException):
+    """
+    Only raised when the process received a SIGUSR2 signal.
+
+    Raised into the blocking background activities of the method only.
+
+    If you catch it, this mean you can clean your activity but you should
+    really raise another exception again to let the Chaos Toolkit
+    quickly terminate.
+    """
+    pass

chaoslib/exit.py

@@ -40,6 +40,7 @@ def my_probe():
 
 WARNING: Only available on Unix/Linux systems.
 """
+from chaoslib.exceptions import InterruptExecution
 from contextlib import contextmanager
 import inspect
 import os
@@ -158,3 +159,4 @@ def _terminate_now(signum: int, frame: FrameType = None) -> None:
     """
     if signum == signal.SIGTERM:
         logger.warning("Caught SIGTERM signal, interrupting experiment now")
+        raise InterruptExecution("SIGTERM signal received")

chaoslib/run.py

@@ -1,5 +1,10 @@
 # -*- coding: utf-8 -*-
 from concurrent.futures import Future, ThreadPoolExecutor, TimeoutError
+try:
+    import ctypes
+    HAS_CTYPES = True
+except ImportError:
+    HAS_CTYPES = False
 from datetime import datetime
 import platform
 import threading
@@ -12,7 +17,8 @@
 from chaoslib.activity import run_activities
 from chaoslib.control import initialize_controls, controls, cleanup_controls, \
     Control, initialize_global_controls, cleanup_global_controls
-from chaoslib.exceptions import ChaosException, InterruptExecution
+from chaoslib.exceptions import ChaosException, ExperimentExitedException, \
+    InterruptExecution
 from chaoslib.exit import exit_signals
 from chaoslib.configuration import load_configuration
 from chaoslib.hypothesis import run_steady_state_hypothesis
@@ -754,10 +760,11 @@ def apply_activities(experiment: Experiment, configuration: Configuration,
                 logger.debug("Waiting for background activities to complete")
                 pool.shutdown(wait=True)
             elif pool:
+                harshly_terminate_pending_background_activities(pool)
                 logger.debug(
                     "Do not wait for the background activities to finish "
                     "as per signal")
-                background_activity_timeout = 0.1
+                background_activity_timeout = 0.2
                 pool.shutdown(wait=False)
 
             for index, run in enumerate(runs):
@@ -828,3 +835,39 @@ def has_steady_state_hypothesis_with_probes(experiment: Experiment) -> bool:
         if probes:
             return len(probes) > 0
     return False
+
+
+def harshly_terminate_pending_background_activities(
+        pool: ThreadPoolExecutor) -> None:
+    """
+    Ugly hack to try to force background activities to terminate now.
+
+    This cano only have an impact over functions that are still in the Python
+    land. Any code outside of the Python VM (say calling a C function, even
+    time.sleep()) will not be impacted and therefore will continue hanging
+    until it does complete of its own accord.
+
+    This could have really bizarre side effects so it's only applied when
+    a SIGUSR2 signal was received.
+    """
+    if not HAS_CTYPES:
+        logger.debug(
+            "Your Python implementation does not provide the `ctypes` "
+            "module and we cannot terminate very harshly running background "
+            "activities.")
+        return
+
+    logger.debug(
+        "Harshly trying to interrupt remaining background activities still "
+        "running")
+
+    # oh and of course we use private properties... might as well when trying
+    # to be ugly
+    for thread in pool._threads:
+        tid = ctypes.c_long(thread.ident)
+        try:
+            gil = ctypes.pythonapi.PyGILState_Ensure()
+            ctypes.pythonapi.PyThreadState_SetAsyncExc(
+                tid, ctypes.py_object(ExperimentExitedException))
+        finally:
+            ctypes.pythonapi.PyGILState_Release(gil)

tests/fixtures/experiments.py

@@ -969,3 +969,37 @@
         }
     ]
 }
+
+
+SimpleExperimentWithBackgroundActivity = {
+    "version": "1.0.0",
+    "title": "Hello world!",
+    "description": "Say hello world.",
+    "method": [
+        {
+            "type": "action",
+            "name": "pretend-we-do-stuff",
+            "background": True,
+            "provider": {
+                "type": "python",
+                "module": "fixtures.longpythonfunc",
+                "func": "be_long",
+                "arguments": {
+                    "howlong": 3
+                }
+            }
+        },
+        {
+            "type": "action",
+            "name": "pretend-we-do-stuff-again",
+            "provider": {
+                "type": "python",
+                "module": "fixtures.longpythonfunc",
+                "func": "be_long",
+                "arguments": {
+                    "howlong": 4
+                }
+            }
+        }
+    ]
+}

tests/test_exit.py

@@ -3,6 +3,7 @@
 import time
 from wsgiref.simple_server import WSGIServer, WSGIRequestHandler
 
+from chaoslib.exit import exit_gracefully, exit_ungracefully
 from chaoslib.run import Runner
 from chaoslib.types import Strategy
 
@@ -116,14 +117,29 @@ def test_wait_for_background_activity_on_graceful_exit():
 
 
 def test_do_not_wait_for_background_activity_on_ungraceful_exit():
-    server = threading.Thread(target=run_http_server_in_background)
-    server.start()
+    def _exit_soon():
+        time.sleep(1.5)
+        exit_ungracefully()
+    t = threading.Thread(target=_exit_soon)
 
-    x = deepcopy(experiments.ExperimentUngracefulExitLongHTTPCall)
+    x = deepcopy(experiments.SimpleExperimentWithBackgroundActivity)
     with Runner(Strategy.DEFAULT) as runner:
+        t.start()
         journal = runner.run(x)
-
         assert journal["status"] == "interrupted"
         assert journal["run"][0]["status"] == "failed"
+        assert "ExperimentExitedException" in journal["run"][0]["exception"][-1]
 
-    server.join()
+
+def test_wait_for_background_activity_to_finish_on_graceful_exit():
+    def _exit_soon():
+        time.sleep(1.5)
+        exit_gracefully()
+    t = threading.Thread(target=_exit_soon)
+
+    x = deepcopy(experiments.SimpleExperimentWithBackgroundActivity)
+    with Runner(Strategy.DEFAULT) as runner:
+        t.start()
+        journal = runner.run(x)
+        assert journal["status"] == "interrupted"
+        assert journal["run"][0]["status"] == "succeeded"