mirror of
https://github.com/adamhathcock/sharpcompress.git
synced 2026-07-08 18:16:30 +00:00
First pass of implementation
This commit is contained in:
Adam Hathcock
992
src/SharpCompress/IO/PooledMemoryStream.cs
Normal file
992
src/SharpCompress/IO/PooledMemoryStream.cs
Normal file
@@ -0,0 +1,992 @@
|
||||
using System;
|
||||
using System.Buffers;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace SharpCompress.IO;
|
||||
|
||||
/// <summary>
|
||||
/// MemoryStream implementation backed by pooled byte arrays.
|
||||
/// </summary>
|
||||
public sealed class PooledMemoryStream : MemoryStream
|
||||
{
|
||||
public const int DefaultBlockSize = 81920;
|
||||
|
||||
private const int MaxStreamLength = int.MaxValue;
|
||||
|
||||
private enum StorageMode
|
||||
{
|
||||
Segmented,
|
||||
Contiguous,
|
||||
}
|
||||
|
||||
private readonly ArrayPool<byte> _arrayPool;
|
||||
private readonly int _blockSize;
|
||||
|
||||
private readonly List<byte[]> _detachedExposedBuffers = new();
|
||||
|
||||
private StorageMode _mode;
|
||||
private List<byte[]>? _blocks;
|
||||
private byte[]? _contiguousBuffer;
|
||||
|
||||
private bool _contiguousBufferExposed;
|
||||
private bool _isOpen;
|
||||
private bool _writable;
|
||||
private bool _expandable;
|
||||
private bool _exposable;
|
||||
|
||||
private int _origin;
|
||||
private int _position;
|
||||
private int _length;
|
||||
private int _capacity;
|
||||
private int _allocatedCapacity;
|
||||
|
||||
public PooledMemoryStream()
|
||||
: this(0) { }
|
||||
|
||||
public PooledMemoryStream(int capacity)
|
||||
: this(capacity, DefaultBlockSize, ArrayPool<byte>.Shared) { }
|
||||
|
||||
public PooledMemoryStream(int capacity, int blockSize)
|
||||
: this(capacity, blockSize, ArrayPool<byte>.Shared) { }
|
||||
|
||||
public PooledMemoryStream(int capacity, int blockSize, ArrayPool<byte> arrayPool)
|
||||
{
|
||||
ThrowHelper.ThrowIfNull(arrayPool, nameof(arrayPool));
|
||||
ThrowHelper.ThrowIfNegative(capacity, nameof(capacity));
|
||||
ThrowHelper.ThrowIfNegativeOrZero(blockSize, nameof(blockSize));
|
||||
|
||||
_arrayPool = arrayPool;
|
||||
_blockSize = blockSize;
|
||||
|
||||
_mode = StorageMode.Segmented;
|
||||
_blocks = new List<byte[]>();
|
||||
_isOpen = true;
|
||||
_writable = true;
|
||||
_expandable = true;
|
||||
_exposable = true;
|
||||
_origin = 0;
|
||||
_position = 0;
|
||||
_length = 0;
|
||||
_capacity = capacity;
|
||||
|
||||
EnsureSegmentedAllocated(capacity);
|
||||
}
|
||||
|
||||
public PooledMemoryStream(byte[] buffer)
|
||||
: this(buffer, writable: true) { }
|
||||
|
||||
public PooledMemoryStream(byte[] buffer, bool writable)
|
||||
: this(buffer, 0, buffer?.Length ?? 0, writable, publiclyVisible: false) { }
|
||||
|
||||
public PooledMemoryStream(byte[] buffer, int index, int count)
|
||||
: this(buffer, index, count, writable: true, publiclyVisible: false) { }
|
||||
|
||||
public PooledMemoryStream(byte[] buffer, int index, int count, bool writable)
|
||||
: this(buffer, index, count, writable, publiclyVisible: false) { }
|
||||
|
||||
public PooledMemoryStream(
|
||||
byte[] buffer,
|
||||
int index,
|
||||
int count,
|
||||
bool writable,
|
||||
bool publiclyVisible
|
||||
)
|
||||
{
|
||||
ThrowHelper.ThrowIfNull(buffer, nameof(buffer));
|
||||
ThrowHelper.ThrowIfNegative(index, nameof(index));
|
||||
ThrowHelper.ThrowIfNegative(count, nameof(count));
|
||||
if (buffer.Length - index < count)
|
||||
{
|
||||
throw new ArgumentException("Offset and length are out of bounds.");
|
||||
}
|
||||
|
||||
_arrayPool = ArrayPool<byte>.Shared;
|
||||
_blockSize = DefaultBlockSize;
|
||||
|
||||
_mode = StorageMode.Segmented;
|
||||
_blocks = new List<byte[]>();
|
||||
_origin = 0;
|
||||
_position = 0;
|
||||
_length = count;
|
||||
_capacity = count;
|
||||
_writable = writable;
|
||||
_expandable = false;
|
||||
_exposable = publiclyVisible;
|
||||
_isOpen = true;
|
||||
|
||||
EnsureSegmentedAllocated(_capacity);
|
||||
if (count > 0)
|
||||
{
|
||||
CopyToSegmented(0, buffer, index, count);
|
||||
}
|
||||
}
|
||||
|
||||
public override bool CanRead => _isOpen;
|
||||
|
||||
public override bool CanSeek => _isOpen;
|
||||
|
||||
public override bool CanWrite => _writable;
|
||||
|
||||
public override long Length
|
||||
{
|
||||
get
|
||||
{
|
||||
EnsureNotClosed();
|
||||
return _length - _origin;
|
||||
}
|
||||
}
|
||||
|
||||
public override long Position
|
||||
{
|
||||
get
|
||||
{
|
||||
EnsureNotClosed();
|
||||
return _position - _origin;
|
||||
}
|
||||
set
|
||||
{
|
||||
ThrowHelper.ThrowIfNegative(value, nameof(value));
|
||||
EnsureNotClosed();
|
||||
ThrowHelper.ThrowIfGreaterThan(value, MaxStreamLength - _origin, nameof(value));
|
||||
|
||||
_position = _origin + (int)value;
|
||||
}
|
||||
}
|
||||
|
||||
public override int Capacity
|
||||
{
|
||||
get
|
||||
{
|
||||
EnsureNotClosed();
|
||||
return _capacity - _origin;
|
||||
}
|
||||
set
|
||||
{
|
||||
if (value < Length)
|
||||
{
|
||||
throw new ArgumentOutOfRangeException(nameof(value));
|
||||
}
|
||||
|
||||
EnsureNotClosed();
|
||||
|
||||
if (!_expandable && value != Capacity)
|
||||
{
|
||||
throw new NotSupportedException("Memory stream is not expandable.");
|
||||
}
|
||||
|
||||
var target = _origin + value;
|
||||
if (target == _capacity)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
SetCapacityAbsolute(target);
|
||||
}
|
||||
}
|
||||
|
||||
public override void Flush() { }
|
||||
|
||||
public override Task FlushAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
return Task.FromCanceled(cancellationToken);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public override long Seek(long offset, SeekOrigin loc)
|
||||
{
|
||||
EnsureNotClosed();
|
||||
|
||||
var anchor = loc switch
|
||||
{
|
||||
SeekOrigin.Begin => _origin,
|
||||
SeekOrigin.Current => _position,
|
||||
SeekOrigin.End => _length,
|
||||
_ => throw new ArgumentException("Invalid seek origin.", nameof(loc)),
|
||||
};
|
||||
|
||||
var target = anchor + offset;
|
||||
if (target < _origin)
|
||||
{
|
||||
throw new IOException("Attempted to seek before the beginning of the stream.");
|
||||
}
|
||||
|
||||
if (target > MaxStreamLength)
|
||||
{
|
||||
throw new ArgumentOutOfRangeException(nameof(offset));
|
||||
}
|
||||
|
||||
_position = (int)target;
|
||||
return _position - _origin;
|
||||
}
|
||||
|
||||
public override void SetLength(long value)
|
||||
{
|
||||
ThrowHelper.ThrowIfNegative(value, nameof(value));
|
||||
ThrowHelper.ThrowIfGreaterThan(value, MaxStreamLength, nameof(value));
|
||||
|
||||
EnsureWritable();
|
||||
|
||||
var newLength = _origin + (int)value;
|
||||
if (newLength > _capacity)
|
||||
{
|
||||
EnsureCapacityForAppend(newLength);
|
||||
}
|
||||
|
||||
if (newLength > _length)
|
||||
{
|
||||
ClearRange(_length, newLength - _length);
|
||||
}
|
||||
|
||||
_length = newLength;
|
||||
if (_position > newLength)
|
||||
{
|
||||
_position = newLength;
|
||||
}
|
||||
}
|
||||
|
||||
public override int Read(byte[] buffer, int offset, int count)
|
||||
{
|
||||
ValidateReadWriteBufferArguments(buffer, offset, count);
|
||||
EnsureNotClosed();
|
||||
|
||||
var available = _length - _position;
|
||||
if (available <= 0)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (count > available)
|
||||
{
|
||||
count = available;
|
||||
}
|
||||
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
Buffer.BlockCopy(_contiguousBuffer!, _position, buffer, offset, count);
|
||||
break;
|
||||
case StorageMode.Segmented:
|
||||
CopyFromSegmented(_position, buffer, offset, count);
|
||||
break;
|
||||
}
|
||||
|
||||
_position += count;
|
||||
return count;
|
||||
}
|
||||
|
||||
public override int ReadByte()
|
||||
{
|
||||
EnsureNotClosed();
|
||||
if (_position >= _length)
|
||||
{
|
||||
return -1;
|
||||
}
|
||||
|
||||
byte value;
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
value = _contiguousBuffer![_position];
|
||||
break;
|
||||
default:
|
||||
{
|
||||
var blockIndex = _position / _blockSize;
|
||||
var blockOffset = _position % _blockSize;
|
||||
value = _blocks![blockIndex][blockOffset];
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
_position++;
|
||||
return value;
|
||||
}
|
||||
|
||||
public override void Write(byte[] buffer, int offset, int count)
|
||||
{
|
||||
ValidateReadWriteBufferArguments(buffer, offset, count);
|
||||
EnsureWritable();
|
||||
|
||||
if (count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var endPosition = _position + count;
|
||||
if (endPosition < 0)
|
||||
{
|
||||
throw new IOException("Stream is too long.");
|
||||
}
|
||||
|
||||
if (endPosition > _capacity)
|
||||
{
|
||||
EnsureCapacityForAppend(endPosition);
|
||||
}
|
||||
|
||||
if (_position > _length)
|
||||
{
|
||||
ClearRange(_length, _position - _length);
|
||||
}
|
||||
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
Buffer.BlockCopy(buffer, offset, _contiguousBuffer!, _position, count);
|
||||
break;
|
||||
case StorageMode.Segmented:
|
||||
CopyToSegmented(_position, buffer, offset, count);
|
||||
break;
|
||||
}
|
||||
|
||||
_position = endPosition;
|
||||
if (_position > _length)
|
||||
{
|
||||
_length = _position;
|
||||
}
|
||||
}
|
||||
|
||||
public override void WriteByte(byte value)
|
||||
{
|
||||
EnsureWritable();
|
||||
|
||||
var endPosition = _position + 1;
|
||||
if (endPosition < 0)
|
||||
{
|
||||
throw new IOException("Stream is too long.");
|
||||
}
|
||||
|
||||
if (endPosition > _capacity)
|
||||
{
|
||||
EnsureCapacityForAppend(endPosition);
|
||||
}
|
||||
|
||||
if (_position > _length)
|
||||
{
|
||||
ClearRange(_length, _position - _length);
|
||||
}
|
||||
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
_contiguousBuffer![_position] = value;
|
||||
break;
|
||||
default:
|
||||
{
|
||||
var blockIndex = _position / _blockSize;
|
||||
var blockOffset = _position % _blockSize;
|
||||
_blocks![blockIndex][blockOffset] = value;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
_position = endPosition;
|
||||
if (_position > _length)
|
||||
{
|
||||
_length = _position;
|
||||
}
|
||||
}
|
||||
|
||||
public override byte[] GetBuffer()
|
||||
{
|
||||
EnsureNotClosed();
|
||||
if (!_exposable)
|
||||
{
|
||||
throw new UnauthorizedAccessException("Memory stream buffer is not publicly visible.");
|
||||
}
|
||||
|
||||
EnsureContiguous();
|
||||
_contiguousBufferExposed = true;
|
||||
return _contiguousBuffer!;
|
||||
}
|
||||
|
||||
public override bool TryGetBuffer(out ArraySegment<byte> buffer)
|
||||
{
|
||||
if (!_exposable)
|
||||
{
|
||||
buffer = default;
|
||||
return false;
|
||||
}
|
||||
|
||||
EnsureNotClosed();
|
||||
|
||||
EnsureContiguous();
|
||||
_contiguousBufferExposed = true;
|
||||
buffer = new ArraySegment<byte>(_contiguousBuffer!, 0, _length);
|
||||
return true;
|
||||
}
|
||||
|
||||
public override byte[] ToArray()
|
||||
{
|
||||
EnsureNotClosed();
|
||||
|
||||
var count = _length - _origin;
|
||||
if (count == 0)
|
||||
{
|
||||
return Array.Empty<byte>();
|
||||
}
|
||||
|
||||
var copy = new byte[count];
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
Buffer.BlockCopy(_contiguousBuffer!, _origin, copy, 0, count);
|
||||
break;
|
||||
case StorageMode.Segmented:
|
||||
CopyFromSegmented(_origin, copy, 0, count);
|
||||
break;
|
||||
}
|
||||
|
||||
return copy;
|
||||
}
|
||||
|
||||
public override void WriteTo(Stream stream)
|
||||
{
|
||||
ThrowHelper.ThrowIfNull(stream, nameof(stream));
|
||||
EnsureNotClosed();
|
||||
|
||||
var count = _length - _origin;
|
||||
if (count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
stream.Write(_contiguousBuffer!, _origin, count);
|
||||
break;
|
||||
case StorageMode.Segmented:
|
||||
{
|
||||
var position = _origin;
|
||||
var remaining = count;
|
||||
while (remaining > 0)
|
||||
{
|
||||
var blockIndex = position / _blockSize;
|
||||
var blockOffset = position % _blockSize;
|
||||
var toWrite = Math.Min(remaining, _blockSize - blockOffset);
|
||||
stream.Write(_blocks![blockIndex], blockOffset, toWrite);
|
||||
position += toWrite;
|
||||
remaining -= toWrite;
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public override Task<int> ReadAsync(
|
||||
byte[] buffer,
|
||||
int offset,
|
||||
int count,
|
||||
CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
return Task.FromCanceled<int>(cancellationToken);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
return Task.FromResult(Read(buffer, offset, count));
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return Task.FromException<int>(ex);
|
||||
}
|
||||
}
|
||||
|
||||
public override Task WriteAsync(
|
||||
byte[] buffer,
|
||||
int offset,
|
||||
int count,
|
||||
CancellationToken cancellationToken
|
||||
)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
return Task.FromCanceled(cancellationToken);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
Write(buffer, offset, count);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return Task.FromException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
#if !LEGACY_DOTNET
|
||||
public override int Read(Span<byte> buffer)
|
||||
{
|
||||
EnsureNotClosed();
|
||||
|
||||
var available = _length - _position;
|
||||
if (available <= 0)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
var count = Math.Min(available, buffer.Length);
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
_contiguousBuffer.AsSpan(_position, count).CopyTo(buffer);
|
||||
break;
|
||||
case StorageMode.Segmented:
|
||||
{
|
||||
var sourcePosition = _position;
|
||||
var destinationOffset = 0;
|
||||
var remaining = count;
|
||||
|
||||
while (remaining > 0)
|
||||
{
|
||||
var blockIndex = sourcePosition / _blockSize;
|
||||
var blockOffset = sourcePosition % _blockSize;
|
||||
var toCopy = Math.Min(remaining, _blockSize - blockOffset);
|
||||
_blocks!
|
||||
[blockIndex]
|
||||
.AsSpan(blockOffset, toCopy)
|
||||
.CopyTo(buffer.Slice(destinationOffset, toCopy));
|
||||
|
||||
sourcePosition += toCopy;
|
||||
destinationOffset += toCopy;
|
||||
remaining -= toCopy;
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
_position += count;
|
||||
return count;
|
||||
}
|
||||
|
||||
public override void Write(ReadOnlySpan<byte> buffer)
|
||||
{
|
||||
EnsureWritable();
|
||||
if (buffer.Length == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var endPosition = _position + buffer.Length;
|
||||
if (endPosition < 0)
|
||||
{
|
||||
throw new IOException("Stream is too long.");
|
||||
}
|
||||
|
||||
if (endPosition > _capacity)
|
||||
{
|
||||
EnsureCapacityForAppend(endPosition);
|
||||
}
|
||||
|
||||
if (_position > _length)
|
||||
{
|
||||
ClearRange(_length, _position - _length);
|
||||
}
|
||||
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
buffer.CopyTo(_contiguousBuffer.AsSpan(_position, buffer.Length));
|
||||
break;
|
||||
case StorageMode.Segmented:
|
||||
{
|
||||
var sourceOffset = 0;
|
||||
var destinationPosition = _position;
|
||||
var remaining = buffer.Length;
|
||||
|
||||
while (remaining > 0)
|
||||
{
|
||||
var blockIndex = destinationPosition / _blockSize;
|
||||
var blockOffset = destinationPosition % _blockSize;
|
||||
var toCopy = Math.Min(remaining, _blockSize - blockOffset);
|
||||
|
||||
buffer
|
||||
.Slice(sourceOffset, toCopy)
|
||||
.CopyTo(_blocks![blockIndex].AsSpan(blockOffset, toCopy));
|
||||
|
||||
sourceOffset += toCopy;
|
||||
destinationPosition += toCopy;
|
||||
remaining -= toCopy;
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
_position = endPosition;
|
||||
if (_position > _length)
|
||||
{
|
||||
_length = _position;
|
||||
}
|
||||
}
|
||||
|
||||
public override ValueTask<int> ReadAsync(
|
||||
Memory<byte> buffer,
|
||||
CancellationToken cancellationToken = default
|
||||
)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
return ValueTask.FromCanceled<int>(cancellationToken);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
return ValueTask.FromResult(Read(buffer.Span));
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return ValueTask.FromException<int>(ex);
|
||||
}
|
||||
}
|
||||
|
||||
public override ValueTask WriteAsync(
|
||||
ReadOnlyMemory<byte> buffer,
|
||||
CancellationToken cancellationToken = default
|
||||
)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
return ValueTask.FromCanceled(cancellationToken);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
Write(buffer.Span);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return ValueTask.FromException(ex);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (_isOpen)
|
||||
{
|
||||
_isOpen = false;
|
||||
_writable = false;
|
||||
_expandable = false;
|
||||
|
||||
if (disposing)
|
||||
{
|
||||
ReturnPooledBuffers();
|
||||
}
|
||||
}
|
||||
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
private void EnsureNotClosed()
|
||||
{
|
||||
if (!_isOpen)
|
||||
{
|
||||
throw new ObjectDisposedException(nameof(PooledMemoryStream));
|
||||
}
|
||||
}
|
||||
|
||||
private void EnsureWritable()
|
||||
{
|
||||
EnsureNotClosed();
|
||||
if (!_writable)
|
||||
{
|
||||
throw new NotSupportedException("Stream does not support writing.");
|
||||
}
|
||||
}
|
||||
|
||||
private void EnsureCapacityForAppend(int requiredLength)
|
||||
{
|
||||
if (requiredLength < 0)
|
||||
{
|
||||
throw new IOException("Stream is too long.");
|
||||
}
|
||||
|
||||
if (requiredLength <= _capacity)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (!_expandable)
|
||||
{
|
||||
throw new NotSupportedException("Memory stream is not expandable.");
|
||||
}
|
||||
|
||||
var nextCapacity = RoundUpToBlockBoundary(requiredLength);
|
||||
SetCapacityAbsolute(nextCapacity);
|
||||
}
|
||||
|
||||
private void SetCapacityAbsolute(int newCapacity)
|
||||
{
|
||||
ThrowHelper.ThrowIfLessThan(newCapacity, _length, nameof(newCapacity));
|
||||
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
if (newCapacity > _allocatedCapacity)
|
||||
{
|
||||
DemoteContiguousToSegmented();
|
||||
EnsureSegmentedAllocated(newCapacity);
|
||||
}
|
||||
break;
|
||||
|
||||
case StorageMode.Segmented:
|
||||
EnsureSegmentedAllocated(newCapacity);
|
||||
break;
|
||||
}
|
||||
|
||||
_capacity = newCapacity;
|
||||
if (_length > _capacity)
|
||||
{
|
||||
_length = _capacity;
|
||||
}
|
||||
if (_position > _capacity)
|
||||
{
|
||||
_position = _capacity;
|
||||
}
|
||||
}
|
||||
|
||||
private void EnsureContiguous()
|
||||
{
|
||||
if (_mode == StorageMode.Contiguous)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var requested = Math.Max(_capacity, 1);
|
||||
var contiguous = _arrayPool.Rent(requested);
|
||||
if (_length > 0)
|
||||
{
|
||||
CopyFromSegmented(0, contiguous, 0, _length);
|
||||
}
|
||||
|
||||
ReturnSegmentedBlocks();
|
||||
|
||||
_mode = StorageMode.Contiguous;
|
||||
_contiguousBuffer = contiguous;
|
||||
_contiguousBufferExposed = false;
|
||||
_allocatedCapacity = contiguous.Length;
|
||||
}
|
||||
|
||||
private void DemoteContiguousToSegmented()
|
||||
{
|
||||
var contiguous = _contiguousBuffer;
|
||||
if (contiguous is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var requiredCapacity = Math.Max(_capacity, _length);
|
||||
_mode = StorageMode.Segmented;
|
||||
_blocks = new List<byte[]>();
|
||||
_contiguousBuffer = null;
|
||||
EnsureSegmentedAllocated(requiredCapacity);
|
||||
|
||||
if (_length > 0)
|
||||
{
|
||||
CopyToSegmented(0, contiguous, 0, _length);
|
||||
}
|
||||
|
||||
if (_contiguousBufferExposed)
|
||||
{
|
||||
_detachedExposedBuffers.Add(contiguous);
|
||||
_contiguousBufferExposed = false;
|
||||
}
|
||||
else
|
||||
{
|
||||
_arrayPool.Return(contiguous);
|
||||
}
|
||||
}
|
||||
|
||||
private void EnsureSegmentedAllocated(int capacity)
|
||||
{
|
||||
if (_mode != StorageMode.Segmented)
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
"Segmented allocation requested while not in segmented mode."
|
||||
);
|
||||
}
|
||||
|
||||
var requiredAllocated = RoundUpToBlockBoundary(capacity);
|
||||
var requiredBlocks = requiredAllocated == 0 ? 0 : requiredAllocated / _blockSize;
|
||||
|
||||
_blocks ??= new List<byte[]>();
|
||||
|
||||
while (_blocks.Count < requiredBlocks)
|
||||
{
|
||||
_blocks.Add(_arrayPool.Rent(_blockSize));
|
||||
}
|
||||
|
||||
while (_blocks.Count > requiredBlocks)
|
||||
{
|
||||
var index = _blocks.Count - 1;
|
||||
var block = _blocks[index];
|
||||
_blocks.RemoveAt(index);
|
||||
_arrayPool.Return(block);
|
||||
}
|
||||
|
||||
_allocatedCapacity = requiredAllocated;
|
||||
}
|
||||
|
||||
private int RoundUpToBlockBoundary(int value)
|
||||
{
|
||||
if (value <= 0)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
var rounded = ((long)value + _blockSize - 1) / _blockSize * _blockSize;
|
||||
if (rounded > MaxStreamLength)
|
||||
{
|
||||
throw new IOException("Stream is too long.");
|
||||
}
|
||||
|
||||
return (int)rounded;
|
||||
}
|
||||
|
||||
private void ClearRange(int absoluteStart, int count)
|
||||
{
|
||||
if (count <= 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
switch (_mode)
|
||||
{
|
||||
case StorageMode.Contiguous:
|
||||
Array.Clear(_contiguousBuffer!, absoluteStart, count);
|
||||
break;
|
||||
case StorageMode.Segmented:
|
||||
{
|
||||
var position = absoluteStart;
|
||||
var remaining = count;
|
||||
while (remaining > 0)
|
||||
{
|
||||
var blockIndex = position / _blockSize;
|
||||
var blockOffset = position % _blockSize;
|
||||
var toClear = Math.Min(remaining, _blockSize - blockOffset);
|
||||
Array.Clear(_blocks![blockIndex], blockOffset, toClear);
|
||||
position += toClear;
|
||||
remaining -= toClear;
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void CopyFromSegmented(
|
||||
int absoluteSourcePosition,
|
||||
byte[] destination,
|
||||
int offset,
|
||||
int count
|
||||
)
|
||||
{
|
||||
var sourcePosition = absoluteSourcePosition;
|
||||
var destinationOffset = offset;
|
||||
var remaining = count;
|
||||
|
||||
while (remaining > 0)
|
||||
{
|
||||
var blockIndex = sourcePosition / _blockSize;
|
||||
var blockOffset = sourcePosition % _blockSize;
|
||||
var toCopy = Math.Min(remaining, _blockSize - blockOffset);
|
||||
Buffer.BlockCopy(
|
||||
_blocks![blockIndex],
|
||||
blockOffset,
|
||||
destination,
|
||||
destinationOffset,
|
||||
toCopy
|
||||
);
|
||||
|
||||
sourcePosition += toCopy;
|
||||
destinationOffset += toCopy;
|
||||
remaining -= toCopy;
|
||||
}
|
||||
}
|
||||
|
||||
private void CopyToSegmented(
|
||||
int absoluteDestinationPosition,
|
||||
byte[] source,
|
||||
int offset,
|
||||
int count
|
||||
)
|
||||
{
|
||||
var sourceOffset = offset;
|
||||
var destinationPosition = absoluteDestinationPosition;
|
||||
var remaining = count;
|
||||
|
||||
while (remaining > 0)
|
||||
{
|
||||
var blockIndex = destinationPosition / _blockSize;
|
||||
var blockOffset = destinationPosition % _blockSize;
|
||||
var toCopy = Math.Min(remaining, _blockSize - blockOffset);
|
||||
Buffer.BlockCopy(source, sourceOffset, _blocks![blockIndex], blockOffset, toCopy);
|
||||
|
||||
sourceOffset += toCopy;
|
||||
destinationPosition += toCopy;
|
||||
remaining -= toCopy;
|
||||
}
|
||||
}
|
||||
|
||||
private void ReturnSegmentedBlocks()
|
||||
{
|
||||
if (_blocks is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
for (var i = 0; i < _blocks.Count; i++)
|
||||
{
|
||||
_arrayPool.Return(_blocks[i]);
|
||||
}
|
||||
|
||||
_blocks.Clear();
|
||||
}
|
||||
|
||||
private void ReturnPooledBuffers()
|
||||
{
|
||||
if (_mode == StorageMode.Segmented)
|
||||
{
|
||||
ReturnSegmentedBlocks();
|
||||
_blocks = null;
|
||||
}
|
||||
|
||||
if (_mode == StorageMode.Contiguous && _contiguousBuffer is not null)
|
||||
{
|
||||
_arrayPool.Return(_contiguousBuffer);
|
||||
_contiguousBuffer = null;
|
||||
}
|
||||
|
||||
for (var i = 0; i < _detachedExposedBuffers.Count; i++)
|
||||
{
|
||||
_arrayPool.Return(_detachedExposedBuffers[i]);
|
||||
}
|
||||
|
||||
_detachedExposedBuffers.Clear();
|
||||
}
|
||||
|
||||
private static void ValidateReadWriteBufferArguments(byte[] buffer, int offset, int count)
|
||||
{
|
||||
ThrowHelper.ThrowIfNull(buffer, nameof(buffer));
|
||||
ThrowHelper.ThrowIfNegative(offset, nameof(offset));
|
||||
ThrowHelper.ThrowIfNegative(count, nameof(count));
|
||||
if (buffer.Length - offset < count)
|
||||
{
|
||||
throw new ArgumentException("Offset and length are out of bounds.");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -60,6 +60,15 @@ internal static class ThrowHelper
|
||||
}
|
||||
}
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public static void ThrowIfGreaterThan(long value, long other, string? paramName = null)
|
||||
{
|
||||
if (value > other)
|
||||
{
|
||||
throw new ArgumentOutOfRangeException(paramName);
|
||||
}
|
||||
}
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public static void ThrowIfGreaterThan(uint value, uint other, string? paramName = null)
|
||||
{
|
||||
|
||||
149
tests/SharpCompress.Test/Streams/PooledMemoryStreamTests.cs
Normal file
149
tests/SharpCompress.Test/Streams/PooledMemoryStreamTests.cs
Normal file
@@ -0,0 +1,149 @@
|
||||
using System;
|
||||
using System.Buffers;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using SharpCompress.IO;
|
||||
using Xunit;
|
||||
|
||||
namespace SharpCompress.Test.Streams;
|
||||
|
||||
public class PooledMemoryStreamTests
|
||||
{
|
||||
[Fact]
|
||||
public void GrowsUsingFixedSizeBlocks()
|
||||
{
|
||||
var pool = new TrackingArrayPool();
|
||||
|
||||
using var stream = new PooledMemoryStream(capacity: 0, blockSize: 8, arrayPool: pool);
|
||||
stream.Write(new byte[20], 0, 20);
|
||||
|
||||
Assert.Equal(3, pool.RentRequests.Count);
|
||||
Assert.All(pool.RentRequests, requested => Assert.Equal(8, requested));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DisposeReturnsRentedBlocksToPool()
|
||||
{
|
||||
var pool = new TrackingArrayPool();
|
||||
var stream = new PooledMemoryStream(capacity: 0, blockSize: 8, arrayPool: pool);
|
||||
|
||||
stream.Write(new byte[17], 0, 17);
|
||||
stream.Dispose();
|
||||
|
||||
Assert.Equal(pool.RentRequests.Count, pool.ReturnedLengths.Count);
|
||||
Assert.All(pool.ReturnedLengths, length => Assert.Equal(8, length));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void GetBufferPromotesToContiguousAndLaterGrowthUsesBlocks()
|
||||
{
|
||||
var pool = new TrackingArrayPool();
|
||||
|
||||
using var stream = new PooledMemoryStream(capacity: 0, blockSize: 8, arrayPool: pool);
|
||||
stream.Write(new byte[10], 0, 10);
|
||||
|
||||
var contiguous = stream.GetBuffer();
|
||||
Assert.Equal(16, contiguous.Length);
|
||||
Assert.Contains(16, pool.RentRequests);
|
||||
|
||||
stream.Position = stream.Length;
|
||||
stream.Write(new byte[10], 0, 10); // grow to 20, forcing demotion + block growth
|
||||
|
||||
var totalBlockRents = pool.RentRequests.Count(size => size == 8);
|
||||
Assert.True(totalBlockRents >= 3);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BufferConstructorCopiesDataAndIsNonExpandable()
|
||||
{
|
||||
var backing = Enumerable.Range(0, 10).Select(i => (byte)i).ToArray();
|
||||
using var stream = new PooledMemoryStream(
|
||||
backing,
|
||||
2,
|
||||
4,
|
||||
writable: true,
|
||||
publiclyVisible: true
|
||||
);
|
||||
|
||||
Assert.Equal(4, stream.Length);
|
||||
Assert.Equal(4, stream.Capacity);
|
||||
|
||||
backing[2] = 255;
|
||||
Assert.Equal(new byte[] { 2, 3, 4, 5 }, stream.ToArray());
|
||||
|
||||
Assert.Throws<NotSupportedException>(() => stream.Capacity = 5);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BufferConstructorTryGetBufferRespectsVisibilityFlag()
|
||||
{
|
||||
var backing = new byte[10];
|
||||
using var hidden = new PooledMemoryStream(
|
||||
backing,
|
||||
1,
|
||||
5,
|
||||
writable: true,
|
||||
publiclyVisible: false
|
||||
);
|
||||
Assert.False(hidden.TryGetBuffer(out _));
|
||||
|
||||
using var visible = new PooledMemoryStream(
|
||||
backing,
|
||||
1,
|
||||
5,
|
||||
writable: true,
|
||||
publiclyVisible: true
|
||||
);
|
||||
Assert.True(visible.TryGetBuffer(out var segment));
|
||||
Assert.Equal(0, segment.Offset);
|
||||
Assert.Equal(5, segment.Count);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SetLengthExtendingClearsGap()
|
||||
{
|
||||
using var stream = new PooledMemoryStream(capacity: 0, blockSize: 8);
|
||||
stream.Position = 5;
|
||||
stream.WriteByte(42);
|
||||
stream.Position = 0;
|
||||
|
||||
var data = stream.ToArray();
|
||||
Assert.Equal(6, data.Length);
|
||||
Assert.Equal(0, data[0]);
|
||||
Assert.Equal(0, data[4]);
|
||||
Assert.Equal(42, data[5]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void MethodsThrowAfterDispose()
|
||||
{
|
||||
using var stream = new PooledMemoryStream(capacity: 0, blockSize: 8);
|
||||
stream.WriteByte(1);
|
||||
stream.Dispose();
|
||||
|
||||
Assert.Throws<ObjectDisposedException>(() => stream.ReadByte());
|
||||
Assert.Throws<ObjectDisposedException>(() => stream.ToArray());
|
||||
Assert.Throws<ObjectDisposedException>(() => stream.GetBuffer());
|
||||
}
|
||||
|
||||
private sealed class TrackingArrayPool : ArrayPool<byte>
|
||||
{
|
||||
public readonly System.Collections.Generic.List<int> RentRequests = new();
|
||||
public readonly System.Collections.Generic.List<int> ReturnedLengths = new();
|
||||
|
||||
public override byte[] Rent(int minimumLength)
|
||||
{
|
||||
RentRequests.Add(minimumLength);
|
||||
return new byte[minimumLength];
|
||||
}
|
||||
|
||||
public override void Return(byte[] array, bool clearArray = false)
|
||||
{
|
||||
ReturnedLengths.Add(array.Length);
|
||||
if (clearArray)
|
||||
{
|
||||
Array.Clear(array, 0, array.Length);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user