Perl进程间数据共享

本文介绍的Perl进程间数据共享内容主体来自于《Pro Perl》的第21章。

IPC简介

经过fork建立多个子进程时，进程间的数据共享是个大问题，要么创建一个进程间通讯的通道，要么找到一个两进程都引用的共享变量。本文将介绍Unix IPC的近亲System V IPC：message queues(消息队列)、semaphores(信号量)和shared memory-segments(共享内存段)。它们都是IPC结构，它们被很是普遍地应用于进程间通讯。它们的帮助文档可参见：

$ perldoc IPC::Msg
$ perldoc IPC::Semaphore
$ perldoc IPC::SharedMem

可是，并不是全部操做系统都支持System V IPC，对于那些不遵照POSIX规范的平台就不支持。固然，也并不是必定要在Unix操做系统上才能使用IPC，只要操做系统支持IPC就能够，并且就算是Unix系统上也并不是必定支持IPC，可使用ipcs命令来查看是否支持：

$ ipcs

------ Message Queues --------
key        msqid      owner      perms      used-bytes   messages

------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status

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

message queues、semaphores和shared memory segments的共同点在于它们的数据都持久存在于内存中，且只要知道资源的ID且有权限访问，就能够被任意一个进程(因此随意多个进程)访问到，不只如此，还能够被其它程序访问到(例如两个perl程序文件)。因为数据在内存中持久，且数据资源进行了ID标识，这可使得程序退出前保存状态，而后重启时再次获取到原来的状态。

严格地说，Perl支持的IPC在于能够调用一些IPC函数：msgctl、msgget、msgrcv、msgsnd、semctl、semget、semop、shmctl、shmget、shmread、shmwrite。它们几乎是对C对应函数的封装，很是底层。尽管这些函数的文档很是丰富，但这些函数并不容易使用，IPC::家族的模块提供了面向对象的IPC功能支持。

IPC::SysV模块

要使用IPC的一些函数，经常须要导入一些IPC::SysV模块中的常量，所以不少使用IPC的程序中，都会：

use IPC::SysV：

它里面定义了不少常量，完整的可参见perldoc -m IPC::SysV，下面是一些常见的常量：

Message Queue(消息队列)

曾经消息队列是进程间通讯的惟一有效方式，它就像管道同样，一端写入一端读取。对于消息队列而言，咱们写入和读取的数据都称之为消息(message)。

能够建立两种类型的消息队列：私有的(private)和公有的(public)

• 私有队列只对建立它的进程和它的子进程能够访问，固然还能够经过权限控制的方式来改变访问权限
  
• 公有队列只有有权限，且知道资源ID的进程能够访问

例如，建立一个私有的消息队列。

use IPC::SysV qw(IPC_PRIVATE IPC_CREAT S_IRWXU);
use IPC::Msg;

my $queue = IPC::Msg->new IPC_PRIVATE S_IRWXU | IPC_CREAT;

IPC_Msg的构造函数new有两个参数，第一个是要建立的消息队列的资源ID，在IPC SysV中常称为KEY，对于私有队列来讲，KEY须要指定为IPC_PRIVATE，第二个参数是访问该队列的权限，S_IRWXU表示队列的全部者(U)能够对该队列进行读(R)、写(W)、执行(X)操做。此处还配合了IPC_CREAT，表示若是队列不存在就建立新队列。

权限部分也能够写成数值格式的：

my $queue = IPC::Msg->new IPC_PRIVATE 0700 | IPC_CREAT;

因此，这里建立的私有队列只有建立者进程和子进程能够执行读写执行的操做。

若是想要建立一个公有队列，须要为该公有队列提供一个资源ID，资源ID是一个数值。下例中给的资源ID是10023，权限是0722，表示建立队列的进程拥有读写执行操做，而资源所在组或其它用户进程只能写队列。

my $q = IPC::Msg->new 10023, 0722 | IPC_CREAT;

