Parallel::ForkManage: 一个简单的并行进程用于fork管理:

Parallel::ForkManage: 一个简单的并行进程用于fork管理:


  use Parallel::ForkManager;

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

  DATA_LOOP:
  foreach my $data (@all_data) {
    # Forks and returns the pid for the child:
    my $pid = $pm->start and next DATA_LOOP;

    ... do some work with $data in the child process ...

    $pm->finish; # Terminates the child process
  }

描述:

这个模块是用于使用在并行操作中进行操作, 被fork的进程数量应该被限制。

典型的使用是下载器从数百 数千个文件中检索


代码用于下载器应该是这样的:

  use LWP::Simple;
  use Parallel::ForkManager;

  ...

  my @links=(
    ["http://www.foo.bar/rulez.data","rulez_data.txt"],
    ["http://new.host/more_data.doc","more_data.doc"],
    ...
  );

  ...

  # Max 30 processes for parallel download
  my $pm = Parallel::ForkManager->new(30);

  LINKS:
  foreach my $linkarray (@links) {
    $pm->start and next LINKS; # do the fork

    my ($link, $fn) = @$linkarray;
    warn "Cannot get $fn from $link"
      if getstore($link, $fn) != RC_OK;

    $pm->finish; # do the exit in the child process
  }
  $pm->wait_all_children;


DESCRIPTION ：

这个模块用于在并行操作中进程操作,fork的进程数应该被限制。

典型的使用一个下载器来检索成百/上千个文件。

  use LWP::Simple;
  use Parallel::ForkManager;

  ...

  my @links=(
    ["http://www.foo.bar/rulez.data","rulez_data.txt"],
    ["http://new.host/more_data.doc","more_data.doc"],
    ...
  );

  ...

  # Max 30 processes for parallel download
  my $pm = Parallel::ForkManager->new(30);

  LINKS:
  foreach my $linkarray (@links) {
    $pm->start and next LINKS; # do the fork

    my ($link, $fn) = @$linkarray;
    warn "Cannot get $fn from $link"
      if getstore($link, $fn) != RC_OK;

    $pm->finish; # do the exit in the child process
  }
  $pm->wait_all_children;



首先你需要实例化ForkManager  使用"new"构造器。


你必须指定创建的最大进程数。 如果你指定为0,然后不进行fork，

这对调试是有好处

下一步,使用$pm->start  来fork.

$pm 返回0对于child process, 和child pid 对于parent process

 "and next" 跳过内部循环在parent process.
 
 NOTE: $pm->start dies if the fork fails.
 
 $pm->finish 中断child process
 
 注意:你不能使用$pm->start  如果你已经在child process里。
 
 如果你需要管理另外的子进程集在child process,
 
 你必须实例化另外一个 Parallel::ForkManager对象。
 
 
 METHODS :
 
 注释的字母表明方法允许在哪里 P对于parent C对child
 
 
 new $processes：
 
实例化一个新的 Parallel::ForkManager object.

你必须指定0（zero), 然后no children 会被forked.


这是用于调试目的。


第2个参数,$tempdir, 是只能用于如果你要children 发回一个引用到一些数据。

如果没有提供,它是设置通过调用 File::Temp::tempdir().


new方法会die如果 temporary directory 不存在或者它不是一个目录

start [ $process_identifier ]

这个方法开始fork, 它返回child process的pid 对于父进程

0对于子进程。

如果$processes 参数对于构造器是0,那么假设你是在child process,

$pm->start simply returns 0.

一个可选的$process_identifier 可以被提供用于这个方法,它是通过run_on_finish使用

finish [ $exit_code [, $data_structure_reference] ]
 
关闭 child 进程通过退出和接收一个可选的exit code（默认code是0）

可以通过回调在父进程中检索

如果第2个参数提供,child 尝试发送它的内容到parent.

set_max_procs $processes

允许你设置一个新的最大的children数量

wait_all_children:

你可以调用这个方法来等待所有的被fork的进程,这是一个堵塞等待:

reap_finished_children:

这是一个非堵塞请求来收割children，执行回调独立于调用"start"或者"wait_all_children". 

is_parent


返回真如果在parent里 返回false在child里:

  $pm->wait_all_children;
  print "22222222222222222222222222
";
  print $pm->is_parent;
  print "
";
  print "22222222222222222222222222
";

22222222222222222222222222
1
22222222222222222222222222
[oracle@node01 bb]$ 

max_procs：

返回奖fork的最大进程数

  $pm->wait_all_children;
  print "22222222222222222222222222
";
  print $pm->is_parent;
  print "
";
  print $pm->max_procs;
  print "
";
  print "22222222222222222222222222
";

22222222222222222222222222
1
30
22222222222222222222222222


running_procs

返回 forked 进程pids 当前被Parallel::ForkManager监控的。


注意 children 是仍旧被汇报为运行的直到 fork manager收割它们，

通过next call 来start或者wait_all_children。


[oracle@node01 bb]$ ps -ef | grep t1.pl
oracle   11066  1970  3 06:42 pts/0    00:00:00 perl t1.pl
oracle   11068 11066  0 06:42 pts/0    00:00:00 perl t1.pl
oracle   11069 11066  0 06:42 pts/0    00:00:00 perl t1.pl
oracle   11070 11066  0 06:42 pts/0    00:00:00 perl t1.pl
oracle   11071 11066  0 06:42 pts/0    00:00:00 perl t1.pl
oracle   11072 11066  0 06:42 pts/0    00:00:00 perl t1.pl
oracle   11074  2278  0 06:42 pts/1    00:00:00 grep t1.pl


