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提供了一个系统调用，称为 富泰克斯, ，许多并发实用程序都基于它，包括最近的 pthread_mutex 实现。每当您编写代码时，您都应该始终考虑使用现有的实现还是自己编写代码对于您的项目来说是更好的选择。

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

它似乎包含一个死锁错误，在多个线程尝试锁定和解锁它数千次之后，线程可能会进入 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

问题是你明确 -1 分配给 x 如果 SubFetch 获取锁失败。这与解锁赛跑。

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的纸质中描述了基于Futex的互斥锁的互斥锁的“Futexes是棘手的”

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 || compareSign（x，1,2））

直接使用futex的值，而drepper使用上一个parameSign（）的返回值。这种差异可能只影响性能。

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

在任何情况下，我都会强烈建议您跟踪Drepper的代码给信函。那篇论文已经支撑了时间的考验，并收到了很多同行评审。你没有从自己的滚动中获得。

Answer 3

这个具有三个线程 A、B 和 C 的场景怎么样？

该场景的初始状态为：

• 线程A持有锁
• 线程 B 尚未争夺锁
• 螺纹C进入 CompareWait()
• x == -1 从C未能获取锁时开始

        A                  B                C
    ==============   ================  ===============
     AddFetch()
     (so x == 0)
                       SubFetch()
                       (so x == -1)

     x = 1

                       x = -1

     Wake()

此时无论B还是C解封，都不会得到如下结果 0 当他们 SubFetch().