Buffering byte data in C#

Question

My app reads bytes from a TCP socket and needs to buffer them up, so that I can extract messages from them later. Due to the nature of TCP I may get partial or multiple messages in one read, so after each read I would like to inspect the buffer and extract as many full messages as are available.

Therefore I want a class that allows me to do the following:

• append arbitrary byte[] data to it
• inspect the content without consuming it, in particular checking the amount of content and also searching for the existence of a certain byte or bytes
• extract and consume part of the data as a byte[], while leaving the rest in there for a future read

I expect that what I want can be done with 1 or more existing classes in the .NET library, but I'm not sure which ones. System.IO.MemoryStream looks close to what I want, but (a) it isn't clear whether it's suited to being used as a buffer (does the read data get removed from the capacity?) and (b) reads and writes seem to happen at the same place - "The current position of a stream is the position at which the next read or write operation could take place." - which is not what I want. I need to be writing to the end and reading from the front.

Answer 1

I suggest you use MemoryStream under the hood, but encapsulate it in another class which stores:

• The MemoryStream
• The current "read" position
• The current "consumed" position

It will then expose:

• Write: set the stream's position to the end, write data, set the stream's position back to the read position
• Read: read data, set the read position to the stream's position
• Consume: update the consumed position (details based on how you're trying to consume); if the consume position is above a certain threshold, copy the existing buffered data into a new MemoryStream and update all the variables. (You probably don't want to copy the buffer on every consume request.)

Note that none of this will be thread-safe without extra synchronization.

Answer 2

Just use a big byte-array and Array.Copy - it should do the trick. If not, use List<byte>.

If you use the array you have to implement an index to it (where you copy additional data) yourself (same for checking the content-size), but it's straightforward.

If you are interested: here is a simple implementation of a "cyclic buffer". The test should run (I threw a couple unit test at it, but it didn't check all critical path):

public class ReadWriteBuffer
{
    private readonly byte[] _buffer;
    private int _startIndex, _endIndex;

    public ReadWriteBuffer(int capacity)
    {
        _buffer = new byte[capacity];
    }

    public int Count
    {
        get
        {
            if (_endIndex > _startIndex)
                return _endIndex - _startIndex;
            if (_endIndex < _startIndex)
                return (_buffer.Length - _startIndex) + _endIndex;
            return 0;
        }
    }

    public void Write(byte[] data)
    {
        if (Count + data.Length > _buffer.Length)
            throw new Exception("buffer overflow");
        if (_endIndex + data.Length >= _buffer.Length)
        {
            var endLen = _buffer.Length - _endIndex;
            var remainingLen = data.Length - endLen;

            Array.Copy(data, 0, _buffer, _endIndex, endLen);
            Array.Copy(data, endLen, _buffer, 0, remainingLen);
            _endIndex = remainingLen;
        }
        else
        {
            Array.Copy(data, 0, _buffer, _endIndex, data.Length);
            _endIndex += data.Length;
        }
    }

    public byte[] Read(int len, bool keepData = false)
    {
        if (len > Count)
            throw new Exception("not enough data in buffer");
        var result = new byte[len];
        if (_startIndex + len < _buffer.Length)
        {
            Array.Copy(_buffer, _startIndex, result, 0, len);
            if (!keepData)
                _startIndex += len;
            return result;
        }
        else
        {
            var endLen = _buffer.Length - _startIndex;
            var remainingLen = len - endLen;
            Array.Copy(_buffer, _startIndex, result, 0, endLen);
            Array.Copy(_buffer, 0, result, endLen, remainingLen);
            if (!keepData)
                _startIndex = remainingLen;
            return result;
        }
    }

    public byte this[int index]
    {
        get
        {
            if (index >= Count)
                throw new ArgumentOutOfRangeException();
            return _buffer[(_startIndex + index) % _buffer.Length];
        }
    }

    public IEnumerable<byte> Bytes
    {
        get
        {
            for (var i = 0; i < Count; i++)
                yield return _buffer[(_startIndex + i) % _buffer.Length];
        }
    }
}

Please note: the code "consumes" on read - if you don't want that just remove the "_startIndex = ..." parts (or make a overload optional parameter and check or whatever).

Answer 3

I think BufferedStream is the solution to the problem. Also it is possible to go unread len byte of data by calling Seek.

BufferdStream buffer = new BufferedStream(tcpStream, size); // we have a buffer of size
...
...
while(...)
{
    buffer.Read(...);
    // do my staff
    // I read too much, I want to put back len bytes
    buffer.Seek(-len, SeekOrigin.End);
    // I shall come back and read later
}

Growing memory

Contrary to BufferedStream where the size is specified initially, MemoryStream can grow.

Remembering streaming data

MemoryStream holds all the dara-read all the time, while BufferedStream only holds a segment of stream data.

Source stream vs byte array

MemoryStream allows to add input-bytes in Write() method which can be Read() in future. While BufferedSteam takes input-bytes from another source-stream specified in the constructor.

Answer 4

Here's another implementation of a buffer I wrote a while ago:

• Resizeable: allowing to queue up data and not throw buffer overflow exception;
• Efficient: uses a single buffer and Buffer.Copy operations to enqueue/dequeue data

Answer 5

Coming to this late, but for posterity:

When I've done this in the past I've take a slight different approach. If your messages have a fixed header size (that tells you how many bytes in the body), and bearing in mind the network stream is already buffering, I perform the operation in two phases:

• a read on the stream for the bytes for the header
• a subsequent read on the stream for the bytes for the body, based on the header
• repeat

This leverages the fact that - for a stream - when you ask for 'n' bytes you'll never get more back, so you can ignore many of the 'opps I read too many, let me put these aside till next time' issues.

Now this isn't the whole story, to be fair. I had an underlying wrapper class over the stream to handle fragmentation issues (ie if asked for 4 bytes, don't return until 4 bytes received, or stream closed). But that bit is fairly easy.

