无锁引用计数

Question

我正在开发一个需要大量使用C API的系统。其中一部分是在任何操作之前和之后对该系统进行初始化和关闭。如果无法完成这两个步骤，将导致系统不稳定。我通过在核心可释放环境类中实现引用计数来实现此操作，如下所示：

public FooEnvironment()
{
  lock(EnvironmentLock)
  {
    if(_initCount == 0)
    {
      Init();  // global startup
    }
    _initCount++;
  }
}

private void Dispose(bool disposing)
{
  if(_disposed)
    return;

  if(disposing)
  {
    lock(EnvironmentLock)
    {
      _initCount--;
      if(_initCount == 0)
      {
        Term(); // global termination
      }
    }
  }
}

这样做很好，并且完成了目标。 但是，由于任何interop操作都必须嵌套在FooEnvironment using块中，我们一直处于锁定状态，并且分析表明，此锁定占据运行时工作量的近50％。 看起来这是一个足够基本的概念，.NET或CLR中必须解决它。 有更好的方法来进行引用计数吗？

- Jeff

---

Interlocked.Increment（以及相关方法）是您正在寻找的吗？ - harold

你正在使用哪个版本的 .Net 框架？ - pstrjds

你的意思是初始化和关闭必须在每个操作之间执行，还是只需在第一次操作之前和最后一次操作之后执行？ - jalf

Interlocked.Increment 可能有效，但我不确定具体如何实现。我还需要能够测试该值并在其为零时采取行动。你能提供一种方法吗？谢谢！ - Jeff

这是4.0版本，初始化和终止只需要在第一次和最后一次操作之前发生（而不是在每次操作之间）。 - Jeff

顺便说一下，interlocked 的成员也不是特别快。 - harold

Answer 1

这比你一开始想象的要棘手得多。我不相信Interlocked.Increment足以胜任你的任务。相反，我希望你需要用CAS（比较并交换）执行一些巫术。

还要注意，当你的程序因为Heisenbugs而崩溃时，很容易做到大部分正确，但大部分正确仍然是完全错误的。

我强烈建议在走这条路之前进行一些真正的研究。如果你搜索“无锁引用计数”，会有一些很好的起点。这篇Dr. Dobbs文章很有用，这个SO问题可能相关。

首先要记住的是，无锁编程非常困难。如果这不是你的专长，请考虑退后一步，并调整你对引用计数粒度的期望。重新思考你的基本引用计数策略可能会比创建可靠的无锁机制更加经济实惠，尤其是当你还不知道无锁技术是否真的会更快时。

- Greg D

---

+1 这些链接很棒，尤其是那个非常出色的评论：“还要注意一点，即使你的程序在大多数情况下都能运行得很好，但当它因为 Heisenbugs 而崩溃时，这种情况仍然是完全错误的。” 当你的客户打电话来说“有时候我启动程序后几秒钟就会崩溃，而其他时候它可以连续运行几天而没有问题”，而你在调试时发现一个放错位置的锁语句时，这种情况永远不是好事。 - pstrjds

@Jeff，你尝试过用SpinLock替换你的锁吗？如果大部分锁定操作都很快，那么这是一种更便宜的锁定机制。 - pstrjds

@Jeff：所有无锁编程都使用处理器特定的指令。 :) .Net为它们提供了一个好听的名字（CompareExchange而不是x86 cmpxchg），但任何无锁操作都必须非常接近底层。如果那篇文章让你感到困惑，我强烈建议不要尝试创建一个无锁引用计数机制。尝试找出是否有一种方法可以比你现在做的更少地进行锁定和引用计数。 - Greg D

@Jeff：锁热了吗（即，是否存在很多争用）？ - Greg D

我不能在没有足够的上下文信息的情况下为你作出回答。我只能告诉你尽量少竞争。例如，当锁定根时是否锁定整个对象图？是否可以要求客户端为性能原因正确处理初始化/反初始化？或者让性能敏感的客户端选择退出自动环境管理是否可行？这并不是一个独家问题集，我可以提供答案，这些都是您必须考虑的事情（还有很多其他问题）来解决您的方案。 - Greg D

Answer 2

正如哈罗德的评论所指出的那样，答案是Interlocked：

public FooEnvironment() {
  if (Interlocked.Increment(ref _initCount) == 1) {
    Init();  // global startup
  }
}

