Implemented draft of new background task system

openflexure_microscope/api/v1/blueprints/task.py

@@ -1,8 +1,10 @@
-from openflexure_microscope.api.v1.views import MicroscopeView
+from openflexure_microscope.api.v1.views import MethodView
 from flask import jsonify, abort, Blueprint
 
+from openflexure_microscope.common import tasks
 
-class TaskListAPI(MicroscopeView):
+
+class TaskListAPI(MethodView):
     def get(self):
         """
         Get list of long-running tasks.
@@ -43,7 +45,7 @@ class TaskListAPI(MicroscopeView):
         :>header Content-Type: application/json
         """
 
-        data = self.microscope.task.state
+        data = tasks.states()
 
         return jsonify(data)
 
@@ -58,13 +60,13 @@ class TaskListAPI(MicroscopeView):
         :>header Content-Type: application/json
         """
 
-        self.microscope.task.clean()
+        tasks.cleanup_tasks()
-        data = self.microscope.task.state
+        data = tasks.states()
 
         return jsonify(data)
 
 
-class TaskAPI(MicroscopeView):
+class TaskAPI(MethodView):
     def get(self, task_id):
         """
         Get JSON representation of a task
@@ -99,9 +101,9 @@ class TaskAPI(MicroscopeView):
 
         """
 
-        task = self.microscope.task.task_from_id(task_id)
+        try:
-
+            task = tasks.tasks()[task_id]
-        if not task_id:
+        except KeyError:
             return abort(404)  # 404 Not Found
 
         # Return task state
@@ -109,26 +111,23 @@ class TaskAPI(MicroscopeView):
 
     def delete(self, task_id):
         """
-        Remove a particular task (fails if task is currently running).
+        Terminate a particular task.
 
