<commit_before>package Coro::ProcessPool;

use strict;
use warnings;
use Carp;

use AnyEvent;
use Coro;
use Coro::Channel;
use Coro::Storable qw(freeze thaw);
use MIME::Base64 qw(encode_base64 decode_base64);
use Sys::Info;
use Coro::ProcessPool::Process;

our $VERSION = 0.06;

if ($^O eq 'MSWin32') {
    die 'MSWin32 is not supported';
}

use fields qw(
    max_procs
    num_procs
    max_reqs
    procs
);

sub new {
    my ($class, %param) = @_;
    my $self = fields::new($class);
    $self->{max_procs} = $param{max_procs} || _cpu_count();
    $self->{max_reqs}  = $param{max_reqs}  || 0;
    $self->{num_procs} = 0;
    $self->{procs}     = Coro::Channel->new();
    return $self;
}

sub _cpu_count {
    my $info = Sys::Info->new();
    my $cpu  = $info->device('CPU');
    return $cpu->count;
}

sub shutdown {
    my $self = shift;
    for (1 .. $self->{num_procs}) {
        my $proc = $self->{procs}->get;
        $proc->terminate;
        --$self->{num_procs};
    }
}

sub start_proc {
    my $self = shift;
    my $proc = Coro::ProcessPool::Process->new();
    $proc->spawn;
    ++$self->{num_procs};
    return $proc;
}

sub kill_proc {
    my ($self, $proc) = @_;
    $proc->terminate;
    --$self->{num_procs};
}

sub process {
    my ($self, $f, $args) = @_;
    ref $f eq 'CODE' || croak 'expected CODE ref to execute';
    $args ||= [];
    ref $args eq 'ARRAY' || croak 'expected ARRAY ref of arguments';

    my $proc;

    # Start a new process if none are available and there are worker slots open
    if ($self->{procs}->size == 0 && $self->{num_procs} < $self->{max_procs}) {
        $proc = $self->start_proc;
    }

    # Otherwise, wait for the next available process
    $proc = $self->{procs}->get unless defined $proc;

    # Restart the process if the worker is exhausted
    if ($self->{max_reqs} > 0 && $proc->{processed} >= $self->{max_reqs}) {
        $self->kill_proc($proc);
        $proc = $self->start_proc;
    }

    # Send the task
    $proc->send($f, $args);

    # Replace process in the pool as soon as result is ready on the connection
    $proc->readable;
    $self->{procs}->put($proc);

    # Collect and return the result
    return $proc->recv;
}

sub map {
    my ($self, $f, @args) = @_;

    my @results;
    my @threads;

    foreach my $i (0 .. $#args) {
        push @threads, async {
            $results[$i] = [ process(@_) ];
        } $self, $f, [$args[$i]];
    }

    $_->join foreach @threads;

    return map { @$_ } @results;
}

sub defer {
    my ($self, $f, $args) = @_;
    my $arr = wantarray;
    my $cv  = AnyEvent->condvar;

    async_pool {
        if ($arr) {
            my @results = eval { process(@_) };
            $cv->croak($@) if $@;
            $cv->send(@results);
        } else {
            my $result = eval { process(@_) };<commit_msg>Added additional error handling in process()<commit_after>package Coro::ProcessPool;

use strict;
use warnings;
use Carp;

use AnyEvent;
use Coro::Channel;
use Coro::ProcessPool::Process;
use Coro::Storable qw(freeze thaw);
use Coro;
use MIME::Base64 qw(encode_base64 decode_base64);
use Sys::Info;

our $VERSION = 0.06;

if ($^O eq 'MSWin32') {
    die 'MSWin32 is not supported';
}

use fields qw(
    max_procs
    num_procs
    max_reqs
    procs
);

sub new {
    my ($class, %param) = @_;
    my $self = fields::new($class);
    $self->{max_procs} = $param{max_procs} || _cpu_count();
    $self->{max_reqs}  = $param{max_reqs}  || 0;
    $self->{num_procs} = 0;
    $self->{procs}     = Coro::Channel->new();
    return $self;
}

sub _cpu_count {
    my $info = Sys::Info->new();
    my $cpu  = $info->device('CPU');
    return $cpu->count;
}

sub shutdown {
    my $self = shift;
    for (1 .. $self->{num_procs}) {
        my $proc = $self->{procs}->get;
        $proc->terminate;
        --$self->{num_procs};
    }
}

sub start_proc {
    my $self = shift;
    my $proc = Coro::ProcessPool::Process->new();
    $proc->spawn;
    ++$self->{num_procs};
    return $proc;
}

sub kill_proc {
    my ($self, $proc) = @_;
    $proc->terminate;
    --$self->{num_procs};
}

sub process {
    my ($self, $f, $args) = @_;
    ref $f eq 'CODE' || croak 'expected CODE ref to execute';
    $args ||= [];
    ref $args eq 'ARRAY' || croak 'expected ARRAY ref of arguments';

    my $proc;

    # Start a new process if none are available and there are worker slots open
    if ($self->{procs}->size == 0 && $self->{num_procs} < $self->{max_procs}) {
        $proc = $self->start_proc;
    }

    # Otherwise, wait for the next available process
    $proc = $self->{procs}->get unless defined $proc;

    # Restart the process if the worker is exhausted
    if ($self->{max_reqs} > 0 && $proc->{processed} >= $self->{max_reqs}) {
        $self->kill_proc($proc);
        $proc = $self->start_proc;
    }

    # Send the task
    eval { $proc->send($f, $args) };

    if ($@) {
        # In the event of an error communicating with the process, assume the
        # process is in an error state and replace it. Once the new process is
        # returned to the pool, croak with the original error message.
        my $error = $@;
        $self->kill_proc($proc);
        $proc = $self->start_proc;
        $self->{procs}->put($proc);
        croak $error;
    } else {
        # Replace process in the pool as soon as result is ready on the connection
        $proc->readable;
        $self->{procs}->put($proc);

        # Collect and return the result
        return $proc->recv;
    }
}

sub map {
    my ($self, $f, @args) = @_;

    my @results;
    my @threads;

    foreach my $i (0 .. $#args) {
        push @threads, async {
            $results[$i] = [ process(@_) ];
        } $self, $f, [$args[$i]];
    }

    $_->join foreach @threads;

    return map { @$_ } @results;
}

sub defer {
    my ($self, $f, $args) = @_;
    my $arr = wantarray;
    my $cv  = AnyEvent->condvar;

    async_pool {
        if ($arr) {
            my @results = eval { process(@_) };
            $cv->croak($@) if $@;
            $cv->send(@results);
        } else {
            my $result = eval { process(@_) };
{'fraction_non_alphanumeric': 0.17256564997553417, 'fraction_numerical': 0.006198010112542815, 'mean_word_length': 2.433370660694289, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 89, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '81915', 'n_tokens_mistral': 2243, 'n_tokens_neox': 2170, 'n_words': 590}