$water = {
map
{
    my $process = $_;
    my @inputs = @ {$processes{$process}[1]};
    $_, {
	prerequisites => [ @inputs ],
	oo_output =>
	    sub
	    {
		my ($data) = shift;
		my ($pid, @fh, $active);
		@fh = map { $data->get($_) } @inputs;
                push @fh, \*STDIN if ((defined $start) and $start eq $process);
                $active = scalar @fh;
		local (*IN, *OUT, $/, $\);
		$/ = 1;
		$\ = '';
		open2 \*OUT, \*IN, $processes{$process}[0] or
		    die "Can't open2 \"${\($processes{$process}[0])}\": $!, $?, $^E; stopped"; ##
		select(IN) or die "Can't select IN: $!, $?, $^E; stopped"; ##
		for (;;)
		{
		    my $piece = join '', map { my $x; (eof $_) ? "\0" : (defined ($x = <$_>)
							      ? $x : (
								      die "Can't read: ", ##
								      "$!, $?, $^E; stopped"
								      )) } @fh;
		    (grep { not eof $_ } @fh) ? (print($piece)
						 or die "Can't print: $!, $?, $^E; stopped") ##
			: (last);
		}
		close (IN) or die "Can't close IN: $!, $?, $^E; stopped"; ##
		return \*OUT;
            } } } keys %processes };

tie %river, 'Data::Flow', $water;
$SIG{CHLD} = sub { wait }; ##
{
    my $output = $river{$final};
    local $/ = undef;
    local $_;
    defined($_ = <$output>) or die "Can't read: $!, $?, $^E; stopped"; ##
    print or die "Can't print: $!, $?, $^E; stopped"; ##
}
exit 0;