2
\$\begingroup\$

This stream class is designed to hold data only until it is read from the stream. As opposed to the MemoryStream which holds data until the stream is closed. It was written some time ago and was based on a sample found online. I can no longer find the source, but I believe I used it as a start and cleaned it up.

I am interested in reviewing the performance both in terms of speed and memory usage (there are a number of spots that create byte arrays and copy data from one array to another).

/// <summary>
/// This stream maintains data only until the data is read, then it is purged from the stream.
/// </summary>
public class MemoryQueueBufferStream : Stream, IDisposable
{
    //Maintains the streams data.  The Queue object provides an easy and efficient way to add and remove data
    //Each item in the queue represents each write to the stream.  Every call to write translates to an item in the queue
    private Queue<byte[]> lstBuffers_m;

    //If the item on the top of the queue is read, but the amount read is less then the size of the data in the queue,
    //the remaining data is put in this buffer because the item from the queue is removed.  Subsequent read calls will first read
    //from this buffer until it is depleted, then the queue will be read again.
    private byte[] binLeftOverFromFirstBuffer_m = null;

    private bool bIsOpen_m;

    public MemoryQueueBufferStream()
    {
        this.bIsOpen_m = true;
        this.lstBuffers_m = new Queue<byte[]>();
    }

    /// <summary>
    /// Reads up to count bytes from the stream, and removes the read data from the stream.
    /// </summary>
    /// <param name="buffer"></param>
    /// <param name="offset"></param>
    /// <param name="count"></param>
    /// <returns></returns>
    public override int Read(byte[] buffer, int offset, int count)
    {
        this.ValidateBufferArgs(buffer, offset, count);
        if (!this.bIsOpen_m)
        {
            throw new ObjectDisposedException(null, this.GetType().Name + " is closed");
        }

        int iRemainingBytesToRead = count;

        //If there was data left over return it now before reading from the queue
        if (binLeftOverFromFirstBuffer_m != null)
        {
            int iBytesRead = ReadFromBufferAndTrim(ref binLeftOverFromFirstBuffer_m, buffer, offset, iRemainingBytesToRead);
            offset += iBytesRead;
            iRemainingBytesToRead -= iBytesRead;
        }

        while ((iRemainingBytesToRead > 0) && (lstBuffers_m.Count > 0))
        {
            //Get data from the queue, and remove it from the queue, and put it in a buffer
            byte[] binNextBuffer = this.lstBuffers_m.Dequeue();

            int bytesRead = ReadFromBufferAndTrim(ref binNextBuffer, buffer, offset, iRemainingBytesToRead);
            offset += bytesRead;
            iRemainingBytesToRead -= bytesRead;

            if (binNextBuffer != null)
            {
                // If the buffer contains more data than requested, we need to hold it for subsequent reads.
                // ReadFromBufferAndTrim has removed the part of the buffer that was already read. The remaining part will
                // be read at the next call (and possibly trimmed again)
                // before going to the next buffer in the queue.
                this.binLeftOverFromFirstBuffer_m = binNextBuffer;
            }
        }

        int totalBytesRead = count - iRemainingBytesToRead;
        return totalBytesRead;
    }

    private static int ReadFromBufferAndTrim(ref byte[] srcBuffer, byte[] destBuffer, int destOffset, int remainingBytesToRead)
    {
        int bytesToRead = Math.Min(srcBuffer.Length, remainingBytesToRead);
        Buffer.BlockCopy(srcBuffer, 0, destBuffer, destOffset, bytesToRead);

        if (bytesToRead < srcBuffer.Length)
        {
            srcBuffer = TrimBufferStart(srcBuffer, bytesToRead);
        }
        else
        {
            srcBuffer = null;
        }

        return bytesToRead;
    }