若是其它进程想要访问这个公有队列，只需经过new方法指定这个公有队列的KEY便可即表示构建这个已有的队列，不要指定IPC_CREAT修饰符，不然表示建立动做(尽管IPC结构存在时不会建立，但他表明了建立这个动做，而非访问动做)。若是要获取的公有队列不存在，则返回undef。以下：

my $q = IPC::Msg->new 10023, 0200;

而对于私有队列，想要知道它的KEY，可使用id()方法：

$KEY = $queue->id

发送和接收消息队列

有了消息队列的对象结构以后，就能够操做这个消息队列，好比发送消息，接收消息等。相关文档参见man msgsnd。

向队列发送消息和从队列中接收消息的方式为：

$queue->snd($type, $wr_msg, [ $flags ]);
$queue->rcv(\$rd_msg, $length, $type, [ $flags ]);

• $wr_msg为想要发送的消息
  
• $rd_msg是从消息队列中读取消息保存到哪一个标量变量中
  
• $type是一个正整数，表示消息队列的类型，可在rcv方法中指定这个正整数表示选择接收哪一个数值类型的消息
  
• $length表示消息队列中最多容许接收多少条消息。若是消息长度大于该值，则rcv返回undef，而且$!设置为E2BIG
  
• $flags是可选的，若是设置为IPC_NOWAIT，则这两个方法不会阻塞，而是当即返回($!设置为EAGAIN)

关于rcv type和flag的规则，参考以下解释。

rcv Type的解释：
整数值             意义
-----------------------------
 0           rcv老是读取队列的第一条消息，无视type

 >0          rcv老是读取该类型的第一条消息。例如，
             type=2，则只读取type=2的消息，若是不存在，
             则一直阻塞直到有type=2的消息。可是能够设
             置IPC_NOWAIT和MSG_EXCEPT常量改变这种模式

 <0          rcv读取类型不大于type绝对值(从小到大)的第
             一条消息。不严谨，但可看示例描述：若是rcv的
             type=-2，则首先读取type=0的第一条消息，如
             果不存在type=0的消息，则继续读取type=1的第
             一条消息，不存在则继续读取type=2的第一条消息

flag的解释：
flag值              意义
------------------------------
MSG_EXCEPT         rcv读取第一条非type值的消息。例如，rcv
                   的type=1，则读取第一条type不为1的消息

MSG_NOERROR        容许消息过长超过$length时截断超出的部分，
                   而不是在这种状况下返回E2BIG错误  

IPC_NOWAIT         rcv在请求的消息类型不存在时不要阻塞等待，
                   而是当即返回，且设置$!的值为EAGAIN

将上面的解释合并起来，很明确的意思是咱们能够经过设置不一样的type来实现多级通讯的消息队列，这一切都交给咱们本身来决定，例如对不一样子进程或线程发送不一样的消息。

获取和设置消息队列的属性

可使用set方法来修改消息队列的权限，它须要一个key-value格式的参数。

例如：

$queue->set(
    uid => $user_id,           # chown
    gid => $group_id,          # chgrp
    mode => $perm,             # 8进制权限位或S_格式的权限
    qbytes => $queue_size,     # 队列最大容量(capacity)
);

另外，可使用stat方法获取队列的属性对象，经过这个属性对象，能够直接修改队列的对应属性。只是须要注意的是，当经过stat对象更改属性时，不会当即应用到消息队列上生效，只有经过set方法设置后，设置才会当即生效。

my $stat = $queue->stat;
$stat->mode(0722);
$queue->set($stat);

最后，若是拥有队列的执行权限，能够经过remove方法销毁这个队列：

$queue->remove;

若是没法删除队列，则remove返回undef，并设置$!。其实删除队列挺重要的，由于若是程序退出，队列可能会继续保留在内存中(前文已经说过了，IPC对象都是持久化在内存中的)。

信号量(semaphore)

信号量也称为信号灯，典故来源于荷兰：火车根据旗标来决定是否通行。就像红灯停、绿灯行的意思，红绿灯就是信号灯，车子就是被阻塞或通行的进程/线程。

