eqsql.task_queues.core module¶

Task Queue protocol and implementations

class eqsql.task_queues.core.Future(eq_sql, eq_task_id, tag=None)¶

Bases: object

Represents the eventual result of an LocalTaskQueue task. Future instances are returned by the LocalTaskQueue.submit_task, and LocalTaskQueue.submit_tasks methods.

Parameters:

eq_sql (TaskQueue) – the LocalTaskQueue instance that created this Future.
eq_task_id (int) – the task id
tag (str) – an optional metadata tag

cancel()¶

Cancels this Future’s task by removing this Future’s task id from the output queue. Cancelation can fail if this Future task has been popped from the output queue before this call completes. Calling this on an already canceled task will return True.

Returns:: True if the task is canceled, otherwise False.

done()¶: Returns True if this Future task has been completed or canceled, otherwise False

property priority: int¶

Gets the priority of this Future task.

Returns:: The priority of this Future task.

result(delay=0.5, timeout=2.0)¶

Gets the result of this future task.

This repeatedly pools the DB for the task result. The polling interval is specified by the delay such that the first interval is defined by the initial delay value which is increased after the first poll. The polling will timeout after the amount of time specified by the timout value is has elapsed.

Parameters:

delay (float) – the initial polling delay value
timeout (float) – the duration after which the query will timeout. If timeout is None, there is no limit to the wait time.

Returns:

ResultStatus.SUCCESS or ResultStatus.FAILURE, and whose second element is either the result of the task, or in the case of failure the reason for the failure (EQ_TIMEOUT, or EQ_ABORT)

Return type:

A tuple whose first element indicates the status of the query

property status: TaskStatus¶

Gets the current status of this Future, one of TaskStatus.QUEUED, TaskStatus.RUNNING, TaskStatus.COMPLETE, or TaskStatus.CANCELED.

Returns:: One of TaskStatus.QUEUED, TaskStatus.RUNNING, TaskStatus.COMPLETE, TaskStatus.CANCELED, or None if the status query fails.

property worker_pool: str | None¶

Gets the id of the worker pool, if any, that this Future task is running on.

Returns:: The id of the worker pool that this Future task is running on, or None if the task hasn’t been selected by a worker pool yet.

class eqsql.task_queues.core.ResultStatus(*values)¶

Bases: IntEnum

Enum defining the status (success or failure) of an EQSQL database operation.

FAILURE = 1¶

SUCCESS = 0¶

exception eqsql.task_queues.core.StopConditionException(msg='StopIterationException', *args, **kwargs)¶: Bases: Exception

class eqsql.task_queues.core.TaskQueue(*args, **kwargs)¶

Bases: Protocol

Task queue protocol for submitting, manipulating and retrieving tasks

are_queues_empty(eq_type=None)¶

Returns whether or not either of the input or output queues are empty, optionally of a specified task type.

Parameters:: eq_type (int) – the optional task type to check for.
Returns:: True if the queues are empty, otherwise False.
Return type:: bool

as_completed(futures, pop=False, timeout=None, n=None, batch_size=1, sleep=0)¶

Returns a generator over the Futures in the futures argument that yields Futures as they complete. The Futures are checked for completion by iterating over all of the ones that have not yet completed and checking for a result. At the end of each iteration, the timeout is checked. Note that adding or removing Futures to or from the futures argument List while iterating may have no effect on this call. A TimeoutError will be raised if the futures do not complete within the specified timeout duration.

Parameters:

futures (List[Future]) – the List of Futures to iterate over and return as they complete.
pop (bool) – if true, completed futures will be popped off of the futures argument List.
timeout (float) – if the time taken for futures to completed is greater than this value, then raise TimeoutError.
n (int) – yield this many completed Futures and then stop iteration.
batch_size (int) – retrieve this many completed futures, before yielding.
sleep (float) – the time, in seconds, to sleep between each iteration over all the Futures.

Return type:

Generator[Future, None, None]

Yields: Futures in the futures argument as they complete.

Examples

>>> futures = []
    // submit some tasks and append the Future instances to futures
    for ft in task_queue.as_completed(futures, timeout=5):
        status, result = ft.result()
        // do something with result

cancel_tasks(futures)¶

Cancels the specified Futures.

Parameters:: futures (List[Future]) – the Futures to cancel.
Returns:: A tuple containing the ResultStatus and the ids of the successfully canceled tasks.
Return type:: Tuple[ResultStatus, int]

get_priorities(futures)¶

Gets the priorities of the specified tasks.

Parameters:: futures (Iterable[Future]) – the futures of the tasks whose priorities are returned.
Returns:: A List of tuples containing the future and priorty for each task, or ResultStatus.FAILURE if the query has failed.
Return type:: List[Tuple[Future, int]]