    private static byte[] TrimBufferStart(byte[] buffer, int firstByteToKeep)
    {
        int trimmedLength = buffer.Length - firstByteToKeep;
        byte[] trimmedBuffer = new byte[trimmedLength];

        Buffer.BlockCopy(buffer, firstByteToKeep, trimmedBuffer, 0, trimmedLength);
        return trimmedBuffer;
    }

    private void ValidateBufferArgs(byte[] buffer, int offset, int count)
    {
        if (offset < 0)
        {
            throw new ArgumentOutOfRangeException("offset", "offset must be non-negative");
        }
        if (count < 0)
        {
            throw new ArgumentOutOfRangeException("count", "count must be non-negative");
        }
        if ((buffer.Length - offset) < count)
        {
            throw new ArgumentException("requested count exceeds available size");
        }
    }

    /// <summary>
    /// Writes data to the stream
    /// </summary>
    /// <param name="buffer">Data to copy into the stream</param>
    /// <param name="offset"></param>
    /// <param name="count"></param>
    public override void Write(byte[] buffer, int offset, int count)
    {
        this.ValidateBufferArgs(buffer, offset, count);
        if (!bIsOpen_m)
        {
            throw new ObjectDisposedException(null, this.GetType().Name + " is closed");
        }

        byte[] newBuffer = new byte[count];
        Buffer.BlockCopy(buffer, offset, newBuffer, 0, count);

        //Add the data to the queue
        this.lstBuffers_m.Enqueue(newBuffer);
    }

    public override bool CanSeek
    {
        get { return false; }
    }

    public override long Position
    {
        get
        {
            throw new NotSupportedException(this.GetType().Name + " is not seekable");
        }
        set
        {
            throw new NotSupportedException(this.GetType().Name + " is not seekable");
        }
    }

    public override bool CanWrite
    {
        get { return true; }
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        throw new NotSupportedException(this.GetType().Name + " is not seekable");
    }

    public override void SetLength(long value)
    {
        throw new NotSupportedException(this.GetType().Name + " length can not be changed");
    }

    public override bool CanRead
    {
        get { return true; }
    }

    public override long Length
    {
        get
        {
            return this.lstBuffers_m.Sum(b => b.Length) + this.binLeftOverFromFirstBuffer_m.Length;
        }
    }

    public override void Flush()
    {
    }

    #region IDisposable Members

    void IDisposable.Dispose()
    {
        this.lstBuffers_m.Clear();
        this.lstBuffers_m = null;
        this.binLeftOverFromFirstBuffer_m = null;
        this.bIsOpen_m = false;
        this.Close();
    }

    #endregion
}
\$\endgroup\$
  • 1
    \$\begingroup\$ "It is based on"-- You just cut off there. Could you kindly add what it's based on into your question? \$\endgroup\$ – Fund Monica's Lawsuit Jun 10 '15 at 0:08
  • \$\begingroup\$ QPaysTaxes - sorry, I have added the missing text - not that it actually adds much :) \$\endgroup\$ – Jeremy Jun 10 '15 at 3:17
4
\$\begingroup\$

