perl多进程实战之三-IPC

    本来说是要讨论文件存储的，但是要穿越一下，因为我觉得IPC更加基础一点，特别是锁这方面，其实用Semaphore都是可以代替的，不过，ipc的话，废话可能就要多一点点了。

 

    IPC就是用来进程间通讯用的东东，linux下，分为三类：消息队列、信号量、共享内存，在linux下，用ipcs命令就可以看到当前系统已经创建的ipc，ipcrm，可以删除已经创建的ipc，当然，要跟id的。既然它分为三类，那就要分三种类型来做讲解了。

 

    首先是消息队列。消息队列比较简单，就是把数据放到队列里面，按照先进先出，后进后出的序列来。消息队列用的人已经很少了，它的大部分功能，都可以由共享内存来替代，这里就不举例子了。

 

    信号量，用的应该是比较多的，主要的作用就在于锁，代码如下：

 

#!/usr/bin/perl

 

use strict qw(vars);
use warnings;
use IPC::SysV qw(S_IRWXU IPC_CREAT);
use IPC::Semaphore;

 

my $sem0;
my $sem1;

 

my $sem = IPC::Semaphore->new(1564, 2, S_IRWXU|IPC_CREAT)
        || die "IPC::Semaphore->new: $!\n";

 

$sem0 = $sem->getval(0);
$sem1 = $sem->getval(1);
print "Sem0: $sem0\n";
print "Sem1: $sem1\n";

 

$sem->setval(0,10);
$sem->setval(1,100);
$sem0 = $sem->getval(0);
$sem1 = $sem->getval(1);
print "Sem0: $sem0\n";
print "Sem1: $sem1\n";

 

$sem->op(0,-1,IPC_NOWAIT) || die "sem setval: $!\n";
$sem->op(1,10,IPC_NOWAIT) || die "sem setval: $!\n";
$sem0 = $sem->getval(0);
$sem1 = $sem->getval(1);
print "Sem0: $sem0\n";
print "Sem1: $sem1\n";

 

 

    执行结果如下：

 

Sem0: 0
Sem1: 0
Sem0: 10
Sem1: 100
Sem0: 9
Sem1: 110

 

    在刚开始，定义一个信号量：

    第一个参数是key，它的key是1564，这个是随便选的，如果再别的进程里也需要连接这个信号量，那么就用相同的key连接即可。

    第二个参数是信号量的数量，上例中是两个，信号量可以看做是在内存里存放的一个纯数字的数组，这个数量，那就是你的数组大小，可以包含多少数字。

    第三个参数是flag，定义权限等，S_IRWXU，意思是创建它的用户可读可写可执行，如果是S_IRWXG，则是属组，如果是S_IRWXO，则是所有人。IPC_CREAT代表如果不存在，则创建。

 

    getval函数，是用来获取信号量中的数值的，参数只有一个，可以看做是这个信号量数组的下标，上例只有两个数，那么就是0或者1。也可以用getall直接导出为数组。

    setval函数就是赋值了，第一个参数为下标，第二个为要赋的值。

    op函数是操作的意思，第一个函数为下标，第二个函数为要做的加加减减的操作，正为加，负为减，第三个为flag

    remove函数可以销毁信号量。

 

    脚本最后并没有销毁信号量，在脚本退出后，可以执行ipcs命令，看到我们创建的这个信号量。

 

# ipcs -s

 

------ Semaphore Arrays --------
key        semid      owner      perms      nsems    
0x0000061c 131072     root      700        2

 

    可以看出它的权限是700，里面有两个数值。

 

    如果我们再执行一次脚本，因为它用了同样的key，所以并不会重新创建，而是连接到已有的信号量上，输出如下：

 

Sem0: 9
Sem1: 110
Sem0: 10
Sem1: 100
Sem0: 9
Sem1: 110

 

    和第一次执行，刚开始输出的0不同，这次的初始值为上次最后的值。如果我们在脚本最后加上“$sem->remove();”，那么信号量将被销毁，脚本执行后，执行ipcs，输出如下：

 

# ipcs -s

 

------ Semaphore Arrays --------
key        semid      owner      perms      nsems

 

    信号量已经被销毁了。

    

    信号量所能存储的信息是很有限的，它最广泛的应用就是在锁上面，比如自旋锁，当一个进程进入一段资源之前，先检查特定的信号量，如果它是0，则将它置为1，进入资源，在退出资源以后，再将它置为0，如果再进入资源前，检查信号量发现是1，那么就等待一段时间，等它成为0以后再进行操作。读写锁也一样，读锁加1，写锁减1，如果发现它不小于1，那么就将信号量加1，然后读取资源，读取完毕后减去1，写的话，只有当发现信号量等于1的时候，才能进行写操作，之前要将它减去1，成为0，阻止其它任何操作。几乎所有的锁操作，都可以用它来完成。

    下面的代码是一段用信号量给DB_File多进程写操作进行加锁的过程，和用DB_File::Lock是同样的效果。

 

#!/usr/bin/perl

# sem for lock

# create by lianming: 2009-08-12

# last modify by lianming: 2009-08-12

 

use strict qw(vars);
use warnings;
use DB_File;
use IPC::SysV qw(IPC_PRIVATE S_IRWXU IPC_CREAT);
use IPC::Semaphore;

 

use POSIX ":sys_wait_h";

 

our $zombies = 0;
our $procs = 0;

 

