1
1

LargeMemoryStream.cs 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274
  1. using System;
  2. using System.Buffers;
  3. using System.IO;
  4. using System.Runtime.CompilerServices;
  5. namespace Masuit.Tools.Systems;
  6. /// <summary>
  7. /// 大型内存流,最大可支持1TB数据,推荐当数据流大于2GB时使用
  8. /// </summary>
  9. public class LargeMemoryStream : Stream
  10. {
  11. /// <summary>
  12. /// 终结器
  13. /// </summary>
  14. ~LargeMemoryStream()
  15. {
  16. Dispose(true);
  17. }
  18. private const int PageSize = 1024000000;
  19. private const int AllocStep = 1024;
  20. private byte[][] _streamBuffers;
  21. private int _pageCount;
  22. private long _allocatedBytes;
  23. private long _position;
  24. private long _length;
  25. private bool _isDisposed;
  26. private int GetPageCount(long length)
  27. {
  28. int pageCount = (int)(length / PageSize) + 1;
  29. if (length % PageSize == 0)
  30. {
  31. pageCount--;
  32. }
  33. return pageCount;
  34. }
  35. private void ExtendPages()
  36. {
  37. if (_streamBuffers == null)
  38. {
  39. _streamBuffers = new byte[AllocStep][];
  40. }
  41. else
  42. {
  43. var streamBuffers = new byte[_streamBuffers.Length + AllocStep][];
  44. Buffer.BlockCopy(_streamBuffers, 0, streamBuffers, 0, _streamBuffers.Length);
  45. _streamBuffers = streamBuffers;
  46. }
  47. _pageCount = _streamBuffers.Length;
  48. }
  49. private void AllocSpaceIfNeeded(long value)
  50. {
  51. switch (value)
  52. {
  53. case < 0:
  54. throw new InvalidOperationException("AllocSpaceIfNeeded < 0");
  55. case 0:
  56. return;
  57. }
  58. int currentPageCount = GetPageCount(_allocatedBytes);
  59. int neededPageCount = GetPageCount(value);
  60. while (currentPageCount < neededPageCount)
  61. {
  62. if (currentPageCount == _pageCount)
  63. ExtendPages();
  64. _streamBuffers[currentPageCount++] = ArrayPool<byte>.Shared.Rent(PageSize);
  65. }
  66. _allocatedBytes = (long)currentPageCount * PageSize;
  67. value = Math.Max(value, _length);
  68. if (_position > (_length = value))
  69. {
  70. _position = _length;
  71. }
  72. }
  73. public override bool CanRead => true;
  74. public override bool CanSeek => true;
  75. public override bool CanWrite => true;
  76. public override long Length => _length;
  77. public override long Position
  78. {
  79. get => _position;
  80. set
  81. {
  82. if (value > _length)
  83. {
  84. throw new InvalidOperationException("Position > Length");
  85. }
  86. if (value < 0)
  87. {
  88. throw new InvalidOperationException("Position < 0");
  89. }
  90. _position = value;
  91. }
  92. }
  93. #if NETCOREAPP || NET452
  94. public Span<byte[]> GetSpan()
  95. {
  96. return _streamBuffers.AsSpan(0, _streamBuffers.Length);
  97. }
  98. public Memory<byte[]> GetMemory()
  99. {
  100. return _streamBuffers.AsMemory(0, _streamBuffers.Length);
  101. }
  102. public ArraySegment<byte[]> ToArraySegment()
  103. {
  104. return new ArraySegment<byte[]>(_streamBuffers, 0, _streamBuffers.Length);
  105. }
  106. #endif
  107. public override void Flush()
  108. {
  109. AssertNotDisposed();
  110. }
  111. public override int Read(byte[] buffer, int offset, int count)
  112. {
  113. AssertNotDisposed();
  114. int currentPage = (int)(_position / PageSize);
  115. int currentOffset = (int)(_position % PageSize);
  116. int currentLength = PageSize - currentOffset;
  117. long startPosition = _position;
  118. if (startPosition + count > _length)
  119. {
  120. count = (int)(_length - startPosition);
  121. }
  122. while (count != 0 && _position < _length)
  123. {
  124. if (currentLength > count)
  125. {
  126. currentLength = count;
  127. }
  128. Buffer.BlockCopy(_streamBuffers[currentPage++], currentOffset, buffer, offset, currentLength);
  129. offset += currentLength;
  130. _position += currentLength;
  131. count -= currentLength;
  132. currentOffset = 0;
  133. currentLength = PageSize;
  134. }
  135. return (int)(_position - startPosition);
  136. }
  137. public override long Seek(long offset, SeekOrigin origin)
  138. {
  139. AssertNotDisposed();
  140. switch (origin)
  141. {
  142. case SeekOrigin.Begin:
  143. break;
  144. case SeekOrigin.Current:
  145. offset += _position;
  146. break;
  147. case SeekOrigin.End:
  148. offset = _length - offset;
  149. break;
  150. default:
  151. throw new ArgumentOutOfRangeException("origin");
  152. }
  153. return Position = offset;
  154. }
  155. public override void SetLength(long value)
  156. {
  157. switch (value)
  158. {
  159. case < 0:
  160. throw new InvalidOperationException("SetLength < 0");
  161. case 0:
  162. _streamBuffers = null;
  163. _allocatedBytes = _position = _length = 0;
  164. _pageCount = 0;
  165. return;
  166. }
  167. int currentPageCount = GetPageCount(_allocatedBytes);
  168. int neededPageCount = GetPageCount(value);
  169. while (currentPageCount > neededPageCount)
  170. {
  171. ArrayPool<byte>.Shared.Return(_streamBuffers[--currentPageCount], true);
  172. _streamBuffers[currentPageCount] = null;
  173. }
  174. AllocSpaceIfNeeded(value);
  175. if (_position > (_length = value))
  176. {
  177. _position = _length;
  178. }
  179. }
  180. public override void Write(byte[] buffer, int offset, int count)
  181. {
  182. AssertNotDisposed();
  183. int currentPage = (int)(_position / PageSize);
  184. int currentOffset = (int)(_position % PageSize);
  185. int currentLength = PageSize - currentOffset;
  186. AllocSpaceIfNeeded(_position + count);
  187. while (count != 0)
  188. {
  189. if (currentLength > count)
  190. {
  191. currentLength = count;
  192. }
  193. Buffer.BlockCopy(buffer, offset, _streamBuffers[currentPage++], currentOffset, currentLength);
  194. offset += currentLength;
  195. _position += currentLength;
  196. count -= currentLength;
  197. currentOffset = 0;
  198. currentLength = PageSize;
  199. }
  200. }
  201. [MethodImpl(MethodImplOptions.AggressiveInlining)]
  202. protected override void Dispose(bool disposing)
  203. {
  204. if (disposing)
  205. {
  206. _isDisposed = true;
  207. Position = 0;
  208. _length = 0;
  209. if (_streamBuffers != null)
  210. {
  211. foreach (var bytes in _streamBuffers)
  212. {
  213. if (bytes != null)
  214. {
  215. ArrayPool<byte>.Shared.Return(bytes);
  216. }
  217. }
  218. _streamBuffers = null;
  219. }
  220. }
  221. base.Dispose(disposing);
  222. }
  223. [MethodImpl(MethodImplOptions.AggressiveInlining)]
  224. private void AssertNotDisposed()
  225. {
  226. if (_isDisposed)
  227. {
  228. throw new ObjectDisposedException(nameof(PooledMemoryStream));
  229. }
  230. }
  231. }