The Persistent Queue - perqueue

The aim of the perqueue package is to be a lightweight but powerful workflow manager that wraps the myqueue package. Thus, perqueue can make use of all the schedulers that myqueue already supports.

The attractive points of using perqueue for workflow management instead of using the workflow eco-system already present in myqueue comes down to at least five areas, where the two differ:

1. perqueue has internal data storage for all submitted tasks, which means that data can be returned from each task and stored with it. This makes it a lot simpler to propagate data from one task to another, and thus dependencies can much more easily access the results or parameters of their dependants.
2. Tasks submitted through perqueue are only submitted to the job scheduling system (by use of myqueue) once all dependencies of that task are met. Therefore, large volume parallel jobs can be submitted without worrying about restrictions to the max number of jobs allowed.
3. Any python file can be run through perqueue without having to touch any environment variables nor installing it as a package. The code is instead run with runpy, and only needs a main() function, which is described in Format of Code.
4. While myqueue marks each task as done, running, failed and timed-out, perqueue also distinguishes between whether a task has met a given criteria or not. Thus, if a workflow is of a filtering nature, where the specific result of a calculation determines whether the following steps are performed, perqueue can easily stop the workflow at that point.
5. Dynamic workflows can be implemented for cyclical and variable width cases. Additionally, dynamic workflow branching is supported.

Some of the upcoming features in the next releases are going to be huge improvements over myqueue, and they will include:

• Extending workflows after submission

Installing perqueue

Since the package is not on PyPI, you need to install it directly from GitLab. At the moment, this is achieved by cloning the project and installing it from source. In the future, there will hopefully be pre-compiled wheels, that one can install directly.

Install from GitLab

If you are a regular user of perqueue, you can use the pip-git interface to install directly from GitLab:

pip install git+https://gitlab.com/asm-dtu/perqueue.git

Otherwise, if you are installing it as a developer, you can do an editable install. Start by cloning the project using the SSH protocol (this requires setting up SSH keys):

git clone git@gitlab.com:asm-dtu/perqueue.git

Then move into the newly created perqueue directory and in there run:

pip install -e .

The -e creates an editable install. The install will be static, if it's omitted.

Either way will install perqueue and its dependencies (which can be found in pyproject.toml). The next thing to do is set up myqueue, if this hasn't already been done. A guide to this can be found in their documentation.

Once perqueue has been installed, you need to initialize the queue database that it relies on. You can do this by running the following command in the project directory:

pq init

This will initialize a queue database in the current directory.

You can find all the options of the pq CLI by calling pq --help

Installing a Specific Version from GitLab

If you need to install a specific release of perqueue, you can clone the repository with the following command instead:

git clone -b <release-tag> --single-branch git@gitlab.com:asm-dtu/perqueue.git

Thereafter, you can just follow the rest of the installation instructions.

Using the perqueue package

The perqueue package is designed to run on top of myqueue, so ensure that you have a basic understanding of that package before venturing too far with perqueue.

To use perqueue, you define a script that contains all the steps of the current workflow. The workflow must correspond to a Directed Acyclic Graph, which means that there can't be any cycles in the workflow. To introduce cyclicity, you make use of the CyclicalGroup. Each step is defined as a Task, which corresponds to a single python script. The dependencies between them are defined with Workflows, that use the DiGraph format from networkx to describe the dependency graph. Each script must have a main() function, which works as an entry point. This is described in Format of Code.

The Command Line Interface

The package includes a command line interface to make it simpler to handle certain operations of the package. Calling pq --help will give you an overview of the functionality, and each subcommand also has help text.

The most commonly used subcommands are list, modify and resubmit. These list the tasks of the queue, modify the arguments or resources of a job, and resubmit a job that has failed or timed out, respectively.
These subcommands can filter the affected tasks directly regarding the perqueue ID, their (most recent) MyQueue ID, the name of the code (or Task) and the state of the task.

For the IDs, you can specify either a single integer, a comma-separated list of integers (e.g., 1,2,3,4 - no white-space) or an inclusive range of integers (1-10).
The name supports both globbing and regex, such that you don't have to specify the full name.

Do beware that when using globbing or regex, you'll have to quote the string, e.g., pq list -n "cod*py". Otherwise things will break.

Creating Workflows with PerQueue

Any workflow that utilizes perqueue is defined in a python script like the one shown here, which you run as a normal Python script for submission:

from pathlib import Path
from perqueue import PersistentQueue, Task, Workflow

code_path<n>: str | Path
args<n>: dict
resources<n>: str = "cores:partition:procs:time"  # Uses same format as MyQueue

task1 = Task(code_path1, args1, resources1, name="start")
task2 = Task(code_path2, args2, resources2, name="left")
task3 = Task(code_path3, args3, resources3, name="right")

with PersistentQueue() as pq:
    pq.submit(Workflow({task1: [], task2: [task1], task3: [task1]}))

# This submits a workflow with the following structure
#         start
#     ______|______
#    |            |
#  left         right

📑 When adding tasks to the PersistentQueue, one must always use it in a with block! This ensures that the jobs are only submitted if no errors occur during the submission. NB: perqueue throws an Exception, if the context manager way is not used.

Note that the code argument is either a str or a pathlib.Path with the relative path from the current, or an absolute path, that points to the python file containing the code of the task.

Examples of actual workflows can be found in the examples folder.

Format of Code

To make code work with the perqueue workflow, it needs just 1 thing:
The code must have a main function with the signature

def main(arg1=..., ..., **kwargs) -> Tuple[bool, Optional[dict]]:

Which in plain speak means, that the function can take any number of key-value parameters (but no non key-value parameters), and must return a tuple. The first element of this tuple must be a boolean (either True or False), and the second element must be either None or a dict.

Data Passing

A nice feature of perqueue is the automatic passing of data from dependencies to the dependants. This means that any data returned as part of the dict of a task is automatically passed on to its child tasks. Let's illustrate this with an example:

Task2 depends on Task1. The dict returned from Task 1 is {"maximum_value": 42, "path_to_data": "path/to/file"}. Thus, the main function of Task2 can be def main(maximum_value=None, **kwargs): and inside the scope of main, the maximum_value will have been set to 42, and kwargs will be {"path_to_data": "path/to/file", ...}. This considerably simplifies passing data between tasks.

Task1(**kwargs)
  |
{"maximum_value": 42, "path_to_data": "path/to/file"}
  |
  v
Task2(maximum_value, **kwargs)
  |
  '- main(maximum_value=42, path_to_data="path/to/file")

The local runner

In addition to the schedulers that myqueue support, perqueue can also run tasks as background processes, which are run on the local system (the login-node on clusters) on a single core. This ensures that fast pieces of code can be run without having to reserve a job with the scheduler. To stay good friends with your system administrator, this runner should only be used for tasks that take less than a minute or two.

In future releases, there will be a hard limit on the runtime of a local task.

IMPORTANT: In the current releases, the local runner should only be used in the beginning of a workflow, due to how they are run. So, if you have already started submitting things to compute nodes, stop using the local runner.

To submit a task to the local runner, specify the resources as "local:2m", i.e., you specify the partition as local.