Perlのforkで子プロセスの同時実行数を制限した並列処理

ゴールデンウィークとお遊び用CentOS7を手に入れたのでお遊びが進む君。
ということで、perlのforkで並列処理をやってみました。並列稼動する子プロセスの数を一定数に制限しながら指定数分の処理を行います。実際にも、1つだけや無制限に子プロを作るというようなことはなく、並列数を一定にして処理をさせることはよくあると思います。
例えば、100個のRSSフィードを取得したい場合、
・100個のフィードを10個づつ10個のフィード取得処理に分割
・子プロの最大同時並列数を5に限定
・1つの子プロで１つのフィード取得処理を実行
・親プロセスは、1つの子プロが終了したら新しい子プロ作って次のフィード取得処理用子プロを作成
・親プロセスは、全10フィード取得処理が完了するまで上記を繰り返し
終了判定の作り方次第で、複数サーバでの同時実行も比較的簡単にできると思います。

で、サンプルソース。Perlだと親プロセスと子プロセスの処理を同じソース内に書いてるのをよく見るのですがメンテナンスがやりにくくなるのと単体テストも難しくなるのでソースを分けています。

親プロセスのソース

fork_p.pl。子プロの制御と処理の終了判定をしています。

01: 02: 03: 04: 05: 06: 07: 08: 09: 10: 11: 12: 13: 14: 15: 16: 17: 18: 19: 20: 21: 22: 23: 24: 25: 26: 27: 28: 29: 30: 31: 32: 33: 34: 35: 36: 37: 38: 39: 40: 41: 42: 43: 44: 45: 46: 47: 48: 49: 50: 51: 52: 53: 54:

#!/usr/bin/perl use strict; use POSIX ":sys_wait_h"; my $TOTAL_PROC = 10; my $MAX_CHILD = 3; my %done = (); my $running_proc = 0; my $no = 1; print "P:[$$] start parent\n"; sub run_child { my $no = shift; print "P:[$$] run_child $no\n"; my $waittime = rand(3); exec("./fork_c.pl $no $waittime"); exit(); } my $running = 0; while ($no <= $TOTAL_PROC) { print "P:[$$] no=$no running=$running done=" .keys(%done). "\n"; my $can_run_cnt = $MAX_CHILD - $running; if ($can_run_cnt > $TOTAL_PROC - keys(%done)) { $can_run_cnt = $TOTAL_PROC - keys(%done); } for (my $i = 0; $i < $can_run_cnt; $i++) { my $pid = fork(); print "P:[$$] fork pid=$pid\n"; if ($pid) { # parent print "P:[$$] parent proc\n"; $running++; $no++; } elsif ($pid == 0) { # child print "P:[$$] start child proc\n"; $done{$no} = $pid; run_child($no); } else { print "*** fork error\n"; exit(); } } my $finished_child = waitpid(-1, 0); $running--; print "P:[$$] finished_child=$finished_child\n"; } print "P:[$$] finished parent\n";

並列実行数の制御のためにwaitpidを利用していますが、実際には最後に書いているように処理が正常・失敗の何かを子プロで吐いて、親プロセスはそれをチェックさせたほうがいいかと思われます。waitpidでの子プロの状態判定は実はややこしいので。
また、waitpid(-1, 0)は、いずれかの子プロの終了を待ちますが、WNOHANGを指定して子プロの状態変化を待つことなく親プロセスの処理を続行させることもできます。

子プロセスのソース

fork_c.pl。実際に並列処理させたいロジックを定義します。サンプルは何もしないので親プロセスから渡された分だけsleepします。

01: 02: 03: 04: 05: 06: 07: 08:

#!/usr/bin/perl my $no = $ARGV[0]; my $waittime = $ARGV[1]; print "C:[$$] start $no. sleep $waittime\n"; sleep($waittime); print "C:[$$] finished $no.\n";

実行結果

[user01@pc02 perl]$ ./fork_p.pl 
P:[5061] start parent
P:[5061] no=1 running=0 done=0
P:[5061] fork pid=5062
P:[5061] parent proc
P:[5061] fork pid=5063
P:[5061] parent proc
P:[5061] fork pid=5064
P:[5061] parent proc
P:[5062] fork pid=0
P:[5062] start child proc
P:[5062] run_child 1
C:[5062] start 1. sleep 1.69095524981368
P:[5063] fork pid=0
P:[5063] start child proc
P:[5063] run_child 2
C:[5063] start 2. sleep 0.0821911322300473
C:[5063] finished 2.
P:[5061] finished_child=5063
P:[5061] no=4 running=2 done=0
P:[5061] fork pid=5065
P:[5061] parent proc
P:[5064] fork pid=0
P:[5064] start child proc
P:[5064] run_child 3
C:[5064] start 3. sleep 1.76854459753941
P:[5065] fork pid=0
P:[5065] start child proc
P:[5065] run_child 4
C:[5065] start 4. sleep 2.15744215830911
C:[5062] finished 1.
P:[5061] finished_child=5062
P:[5061] no=5 running=2 done=0
P:[5061] fork pid=5066
P:[5061] parent proc
P:[5066] fork pid=0
P:[5066] start child proc
P:[5066] run_child 5
C:[5066] start 5. sleep 2.0960794109733
C:[5064] finished 3.
P:[5061] finished_child=5064
P:[5061] no=6 running=2 done=0
P:[5061] fork pid=5067
P:[5061] parent proc
P:[5067] fork pid=0
P:[5067] start child proc
P:[5067] run_child 6
C:[5067] start 6. sleep 0.725568342243616
C:[5067] finished 6.
P:[5061] finished_child=5067
P:[5061] no=7 running=2 done=0
P:[5061] fork pid=5068
P:[5061] parent proc
P:[5068] fork pid=0
P:[5068] start child proc
P:[5068] run_child 7
C:[5068] start 7. sleep 0.0705134402842411
C:[5068] finished 7.
P:[5061] finished_child=5068
P:[5061] no=8 running=2 done=0
P:[5061] fork pid=5069
P:[5061] parent proc
P:[5069] fork pid=0
P:[5069] start child proc
P:[5069] run_child 8
C:[5069] start 8. sleep 0.558187886129428
C:[5069] finished 8.
P:[5061] finished_child=5069
P:[5061] no=9 running=2 done=0
P:[5061] fork pid=5070
P:[5061] parent proc
P:[5070] fork pid=0
P:[5070] start child proc
P:[5070] run_child 9
C:[5070] start 9. sleep 1.29167993807915
C:[5065] finished 4.
P:[5061] finished_child=5065
P:[5061] no=10 running=2 done=0
P:[5061] fork pid=5071
P:[5061] parent proc
P:[5071] fork pid=0
P:[5071] start child proc
P:[5071] run_child 10
C:[5071] start 10. sleep 0.447920087945942
C:[5071] finished 10.
P:[5061] finished_child=5071
P:[5061] finished parent
[user01@pc02 perl]$ C:[5070] finished 9.
C:[5066] finished 5.

見ての通り、最後の終了判定がイマイチなので子プロが生きているのに親プロが死んでいます。waitpidで子プロの終了を待ってますが、実際には子プロが終了時に、OK/NGの何かを返す/doneファイルを作る、などして親プロセスはそれをチェックするのがいいかと思われます。