mrfifo 0.7

Map-Reduce parallelism over FIFOs (named pipes)

Mr. FIFO

Mr. FIFO is a python/cython package that builds on top of the multiprocessing module, providing high throughput and very low-overhead, directional, point-to-point interprocess communication. It essentially provides Map-Reduce parallelism over FIFOs (aka named pipes).

Abstract

Some problems can very nicely be implemented in python and are easily parallelizable, but require high throughput. The python multiprocessing module primitives (Queue, Pipe, etc.) however suffer from significant overhead (pickling/un-pickling etc.) and can become a severe bottleneck. Mr. FIFO provides few very low overhead function primitives (implemented in Cython) that exploit the time-tested and extremely optimized inter-process communication framework known as named pipes or FIFOs (first-in first-out). The development is heavily geared towards bioinformatics tasks (processing large BAM files for instance), but is in no way restricted to that.

Install

   pip install mrfifo

should do the trick.

Example workflow

Here's an instructive example that shows off several of the main features. The following code decompresses a BAM file, distributes the SAM lines to four separate worker processes running in parallel, and then re-assembles a new BAM with the exact same content. The workers execute the function pass_through which iterates over its input, counts the number of lines (SAM records) it encounters, and writes the line to an output. After exhausting the input, the function returns the count and exits. The magic happens before and after the calls to pass_through. For each of the calls defining the workflow (called "parts" of the workflow), a separate multiprocessing.Process is started. In fact, in reverse order, so starting with the .funnel() and then working its way up to the .BAM_reader(). However, before any processes get started, the workflow object figures out how many named pipes it needs to create (in a safe temp directory), by way of the mf.FIFO objects being created and passed into each part. Unless, explicitly specified (using manage_pipes=False), these pipes are additionally managed by mrfifo, to avoid deadlocks. For instance, pass_through is already called with two opened file objects, one for reading and one for writing, which have been created by mrfifo to connect it with the .distribute part on the input side and the .collect part on the output side. Note that return values and exceptions (including tracebacks) are automatically captured by the workflow object to simplify your program and to avoid hanging processes waiting for a pipe to be closed. A little extra is the BAM header. This gets detected and separated from the bulk of the data by the .distribute part and is sent to its own FIFO named header. The header thus bypasses the workers and is merged with their outputs only at the .collect part. However, the header could alternatively have been broadcast to all workers as well. See the documentation for more details.

    def pass_through(input, output):
        i = 0
        for line in input:
            i += 1
            output.write(line)

        return i

    def is_header(line):
        return line.startswith("@")

    def test_bam_reconstruct(chunk_size=1, n=4):
        import mrfifo as mf
        w = (
            mf.Workflow("BAM_reconstruct")
            .BAM_reader(input="test_data/tiny_test.bam")
            .distribute(
                input=mf.FIFO("input_sam", "rt"),
                outputs=mf.FIFO("dist{n}", "wt", n=n),
                chunk_size=chunk_size,
                header_detect_func=is_header,
                header_broadcast=False,
                header_fifo=mf.FIFO("header", "wt"),
            )
            .workers(
                input=mf.FIFO("dist{n}", "rt"),
                output=mf.FIFO("out{n}", "wt"),
                func=pass_through,
                n=n,
            )
            .collect(
                inputs=mf.FIFO("out{n}", "rt", n=n),
                header_fifo=mf.FIFO("header", "rt"),
                output=mf.FIFO("out_sam", "wt"),
                chunk_size=chunk_size,
            )
            .funnel(
                input=mf.FIFO("out_sam", "rt"),
                output="test_data/reconstruct.bam",
                _manage_fifos=False,
                func=mf.parts.bam_writer,
            )
            .run()
        )

Please note that we assign the arguments of pass_through to mf.FIFO instances. This tells mrfifo that input and output are considered internal plumbing between the various parts executed in parallel. As such, Mr.FIFO will actually create named pipes in a temporary directory for each connection between parts. Unless explicitly turned of with manage_pipes=False, Mr. FIFO will even open and close the named pipes to avoid deadlocks in case of any exceptions.

The example above actually creates a number of sub-processes:

• BAM_reconstruct.bam_reader0: reads a BAM file from test_data/tiny_test.bam and decompresses to the named pipe input_sam
• BAM_reconstruct.dist0: reads line-by-line from input_sam and distributes in a round-robin fashion to four named pipes ['dist0', 'dist1', 'dist2', 'dist3']. The BAM header is separately sent to its own FIFO named header.
• BAM_reconstruct.worker{i}: four processes, each reading from dist{i} and writing to out{i} where i goes from 0 to 3.
• BAM_reconstruct.collect0: reads line-by-line, round-robin from ['counted0', ..., 'counted3'] and writes the line to out_sam. But not before passing on the contents of the header!
• BAM_reconstruct.funnel0: sends the content of out_sam to a samtools view process which re-creates a compressed BAM file with the identical original context.

API

Under the hood, the workflow creates mrfifo.Job instances for each process. These are thin wrappers around mutliprocessing.Process, which allow to keep track of the inputs and outputs, collect return values and unhandled exceptions. func is a callable that must not especially be prepared for mrfifo. In general, it should probably have an argument that can be iterated over (line-wise) so that we can connect it to a FIFO. The API to compose the workflow consists of useful shorthands to create meaningfully different jobs. But there may be different ways to combine code with the offered primitives. The currently proposed logic is as follows:

.reader

Readers sequentially process one or more input files and write to a single output. In the above example we use a as input, but we could also have provided /dev/stdin. Typically, the entry-point to your workflow will be defined by one or more .reader() calls.

.distribute

A distributor splits an input into multiple outputs by cycling through the outputs as it iterates over chunks of input. The default distributor is implemented in cython, tries to allocate large pipe buffers and has low overhead. For more complex input streams, it offers detection and separate treatment of a header. The header can either be broadcast to each of the outputs, or sent to a dedicated output that is exclusively used for the header (see the documentation). The code also supports very basic pattern matching and an output lookup mechanism which allows, for example, to split BAM records by the first few bases of the Unique Molecular Identifier (UMI) or Cell Barcode (CB). This is very useful if the downstream workers, running in parallel need to ensure that records with the same UMI or CB are always sent to the same worker.

.workers

This is the only API function that creates multiple jobs from one call. inputs and outputs are expanded for each worker using the n variable in the inputs and outputs argument. Note that the input FIFO contains a wildcard that will be replaced with the appropriate number for each worker. This allows to route multiple inputs (from multiple distributors) into workers while keeping the input streams in sync.

.collect

Collectors do the opposite of distributors. They read round-robin from multiple inputs and write to a single output. This output can be a file and may represent the output of your workflow. It can of course also be a FIFO.

.funnel

A funnel is just a job with a single output, but possibly multiple inputs. Any kind of processing may fit in here. Here, for example, we re-compress already collected output.

w.results_dict

Note that after a workflow is complete, you can find the return values of the functions assigned to subprocesses in the dictionary w.results_dict[<job-name>]. Similarly, exception tracebacks can be retrieved from w.exc_dict.