在托管环境中划分内存

Question

我正在用C#进行流体模拟.每个循环我需要计算空间中离散点处流体的速度.作为计算的一部分,我需要几十千字节的空间来容纳一些double []数组(数组的确切大小取决于一些输入数据).数组仅在使用它们的方法的持续时间内需要,并且有一些不同的方法需要这样的临时空间.

在我看来,有一些不同的解决方案来构建临时数组:

1. 每次调用方法时,使用'new'从堆中获取内存.这就是我最初做的事情,但是它给垃圾收集器带来了很大的压力,而每秒一到两次的几毫秒尖峰真的很烦人.

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

3. 在不安全的上下文中使用stackalloc从程序堆栈中分配临时内存.这可以正常工作,除了我需要使用/ unsafe进行编译并在我的代码中不断地散布不安全的块,我想避免.

4. 程序启动时预先分配私有数组.这很好,除非我不一定知道我需要的数组的大小,直到我可以查看一些输入数据.它变得非常混乱,因为你不能将这些私有变量的范围限制为单一方法,因此它们不断地污染命名空间.并且随着需要暂存内存的方法数量的增加,它的扩展性很差,因为我分配的内存很多,只占用了一小部分时间.

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

...

有没有人有类似问题的经验？从经验中提供的任何建议/课程？

Answer 1

我同情你的情况; 当我在Roslyn工作时,我们非常仔细地考虑了分配临时工作阵列时收集压力带来的潜在性能问题.我们解决的解决方案是汇集策略.

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

编写线程安全池并不太难,或者您可以使池线程静态并且每个线程有一个池.

棘手的一点是在池中恢复内存.有几种方法可以解决这个问题.首先,如果他们不想承担收集压力,你可以简单地要求池化内存的用户在完成数组时调用"回到池中"方法.

另一种方法是在数组周围编写一个外观包装器,使它实现IDisposable,这样你就可以使用"using"(*),并在上面创建一个终结器,将对象放回池中,然后重新启动它.(确保让终结者回到"我需要最终确定"位.)复活的终结者让我感到紧张; 我个人更喜欢前一种方法,这就是我们在罗斯林所做的.

(*)是的,这违反了"使用"应表明非托管资源正在返回操作系统的原则.基本上我们通过自己的管理将托管内存视为非托管资源,所以它并没有那么糟糕.

Answer 2

您可以将使用这些临时数组的代码包装在 using 语句中，如下所示：

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

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

不幸的是，上述情况似乎并不正确！相反，您可以尝试使用辅助函数来返回适当大小的数组（大于该大小的最接近的 2 的幂），如果它不存在，则创建它。这样，您就只有对数数量的数组。如果您希望它是线程安全的，那么您将需要遇到更多麻烦。

它可能看起来像这样：（使用算法中的 pow2roundup 来查找大于或等于给定值的最小二的幂）

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];
}