在编程语言中，信号量是一个整数值，若是是正数，表示有多少个信号量，也表示可以使用的信号灯数量，也能够是0或负数。信号量要结合PV操做(两个原语)才能真正起做用，P是减一个信号灯操做，V是加一个信号灯操做。

信号量的规则是这样的：

• 若是一个进程请求P操做(减1操做，即请求一个信号灯)，若是减去以后为负数，则该进程被阻塞，若是减去以后为0或正数，则放行该进程
  
• 若是一个进程请求V操做(加1操做，即释放或增长一个信号灯)，进程直接放行
  
• 若是请求V操做，若是加1以后仍为0或负数，则放行该进程的同时还唤醒另外一个被阻塞的进程。若是加1后为正数，则直接添加一个信号灯资源

总结起来很简单：若是当前没有信号灯资源(小于或等于0)，那么请求信号灯(P原语)的进程就会被阻塞；若是有信号灯资源(大于0)，就直接放行。若是一个进程原本就是来增长信号灯资源的(V原语)，那么这个进程固然要放行，由于添加了一个信号灯，那么还能够拥有唤醒一个被阻塞进程的能力(若是有被阻塞进程的话)。

若是限制只使用1个信号灯，那么信号量就实现了锁的机制：P是申请锁操做，只有在有值为1的时候才能申请锁，不然被阻塞；V是释放锁，一直被放行。

其实，只要把信号灯理解为一种有限的资源就很容易理解信号量的机制。

固然，具体到信号量的实现上就不必定遵照上面的操做，例如能够一次加N或一次减N，而不是以1做为操做单位。

对于Sys V IPC中信号量来讲：

1. 每个信号量结构(或信号量对象，经过KEY来标识)能够有多路信号量，每一路信号量经过信号量序号semnum标识分类。信号量序号从0开始，每一号的信号量上都是互相独立的，都有本身的信号量值(信号灯的数量，semval)以及控制进程是否阻塞的信号量操做，操做0号信号量不会影响1号信号量
   
2. 使用semop函数来操做某个信号量结构，semop函数能够一次性操做多路信号量，每一路信号量都要求3个值：sem_num, sem_op, flag
   • sem_num：指定要操做哪路信号量
     
   • sem_op：是一个整数值，用来表示信号量的操做模式，能够是0、正数、负数
     • 正数N：表示增长N个信号灯资源
       
     • 0：表示等待信号灯的数量为0，等待过程当中一直阻塞
       
     • 负数-N：表示等待信号灯的数量大于等于N，等待过程当中一直阻塞
       
   • flag：可被信号量识别的flag除了0外只有两个：IPC_NOWAIT和SEM_UNDO
     • 0：若是该op不能成功，则一直等待(阻塞)直到能够成功
       
     • IPC_NOWAIT：不阻塞，而是当即返回并设置操做信息为EAGAIN（对于Perl来讲设置$!）
       
     • SEM_UNDO：以该flag执行op时，在进程退出时(不管是正常仍是异常退出)自动归还信号灯。例如已有信号灯10，以undo方式执行减二、加3，最后信号灯数量为11，当退出时反向操做，又变回10。使用sem_undo能够有效避免进程异常时永久锁住资源不释放的问题
       
3. 因为减法操做"-N"在减的过程当中一直没有减下去，而是一直阻塞，因此不会出现负数信号灯，而是以等待加法操做的进程数来衡量，这和前面的信号灯机制是不同的
   
4. semop的操做是原子的，要么多路信号量所有操做成功，要么所有失败

多路信号量的模式以下图所示：

SysV IPC经过这样的信号量规则，可让进程之间进行协做，例如一个进程能够经过设置信号量的值来控制另外一个进程是执行仍是阻塞，不只如此，还能够控制进程间的共享资源。这是一个很是大的话题，这里给个简单的信号量控制进程协做的示例来解释进程间如何访问共享资源。