get_status(futures)¶

Gets the status (queued, running, etc.) of the specified tasks

Parameters:: futures (Iterable[Future]) – the futures of the tasks to get the status of.
Returns:: A List of Tuples containing the status of the tasks. The first element of the tuple will be the task id, and the second element will be the status of that task as a TaskStatus object.
Return type:: List[Tuple[Future, TaskStatus]]

get_worker_pools(futures)¶

Gets the worker pools on which the specified list of Futures are running, if any. All the specified Futures must have been submitted by the same task queue (i.e., they are all in the same eqsql database).

Returns:: A list of two element tuples. The first element is a Future, and the second is that Futures’s worker pool, or None, if the Future hasn’t been selected for execution yet.
Parameters:: futures (List[Future])
Return type:: List[Tuple[Future, str | None]]

pop_completed(futures, timeout=None, sleep=0)¶

Pops and returns the first completed future from the specified List of Futures.

Parameters:

futures (List[Future]) – the List of Futures to check for a completed one. The completed Future will be popped from this list.
timeout – a TimeoutError will be raised if a completed Future cannot be returned by after this amount time.
sleep (float) – the time, in seconds, to sleep between each iteration over all the Futures, when looking for a completed one.

Returns:

The first completed Future from the specified List of Futures.

Return type:

Future

query_result(eq_task_id, delay=0.5, timeout=2.0)¶

Queries for the result of the specified task.

The query repeatedly polls for a result. The polling interval is specified by the delay such that the first interval is defined by the initial delay value which is increased after the first poll. The polling will timeout after the amount of time specified by the timout value is has elapsed.

Parameters:

eq_task_id (int) – the id of the task to query
delay (float) – the initial polling delay value
timeout (float) – the duration after which the query will timeout. If timeout is None, there is no limit to the wait time.

Returns:

ResultStatus.SUCCESS or ResultStatus.FAILURE, and whose second element is either the result of the task, or in the case of failure the reason for the failure (EQ_TIMEOUT, or EQ_ABORT)

Return type:

A tuple whose first element indicates the status of the query

submit_task(exp_id, eq_type, payload, priority=0, tag=None)¶

Submits work of the specified type and priority with the specified payload, returning the status and the Future encapsulating the submission.

Parameters:

exp_id (str) – the id of the experiment of which the work is part.
eq_type (int) – the type of work
payload (str) – the work payload
priority (int) – the priority of this work
tag (str) – an optional metadata tag for the task

Returns:

A tuple containing the status (ResultStatus.FAILURE or ResultStatus.SUCCESS) of the submission and if successful, a Future representing the submitted task otherwise None.

Return type:

Tuple[ResultStatus, Future | None]

submit_tasks(exp_id, eq_type, payload, priority=0, tag=None)¶

Submits work of the specified type and priority with the specified payloads, returning the status and the futures encapsulating the submission.

Parameters:

exp_id (str) – the id of the experiment of which the work is part.
eq_type (int) – the type of work
payload (List[str]) – a list of the work payloads
priority (int) – the priority of this work
tag (str) – an optional metadata tag for the tasks

Returns:

A tuple containing the status (ResultStatus.FAILURE or ResultStatus.SUCCESS) of the submission and the list of futures for the submitted tasks. If the submission fails, the list of futures will contain the futures that submitted sucessfully.

Return type:

Tuple[ResultStatus, List[Future]]

update_priorities(futures, new_priority)¶

Updates the priority of the specified Futures to the new_priority.

Parameters:

futures (List[Future]) – the Futures to update.
new_priority (int | List[int]) – the priority to update to. If this is a single integer then all the specified tasks are updated to that priority. If this is a List of ints then each task is updated with the corresponding priority, i.e., the first task in the futures is updated with the first priority in the new_priority List.

Returns:

If the update is successful, the Tuple will contain ResultStatus.SUCCESS and: the eq_task_ids of the tasks whose priority was successfully updated, otherwise (ResultStatus.FAILURE, []).

Return type:

Tuple[ResultStatus, int]

class eqsql.task_queues.core.TaskStatus(*values)¶

Bases: IntEnum

Enum defining the status of a task: queued, etc. These are used in the database to store the status of a task.

CANCELED = 3¶

COMPLETE = 2¶

QUEUED = 0¶

REQUEUED = 4¶

RUNNING = 1¶

exception eqsql.task_queues.core.TimeoutError(msg='TimeoutError', *args, **kwargs)¶

Bases: Exception

Exception used to indicate that a query has timed out.

eqsql.task_queues.core module¶

Table of Contents

Previous topic

Next topic

This Page