c# - 如何消除此循环矢量化的数组边界检查？

Question

我的任务是从二进制文字 0x0 上的数据库表中拆分多个运行的 varbinary(8000) 列。但是，这可能会改变，所以我想保留这个变量。我想使用 SQLCLR 作为流式表值函数快速执行此操作。我知道我的每个字符串总是至少有几千个字节。

编辑:我更新了我的算法。为了避免内循环展开的麻烦。但是要说服 CLR 对寄存器分配做出正确的选择是极其困难的。如果有一种简单的方法可以让 CLR 相信 j 和 i 真的是同一个东西，那就太棒了。但相反，它确实做了一些愚蠢的事情。优化第一个路径循环会很好。但是您不能在循环中使用 goto。

我决定采用 C 函数 memchr 的 64 位实现。基本上不是一次扫描一个字节并进行比较，我使用一些位旋转一次扫描 8 个字节。作为引用，Array.IndexOf<Byte>为一个答案做类似 4 字节扫描的事情，我只想继续这样做。 有几点需要注意:

1. 内存压力是 SQLCLR 函数中一个非常现实的问题。 String.Split已经出来了，因为它预先分配了很多我真的很想避免的内存。它也适用于 UCS-2 字符串，这需要我将我的 ascii 字符串转换为 unicode 字符串，因此在返回时将我的数据视为 lob 数据类型。 (SqlChars/SqlString 在变成lob类型前只能返回4000字节)。

2. 我想直播。避免String.Split的另一个原因它一次完成所有工作，造成很大的内存压力。在包含大量定界符的代码上，纯 T-SQL 方法将开始击败它。

3. 我想保持它“安全”。所以都管理好了。安检好像有很大的罚款。

Buffer.BlockCopy真的很快，而且似乎最好先支付一次费用，而不是不断支付 BitConverter 的费用。这也比将我的输入转换为字符串并保留该引用更便宜。

代码非常快，但似乎我要为很多绑定(bind)检查付费，在初始循环和关键部分找到匹配项时。由于代码中有很多定界符，我倾向于输给一个更简单的 C# 枚举器，它只进行字节比较。

这是我的代码，

class SplitBytesEnumeratorA : IEnumerator
{
    // Fields
    private readonly byte[] _bytes;
    private readonly ulong[] _longs;
    private readonly ulong _comparer;
    private readonly Record _record = new Record();
    private int _start;
    private readonly int _length;

    // Methods
    internal SplitBytesEnumeratorA(byte[] bytes, byte delimiter)
    {
        this._bytes = bytes;
        this._length = bytes.Length;
        // we do this so that we can avoid a spillover scan near the end.
        // in unsafe implementation this would be dangerous as we potentially
        // will be reading more bytes than we should.

        this._longs = new ulong[(_length + 7) / 8];
        Buffer.BlockCopy(bytes, 0, _longs, 0, _length);
        var c = (((ulong)delimiter << 8) + (ulong)delimiter);
        c = (c << 16) + c;
        // comparer is now 8 copies of the original delimiter.
        c |= (c << 32);
        this._comparer = c;
    }

    public bool MoveNext()
    {
        if (this._start >= this._length) return false;
        int i = this._start;
        var longs = this._longs;
        var comparer = this._comparer;
        var record = this._record;
        record.id++;
        // handle the case where start is not divisible by eight.
        for (; (i & 7) != 0; i++)
        {
            if (i == _length || _bytes[i] == (comparer & 0xFF))
            {
                record.item = new byte[(i - _start)];
                Buffer.BlockCopy(_bytes, _start, record.item, 0, i - _start);
                _start = i + 1;
                return true;
            }
        }

        // main loop. We crawl the array 8 bytes at a time.

        for (int j=i/8; j < longs.Length; j++)
        {
            ulong t1 = longs[j];
            unchecked
            {
                t1 ^= comparer;
                ulong t2 = (t1 - 0x0101010101010101) & ~t1;
                if ((t2 & 0x8080808080808080) != 0)
                {
                    i =j*8;
                    // make every case 3 comparison instead of n. Potentially better. 
                    // This is an unrolled binary search.
                    if ((t2 & 0x80808080) == 0)
                    {
                        i += 4;
                        t2 >>= 32;
                    }

                    if ((t2 & 0x8080) == 0)
                    {
                        i += 2;
                        t2 >>= 16;
                    }

                    if ((t2 & 0x80) == 0)
                {
                i++;
                }
                record.item = new byte[(i - _start)];
                // improve cache locality by not switching collections.
                Buffer.BlockCopy(longs, _start, record.item, 0, i - _start);                _start = i + 1;
                return true;
            }
        }
        // no matches found increment by 8
    }
    // no matches left. Let's return the remaining buffer.
    record.item = new byte[(_length - _start)];
    Buffer.BlockCopy(longs, _start, record.item, 0, (_length - _start));
    _start = _bytes.Length;
    return true;
    }

    void IEnumerator.Reset()
    {
        throw new NotImplementedException();
    }

    public object Current
    {
        get
        {
            return this._record;
        }
    }
}

// We use a class to avoid boxing .
class Record
{
    internal int id;
    internal byte[] item;
}

Answer 1

跳出框框思考，您是否考虑过将字符串转换为 XML 并使用 XQuery 进行拆分？

例如，您可以传入分隔符和(空气代码):

DECLARE @xml as xml
DECLARE @str as varchar(max)
SET @str = (SELECT CAST(t.YourBinaryColumn AS varchar(max) FROM [tableName] t) 
SET @xml = cast(('<X>'+replace(@str,@delimiter,'</X><X>')+'</X>') as xml)

这会将二进制文件转换为字符串并将分隔符替换为 XML 标记。然后:

SELECT N.value('.', 'varchar(10)') as value FROM @xml.nodes('X') as T(N)

将获取单个“元素”，即每个分隔符出现之间的数据。

也许这个想法可以按原样使用，或者作为您可以在此基础上构建的催化剂。