信号量 Semaphores

Semaphores 这个概念也是由迪杰斯特拉发明的。在这节你将会看到,信号量可以用来实现锁和条件变量。

定义

A semaphore is an object with an integer value that we can manipulate with two routines;

这两个 routines 在 POSIX 中分别叫是 sem_waitsem_post

声明一个信号量并且初始化它的值为 1。

#include <semaphore.h>

sem_t s;
sem_init(&s, 0, 1);
int sem_wait(sem_t * s) {
    // decrement the value of semaphores by one
    // wait if value of semaphores is negative
}

// Second, we can see that sem post() does not wait for some particular condition to hold like sem wait() does
int sem_post(sem_t * s) {
    // increment the value of semaphores by one
    // if there are one or more threads waiting, wake one
}

First, we can see that sem wait() will either return right away or it will cause the caller to suspend execution waiting for a subsequent post.
Second, we can see that sem post() does not wait for some particular condition to hold like sem wait() does.
Third, the value of the semaphore, when negative, is equal to the number of waiting threads.

Binary Semaphores (Locks)

sem_t m;
sem_init(&m, 0, X); // initialize semaphore to X; what should X be?

sem_wait(&m);
// critical section here
sem_post(&m);

一个问题, What should X be? —— 答案是 1。

  1. 当第一个线程调用 sem_wait() 的时候,它会把对信号量的值减一,此时等于 0;然后它完成了运行,退出,sem_post() 再把信号量的值加一
  2. 如果当第一个线程还没结束时,有第二个线程争抢 critical section,它首先调用 sem_wait(),然而这时候信号量的值由于被它减一了,等于了 -1,是负数,sem_wait() 得让它wait((putting itself to sleep and relinquishing the processor)
  3. 第一个线程此时结束了任务,sem_post() ,incrementing the value of the semaphore back to zero, and then wake the waiting thread

01
02

如果稍微换个场景,多于 3 个线程争抢 critical section,此时信号量的初始值等于几呢?——答案还是 1。
这也是得名于 Binary Semaphores 的原因。

看上面引用的第三点, the value of the semaphore, when negative, is equal to the number of waiting threads

信号量实现条件变量

sem_t s;

void * child(void * arg) {
    printf("child\n");
    sem_post(&s); // signal here: child is done
    return NULL;
}

int main(int argc, char * argv[]) {
    sem_init(&s, 0, X); // what should X be?
    printf("parent: begin\n");
    pthread_t c;
    Pthread_create(c, NULL, child, NULL);
    sem_wait(&s); // wait here for child
    printf("parent: end\n");
    return 0;
}

还是同样的问题, what should the initial value of this semaphore be? —— 答案是 0。

也是分两种情况来分析,第一种是主线程运行到 sem_wait 之后子线程执行,第二种是子线程先于主线程的 sem_wait 执行。

03

生产者/消费者问题

这里我只摘抄了 work 的代码

sem_t empty;
sem_t full;
sem_t mutex;

void * producer(void * arg) {
    int i;
    for (i = 0; i < loops; i++) {
        sem_wait(&empty);   // p1
        sem_wait(&mutex);   // p1.5
        put(i);             // p2
        sem_post(&mutex);   // p2.5
        sem_post(&full);    // p3
    }
}

void * consumer(void * arg) {
    int i;
    for (i = 0; i < loops; i++) {
        sem_wait(&full);    // c1
        sem_wait(&mutex);   // c1.5
        int tmp = get();    // c2
        sem_post(&mutex);   // c2.5
        sem_post(&empty);   // c3
        printf("%d\n", tmp);
    }
}

int main(int argc, char * argv[]) {
    sem_init(&empty, 0, MAX);   // MAX buffers are empty to begin with...
    sem_init(&full, 0, 0);      // ... and 0 are full
    sem_init(&mutex, 0, 1);     // mutex=1 because it is a lock
}

读写锁

typedef struct _rwlock_t {
    sem_t lock;             // binary semaphore (basic lock)
    sem_t writelock;        // used to allow ONE writer or MANY readers int readers;
    in critical section     // count of readers reading
} rwlock_t;

void rwlock_init(rwlock_t * rw) {
    rw->readers = 0;
    sem_init(&rw->lock, 0, 1);
    sem_init(&rw->writelock, 0, 1);
}

void rwlock_acquire_readlock(rwlock_t * rw) {
    sem_wait(&rw->lock);
    rw->readers++;
    if (rw->readers == 1)
        sem_wait(&rw->writelock); // first reader acquires writelock
    sem_post(&rw->lock);
}

void rwlock_release_readlock(rwlock_t * rw) {
    sem_wait(&rw->lock);
    rw->readers--;
    if (rw->readers == 0)
        sem_post(&rw->writelock); // last reader releases writelock
    sem_post(&rw->lock);
}

void rwlock_acquire_writelock(rwlock_t * rw) { sem_wait(&rw->writelock); }

void rwlock_release_writelock(rwlock_t * rw) { sem_post(&rw->writelock); }
  1. The important step then taken within rwlock acquire readlock() occurs when the first reader acquires the lock; in that case, the reader also acquires the write lock by calling sem wait() on the writelock semaphore, and then finally releasing the lock by calling sem post().
  2. once a reader has acquired a read lock, more readers will be allowed to acquire the read lock too;
  3. however, any thread that wishes to acquire the write lock will have to wait until all readers are finished; the last one to exit the critical section calls sem post() on “writelock” and thus enables a waiting writer to acquire the lock.

这个读写锁实现的有些问题,第一个是性能,可以看到,它的加锁行为非常多,读者也是要加锁的;第二个是公平性问题,在读者很多写者很少的时候,写者线程的等待时间会很长。*这个读写锁的实现使我联想到 SQLite 的实现,读者和写者遵循的规则也是类似的。*

哲学家就餐问题

如何实现信号量

哈哈终于讲到底层原语是怎么实现的了。这个方法叫做 Zemaphores

typedef struct __Zem_t {
    int value;
    pthread_cond_t cond;
    pthread_mutex_t lock;
} Zem_t;

// only one thread can call this
void Zem_init(Zem_t * s, int value) {
    s->value = value;
    Cond_init(&s->cond);
    Mutex_init(&s->lock);
}

void Zem_wait(Zem_t * s) {
    Mutex_lock(&s->lock);
    while (s->value <= 0)
        Cond_wait(&s->cond, &s->lock);
    s->value--;
    Mutex_unlock(&s->lock);
}

void Zem_post(Zem_t * s) {
    Mutex_lock(&s->lock);
    s->value++;
    Cond_signal(&s->cond);
    Mutex_unlock(&s->lock);
}

总结

引用原文的话吧

Semaphores are a powerful and flexible primitive for writing concurrent programs. Some programmers use them exclusively, shunning locks and condition variables, due to their simplicity and utility.

很多翻译把它翻译成信号量,因为它在锁和条件变量的基础上增加了”量“的功能,其实也不一定需要对吧,我们可以自行维护这个量的增删,并通知到与“量”相关的线程

引用

《Semaphores》

Comments
Write a Comment