Some quick shots

  • ValidateBufferArgs() is not needed, because the Buffer.BlockCopy() method will handle these cases in the same manner. In addition, why do you call the ValidtaeBufferArgs() method before you check for if (!bIsOpen_m) inside the Write() method ?

  • the naming of your member variables is well strange. First you are using hungarian notation which is an anti pattern if it is used this way. There are ways how this could be acceptable but not like this

    private bool bIsOpen_m; indicating by the b prefix that it is a bool.

    Next, you postfix the member variables with a _m and you always use this. with them. If you pre- or postfix the variables with either m_ or _m than there is no need to use this..

    EDIT:
    if you e.g prefix an int variable like

    int iLength = 1;  
    

    and you later decide that a long would be the better fit, you for sure will change the type, but the possibility exists (at a high level) that you miss to change the variables name too and you would have

    long iLength =1;  
    

    in your code which is just lying.

    Please read also: what-is-the-benefit-of-not-using-hungarian-notation

    From the first answer

    Because its orginal intention (see http://www.joelonsoftware.com/articles/Wrong.html and http://fplanque.net/Blog/devblog/2005/05/11/hungarian_notation_on_steroids) has been misunderstood and it has been (ab)used to help people remember what type a variable is when the language they use is not statically typed. In any statically typed language you do not need the added ballast of prefixes to tell you what type a variable is. In many untyped script languages it can help, but it has often been abused to the point of becoming totally unwieldy. Unfortunately, instead of going back to the original intent of Hungarian notation, people have just made it into one of those "evil" things you should avoid.

    From the second answer

    It uselessly repeats information with no benefit and additional maintenance overhead. What happens when you change your int to a different type like long, now you have to search and replace your entire code base to rename all the variables or they are now semantically wrong which is worse than if you hadn't duplicated the type in the name.

    A fine read about hungarian notation (how it can be done right) is in the first answer: http://www.joelonsoftware.com/articles/Wrong.html

  • you have many places where it is obvious what type you are assigning by checking the right side of the assignment. In such cases you should use the var type and let the compiler determine the correct type.

    byte[] newBuffer = new byte[count];  
    

    would e.g become

    var newBuffer = new byte[count];
    
  • IDisposable

    Please read: proper-use-of-the-idisposable-interface

    If the user calls Dispose() on your object, then everything has been cleaned up. Later on, when the garbage collector comes along and calls Finalize, it will then call Dispose again.

    Not only is this wasteful, but if your object has junk references to objects you already disposed of from the last call to Dispose(), you'll try to dispose them again!

  • Regions

    Please read are-regions-an-antipattern-or-code-smell

    Is there a good use for regions?

    No. There was a legacy use: generated code. Still, code generation tools just have to use partial classes instead. If C# has regions support, it's mostly because this legacy use, and because now that too many people used regions in their code, it would be impossible to remove them without breaking existent codebases.

    Think about it as about goto. The fact that the language or the IDE supports a feature doesn't mean that it should be used daily. StyleCop SA1124 rule is clear: you should not use regions. Never.

\$\endgroup\$
  • \$\begingroup\$ Can you explain the Hungarian notation being an anti0pattern when used that way? I haven't heard that terminology before \$\endgroup\$ – Jeremy Jun 10 '15 at 18:45
  • \$\begingroup\$ I have updated my answer. \$\endgroup\$ – Heslacher Jun 11 '15 at 6:33
1
\$\begingroup\$

I have taken into account some of the ideas @Heslacher mentioned:

  • I removed regions.
  • I removed the IDisposable interface and related functions. Taking a closer look there's nothing unmanaged here that needs cleanup, so I felt it was not necessary.
  • I refactored some of the code. Instead of removing data from the stream as it's read, I am maintaining an index that tells me where in each chunk I need to read on the next call. This drastically reduces the number of times I need to do a BlockRead and instantiate byte arrays, which I think has a performance advantage.

/// <summary>
/// This stream maintains data only until the data is read, then it is purged from the stream.
/// </summary>
public class MemoryQueueBufferStream : Stream
{
    /// <summary>
    /// Represents a single write into the MemoryQueueBufferStream.  Each write is a seperate chunk
    /// </summary>
    private class Chunk
    {
        /// <summary>
        /// As we read through the chunk, the start index will increment.  When we get to the end of the chunk,
        /// we will remove the chunk
        /// </summary>
        public int ChunkReadStartIndex { get; set; }

        /// <summary>
        /// Actual Data
        /// </summary>
        public byte[] Data { get; set; }
    }

    //Maintains the streams data.  The Queue object provides an easy and efficient way to add and remove data
    //Each item in the queue represents each write to the stream.  Every call to write translates to an item in the queue
    private Queue<Chunk> lstBuffers_m;

    public MemoryQueueBufferStream()
    {
        this.lstBuffers_m = new Queue<Chunk>();
    }

    /// <summary>
    /// Reads up to count bytes from the stream, and removes the read data from the stream.
    /// </summary>
    /// <param name="buffer"></param>
    /// <param name="offset"></param>
    /// <param name="count"></param>
    /// <returns></returns>
    public override int Read(byte[] buffer, int offset, int count)
    {
        this.ValidateBufferArgs(buffer, offset, count);

        int iRemainingBytesToRead = count;

        int iTotalBytesRead = 0;

        //Read until we hit the requested count, or until we hav nothing left to read
        while (iTotalBytesRead <= count && lstBuffers_m.Count > 0)
        {
            //Get first chunk from the queue
            Chunk chunk = this.lstBuffers_m.Peek();

            //Determine how much of the chunk there is left to read
            int iUnreadChunkLength = chunk.Data.Length - chunk.ChunkReadStartIndex;

            //Determine how much of the unread part of the chunk we can actually read
            int iBytesToRead = Math.Min(iUnreadChunkLength, iRemainingBytesToRead);

            if (iBytesToRead > 0)
            {
                //Read from the chunk into the buffer
                Buffer.BlockCopy(chunk.Data, chunk.ChunkReadStartIndex, buffer, offset + iTotalBytesRead, iBytesToRead);

                iTotalBytesRead += iBytesToRead;
                iRemainingBytesToRead -= iBytesToRead;

                //If the entire chunk has been read,  remove it
                if (chunk.ChunkReadStartIndex + iBytesToRead >= chunk.Data.Length)
                {
                    this.lstBuffers_m.Dequeue();
                }
                else
                {
                    //Otherwise just update the chunk read start index, so we know where to start reading on the next call
                    chunk.ChunkReadStartIndex = chunk.ChunkReadStartIndex + iBytesToRead;
                }
            }
            else
            {
                break;
            }
        }

        return iTotalBytesRead;
    }

    private void ValidateBufferArgs(byte[] buffer, int offset, int count)
    {
        if (offset < 0)
        {
            throw new ArgumentOutOfRangeException("offset", "offset must be non-negative");
        }
        if (count < 0)
        {
            throw new ArgumentOutOfRangeException("count", "count must be non-negative");
        }
        if ((buffer.Length - offset) < count)
        {
            throw new ArgumentException("requested count exceeds available size");
        }
    }

    /// <summary>
    /// Writes data to the stream
    /// </summary>
    /// <param name="buffer">Data to copy into the stream</param>
    /// <param name="offset"></param>
    /// <param name="count"></param>
    public override void Write(byte[] buffer, int offset, int count)
    {
        this.ValidateBufferArgs(buffer, offset, count);

        //We don't want to use the buffer passed in, as it could be altered by the caller
        byte[] bufSave = new byte[count];
        Buffer.BlockCopy(buffer, offset, bufSave, 0, count);

        //Add the data to the queue
        this.lstBuffers_m.Enqueue(new Chunk() { ChunkReadStartIndex = 0, Data = bufSave });
    }

    public override bool CanSeek
    {
        get { return false; }
    }

    /// <summary>
    /// Always returns 0
    /// </summary>
    public override long Position
    {
        get
        {
            //We're always at the start of the stream, because as the stream purges what we've read
            return 0;  
        }
        set
        {
            throw new NotSupportedException(this.GetType().Name + " is not seekable");
        }
    }

    public override bool CanWrite
    {
        get { return true; }
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        throw new NotSupportedException(this.GetType().Name + " is not seekable");
    }

    public override void SetLength(long value)
    {
        throw new NotSupportedException(this.GetType().Name + " length can not be changed");
    }

    public override bool CanRead
    {
        get { return true; }
    }

    public override long Length
    {
        get
        {

            if (this.lstBuffers_m == null)
            {
                return 0;
            }

            if (this.lstBuffers_m.Count == 0)
            {
                return 0;
            }

            return this.lstBuffers_m.Sum(b => b.Data.Length - b.ChunkReadStartIndex);
        }
    }

    public override void Flush()
    {
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.