private void Dispose(bool disposing) {
  if(_disposed)
    return;

  if (disposing) {
    if (0 == Interlocked.Decrement(ref _initCount)) {
      Term(); // global termination
    }
  }
}

无论是增量还是减量，都会返回新的计数（仅适用于此类用法），因此需要进行不同的检查。

但请注意：如果需要并发保护，则这将无法正常工作。Interlocked操作本身是安全的，但其他任何操作都不是（包括不同线程相对顺序的Interlocked调用）。在上面的代码中，Init()仍然可以在另一个线程完成构造函数后运行。

- Richard

---

在上面的代码中，即使另一个线程已经完成构造函数，Init()仍然可以继续运行。这就是为什么无锁代码应该留给专家的原因。 :) - Greg D

实际上，我在Init和Term内部加锁也没问题。这解决了每次都需要加锁的大问题。谢谢！ - Jeff

@Jeff：那并不能解决问题。假设你在init中加锁，第二个线程永远不会看到那个锁，因为它从未运行过init。它仍然会在初始化完成之前进入组件。 - Greg D

我明白了，很好的观点。所以这并没有完全解决问题。 - Jeff

Answer 3

我相信这将为您提供一种安全的方式，使用Interlocked.Increment/Decrement。

注意：这是过于简化了，如果Init()抛出异常，下面的代码可能会导致死锁。当计数器归零时，在Dispose中存在竞争条件，初始化被重置并且构造函数再次被调用。我不知道您的程序流程，所以如果您有可能在几个dispose调用之后再次进行初始化，那么您最好使用一个更便宜的锁，比如SpinLock，而不是InterlockedIncrement。

static ManualResetEvent _inited = new ManualResetEvent(false);
public FooEnvironment()
{
    if(Interlocked.Increment(ref _initCount) == 1)
    {
        Init();  // global startup
        _inited.Set();
    }

    _inited.WaitOne();
}

private void Dispose(bool disposing)
{
    if(_disposed)
        return;

    if(disposing)
    {
        if(Interlocked.Decrement(ref _initCount) == 0)
        {
            _inited.Reset();
            Term(); // global termination
        }
    }
}

编辑：
在进一步思考后，您可能需要考虑一些应用程序重新设计。 相反地，只需在应用程序启动时进行单个调用以管理Init和Term，并在应用程序关闭时进行对Term的调用，然后您就可以完全删除锁定的需要了。 如果锁定显示为您执行时间的50％，那么似乎您总是想要调用Init，所以请直接调用它并继续进行。

- pstrjds

---

请注意，Init() 不能保证在第二个线程进入、检查引用计数并开始运行（尚未初始化环境）之前完成。 - Greg D

@Greg D - 很好的观点。我知道感觉好像我漏掉了什么。再次证明多线程编程很难。 - pstrjds

@Greg D - 那我可以理解你不会陷入试图找出解决方案的陷阱了吗？ :) - Jeff

@Greg D - 嗯，我确实提出了一个仍然有缺陷的替代方案，尽管根据程序流程，我的替代方案可能是安全的（加上一些异常处理），但从外观来看，最好的选择是更便宜的锁。 - pstrjds

@Jeff: 我不够聪明来解决这个问题。我所做的任何无锁编程都仅限于“娱乐目的”。 :) 我建议的解决方案是重新思考您的引用计数策略周围的粒度。 - Greg D

Answer 4

您可以使用以下代码使其几乎无锁。这肯定会降低争用，如果这是您的主要问题，那么这将是您需要的解决方案。

此外，我建议从析构函数/终结器中调用Dispose（以防万一）。我已更改了您的Dispose方法 - 无论disposing参数如何，都应该释放非托管资源。请参阅this以了解有关正确处理对象的详细信息。

希望这能帮到您。

public class FooEnvironment
{
    private static int _initCount;
    private static bool _initialized;
    private static object _environmentLock = new object();

    private bool _disposed;

    public FooEnvironment()
    {
        Interlocked.Increment(ref _initCount);

        if (_initCount > 0 && !_initialized)
        {
            lock (_environmentLock)
            {
                if (_initCount > 0 && !_initialized)
                {
                    Init(); // global startup
                    _initialized = true;
                }
            }
        }
    }

