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

Posted 2023-02-16

【中文标题】将结果数据附加到 Parallel::ForkManager Perl 中的标量变量

【英文标题】：Append resulted data to scalar variable in Parallel::ForkManager Perl

【发布时间】：2019-12-27 17:25:06

【问题描述】：

我有一个代码，它按预期工作。但是我很难将每个已执行命令的输出存储在$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 循环内部获取的。

【问题讨论】：

外部$result指的是父进程的$result而不是子进程$result。我认为您需要使用 run_on_finish()

将结果传递回父级

请参阅this post 以获得答案（例如）

【参考方案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"; 

【参考方案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 返回）Futures 的管理方式有很大的灵活性，但默认情况下，任何执行流程的失败都会导致整个 Future 立即失败（导致 get 到抛出异常）并且任何非零退出都将被忽略。