Semaphores (信号量)

Semaphores

注意：竞争属于协作，竞争是协作的特例，因此使用信号量可以解决竞争关系

信号量与PV操作

信号量

  • 信号量（Semaphore）是一种比互斥锁更强大的同步工具，它可以提供更高级的方法来同步并发进程。
  • A semaphore S is an integer(整数) variable that, apart from initialization, is accessed only through two standard atomic operations: P (proberen(测试) in Dutch) and V(verhogen(增加) in Dutch).
    信号量 S 是一个整数变量，除了在初始化之外，信号量S被访问只能依靠两个原子操作 P 和 V

信号量的实现

伪代码：

1
2
3
4
5
6
7
8
9
P(s){   //测试信号量是否大于0，若大于0，执行自减操作
    while(s<=0)
        do nothing;
    s--; 
}

V(s){   //使信号量增加
    s++;
}

信号量的使用

P V 操作必须成对出现

BINARY SEMAPHORE

  • 顾名思义，二值信号量的值只能是0或1，通常将其初始化为1，用于实现互斥锁的功能。

伪代码：

1
2
3
4
5
6
semphore mutex = 1;
process p{
    p(mutex);            //在 binary semaphore 中等价于 mutex 中的 lock
    //critical section
    v(mutex);            // unlock
}

COUNTIING SEMAPHORE

  • 计数信号量又称一般信号量，其取值可以是任意数值，用于控制并发 进程对共享资源的访问。
1
2
3
4
5
6
semaphore road = 2;
process Car{
    P(road);
    //pass the fork in road;
    V(road);
}

实践

要使用信号量，请先包含头文件 semaphore.h

sem_t：信号量的数据类型

int sem_init(sem_t *sem, int pshared, unsigned int val);
该函数第一个参数为信号量指针，第二个参数为信号量 类型（一般设置为0），第三个为信号量初始值。第二 个参数pshared为0时，该进程内所有线程可用，不为0 时不同进程间可用。

int sem_wait(sem_t *sem);
该函数申请一个信号量，当前无可用信号量则等待，有可用信号量时占用一个信号量，对信号量的值减1。(相当于 P 操作)

int sem_post(sem_t *sem);
该函数释放一个信号量，信号量的值加1。 (相当于 V 操作)

int sem_destory(sem_t *sem);
该函数用于销毁信号量

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
int tickAmount = 2;
sem_t sem;
void* ticketAgent(void *arg){
    sem_wait(&sem);
    int t = tickAmount;
    if(t > 0){
        printf("One ticket sold\n");
        t--;
    }
    else{
        printf("ticket sold out\n");
    }
    tickAmount = t;
    sem_post(&sem);
    pthread_exit(0);
}
int main()
{
    sem_init(&sem,0,1);
    pthread_t arr[2];
    for(int i=0;i<2;i++)
    {
        pthread_create(arr+i,NULL,ticketAgent,NULL);
    }
    for (int i = 0;i < 2; ++i) {
        pthread_join(arr[i],NULL);
    }
    sem_destroy(&sem);
}

信号量实现同步

  • 同步问题实质是将异步的并发进程按照某种顺序执行
  • 解决同步的本质就是要找到并发进程的交互点，利用P操作的等待特点来调节进程的执行速度
  • 通常初始值为0的信号量可以让进程直接进行等待状态，直到另一个进程唤醒他。

生产者消费者问题

  • 生产者(P)与消费者(C)共用一个缓冲区，生产者不能往 “满”的缓冲区中放产品，消费者不能从“空”的缓冲区中取产品。

使用信号量实现：当开始执行时假定BUFFER为空，那么 P 可以执行 C 不能执行，显然有两个控制流，故需要两个信号量来实现同步，因为 C 在程序开始时要等待，故将控制 C 的信号量初始值设为 0，当BUFFER满时，P将不能工作(进入忙式等待)，等实现等待的只有 P操作 ，所以将控制 P 的信号量初始值设为 BUFFER 的大小，对 P 进程执行 P操作时会使得对应的信号量减1，之后要通知 C进程，使用 V(C进程的信号量)通知C进程，C进程在 P操作后会使信号量减1，之后要通知 P进程，V(P进程的信号量)

单缓冲解决方案

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
Semaphore empty = 1;
Semaphore full = 0;

Producer{
    while(true){
        //product;
        P(empty);
        //pull the product into buffer
        V(full);
    }
}

Consumer{
    while(true){
        P(full);
        //pick product from buffer
        V(empty);
    }
}

THE BOUNDED-BUFFER(有界缓冲区) PROBLEM

OUT：指向消费者使用的位置

IN： 指向生产者使用的位置

IN 与 OUT 形成了循环队列，IN 相当于队头（生产者生产后IN前移），OUT相对于队尾（消费者消费后OUT前移），保证 IN >= OUT 及生产者生产的大于消费者消费的

伪代码： 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
item B[k];
semaphore empty = k;
semaphore full = 0;
semaphore mutex = 1;
int in = 0 , out = 0;
Process producer_i{
    make a product
    P(empty);
    P(mutex);
    B[in] = product;     //存在共享数据的修改，加锁
    in = (in + 1) % k;
    V(mutex);
    V(full);
}
Process consumer_i{
    P(full);
    P(mutex);
    product = B[out];     //存在共享数据的修改，加锁
    out = (out + 1) % k;
    V(mutex);
    V(empty);
    consume a product;
}

