futex_wake 怎么会返回 0答案

【问题标题】：How could futex_wake return 0futex_wake 怎么会返回 0
【发布时间】：2018-09-23 07:23:57
【问题描述】：

我使用 futex 实现了信号量。以下程序经常在 sem_post() 中的断言处失败。虽然返回值应该是 1，但它有时会返回 0。这怎么会发生？

当我使用 POSIX 信号量时，程序总是成功完成。

我使用的是 Linux 2.6.32-642.6.1.el6.x86_64

#include <cstdio>
#include <cstdlib>
#include <cassert>
#include <ctime>
#include <linux/futex.h>
#include <sys/syscall.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>
#include <errno.h>

using namespace std;

#if 0
 #include <semaphore.h>
#else
typedef volatile int sem_t;

void sem_init(sem_t* sem, int shared, int value)
{
    *sem = value;
}

void sem_post(sem_t* sem)
{
    while (1)
    {
        int value = *sem;
        if (__sync_bool_compare_and_swap(sem, value, value >= 0 ? value+1 : 1))
        {
            if (value < 0)      // had contender
            {
                int r = syscall(SYS_futex, sem, FUTEX_WAKE, 1, NULL, 0, 0);
                if (r != 1)
                    fprintf(stderr, "post r=%d err=%d sem=%d %d\n", r,errno,value,*sem);
                assert(r == 1);
            }
            return;
        }
    }
}

int sem_wait(sem_t* sem)
{
    while (1)
    {
        int value = *sem;
        if (value > 0   // positive means no contender
            && __sync_bool_compare_and_swap(sem, value, value-1))
            return 0;
        if (value <= 0
            && __sync_bool_compare_and_swap(sem, value, -1))
        {
            int r= syscall(SYS_futex, sem, FUTEX_WAIT, -1, NULL, 0, 0);
            if (!r) {
                assert(__sync_fetch_and_sub(sem, 1) > 0);
                return 0;
            }
            printf("wait r=%d errno=%d sem=%d %d\n", r,errno, value,*sem);
        }
    }
}

void sem_getvalue(sem_t* sem, int* value)
{
    *value = *sem;
}

#endif

// return current time in ns
unsigned long GetTime()
{
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    return ts.tv_sec*1000000000ul + ts.tv_nsec;
}

void Send(sem_t* sem, unsigned count)
{
    while (count--)
        sem_post(sem);
}

void Receive(sem_t* sem, unsigned count)
{
    while (count--)
        sem_wait(sem);
}


int main()
{
    sem_t* sem = reinterpret_cast<sem_t*>(mmap(NULL, sizeof(sem_t), PROT_READ|PROT_WRITE, MAP_ANONYMOUS|MAP_SHARED, -1, 0));
    assert(sem != MAP_FAILED);
    sem_init(sem, 1, 0);
    unsigned count = 10485760;

    int pid = fork();
    assert(pid != -1);
    if (!pid)   // child
    {
        Send(sem, count);
        _exit(EXIT_SUCCESS);
    }
    else    // parent
    {
        unsigned long t0 = GetTime();
        Receive(sem, count);
        printf("t=%g ms\n", (GetTime()-t0)*1e-6);
        wait(NULL);
        int v;
        sem_getvalue(sem, &v);
        assert(v == 0);
    }
}

【问题讨论】：

标签： linux semaphore futex

【解决方案1】：

__sync_bool_compare_and_swap(sem, value, -1) 和 __sync_fetch_and_sub(sem, 1) 似乎有问题。我们需要记住，sem_wait 可能会被多个线程同时调用（尽管在您的测试用例中只有一个线程调用它）。

如果我们能够承受繁忙轮询的开销，我们可以删除futex 并生成以下代码。它也比futex 版本更快（t=347 ms，而futex 版本为 t=914 ms）。

void sem_post(sem_t* sem)
{
    int value = __sync_fetch_and_add(sem, 1);
}

int sem_wait(sem_t* sem)
{
    while (1)
    {
        int value = *sem;
        if (value > 0) // positive means no contention
        {
            if (__sync_bool_compare_and_swap(sem, value, value-1)) {
                return 0; // success
            }
        }
        // yield the processor to avoid deadlock
        sched_yield();
    }
}

代码的工作方式如下：共享变量*sem 始终为非负数。当一个线程从 0 到 1 发布信号量时，所有等待该信号量的线程都可以尝试，但只有一个线程会在compare_and_swap 中成功。

【讨论】：

您的代码有效，但没有回答我原来的问题。我知道 sched_yield() 更快，但是在 cpu 使用更重要的情况下（没有足够的内核并且对延迟的要求较低）futex 是首选。在我的代码中，__sync_bool_compare_and_swap(sem, value, -1) 和 __sync_fetch_and_sub(sem,1) 在一个线程中，那么 assert() 怎么会失败呢？

【解决方案2】：

当sem 上没有线程等待时，对syscall(SYS_futex, sem, FUTEX_WAKE, 1, NULL, 0, 0) 的调用将返回0。在您的代码中，这是可能的，因为当 *sem 为负数时，您在 sem_post 中调用该 futex 行，如果没有任何线程实际上正在休眠，则可能是这种情况：

如果在调用sem_wait 时*sem 为零，您将继续执行__sync_bool_compare_and_swap(sem, value, -1)，它将*sem 设置为-1。然而，此时该线程尚未休眠。因此，当另一个线程调用sem_post 时（在调用sem_wait 的线程进入futex 系统调用之前）你的断言失败将会发生。

【讨论】：