1.进程A建立一个信号量，其值为1，并建立一个共享资源(如一个文件或IPC共享内存段)。因为这个资源可能涉及到不少初始化，因此如今不当即访问这个资源
2.进程B启动，将信号量的值减为0(即获取锁)，而后访问共享资源
3.进程A如今尝试减小信号量的值(即申请锁)并访问共享资源，因为当前信号量的值为0，不足以完成减法，因此进程A被阻塞
4.进程B完成了共享资源的访问，并增长信号量(释放锁)的值，这个操做是必定会成功的
5.进程A如今能够执行减法操做(获取锁)了，由于信号量的值已经变成了1，因而可以访问共享资源
6.进程B尝试第二次访问共享资源，但它会被阻塞，由于信号量的值被进程A减为0
7.进程A完成共享资源的访问，并增长信号量的值
8.进程B访问共享资源并减小信号量

尽管共享资源和信号量没有直接的关联关系，可是信号量在这里充当了看门狗，只要想访问共享资源，都须要从信号量这里获取访问权限。

从上面的操做上能够发现，减法操做和加法操做顺序必须不能错(先减后加，即PV)，并且减法、执行和加法的操做必须在同一个临界区内执行，便是一个原子操做。

建立信号量

建立信号量须要经过IPC::Semaphore模块，固然，还须要导入IPC::SysV提供使用IPC结构时须要的常量。

一样，Semaphore做为一种SysV IPC结构，它也有公有和私有两种信号量类型，且也使用KEY来标识信号量资源，使用权限来控制访问、修改信号量。其实建立和获取消息队列、信号量和共享内存这3种IPC结构的方式都是一致的，都有公有私有的区别，都是用KEY来标识，都使用权限位来控制访问能力，

例如，建立私有信号量并获取它的id标识符：

use IPC::SysV qw(IPC_CREAT IPC_PRIVATE S_IRWXU);
use IPC::Semaphore;

my $size = 4;
my $sem = IPC::Semaphore->new $size, IPC_CREAT | S_IRWXU;

$id = $sem->id;

这里的$size=4表示初始化该信号量对象时有4路信号量。

建立公有信号量，并设置其KEY为10023：

my $sem = IPC::Semaphore->new 10023, 4, 0744 | IPC_CREAT;

有了信号量ID，其它进程就能够获取到对应的信号量结构，注意不要加上IPC_CREAT修饰符：

my $sem = IPC::Semaphore->new 10023, 0400;

操做信号量

有了信号量结构，就能够操做这个信号量。有如下几个常见方法：

getall    返回当前信号量对象中全部路信号量的信
          号量值(即信号灯数量)放进一个列表
          my @semval = $sem->getall;

getval    返回当前信号量对象指定序号的信号量值
          例如返回第4路信号量的信号灯数量
          my $semval = $sem->getval(3);

setall    设置当前信号量对象中全部路信号量的
          信号量值。例如清空上例建立的4路信号量
          $sem->setall( (0) x 4 );

setval    设置指定某路信号量的信号量值
          例如设置第4路信号量的信号灯为1
          $sem->setval(3, 1);

set       设置信号量对象的UseID、GroupID以及权限
          例如 $sem->set(
              uid => $usr_id,
              gid => $grp_id,
              mode => $perm,
          );

stat      获取当前信号量对象的stat对象，能够经过stat
          对象简单地修改信号量属性。例如
          $semstat = $sem->stat;
          $semstat->mode(0744);
          $sem->set($semstat);

getpid    返回在此信号量对象上最近一次执行semop操做的进程PID
          PID that did last op

getncnt   返回等待某路信号量的值增长的进程数量
          waiting for increase
          例如 $ncnt = $sem->getncnt;

getzcnt   返回等待某路信号量的值为0的进程数量
          waiting for zero
          例如 $zcnt = $sem->getzcnt;

op        信号量操做，见下文

对于Perl而言，有了信号量，还须要结合op方法来执行"PV"操做，规则在前面介绍SysV IPC信号量规则的时候已经介绍过了。给个示例：

$sem->op(
    0, -1, 0,
    1, 1, 0,
    3, 0, 0,
);