-        .. :quickref: Tasks; Remove a task
+        .. :quickref: Tasks; Terminate a task
 
         :>header Accept: application/json
 
         :>header Content-Type: application/json
         """
 
-        success = self.microscope.task.delete(task_id)
+        try:
+            task = tasks.tasks()[task_id]
+        except KeyError:
+            return abort(404)  # 404 Not Found
 
-        if success:
+        task.terminate()
-            data = {"status": "success", "message": "task successfully deleted"}
-        else:
-            data = {
-                "status": "fail",
-                "message": "task could not be deleted, as it is currently running or does not exist",
-            }
 
-        return jsonify(data)
+        return jsonify(task.state)
 
 
 def construct_blueprint(microscope_obj):
@@ -136,11 +135,11 @@ def construct_blueprint(microscope_obj):
     blueprint = Blueprint("task_blueprint", __name__)
 
     blueprint.add_url_rule(
-        "/", view_func=TaskListAPI.as_view("task_list", microscope=microscope_obj)
+        "/", view_func=TaskListAPI.as_view("task_list")
     )
 
     blueprint.add_url_rule(
-        "/<task_id>/", view_func=TaskAPI.as_view("task", microscope=microscope_obj)
+        "/<task_id>/", view_func=TaskAPI.as_view("task")
     )
 
     return blueprint

openflexure_microscope/common/tasks/__init__.py

@@ -0,0 +1,2 @@
+from .pool import tasks, states, current_task, update_task_progress, cleanup_tasks, remove_task, update_task_data, taskify
+from .thread import ThreadTerminationError

openflexure_microscope/common/tasks/pool.py

@@ -0,0 +1,118 @@
+import threading
+import logging
+from functools import wraps
+
+from .thread import TaskThread
+
+
+class TaskMaster:
+    def __init__(self, *args, **kwargs):
+        self._tasks = []
+
+    @property
+    def tasks(self):
+        """
+        Returns:
+            dict: Dictionary of TaskThread objects. Key is TaskThread ID.
+        """
+        return {t.id: t for t in self._tasks}
+
+    @property
+    def states(self):
+        """
+        Returns:
+            dict: Dictionary of TaskThread.state dictionaries. Key is TaskThread ID.
+        """
+        return {t.id: t.state for t in self._tasks}
+
+    def new(self, f, *args, **kwargs):
+        task = TaskThread(target=f, args=args, kwargs=kwargs)
+        self._tasks.append(task)
+        return task
+
+    def remove(self, task_id):
+        for task in self._tasks:
+            if (task.id == task_id) and not task.isAlive():
+                del task
+
+    def cleanup(self):
+        for task in self._tasks:
+            if not task.isAlive():
+                del task
+
+
+# Task management
+
+def tasks():
+    """
+    Dictionary of tasks in default taskmaster
+    Returns:
+        dict: Dictionary of tasks in default taskmaster
+    """
+    global _default_task_master
+    return _default_task_master.tasks
+
+
+def states():
+    """
+    Dictionary of TaskThread.state dictionaries. Key is TaskThread ID.
+    Returns:
+        dict: Dictionary of task states in default taskmaster
+    """
+    global _default_task_master
+    return _default_task_master.states
+
+def cleanup_tasks():
+    global _default_task_master
+    return _default_task_master.cleanup()
+
+
+def remove_task(task_id: str):
+    global _default_task_master
+    return _default_task_master.remove(task_id)
+
+
+# Operations on the current task
+
+def current_task():
+    current_task_thread = threading.current_thread()
+    if not isinstance(current_task_thread, TaskThread):
+        return None
+    return current_task_thread
+
+
+def update_task_progress(progress: int):
+    if current_task():
+        current_task().update_progress(progress)
+    else:
+        logging.info("Cannot update task progress of __main__ thread. Skipping.")
+
+
+def update_task_data(data: dict):
+    if current_task():
+        current_task().update_data(data)
+    else:
+        logging.info("Cannot update task data of __main__ thread. Skipping.")
+
+
+# Main "taskify" functions
+
+def taskify(f):
+    """
+    A decorator that wraps the passed in function
+    and surpresses exceptions should one occur
+    """
+
+    @wraps(f)
+    def wrapped(*args, **kwargs):
+        task = _default_task_master.new(
+            f, *args, **kwargs
+        )  # Append to parent object's task list
+        task.start()  # Start the function
+        return task
+
+    return wrapped
+
+
+# Create our default, protected, module-level task pool
+_default_task_master = TaskMaster()

openflexure_microscope/common/tasks/thread.py

@@ -0,0 +1,200 @@
+import ctypes
+import datetime
+import logging
+import traceback
+import uuid
+
+import threading
+
+_LOG = logging.getLogger(__name__)
+
+
+class ThreadTerminationError(SystemExit):
+    """Sibling of SystemExit, but specific to thread termination."""
+
+
+class TaskThread(threading.Thread):
+    def __init__(self, target=None, name=None, args=(), kwargs={}, daemon=True):
+        threading.Thread.__init__(
+            self,
+            group=None,
+            target=target,
+            name=name,
+            args=args,
+            kwargs=kwargs,
+            daemon=daemon,
+        )
+
+        # A UUID for the TaskThread (not the same as the threading.Thread ident)
+        self._ID = uuid.uuid4().hex  # Task ID
+
+        # Make _target, _args, and _kwargs available to the subclass
+        self._target = target
+        self._args = args
+        self._kwargs = kwargs
+
+        # Nice string representation of target function
+        self.target_string = f"{self._target}(args={self._args}, kwargs={self._kwargs})"
+
+        # Private state properties
+        self._status: str = "idle"  # Task status
+        self._return_value = None  # Return value
+        self._start_time = None  # Task start time
+        self._end_time = None  # Task end time
+        
+        # Public state properties
+        self.progress: int = None  # Percent progress of the task
+        self.data = {}  # Dictionary of custom data added during the task
+
+        # Stuff for handling termination
+        self._running_lock = (
+            threading.Lock()
+        )  # Lock obtained while self._target is running
+        self._killed = (
+            threading.Event()
+        )  # Event triggered when thread is manually terminated
+
+    @property
+    def id(self):
+        """
+        Return ID of current TaskThread
+        """
+        return self._ID
+
+    @property
+    def state(self):
+        return {
+            "function": self.target_string,
+            "id": self._ID,
+            "status": self._status,
+            "progress": self.progress,
+            "data": self.data,
+            "return": self._return_value,
+            "start_time": self._start_time,
+            "end_time": self._end_time,
+        }
+
+    def update_progress(self, progress: int):
+        # Update progress of the task
+        self.progress = progress
+
+    def update_data(self, data: dict):
+        # Store data to be used before task finishes (eg for real-time plotting)
+        self.data.update(data)
+
+    def _thread_proc(self, f):
+        """
+        Wraps the target function to handle recording `status` and `return` to `state`.
+        Happens inside the task thread.
+        """
+
+        def wrapped(*args, **kwargs):
+            nonlocal self
+
+            self._status = "running"
+            self._start_time = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
+            try:
+                self._return_value = f(*args, **kwargs)
+                self._status = "success"
+            except Exception as e:
+                logging.error(e)
+                logging.error(traceback.format_exc())
+                self._return_value = str(e)
+                self._status = "error"
+            finally:
+                self._end_time = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
+
+        return wrapped
+
+    def run(self):
+        """
+        Overrides default threading.Thread run() method
+        """
+        logging.debug((self._args, self._kwargs))
+        try:
+            with self._running_lock:
+                if self._killed.is_set():
+                    raise ThreadTerminationError()
+                if self._target:
+                    self._thread_proc(self._target)(*self._args, **self._kwargs)
+        finally:
+            # Avoid a refcycle if the thread is running a function with
+            # an argument that has a member that points to the thread.
+            del self._target, self._args, self._kwargs
+
+    def wait(self):
+        """
+        Start waiting for the task to finish before returning
+        """
+        print("Joining thread {}".format(self))
+        self.join()
+        return self._return_value
+
+    def async_raise(self, exc_type):
+        """Raise an exception in this thread."""
+        # Should only be called on a started thread, so raise otherwise.
+        assert self.ident is not None, "Only started threads have thread identifier"
+
+        # If the thread has died we don't want to raise an exception so log.
+        if not self.is_alive():
+            _LOG.debug(
+                "Not raising %s because thread %s (%s) is not alive",
+                exc_type,
+                self.name,
+                self.ident,
+            )
+            return
+
+        result = ctypes.pythonapi.PyThreadState_SetAsyncExc(
+            ctypes.c_long(self.ident), ctypes.py_object(exc_type)
+        )
+        if result == 0 and self.is_alive():
+            # Don't raise an exception an error unnecessarily if the thread is dead.
+            raise ValueError("Thread ID was invalid.", self.ident)
+        elif result > 1:
+            # Something bad happened, call with a NULL exception to undo.
+            ctypes.pythonapi.PyThreadState_SetAsyncExc(self.ident, None)
+            raise RuntimeError(
+                "Error: PyThreadState_SetAsyncExc %s %s (%s) %s"
+                % (exc_type, self.name, self.ident, result)
+            )
+
+    def _is_thread_proc_running(self):
+        """
+        Test if thread funtion (_thread_proc) is running,
+        by attemtping to acquire the lock _thread_proc acquires at runtime.
+        Returns:
+            bool: If _thread_proc is currently running
+        """
+        could_acquire = self._running_lock.acquire(0)
+        if could_acquire:
+            self._running_lock.release()
+            return False
+        return True
+
+    def terminate(self):
+        """
+        Raise ThreadTerminatedException in the context of the given thread,
+        which should cause the thread to exit silently.
+        """
+        _LOG.warning(f"Terminating thread {self}")
+        self._killed.set()
+        if not self.is_alive():
+            logging.debug("Cannot kill thread that is no longer running.")
+            return
+        if not self._is_thread_proc_running():
+            logging.debug(
+                "Thread's _thread_proc function is no longer running, "
+                "will not kill; letting thread exit gracefully."
+            )
+            return
+        self.async_raise(ThreadTerminationError)
+
+        # Wait for the thread for finish closing. If the threaded function has cleanup code in a try-except,
+        # this pause allows it to finish running before the main process can continue.
+        while self._is_thread_proc_running():
+            pass
+
+        # Set state to terminated
+        self._status = "terminated"
+        self.progress = None

openflexure_microscope/devel/__init__.py

@@ -10,6 +10,9 @@ from openflexure_microscope.plugins import MicroscopePlugin
 from openflexure_microscope.api.v1.views import MicroscopeViewPlugin
 from openflexure_microscope.api.utilities import JsonResponse
 
+# Task management
+from openflexure_microscope.common.tasks import current_task, update_task_progress, update_task_data, taskify
+
 # Exceptions
 from openflexure_microscope.exceptions import TaskDeniedException
 

openflexure_microscope/plugins/default/autofocus/api.py

@@ -6,6 +6,7 @@ from openflexure_microscope.devel import (
     JsonResponse,
     request,
     jsonify,
+    taskify
 )
 
 
@@ -23,7 +24,7 @@ class AutofocusAPI(MicroscopeViewPlugin):
         dz = payload.param("dz", default=np.linspace(-300, 300, 7), convert=np.array)
 
         logging.info("Running autofocus...")
-        task = self.microscope.task.start(self.plugin.autofocus, dz)
+        task = taskify(self.plugin.autofocus)(dz)
 
         # return a handle on the autofocus task
         return jsonify(task.state), 202
@@ -40,9 +41,7 @@ class FastAutofocusAPI(MicroscopeViewPlugin):
             backlash = 0
 
         logging.info("Running autofocus...")
-        task = self.microscope.task.start(
+        task = taskify(self.plugin.fast_autofocus)(dz, backlash=backlash)
-            self.plugin.fast_autofocus, dz, backlash=backlash
-        )
 
         # return a handle on the autofocus task
         return jsonify(task.state), 202

openflexure_microscope/plugins/default/camera_calibration/plugin.py

@@ -4,6 +4,7 @@ from openflexure_microscope.devel import (
     JsonResponse,
     request,
     jsonify,
+    taskify
 )
 
 import logging
@@ -14,7 +15,7 @@ from .recalibrate_utils import recalibrate_camera, auto_expose_and_freeze_settin
 class RecalibrateAPIView(MicroscopeViewPlugin):
     def post(self):
         logging.info("Starting microscope recalibration...")
-        task = self.microscope.task.start(self.plugin.recalibrate)
+        task = taskify(self.plugin.recalibrate)
 
         # Return a handle on the autofocus task
         return jsonify(task.state), 202

openflexure_microscope/plugins/default/scan/api.py

@@ -4,6 +4,7 @@ from openflexure_microscope.devel import (
     request,
     jsonify,
     abort,
+    taskify
 )
 
 import logging
@@ -43,8 +44,7 @@ class TileScanAPI(MicroscopeViewPlugin):
         tags = payload.param("tags", default=[], convert=list)
 
         logging.info("Running tile scan...")
-        task = self.microscope.task.start(
+        task = taskify(self.plugin.tile)(
-            self.plugin.tile,
             basename=filename,
             temporary=temporary,
             step_size=step_size,
@@ -59,5 +59,5 @@ class TileScanAPI(MicroscopeViewPlugin):
             tags=tags,
         )
 
-        # return a handle on the autofocus task
+        # return a handle on the scan task
         return jsonify(task.state), 202

openflexure_microscope/plugins/testing/form_example/api.py

@@ -1,11 +1,9 @@
 from openflexure_microscope.devel import (
     MicroscopeViewPlugin,
-    TaskDeniedException,
     JsonResponse,
-    Response,
     request,
-    escape,
     jsonify,
+    taskify
 )
 
 import logging
@@ -56,9 +54,7 @@ class TaskAPI(MicroscopeViewPlugin):
         val_int = payload.param("run_time", default=5, convert=int)
 
         logging.info("Running task...")
-        task = self.microscope.task.start(
+        task = taskify(self.plugin.generate_random_numbers_for_a_while)(val_int)
-            self.plugin.generate_random_numbers_for_a_while, val_int
-        )
 
         # return a handle on the autofocus task
         return jsonify(task.state), 202

openflexure_microscope/task.py

@@ -1,16 +1,10 @@
-from threading import Thread
-import datetime
 import logging
-import traceback
+from openflexure_microscope.common.tasks import tasks, states, taskify, cleanup_tasks, remove_task
-import uuid
-
-from openflexure_microscope.exceptions import TaskDeniedException
-from openflexure_microscope.utilities import entry_by_id
 
 
 class TaskOrchestrator:
     """
