C/C++ 线程条件的神奇差异

Question

我想用 C/C++ 编写简单的多线程应用程序。函数 funProducent 产生 100 个值，如果随机生成的值在给定范围内，则将 char 添加到缓冲区。函数 funKonzument 从缓冲区获取值。这是我的代码：

#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>

#define BUFFER_LIMIT 20

struct struktura{
    pthread_mutex_t mutex;
    pthread_cond_t bufferNotFull;
    pthread_cond_t bufferNotEmpty;
    int bufferIndex;
    char * buffer;
    int junk;
};

void * funProducent(void *arg){
    struktura * data = (struktura *) arg;
    int i = 0;
    while (i < 100) {
        pthread_mutex_lock(&data->mutex);
        if(data->bufferIndex == BUFFER_LIMIT - 1){
            pthread_cond_wait(&data->bufferNotFull, &data->mutex);
        }
        int randomValue = (rand() % 20) + 1;
        if( randomValue < 13 ){
            data->buffer[++data->bufferIndex] = 'a';
            printf("%2d : Producent at index %d added %c\n", i, data->bufferIndex, data->buffer[data->bufferIndex]);
            pthread_cond_signal(&data->bufferNotEmpty);
        } else {
            data->junk++;
        }
        pthread_mutex_unlock(&data->mutex);
        i++;
    }
    printf("producent is done\n");
}

void * funKonzument(void *arg){
    struktura * data = (struktura *) arg;
    int i = 0;
    while (i + data->junk < 100) {  
        printf("%d\n", i + data->junk);
        pthread_mutex_lock(&data->mutex);
        if(data->bufferIndex < 0){
            pthread_cond_wait(&data->bufferNotEmpty, &data->mutex);
        }
        printf("%2d : Konzument at index %d consumed %c\n", i, data->bufferIndex, data->buffer[data->bufferIndex]);
        data->bufferIndex--;
        pthread_cond_signal(&data->bufferNotFull);
        pthread_mutex_unlock(&data->mutex);
        i++;
    }
    printf("konzument is done\n");
}

int main(int argc, char** argv) {

pthread_t threadProducent, threadKonzument;
struktura threadData;
threadData.buffer = (char *) malloc(sizeof(char) * BUFFER_LIMIT);
threadData.bufferIndex = -1;
threadData.bufferNotFull = PTHREAD_COND_INITIALIZER;
threadData.bufferNotEmpty = PTHREAD_COND_INITIALIZER;
threadData.mutex = PTHREAD_MUTEX_INITIALIZER;
threadData.junk = 0;

pthread_create(&threadProducent, NULL, funProducent, &threadData);
pthread_create(&threadKonzument, NULL, funKonzument, &threadData);

pthread_join(threadProducent, NULL);
pthread_join(threadKonzument, NULL);

free(threadData.buffer);
pthread_mutex_destroy(&threadData.mutex);
pthread_cond_destroy(&threadData.bufferNotFull);
pthread_cond_destroy(&threadData.bufferNotEmpty);
return 0;
}

当我尝试运行此代码时，有时它会卡在 funKonzument 中：

pthread_cond_wait(&data->bufferNotEmpty, &data->mutex);

但是...当我在 funProducent 方法中更改条件时：

if( randomValue < 13 )

到

if( randomValue > 8 )

一切正常。谁能解释一下这两个条件之间有什么神奇的区别？

Answer 1

您可能正遭受虚假唤醒和垃圾计数器的一些问题。我刚刚删除了那个计数器并添加了一个 cond 等待循环函数（和一个小锁上下文管理器），然后挂起似乎已经停止了。

#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <stdexcept>
#include <functional>

#define BUFFER_LIMIT 20

struct struktura{
    pthread_mutex_t mutex;
    pthread_cond_t bufferNotFull;
    pthread_cond_t bufferNotEmpty;
    int bufferIndex;
    char * buffer;
};

// a lock context manager
class mlock {
    pthread_mutex_t* mtx;
public:
    mlock(pthread_mutex_t& Mtx) :
        mtx(&Mtx)
    {
        int rv=pthread_mutex_lock(mtx);
        if(rv) throw std::runtime_error(std::to_string(rv));
    }
    mlock(const mlock&) = delete;
    mlock(mlock&&) = delete;
    mlock& operator=(const mlock&) = delete;
    mlock& operator=(mlock&&) = delete;
    ~mlock() {
        pthread_mutex_unlock(mtx);
    }
};

// silly loop to take care of spurious wakes
void cwait(pthread_cond_t& c, pthread_mutex_t& m, std::function<bool()> f) {
    while(f()) pthread_cond_wait(&c, &m);
}

void* funProducent(void *arg){
    struktura* data = static_cast<struktura*>(arg);
    int i = 0;
    while(i < 100) {
        mlock dummy(data->mutex);
        cwait(data->bufferNotFull, data->mutex, [&](){return data->bufferIndex == BUFFER_LIMIT - 1;});

        int randomValue = (rand() % 20) + 1;
        if( randomValue < 13 ){
            data->buffer[++data->bufferIndex] = 'a';
            printf("%2d : Producent at index %d added %c\n", i, data->bufferIndex, data->buffer[data->bufferIndex]);
            i++;
            pthread_cond_signal(&data->bufferNotEmpty);
        }
    }
    printf("producent is done\n");
    return nullptr;
}

void* funKonzument(void *arg){
    struktura* data = static_cast<struktura*>(arg);
    int i = 0;
    while(i < 100) {
        mlock dummy(data->mutex);
        cwait(data->bufferNotEmpty, data->mutex, [&](){return data->bufferIndex<0;});

        printf("\t\t\t%2d : Konzument at index %d consumed %c\n", i, data->bufferIndex, data->buffer[data->bufferIndex]);
        data->bufferIndex--;
        i++;
        pthread_cond_signal(&data->bufferNotFull);
    }
    printf("\t\t\tkonzument is done\n");
    return nullptr;
}

int main() {
    pthread_t threadProducent, threadKonzument;
    struktura threadData;
    threadData.buffer = (char *) malloc(sizeof(char) * BUFFER_LIMIT);
    threadData.bufferIndex = -1;
    threadData.bufferNotFull = PTHREAD_COND_INITIALIZER;
    threadData.bufferNotEmpty = PTHREAD_COND_INITIALIZER;
    threadData.mutex = PTHREAD_MUTEX_INITIALIZER;

    pthread_create(&threadProducent, NULL, funProducent, &threadData);
    pthread_create(&threadKonzument, NULL, funKonzument, &threadData);

    pthread_join(threadProducent, NULL);
    pthread_join(threadKonzument, NULL);

    free(threadData.buffer);
    pthread_mutex_destroy(&threadData.mutex);
    pthread_cond_destroy(&threadData.bufferNotFull);
    pthread_cond_destroy(&threadData.bufferNotEmpty);
    return 0;
}