op能够一次性操做某信号量对象的多路信号量，每一路信号量由3个元素组成一个小列表，例如上面的0, -1, 0表示操做第一路信号量，其中第一个元素表示sem_num，即第几路信号量，-1是sem_op，值为-1表示要等待信号量的值至少为1以后才不会阻塞，最后一个元素0表示flag(flag的解释见前文)。

例如，想要使用信号量来实现锁机制，锁机制只需一路信号量且一个信号灯便可：

sub access_resource {
    # 访问资源，执行减法操做来获取锁，若是已经为0，则本身被阻塞
    $sem->op(0, -1, 0);

    ... 访问资源 ...

    # 访问完成，执行加法操做来释放锁
    $sem->op(0, 1, 0);
}

最后，信号量和消息队列相似，都应该在不须要的时候清空它(好比最后一个进程退出且肯定再也不使用它的时候)。

共享内存段

Shared Memory Segments是IPC的第三种结构，和IPC::Msg和IPC::Semaphore对应的是IPC::SharedMem，可是它们都太底层了。对于共享内存来讲，Perl中有更高层次的IPC::Shareable模块，使得共享内存操做更加方便，它的实现使用了tie机制，能够简单地附加(attach)一个变量到共享内存段上并轻松地访问它。但可能须要先安装它：

> cpan IPC::Shareable

tie方法有4个参数：

• (1).一个待附加的变量(变量、数组、hash等，但它们中能够有更复杂的数据结构，如引用)
  
• (2).IPC::Shareable
  
• (3).一个IPC结构的key，key能够是一个数值或字符串，但最多只能是4个字符，超出的字符将忽略，因此abcd和abcde表明的是同一个key
  
• (4).Options，它是可选的hash引用，该hash引用中包含了一个或多个key/value对，稍后解释

例如，下面的代码中建立并tie了一个Hash变量(local_hash)到共享内存段上。

use IPC::SysV;
use IPC::Shareable;

our %local_hash;
tie %local_hash, "IPC::Shareable", "mykey", {create => 1};

$local_hash{hashkey} = "hashvalue";

tie一个共享内存段后，在该tied变量之下会有一个tie对象，它能够经过tie的返回值或tied()函数来获取。下面两种获取tie对象的方式是等价的：

$mytie = tie $sv, 'IPC::Shareable', 'mykey', {...};
$mytie = tied $sv;

下面是关于tie方法第四个参数Options的说明，它是一个hash引用，该hash中可定义的key包括以及它们的默认值为：

{
    key       => IPC_PRIVATE,
    create    => 0,
    exclusive => 0,
    destroy   => 0,
    mode      => 0666,
    size      => IPC::Shareable::SHM_BUFSIZ(),
}

一个个解释这6个key，有些布尔值类型的，只需提供Perl中的false值或true值便可，例如数值的0和空字符串均可以表示false。

key
在前面介绍tie方法时解释了有4个参数，可是其实能够将第三个参数KEY放进这个hash引用中，并使用key来指定KEY。例如使用3个参数建立一个共享内存变量：
tie %myhash, "IPC::Shareable", {key => "mykey"};
默认值为IPC_PRIVATE，其它进程没法访问该共享内存变量

create
当key不存在时就建立，默认为false，表示不会尝试建立key，因此必需要求key是已经存在的，不然将失败并返回undef

exclusive
若是key存在时，则不建立，而且失败返回undef。默认值为false，这时即便key已存在也不会失败  

mode
八进制的权限位，控制key被建立时的权限。例如0666表示对owner、group、other均可读、可写，而0600表示只对owner可读可写。默认值为0666

destroy
设置为true时，当调用tie的进程退出时将自动销毁该tie建立的共享内存段。默认值为false

size
指定共享内存段分配的大小，默认值为IPC::Shareable::SHM_BUFSIZ()，默认值为65536字节

关于IPC::Shareable的锁机制

IPC::Shareable提供了程序级别的锁机制，它直接拷贝了IPC::ShareLite中的锁机制，但它们的底层是使用IPC::Semaphore实现的，因此若是想要实现本身的锁机制，能够直接使用IPC::Semaphore。能够直接调用shlock()和shunlock()方法来获取锁和释放锁。但在使用它们以前，须要先获取到tie对象，前面说过如何获取tie对象。

