条件下减少原子计数器

Question

我希望能够实现以下内容：

        inline void DecrementPendingWorkItems()
        {
            if(this->pendingWorkItems != 0) //make sure we don't underflow and get a very high number
            {
                ::InterlockedDecrement(&this->pendingWorkItems);
            }
        }

我该如何做，以使这两个操作作为一个原子块执行，而不使用锁？

- Ghita

---

与您实际问题无关，但如果不希望 pendingWorkItems 为零或负数，您应该使用 > 比较运算符。它可以防止 pendingWorkItems 变为负数的错误。 - Some programmer dude

@JoachimPileborg 那个变量是无符号的，因为所使用的 API 要求如此。 - Ghita

@Ghita InterlockedDecrement 接受指向 LONG 的指针，实际上是一个有符号值。 - dgnuff

Answer 1

您可以检查InterlockedDecrement()的结果，如果结果为负数（或者小于等于0更合适），则通过调用InterlockedIncrement()来撤消减量。在其他情况下，正确的代码应该没问题。

- Alexey Frunze

Answer 2

最简单的解决方案就是在整个部分（以及对this->pendingWorkItems的所有其他访问）周围使用互斥锁。如果由于某种原因这不可接受，那么您可能需要使用比较和交换：

void decrementPendingWorkItems()
{
    int count = std::atomic_load( &pendingWorkItems );
    while ( count != 0
            && ! std::atomic_compare_exchange_weak( 
                    &pendingWorkItems, &count, count - 1 ) ) {
    }
}

（这假设pendingWorkItems的类型为std::atomic_int。）

- James Kanze

Answer 3

有一种叫做“自旋锁”的东西。这是一种非常轻量级的同步方式。

具体思路如下：

//
//    This lock should be used only when operation with protected resource
//  is very short like several comparisons or assignments.
//
class SpinLock
{
 public:

      __forceinline SpinLock() { body = 0; }
      __forceinline void Lock()
           {
             int spin = 15;
             for(;;) {
               if(!InterlockedExchange(&body, 1)) break;
               if(--spin == 0) { Sleep(10); spin = 29; }
             }
           }

      __forceinline void Unlock() { InterlockedExchange(&body, 0); }

 protected:

    long   body;

};

样本中的实际数字并不重要，这个锁非常高效。

- Kirill Kobelev

---

主体应该是“易失性”的，不是吗？ - smerlin

不，其实不是。这个在生产环境中运行了好几年。关于volatile的讨论可能会在这里进行，但不幸的是它超出了评论的格式。 - Kirill Kobelev

在进行易失性读取时，您应该进行自旋，并且只有在读取指示需要尝试获取锁时才使用CAS。此外，您应始终使用_mm_pause来发出自旋循环处理器提示，否则会不必要地消耗CPU资源（并使用非线性退避）。 - Necrolis

请注意，我没有直接读取/写入数据字段。这一切都隐藏在InterlockedExchange中。@Necrolis，您的评论适用于那些决定编写自己的InterlockedExchange的人。 - Kirill Kobelev

Answer 4

您可以在循环中使用InterlockedCompareExchange：

    inline void DecrementPendingWorkItems() {
        LONG old_items = this->pendingWorkingItems;
        LONG items;
        while ((items = old_items) > 0) {
            old_items = ::InterlockedCompareExchange(&this->pendingWorkItems,
                                                     items-1, items);
            if (old_items == items) break;
        }
    }

InterlockedCompareExchange 函数的作用是：

  if pendingWorkItems matches items, then
    set the value to items-1 and return items
  else return pendingWorkItems

这是原子性完成的，也被称为比较并交换。

- jxh

Answer 5

使用原子CAS。

http://msdn.microsoft.com/en-us/library/windows/desktop/ms683560(v=vs.85).aspx

你可以使它无锁，但不能等待自由。

正如Kirill所建议的，在你的情况下这类似于自旋锁。

我认为这可以满足你的需求，但在使用之前我建议仔细考虑所有可能性，因为我根本没有测试过它：

inline bool
InterlockedSetIfEqual(volatile LONG* dest, LONG exchange, LONG comperand)
{
    return comperand == ::InterlockedCompareExchange(dest, exchange, comperand);
}

inline bool InterlockedDecrementNotZero(volatile LONG* ptr)
{
    LONG comperand;
    LONG exchange;
    do {
        comperand = *ptr;
        exchange = comperand-1;
        if (comperand <= 0) {
            return false;
        }
    } while (!InterlockedSetIfEqual(ptr,exchange,comperand));
    return true;
}

现在还有一个问题，即为什么您的待处理工作项会变成负数。您应该确保增量的数量与减量的数量相匹配，这样一切都会很好。如果违反了这个约束条件，我建议添加一个assert或异常。

- Pete

---

小细节提示：在处理潜在竞态条件时使用标准的C++ assert是无意义的，因为许多竞态条件仅会在发布构建中出现，而assert在发布构建中将不可用。在这种情况下，应编写自己的assert版本，该版本在所有配置中均可用。 - smerlin

同意。使用异常会更好。 - Pete

如果发生下溢的原因（可能是一个 bug），我不希望在这种情况下导致另一个更危险的 bug。这只是一种安全措施。 - Ghita