Есть ли многопроцессорный модуль для Perl?

Существует ли многопроцессорный модуль для Perl? Что-то, что имеет сходную функциональность с тем, что предлагает Python модуль многопроцессорности.

Я понимаю, что я мог бы создать аналогичную функциональность с помощью Perl, но я ищу что-то уже реализованное.

Ответы

Ответ 1

Я думаю, Forks::Super близок. Он имеет несколько функций для запуска произвольной подпрограммы (или внешней команды) в фоновом режиме, мониторинга и сигнализации фонового процесса, а также для межпроцессного взаимодействия немного менее болезненным.

use Forks::Super;

sub do_something { my @args = @_; ... }
$process = fork { sub => \&do_something, args => [@args] };
$process->wait;


my $x = 42;
my @y = ();
my %z = ();
sub do_something_else {
    $x = 19;
    @y = qw(foo bar);
    %z = (foo => 'bar');
}
$process = fork { sub => 'do_something_else', share => [\$x, \@y, \%z ] };
$process->wait;
# $x, @y, and %z are now updated with changes made in background process


# create your own pipes to use for IPC
use IO::Handle;
pipe my $child_read, my $parent_write;
pipe my $parent_read, my $child_write;
$parent_write->autoflush(1);
$child_write->autoflush(1);
sub square {
    while (my $x = <$child_read>) {
        print {$child_write} $x ** 2, "\n";
    }
    close $child_write;
}
$process = fork { sub => 'square' };
print {$parent_write} "9\n";
my $result = <$parent_read>;    # should be "81\n";
close $parent_write;

# or use the standard I/O handles for IPC
sub square_root {
    sleep 1 && seek STDIN,0,1 while eof(STDIN); # ok, this is a workaround for an existing bug :-(
    while (my $x = <STDIN>) {
        print sqrt($x), "\n";
    }
}
$process = fork { sub => 'square_root', child_fh => 'in,out,block' };
$process->write_stdin("81\n");
$result = $process->read_stdout(); #  =>  "9\n"

Оба модуля multiprocessing и Forks::Super имеют множество функций. Какие именно вас интересуют?

Я являюсь автором Forks::Super, и моя цель - включить любые функции для параллельной обработки, которые люди находят полезными, поэтому, если в Perl есть функция в multiprocessing, которую вы хотите на Perl, сообщите мне.

Ответ 2

forks предоставляет такой же удивительный интерфейс, как threads, но использует процессы вместо потоков.

use forks;  # Or: use threads;
use Thread::Queue;

my $q = Thread::Queue->new();

my @workers;
for (1..NUM_WORKERS) {
   push @workers, async {
      while (defined(my $job = $q->dequeue())) {
         ...
      }
   };
}

$q->enqueue(...);

$q->enqueue(undef) for @workers;
$_->join for @workers;

Сравнение вилок с Forks:: Super.

Имейте в виду, что это предположительно случаи, когда Forks:: Super превосходит!

use Forks::Super;
sub do_something { my @args = @_; ... }
$process = fork { sub => \&do_something, args => [@args] };
$process->wait;

может быть записано как

use forks;
sub do_something { my @args = @_; ... }
$process = async { do_something(@args) };
$process->join;

---

use Forks::Super;
my $x = 42;
my @y = ();
my %z = ();
sub do_something_else {
    $x = 19;
    @y = qw(foo bar);
    %z = (foo => 'bar');
}
$process = fork { sub => 'do_something_else', share => [\$x, \@y, \%z ] };
$process->wait;

может быть записано как

use forks;
use forks::shared;
my $x :shared = 42;
my @y :shared = ();
my %z :shared = ();
sub do_something_else {
    $x = 19;
    @y = qw(foo bar);
    %z = (foo => 'bar');
}
$process = async { do_something_else() };
$process->join;

---

use Forks::Super;
use IO::Handle;
pipe my $child_read, my $parent_write;
pipe my $parent_read, my $child_write;
$parent_write->autoflush(1);
$child_write->autoflush(1);
sub square {
    while (my $x = <$child_read>) {
        chomp($x);
        print {$child_write} $x ** 2, "\n";
    }
    close $child_write;
}
$process = fork { sub => 'square' };
print { $parent_write } "9\n";
chomp( my $result = <$parent_read> );  # 81
close $parent_write;
$process->wait;

может быть записано как

use forks;
use Threads::Queue;
my $req = Threads::Queue->new();
my $resp = Threads::Queue->new();
sub square { $_[0] ** 2 }
$process = async {
    while (defined(my $x = $req->dequeue())) {
        $resp->enqueue( square($x) );
    }
};
$req->enqueue(9);
my $result = $resp->dequeue();  # 81
$resp->enqueue(undef);
$process->join;

---

use Forks::Super;
sub square_root {
    sleep 1 && seek STDIN,0,1 while eof(STDIN); # ok, this is a workaround for an existing bug :-(
    while (my $x = <STDIN>) {
        chomp($x);
        print sqrt($x), "\n";
    }
}
$process = fork { sub => 'square_root', child_fh => 'in,out,block' };
$process->write_stdin("81\n");
chomp( $result = $process->read_stdout() );  # 9
$process->close_fh('stdin');
$process->wait;

может быть записано как

use forks;
use Threads::Queue;
my $req = Threads::Queue->new();
my $resp = Threads::Queue->new();
$process = async {
    while (defined(my $x = $req->dequeue())) {
        $resp->enqueue( sqrt($x) );
    }
};
$req->enqueue(81);
my $result = $resp->dequeue();  # 9
$resp->enqueue(undef);
$process->join;

Ответ 3

Как насчет POE: Perl Object Environment? Он поддерживает асинхронные дочерние процессы.