fedbiomed.common.tasks_queue

Module: fedbiomed.common.tasks_queue

Queue module that contains task queue class that is a wrapper to the persistqueue python library.

Classes

TasksQueue

CLASS 
TasksQueue(messages_queue_dir, tmp_dir)

A disk-persistent Queue object, ensuring queue will remain on disk even if program crashes.

Relies on persistqueue package.

Parameters:

Name Type Description Default
messages_queue_dir str

directory where enqueued data should be persisted.

 required
tmp_dir str

indicates where temporary files should be stored.

 required
Source code in fedbiomed/common/tasks_queue.py 
def __init__(self, messages_queue_dir: str, tmp_dir: str):
    """Construct disk-persistent Queue.

    Args:
        messages_queue_dir: directory where enqueued data should be persisted.
        tmp_dir: indicates where temporary files should be stored.
    """
    try:
        self.queue = persistqueue.Queue(messages_queue_dir, tempdir=tmp_dir)
    except ValueError as e:
        msg = ErrorNumbers.FB603.value + ": cannot create queue (" + str(e) + ")"
        logger.critical(msg)
        raise FedbiomedTaskQueueError(msg)

Attributes

queue instance-attribute 
queue = persistqueue.Queue(
    messages_queue_dir, tempdir=tmp_dir
)

Functions

add(task)

Adds a task to the queue

Parameters:

Name Type Description Default
task dict

a dict describing the task to be added

 required
Source code in fedbiomed/common/tasks_queue.py 
def add(self, task: dict):
    """Adds a task to the queue

    Args:
        task: a dict describing the task to be added
    """
    try:
        self.queue.put(task)
    except persistqueue.exceptions.Full:
        msg = ErrorNumbers.FB603.value + ": queue is full"
        logger.critical(msg)
        raise FedbiomedTaskQueueError(msg)

get(block=True)

Get the current task in the queue

Parameters:

Name Type Description Default
block Optional[bool]

if True, block if necessary until an item is available. Defaults to True.

 True

Returns:

Type Description
dict

Dictionary object stored in queue

Raises:

Type Description
FedbiomedTaskQueueError

If queue is empty
Source code in fedbiomed/common/tasks_queue.py 
def get(self, block: Optional[bool] = True) -> dict:
    """Get the current task in the queue

    Args:
        block: if True, block if necessary until an item is available. Defaults to True.

    Returns:
        Dictionary object stored in queue

    Raises:
        FedbiomedTaskQueueError: If queue is empty
    """
    try:
        return self.queue.get(block)
    except persistqueue.exceptions.Empty:
        msg = ErrorNumbers.FB603.value + ": queue is empty"
        logger.debug(msg)
        raise FedbiomedTaskQueueError(msg)

qsize()

Retrieve the size of the queue

Returns:

Type Description
int

size of the queue
Source code in fedbiomed/common/tasks_queue.py 
def qsize(self) -> int:
    """Retrieve the size of the queue

    Returns:
        size of the queue
    """
    return self.queue.qsize()

task_done()

Indicate whether a formerly enqueued task is complete

Returns:

Type Description
Any

True if task is complete
Source code in fedbiomed/common/tasks_queue.py 
def task_done(self) -> Any:
    """Indicate whether a formerly enqueued task is complete

    Returns:
        True if task is complete
    """
    try:
        return self.queue.task_done()
    except ValueError:
        # persistqueue raises it if task_done called too many times we can ignore it
        return