您當前位置：首頁 > php開源 > php教程 > 進程間通信筆記（6）—POSIX信號量

進程間通信筆記（6）—POSIX信號量

來源：程序員人生發布時間：2016-12-06 10:47:55 閱讀次數：2805次

1.概述

信號量（semaphore）是1種提供不同進程間或1個給定進程不同線程之間的同步。
仍然分為POSIX信號量和SystemV信號量，這里先學習POSIX信號量。

POSIX信號量又分為著名信號量和基于內存的信號量（無名信號量）。區分在因而否需要使用POSIX IPC名字來標識。

NOTE：Linux操作系統中，POSIX著名信號量創建在虛擬文件系統中 1般掛載在/dev/shm，其名字以sem.somename的情勢存在。

2.信號量操作

早在學操作系統那會，就直到信號量的PV操作，總結1下大概是這么回事：

P操作，也叫做等待（wait）1個信號量，該操作會測試信號量的值，如果其值小于或等于0，將把當前進程/線程投入眠眠，當該信號量變得大于0后就將它減1。
偽代碼以下，這兩步必須是原子操作。

while(sem <=0);
sem--;

V操作，掛出（post）1個信號量，該操作將信號量值加1
偽代碼以下：

sem++;

信號量初始化的值的大小1般用于表示可用資源的數（例如緩沖區大小，以后代碼中體現）;如果初始化為1，則稱之2值信號量，2值信號量的功能就有點像互斥鎖了。

不同的是：互斥鎖的加鎖和解鎖必須在同1線程履行，而信號量的掛出卻沒必要由履行等待操作的線程履行。

3.POSIX信號量編程

POSIX信號量編程，離不開以下函數：

1.sem_init()初始化無名信號量

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
//pshared指定該信號量用于進程還是線程同步
//0-表示用于線程同步（所有線程可見）
//非0-表示用于進程同步（需要放在同享內存中）

2.sem_open()初始化著名信號量

#include <fcntl.h>           /* For O_* constants */
#include <sys/stat.h>        /* For mode constants */
#include <semaphore.h>
sem_t *sem_open(const char *name, int oflag);
sem_t *sem_open(const char *name, int oflag, mode_t mode, unsigned int value);
Link with -pthread.
//套路跟之前的消息隊列類似
//oflag可以設置為O_CREAT | O_EXCL (如果存在則返回毛病)。
//mode可以設置為0644 自己可讀寫，其他用戶和組內用戶可讀
//value表示信號量初始化的值

3.sem_wait()和sem_post()等待和掛出函數

#include <semaphore.h>
int sem_wait(sem_t *sem);//P操作
int sem_post(sem_t *sem);//V操作
Link with -pthread.

4.sem_timedwait()超時等待

 int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);
 //跟sem_wait()類似，不過，如果P操作不能立即履行，該函數將投入眠眠并等待abs_timeout中指定的時間。
 //如果超時照舊沒法履行P操作，則返回timeout毛病

3.1 基于內存的信號量（無名信號量）

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <time.h>
#include <assert.h>
#include <errno.h>
#include <signal.h>

sem_t sem;

#define handle_error(msg) \
    do { perror(msg); exit(EXIT_FAILURE); } while (0)//毛病處理宏

static void handler(int sig)//信號處理函數
{//信號處理函數中將信號掛出
    printf("in signal handler \r\n");
    write(STDOUT_FILENO, "sem_post() from handler\n", 24);
    if (sem_post(&sem) == -1) {
        write(STDERR_FILENO, "sem_post() failed\n", 18);
        _exit(EXIT_FAILURE);
    }
}

