pthread 信號量

本文介紹 semaphore.h 的相關 API ：

• 與 Named Semaphore 相關：sem_open, sem_close, sem_unlink .
• 與 Unnamed Semaphore 相關：sem_init, sem_destroy .

API

sem_init

函數原型：

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

做用：

sem_init() initializes the unnamed semaphore at the address pointed to by sem. The value argument specifies the initial value for the semaphore.

The pshared argument indicates whether this semaphore is to be shared between the threads of a process, or between processes.

If pshared has the value 0, then the semaphore is shared between the threads of a process, and should be located at some address that is visible to all threads (e.g., a global variable, or a variable allocated dynamically on the heap).

If pshared is nonzero, then the semaphore is shared between processes, and should be located in a region of shared memory (see shm_open(3), mmap(2), and shmget(2)). (Since a child created by fork(2) inherits its parent's memory mappings, it can also access the semaphore.) Any process that can access the shared memory region can operate on the semaphore using sem_post(3), sem_wait(3), and so on.

-- Manual on Ubuntu.

初始化一個匿名信號量。

其中，pshared 爲 0 表示該信號量在某個進程中的多個線程之間共享，該信號量應當可以被全部的線程訪問，例如是一個全局變量或者是在堆上動態分配的變量；若不爲 0 ，表示在進程間共享，那麼該信號量應該位於共享內存 shared memory 中。因爲 fork 操做產生的子進程，會自動繼承父進程的 memory mapping，因此這些 fork 進程也能訪問該信號量。

sem_destroy

函數原型：

int sem_destroy(sem_t *sem);

做用：

sem_destroy() destroys the unnamed semaphore at the address pointed to by sem. Only a semaphore that has been initialized by sem_init(3) should be destroyed using sem_destroy().

Destroying a semaphore that other processes or threads are currently blocked on (in sem_wait(3)) produces undefined behavior.

Using a semaphore that has been destroyed produces undefined results, until the semaphore has been reinitialized using sem_init(3).

-- Manual on Ubuntu.

銷燬一個已初始化的匿名信號量。

若是該信號量上還存在被阻塞的進程或線程，銷燬該信號量將是一個 Undefined behavior .

使用一個已銷燬的信號量一樣是一個 Undefined behavior .

sem_wait / trywait / timedwait

函數原型：

int sem_wait(sem_t *sem);
int sem_trywait(sem_t *sem);
int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);

做用：

sem_wait() decrements (locks) the semaphore pointed to by sem. If the semaphore's value is greater than zero, then the decrement proceeds, and the function returns, immediately. If the semaphore curently has the value zero, then the call blocks until either it becomes possible to perform the decrement (i.e., the semaphore value rises above zero), or a signal handler interrupts the call.

sem_trywait() is the same as sem_wait(), except that if the decrement cannot be immediately performed, then call returns an error (errno set to EAGAIN) instead of blocking.

sem_timedwait() is the same as sem_wait(), except that abs_timeout specifies a limit on the amount of time that the call should block if the decrement cannot be immediately performed. The abs_timeout argument points to a structure that specifies an absolute timeout in seconds and nanoseconds since the Epoch, 1970-01-01 00:00:00 +0000 (UTC). This structure is defined as follows:

struct timespec {
    time_t tv_sec;      /* Seconds */
    long   tv_nsec;     /* Nanoseconds [0 .. 999999999] */
};

If the timeout has already expired by the time of the call, and the semaphore could not be locked immediately, then sem_timedwait() fails with a timeout error (errno set to ETIMEDOUT).

If the operation can be performed immediately, then sem_timedwait() never fails with a timeout error, regardless of the value of abs_timeout. Furthermore, the validity of abs_timeout is not checked in this case.

-- Manual on Ubuntu.

若是信號量的值 sem_val 大於 0 ，sem_wait 會對 sem_val 減 1 ，函數當即返回。若是 sem_val 爲 0 ，sem_wait 將阻塞調用者（某個進程或線程），直到 sem_val 從新大於 0 或者調用者被 Linux 系統中的系統調用 signal() 函數中斷 。

對於 trywait ，若是對 sem_val 的減 1 操做不能完成，trywait 不會阻塞調用者，而是返回一個 error，並設置 errno 爲 EAGAIN .

對於 timedwait ，若是 sem_val 爲 0，則阻塞等待，當阻塞時長超過 abs_timeout 返回失敗 (errno 設置爲 ETIMEDOUT) .

sem_post

函數原型：

int sem_post(sem_t *sem);

做用：

sem_post() increments (unlocks) the semaphore pointed to by sem. If the semaphore's value consequently becomes greater than zero, then another process or thread blocked in a sem_wait(3) call will be woken up and proceed to lock the semaphore.

對信號量的值 sem_val 進行加 1 操做。若是從新大於 0 ，那麼喚醒某個阻塞的線程或進程，該線程/進程將從新鎖定該信號量。

sem_getvalue

函數原型：

int sem_getvalue(sem_t *sem, int *sval);

做用：

sem_getvalue() places the current value of the semaphore pointed to sem into the integer pointed to by sval.

獲取 sem 的值，並放到 sval 指向的內存。

Examples

Apple-Orange Problem

採用信號量機制，解決「蘋果-橙子」問題：一個能放 N（這裏 N 設爲 3）個水果的盤子，爸爸只往盤子裏放蘋果，媽媽只放橙子，女兒只吃盤子裏的橙子，兒子只吃蘋果。

代碼實現：

