将结果数据追加到Parallel :: ForkManager Perl中的标量变量中

Question

我有一个代码，该代码可以正常工作。但是我很难将每个已执行命令的输出存储在$result = $ssh->capture($command_to_execute);中。它使用Parallel::ForkManager模块通过将不同的文件用作输入参数来在不同的主机中运行命令。

一旦命令执行完成，我希望结果输出应存储在$result变量中。它应该将每个主机结果附加到相同的变量中，最后我要处理$result中的值。我正在使用.=将结果数据附加到$result，但它确实起作用了。

在此处粘贴我的代码以供参考：

.
.
.
my $result;
my $pm = Parallel::ForkManager->new(5);

DATA_LOOP:
foreach my $n (1..$num_buckets) {
        my $pid = $pm->start and next DATA_LOOP;

        $command_to_execute = $my_command." ".$Files{$n};
        my $ssh = SSH_Connection( $list_of_ips[$n-1], 'username', 'pass' );
        $result = $ssh->capture($command_to_execute);
        $result .= "Result from File:$Files{$n} and Host:$list_of_ips[$n-1] is $result\n"; 
        print "Result: INSIDE: $result";
        $pm->finish;
}
$pm->wait_all_children;
print "Result: OUTSIDE: $result";
print "Done\n";

sub SSH_Connection {
    my ( $host, $user, $passwd ) = @_;
    my $ssh = Net::OpenSSH->new($host,
                                user => $user,
                                password => $passwd,
                                master_opts => [-o => "StrictHostKeyChecking=no"]
    );
    $ssh->error and die "Couldn't establish SSH connection: ". $ssh->error;

    return $ssh;
}

print "Result: INSIDE: $result";可以一张一张打印结果。但是print "Result: OUTSIDE: $result"; 为空，实际上应该具有从$results循环内部获取的for的组合结果。

Answer 1

如Parallel::ForkManager的文档中所示，要从子级获取结果，您需要提供对该结果的引用作为finish的另一个参数。

$pm->finish(0, [$Files{$n}, $list_of_ips[$n-1], $result]);

使用run_on_finish收集结果：

my $result;
$pm->run_on_finish( sub {
    my ($pid, $exit_code, $ident, $exit_signal, $core_dump, $single_result) = @_;
    $result .= "Result from File: $single_result->[0] and Host: $single_result->[1]"
             . " is $single_result->[2]\n"; 

Answer 2

[每次运行$pm->start，您正在分叉一个新进程来运行代码，直到$pm->finish。除了Parallel :: ForkManager提供的将数据发送回父级的机制之外，此分叉的进程不会以任何方式影响父级进程。此机制在https://metacpan.org/pod/Parallel::ForkManager#RETRIEVING-DATASTRUCTURES-from-child-processes中进行了描述。

$pm->run_on_finish(sub {
  my ($pid, $exit_code, $ident, $exit_signal, $core_dump, $data) = @_;
  my $result = $$data;
  ...
});

DATA_LOOP:
foreach my $n (1..$num_buckets) {
        my $pid = $pm->start and next DATA_LOOP;
        ...
        $pm->finish(0, \$result);
}

实际上，如果您愿意进行一些重组，则不需要分叉。 Net :: OpenSSH可以提供可以由事件循环（例如IO::Async::Loop）同时管理的命令，因此所有Perl操作都将在同一进程中发生（但不一定按它们出现的顺序）。由于IO::Async::Loop->run_process返回Future，因此Future::Utils提供了一种管理这些命令的并发性的方法。

use strict;
use warnings;
use Net::OpenSSH;
use IO::Async::Loop;
use Future::Utils 'fmap_concat';

my $loop = IO::Async::Loop->new;

my $future = fmap_concat {
  my $n = shift;
  ...
  my $remote_command = $ssh->make_remote_command($command_to_execute);
  return $loop->run_process(command => $remote_command, capture => ['stdout'])
    ->transform(done => sub { "Result from File:$Files{$n} and Host:$list_of_ips[$n-1] is $_[0]\n"; });
} foreach => [1..$num_buckets], concurrent => 5;

my @results = $future->get;

[个人和整体（由fmap返回）Future的管理方式具有很大的灵活性，但是默认情况下，执行流程的任何失败都会导致整个Future立即失败（导致get抛出）异常），则任何非零出口都将被忽略。