int main(int argc, char *argv[])
{
    struct sigaction sa;//注冊信號處理函數
    struct timespec ts;//超時
    int s;

    if (argc != 3) {
        fprintf(stderr, "Usage: %s <alarm-secs> <wait-secs>\n",
            argv[0]);
        exit(EXIT_FAILURE);
    }//第1個函數

    if (sem_init(&sem, 0, 0) == -1)
    handle_error("sem_init");


    sa.sa_handler = handler;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = 0;
    if (sigaction(SIGALRM, &sa, NULL) == -1)//設置定時器信號處理
        handle_error("sigaction");
    alarm(atoi(argv[1])); //開啟定時器

    if (clock_gettime(CLOCK_REALTIME, &ts) == -1)
        handle_error("clock_gettime");

    ts.tv_sec += atoi(argv[2]);


    while ((s = sem_timedwait(&sem, &ts)) == -1 && errno == EINTR)//超時等待信號量
        continue;       


    if (s == -1) {
        if (errno == ETIMEDOUT)
            printf("sem_timedwait() timed out\n");
        else
            perror("sem_timedwait");
    } else
        printf("sem_timedwait() succeeded\n");

    exit((s == 0) ? EXIT_SUCCESS : EXIT_FAILURE);
}

命令行參數有兩個，第1個是定時器發出SIGALRM的時間，1個是信號量阻塞等待的時間。
并且，在接收SIGALRM信號的處理函數中，將信號量掛出，這樣：
如果定時器發出SIGALRM時間大于信號阻塞等待的時間，sem_timedwait()將會返回timeout毛病。反之，如果在此之前進入信號處理函數，將信號量掛出，則將調用成功。如圖：

3.2 著名信號量

著名信號量要指定1個名字somename，打開成功后將以sem.somename的情勢存在于/dev/shm/目錄下。
書中用2值信號量做互斥同步，這里我直接用mutex。

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <time.h>
#include <assert.h>
#include <errno.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>

#define MAXSIZE 10
void *produce(void *); 
void *consume(void *);

typedef void *(*handler_t)(void *);//線程函數指針

struct _shared 
{
    int buff[MAXSIZE];

    sem_t *nempty;
    sem_t *nstored;
};//同享緩沖
typedef struct _shared shared;
shared shared_;
pthread_mutex_t mutex;//互斥鎖
int nitems;//生產和消費的數目

int main(int argc,char **argv)
{
    if (argc !=2)
    {
        printf("usage:namedsem <#items>\r\n");
        exit(-1);
    }
    nitems=atoi(argv[1]);
    const char *const SEM_NEMPTY = "nempty";//信號量的“名字”
    const char *const SEM_NSTORED = "nstored";//信號量的“名字”
    pthread_t tid_produce;//生產者線程 id
    pthread_t tid_consume;//消費者線程 id
    //初始化信號量和互斥鎖
    pthread_mutex_init(&mutex, NULL); 
    shared_.nstored=sem_open(SEM_NSTORED,O_CREAT|O_EXCL,0644,0);
    shared_.nempty=sem_open(SEM_NEMPTY,O_CREAT|O_EXCL,0644,MAXSIZE);
    memset(shared_.buff,0x00,MAXSIZE);

    //線程創建
    handler_t handler=produce;
    pthread_setconcurrency(2);
    if((pthread_create(&tid_produce,NULL,handler,NULL))<0)
    {
        printf("pthread_create error\r\n");
        exit(-1);
    }
//    sleep(5);
    handler=consume;
    if((pthread_create(&tid_consume,NULL,handler,NULL))<0)
    {
        printf("pthread_create error\r\n");
        exit(-1);
    }

    //線程回收
    pthread_join(tid_produce,NULL);
    pthread_join(tid_consume,NULL);

    //信號量鎖燒毀
    sem_unlink(SEM_NEMPTY);
    sem_unlink(SEM_NSTORED);
    pthread_mutex_destroy(&mutex);
    exit(0);
}



void *produce(void *args)
{
    int i;
    for(i=0;i<nitems;i++)
    {
        sem_wait(shared_.nempty);
        pthread_mutex_lock(&mutex);
        shared_.buff[i%MAXSIZE]=i;
        printf("add an item\r\n");
        pthread_mutex_unlock(&mutex);
        sem_post(shared_.nstored);
    }
    return NULL;
}


void *consume(void *args)
{
    int i;
    for(i=0;i<nitems;i++)
    {
        sem_wait(shared_.nstored);
        pthread_mutex_lock(&mutex);
        printf("consume an item\r\n");
        if(shared_.buff[i%MAXSIZE]!=i)
            printf("buff[%d]=%d\r\n",i,shared_.buff[i%MAXSIZE]);
        pthread_mutex_unlock(&mutex);
        sem_post(shared_.nempty);
    }
    return NULL;
}