perl - 如何等待子进程在父进程中设置变量？

Question

 use Parallel::ForkManager;    
 my $number_running = 0;
 my $pm = new Parallel::ForkManager(30); 
 $pm->run_on_start( sub { ++$number_running; } );
 $pm->run_on_finish( sub { --$number_running; } );
 for (my $i=0; $i<=100; $i++)
 {
     if ($number_running == 5) { while ($number_running > 0) {} }  # waits forever
     $pm->start and next;
     print $i;
     $pm->finish;
 }

以上代码使用 Parallel::ForkManager使用并行进程在 for 循环中执行代码。它正在计算有多少子进程正在运行并设置 $number_running相应地变化。一旦 5 个子进程正在运行，我希望它等到 0 个子进程正在运行后再继续。

for 循环中的第一行旨在实现此目的，但它会永远等待该行。这就像子进程对变量所做的更改不适用于该代码行。我究竟做错了什么？注意:我知道 wait_all_children但我不想使用它。

Answer 1

短 回调 run_on_finish通常不会触发每个 child 的退出，所以 $number_running不会减少，因此它无法控制循环。解决此问题的方法:

使用 reap_finished_children为了在个别 child 退出时进行交流，以便 run_on_finish当每个 child 退出时，确实可以运行

使用 wait_for_available_procs等待整个批处理完成后再开始一个新批处理

至于标题，子进程不能在父进程中设置任何内容(父进程也不能在子进程中设置)。他们必须按照上面针对本模块中的此计划概述的方式进行通信以执行操作。

回调 run_on_start与每个新进程一起运行并且计数器递增。但是回调 run_on_finish永远不会触发，因此计数器永远不会递减。因此一旦达到 5代码位于 while环形。请注意， parent 和 child 是独立的进程，因此不知道彼此的变量并且无法更改它们。
回调 run_on_finish通常由 wait_all_children 触发在所有进程被 fork 之后。它的工作也完成了
当最大数量的进程运行并且一个退出时。这是在 start 中完成的调用 wait_one_child (调用 on_finish ，见下文)。
或者，这可以通过调用 reap_finished_children 随意完成。方法

This is a non-blocking call to reap children and execute callbacks independent of calls tostart or wait_all_children. Use this in scenarios where start is called infrequently but you would like the callbacks executed quickly.

这解决了 的主要问题。个别 child 退出时如何沟通 (如评论中所述)，而不是 wait_all_children .
这是一个如何使用它的示例，以便回调在子退出时正确运行。大量代码仅用于诊断(打印)。

use warnings;
use strict;
use feature 'say';
use Parallel::ForkManager;    
$| = 1;

my $total_to_process = 3;  # only a few for this test
my $number_running   = 0;    
my @ds;

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

$pm->run_on_start( sub {
    ++$number_running;
    say "Started $_[0], total: $number_running";
});
$pm->run_on_finish( sub {
    --$number_running;
    my ($pid, $code, $iden, $sig, $dump, $rdata) = @_;
    push @ds, "gone-$pid";
    say "Cleared $pid, ", ($rdata->[0] // ''), ($code ? " exit $code" : '');
});

foreach my $i (1 .. $total_to_process)
{
    $pm->start and next;
    run_job($i);
    $pm->finish(10*$i, [ "kid #$i" ]);
}
say "Running: ", map { "$_ " } $pm->running_procs;  # pid's of children

# Reap right as each process exits, retrieve and print info
my $curr = $pm->running_procs;
while ($pm->running_procs) 
{
    $pm->reap_finished_children;    # may be fewer now
    if ($pm->running_procs < $curr) {
        $curr = $pm->running_procs;
        say "Remains: $number_running. Data: @ds";
    }
    sleep 1;  # or use Time::HiRes::sleep 0.1;
}

sub run_job {
    my ($num) = @_;
    my $sleep_time = ($num == 1) ? 1 : ($num == 2 ? 10 : 20);
    sleep $sleep_time;
    say "\tKid #$num slept for $sleep_time, exiting";
}

使用该方法相当于调用 waitpid -1, POSIX::WNOHANG在 fork 之后的循环中.这比最大 ( 30 ) 进程 fork 更少，以便更容易地查看输出并证明回调在子退出时正确运行。更改这些数字以查看其完整操作。
子进程以 10*$i 退出，以便能够跟踪输出中的子进程。匿名数组中返回的数据 [...]是标识子进程的字符串。尽快 reap_finished_children通话完成 $number_running在回调中减少。这就是拥有 $curr 的原因。变量，再次用于诊断。
这打印

start: Started 4656, running: 1start: Started 4657, running: 2start: Started 4658, running: 3Running: 4656 4658 4657         Kid #1 slept for 1, exitingCleared 4656, kid #1 exit 10Remains: 2. Data: gone-4656        Kid #2 slept for 10, exitingCleared 4657, kid #2 exit 20Remains: 1. Data: gone-4656 gone-4657        Kid #3 slept for 20, exitingCleared 4658, kid #3 exit 30Remains: 0. Data: gone-4656 gone-4657 gone-4658

---

The direct question is of how to wait for the whole batch to finish before starting a new one. This can be done directly by wait_for_available_procs($n)

Wait until $n available process slots are available. If $n is not given, defaults to 1.

If $MAX is used for $n, that many slots will become available only once the whole batch completed. What to use for $n can also be decided at runtime.

---

Some details of module's operation

When a child exits the SIGCHLD signal is sent to the parent, which it must catch in order to know that the child is gone (and to avoid zombies, in the first place). This is done by using wait or waitpid, in code or in the SIGCHLD handler (but only at one place). See fork, Signals in perlipc, waitpid and wait.

We see from P::FM's source that this is done in wait_one_child (via _waitpid sub)

sub wait_one_child { my ($s,$par)=@_;  

  my $kid;
  while (1) {
    $kid = $s->_waitpid(-1,$par||=0);
    last if $kid == 0 || $kid == -1; # AS 5.6/Win32 returns negative PIDs
    redo if !exists $s->{processes}->{$kid};
    my $id = delete $s->{processes}->{$kid};
    $s->on_finish( $kid, $? >> 8 , $id, $? & 0x7f, $? & 0x80 ? 1 : 0);
    last;
  }
  $kid;
};  

用于 wait_all_children

sub wait_all_children { my ($s)=@_;

  while (keys %{ $s->{processes} }) {
    $s->on_wait;
    $s->wait_one_child(defined $s->{on_wait_period} ? &WNOHANG : undef);
  };
}

方法 reap_finished_children上面使用的是此方法的同义词。
方法 wait_one_child start 使用获取信号的当最大进程数被填满并且一个退出时，获取子进程。这就是模块知道何时可以启动另一个进程并尊重其最大值的方式。 (它也被其他一些等待进程的例程使用。
)。这是 run_on_finish被 $s->on_finish( $kid, ... ) 触发

sub on_finish {

  my ($s,$pid,@par)=@_;
  my $code=$s->{on_finish}->{$pid} || $s->{on_finish}->{0} or return 0;
  $code->($pid,@par);
};

回调在 coderef $code 中, 从对象的 on_finish 中检索 key ，它本身在子 run_on_finish 中设置.一旦该子运行，这就是设置回调的方式。
用户可用的方法是 wait_all_children和 reap_finished_children .
由于发布的代码中没有使用这些， $number_running没有更新所以 while是一个无限循环。回想一下变量 $number_running在父进程中不能被子进程直接更改。