注意：

  • empty 和 full 的 PV 操作并不在同一进程中，称为同步信号量
  • mutex 的 PV操作在同一进程中，称为互斥信号量

苹果橘子问题

问题描述：

桌上有一只盘子，每次只能放入一只水果

爸爸专向盘子中放苹果，妈妈专向盘子中放桔子

儿子专等吃盘子中的桔子，女儿专等吃盘子里的苹果

这里爸爸 妈妈 相当于时生产者，儿子 女儿相当于是消费者

father 和 mother 是竞争关系，竞争盘子

father 和 daughter 是协作关系

mother 和 sun 是协作关系

伪代码:

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
semaphore sp = 1;        //盘子里允许放一个水果，竞争关系
semaphore sg1 = 0;       //盘子里没有橘子
semaphore sg2 = 0;       //盘子里没有苹果

Process father { 
削⼀个苹果；
 P(sp);      //占用盘子
 V(sg2);     //把苹果放⼊plate；通知儿子
} 
Process mother { 
剥⼀个桔⼦；
 P(sp); 
 V(sg1);     //把桔⼦放⼊plate；通知女儿
}
Process daughter { 
 P(sg2); 
 从plate中取苹果；
 V(sp); 
吃苹果；通知生产者可以继续生产了
}
Process son { 
 P(sg1); 
 从plate中取桔⼦；
 V(sp); 
吃桔⼦；通知生产者可以继续生产了
}

其本质还是生产者消费者模型

生产者消费者：

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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
#define BUFSIZE 10
sem_t sem_empty;           //生产者所使用的
sem_t sem_full;            //消费者所使用的
sem_t mutex;
int in = 0;  //表示队列中生产者生产的内存
int out = 0; //表示队列中要消费的内容
int arr[BUFSIZE];

void *prodecer(void* arg)  //生产者线程
{
    while (1) {
        sem_wait(&sem_empty);

        sem_wait(&mutex);
        arr[in] = 1;
        printf("prodecer : %d\n", in);
        in = (in + 1) % BUFSIZE;
        sem_post(&mutex);

        sem_post(&sem_full);
        sleep(1);
    }
}

void *consumer(void* arg)  //消费者线程
{
    printf("consumer -------------\n");
    while (1) {
        sem_wait(&sem_full);

        sem_wait(&mutex);
        printf("consumer: %d\n", out);
        out = (out + 1) % BUFSIZE;
        sem_post(&mutex);

        sem_post(&sem_empty);
        sleep(1);
    }
}

int main()
{
    sem_init(&sem_empty,0,BUFSIZE);
    //初始值是BUFSIZE保证了环形队列不会出现尾指针越过头指针的情况发生
    sem_init(&sem_full,0,0);
    sem_init(&mutex,0,1);

    pthread_t pro[2];
    pthread_t con[2];

    for(int i=0;i<2;i++) {
        pthread_create(pro + i, NULL, prodecer, NULL);
        pthread_create(con + i, NULL, consumer, NULL);
    }

    for (int i = 0; i < 2; ++i) {
        pthread_join(con[i],NULL);
        pthread_join(pro[i],NULL);
    }

    sem_destroy(&sem_empty);
    sem_destroy(&sem_full);
    sem_destroy(&mutex);
}

苹果橘子问题：

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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
enum plate{apple = 1, origen};

sem_t sp;   //父母竞争盘子
sem_t sp1;  //父亲通知女儿
sem_t sp2;   //母亲通知儿子
enum plate plate;

void* father(void *arg)
{
    sem_wait(&sp);
    plate = apple;
    printf("father\n");
    sem_post(&sp1);
}

void* mother(void *arg)
{
    sem_wait(&sp);
    plate = origen;
    printf("mother\n");
    sem_post(&sp2);
}

void* son(void *arg)
{
    sem_wait(&sp1);
    plate = 0;
    printf("son\n");
    sem_post(&sp);
}

void* daughter(void *arg)
{
    sem_wait(&sp2);
    plate = 0;
    printf("daughrer\n");
    sem_post(&sp);
}

int main()
{
    sem_init(&sp,0,1);
    sem_init(&sp1,0,0);
    sem_init(&sp2,0,0);

    pthread_t arr1[10] ;
    pthread_t arr2[10] ;
    pthread_t arr3[10] ;
    pthread_t arr4[10] ;
    for (int i = 0; i < 10; ++i) {
        pthread_create(arr1+i,NULL,father,NULL);
        pthread_create(arr2+i,NULL,mother,NULL);
        pthread_create(arr3+i,NULL,son,NULL);
        pthread_create(arr4+i,NULL,daughter,NULL);
    }
    for (int i = 0; i < 10; ++i) {
        pthread_join(arr1[i],NULL) ;
        pthread_join(arr2[i],NULL) ;
        pthread_join(arr3[i],NULL) ;
        pthread_join(arr4[i],NULL) ;
    }

    sem_destroy(&sp);
    sem_destroy(&sp1);
    sem_destroy(&sp2);
}

参考：信号量上下