Linux多线程实践（五 ）Posix信号量和互斥锁解决生产者消费者问题


Posix信号量和System V信号量的一点区别：

system v 信号量只能用于进程间同步，而posix 信号量除了可以进程间同步，还可以线程间同步。system v 信号量每次PV操作可以是N，但Posix 信号量每次PV只能是1。除此之外，posix 信号量还有命名和匿名之分（man 7 sem_overview）：

Posix 信号量

有名信号量

无名信号量

sem_open

sem_init

sem_close

sem_destroy

sem_unlink

 

sem_wait

sem_post
有名信号量

 

 

#include            /* For O_* constants */  
#include         /* For mode constants */  
#include   
sem_t *sem_open(const char *name, int oflag);  
sem_t *sem_open(const char *name, int oflag, mode_t mode, unsigned int value);  
int sem_close(sem_t *sem);  
int sem_unlink(const char *name);

 

与Posix类IPC用法类似: 名字以/somename形式标识，且只能有一个/ ，并且总长不能超过NAME_MAX-4 (i.e., 251)。

Posix有名信号量需要用sem_open 函数创建或打开，PV操作分别是sem_wait 和 sem_post，可以使用sem_close 关闭，删除用sem_unlink。

有名信号量用于不需要共享内存的进程间同步(可以通过名字访问), 类似System V 信号量。

匿名信号量

 

#include   
int sem_init(sem_t *sem, int pshared, unsigned int value);  
int sem_destroy(sem_t *sem);

 

匿名信号量只存在于内存中, 并要求使用信号量的进程必须可以访问内存; 这意味着他们只能应用在同一进程中的线程, 或者不同进程中已经映射相同内存内容到它们的地址空间中的线程.

存放在一块共享内存中，如果是线程共享，这块区域可以是全局变量；如果是进程共享，可以是system v 共享内存（shmget 创建，shmat 映射），也可以是 posix 共享内存（shm_open 创建，mmap 映射）。

匿名信号量必须用sem_init 初始化，sem_init 函数其中一个参数pshared决定了线程共享（pshared=0）还是进程共享（pshared!=0），也可以用sem_post 和sem_wait 进行操作，在共享内存释放前，匿名信号量要先用sem_destroy 销毁。

Posix信号量PV操作
int sem_wait(sem_t *sem);   //P操作  
int sem_post(sem_t *sem);   //V操作

wait操作实现对信号量的减1, 如果信号量计数原先为0则会发生阻塞;

post操作将信号量加1, 在调用sem_post时, 如果在调用sem_wait中发生了进程阻塞, 那么进程会被唤醒并且sem_post增1的信号量计数会再次被sem_wait减1;

Posix互斥锁

 

#include   
int pthread_mutex_init(pthread_mutex_t *mutex,   
                       const pthread_mutexattr_t *mutexattr);       //互斥锁初始化, 注意:函数成功执行后，互斥锁被初始化为未锁住状态。  
int pthread_mutex_lock(pthread_mutex_t *mutex); //互斥锁上锁  
int pthread_mutex_trylock(pthread_mutex_t *mutex);  //互斥锁判断上锁  
int pthread_mutex_unlock(pthread_mutex_t *mutex);   //互斥锁解锁  
int pthread_mutex_destroy(pthread_mutex_t *mutex);  //消除互斥锁

 

互斥锁是用一种简单的加锁方法来控制对共享资源的原子操作。这个互斥锁只有两种状态，也就是上锁／解锁，可以把互斥锁看作某种意义上的全局变量。在同一时刻只能有一个线程掌握某个互斥锁，拥有上锁状态的线程能够对共享资源进行操作。若其他线程希望上锁一个已经被上锁的互斥锁，则该线程就会阻塞，直到上锁的线程释放掉互斥锁为止。可以说，这把互斥锁保证让每个线程对共享资源按顺序进行原子操作。

