Added ActionView class attribute to override thread default_stop_timeout

Author: Joel Collins

src/labthings/actions/thread.py

@@ -20,7 +20,15 @@ class ActionThread(threading.Thread):
     A native thread with extra functionality for tracking progress and thread termination.
     """
 
-    def __init__(self, target=None, name=None, args=None, kwargs=None, daemon=True):
+    def __init__(
+        self,
+        target=None,
+        name=None,
+        args=None,
+        kwargs=None,
+        daemon=True,
+        default_stop_timeout: int = 5,
+    ):
         threading.Thread.__init__(
             self,
             group=None,
@@ -42,6 +50,7 @@ def __init__(self, target=None, name=None, args=None, kwargs=None, daemon=True):
         self.started = threading.Event()
         # Event to track if the user has requested stop
         self.stopping = threading.Event()
+        self.default_stop_timeout = default_stop_timeout
 
         # Make _target, _args, and _kwargs available to the subclass
         self._target = target
@@ -291,15 +300,18 @@ def terminate(self, exception=ActionKilledException):
         self.progress = None
         return True
 
-    def stop(self, timeout=5, exception=ActionKilledException):
+    def stop(self, timeout=None, exception=ActionKilledException):
         """Sets the threads internal stopped event, waits for timeout seconds for the
         thread to stop nicely, then forcefully kills the thread.
 
-        :param timeout: Time to wait before killing thread forecefully. Defaults to 5.
+        :param timeout: Time to wait before killing thread forecefully. Defaults to ``self.default_stop_timeout``
         :type timeout: int
         :param exception:  (Default value = TaskKillException)
 
         """
+        if timeout is None:
+            timeout = self.default_stop_timeout
+
         self.stopping.set()
         timeout_tracker = TimeoutTracker(timeout)
         # While the timeout hasn't expired

src/labthings/default_views/actions.py

@@ -42,7 +42,7 @@ def get(self, task_id):
 
         return ActionSchema().dump(task)
 
-    @use_args({"timeout": fields.Int(missing=5)})
+    @use_args({"timeout": fields.Int()})
     def delete(self, args, task_id):
         """Terminate a running task.
         
@@ -52,7 +52,7 @@ def delete(self, args, task_id):
         :param task_id: 
 
         """
-        timeout = args.get("timeout", 5)
+        timeout = args.get("timeout", None)
         task_dict = current_labthing().actions.to_dict()
 
         if task_id not in task_dict:

src/labthings/views/__init__.py

@@ -162,6 +162,7 @@ class ActionView(View):
 
     # Action handling
     wait_for: int = 1  # Time in seconds to wait before returning the action as pending/running
+    default_stop_timeout: int = None  # Time in seconds to wait for the action thread to end after a stop request before terminating it forcefully
 
     # Internal
     _cls_tags = {"actions"}
@@ -278,6 +279,9 @@ def dispatch_request(self, *args, **kwargs):
         )
         # Make a task out of the views `post` method
         task = pool.spawn(meth, *args, **kwargs)
+        # Optionally override the threads default_stop_timeout
+        if self.default_stop_timeout is not None:
+            task.default_stop_timeout = self.default_stop_timeout
 
         # Keep a copy of the raw, unmarshalled JSON input in the task
         try:

tests/test_views.py

@@ -3,6 +3,7 @@
 from werkzeug.wrappers import Response as ResponseBase
 from flask import make_response
 
+import time
 import json
 
 import pytest
@@ -157,7 +158,7 @@ def get(self):
         assert Index().get_value() == {"json": "body"}
 
 
-def test_action_view_get_responses(app_ctx):
+def test_action_view_get_responses():
     class Index(views.ActionView):
         def post(self):
             return {}
@@ -166,3 +167,25 @@ def post(self):
     assert 201 in responses
     assert "application/json" in responses[201]["content"]
     assert "schema" in responses[201]["content"]["application/json"]
+
+
+def test_action_view_stop(app):
+    class Index(views.ActionView):
+        default_stop_timeout = 0
+
+        def post(self):
+            while True:
+                time.sleep(1)
+
+    app.add_url_rule("/", view_func=Index.as_view("index"))
+    c = app.test_client()
+
+    response = c.post("/")
+    assert response.status_code == 201
+    assert response.json.get("status") == "running"
+    # Assert we only have a single running Action thread
+    assert len(Index._deque) == 1
+    action_thread = Index._deque[0]
+    assert action_thread.default_stop_timeout == 0
+    action_thread.stop()
+    assert action_thread.status == "terminated"