For mandy

pp_with_pipe.pl

#!/usr/bin/env perl

use strict;
use warnings;

use Data::Dump qw(dump);
use IO::Pipe;
use Parallel::Prefork;
use Time::HiRes;

my @servers = map {sprintf('svr%03.d', $_)} (1..105);

my $job_id = 0;
my $workers = 10;
my @pipes;

my $pp = Parallel::Prefork->new(+{
    max_workers => $workers,
    err_respawn_interval => 10,
    trap_signals => +{
        HUP => 'TERM',
        TERM => 'TERM',
    },
    before_fork => sub {
        my $pp = shift;

        $pp->{job_id} = $job_id;
        $pp->{pipes} ||= [];

        if ($job_id < $workers) {
            $pipes[$job_id] = IO::Pipe->new;;
            $job_id++;
        }
    }
});

my $signal_trapped = 0;
$SIG{USR1} = sub {
    $signal_trapped++;
    # warn "USR1 $signal_trapped";
    if ($signal_trapped >= $workers) {
        kill TERM => $$;
    }
};

while ($pp->signal_received ne 'TERM') {
    $pp->start(
        sub {
            local $SIG{TERM} = 'IGNORE';
            kill USR1 => $pp->manager_pid;

            my $job_id = $pp->{job_id};

            if ($job_id >= $workers) {
                warn "Do not spawn";
                return -1;
            }

            my $pipe = $pipes[$job_id];
            $pipe->writer;

            for (my $i = $job_id; $i < scalar(@servers); $i += $workers) {
                my $server = $servers[$i];
                print $pipe "$server ($$)\n";
                Time::HiRes::sleep(rand(0.1));
            }

            close($pipe);
            return 0;
        }
    );
}

$pp->wait_all_children;

for my $pipe (@pipes) {
    $pipe->reader;
    while (<$pipe>) {
        print $_;
    }
}