eqsql.task_queues.service_queue module

class eqsql.task_queues.service_queue.ServiceTaskQueue(service_url, db_params)

Bases: object

Task queue protocol for submitting, manipulating and retrieving tasks

Parameters:

  • service_url (str)

  • db_params (DBParameters)

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. Assuming a batch_size > 1, the remotely executing code will retrieve this many results before returning them to the locally executing code. Consequently, this greatly reduces communication overhead, and can improve the performance of a remote queue.

  • 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]

eqsql.task_queues.service_queue.init_task_queue(service_url, db_host, db_user, db_port, db_name, password=None, retry_threshold=0)

Initializes and returns an LocalTaskQueue class instance with the specified parameters.

Parameters:

  • host – the eqsql database host

  • user – the eqsql database user

  • port – the eqsql database port

  • db_name (str) – the eqsql database name

  • password (str) – the eqdql database password (if there is one)

  • retry_threshold – if a DB connection cannot be established (e.g, there are currently too many connections), then retry retry_threshold many times to establish a connection. There will be random few second delay betwen each retry.

  • service_url (str)

  • db_host (str)

  • db_user (str)

  • db_port (int)

Returns:

An ServiceTaskQueue instance

Return type:

ServiceTaskQueue