其中，互斥锁可以分为快速互斥锁(默认互斥锁)、递归互斥锁和检错互斥锁。这三种锁的区别主要在于其他未占有互斥锁的线程在希望得到互斥锁时是否需要阻塞等待。快速锁是指调用线程会阻塞直至拥有互斥锁的线程解锁为止。递归互斥锁能够成功地返回，并且增加调用线程在互斥上加锁的次数，而检错互斥锁则为快速互斥锁的非阻塞版本，它会立即返回并返回一个错误信息。

解决生产者消费者问题

关于这个经典问题就不再叙述了，自行baidu吧。我们使用伪代码理一下流程，然后使用上面的API实现就很容易了。

 

简单概述： 一组生产者，一组消费者，公用n个环形缓冲区 。

在这个问题中，不仅生产者与消费者之间要同步，而且各个生产者之间、各个消费者之间还必须互斥地访问缓冲区。

定义四个信号量：

empty——表示缓冲区是否为空，初值为n。

full——表示缓冲区中是否为满，初值为0。

mutex1——生产者之间的互斥信号量，初值为1。

mutex2——消费者之间的互斥信号量，初值为1。

设缓冲区的编号为1～n-1，定义两个指针in和out，分别是生产者进程和消费者进程使用的指针，指向下一个可用的缓冲区。

生产者进程 while(TRUE)
{       
     生产一个产品; 
     P(empty); 
     P(mutex1)； 
     产品送往buffer（in）；  
 in=(in+1)mod n；  
 V(mutex1);  
 V(full); 
}

