Refactored tasks.TaskThread to actions.ActionThread

Author: Joel Collins

src/labthings/__init__.py

@@ -19,11 +19,11 @@
 from .sync import CompositeLock
 from .sync import ClientEvent
 
-# Task management functions
-from .tasks import current_task
-from .tasks import update_task_progress
-from .tasks import update_task_data
-from .tasks import TaskKillException
+# Action threads
+from .actions import current_action
+from .actions import update_action_progress
+from .actions import update_action_data
+from .actions import ActionKilledException
 
 # Schema and field
 from .schema import Schema
@@ -47,11 +47,10 @@
     "StrictLock",
     "CompositeLock",
     "ClientEvent",
-    "current_task",
-    "current_task_stopped"
-    "update_task_progress"
-    "update_task_data"
-    "TaskKillException",
+    "current_action",
+    "update_action_progress",
+    "update_action_data",
+    "ActionKilledException",
     "extensions",
     "views",
     "fields",

src/labthings/actions/__init__.py

@@ -0,0 +1,14 @@
+__all__ = [
+    "current_action",
+    "update_action_progress",
+    "update_action_data",
+    "ActionKilledException"
+]
+
+from .pool import (
+    Pool,
+    current_action,
+    update_action_progress,
+    update_action_data,
+)
+from .thread import ActionThread, ActionKilledException

src/labthings/tasks/pool.py src/labthings/actions/pool.pysrc/labthings/tasks/pool.py renamed to src/labthings/actions/pool.py

@@ -2,28 +2,28 @@
 from functools import wraps
 
 import threading
-from .thread import TaskThread
+from .thread import ActionThread
 
 
-# TODO: Handle discarding old tasks. Action views now use deques
+# TODO: Handle discarding old actions. Action views now use deques
 class Pool:
     """ """
 
     def __init__(self):
         self.threads = set()
 
-    def add(self, thread: TaskThread):
+    def add(self, thread: ActionThread):
         """
 
-        :param thread: TaskThread: 
+        :param thread: ActionThread: 
 
         """
         self.threads.add(thread)
 
-    def start(self, thread: TaskThread):
+    def start(self, thread: ActionThread):
         """
 
-        :param thread: TaskThread: 
+        :param thread: ActionThread: 
 
         """
         self.add(thread)
@@ -37,7 +37,7 @@ def spawn(self, function, *args, **kwargs):
         :param **kwargs: 
 
         """
-        thread = TaskThread(target=function, args=args, kwargs=kwargs)
+        thread = ActionThread(target=function, args=args, kwargs=kwargs)
         self.start(thread)
         return thread
 
@@ -55,7 +55,7 @@ def tasks(self):
         """
 
 
-        :returns: List of TaskThread objects.
+        :returns: List of ActionThread objects.
 
         :rtype: list
 
@@ -66,7 +66,7 @@ def states(self):
         """
 
 
-        :returns: Dictionary of TaskThread.state dictionaries. Key is TaskThread ID.
+        :returns: Dictionary of ActionThread.state dictionaries. Key is ActionThread ID.
 
         :rtype: dict
 
@@ -77,7 +77,7 @@ def to_dict(self):
         """
 
 
-        :returns: Dictionary of TaskThread objects. Key is TaskThread ID.
+        :returns: Dictionary of ActionThread objects. Key is ActionThread ID.
 
         :rtype: dict
 
@@ -117,22 +117,22 @@ def join(self):
 # Operations on the current task
 
 
-def current_task():
-    """Return the Task instance in which the caller is currently running.
+def current_action():
+    """Return the ActionThread instance in which the caller is currently running.
     
-    If this function is called from outside a Task thread, it will return None.
+    If this function is called from outside an ActionThread, it will return None.
 
 
-    :returns: TaskThread -- Currently running Task thread.
+    :returns: ActionThread -- Currently running ActionThread.
 
     """
-    current_task_thread = threading.current_thread()
-    if not isinstance(current_task_thread, TaskThread):
+    current_action_thread = threading.current_thread()
+    if not isinstance(current_action_thread, ActionThread):
         return None
-    return current_task_thread
+    return current_action_thread
 
 
-def update_task_progress(progress: int):
+def update_action_progress(progress: int):
     """Update the progress of the Task in which the caller is currently running.
     
     If this function is called from outside a Task thread, it will do nothing.
@@ -141,13 +141,13 @@ def update_task_progress(progress: int):
     :param progress: int: 
 
     """
-    if current_task():
-        current_task().update_progress(progress)
+    if current_action():
+        current_action().update_progress(progress)
     else:
         logging.info("Cannot update task progress of __main__ thread. Skipping.")
 
 
-def update_task_data(data: dict):
+def update_action_data(data: dict):
     """Update the data of the Task in which the caller is currently running.
     
     If this function is called from outside a Task thread, it will do nothing.
