From e4a78046d7b782cafd4f5a75da7d69ee3dfe2726 Mon Sep 17 00:00:00 2001 From: jtc42 Date: Fri, 1 Nov 2019 16:06:49 +0000 Subject: [PATCH] Implemented draft of new background task system --- .../api/v1/blueprints/task.py | 43 ++-- openflexure_microscope/common/__init__.py | 0 .../common/tasks/__init__.py | 2 + openflexure_microscope/common/tasks/pool.py | 118 +++++++++++ openflexure_microscope/common/tasks/thread.py | 200 ++++++++++++++++++ openflexure_microscope/devel/__init__.py | 3 + .../plugins/default/autofocus/api.py | 7 +- .../default/camera_calibration/plugin.py | 3 +- .../plugins/default/scan/api.py | 6 +- .../plugins/testing/form_example/api.py | 8 +- openflexure_microscope/task.py | 139 ++---------- 11 files changed, 370 insertions(+), 159 deletions(-) create mode 100644 openflexure_microscope/common/__init__.py create mode 100644 openflexure_microscope/common/tasks/__init__.py create mode 100644 openflexure_microscope/common/tasks/pool.py create mode 100644 openflexure_microscope/common/tasks/thread.py diff --git a/openflexure_microscope/api/v1/blueprints/task.py b/openflexure_microscope/api/v1/blueprints/task.py index 12f65700..4cc68e90 100644 --- a/openflexure_microscope/api/v1/blueprints/task.py +++ b/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() - data = self.microscope.task.state + tasks.cleanup_tasks() + 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) - - if not task_id: + try: + task = tasks.tasks()[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: - 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", - } + task.terminate() - 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( - "//", view_func=TaskAPI.as_view("task", microscope=microscope_obj) + "//", view_func=TaskAPI.as_view("task") ) return blueprint diff --git a/openflexure_microscope/common/__init__.py b/openflexure_microscope/common/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/openflexure_microscope/common/tasks/__init__.py b/openflexure_microscope/common/tasks/__init__.py new file mode 100644 index 00000000..ce5851df --- /dev/null +++ b/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 diff --git a/openflexure_microscope/common/tasks/pool.py b/openflexure_microscope/common/tasks/pool.py new file mode 100644 index 00000000..efe286be --- /dev/null +++ b/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() diff --git a/openflexure_microscope/common/tasks/thread.py b/openflexure_microscope/common/tasks/thread.py new file mode 100644 index 00000000..b7a3c1bf --- /dev/null +++ b/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 diff --git a/openflexure_microscope/devel/__init__.py b/openflexure_microscope/devel/__init__.py index 3e5f1c92..3ecd85ec 100644 --- a/openflexure_microscope/devel/__init__.py +++ b/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 diff --git a/openflexure_microscope/plugins/default/autofocus/api.py b/openflexure_microscope/plugins/default/autofocus/api.py index 34ff672d..af0c3981 100644 --- a/openflexure_microscope/plugins/default/autofocus/api.py +++ b/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( - self.plugin.fast_autofocus, dz, backlash=backlash - ) + task = taskify(self.plugin.fast_autofocus)(dz, backlash=backlash) # return a handle on the autofocus task return jsonify(task.state), 202 diff --git a/openflexure_microscope/plugins/default/camera_calibration/plugin.py b/openflexure_microscope/plugins/default/camera_calibration/plugin.py index b4ded10a..37281973 100644 --- a/openflexure_microscope/plugins/default/camera_calibration/plugin.py +++ b/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 diff --git a/openflexure_microscope/plugins/default/scan/api.py b/openflexure_microscope/plugins/default/scan/api.py index 7d338a14..3581afb1 100644 --- a/openflexure_microscope/plugins/default/scan/api.py +++ b/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( - self.plugin.tile, + task = taskify(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 diff --git a/openflexure_microscope/plugins/testing/form_example/api.py b/openflexure_microscope/plugins/testing/form_example/api.py index c05b3c4d..f6426ad1 100644 --- a/openflexure_microscope/plugins/testing/form_example/api.py +++ b/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( - self.plugin.generate_random_numbers_for_a_while, val_int - ) + task = taskify(self.plugin.generate_random_numbers_for_a_while)(val_int) # return a handle on the autofocus task return jsonify(task.state), 202 diff --git a/openflexure_microscope/task.py b/openflexure_microscope/task.py index 4e92169d..859b8532 100644 --- a/openflexure_microscope/task.py +++ b/openflexure_microscope/task.py @@ -1,16 +1,10 @@ -from threading import Thread -import datetime import logging -import traceback -import uuid - -from openflexure_microscope.exceptions import TaskDeniedException -from openflexure_microscope.utilities import entry_by_id +from openflexure_microscope.common.tasks import tasks, states, taskify, cleanup_tasks, remove_task 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): - self.tasks = [] + @property + 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 = {} - - for task in self.tasks: - state[task.id] = task.state - - return state + logging.warning("TaskOrchestrator class is deprecated. Please use common.tasks.tasks() instead.") + return states() 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 - task = Task(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 + logging.warning("TaskOrchestrator class is deprecated. Please use common.tasks.taskify() instead.") + return taskify(function)(*args, **kwargs) 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 - - 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 + logging.warning("TaskOrchestrator class is deprecated. Please use common.tasks.remove_task() instead.") + remove_task(task_id)