Parallel::ForkManager and aggregate results

pf_with_data_structure.pl

#!/usr/bin/env perl

use strict;
use warnings;

use Data::Dump qw(dump);
use Parallel::ForkManager 0.7.6;

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

my $pm = Parallel::ForkManager->new(10);
$pm->run_on_finish(sub {
    my ($pid, $exit_status, $ident, $exit_signal, $core_dump, $data_ref) = @_;
    $results{$ident} = $data_ref;
});

for my $svr (@servers) {
    my $pid = $pm->start($svr) and next;
    $pm->finish(0, +{ pid => $$, val => rand(1) });
}

$pm->wait_all_children;

dump(\%results);

pf_with_sqlite.pl

#!/usr/bin/env perl

use strict;
use warnings;

use Data::Dump qw(dump);
use DBIx::Connector;
use File::Temp qw(tempfile);
use Parallel::ForkManager 0.7.6;

my @servers = map { sprintf("svr%03d", $_) } (1 .. 1000);
my (undef, $temp_db) = tempfile("mandyXXXXX", UNLINK => 1, TMPDIR => 1, OPEN => 0);

my $conn = DBIx::Connector->new(
    "dbi:SQLite:db=$temp_db",
    "", "",
    +{
        AutoCommit                       => 0,
        RaiseError                       => 1,
        ShowErrorStatement               => 1,
        AutoInactiveDestroy              => 1,
        sqlite_use_immediate_transaction => 1,
    }
) or die($DBI::errstr);

$conn->txn(fixup => sub {
    my $dbh = shift;
    $dbh->do(<< 'SQL');
CREATE TABLE store_results(
  svr TEXT not null,
  result TEXT not null
)
SQL
    $dbh->commit;
});

my $pm = Parallel::ForkManager->new(10);

for my $svr (@servers) {
    my $pid = $pm->start and next;

    eval {
        $conn->txn(
            fixup => sub {
                my $dbh = shift;
                $dbh->do("INSERT INTO store_results(svr, result) VALUES(?, ?)", undef, $svr, sprintf("%03.3f %d", rand(1), $$));
                $dbh->commit;
            },
        );
    };
    if (my $e = $@) {
        warn $e;
    }

    $pm->finish;
}

$pm->wait_all_children;

my $rs = $conn->dbh->selectcol_arrayref("SELECT * FROM store_results", +{ Columns => [1,2] });
my %results = @$rs;

dump(\%results);

$conn->disconnect;