消费者进程 
while(TRUE)
{   
   P(full)  
   P(mutex2)；  
   从buffer（out）中取出产品；
out=(out+1)mod n；  
  V（mutex2）；  
  V(empty);  
  消费该产品;
｝
#include 
#include 
#include 
#include 

#include 
#include 
#include 
#include 

#define ERR_EXIT(m) \
        do \
        { \
                perror(m); \
                exit(EXIT_FAILURE); \
        } while(0)

#define CONSUMERS_COUNT 1
#define PRODUCERS_COUNT 1
#define BUFFSIZE 10

int g_buffer[BUFFSIZE];

unsigned short in = 0;
unsigned short out = 0;
unsigned short produce_id = 0;
unsigned short consume_id = 0;

sem_t g_sem_full;
sem_t g_sem_empty;
pthread_mutex_t g_mutex;

pthread_t g_thread[CONSUMERS_COUNT + PRODUCERS_COUNT];

void *consume(void *arg)
{
    int i;
    int num = (int)arg;
    while (1)
    {
        printf("%d wait buffer not empty\n", num);
        sem_wait(&g_sem_empty);
        pthread_mutex_lock(&g_mutex);

        for (i = 0; i < BUFFSIZE; i++)
        {
            printf("%02d ", i);
            if (g_buffer[i] == -1)
                printf("%s", "null");
            else
                printf("%d", g_buffer[i]);

            if (i == out)
                printf("\t<--consume");

            printf("\n");
        }
        consume_id = g_buffer[out];
        printf("%d begin consume product %d\n", num, consume_id);
        g_buffer[out] = -1;
        out = (out + 1) % BUFFSIZE;
        printf("%d end consume product %d\n", num, consume_id);
        pthread_mutex_unlock(&g_mutex);
        sem_post(&g_sem_full);
        sleep(1);
    }
    return NULL;
}

void *produce(void *arg)
{
    int num = (int)arg;
    int i;
    while (1)
    {
        printf("%d wait buffer not full\n", num);
        sem_wait(&g_sem_full);
        pthread_mutex_lock(&g_mutex);
        for (i = 0; i < BUFFSIZE; i++)
        {
            printf("%02d ", i);
            if (g_buffer[i] == -1)
                printf("%s", "null");
            else
                printf("%d", g_buffer[i]);

            if (i == in)
                printf("\t<--produce");

            printf("\n");
        }

        printf("%d begin produce product %d\n", num, produce_id);
        g_buffer[in] = produce_id;
        in = (in + 1) % BUFFSIZE;
        printf("%d end produce product %d\n", num, produce_id++);
        pthread_mutex_unlock(&g_mutex);
        sem_post(&g_sem_empty);
        sleep(5);
    }
    return NULL;
}

int main(void)
{
    int i;
    for (i = 0; i < BUFFSIZE; i++)
        g_buffer[i] = -1;

    sem_init(&g_sem_full, 0, BUFFSIZE);
    sem_init(&g_sem_empty, 0, 0);

    pthread_mutex_init(&g_mutex, NULL);


    for (i = 0; i < CONSUMERS_COUNT; i++)
        pthread_create(&g_thread[i], NULL, consume, (void *)i);

    for (i = 0; i < PRODUCERS_COUNT; i++)
        pthread_create(&g_thread[CONSUMERS_COUNT + i], NULL, produce, (void *)i);

    for (i = 0; i < CONSUMERS_COUNT + PRODUCERS_COUNT; i++)
        pthread_join(g_thread[i], NULL);

    sem_destroy(&g_sem_full);
    sem_destroy(&g_sem_empty);
    pthread_mutex_destroy(&g_mutex);

    return 0;
}
这里是演示线程间同步，现在上述程序生产者消费者各一个线程，但生产者睡眠时间是消费者的5倍，故消费者会经常阻塞在sem_wait(&g_sem_empty) 上面，因为缓冲区经常为空，可以将PRODUCTORS_COUNT 改成5，即有5个生产者线程和1个消费者线程，而且生产者睡眠时间还是消费者的5倍，从动态输出可以看出，基本上就动态平衡了，即5个生产者一下子生产了5份东西，消费者1s消费1份，刚好在生产者继续生产前消费完。

自旋锁和读写锁

（1）自旋锁(Spin lock)

自旋锁与互斥锁有点类似，只是自旋锁不会引起调用者睡眠，如果自旋锁已经被别的执行单元保持，调用者就一直循环在那里看是 否该自旋锁的保持者已经释放了锁，"自旋"一词就是因此而得名。其作用是为了解决某项资源的互斥使用。因为自旋锁不会引起调用者睡眠，所以自旋锁的效率远 高于互斥锁。虽然它的效率比互斥锁高，但是它也有些不足之处：
1、自旋锁一直占用CPU，他在未获得锁的情况下，一直运行－－自旋，所以占用着CPU，如果不能在很短的时 间内获得锁，这无疑会使CPU效率降低。
2、在用自旋锁时有可能造成死锁，当递归调用时有可能造成死锁，调用有些其他函数也可能造成死锁，如 copy_to_user()、copy_from_user()、kmalloc()等。
因此我们要慎重使用自旋锁，自旋锁只有在内核可抢占式或SMP的情况下才真正需要，在单CPU且不可抢占式的内核下，自旋锁的操作为空操作。自旋锁适用于锁使用者保持锁时间比较短的情况下。
自旋锁的用法如下：
首先定义：spinlock_t x;
然后初始化：spin_lock_init(spinlock_t *x); //自旋锁在真正使用前必须先初始化
在2.6.11内核中将定义和初始化合并为一个宏：DEFINE_SPINLOCK(x)

获得自旋锁：spin_lock(x); //只有在获得锁的情况下才返回，否则一直“自旋”
spin_trylock(x); //如立即获得锁则返回真，否则立即返回假
释放锁： spin_unlock(x);

（2）读写锁

1、只要没有线程持有给定的读写锁用于写，那么任意数目的线程可以持有读写锁用于读
2、仅当没有线程持有某个给定的读写锁用于读或用于写时，才能分配读写锁用于写
3、读写锁用于读称为共享锁，读写锁用于写称为排它锁
pthread_rwlock_init
pthread_rwlock_destroy
int pthread_rwlock_rdlock
int pthread_rwlock_wrlock
int pthread_rwlock_unlock

关于linux的锁的相关问题可以参考IBM的文档：

http://www.ibm.com/developerworks/cn/linux/l-cn-mthreadps/index.html

相关内容

    暂无相关文章