例如，下面两种加锁方式是等价的：

$mytie = tie $sv, 'IPC::Shareable', 'mykey', {...};
...
$mytie->shlock;


tie $sv, 'IPC::Shareable', 'mykey', {...};
...
(tied $sv)->shlock;

IPC::Shareable提供了独占锁(LOCK_EX)、共享锁(LOCK_SH)、非阻塞锁(LOCK_NB，没法获取锁的时候当即返回0)三种锁，只要将它们做为shlock方法的参数便可申请对应模式的锁。多个共享锁能够共存，但独占锁和独占锁、共享锁都互斥。此外，还能够为shlock()提供LOCK_UN参数来实现shunlock()，它们是等价的。若是shlock()不提供任何参数，则默认为LOCK_EX。

在使用这几个常量以前，须要先导入(all或lock或flock标签均可以)。例如：

use IPC::Shareable qw(:all);

if ( (tied $sv)->shlock(LOCK_SH | LOCK_NB) ){
    print "The value is $sv\n";
    (tied $sv)->shlock(LOCK_UN);   # (tied $sv)->shunlock;
} else {
    print "Another process has an exclusive lock now\n";
}

上面的示例中结合了共享锁和非阻塞锁，其实独占锁和共享锁均可以结合非阻塞锁：

shlock(LOCK_EX | LOCK_NB)   # 获取独占锁失败时当即返回0，表示资源已锁定
shlock(LOCK_SH | LOCK_NB)   # 获取共享锁失败时当即返回0，表示资源已被独占锁锁定

清除共享内存段

tie的第四个参数中，能够设置destory选项，该选项使得调用tie的进程退出时自动删除对应的tie对象以及其对应的共享内存段。

除了destory，还有remove、clean_up和clean_up_all能够用来移除共享内存段。

(tied $sv)->remove;
IPC::Shareable->clean_up;
IPC::Shareable->clean_up_all;

remove方法能够删除tie对象对应的共享内存段。无视destory选项的设置，无视是哪一个进程。

clean_up是一个类方法，只删除调用该方法的进程所建立的共享内存段。非该进程所建立的，clean_up不会删除。

clean_up_all移除该进程所能看见的全部共享内存段，而不限于该建立者进程。

若是要清除共享内存段、消息队列、信号量，可使用ipcrm命令。

共享内存段示例

在server.pl文件中：

#!usr/bin/perl -w

use strict;
use IPC::Shareable;

my $key = 'data';
my %options = (
        create => 1,
        exclusive => 1,
        mode => 0644,
        destory => 1,
);

my %colors;
tie %colors, 'IPC::Shareable', $key, { %options } 
    or die "Sever: tied failed";

%colors = (
        red => [
                'fire truck',
                'leaves in the fall',
        ],
        blue => [
                'sky',
                'police cars',
        ],
);

((print "Server: there are 2 colors\n"), sleep 2) while scalar keys %colors == 2;

print "Server: here are all my colors:\n";
foreach my $c (keys %colors){
        print "Server: these are $c: ",
                join(', ', @{$colors{$c}}), "\n";
}

exit;

在client.pl文件中：

#!/usr/bin/perl -w
#
use strict;

use IPC::Shareable;

my $key = 'data';

my %options = (
        create => 0,      # 不建立，直接获取data
        exclusive => 0, 
        mode => 0644, 
        destory => 0,
);

my %colors;

tie %colors, "IPC::Shareable", $key, { %options } or
        die "Client: tied failed\n";

foreach my $c (keys %colors){
        print "Client: these are $c: ",
                join(', ', @{$colors{$c}}), "\n";
}

delete $colors{'red'};  # 删除一个key/value
exit;

逻辑很简单，只为了证实不一样进程间能够获取同一个共享数据段，且进程退出以后数据还可以继续保留在共享内存中。

执行它们：

$ perl server.pl &
$ perl client.pl  # 将输出
$ ipcs