1
\$\begingroup\$

A custom stream that is used to perform read and write operations directly into the page cloud blob

public sealed class WindowsAzureCloudPageBlobStream : Stream
{
    // 4 MB is the top most limit for page blob write operations
    public const int MaxPageWriteCapacity = 4 * 1024 * 1024;

    // Every operation on a page blob has to manipulate a value which is rounded up to 512 bytes
    private const int PageBlobPageAdjustmentSize = 512;

    private CloudPageBlob _pageBlob;
    private long _streamLength;

    public WindowsAzureCloudPageBlobStream(CloudPageBlob pageBlob) : this(pageBlob, PageBlobPageAdjustmentSize)
    {
    }

    public WindowsAzureCloudPageBlobStream(CloudPageBlob pageBlob, long initialCapacity)
    {
        _pageBlob = pageBlob;
        Position = 0;

        if (_pageBlob.Exists())
        {
            _streamLength = GetCurrentCapacity();
        }
        else
        {
            _pageBlob.Create(RoundUpToPageBlobSize(initialCapacity));
            _streamLength = 0;
        }
    }

    public override bool CanRead
    {
        get
        {
            return _pageBlob != null;
        }
    }

    public override bool CanSeek
    {
        get
        {
            return _pageBlob != null;
        }
    }

    public override bool CanWrite
    {
        get
        {
            return _pageBlob != null;
        }
    }

    public override long Length
    {
        get
        {
            return _streamLength;
        }
    }

    public override long Position { get; set; }

    public override void Flush() { /*No need to flush*/ }

    public override long Seek(long offset, SeekOrigin origin)
    {
        switch (origin)
        {
            case SeekOrigin.Begin:
                if (offset > Length)
                {
                    var errorMessage = SafeFormatter.Format(CloudLocalization.Culture, CloudLocalization.OffsetOutOfRange, Length, offset);
                    throw new InvalidOperationException(errorMessage);
                }

                Position = offset;
                break;

            case SeekOrigin.Current:
                if (Position + offset > Length)
                {
                    var errorMessage = SafeFormatter.Format(CloudLocalization.Culture, CloudLocalization.OffsetOutOfRange, Length, Position + offset);
                    throw new InvalidOperationException(errorMessage);
                }

                Position += offset;
                break;

            case SeekOrigin.End:
                if (Length + offset > Length)
                {
                    var errorMessage = SafeFormatter.Format(CloudLocalization.Culture, CloudLocalization.OffsetOutOfRange, Length, Length + offset);
                    throw new InvalidOperationException(errorMessage);
                }

                Position = Length + offset;
                break;

            default: throw new InvalidOperationException(SafeFormatter.Format(CloudLocalization.Culture, CloudLocalization.SeekOperationNotSupported, origin));
        }

        return Position;
    }

