如何安排pthread_cond_signal()使其始终跟随另一个线程上的pthread_cond_wait()？

Question

我正在测试一个生产者 - 消费者场景,生产者在尝试写入已满的队列时会阻塞.我想测试生产者线程是否正确唤醒并在消费者从完整队列中读取后按预期工作*.队列写入API在检测队列已满时调用pthread_cond_wait(),并且读取API在从队列读取后发出条件变量信号.

如何确保序列3在我的测试环境中的任何其他操作序列中出现？

*是的我想分别测试这个有限的场景; 还有其他测试可以测试整个队列的功能,这个测试是除此之外的测试.

更多细节 -
有一个管理队列的互斥锁.有2个条件变量 - 一个用于信号写入(所有写入),一个用于信号读取(所有读取).在对读condvar queue_write API块,如果队列已满.该上写condvar queue_read API块,如果队列为空.所有信号都发生在互斥体的支持下.
队列中有很多细微差别,但为了设置这个问题的上下文,这是对队列功能的充分总结.

Answer 1

编辑（请注意，省略了 pthread 调用的错误处理）

您可以通过使用评论中提到的函数测试队列是否已满来实现此目的。对于这个答案，我假设它是bool is_queue_full(const queue*).

在您的测试用例中，当且仅当队列已满后，您可以通过创建生产者并创建消费者来保证场景 3。就像 bool is_queue_full(queue*); //不得使用互斥体本身，也许将其标记为仅供实习生使用

struct queue {
    /* Actual queue stuff */
    pthread_mutex_t queue_mutex;
    pthread_cond_t read_condvar;
    pthread_cond_t write_condvar;
};

void wait_until_queue_is_full (queue *q) {

    pthread_mutex_lock(&q->queue_mutex);
    while (!is_queue_full(q)){ //Use in loop because of Spurious wakeups
        pthread_cond_wait(&q->write_condvar,&q->queue_mutex);
    }
    pthread_mutex_unlock(&q->queue_mutex);
}

bool test_writer_woke_up(queue *q);

bool test_case(){
    queue *q = create_queue();
    producer *p = create_producer(q);

    wait_until_queue_is_full(q);

    return test_writer_woke_up(q); //or cache the result and destroy your queue, but if your testrunner process will quit anyway...
}

wait_until_queue_is_full只会检查队列是否已满，如果没有，将像任何读者一样等待，直到您的作者（又名生产者）将其填满。然后你的测试用例可以产生类似test_writer_woke_up void intern_consume_stuff(queue q);的消费者 /您的实习生函数从队列中获取内容，但不关心同步（又名互斥体和 condvar）*/

bool test_writer_woke_up(queue *q){
    pthread_mutex_lock(&q->queue_mutex); //Could be omitted in this testcase (together with the 1 unlock below of course)
    void intern_consume_stuff(queue *q);
    pthread_mutex_unlock(&q->queue_mutex); //Could be omitted in this testcase (together with the 1 lock above of course)
    pthread_cond_signal(&q->read_condvar);

    /* Adjust these as you like to give your producer/writer time to wake up and produce something 
    */
    unsigned retry_count = 5;
    unsigned sleep_time = 1; 

    //timed cond wait approach
    for (; retry_count > 0; --retry_count){
        pthread_mutex_lock(&q->queue_mutex);
        struct timespec ts;
        clock_gettime(CLOCK_REALTIME, &ts);
        ts.tv_sec += sleep_time;
        int timed_cond_rc = 0;
        while (!is_queue_full(q) && timed_cond_rc == 0) {
            timed_cond_rc = pthread_cond_timedwait(&q->write_condvar, &q->queue_mutex, &ts);
        }
        if (is_queue_full(q)) {
            pthread_mutex_unlock(&q->queue_mutex);
            return true;
        }
        assert(timed_cond_rc == ETIMEDOUT);
        continue;
    }
    return false;
}

如果使用了绝对时间等待，因为您必须重新计算相对时间，或者为了简化事情，您可以用这种简单的方法替换 for 循环

//naive busy approach
for (; retry_count > 0; --retry_count){
    pthread_mutex_lock(q->queue_mutex);
    const bool queue_full_result = is_queue_full(q);
    pthread_mutex_unlock(q->queue_mutex);

    if (queue_full_result){
        return true;
    } else {
        pthread_yield();
        sleep(sleep_time);
    }
}