Action threads

Many actions in your LabThing may perform tasks that take a long time (compared to the expected response time of a web request). For example, if you were to implement a timelapse action, this inherently runs over a long time.

This introduces a couple of problems. Firstly, a request that triggers a long function will, by default, block the Python interpreter for the duration of the function. This usually causes the connection to timeout, and the response will never be revieved.

Action threads are introduced to manage long-running functions in a way that does not block HTTP requests. Any API Action will automatically run as a background thread.

Internally, the labthings.LabThing object stores a list of all requested actions, and their states. This state stores the running status of the action (if itis idle, running, error, or success), information about the start and end times, a unique ID, and, upon completion, the return value of the long-running function.

By using threads, a function can be started in the background, and it’s return value fetched at a later time once it has reported success. If a long-running action is started by some client, it should note the ID returned in the action state JSON, and use this to periodically check on the status of that particular action.

API routes have been created to allow checking the state of all actions (GET /actions), a particular action by ID (GET /actions/<action_id>), and terminating or removing individual actions (DELETE /actions/<action_id>).

All actions will return a serialized representation of the action state when your POST request returns. If the action completes within a default timeout period (usually 1 second) then the completed action representation will be returned. If the action is still running after this timeout period, the “in-progress” action representation will be returned. The final output value can then be retrieved at a later time.

Most users will not need to create instances of this class. Instead, they will be created automatically when a function is started by an API Action view.

class labthings.actions.ActionThread(action: str, target: Optional[Callable] = None, name: Optional[str] = None, args: Optional[Iterable[Any]] = None, kwargs: Optional[Dict[str, Any]] = None, daemon: bool = True, default_stop_timeout: int = 5, log_len: int = 100, http_error_lock: Optional[_thread.allocate_lock] = None)

A native thread with extra functionality for tracking progress and thread termination.

Arguments: * action is the name of the action that’s running * target, name, args, kwargs and daemon are passed to threading.Thread

(though the defualt for daemon is changed to True)

  • default_stop_timeout specifies how long we wait for the target function to stop nicely (e.g. by checking the stopping Event )

  • log_len gives the number of log entries before we start dumping them

  • http_error_lock allows the calling thread to handle some errors initially. See below.

## Error propagation If the target function throws an Exception, by default this will result in: * The thread terminating * The Action’s status being set to error * The exception appearing in the logs with a traceback * The exception being raised in the background thread. However, HTTPException subclasses are used in Flask/Werkzeug web apps to return HTTP status codes indicating specific errors, and so merit being handled differently.

Normally, when an Action is initiated, the thread handling the HTTP request does not return immediately - it waits for a short period to check whether the Action has completed or returned an error. If an HTTPError is raised in the Action thread before the initiating thread has sent an HTTP response, we don’t want to propagate the error here, but instead want to re-raise it in the calling thread. This will then mean that the HTTP request is answered with the appropriate error code, rather than returning a 201 code, along with a description of the task (showing that it was successfully started, but also showing that it subsequently failed with an error).

In order to activate this behaviour, we must pass in a threading.Lock object. This lock should already be acquired by the request-handling thread. If an error occurs, and this lock is acquired, the exception should not be re-raised until the calling thread has had the chance to deal with it.

property cancelled: bool

Alias of stopped

property dead: bool

Has the thread finished, by any means (return, exception, termination).

property exception: Optional[Exception]

The Exception that caused the action to fail.

get(block: bool = True, timeout: Optional[int] = None)

Start waiting for the task to finish before returning

Parameters

  • block – (Default value = True)

  • timeout – (Default value = None)

property id: uuid.UUID

UUID for the thread. Note this not the same as the native thread ident.

property output: Any

Return value of the Action function. If the Action is still running, returns None.

run()

Overrides default threading.Thread run() method

property status: str

Current running status of the thread.

Status

Meaning

pending

Not yet started

running

Currently in-progress

completed

Finished without error

cancelled

Thread stopped after a cancel request

error

Exception occured in thread
stop(timeout=None, exception=<class 'labthings.actions.thread.ActionKilledException'>) → bool

Sets the threads internal stopped event, waits for timeout seconds for the thread to stop nicely, then forcefully kills the thread.

Parameters

  • timeout (int) – Time to wait before killing thread forecefully. Defaults to self.default_stop_timeout

  • exception – (Default value = ActionKilledException)

property stopped: bool

Has the thread been cancelled

terminate(exception=<class 'labthings.actions.thread.ActionKilledException'>) → bool
Parameters

exception – (Default value = ActionKilledException)

Raises

which – should cause the thread to exit silently

update_data(data: Dict[Any, Any])
Parameters

data – dict:

update_progress(progress: int)

Update the progress of the ActionThread.

Parameters

progress – int: Action progress, in percent (0-100)

Accessing the current action thread

A function running inside a labthings.actions.ActionThread is able to access the instance it is running in using the labthings.current_action() function. This allows the state of the Action to be modified freely.

labthings.current_action()

Return the ActionThread instance in which the caller is currently running.

If this function is called from outside an ActionThread, it will return None.

Returns

labthings.actions.ActionThread – Currently running ActionThread.

Updating action progress

Some client applications may be able to display progress bars showing the progress of an action. Implementing progress updates in your actions is made easy with the labthings.update_action_progress() function. This function takes a single argument, which is the action progress as an integer percent (0 - 100).

If your long running function was started within a labthings.actions.ActionThread, this function will update the state of the corresponding labthings.actions.ActionThread instance. If your function is called outside of an labthings.actions.ActionThread (e.g. by some internal code, not the web API), then this function will silently do nothing.

labthings.update_action_progress(progress: int)

Update the progress of the ActionThread in which the caller is currently running.

If this function is called from outside an ActionThread, it will do nothing.

Parameters

progress – int: Action progress, in percent (0-100)