Fallback protection without blocking ByteArray queue

You will probably get a lot more speedup if, instead of creating and consuming bytes, you work in chunks. In this case, the "lock-engagement" code probably doesn't matter at all — in fact, the traditional blocking solution may be preferred. I will try to demonstrate.

In C #, an unblocked, single, single, bounded queue is defined. (Listing A)
There are no esoteric blocking operations, not even explicit memory barriers. Let's say that at first glance it is as fast and without blocking as it gets. Is not it? Now compare it to the blocking solution given by Mark Gravell here .

We will be using a dual-processor machine that does not have a shared L3 cache between the cores. We expect a maximum of 2x speedup. A 2x speedup would really mean that a non-blocking solution works perfectly within the theoretical framework.
To create an ideal environment for non-blocking code, we'll even set the processor's proximity to the manufacturer and consumer thread using the utility class here.
The resulting test code is in (Listing B).

It produces approx. 10 MB in one stream, consuming it in another.
The queue size is fixed at 32 KB. If it's full, the producer waits.
A typical test run on my machine looks like this:

LockFreeByteQueue: 799ms
ByteQueue: 1843 ms

Stopping without blocking is faster. Wow, that's more than twice as fast! It's something to brag about. :)
Let's see what happens. The blocking queue mark does just that. He locks. It does this for every byte.

Do we need to block every byte and push data bytes by byte? It certainly comes in chunks on the net (like about 1,000 packages). Even if it does indeed come in bytes from an internal source, the manufacturer can easily package it into good chunks.
Let's just do this - instead of creating and consuming byte bytes, let's work in chunks and add two more micro-benchmark tests (Listing C, just paste it in the breakpoint tag).
Now a typical run looks like this:

LockFreePageQueue: 33ms
PageQueue: 25ms

Now both of them are actually 20x faster than the non-blocking source code - the Marc solution with added chunking is actually faster than the non-blocking code with chunking!
Instead of going with a locked structure that would lead to a 2x speedup, we tried to create another solution that works great with blocking and resulted in a 20x (!) Speedup.
The key to a lot of problems is not so much blocking avoidance - it's much more about avoiding sharing and minimizing blocking. In the above case, we can avoid sharing during the byte copy period.
We can work in a private framework most of the time and then insert one pointer, thus reducing the total space and time to one insertion of one pointer into the queue.

In Listing A, the unblocked, single, single user queue:

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

      

        
        
        
      

    

Listing B, micro benchmark:

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();
            }
        }
    }
}

      

        
        
        
      

    

In Listing C, the tests are verified:

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);