    public override void SetLength(long value)
    {
        var newLength = RoundUpToPageBlobSize(value);
        if (newLength == Length) { return; }

        var capacity = GetCurrentCapacity();
        if (newLength > capacity) { Resize(newLength); }

        _streamLength = newLength;
        if (Position > newLength) { Position = newLength; }
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        using (var readStream = _pageBlob.OpenRead())
        {
            readStream.Seek(Position, SeekOrigin.Begin);

            var readCount = readStream.Read(buffer, offset, count);
            Position += readCount;

            return readCount;
        }
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        var additionalOffset = 0;
        var bytesToWriteTotal = count;

        while (bytesToWriteTotal > 0)
        {
            var bytesToWriteTotalAdjusted = RoundUpToPageBlobSize(bytesToWriteTotal);

            // If we do not have enough space in the cloud, we'll reserve more
            var capacity = GetCurrentCapacity();
            var delta = Position + bytesToWriteTotalAdjusted - capacity;
            if (delta > 0) { Resize(capacity + delta); }

            // Azure does not allow us to write as many bytes as we want
            // Max allowed size per write is 4MB
            var bytesToWriteNow = Math.Min((int)bytesToWriteTotalAdjusted, MaxPageWriteCapacity);
            var adjustmentBuffer = new byte[bytesToWriteNow];

            // In case we need to write a number of bytes that cannot be splitted into 512-byte blocks
            // We need to adjust the buffer:
            // 1. Read from current offset a chunk that can be splitted into 512-byte blocks
            // 2. Merge new data with read data
            // 3. Write the modified data back
            if (bytesToWriteTotal < bytesToWriteTotalAdjusted && bytesToWriteTotal < MaxPageWriteCapacity)
            {
                var currentPosition = Position;
                var bytesRead = Read(adjustmentBuffer, 0, adjustmentBuffer.Length);

                // If we fail to read the required amount of bytes, it means that our stream is not valid
                if (bytesRead != adjustmentBuffer.Length)
                {
                    var errorMessage = SafeFormatter.Format(CloudLocalization.Culture, CloudLocalization.ReadWriteMissmatch, bytesRead, adjustmentBuffer.Length);
                    throw new InvalidOperationException(errorMessage);
                }

                Position = currentPosition;
                Array.Copy(buffer, offset + additionalOffset, adjustmentBuffer, 0, bytesToWriteTotal);
            }
            else
            {
                Array.Copy(buffer, offset + additionalOffset, adjustmentBuffer, 0, bytesToWriteNow);
            }

            // We cannot write in the middle of the page, need to adjust our position:
            // 1. Set position to the closest full block offset
            // 2. Read additional data from stream
            // 3. Adjust current buffer to include read data
            var positionSetToStartOfPage = (Position / PageBlobPageAdjustmentSize) * PageBlobPageAdjustmentSize;
            var steps = Position - positionSetToStartOfPage;
            if (steps != 0)
            {
                Position = positionSetToStartOfPage;
                var positionCorrectionBuffer = new byte[steps];

                // If we fail to read the required amount of bytes, it means that our stream is not valid
                var bytesRead = Read(positionCorrectionBuffer, 0, (int)steps);
                if (bytesRead != steps)
                {
                    var errorMessage = SafeFormatter.Format(CloudLocalization.Culture, CloudLocalization.ReadWriteMissmatch, bytesRead, steps);
                    throw new InvalidOperationException(errorMessage);
                }

                Position = positionSetToStartOfPage;
                var tempBuffer = new byte[adjustmentBuffer.Length];

                Array.Copy(positionCorrectionBuffer, 0, tempBuffer, 0, positionCorrectionBuffer.Length);
                Array.Copy(adjustmentBuffer, 0, tempBuffer, positionCorrectionBuffer.Length, adjustmentBuffer.Length - steps);

                adjustmentBuffer = tempBuffer;
            }

            using (var memoryStream = new MemoryStream(adjustmentBuffer, 0, bytesToWriteNow, false, false))
            {
                _pageBlob.WritePages(memoryStream, Position);
            }

            // If we had to make adjustments to stream position, we failed to write some of the data from initial buffer
            // Mark that data as non-processed and try to write it during the next iteration
            if (steps != 0) { bytesToWriteNow -= (int)steps; }

            Position += bytesToWriteNow;
            additionalOffset += bytesToWriteNow;
            bytesToWriteTotal -= bytesToWriteNow;

            if (Position > Length) { _streamLength = Position; }
        }
    }

    protected override void Dispose(bool disposing)
    {
        // Free the overused space
        var capacity = GetCurrentCapacity();
        if (Length < capacity) { Resize(Length); }

        _pageBlob = null;
        base.Dispose(disposing);
    }

    private void Resize(long value)
    {
        var newCapacity = RoundUpToPageBlobSize(value);
        _pageBlob.Resize(newCapacity);

        if (Length > newCapacity) { _streamLength = newCapacity; }
        if (Position > newCapacity) { Position = newCapacity; }
    }

    private long GetCurrentCapacity()
    {
        _pageBlob.FetchAttributes();
        return _pageBlob.Properties.Length;
    }

    private static long RoundUpToPageBlobSize(long size) { 
       return (size + PageBlobPageAdjustmentSize - 1) & ~(PageBlobPageAdjustmentSize - 1); }
    }
}

How can I improve my algorithm? Do I have any weaknesses in the solution?

\$\endgroup\$
1
\$\begingroup\$

If any of your constructors is called with an passed pageBlob which is null you get an NullReferenceException at the point of calling if (_pageBlob.Exists()) which is bad. You should throw an ArgumentNullExceptionat the start of that method if pageBlob is null.

Based on that I can't see how the CanXXX properties should return false. If the object (WindowsAzureCloudPageBlobStream) is disposed these properties should throw an ObjectDisposedException.


public override long Seek(long offset, SeekOrigin origin)

You don't check if the given offset is negativ. Assume the passed in origin == SeekOrigin.Begin and the offset == -1 now Position == -1 which will in the Read() method by seeking to Position result in an IOException because that makes an attempt to place the Position before the beginning of the stream.

| improve this answer | |
\$\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.