In my mind the key is to decouple the message handling with the stream mechanics, and if you stop attempting to consume the message as a single ReadBytes() from a stream, life becomes much simpler.

[all of this is true whether your reads are blocking, or async (APM/await)]

Answer 6

There are only three answers here which provide code. One of them is clumsy and the others do not answer the question.

Here a class that you can just copy and paste:

/// <summary>
/// This class is a very fast and threadsafe FIFO buffer
/// </summary>
public class FastFifo
{
    private List<Byte> mi_FifoData = new List<Byte>();

    /// <summary>
    /// Get the count of bytes in the Fifo buffer
    /// </summary>
    public int Count
    {
        get 
        { 
            lock (mi_FifoData)
            {
                return mi_FifoData.Count; 
            }
        }
    }

    /// <summary>
    /// Clears the Fifo buffer
    /// </summary>
    public void Clear()
    {
        lock (mi_FifoData)
        {
            mi_FifoData.Clear();
        }
    }

    /// <summary>
    /// Append data to the end of the fifo
    /// </summary>
    public void Push(Byte[] u8_Data)
    {
        lock (mi_FifoData)
        {
            // Internally the .NET framework uses Array.Copy() which is extremely fast
            mi_FifoData.AddRange(u8_Data);
        }
    }

    /// <summary>
    /// Get data from the beginning of the fifo.
    /// returns null if s32_Count bytes are not yet available.
    /// </summary>
    public Byte[] Pop(int s32_Count)
    {
        lock (mi_FifoData)
        {
            if (mi_FifoData.Count < s32_Count)
                return null;

            // Internally the .NET framework uses Array.Copy() which is extremely fast
            Byte[] u8_PopData = new Byte[s32_Count];
            mi_FifoData.CopyTo(0, u8_PopData, 0, s32_Count);
            mi_FifoData.RemoveRange(0, s32_Count);
            return u8_PopData;
        }
    }

    /// <summary>
    /// Gets a byte without removing it from the Fifo buffer
    /// returns -1 if the index is invalid
    /// </summary>
    public int PeekAt(int s32_Index)
    {
        lock (mi_FifoData)
        {
            if (s32_Index < 0 || s32_Index >= mi_FifoData.Count)
                return -1;

            return mi_FifoData[s32_Index];
        }
    }
}

Answer 7

It sounds like you want to read from the socket into a MemoryStream buffer, and then 'pop' data out of the buffer and reset it each time a certain byte is encountered. It would look something like this:

void ReceiveAllMessages(Action<byte[]> messageReceived, Socket socket)
{
    var currentMessage = new MemoryStream();
    var buffer = new byte[128];

    while (true)
    {
        var read = socket.Receive(buffer, 0, buffer.Length);
        if (read == 0)
            break;     // Connection closed

        for (var i = 0; i < read; i++)
        {
            var currentByte = buffer[i];
            if (currentByte == END_OF_MESSAGE)
            {
                var message = currentMessage.ToByteArray();
                messageReceived(message);

                currentMessage = new MemoryStream();
            }
            else
            {
                currentMessage.Write(currentByte);
            }
        }
    }
}

Answer 8

You could do this with a Stream wrapping a ConcurrentQueue<ArraySegment<byte>> (keep in mind this makes it forward only). However I really dislike the idea of keeping data in memory before doing something with it; it opens you up to a bunch of attacks (intentional or not) regarding the size of the message. You might also want to Google "circular buffer".

You should actually be writing code that does something meaningful with the data as soon as it is received: 'Push Parsing' (this is what, for example, SAX supports). As an example of how you would do this with text:

private Encoding _encoding;
private Decoder _decoder;
private char[] _charData = new char[4];

public PushTextReader(Encoding encoding)
{
    _encoding = encoding;
    _decoder = _encoding.GetDecoder();
}

// A single connection requires its own decoder
// and charData. That connection should never
// call this method from multiple threads
// simultaneously.
// If you are using the ReadAsyncLoop you
// don't need to worry about it.
public void ReceiveData(ArraySegment<byte> data)
{
    // The two false parameters cause the decoder
    // to accept 'partial' characters.
    var charCount = _decoder.GetCharCount(data.Array, data.Offset, data.Count, false);
    charCount = _decoder.GetChars(data.Array, data.Offset, data.Count, _charData, 0, false);
    OnCharacterData(new ArraySegment<char>(_charData, 0, charCount));
}

If you must be able to accept complete messages before deserializing them, you can use a MemoryMappedFile, which has the advantage that the sending entity won't be able to out-of-memory your server. What gets tricky is resetting the file back to zero; because that can bunch of issues. One way to tackle this is by:

TCP Receiver End

1. Write to the current stream.
2. If the stream exceeds a certain length move to a new one.

Deserialization End

1. Read from the current stream.
2. Once you have emptied the stream destroy it.

The TCP receiver end is very simple. The deserializer end will need some elementary buffer stitching logic (remember to use Buffer.BlockCopy and not Array.Copy).

Side note: Sounds like a fun project, if I have the time and remember I might go ahead and implement this system.