-    Class responsible for spawning threaded tasks, and storing their returns.
+    DEPRECATED: Class responsible for spawning threaded tasks, and storing their returns.
     A microscope should contain exactly one instance of `TaskOrchestrator`.
 
     Attributes:
@@ -18,26 +12,24 @@ class TaskOrchestrator:
 
     """
 
-    def __init__(self):
+    @property
-        self.tasks = []
+    def tasks(self):
+        logging.warning("TaskOrchestrator class is deprecated. Please use common.tasks.tasks().values() instead.")
+        return tasks().values()
 
     @property
     def state(self):
         """
         Returns a list of dictionary representations of all tasks in the session.
         """
-        state = {}
+        logging.warning("TaskOrchestrator class is deprecated. Please use common.tasks.tasks() instead.")
-
+        return states()
-        for task in self.tasks:
-            state[task.id] = task.state
-
-        return state
 
     def task_from_id(self, task_id: str):
         """
         Returns a particular task object with the specified `task_id`.
         """
-        return entry_by_id(task_id, self.tasks)
+        return tasks()[task_id]
 
     def start(self, function, *args, **kwargs):
         """
@@ -49,119 +41,20 @@ class TaskOrchestrator:
             function (function): The target function to run in the background
             args, kwargs: Arguments that will be passed to `function` at runtime.
         """
-        # Create a task object
+        logging.warning("TaskOrchestrator class is deprecated. Please use common.tasks.taskify() instead.")
-        task = Task(function, *args, **kwargs)
+        return taskify(function)(*args, **kwargs)
-
-        # If the task isn't allowed to run
-        if not self.start_condition(task):
-            raise TaskDeniedException("Unable to start this task. Aborting.")
-        else:
-            self.tasks.append(task)
-            task.start()
-            return task
 
     def clean(self):
         """
         Remove all inactive tasks from the task array.
         This will remove tasks that either finished or never started.
         """
-        self.tasks = [task for task in self.tasks if task._running]
+        logging.warning("TaskOrchestrator class is deprecated. Please use common.tasks.cleanup_tasks() instead.")
+        cleanup_tasks()
 
-    def delete(self, task_id):
+    def delete(self, task_id: str):
         """
         Delete a given task by ID, only if task is not currently running.
         """
-        success = False
+        logging.warning("TaskOrchestrator class is deprecated. Please use common.tasks.remove_task() instead.")
-
+        remove_task(task_id)
-        for task in self.tasks:
-            if task.id == task_id and not task._running:
-                self.tasks.remove(task)
-                success = True
-
-        return success
-
-    def start_condition(self, task_obj):
-        """
-        Method returning bool describing if a particular task is allowed to run.
-        Currently always allows tasks to start, as locks determine access to hardware.
-        """
-        # return not any([task._running for task in self.tasks])
-        return True
-
-
-class Task:
-    """
-    Class responsible for running a task function in a thread, and handling return or errors.
-    Tasks should be created by an instance of :py:class:`openflexure_microscope.task.TaskOrchestrator`.
-
-    Args:
-        function (function): Function to be called in the task thread.
-
-    Attributes:
-        task (function): Function to be called in the task thread.
-        args: Positional arguments to be passed to the task function
-        kwargs: Keyword arguments to be passed to the task function.
-        _running (bool): If task is currently running
-        id (str): Unique ID for the task
-        state (dict): Dictionary describing the full state of the task.
-            `status` will be one of 'idle'', 'running', 'error', or 'success'.
-            `return` eventually holds the return value of the task function.
-    """
-
-    def __init__(self, function, *args, **kwargs):
-        # The task long-running method
-        self.task = function
-        self.args = args
-        self.kwargs = kwargs
-
-        self._running = False
-
-        self.id = uuid.uuid4().hex
-
-        self.state = {
-            "id": self.id,
-            "status": "idle",
-            "return": None,
-            "start_time": None,
-            "end_time": None,
-        }
-
-    def process(self, f):
-        """
-        Wraps the target function to handle recording `status` and `return` to `state`.
-        """
-
-        def wrapped(*args, **kwargs):
-            self.state["status"] = "running"
-            try:
-                r = f(*args, **kwargs)
-                s = "success"
-            except Exception as e:
-                logging.error(e)
-                logging.error(traceback.format_exc())
-                r = str(e)
-                s = "error"
-            self.state["return"] = r
-            self.state["status"] = s
-
-        return wrapped
-
-    def run(self):
-        """
-        Records the task start and end times to `state`, and runs the task wrapped in `process`.
-        """
-        self.state["start_time"] = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
-        self.process(self.task)(*self.args, **self.kwargs)
-        self._running = False
-        self.state["end_time"] = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
-
-    def start(self):  # Start and draw
-        """
-        Spawns a thread, and starts `run()` in that thread.
-        """
-        # If not already running
-        if not self._running:
-            self._running = True  # Set start command
-            thread = Thread(target=self.run)  # Define thread
-            thread.daemon = True  # Stop this thread when main thread closes
-            thread.start()  # Start thread