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?
