`ll.taskswarm` – Running tasks in parallel

Overview

ll.taskswarm provides an easy way to execute multiple Python functions in parallel using the multiprocessing module. It is designed to improve performance by utilizing multiple CPU cores efficiently.

Features

Parallel execution of multiple functions or other callables;
Automatic process management;
Supports passing arguments to functions;
Configurable number of parallel processes;
Tasks can dynamically add new tasks during execution;
Live progress output on the terminal (and in the iTerm2 session status via ll.iterm2).

Limitations

Functions/callables, their arguments (and the object in case of bound methods) must be picklable (depending on the start method used by multiprocessing).
No return values are collected; tasks are executed for their side effects (like created files, executed HTTP or database requests).
Overhead from creating processes may outweigh benefits of parallelization.

Usage

A task is a function (or other callable) that gets a Task object as its first argument. It may report its progress via Task.log() and submit further tasks via Task.submit(). The task is finished when the function returns (or raises an exception).

Tasks are submitted to a Swarm object inside a with block. The tasks are started when the with block is exited, so the with block itself only collects the initial tasks:

import sys, json, pathlib

from ll import taskswarm


def walk(task, dir):
        task.log(t"Walking {dir}")
        for file in dir.iterdir():
                if file.is_dir():
                        task.submit(walk, file.name, file)
                elif file.suffix == ".json":
                        task.submit(prettyprint, file.name, file)
        task.log(t"Walked {dir}")


def prettyprint(task, file):
        task.log(t"Pretty printing {file}")
        data = json.loads(file.read_text())
        file.write_text(json.dumps(data, indent="       "))
        task.log(t"Pretty printed {file}")


if __name__ == "__main__":
        with taskswarm.Swarm("Pretty printing JSON files", 20, False) as swarm:
                dir = pathlib.Path(sys.argv[1])
                swarm.submit(walk, str(dir), dir)

Each task runs in its own multiprocessing.Process. At most processes (20 in the example) tasks run at the same time, additional tasks are queued until a slot becomes free. The swarm is finished when all tasks (including those submitted by other tasks) have finished. Then a summary of the total, mean and longest wait and run times is printed.

Log messages and task names can be strings or t-strings (i.e. string.templatelib.Template objects). A t-string is passed to the swarm unchanged and formatted there by Swarm.format(): Each interpolated value is formatted (and colored) according to its type (again by Swarm.format()), so the task function doesn’t have to do that itself. When the type isn’t enough to decide how a value should be formatted (e.g. a database schema name that is a plain str), the format spec selects the formatting: t"Exporting {name:schema}" calls the method format_schema of the swarm (which a subclass can add). Since the t-string is sent to the main process via a multiprocessing.Queue, all interpolated values must be picklable. Values that aren’t must be preformatted by the task (e.g. t"Exporting {str(obj):sqlobject}").

Output

While the swarm is running, its progress is printed to stdout. There are two output modes (selected via the continuous argument of Swarm):

Continuous output (continuous=True): Every event (a task being started, logging a message or finishing) is printed as a new line, prefixed with the elapsed time and the number of waiting, running and finished tasks. This mode is appropriate when the output is redirected to a file or a CI log.
In-place output (continuous=False): One line per process slot is reserved on the terminal and updated in place (via ANSI cursor movement) with the current state of the task running in that slot. Below these lines a status line shows the numbers of waiting, running and finished tasks and the current and mean load (i.e. the number of running processes). This mode requires a terminal that understands ANSI escape sequences.

In both modes values are colored via ANSI escape sequences (see Swarm.format()), unless Swarm.plain_output is true, which is the case on Windows or if stdout isn’t a terminal. The formatting can be customized by overwriting the methods Swarm.format(), Swarm.format_sep() and Swarm.format_task() in a subclass, or by adding format_<spec> methods that can be selected via the format spec in t-strings (see above).

When running in iTerm2, the title of the swarm and the current progress are also shown in the iTerm2 session status (see ll.iterm2.set_session_status()), so that the progress is visible in the tab even when another tab is active.

class ll.taskswarm.Swarm[source]

Bases: object

A Swarm executes tasks in parallel processes.

A Swarm is used as a context manager: Inside the with block tasks are submitted via submit(), when the block is exited the tasks are executed (and further tasks submitted by running tasks via Task.submit() are executed too). The with block ends when all tasks are done. If an exception is raised inside the with block, no tasks are started.

The constructor arguments are:

titlestring: The title of the swarm. It is shown in the iTerm2 session status while the swarm is running.
processesint: The maximum number of tasks that run in parallel.
continuousbool: Selects the output mode: continuous output (one line per event) if true, in-place output (one line per process slot) if false. See the module documentation for details.

After the with block the attributes tasks, run_time and mean_load can be used to inspect the result.

