Ekorn-Pipeline

Stages for any function: let the caller finish the job.

A function that hands out values can't know everything its callers will want done with them. Ekorn::Pipeline lets it take a list of stages from its caller, small functions that each see one value at a time, and run what it would return through them. The caller says, beside the call, what they want; the function does what only it can, and no more.

use Ekorn::Pipeline;
use Ekorn::Pipeline::Stages::Flow;    # take and drop

# The author's function: it reads log lines, and nothing else.
sub log_lines ($file, @stages) {
    open my $fh, '<', $file or die "Can't read $file: $!\n";
    return Ekorn::Pipeline->new(@stages)
        ->iterator(sub { scalar(<$fh>) // () });
}

# A caller's use of it: the first ten errors, as their messages.
my $errors = log_lines(
    'app.log',
    sub { m( \A ERROR \  (.*) )x ? $1 : () },    # only errors, as messages
    take(10),                                    # ten, then stop reading
);
while (my($message) = $errors->()) { say $message }

The problem it solves

Every function that returns a list meets the same pressure. One caller wants only some of the values, another wants them in a different shape, a third wants the first ten, a fourth wants them in batches of 500. The author has two ways to answer, and both cost something.

• Grow options. active_only, fields, limit, offset, as_hash, batch_size, callback. Each is code the author writes, tests, documents and keeps, and each interacts with the others: does limit count before or after active_only? The list is never finished, because the next caller wants something nobody thought of, and every caller pays, in reading the documentation, for the options they'll never use.
• Leave it to the caller. The function returns everything, and each caller writes a loop around the call: a temporary array, a counter, a flag, a last. The calling code fills with noise that has nothing to do with what it's for, the same few loops are written over and over, and the function can't stop early, because by the time the caller has the values, all the work of making them is done.

Stages are a third way. The function takes a list of them as its last arguments, and runs each value it makes through them, so the caller's processing happens inside the function, one value at a time, as the values are made:

# Options
sub users (%opt) {
    my @users = load_users();
    @users = grep { $_->{active} } @users if $opt{active_only};
    @users = map  { $_->{name} } @users   if $opt{names_only};
    @users = splice @users, 0, $opt{limit} if $opt{limit};
    return @users;
}

# Stages
sub users (@stages) {
    return Ekorn::Pipeline->new(@stages)->(load_users());
}

my @names = users(sub { $_->{active} ? $_ : () }, sub { $_->{name} }, take(10));

The second users does everything the first does, and everything it doesn't, without a line more. Its author is done with it: whatever a caller wants next, the caller can write, as a stage, without asking.

What everyone gets

• The caller says it once, beside the call. No loop, no temporary array, no counter, no flag: what the caller wants done with the values is written where they asked for them, in the order it happens, and reads as a list of steps.
• The work stops when the caller has enough. A stage that stops ends the function's work too, so a caller who wants ten results reads ten lines, fetches one page, or runs one query's worth of rows, not all of them. Nothing is held between stages, so a function over a million values needs no more memory than one over ten.
• The author keeps an interface that doesn't grow. One list of stages takes the place of every option a caller might ask for next, so the function's signature, documentation and tests stay the size of what only it can do. A new need is met in the caller's code, and breaks nobody else's.
• The author keeps what only it should touch. The file is opened and closed, the pages fetched, the connection or transaction handled, by the function, as it always was. The caller shapes the values without ever seeing the handle they came from.
• Order answers the questions options raise. Does limit count before or after active_only? With stages, the caller wrote them in an order, and that's the order they run in.
• A stage is written once, for every function. Stages are plain code references, so a stage written for one function works with every function that takes stages, whoever wrote it, and the canned stages here work with all of them.

Stages

A stage is a code reference. It's called with one value, as its argument and in $_, and returns a list:

• nothing, to drop the value: a filter;
• one value, to pass it on, changed or not: a map;
• several, each of which goes on through the rest of the stages by itself: a flatMap.

One shape covers all three, so a pipeline is just a list of stages, and a stage needs no wrapper to say which kind it is:

sub { $_ % 2 ? $_ : () }                # filter: the odd numbers
sub { $_ * 10 }                         # map: ten times each
sub ($user) { @{ $user->{roles} } }     # flatMap: each of a user's roles

Returning a list, rather than a value or undef, is what makes this work. An empty list is plainly "nothing", so undef, 0 and the empty string are values like any other, and can be passed on.

One value at a time

Each value goes through every stage before the next one starts, as it would in a for loop. So the values come out in order, a stage sees them in order, and nothing is held between stages: a pipeline over a million values needs no more memory than one over ten.

A pipeline runs two ways:

• Pushed: called with values, it returns the list that comes out of the last stage.

    my @results = $pipeline->(@values);
  

• Pulled: given a source, a function returning the next value each time it's called, iterator returns a function handing out the results one per call. The source is called only when a result is wanted, so a function reading a file, a socket or a database reads only as much as its caller asks for.

    my $next = $pipeline->iterator(sub { @queue ? shift @queue : () });
    while (my($result) = $next->()) { ... }
  

Stopping early

A stage can end the pipeline with Ekorn::Pipeline::Stop. The values already out are kept, any values the stop carries go on through the rest of the stages, and nothing else is made: once a stage has stopped it, a pulled pipeline never calls its source again.

use Ekorn::Pipeline::Stop;

sub ($user) { $user->{id} == $wanted ? Stop($user) : () }    # find one, then stop

This is what lets a caller's processing save the author's work. With take(10), the function in the example at the top reads ten error lines and however many lines lie between them, and leaves the rest of the file unread. No option for it was written, and the caller wrote no last.

It counts for more where each value costs more. A function over a paged web API fetches a page only when its caller wants a result from it:

sub search ($query, @stages) {
    my $page = 0;
    return Ekorn::Pipeline->new(@stages)->iterator(sub {
        return @{ fetch_page($query, ++$page) };    # a request; empty after the last page
    });
}

my $hits = search('squirrels', take(10));
while (my($hit) = $hits->()) { say $hit->{title} }    # one request, if a page holds ten

Without stages, that function either fetches every page, or grows max_results, max_pages and on_result options, each of which its author has to write and keep.

Stages with state, and with an end

A stage is a closure, so it can keep state, such as a count. But a plain closure keeps it from one run to the next, and can't tell when the last value has gone by. Ekorn::Pipeline::Gather makes a stage afresh at the start of each run, and lets it have an end function, called once the values run out, whose values go on through the rest of the stages:

use Ekorn::Pipeline::Gather;

sub batches_of ($size) {
    return gather {
        my @batch;    # this run's
        return (
            sub ($x) { push @batch, $x; @batch == $size ? [splice @batch] : () },
            sub ()   { @batch ? [@batch] : () },    # the last, short batch
        );
    };
}

That's how batch, collect, take and drop are made.

Stages are values

Stages are code references, so they can be kept in variables, built by functions and passed around. A function that returns a stage is a word in a vocabulary, which grows in the caller's code, not the author's:

sub only_active ()    { sub { $_->{active} ? $_ : () } }
sub value_of ($name)  { sub { $_->{$name} } }

my @page  = (drop(20), take(10));
my @names = users(only_active, value_of('name'), @page);

The same stages work with every function that takes them, so one written for users works for log_lines, or for a function on CPAN its author never heard of.

Writing a function that takes stages

• Take the stages last. Whatever arguments the function needs come first, and the stages after them, as sub log_lines ($file, @stages). A caller who wants nothing done passes none, and a pipeline with no stages returns its values as they are, so there's nothing to special-case.
• Push what you have; pull what you make. If the values are already in memory, call the pipeline with them. If they're made one at a time, from a file, a socket, a database or a computation, give iterator a source, and return the iterator, so a stage that stops saves the work of making the rest.
• Let the stages do the optional work. If you're about to add an option that filters, reshapes or limits what's returned, a stage does it already, and it's one fewer thing for you to maintain. Keep the arguments only the function can act on, such as which file to read, or a filter a database can apply with an index.
• Report your callers' mistakes at their lines. A module whose @CARP_NOT holds @Ekorn::Pipeline::Errors::INTERNAL has a stage that dies reported at the caller's line, where the stage was written, rather than at the module's own line that ran the pipeline.

A class of your own can have pipe and tap methods, from Ekorn::Pipeline::Pipes, which run an object through stages: pipe returns what comes out, and tap returns the object, so a chain of method calls can carry on after it.

A pipeline's stages go wherever stages go, so a set of stages made once can be spliced into another pipeline, or given to any function that takes stages:

my $active_names = Ekorn::Pipeline->new(sub { $_->{active} ? $_ : () }, sub { $_->{name} });
my @names = users($active_names->stages, take(10));

Errors

A stage that dies is named by its place in the pipeline, at the line of the caller's code that ran it, with its message below:

Pipeline stage #2 failed at app.pl line 12:
    No name for user 7 at app.pl line 9.

An exception object is passed on as it is. Either way the run ends, pushed or pulled: an iterator returns nothing more. A mistake in how a pipeline or a canned stage is made, such as a stage that isn't a code reference or take(-1), is reported at the caller's line too.

Canned stages

• Ekorn::Pipeline::Stages::Flow: take, drop, take_while and drop_while, which pass on values by their place or by a test.
• Ekorn::Pipeline::Stages::Batch: batch and collect, which gather values and pass them on together.
• Ekorn::Pipeline::Stages::Taps: peek, count_into and dump_to, which look at each value and pass it on, for logging, counting and debugging.

Stage modules of your own

Stages are plain code references, so anyone can make them, in any namespace. Where a module of them goes depends on what its stages need to know:

• Stages for any values go under Ekorn::Pipeline::Stages, as the canned stages do. They work whatever the values are, and wherever they come from.
• Stages for one kind of value, or one source, go with the module that knows about it, under its own namespace. DBIx::Squirrel's stages for rows, pluck and as_hash, which need a statement's column names, are in DBIx::Squirrel::Stages::Rows; stages for a log format of your own might be in My::Log::Stages.

A module of stages, wherever it goes, is built from the same parts as the canned stages: Ekorn::Pipeline::Gather for state and end functions, Ekorn::Pipeline::Stop to end early, and croakf, from Ekorn::Pipeline::Errors, to report a mistake in making a stage at the line of the caller who made it:

package My::Stages::Words;

use v5.36;
use Exporter qw(import);
use Ekorn::Pipeline::Errors qw(croakf);

our @EXPORT_OK = qw(longer_than);

sub longer_than ($length) {
    croakf 'Expected a length of 0 or more, not %s', $length // 'undef'
        unless defined $length && $length =~ m( \A \d+ \z )xa;
    return sub ($word) { length $word > $length ? $word : () };
}

When to use something else

• Code with no other caller. A loop where the values are made and used in one place, by one author, is clearer as a plain for, grep or map. Stages pay off where a function and its callers are written apart.
• Tight loops over cheap values. Each stage is a call of a code reference for each value. That's nothing beside reading a file or a socket, but it shows in a loop doing arithmetic on millions of numbers in memory.
• Work that needs every value at once, such as sorting. A stage sees one value at a time, so a sort waits for them all, as collect { sort @_ } does, and saves nothing. Where the source can sort, as a database can with ORDER BY, let it.

Where it came from

The pipeline began in DBIx::Squirrel, where stages declared beside a query shape the rows it returns. None of it needs a database, so it's here on its own, for any function to use, and to grow beyond what rows need. DBIx::Squirrel is moving onto it, in place of its own copy.

Documentation

The how-to shows how to do common tasks, one recipe at a time, for callers, function authors and stage authors. The reference describes every module, function and method, and how a pipeline behaves, in full, with notes for stage authors. Each module's POD, starting with Ekorn::Pipeline, says the same more briefly, and is the authority where they differ.

Installing

It isn't on CPAN yet. To install it from GitHub:

cpanm https://github.com/nukopian/Ekorn-Pipeline.git

It needs Perl 5.36 or later.

Licence

Copyright (c) 2024-2026 by Iain Campbell. This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself. See the file LICENSE.

Acknowledgements

Written by Iain Campbell, with Claude, Anthropic's AI model, which reviews the code, writes parts of it where Iain asks, often from his code in DBIx::Squirrel, and helps with the Git Ops, the repository's setup, and the tests and documentation.