锁 条件变量 信号量

锁-条件变量-信号量

用一个例子引入，用于输出嵌套深度不超过n的括号序列。

void Tproduce() {
  while (1) {
    mutex_lock(&lk);
    if (!CAN_PRODUCE) {
      cond_wait(&cv, &lk);
    }
    printf("("); count++;
    cond_signal(&cv);
    mutex_unlock(&lk);
  }
}

void Tconsume() {
  while (1) {
    mutex_lock(&lk);
    if (!CAN_CONSUME) {
      cond_wait(&cv, &lk);
    }
    printf(")"); count--;
    cond_signal(&cv);
    mutex_unlock(&lk);
  }
}

上面代码错误的原因在于被唤醒时，没有循环检查条件，导致产生了不合法的序列

修改了之后呢？

void Tproduce() {
  while (1) {
    mutex_lock(&lk);
    while (!CAN_PRODUCE) {
      cond_wait(&cv, &lk);
    }
    printf("("); count++;
    cond_signal(&cv);
    mutex_unlock(&lk);
  }
}

void Tconsume() {
  while (1) {
    mutex_lock(&lk);
    while (!CAN_CONSUME) {
      cond_wait(&cv, &lk);
    }
    printf(")"); count--;
    cond_signal(&cv);
    mutex_unlock(&lk);
  }
}

这样的代码在运行时会阻塞，什么也输出不了，发生了死锁。
原因在于使用cond_signal时，只会随机唤醒一个线程，生产者和消费者都有可能被唤醒。
一个极端的例子n=1，生产者消费者各有两个P1, P2, C1, C2
可能的执行序列为
P1执行，输出'('，此时P2，C1，C2还未执行，因此没有线程被唤醒，再次进入循环后wait
P2执行，wait
C1执行，输出')'，唤醒P2，再次进入循环wait
C2执行，wait
P2执行，输出'('，唤醒P1，再次进入循环wait
P1执行，wait
至此，所有线程都在wait，死锁。
根本原因就是P1丢失了一次唤醒并且唤醒的进程是随机的，改正的方法就是使用cond_broadcast，唤醒所有线程。

void Tproduce() {
  while (1) {
    mutex_lock(&lk);
    while (!CAN_PRODUCE) {
      cond_wait(&cv, &lk);
    }
    printf("("); count++;
    cond_broadcast(&cv);
    mutex_unlock(&lk);
  }
}

void Tconsume() {
  while (1) {
    mutex_lock(&lk);
    while (!CAN_CONSUME) {
      cond_wait(&cv, &lk);
    }
    printf(")"); count--;
    cond_broadcast(&cv);
    mutex_unlock(&lk);
  }
}

到这里终于得到了勉强正确的代码，但broadcast显然是不太合理的。

但将信号量看做条件变量的一种特例，却不存在这样的问题。信号量会准确地执行V操作，不会丢掉唤醒机会。信号量的伪代码如下

void P(sem_t *sem){
    wait_until(sem.count > 0){
        sem.count--;
    }
}

void V(sem_t *sem){
    sem.count++;
}

P失败时立即睡眠，V后唤醒等待线程。
count = 1就变成了互斥锁

另一个例题如何打印随机的<><和><>，中间用_隔开
使用锁和条件变量时，将打印的过程视为在状态机上输入字符(这里是输出)，进入后一个可行的状态。

mutex_t lk = MUTEX_INIT();
cond_t cv = COND_INIT();
enum {A=1,B,C,D,E,F};
int state = A;
struct mov{
    int from, ch, to;
};
struct mov moves[] = {
    {A,'<',B},
    {B,'>',C},
    {C,'<',D},
    {A,'>',E},
    {E,'<',F},
    {F,'>',D},
    {D,'_',A},
};
const char* roles = ".<<<<<>>>>___";
int sz;
void fish(int id){
    char role = roles[id];
    while(1){
        mutex_lock(&lk);
        for(;;){
            struct mov* p = moves;
            for(;p!=moves+sz;p++){
                if(p->from == state && p->ch == role){
                    break;
                }
            }
            if(p==moves+sz){
                cond_wait(&cv,&lk);
            }else{
                state = p->to;
                printf("%c",role);
                cond_broadcast(&cv);
                break;
            }
        }
        mutex_unlock(&lk);
    }
}
int main(){
    sz = sizeof(moves)/sizeof(struct mov);
    int n = strlen(roles);
    for(int i=0;i<n;i++){
        create(fish);
    }
}

但是，如果我们想用信号量实现呢？似乎不是很直观，因为信号量只能保证互斥和简单的同步，而不能保证状态机的正确顺序。
此时PV操作显然不能对状态进行，而要对待输出字符进行，对每个字符维护一个条件变量，P操作尝试下一个输出字符，到达可行状态后，在这个状态所有能前进的字符中选择一个，进行V操作。
这个不太好想，要用到计算图的思想。

void fish(int id){
    char role = roles[id];
    for(;;){
        P(&sems[role]);
        struct mov* p = moves;
        for(;p!=moves+sz;p++){
            if(p->from == state && p->ch == role){
                break;
            }
        }
        state = p->to;
        printf("%c",role);
        p = moves;
        int choices[4],n=0;
        for(;p!=moves+sz;p++){
            if(p->from == state){
                choices[n++] = p->ch;
            }
        }
        int c = rand()%n;
        V(&sems[choices[c]]);
    }
}

感觉记性真的越来越不好了，有些明明已经较深入学习过的内容还是忘了。