#include <stdio.h>
#include <semaphore.h>
#include <pthread.h>
const int n = 3; // 果盤的容量
const int k = 3; // 每一個人重複 k 次操做
sem_t apple, orange, empty;

void *father_worker(void *arg)
{
    int i = 0;
    for (i = 0; i < k; i++)
    {
        sem_wait(&empty);
        puts("[Father] apple++");
        sem_post(&apple);
    }
    return NULL;
}

void *mother_worker(void *arg)
{
    int i = 0;
    for (i = 0; i < k; i++)
    {
        sem_wait(&empty);
        puts("[Mother] orange++");
        sem_post(&orange);
    }
    return NULL;
}
void *son_worker(void *arg)
{
    int i = 0;
    for (i = 0; i < k; i++)
    {
        sem_wait(&apple);
        puts("\t[Son] apple--");
        sem_post(&empty);
    }
    return NULL;
}
void *daughter_worker(void *arg)
{
    int i = 0;
    for (i = 0; i < k; i++)
    {
        sem_wait(&orange);
        puts("\t[Daughter] orange--");
        sem_post(&empty);
    }
    return NULL;
}

int main()
{
    sem_init(&empty, 0, n);
    sem_init(&apple, 0, 0);
    sem_init(&orange, 0, 0);

    pthread_t father, mother, son, daughter;
    pthread_create(&father, NULL, father_worker, NULL);
    pthread_create(&mother, NULL, mother_worker, NULL);
    pthread_create(&son, NULL, son_worker, NULL);
    pthread_create(&daughter, NULL, daughter_worker, NULL);

    pthread_join(father, NULL), pthread_join(mother, NULL);
    pthread_join(son, NULL), pthread_join(daughter, NULL);

    sem_destroy(&empty), sem_destroy(&apple), sem_destroy(&orange);
    return 0;
}

PC Problem

解決上一篇文章 提到的 PC 問題。

• 全局變量

// global variables
#define CAPACITY 4 // buffer 的容量
#define N 8        // 依據題意，須要轉換 8 個字符
sem_t mutex1, empty1, full1;
sem_t mutex2, empty2, full2;
buffer_t buf1, buf2;

mutex1 和 mutex2 的做用至關於 pthead_mutex_t ，是爲了保證只有一個線程訪問 buf1 或者 buf2 。

empty1, empty2 的信號量值將被初始化爲 buff 的容量，表示可用的臨界資源的數量（指 buff 中空閒位置的個數）。

full1, full2 的信號量的值將被初始化爲 0，表示可用的臨界資源的數量（指 buff 中可取走的 item 的數量）。

• buffer_t

typedef struct
{
    char items[CAPACITY];
    int in, out;
} buffer_t;

void buffer_init(buffer_t *b) { b->in = b->out = 0; }
int buffer_is_full(buffer_t *b) { return ((b->in + 1) % CAPACITY) == (b->out); }
int buffer_is_empty(buffer_t *b) { return b->in == b->out; }
void buffer_put_item(buffer_t *buf, char item)
{
    buf->items[buf->in] = item;
    buf->in = (buf->in + 1) % CAPACITY;
}
char buffer_get_item(buffer_t *buf)
{
    char item = buf->items[buf->out];
    buf->out = (buf->out + 1) % CAPACITY;
    return item;
}

• producer

void *producer(void *arg)
{
    int i;
    char c;
    for (i = 0; i < N; i++)
    {

        sem_wait(&empty1);
        sem_wait(&mutex1);
        c = 'a' + i;
        buffer_put_item(&buf1, c);
        printf("[Producer] Put item [%c] into buf1.\n", c);
        sem_post(&mutex1);
        sem_post(&full1);
    }
    return NULL;
}

• consumer

void *consumer(void *arg)
{
    int i;
    char c;
    for (i = 0; i < N; i++)
    {

        sem_wait(&full2);
        sem_wait(&mutex2);
        c = buffer_get_item(&buf2);
        printf("\t[Consumer] Get item [%c] from buf2.\n", c);
        sem_post(&mutex2);
        sem_post(&empty2);
    }
    return NULL;
}

• calculator

void *calcultor(void *arg)
{
    int i;
    char c;
    for (i = 0; i < N; i++)
    {
        sem_wait(&full1);
        sem_wait(&mutex1);
        c = buffer_get_item(&buf1);
        sem_post(&mutex1);
        sem_post(&empty1);

        sem_wait(&empty2);
        sem_wait(&mutex2);
        buffer_put_item(&buf2, 'A' + c - 'a');
        sem_post(&mutex2);
        sem_post(&full2);
    }
}

• main函數

int main()
{
    buffer_init(&buf1), buffer_init(&buf2);
    sem_init(&mutex1, 0, 1), sem_init(&mutex2, 0, 1);
    sem_init(&empty1, 0, CAPACITY), sem_init(&empty2, 0, CAPACITY);
    sem_init(&full1, 0, 0), sem_init(&full2, 0, 0);

    pthread_t calc, prod, cons;
    pthread_create(&calc, NULL, calcultor, NULL);
    pthread_create(&prod, NULL, producer, NULL);
    pthread_create(&cons, NULL, consumer, NULL);

    pthread_join(prod, NULL), pthread_join(calc, NULL), pthread_join(cons, NULL);
    sem_destroy(&mutex1), sem_destroy(&mutex2);
    sem_destroy(&empty1), sem_destroy(&empty2);
    sem_destroy(&full1), sem_destroy(&full2);
    return 0;
}