Linux 3.0：futex-lock 死锁错误？

Question

// SubFetch(x,y) = atomically x-=y and return x (__sync_sub_and_fetch)
// AddFetch(x,y) = atomically x+=y and return x (__sync_add_and_fetch)
// CompareWait(x, y) = futex(&x, FUTEX_WAIT, y) wait on x if x == y
// Wake(x, y) = futex(&x, FUTEX_WAKE, y) wake up y waiters

struct Lock
{
Lock() : x(1) {}

void lock()
{
    while (true)
    {
        if (SubFetch(x, 1) == 0)
            return;

        x = -1;

        CompareWait(x, -1);
    }
}

void unlock()
{
    if (AddFetch(x, 1) == 1)
        return;

    x = 1;

    Wake(x, 1);
}

private:
    int x;
};

Linux 3.0 提供了一个名为 futex 的系统调用，许多并发实用程序都基于该调用，包括最近的 pthread_mutex 实现。每当您编写代码时，您都应该始终考虑使用现有实现还是自己编写代码是您项目的更好选择。

上面是一个基于futex和man futex(7)中语义描述的Lock（互斥量，1个允许计数信号量）的实现

它似乎包含一个死锁错误，在多个线程尝试锁定和解锁它几千次后，线程可能会进入 x == -1 并且所有线程都卡在 CompareWait 中的状态，但是没有人正在持有锁。

任何人都可以看到错误在哪里吗？

更新：我有点惊讶 futex(7)/semantics 如此糟糕。我完全重写了Lock如下......现在这样正确吗？

// CompareAssign(x,y,z) atomically: if (x == y) {x = z; ret true; } else ret false;

struct Lock
{
Lock() : x(0) {}

void lock()
{
    while (!CompareAssign(x, 0, 1))
        if (x == 2 || CompareAssign(x, 1, 2))
            CompareWait(x, 2);
}

void unlock()
{
    if (SubFetch(x, 1) == 0)
        return;

    x = 0;

    Wake(x, 1);
}

private:
int x;
};

这里的想法是x有以下三种状态：

0: unlocked
1: locked & no waiters
2: locked & waiters

Answer 1

问题在于，如果SubFetch 未能获得锁，您将显式-1 分配给x。这与解锁赛跑。

1. 线程 1 获取锁。 x==0。
2. 线程 2 尝试获取锁。 SubFetch 将 x 设置为 -1，然后线程 2 被挂起。
3. 线程 1 释放锁。 AddFetch 将 x 设置为 0，因此代码随后将 x 显式设置为 1 并调用 Wake。
4. 线程 2 唤醒并将 x 设置为 -1，然后调用 CompareWait。

线程 2 现在处于等待状态，x 设置为 -1，但周围没有人唤醒它，因为线程 1 已经释放了锁。

Answer 2

Ulrich Drepper 的论文“Futexes are tricky”中描述了基于 futex 的 Mutex 的正确实现

http://people.redhat.com/drepper/futex.pdf

它不仅包括代码，而且还非常详细地解释了为什么它是正确的。论文中的代码：

class mutex
{
 public:
 mutex () : val (0) { }
 void lock () {
   int c;
   if ((c = cmpxchg (val, 0, 1)) != 0)
     do {
       if (c == 2 || cmpxchg (val, 1, 2) != 0)
         futex_wait (&val, 2);
     } while ((c = cmpxchg (val, 0, 2)) != 0);
 }
 void unlock () {
//NOTE: atomic_dec returns the value BEFORE the operation, unlike your SubFetch !
   if (atomic_dec (val) != 1) {
     val = 0;
     futex_wake (&val, 1);
   }
 }
 private:
   int val;
};

将论文中的代码与您的代码进行比较，我发现了不同

你有

if (x == 2 || CompareAssign(x, 1, 2))

直接使用 futex 的值，而 Drepper 使用前一个 CompareAssign() 的返回值。这种差异可能只会影响性能。

您的解锁代码也不同，但在语义上似乎是等价的。

无论如何，我强烈建议您严格遵守 Drepper 的代码。那篇论文经受住了时间的考验，并获得了很多同行评议。自己滚动不会有任何收获。

Answer 3

这个场景有三个线程，A、B 和 C。

这个场景的初始状态有：

• 线程 A 持有锁
• 线程 B 还没有竞争锁
• CompareWait() 中的线程 C
• x == -1 从 C 获取锁失败时开始

甲乙丙 ============================================== AddFetch() （所以 x == 0） 子提取（） （所以 x == -1） x = 1 x = -1 唤醒（）

此时无论B还是C都解封了，SubFetch()时都不会得到0的结果。