$SIG{CHLD} = sub { $zombies++ };

 

sub REAPER {
    my $pid;
    while (($pid = waitpid(-1, WNOHANG)) > 0) {
        $zombies --;
    }
}

 

my $sem = IPC::Semaphore->new(7456, 1, S_IRWXU|IPC_CREAT)
        || die "IPC::Semaphore->new: $!\n";

 

$sem->setval(0,0) || die "sem setval: $!\n";

 

sub child {
        my $sem_c = IPC::Semaphore->new(7456, 1, S_IRWXU)
                || die "IPC::Semaphore->new: $!\n";

 

        my $cnt = $_[0];
        my $sem_v;
        my %hash;
        my $file_name = "/opt/B_db_test/DB/test_btree";
        my ($k, $v);

 

        for (my $i = 1; $i < 10; $i++) {
                $k = $cnt*10+$i;
                $v = $k*2;
                do {
                        $sem_v = $sem_c->getval(0);
                } while ($sem_v != 0);
                $sem_c->op(0,1,IPC_NOWAIT);
                my $db = tie(%hash, 'DB_File', $file_name, O_CREAT|O_RDWR, 0666, $DB_BTREE)
                        || die "Cannot open $file_name: $!\n";
                $hash{$k} = $v;
                undef $db;
                untie %hash;
                $sem_c->op(0,-1,IPC_NOWAIT);
        }
        my $time_end = time();
        print "$time_end: complete\n";
        exit 0;
}

 

for ($procs; $procs<5; $procs++) {
        my $pid = fork();

 

        if (!defined($pid)) {
                print "Fork Error: $!\n";
                exit 1;
        }

 

        if ($pid == 0) {
                &child($procs);
                exit 0;
        } else {
                my $time = time();
                print "$time:$procs process forked!\n";
                &REAPER if ($zombies > 0);
                sleep(0.02);
        }
}

 

exit 0;

 

    共享内存，存放数据的话，共享内存应该算是主力了，因为它相对而言更加灵活，不过在多进程操作的时候，需要额外加锁，很多情况下，这个锁都是通过信号量加的。

 

#!/usr/bin/perl
# test share memory
# create by lianming: 2009-08-14
# last modify by lianming: 2009-08-14

 

use strict qw(vars);
use warnings;
use IPC::ShareLite;

 

my $share = IPC::ShareLite->new(
        -key     => 1971,
        -create  => 'yes',
        -destroy => 'yes'
) or die $!;

 

my $b = "Test for share memory";

 

$share->store($b);
my $test = $share->fetch;
if ($test ne "") {
        print "$test\n";
}

 

    输出结果为：

 

Test for share memory

 

    IPC::ShareLite是对共享内存的pm包进行了一次封装，使用上更加人性化吧！

    创建共享内存，创建的过程是一个对哈希进行赋值的过程：

    1、key，使用相同的key，可以再不同的进程中访问同一段共享内存

    2、create，是否创建

    3、destroy，是否销毁

    4、exclusive，如果它为true的话，那么如果这段共享内存存在了，它会报错

    5、mode，权限

    6、size，共享内存的大小，默认为65536Byte，linux支持的最大共享内存可以通过调整内核参数来设置，位置在/proc/sys/kernel下，shmall、shmmax、shmmin（消息队列的内核参数也在此目录下调整，为msgmax、msgmnb、msgmni）

 

    store函数，就是将一个变量存入共享内存。

    fetch是从共享内存中获取变量。

 

    如果destroy设置为0的话，那么在脚本退出后，可以通过ipcs命令看到我们创建的共享内存。

 

# ipcs -s -m

 

------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status     
0x000007b3 98304      root      666        65536      0                      

 

------ Semaphore Arrays --------
key        semid      owner      perms      nsems    
0x000007b3 294913     root      666        3   

 

    但是我们会发现，不仅仅是一段共享内存，它还创建了一个信号量，这个信号量和共享内存的 key是相同的，它就是用来对共享内存加锁的。ShareLite的锁做的有问题，我在调用lock和unlock的时候会报错。

    Use of uninitialized value in subroutine entry at /usr/lib/perl5/site_perl/5.8.5/i386-linux-thread-multi/IPC/ShareLite.pm line 361.

    不过，既然是通过信号量加锁，那倒是也无所谓，自己多写两行代码罢了。

 

    还有一个比较麻烦的地方，就是它无法直接存储数组和哈希，存储数组和哈希要通过另外一个包实现：

 

#!/usr/bin/perl

 

use strict qw(vars);
use warnings;
use IPC::ShareLite;
use Storable qw(freeze thaw);

 

my $share = IPC::ShareLite->new(
        -key     => 1971,
        -create  => 'yes',
        -destroy => 'no'
) or die $!;

 

my @a = ("This", "is", "stored", "in", "shared", "memory");
my $b = \@a;

 

$share->store(freeze($b));
my $test = $share->fetch;
if ($test ne "") {
        my $str = thaw($share->fetch) || print "Wrong\n\n";
        print "@{$str}\n";
}

 

    输出结果为：

 

This is stored in shared memory

 

    哈希值也是一样的。

    需要先包含Storable，在调用store之前，先调用freeze函数，它的参数为你想存储的哈希或者数组的引用。在调用了fetch之后，要将获取到的值作为参数调用thaw函数，还原为引用的格式，然后才能使用。