    private void Dispose(bool disposing)
    {
        if (_disposed)
            return;

        if (disposing)
        {
            // Dispose managed resources here
        }

        Interlocked.Decrement(ref _initCount);

        if (_initCount <= 0 && _initialized)
        {
            lock (_environmentLock)
            {
                if (_initCount <= 0 && _initialized)
                {
                    Term(); // global termination
                    _initialized = false;
                }
            }
        }

        _disposed = true;
    }

    ~FooEnvironment()
    {
        Dispose(false);
    }
}

- Maciej

---

谢谢，不幸的是我认为在一个线程的构造函数中检查_initCount == 0和在另一个线程中进行dispose之间存在小的竞争条件（因此我们认为环境正在运行，而实际上并非如此）。 - Jeff

Answer 5

可能在类中使用一个通用的静态变量。静态变量只是一件事，不特定于任何对象。

- iefpw

---

EnvironmentLock和_initCount都是静态的。 - Jeff

Answer 6

Interlocked类的方法比锁定语句稍微快一些，但在多核机器上，速度优势可能不是很大，因为Interlocked指令必须绕过内存缓存层。

当代码未使用和/或程序退出时，调用Term()函数有多重要？

通常情况下，您可以只在包装其他API的类的静态构造函数中调用一次Init()，而不必担心调用Term()。例如：

static FooEnvironment() { 
    Init();  // global startup 
}

CLR 会确保在封闭类的任何其他成员函数之前，静态构造函数将被调用一次。

此外，还可以挂钩某些（但不是全部）应用程序关闭场景的通知，从而使在干净的关闭时调用 Term() 成为可能。请参阅本文。http://www.codeproject.com/Articles/16164/Managed-Application-Shutdown

- Burt Harris

Answer 7

使用Threading.Interlocked.Increment会比获取锁、执行增量操作和释放锁稍微快一点，但差别不是非常大。在多核系统上，任何一种操作的昂贵部分都是强制内存缓存在核之间同步。 Interlocked.Increment的主要优势不是速度，而是它将在有限的时间内完成。相比之下，如果想要获取锁、执行增量操作并释放锁，即使锁仅用于保护计数器，如果其他线程获取锁然后被拦截，就有可能永远等待。

你没有提到你使用的 .net 版本，但是有一些 Concurrent 类可能会有用。根据你分配和释放资源的模式，一个看起来有点棘手但可能很好用的类是 ConcurrentBag 类。它有点像队列或堆栈，但没有保证事物会按任何特定顺序出现。在你的资源包装器中包含一个指示它是否仍然有效的标志，并在资源本身中包含对包装器的引用。当创建资源用户时，将一个包装器对象扔进袋子里。当不再需要资源用户时，设置“无效”标志。只要袋子里有至少一个“有效”标志被设置的包装器对象，或者资源本身持有对有效包装器的引用，资源就应该保持活动状态。如果在删除项目时，资源似乎没有持有有效的包装器，请获取锁定，如果资源仍然没有持有有效的包装器，则从袋子里取出包装器，直到找到一个有效的包装器为止，然后将其与资源存储在一起（如果没有找到，则销毁资源）。如果在删除项目时，资源持有有效的包装器，但袋子似乎可能持有过多的无效项，请获取锁定，将袋子的内容复制到数组中，并将有效项扔回袋子里。保持计数，以便可以判断何时进行下一次清理。

这种方法可能比使用锁或Threading.Interlocked.Increment更复杂，需要考虑很多边角情况，但它可能提供更好的性能，因为ConcurrentBag旨在减少资源争用。如果处理器1对某个位置执行Interlocked.Increment，然后处理器2也这样做，处理器2将不得不指示处理器1从其缓存中刷新该位置，等待直到处理器1完成此操作，通知所有其他处理器它需要控制该位置，将该位置加载到其缓存中，最后开始递增。在所有这些操作发生之后，如果处理器1需要再次递增该位置，则需要执行相同的一般步骤序列。所有这些都非常慢。相比之下，ConcurrentBag类被设计为多个处理器可以将东西添加到列表中而不会发生缓存冲突。在添加和删除之间的某个时刻，它们将被复制到一个一致的数据结构中，但是这样的操作可以批量执行，以便产生良好的缓存性能。

我从未尝试过使用ConcurrentBag的上述方法，因此我不知道它实际上会产生什么样的性能，但根据使用模式，可能可以提供比通过引用计数获得的更好的性能。

- supercat