托管环境中的临时内存

Question

我正在c#中进行流体模拟。每个循环我需要计算流体在空间中离散点的速度。作为该计算的一部分，我需要几十kb的临时空间来保存一些double[]数组(数组的确切大小取决于某些输入数据)。数组只需要在使用它们的方法的持续时间内使用，并且有一些不同的方法需要像这样的刮擦空间。

在我看来，有几种不同的方法可以用来构造临时数组:

1. 每次调用该方法时，使用'new'从堆中获取内存。这就是我一开始所做的，但是它给垃圾收集器带来了很大的压力，并且每秒一次或两次几毫秒的峰值真的很烦人。

2. 在调用方法时将scratch数组作为参数传递。问题是，这迫使用户管理它们，包括适当地调整它们的大小，这是一个巨大的痛苦。它使得使用或多或少的刮擦存储器变得困难，因为它改变了API。

3. 在不安全的上下文中使用stackalloc从程序堆栈中分配临时内存。这将工作得很好，除了我需要使用/unsafe编译并不断地在我的代码中散布不安全块，这是我想避免的。

4. 程序启动时预分配私有数组一次。这很好，除了我不一定知道我需要的数组的大小，直到我可以看到一些输入数据。由于不能将这些私有变量的作用域限制为单个方法，所以它们会不断污染命名空间，因此会变得非常混乱。而且随着需要临时内存的方法数量的增加，它的可伸缩性也很差，因为我分配了大量只使用一小部分时间的内存。

5. 创建某种中央池，并从池中分配临时内存数组。这样做的主要问题是，我没有看到从中央池分配动态大小的数组的简单方法。我可以使用起始偏移量和长度，并让所有的临时内存基本上共享一个大数组，但我有很多现有的代码假设双[]。我必须小心地使这样的池线程安全。

…

有人遇到过类似的问题吗?从这次经历中有什么建议/教训吗?

Answer 1

我同情你的处境;当我在开发Roslyn时，我们非常仔细地考虑了分配临时工作数组所带来的收集压力所导致的潜在性能问题。我们决定采用的解决方案是池化策略。

在编译器中，数组的大小往往很小，因此经常重复。在您的情况下，如果您有大型数组，那么我要做的是遵循Tom的建议:简化管理问题并浪费一些空间。当向池请求大小为x的数组时，请将x四舍五入到最接近的2的幂并分配该大小的数组，或者从池中取出一个数组。调用者得到的数组有点太大了，但是它们可以被写入来处理这个问题。在池中搜索大小合适的数组应该不会太难。或者您可以维护一堆池，一个池用于大小为1024的数组，一个用于大小为2048的数组，等等。

编写一个线程安全池并不太难，或者您可以将池设置为线程静态，每个线程一个池。

棘手的部分是将内存恢复到池中。有几种方法可以解决这个问题。首先，如果用户不想承担收集压力的代价，可以简单地要求池内存的用户在使用完数组后调用"返回池"方法。

另一种方法是在数组周围编写一个外观包装器，使其实现IDisposable，以便您可以使用"using"(*)，并在其上创建一个终结器，将对象放回池中，使其复活。(确保将定稿器调回到"我需要定稿"的位置。)复活的结局让我紧张;我个人更喜欢前一种方法，这就是我们在Roslyn所做的。

---

(*)是的，这违反了"using"应该表明非托管资源正在返回给操作系统的原则。从本质上讲，我们通过自己的管理将托管内存视为非托管资源，所以这并不是那么糟糕。

Answer 2

您可以将使用这些scratch数组的代码包装在using语句中，像这样:

using(double[] scratchArray = new double[buffer])
{
    // Code here...
}

这将通过在using语句的末尾调用解构函数来显式地释放内存。

不幸的是，上面似乎不是真的!取而代之的是，您可以尝试使用一个辅助函数来返回适当大小的数组(最接近于比大小大2的幂)，如果它不存在，则创建它。这样，你就只有对数个数的数组了。如果你想让它是线程安全的，你就需要多花点时间。

它可能看起来像这样:(使用算法中的pow2roundup来查找大于或等于给定值的2的最小幂)

private static Dictionary<int,double[]> scratchArrays = new Dictionary<int,double[]>();
/// Round up to next higher power of 2 (return x if it's already a power of 2).
public static int Pow2RoundUp (int x)
{
    if (x < 0)
        return 0;
    --x;
    x |= x >> 1;
    x |= x >> 2;
    x |= x >> 4;
    x |= x >> 8;
    x |= x >> 16;
    return x+1;
}
private static double[] GetScratchArray(int size)
{
    int pow2 = Pow2RoundUp(size);
    if (!scratchArrays.ContainsKey(pow2))
    {
        scratchArrays.Add(pow2, new double[pow2]);
    }
    return scratchArrays[pow2];
}

编辑:线程安全版本:这仍然会有被垃圾收集的东西，但它将是特定于线程的，并且应该减少开销。

[ThreadStatic]
private static Dictionary<int,double[]> _scratchArrays;
private static Dictionary<int,double[]> scratchArrays
{
    get
    {
        if (_scratchArrays == null)
        {
            _scratchArrays = new Dictionary<int,double[]>();
        }
        return _scratchArrays;
    }
}
/// Round up to next higher power of 2 (return x if it's already a power of 2).
public static int Pow2RoundUp (int x)
{
    if (x < 0)
        return 0;
    --x;
    x |= x >> 1;
    x |= x >> 2;
    x |= x >> 4;
    x |= x >> 8;
    x |= x >> 16;
    return x+1;
}
private static double[] GetScratchArray(int size)
{
    int pow2 = Pow2RoundUp(size);
    if (!scratchArrays.ContainsKey(pow2))
    {
        scratchArrays.Add(pow2, new double[pow2]);
    }
    return scratchArrays[pow2];
}