eqsql.task_queues.local_queue module¶

Local task queue implementations. A local task queue is one which communicates directly with the database, rather than through some intermediary service (e.g., Globus Computer or a RESTful API)

class eqsql.task_queues.local_queue.LocalTaskQueue(db, logger)¶

Bases: object

Creates an LocalTaskQueue task queue connected to the specified database, logging to the specified logger. LocalTaskQueue tasks queues should be created with init_task_queue().

Parameters:

db (WorkflowSQL) – the database to submit and retrieve tasks from
logger (Logger) – the logger to use for logging

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]

clear_queues()¶

Clears the input and output queues and sets the status of those tasks in the tasks table to CANCELED.

NOTE: this is only a convenience method for resetting the queues to a coherent starting state, and should NOT be used to cancel tasks.

close()¶: Closes the DB connection, and terminates this LocalTaskQueue instance.

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

pop_in_queue(cur, eq_task_id, delay, timeout)¶

Pops the specified task off of the db in queue.

This call repeatedly polls for a task with the specified id. The polling interval is specified by the delay such that the first interval is defined by the initial delay value which is then incremented after each poll. The polling will timeout after the amount of time specified by the timout value is has elapsed.

Parameters:

cur – the database cursor
eq_task_id (int) – id of the task to pop
delay (float) – the initial polling delay value
timeout (float) – the duration after which this call will timeout and return. If timeout is None, there is no limit to the wait time.

Return type:

Tuple[ResultStatus, int | str]

Returns: A two element tuple where the first elements is one of: ResultStatus.SUCCESS or ResultStatus.FAILURE. On success the second element will be eq task id. On failure, the second element will be one of EQ_ABORT or EQ_TIMEOUT depending on the cause of the failure.

pop_out_queue(cur, eq_type, n, delay, timeout)¶

Pops the highest priority task of the specified work type off of the db out queue.

This call repeatedly polls for a task of the specified type. The polling interval is specified by the delay such that the first interval is defined by the initial delay value which is then incremented after each poll. The polling will timeout after the amount of time specified by the timout value is has elapsed.

Parameters:

eq_type (int) – the type of the work to pop from the queue
n (int) – the maximum number of tasks to pop from the queue
delay (float) – the initial polling delay value
timeout (float) – the duration after which this call will timeout and return. If timeout is None, there is no limit to the wait time.

Return type:

Tuple[ResultStatus, int | str]

Returns: A two element tuple where the first elements is one of: ResultStatus.SUCCESS or ResultStatus.FAILURE. On success the second element will be the popped eq task id. On failure, the second element will be one of EQ_ABORT or EQ_TIMEOUT depending on the cause of the failure.

push_in_queue(cur, eq_task_id, eq_type)¶

Pushes the specified task onto the input queue.

Parameters:

cur – the database cursor used to execute the sql push
eq_task_id (int) – the id of the task
eq_type (int) – the type of the task

Returns:

ResultStatus.SUCCESS if the task was successfully pushed onto the input queue, otherwise ResultStatus.FAILURE.

Return type:

ResultStatus

push_out_queue(cur, eq_task_id, eq_type, priority=0)¶

Pushes the specified task onto the output queue with the specified priority.

Parameters:

cur – the database cursor used to execute the sql push
eq_task_id (int) – the id of the task
eq_type (int) – the type of the task
priority (int) – the priority of the task

query_more_tasks(eq_type, eq_task_ids, batch_size, threshold=1, worker_pool='default', delay=0.5, timeout=2.0)¶

Queries for tasks of the specified type, returning up to batch_size number of tasks. The exact number of task to return is batch_size - X where X is the number of currently running tasks from those in eq_task_ids. The intention here is that a worker pool may have a limited amount of capacity and should not get more tasks than that capacity. eq_task_ids keeps track of the number tasks the worker pool is working on and batch_size is the maximum amount of work (tasks) the worker pool wants to execute. When the difference between batch size and the number of running tasks is greater than the threshold then query for tasks.

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

Parameters:

eq_type (int) – the type of the work to query for.
eq_task_ids (Iterable[int]) – the possibly running task ids used to determine the number of tasks to return.
batch_size (int) – the maximum amount of tasks to return
threshold (int) – the number of free “slots” (difference between running workers and batch_size) that must be available before tasks are queried for. This must be less than or equal to batch size.
worker_pool (str) – the id of the worker pool querying for the tasks
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:

A two element Tuple where the first element is a List of the ids of the currently running tasks from those specified in eq_task_ids plus the ids of any new tasks, and the second element is a List of Dictionaries as returned by query_task().

Return type:

Tuple[List[int], List[Dict]]

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

query_task(eq_type, n=1, worker_pool='default', delay=0.5, timeout=2.0)¶

Queries for the highest priority task of the specified type.

The query repeatedly polls for n number of tasks. The polling interval is specified by the delay such that the first interval is defined by the initial delay value which is then incremented after each poll. The polling will timeout after the amount of time specified by the timout value is has elapsed.

Parameters:

eq_type (int) – the type of the task to query for
n (int) – the maximum number of tasks to query for
worker_pool (str) – the id of the worker pool query for the tasks
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:

If n == 1, a single dictionary will be returned, otherwise a List of dictionaries will be returned. In both cases, the dictionary(ies) contain the following. If the query results in a status update, the dictionary will have the following format: {'type': 'status', 'payload': P} where P is one of 'EQ_STOP', 'EQ_ABORT', or 'EQ_TIMEOUT'. If the query finds work to be done then the dictionary will be: {'type': 'work', 'eq_task_id': eq_task_id, 'payload': P} where P is the parameters for the work to be done.

Return type:

List[Dict] | Dict

report_task(eq_task_id, eq_type, result)¶

Reports the result of the specified task of the specified type

Parameters:

eq_task_id (int) – the id of the task whose results are being reported.
eq_type (int) – the type of the task whose results are being reported.
result (str) – the result of the task.

Returns:

ResultStatus.SUCCESS if the task was successfully reported, otherwise ResultStatus.FAILURE.

Return type:

ResultStatus

select_task_payload(cur, eq_task_ids, worker_pool_id='default')¶

Selects the json_out payload associated with the specified task ids in the eq_tasks table, setting the start time of the tasks to the current time, the status of the tasks to TaskStatus.RUNNING, and the worker_pool to the specified worker_pool.

Parameters:

cur – the database cursor used to execute the sql select
eq_task_ids (Iterable[int]) – the ids of the tasks to get the json_out for
worker_pool_id (str) – the id of the worker pool asking for the payload

Returns:

If successful, a list of tuples containing an eq_task_id and the json_out payload, otherwise raise an exception.

Return type:

List[Tuple[int, str]]

select_task_result(cur, eq_task_id)¶

Selects the result (json_in) payload associated with the specified task id in the eq_tasks table.

Parameters:: eq_task_id (int) – the id of the task to get the json_in for
Returns:: The result payload for the specified task id, if successfull, otherwise raise an exception.
Return type:: str

stop_worker_pool(eq_type)¶

Stops any workers pools associated with the specified work type by pushing EQ_STOP into the queue.

Parameters:: eq_type (int) – the work type for the pools to stop
Returns:: ResultStatus.SUCCESS if the stop message was successfully pushed, otherwise ResultStatus.FAILURE.
Return type:: ResultStatus

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. This is essentially a convenience wrapper around submit_task() for submitting a list of payloads.

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]

update_task(cur, eq_task_id, payload)¶

Updates the specified task in the eq_tasks table with the specified result payload (the json_in). This also updates the time_stop to the time when the update occurred.

Parameters:

eq_task_id (int) – the id of the task to update
payload (str) – the payload to update the task with

eqsql.task_queues.local_queue.init_task_queue(host, user, port, db_name, password=None, retry_threshold=0, log_level=30)¶

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

Parameters:

host (str) – the eqsql database host
user (str) – the eqsql database user
port (int) – 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.
log_level – the logging threshold level.

Returns:

An LocalTaskQueue instance

Return type:

TaskQueue

eqsql.task_queues.local_queue.random() → x in the interval [0, 1).¶

eqsql.task_queues.local_queue module¶

Table of Contents

Previous topic

Next topic

This Page