@@ -156,7 +156,7 @@ def update_task_data(data: dict):
     :param data: dict: 
 
     """
-    if current_task():
-        current_task().update_data(data)
+    if current_action():
+        current_action().update_data(data)
     else:
         logging.info("Cannot update task data of __main__ thread. Skipping.")

src/labthings/tasks/thread.py src/labthings/actions/thread.pysrc/labthings/tasks/thread.py renamed to src/labthings/actions/thread.py

@@ -11,11 +11,11 @@
 _LOG = logging.getLogger(__name__)
 
 
-class TaskKillException(SystemExit):
+class ActionKilledException(SystemExit):
     """Sibling of SystemExit, but specific to thread termination."""
 
 
-class TaskThread(threading.Thread):
+class ActionThread(threading.Thread):
     """
     A native thread with extra functionality for tracking progress and thread termination.
     """
@@ -35,7 +35,7 @@ def __init__(self, target=None, name=None, args=None, kwargs=None, daemon=True):
         args = args or ()
         kwargs = kwargs or {}
 
-        # A UUID for the TaskThread (not the same as the threading.Thread ident)
+        # A UUID for the ActionThread (not the same as the threading.Thread ident)
         self._ID = uuid.uuid4()  # Task ID
 
         # Event to track if the task has started
@@ -126,7 +126,7 @@ def run(self):
             with self._running_lock:
                 # Don't run if the thread was stopped before starting
                 if self.stopping.is_set():
-                    raise TaskKillException
+                    raise ActionKilledException
                 if self._target:
                     self._thread_proc(self._target)(*self._args, **self._kwargs)
         finally:
@@ -161,7 +161,7 @@ def wrapped(*args, **kwargs):
             try:
                 self._return_value = f(*args, **kwargs)
                 self._status = "success"
-            except (TaskKillException, SystemExit) as e:
+            except (ActionKilledException, SystemExit) as e:
                 logging.error(e)
                 # Set state to terminated
                 self._status = "terminated"
@@ -242,7 +242,7 @@ def _is_thread_proc_running(self):
             return False
         return True
 
-    def terminate(self, exception=TaskKillException):
+    def terminate(self, exception=ActionKilledException):
         """
 
         :param exception:  (Default value = TaskKillException)
@@ -271,7 +271,7 @@ def terminate(self, exception=TaskKillException):
         self.progress = None
         return True
 
-    def stop(self, timeout=5, exception=TaskKillException):
+    def stop(self, timeout=5, exception=ActionKilledException):
         """Sets the threads internal stopped event, waits for timeout seconds for the
         thread to stop nicely, then forcefully kills the thread.
 
@@ -346,3 +346,7 @@ def emit(self, record):
         # We probably need to check the size of the list...
         # TODO: think about whether any of the keys are security flaws
         # (this is why I don't dump the whole logrecord)
+
+
+# Backwards compatibility
+ActionThread = ActionThread

src/labthings/core/tasks/__init__.py

@@ -1 +1,4 @@
-from ...tasks import *
+from ...actions import current_action as current_task
+from ...actions import update_action_progress as update_task_progress
+from ...actions import update_action_data as update_task_data
+from ...actions import ActionKilledException as ThreadTerminationError

src/labthings/core/tasks/pool.py

@@ -1 +1,3 @@
-from ...tasks.pool import *
+from ...actions.pool import current_action as current_task
+from ...actions.pool import update_action_progress as update_task_progress
+from ...actions.pool import update_action_data as update_task_data

src/labthings/core/tasks/thread.py

@@ -1 +1 @@
-from ...tasks.thread import *
+from ...actions.thread import *

src/labthings/default_views/tasks.py

@@ -8,9 +8,7 @@
 
 
 class TaskList(View):
-    """List of all background tasks from the session"""
-
-    tags = ["tasks"]
+    """List of all background actions from the session"""
 
     def get(self):
         """ """
@@ -29,8 +27,6 @@ class TaskView(View):
 
     """
 
-    tags = ["tasks"]
-
     def get(self, task_id):
         """Show status of a session task
         

src/labthings/labthing.py

@@ -22,7 +22,7 @@
 from .td import ThingDescription
 from .event import Event
 
-from .tasks.pool import Pool
+from .actions.pool import Pool
 
 from .views.builder import property_of, action_from
 
@@ -270,8 +270,8 @@ def _create_base_routes(self):
         self.add_view(ExtensionList, "/extensions", endpoint=EXTENSION_LIST_ENDPOINT)
         self.add_root_link(ExtensionList, "extensions")
         # Add task routes
-        self.add_view(TaskList, "/tasks", endpoint=TASK_LIST_ENDPOINT)
-        self.add_view(TaskView, "/tasks/<task_id>", endpoint=TASK_ENDPOINT)
+        self.add_view(TaskList, "/actions", endpoint=TASK_LIST_ENDPOINT)
+        self.add_view(TaskView, "/actions/<task_id>", endpoint=TASK_ENDPOINT)
         # Add action routes
         self.add_view(ActionQueue, "/actions", endpoint=ACTION_LIST_ENDPOINT)
         self.add_root_link(ActionQueue, "actions")

src/labthings/schema.py

@@ -63,7 +63,7 @@ def dump(self, value):
 
 
 class TaskSchema(Schema):
-    """Legacy schema for background tasks. Will eventually be replaced by ActionSchema,"""
+    """Legacy schema for background actions. Will eventually be replaced by ActionSchema,"""
 
     _ID = fields.String(data_key="id")
     target_string = fields.String(data_key="function")