1 
// 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 (mutex, 1 permit count semaphore) 的实现

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

3 回答 3

4

问题是x如果SubFetch无法获取锁，您会显式分配 -1 。这与解锁比赛。

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

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

于 2012-03-08T16:49:46.333 回答
3

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 的代码。那篇论文经受住了时间的考验，并获得了很多同行评议。你自己滚动一无所获。

于 2012-03-13T21:13:36.300 回答
0

使用三个线程 A、B 和 C 的情况如何。

这个场景的初始状态有：

  • 持有锁的线程A
  • 线程 B 还没有竞争锁
  • 将 C 线插入CompareWait()
  • x == -1从 C 获取锁失败开始
        美国广播公司
    ==============================================
     AddFetch()
     （所以 x == 0）
                       子提取（）
                       （所以 x == -1）

     x = 1

                       x = -1

     唤醒（）

此时无论 B 还是 C 是否畅通，都不会得到什么0时候的结果SubFetch()。

于 2012-03-08T16:55:56.060 回答