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），即使 Dequeue() 肯定会耗尽而 Enqueue() 完全停止（忙于重试），fifo 仍然是满的。

在代码示例中，有 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 调用来管理您的读写线程。

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

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

Answer 2

如果它不是 sem_wait() 中的错误，那么是什么阻止了它 信号量？

您的程序的不耐烦阻止了它。不能保证Dequeue() 线程在给定的重试次数内被调度。如果你改变了

            assert(retry++ < 2);

到

            retry++;

您会看到程序有时会在重试 8 次甚至更多次后愉快地继续阅读器进程。

为什么 Enqueue 必须重试？

它必须重试仅仅是因为main线程的Dequeue()到那时还没有被调度。

出列速度比所有写入器的总和快得多。

你的程序表明这个假设有时是错误的。虽然显然Dequeue() 的执行时间比编写者的要短得多（由于usleep(t)），但这并不意味着Dequeue() 被完全公平调度程序更频繁地调度——这也是主要原因是您使用了不确定的调度策略。 man sched_yield:

sched_yield() 旨在与读取时间调度策略一起使用 （即 SCHED_FIFO 或 SCHED_RR）。将 sched_yield() 与 SCHED_OTHER 等非确定性调度策略是 未指定，很可能意味着您的应用程序设计已损坏。

如果你插入

    struct sched_param param = { .sched_priority = 1 };
    if (sched_setscheduler(0, SCHED_FIFO, &param) < 0)
        perror("sched_setscheduler");

在main() 的开头，您可能会看到您的程序按预期执行（当以适当的权限运行时）。