.net - 线程安全无锁互字节数组队列

标签 .net multithreading thread-safety queue lock-free

应该传输一个字节流,并且有一个生产者线程和一个消费者线程。 大多数时候,生产者的速度高于消费者,并且我需要足够的缓冲数据来保证应用程序的 QoS。 我读到了我的问题,并且有诸如共享缓冲区、PipeStream .NET 类之类的解决方案... 此类将在服务器上实例化多次,因此我需要优化的解决方案。 使用 ByteArray 队列是个好主意吗?

如果是,我将使用优化算法来猜测队列大小和每个 ByteArray 容量,理论上它适合我的情况。

如果没有,最好的方法是什么?

请告诉我是否有 C# 或 VB 中的 ByteArray 队列的良好无锁线程安全实现。

提前致谢

最佳答案

如果您不是逐字节地生成和消耗字节,而是按 block 工作,您可能会获得更多的加速。在这种情况下,代码的“无锁性”可能根本不重要——事实上,传统的锁定解决方案可能更可取。我会尝试演示一下。

C# 中给出了无锁、单个 生产者、单个 消费者、有界 队列。 ( list A)
没有深奥的互锁操作,甚至没有显式的内存屏障。可以这么说,乍一看它是尽可能快且无锁的。不是吗?
现在让我们将其与锁定解决方案进行比较 that Marc Gravell has given, here .

我们将使用双 CPU 机器,核心之间没有共享 L3 缓存。 我们预计最多 2 倍加速。 2 倍加速确实意味着无锁解决方案在理论范围内表现理想。
为了为无锁代码创造一个理想的环境,我们甚至会使用here中的实用程序类来设置生产者和消费者线程的CPU亲和性。 .
测试的结果代码如( list B)所示。

它正在生产大约。一个线程上 10MBytes,而另一线程上则消耗它。
队列大小固定为 32KBytes。如果已满,生产者就会等待。
在我的机器上典型的测试运行如下所示:

LockFreeByteQueue: 799ms
ByteQueue: 1843ms

无锁队列速度更快。哇,速度快了 2 倍多!这是值得夸耀的事情。 :)
让我们看看发生了什么。 Marc 的锁定队列就是这样做的。它锁住了。它对每个字节都执行此操作。

我们真的需要锁定每个字节并逐字节推送数据吗?它肯定会以 block 的形式到达网络(例如一些大约 1k 的数据包)。即使它确实是从内部源逐字节到达的,生产者也可以轻松地将其打包成漂亮的 block 。
让我们这样做 - 不要逐字节地生成和消费,而是分块工作并向微基准添加另外两个测试( list C,只需将其插入基准主体中)。
现在典型的运行如下所示:

LockFreePageQueue: 33ms
PageQueue: 25ms

现在,它们实际上都比原始无锁代码快 20 倍 - Marc's solution现在,添加分块的无锁代码实际上更快!
我们没有采用会导致 2 倍加速的无锁结构,而是尝试了另一种解决方案,它可以很好地使用锁定并导致 20 倍(!)加速。
许多问题的关键并不在于避免锁定,而在于避免共享和最小化锁定。在上述情况下,我们可以在字节复制期间避免共享。
我们可以在大部分时间处理私有(private)结构,然后将单个指针排入队列,从而将共享空间和时间缩小到将单个指针插入队列的单次插入。

list A,一个无锁、单个生产者、单个消费者队列:

public class BoundedSingleProducerSingleConsumerQueue<T>
{
    T[] queue;
    volatile int tail;
    volatile int head;

    public BoundedSingleProducerSingleConsumerQueue(int capacity)
    {
        queue = new T[capacity + 1];
        tail = head = 0;
    }

    public bool TryEnqueue(T item)
    {
        int newtail = (tail + 1) % queue.Length;
        if (newtail == head) return false;
        queue[tail] = item;
        tail = newtail;
        return true;
    }

    public bool TryDequeue(out T item)
    {
        item = default(T);
        if (head == tail) return false;
        item = queue[head];
        queue[head] = default(T);
        head = (head + 1) % queue.Length;
        return true;
    }
}

列表 B,一个微基准:

class Program
{
    static void Main(string[] args)
    {
        for (int numtrials = 3; numtrials > 0; --numtrials)
        {
            using (ProcessorAffinity.BeginAffinity(0))
            {
                int pagesize = 1024 * 10;
                int numpages = 1024;
                int totalbytes = pagesize * numpages;

                BoundedSingleProducerSingleConsumerQueue<byte> lockFreeByteQueue = new BoundedSingleProducerSingleConsumerQueue<byte>(1024 * 32);
                Stopwatch sw = new Stopwatch();
                sw.Start();
                ThreadPool.QueueUserWorkItem(delegate(object state)
                {
                    using (ProcessorAffinity.BeginAffinity(1))
                    {
                        for (int i = 0; i < totalbytes; i++)
                        {
                            while (!lockFreeByteQueue.TryEnqueue((byte)(i & 0xFF))) ;
                        }
                    }
                });
                for (int i = 0; i < totalbytes; i++)
                {
                    byte tmp;
                    while (!lockFreeByteQueue.TryDequeue(out tmp)) ;
                }
                sw.Stop();
                Console.WriteLine("LockFreeByteQueue: {0}ms", sw.ElapsedMilliseconds);


                SizeQueue<byte> byteQueue = new SizeQueue<byte>(1024 * 32);
                sw.Reset();
                sw.Start();
                ThreadPool.QueueUserWorkItem(delegate(object state)
                {
                    using (ProcessorAffinity.BeginAffinity(1))
                    {
                        for (int i = 0; i < totalbytes; i++)
                        {
                            byteQueue.Enqueue((byte)(i & 0xFF));
                        }
                    }
                });

                for (int i = 0; i < totalbytes; i++)
                {
                    byte tmp = byteQueue.Dequeue();
                }
                sw.Stop();
                Console.WriteLine("ByteQueue: {0}ms", sw.ElapsedMilliseconds);

                Console.ReadKey();
            }
        }
    }
}

list C,分块测试:

BoundedSingleProducerSingleConsumerQueue<byte[]> lockfreePageQueue = new BoundedSingleProducerSingleConsumerQueue<byte[]>(32);
sw.Reset();
sw.Start();
ThreadPool.QueueUserWorkItem(delegate(object state)
{
    using (ProcessorAffinity.BeginAffinity(1))
    {
        for (int i = 0; i < numpages; i++)
        {
            byte[] page = new byte[pagesize];
            for (int j = 0; j < pagesize; j++)
            {
                page[j] = (byte)(i & 0xFF);
            }
            while (!lockfreePageQueue.TryEnqueue(page)) ;
        }
    }
});
for (int i = 0; i < numpages; i++)
{
    byte[] page;
    while (!lockfreePageQueue.TryDequeue(out page)) ;
    for (int j = 0; j < pagesize; j++)
    {
        byte tmp = page[j];
    }
}
sw.Stop();
Console.WriteLine("LockFreePageQueue: {0}ms", sw.ElapsedMilliseconds);

SizeQueue<byte[]> pageQueue = new SizeQueue<byte[]>(32);

ThreadPool.QueueUserWorkItem(delegate(object state)
{
    using (ProcessorAffinity.BeginAffinity(1))
    {
        for (int i = 0; i < numpages; i++)
        {
            byte[] page = new byte[pagesize];
            for (int j = 0; j < pagesize; j++)
            {
                page[j] = (byte)(i & 0xFF);
            }
            pageQueue.Enqueue(page);
        }
    }
});
sw.Reset();
sw.Start();
for (int i = 0; i < numpages; i++)
{
    byte[] page = pageQueue.Dequeue();
    for (int j = 0; j < pagesize; j++)
    {
        byte tmp = page[j];
    }
}
sw.Stop();
Console.WriteLine("PageQueue: {0}ms", sw.ElapsedMilliseconds);

关于.net - 线程安全无锁互字节数组队列,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/2613188/

相关文章:

Ruby 线程安全线程创建

java - java中的并发有序列表

c# - 无法建立 SSL 连接 : System. Security.Authentication.AuthenticationException

c# - 如何在 WPF 中右对齐 RadioButton

.net - 如何使用 Socket 和 Reactive 扩展(Rx)从连接的客户端套接字获取接收到的消息缓冲区

linux - IPC 的共享内存和线程的共享内存有什么区别?

python-3.x - 让某种形式的 keras 多处理/线程在 Windows 上工作

.net - sn.exe 可以使用 Windows 证书存储吗?

linux - 良好的 POSIX 线程编程视频培训

windows - 如何在 Perl 中线程安全地设置 STDOUT 编码?