使用Parallel::ForkManager进行命令的并行执行

Question

使用Parallel::ForkManager进行命令的并行执行

3

我想知道我对以下脚本/逻辑的理解是否正确。

我有一些节点的列表，需要在每个节点上运行特定的命令，通过利用我拥有的服务器数量来并行进行SSH到这些节点。我的 node_list.txt 文件包含节点列表：

node1
node2
.
.
node49
node50

我已经在一个数组@hosts中定义了若干个服务器，我需要通过SSH连接到每个节点并且执行命令。我会将node_file.txt文件分成等份（称为$node_list_X.txt），分别存储在可用的服务器上。

一旦我有了这些文件(node_list_1.txt,node_list_2.txt,node_list_3.txt,node_list_4.txt)，我将登录到已经定义好的每个服务器，并通过并行传递node_list_X.txt文件，在每个主机上执行某些命令。

为了并行执行操作，我使用了Parallel::ForkManager Perl模块。

假设在每个主机上 -

192.168.0.1 -> node_list_1.txt (13 nodes)
192.168.0.2 -> node_list_2.txt (13 nodes)
192.168.0.3 -> node_list_3.txt (12 nodes)
192.168.0.4 -> node_list_4.txt (12 nodes)

将同时运行。

以下是脚本:

...
my @hosts = ("192.168.0.1", "192.168.0.2", "192.168.0.3","192.168.0.4");

open(my $node_fh, '<', $node_file)
        or die "can't open $node_file: $!";

my @lines =  <$node_fh>;

my %Files;

my $num_buckets = scalar @hosts;

my $per_bucket = int( @lines / $num_buckets );
my $num_extras =      @lines % $num_buckets;
my $path = "/home/user/vinod/test/";

for my $bucket_num (0..$num_buckets-1) {
   my $num_lines = $per_bucket;
   if ($num_extras) {
      ++$num_lines;
      --$num_extras;
   }

   last if($num_lines == 0);
   my $qfn = $path."node_list_${bucket_num}.txt";
   open(my $fh, '>', $qfn)
      or die("Can't create \"$qfn\": $!\n");

   $fh->print(splice(@lines, 0, $num_lines));
   $Files{$bucket_num} = $qfn;
}
print Dumper(\%Files);

my $command = #"defining my command here";

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

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

        $command_to_execute = $command." ".$Files{$n};
        $ssh = SSH_Connection( $hosts[$n-1], "user", "password" );
        $result = $ssh->capture($command_to_execute);
      
        $pm->finish;       
    }
}
$pm->wait_all_children;
undef $ssh;

#SSH Connect
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;
}

这里一切正常。

当我定义$pm对象时，并行进程设置为5。

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

这是否意味着在特定服务器（例如：192.168.0.1）中，它应该运行5个并行进程。这意味着它应该从node_list_1.txt （共13个节点）文件中获取5个节点，并执行命令？

如果我理解不正确，那么有什么可能的解决方案可以使用多线程在每个服务器上并行运行命令？

- vkk05

3个回答

2

如果你想在许多不同的服务器上运行作业，请考虑使用适当的作业队列。Perl的Minion非常好用。各种服务器可以连接到它，以各种方式请求作业，并发送结果回来。

- brian d foy

看起来很有趣。请问我可以在哪里获取更多关于“Minion”的信息/示例？ - vkk05

2

您是否考虑使用Net::OpenSSH::Parallel？

据我所知，它直接支持您想要做的事情，并且能够处理大量并行连接、安排任务、处理错误和重试失败的命令等。

更新: 但它是否允许我在每个主机内并行运行作业？

但是，您真正想做什么呢？将工作分配给一组工作者吗？在这种情况下，brian d foy的解决方案可能是更好的选择。

无论如何，Net::OpenSSH::Parallel从未旨在实现这一点，但它确实可以做到：

my @hosts = ...;
my @tasks = ...;
my $n_workers = 5;

my $ossh = Net::OpenSSH::Parallel->new;

for my $host (@hosts) {
  for my $ix (0..$n_workers) {
    $ossh->add_host("$host-$ix", host => $host);
  }
}

my $fetch_task = sub {
  my ($pssh, $label) = @_;
  if (defined (my $task = shift @tasks)) {
    $ossh->push($label, cmd => $task);
    $ossh->push($label, sub => $fetch_task);
  }
}

$ossh->push('*', sub => $fetch_task)

$ossh->run

- salva

当然可以。但它是否允许我在每个主机内并行运行作业？ - vkk05

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- ikegami · Accepted Answer

这是否意味着在特定服务器（例如：192.168.0.1）中，它应该运行5个并发进程？

不是的。P::FM对服务器一无所知。它管理进程，“->new(5)”表示如果其中5个进程仍在执行，则“->start”将等待其创建的一个进程完成后再创建一个新进程。

如何通过多线程并行地在每个服务器上运行命令?

假设你的意思是一般的多任务而不是具体的多线程（因为你没有使用线程），可以按以下方式为每个主机创建一个进程：

my %children;
my $error = 0;
for my $host (@hosts) {
    my $pid = fork();
    if (!defined($pid)) {
       warn("Can't execute on $host: Can't fork: $!\n");
       next;
    }

    if ($pid) {
       ++$children{$pid};
       next;
    }

    if (!eval {
       do_it($host);
       return 1;  # No exception
    }) {
       warn("Error executing commands on $host: $@");
    }
}

while (%children) {
   ( my $pid = wait() ) >= 0
      or die("Can't wait: $!\n");

   delete($children{$pid});   
}