首页 > 脚本语言 > perl > Perl 中的 IPC::Semaphore 信号量的操作
2015
07-24

Perl 中的 IPC::Semaphore 信号量的操作

什么是信号量?
信号量(semaphore)的数据结构为一个值和一个指针,指针指向等待该信号量的下一个进程。信号量的值与相应资源的使用情况有关。当它的值大于0时,表示当前可用资源的数量;当它的值小于0时,其绝对值表示等待使用该资源的进程个数。

注意,信号量的值仅能由PV操作来改变。
一般来说,信号量S>=0时,S表示可用资源的数量。执行一次P操作意味着请求分配一个单位资源,因此S的值减1;当S<0时,表示已经没有可用资源,请求者必须等待别的进程释放该类资源,它才能运行下去。而执行一个V操作意味着释放一个单位资源,因此S的值加1;若S<0,表示有某些进程正在等待该资源,因此要唤醒一个等待状态的进程,使之运行下去。

PV操作的含义
PV操作由P操作原语和V操作原语组成(原语是不可中断的过程),这些操作会对对信号量进行相关的操作
具体定义如下:
假设有信号量 S. 然后我们分别来讲 P 和 V 操作.
P(S):

将信号量S的值减1,即S=S-1;
如果S>=0,则该进程继续执行;否则该进程置为等待状态,排入等待队列。
V(S):

将信号量S的值加1,即S=S+1;
如果S>0,则该进程继续执行;否则释放队列中第一个等待信号量的进程。
PV操作的意义:我们用信号量及PV操作来实现进程的同步和互斥。PV操作属于进程的低级通信。 互斥时,进程的用户实现互斥需要成对出现, 默认互斥信号量的初值一般为1。

我们来看看 Perl 模块有关信号量的模块.

创建信号量
这个地方, 我们需要引入二个模块, IPC::SysV 是专门用来引入这种信号量操作的常量, 由  IPC::Semaphore 来操作信号量, 如果我们还要多进程共享,可能还需要引入 IPC::SysV::ftok 这个模块来对本地文件进行操作.

1
2
3
4
use IPC::SysV qw(IPC_PRIVATE S_IRUSR S_IWUSR IPC_CREAT);
use IPC::Semaphore;
 
$sem = IPC::Semaphore->new(IPC_PRIVATE, 10, S_IRUSR | S_IWUSR | IPC_CREAT);

这个地方第一个参数为信号量的名字, 第二个参数为信号量上最多的资源的数量, 所以这个地方这样设置就可以有 10 个信号量.
注意: 多个进程做同步之类时, 我们需要给第一个参数指定为一个文件.通常使用 IPC::SysV::ftok 来替换 IPC_PRIVATE.这样其它进程才能得到这个信号量的编号, 在多个进程中才能相互关联起来.

IPC::Semaphore 模块的基本操作
常用的几个信号量操作的方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
$sem->setall( (0) x 10);
设置上面 10 个信号量都为 0.当然我们也可以单个信号量来设置, 如下
 
 
$sem->setval(0, 0);
设置第 0 信号量的值为 0, 因为一共这个信号量才 1 个, 所以从零开始. 这样的话, 如果使用 P/V 操作的话,最开始操作的时候谁也无法取得资源.
 
 
@sem = $sem->getall;
取得所有信号量的状态.
 
进行信号量的 PV 操作主要是周 op 方法
 
 
$sem->op(0, -1, SEM_UNDO);

这样会操作这 10 个信号中的第零个信号量减 1.这个就是所谓的 PV 的操作.这个最后一个参数是 semop 中的 SEM_UNDO 操作, 这个会在进程退出时自动还原所有操作.这个地方建议
直接使用 IPC::SysV 来给 SEM_UNDO 和 IPC_NOWAIT 这几个常用的参数都导出来. 在这的就可以实现非阻塞返回值.

我们现在来看个日本人写的一个例子, 来做详细分析:

1
2
3
4
5
6
7
8
9
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
#!/usr/bin/env perl
use strict;
use warnings;
use 5.010;
use IPC::Semaphore;
use IPC::SysV qw/ IPC_PRIVATE IPC_CREAT S_IWUSR SEM_UNDO /;
use Parallel::ForkManager;
use Time::HiRes ();
 
my $process = 10;
my $pm  = Parallel::ForkManager->new($process);
 
# 创建一个信号量, 最多可以有 1 的资源
# 第一个参数为信号量的名字, 第二个参数为信号量上最多的资源的数量
# 公用信号量: 实现进程间的互斥, 初值=1或资源的数目
# 私用信号量: 实现进程间的同步, 初值=0或某个整数
my $sem = IPC::Semaphore->new(IPC_PRIVATE, 1, IPC_CREAT | S_IWUSR);
 
# 设置第 0 信号量的值为 0, 因为一共这个信号量才 1 个, 所以从零开始. 这样最开始操作的时候谁也无法取得.
$sem->setval(0, 0);
 
for (1..$process) {
if ($pm->start) {
Time::HiRes::sleep(0.2)# fork 的延时
next;
}
 
# 操作第 0 个信号量, 进行减少
# 这使用了 semop 的 SEM_UNDO 操作, 这个会在进程退出时自动还原所有操作.
# 这还可以使用 IPC_NOWAIT 就可以实现非阻塞返回值
# 这个 P/V 操作因为减少时小于等于 0 , 所以这个时候进程并不会工作.
$sem->op(0, -1, SEM_UNDO);
 
# 子进程开始处理的时间
say "[$$] ", Time::HiRes::time();
 
# 子进程退出, 这时会通过 SEM_UNDO 来解锁
$pm->finish;
}
 
# 设置第 0 个信号量的默认值为进程数量
$sem->setval(0, $process);
$pm->wait_all_children;
 
# 删除信号量
$sem->remove;

所有的程序细节, 我都写了注释, 这个地方, 我们是用来做同步, 让所有的进程都 Fork 完了, 然后都等到父进程 setval 给信号量设置值以后, 这些子进程才开始执行.我可以见到如下的输出, 这个输出可以见到, 从时间上来看, 子进程都是一起执行的.

1
2
3
4
5
6
7
8
9
10
[30007] 1423541569.65873
[30010] 1423541569.65877
[30017] 1423541569.65879
[30009] 1423541569.65993
[30016] 1423541569.66185
[30011] 1423541569.66185
[30015] 1423541569.66452
[30012] 1423541569.66453
[30014] 1423541569.66596
[30013] 1423541569.66729

这时, 假设我们不使用信号量来控制看看.代码实现如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
use Parallel::ForkManager;
use Time::HiRes ( );
 
my $process = 10;
my $pm = Parallel::ForkManager->new($process);
 
for (1..$process) {
if ($pm->start) {
Time::HiRes::sleep(0.2);
next;
}
 
say "[$$] ", Time::HiRes::time( );
 
$pm->finish;
}
 
$pm->wait_all_children;
这时输出会变成, 每过 0.2 秒有一个子进程执行, 并不能同步一起来执行.
 
 
[17905] 1423552414.9412
[17906] 1423552415.14185
[17907] 1423552415.34264
[17910] 1423552415.5436
[17911] 1423552415.74442
[17912] 1423552415.94519
[17913] 1423552416.14596
[17914] 1423552416.3467
[17915] 1423552416.54749
[17916] 1423552416.74825

以上的例子可能实际中并不会有大的作用, 但通过这个我们可以很好的了解  IPC::Semaphore 这个模块和操作系统的信号量到底是怎么回事, 怎么样工作的.

最后编辑:
作者:saunix
大型互联网公司linux系统运维攻城狮,专门担当消防员

留下一个回复