c++ - sem_wait() 在 linux 上唤醒失败

Question

我有一个使用共享 FIFO 的实时应用程序。有多个写入进程和一个读取进程。数据定期写入 FIFO 并不断排出。理论上 FIFO 永远不会溢出，因为读取速度比所有写入器的总和还要快。但是，FIFO 会溢出。

我尝试重现该问题，最终得出以下(简化的)代码:

#include <stdint.h>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <cassert>
#include <pthread.h>
#include <semaphore.h>
#include <sys/time.h>
#include <unistd.h>


class Fifo
{
public:
    Fifo() : _deq(0), _wptr(0), _rptr(0), _lock(0)
    {
        memset(_data, 0, sizeof(_data));
        sem_init(&_data_avail, 1, 0);
    }

    ~Fifo()
    {
        sem_destroy(&_data_avail);
    }

    void Enqueue()
    {
        struct timeval tv;
        gettimeofday(&tv, NULL);
        uint64_t enq = tv.tv_usec + tv.tv_sec * 1000000;
        while (__sync_lock_test_and_set(&_lock, 1))
            sched_yield();
        uint8_t wptr = _wptr;
        uint8_t next_wptr = (wptr + 1) % c_entries;
        int retry = 0;
        while (next_wptr == _rptr)      // will become full
        {
            printf("retry=%u enq=%lu deq=%lu count=%d\n", retry, enq, _deq, Count());
            for (uint8_t i = _rptr; i != _wptr; i = (i+1)%c_entries)
                printf("%u: %lu\n", i, _data[i]);
            assert(retry++ < 2);
            usleep(500);
        }
        assert(__sync_bool_compare_and_swap(&_wptr, wptr, next_wptr));
        _data[wptr] = enq;
        __sync_lock_release(&_lock);
        sem_post(&_data_avail);
    }

    int Dequeue()
    {
        struct timeval tv;
        gettimeofday(&tv, NULL);
        uint64_t deq = tv.tv_usec + tv.tv_sec * 1000000;
        _deq = deq;
        uint8_t rptr = _rptr, wptr = _wptr;
        uint8_t next_rptr = (rptr + 1) % c_entries;
        bool empty = Count() == 0;
        assert(!sem_wait(&_data_avail));// bug in sem_wait?
        _deq = 0;
        uint64_t enq = _data[rptr];     // enqueue time
        assert(__sync_bool_compare_and_swap(&_rptr, rptr, next_rptr));
        int latency = deq - enq;        // latency from enqueue to dequeue
        if (empty && latency < -500)
        {
            printf("before dequeue: w=%u r=%u; after dequeue: w=%u r=%u; %d\n", wptr, rptr, _wptr, _rptr, latency);
        }
        return latency;
    }

    int Count()
    {
        int count = 0;
        assert(!sem_getvalue(&_data_avail, &count));
        return count;
    }

    static const unsigned c_entries = 16;

private:
    sem_t _data_avail;
    uint64_t _data[c_entries];
    volatile uint64_t _deq;     // non-0 indicates when dequeue happened
    volatile uint8_t _wptr, _rptr;      // write, read pointers
    volatile uint8_t _lock;     // write lock
};


static const unsigned c_total = 10000000;
static const unsigned c_writers = 3;

static Fifo s_fifo;


// writer thread
void* Writer(void* arg)
{
    for (unsigned i = 0; i < c_total; i++)
    {
        int t = rand() % 200 + 200;     // [200, 399]
        usleep(t);
        s_fifo.Enqueue();
    }
    return NULL;
}

int main()
{
    pthread_t thread[c_writers];
    for (unsigned i = 0; i < c_writers; i++)
        pthread_create(&thread[i], NULL, Writer, NULL);

    for (unsigned total = 0; total < c_total*c_writers; total++)
        s_fifo.Dequeue();
}

当 Enqueue() 溢出时，调试打印表明 Dequeue() 卡住(因为 _deq 不为 0)。 Dequeue() 唯一会卡住的地方是 sem_wait()。然而，由于 fifo 已满(也由 sem_getvalue() 确认)，我不明白这是怎么发生的。即使在多次重试(每次等待 500us)之后，fifo 仍然是满的，尽管 Dequeue() 肯定会耗尽而 Enqueue() 完全停止(忙于重试)。

在代码示例中，有 3 个写入器，每个写入器每 200-400us 写入一次。在我的电脑上(8 核 i7-2860，运行 centOS 6.5 内核 2.6.32-279.22.1.el6.x86_64，g++ 4.47 20120313)，代码会在几分钟内失败。我也尝试了其他几个 centOS 系统，它也以同样的方式失败。

我知道让 fifo 变大可以降低溢出概率(事实上，程序仍然失败，c_entries=128)，但在我的实时应用程序中，入队-出队延迟有硬约束，因此必须排出数据迅速地。如果它不是 sem_wait() 中的错误，那么是什么阻止它获取信号量？

附言如果我更换

        assert(!sem_wait(&_data_avail));// bug in sem_wait?

与

        while (sem_trywait(&_data_avail) < 0) sched_yield();

然后程序运行正常。所以看起来 sem_wait() 和/或调度程序有问题。

Answer 1

您需要结合使用 sem_wait/sem_post 调用才能管理您的读写线程。

您的入队线程仅执行 sem_post 而您的出队线程仅执行 sem_wait 调用。您需要将 sem_wait 添加到入队线程，并在出队线程上添加 sem_post。

很久以前，我实现了让多个线程/进程能够读取一些共享内存并且只有一个线程/进程写入共享内存的能力。我使用了两个信号量，一个写信号量和一个读信号量。读线程会一直等到写信号量没有被设置，然后它会设置读信号量。写入线程将设置写入信号量，然后等待读取信号量未设置。读写线程在完成任务后将取消设置信号量。读信号量一次可以有n个线程锁定读信号量，而写信号量一次可以由一个线程锁定。