11067
11068
11071
11070
11069$a====8

wait_for_available_procs( $n ):


等待直到$n 可用进程sots 是可用的,如果$n 没有给定 默认是1


CALLBACKS   回调:

你可以定义回调在代码里, 被调用在事件像 开始一个进程或者当finish的时候。

调用可以定义下面的方法:

run_on_finish $code [, $pid ]

你可以定义一个子例程, 当一个child 是被终止时被调用 它是在父进程中调用:

$code 的参数如下:

 - pid of the process, which is terminated
  - exit code of the program
  - identification of the process (if provided in the "start" method)
  - exit signal (0-127: signal name)
  - core dump (1 if there was core dump at exit)
  - datastructure reference or undef (see RETRIEVING DATASTRUCTURES)


run_on_start $code

你可以定义一个子例程当一个child 启动时调用,它调用在一个child 的成功启动在parent 进程

$code的参数如下:


这个$code的参数如下:

  - pid of the process which has been started
  - identification of the process (if provided in the "start" method)


你可以定义一个子例程,当child process 需要等待startup时被调用。


如果 $period没有被定义,然后一个调用时完成对于每个child.

如果 $period是被定义,然后$code 是被周期性的调用。

module 等待$period 数秒在两个请求之间,

BLOCKING CALLS

当它等待child processes终止,Parallel::ForkManager 是在一个fork和一个hard place(如果你excuse the terrible pun)


底层Perl waitpid 函数 模块依赖可以堵塞直到任何一个指定的或者任何child process 终止


这意味着module 可以做两件事情中的一个 当它等待其中一个child processes终止:

只等待它自己的child processes:

并行get:

这个小的例子可以用于get URLs并行的:

[oracle@node01 cc]$ time perl a1.pl 
$c====Parallel::ForkManager
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/1 5.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/5 9.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/4 8.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/3 7.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/2757485/3 3.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/2757485/1 1.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/7 11.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/8 12.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/9 13.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/6 10.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/1756685/2 6.txt
$linkarray=== http://blog.csdn.net/zhaoyangjian724/article/category/2757485/2 2.txt

real	0m8.121s
user	0m0.939s
sys	0m0.452s

 use LWP::Simple;
 use Parallel::ForkManager;
  my @links=(
    ["http://blog.csdn.net/zhaoyangjian724/article/category/2757485/1","1.txt"],
    ["http://blog.csdn.net/zhaoyangjian724/article/category/2757485/2","2.txt"],
    ["http://blog.csdn.net/zhaoyangjian724/article/category/2757485/3","3.txt"],
    ["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/1","5.txt"],
["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/2","6.txt"],
["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/3","7.txt"],
["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/4","8.txt"],
["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/5","9.txt"],
["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/6","10.txt"],
["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/7","11.txt"],
["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/8","12.txt"],
["http://blog.csdn.net/zhaoyangjian724/article/category/1756685/9","13.txt"]
  );

sub getsrore{
  my $link=shift;
  my $fn=shift;
  open fh1,">$fn" || die "open csdn file failed:$!"; 
  my $ua = LWP::UserAgent->new;
  $ua->timeout(10);
  $ua->env_proxy;
  $ua->agent("Mozilla/8.0");
  my $response = $ua->get("$link");
  print fh1  $response->content;  
  close fh1;  
  };
  
  # Max 30 processes for parallel download
  my $pm = Parallel::ForkManager->new(30);

  LINKS:
  foreach my $linkarray (@links) {
    $pm->start and next LINKS; # do the fork
    print "$linkarray=== @$linkarray
";
    my ($link, $fn) = @$linkarray;
      &getstore($link, $fn); 

    $pm->finish; # do the exit in the child process
  }
  $pm->wait_all_children;


Callbacks：
使用回调程序的例子得到child exit codes：

 use strict;
  use Parallel::ForkManager;

  my $max_procs = 5;
  my @names = qw( Fred Jim Lily Steve Jessica Bob Dave Christine Rico Sara );
  # hash to resolve PID's back to child specific information

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

  # Setup a callback for when a child finishes up so we can
  # get it's exit code
  $pm->run_on_finish( sub {
      my ($pid, $exit_code, $ident) = @_;
      print "** $ident just got out of the pool ".
        "with PID $pid and exit code: $exit_code
";
  });

  $pm->run_on_start( sub {
      my ($pid, $ident)=@_;
      print "** $ident started, pid: $pid
";
  });

  $pm->run_on_wait( sub {
      print "** Have to wait for one children ...
"
    },
    0.5
  );

  NAMES:
  foreach my $child ( 0 .. $#names ) {
    my $pid = $pm->start($names[$child]) and next NAMES;

    # This code is the child process
    print "This is $names[$child], Child number $child
";
    sleep ( 2 * $child );
    print "$names[$child], Child $child is about to get out...
";
    sleep 1;
    $pm->finish($child); # pass an exit code to finish
  }

  print "Waiting for Children...
";
  $pm->wait_all_children;
  print "Everybody is out of the pool!
";