All output of the swarm goes through the methods format(), format_sep() and format_task(), so the formatting can be changed by overwriting them in a subclass.

The following attributes influence the formatting and can be changed after the constructor call:

plain_outputbool: If true, no ANSI escape sequences are output. Defaults to true on Windows or if stdout isn’t a terminal.
currentdirpathlib.Path: format() prints paths relative to this directory (if possible). Defaults to the current directory when the swarm was created.

property current_load: The number of tasks currently running.

property mean_load

The mean number of running tasks over the runtime of the swarm (or None if the swarm hasn’t run yet).

The load is sampled whenever a task event (start, log message or done) is processed, so this is an average over events, not over time.

property run_time: The total run time of the swarm as a timedelta (or None if the swarm hasn’t finished yet).

format(obj, spec='')[source]

Format obj (a log message, a task name or a value interpolated in one of them) for the output of the swarm (with ANSI colors).

If obj is a t-string (i.e. a string.templatelib.Template object) each interpolated value is converted according to its conversion (!r, !s or !a) and then formatted with format() (passing the format spec); spec itself is ignored in this case.

Otherwise spec is the format spec from a t-string and is used as follows:

If spec isn’t empty and the swarm has a method named format_<spec>, this method is called with obj. This is how a subclass can support formatting values whose type isn’t enough to decide how they should be formatted (e.g. {name:schema} calls self.format_schema(name)).
Otherwise the formatting depends on the type of obj: pathlib.Path objects are output relative to currentdir (if possible) in yellow, numbers and datetime objects in bold magenta (with spec as the format spec, or , for int and ,.01f for float if spec is empty), timedelta objects as HH:MM:SS with the leading zeros in normal magenta and the rest in bold magenta (spec is ignored). Everything else is formatted with the builtin format() and spec (so an unknown spec raises a ValueError).

If plain_output is true, no ANSI escape sequences are output.

Overwrite this method in a subclass to change the formatting or to support additional types.

format_sep(text)[source]

Format the separator text for the output of the swarm (i.e. in dark grey).

If plain_output is true, no ANSI escape sequences are output.

format_task(task)[source]: Format the “full name” of task (i.e. the names of all tasks in Task.path() joined with separators) for the output of the swarm.

submit(func, name, *args, **kwargs)[source]

Submit a new task to the swarm and return the Task object.

func is the callable that gets executed in a separate process. It will be called as func(task, *args, **kwargs) (where task is the Task object). The task is finished when func returns or raises an exception. name is the name of the task used in the output. It can be a string or a t-string (see format()).

Tasks submitted before the with block is exited are started (in submission order, as far as free slots are available) when the block is exited. This method may only be called from the main process; a running task submits further tasks via Task.submit().

__exit__(exc_type, exc_value, traceback)[source]

Run all submitted tasks (and the tasks they submit) until all of them are done, then print a summary.

Events from the tasks (log messages, done messages and submissions of new tasks) arrive via a multiprocessing.Queue and are processed in the main process.

If an exception was raised inside the with block, no tasks are started and the exception propagates.

class ll.taskswarm.Task[source]

Bases: object

A Task object represents one task submitted to a Swarm.

It is created by Swarm.submit() (or Task.submit()) and passed as the first argument to the task function. The task function uses it to communicate with the swarm (which runs in the main process): log() outputs a progress message and submit() submits further tasks. The task is finished when the task function returns.

Everything sent to the swarm (log messages, names and arguments of new tasks) is pickled by the task itself before it’s put into the multiprocessing.Queue, so that unpicklable values raise an exception in the task function (instead of silently losing the message in the feeder thread of the queue).

The following attributes are available:

process_idint: The unique id of the task inside the swarm.
namestring or t-string: The name of the task (used in the output, see Swarm.format()).
parentTask or None: The task that submitted this task (or None for tasks submitted via Swarm.submit()).
submitted_at, started_at, finished_atdatetime or None: When the task was submitted, started and finished. Note that these are only maintained in the main process, so they are not available inside the task function.

__str__()[source]: The names of all tasks in path() joined with ::, i.e. the “full name” of the task including its ancestors (without any formatting, see Swarm.format_task() for that).

property plain_name

The name of the task as a plain string.

If the name is a t-string, the interpolated values are simply converted with str (format specs are ignored).

property wait_time: How long the task waited for a free slot as a timedelta (or None if it hasn’t started yet).

property run_time: How long the task ran as a timedelta (or None if it hasn’t finished yet).

path()[source]: Yield the chain of tasks from the root task down to this task (i.e. all ancestors via parent followed by the task itself).

log(message)[source]

Output the progress message message for this task.

message can be a string or a t-string (see Swarm.format()). All interpolated values of a t-string must be picklable; preformat those that aren’t (e.g. t"{str(obj):sqlobject}").