在C#中拆分数组的最快(便携)方式

Question

我正在编写一个完全托管的Mercurial库(将用于完全托管的Mercurial Server for Windows,即将推出),而我遇到的最严重的性能问题之一就是奇怪地将数组分成几部分.

这个想法如下:有一个字节数组,大小范围从几百字节到一兆字节,我需要做的就是将它拆分成由我的特定情况下的\n字符分隔的部分.

现在什么dotTrace显示我的是,我的"优化"版本的Split(代码是正确的,这里的天真的版本,我开始用)占用11秒2300个电话(有由dotTrace本身引入了一个明显的性能损失,但一切都达规模).

这是数字:

• unsafe版本:调用11 297ms2 312
• 托管("天真")版本:呼叫的20 001ms2 312

所以这里是:最快的(最好是可移植的,意味着支持x86和x64)在C#中拆分数组的方法.

Answer 1

对于Split，在32位机器上处理ulong确实很慢，所以一定要减少到uint。如果您确实想要 ulong，请实现两个版本，一种用于 32 位，一种用于 64 位。

您还应该测量一次处理字节是否更快。

需要分析内存分配的成本。如果它足够大，请尝试在多个调用中重用内存。

其他：

ToString：使用 "(" + Offset.ToString() + ", " + Length.ToString() + ")" 更快；

GetHashCode：尝试固定(byte * b = & buffer[offset])

如果多次使用，这个版本应该非常快。关键点：内部数组扩展至正确大小后无需新的内存分配，最少的数据复制。

class ArraySplitter
{
    private byte[] m_data;
    private int    m_count;
    private int[]  m_stops;

    private void AddRange(int start, int stop)
    {
        // Skip empty range
        if (start > stop)
        {
            return;
        }

        // Grow array if needed
        if ((m_stops == null) || (m_stops.Length < (m_count + 2)))
        {
            int[] old = m_stops;

            m_stops = new int[m_count * 3 / 2 + 4];

            if (old != null)
            {
                old.CopyTo(m_stops, 0);
            }
        }

        m_stops[m_count++] = start;
        m_stops[m_count++] = stop;
    }

    public int Split(byte[] data, byte sep)
    {
        m_data  = data;
        m_count = 0;      // reuse m_stops

        int last = 0;

        for (int i = 0; i < data.Length; i ++)
        {
            if (data[i] == sep)
            {
                AddRange(last, i - 1);
                last = i + 1;
            }
        }

        AddRange(last, data.Length - 1);

        return m_count / 2;
    }

    public ArraySegment<byte> this[int index]
    {
        get
        {
            index *= 2;
            int start = m_stops[index];

            return new ArraySegment<byte>(m_data, start, m_stops[index + 1] - start + 1);
        }
    }
}

测试程序：

    static void Main(string[] args)
    {
        int count = 1000 * 1000;

        byte[] data = new byte[count];

        for (int i = 0; i < count; i++)
        {
            data[i] = (byte) i;
        }

        Stopwatch watch = new Stopwatch();

        for (int r = 0; r < 10; r++)
        {
            watch.Reset();
            watch.Start();

            int len = 0;

            foreach (var seg in data.MySplit(13))
            {
                len += seg.Count;
            }

            watch.Stop();

            Console.WriteLine("MySplit      : {0} {1,8:N3} ms", len, watch.Elapsed.TotalMilliseconds);

            watch.Reset();
            watch.Start();

            ArraySplitter splitter = new ArraySplitter();

            int parts = splitter.Split(data, 13);

            len = 0;

            for (int i = 0; i < parts; i++)
            {
                len += splitter[i].Count;
            }

            watch.Stop();
            Console.WriteLine("ArraySplitter: {0} {1,8:N3} ms", len, watch.Elapsed.TotalMilliseconds);
        }
    }

结果：

MySplit      : 996093    9.514 ms
ArraySplitter: 996093    4.754 ms
MySplit      : 996093    7.760 ms
ArraySplitter: 996093    2.710 ms
MySplit      : 996093    8.391 ms
ArraySplitter: 996093    3.510 ms
MySplit      : 996093    9.677 ms
ArraySplitter: 996093    3.468 ms
MySplit      : 996093    9.685 ms
ArraySplitter: 996093    3.370 ms
MySplit      : 996093    9.700 ms
ArraySplitter: 996093    3.425 ms
MySplit      : 996093    9.669 ms
ArraySplitter: 996093    3.519 ms
MySplit      : 996093    9.844 ms
ArraySplitter: 996093    3.416 ms
MySplit      : 996093    9.721 ms
ArraySplitter: 996093    3.685 ms
MySplit      : 996093    9.703 ms
ArraySplitter: 996093    3.470 ms