Bộ đệm FIFO/Hàng đợi chuyên về các dòng byte

Question

Có cấu trúc dữ liệu .NET/kết hợp các lớp cho phép dữ liệu byte được nối vào cuối bộ đệm hay không, tất cả các lần xem và đọc đều bắt đầu, rút ​​ngắn bộ đệm khi Tôi đọc?

Lớp MemoryStream có vẻ như là một phần của điều này, nhưng tôi cần duy trì các vị trí riêng biệt để đọc và viết và không tự động hủy dữ liệu ngay khi bắt đầu đọc.

Câu trả lời đã được đăng để trả lời this question về cơ bản những gì tôi đang cố gắng làm nhưng tôi thích điều gì đó tôi có thể làm I/O không đồng bộ trên các thành phần khác nhau của cùng một quy trình, giống như một ống thông thường hoặc thậm chí là luồng mạng (tôi cần lọc/xử lý dữ liệu trước).

Answer 1

Tôi sẽ đăng bản sao bị tước bỏ một số logic mà tôi đã viết cho một dự án tại nơi làm việc một lần. Ưu điểm của phiên bản này là nó hoạt động với một danh sách liên kết dữ liệu đệm và do đó bạn không phải cache một lượng lớn bộ nhớ và/hoặc sao chép bộ nhớ xung quanh khi đọc. hơn nữa, luồng của nó an toàn và hoạt động giống như một luồng mạng, đó là: Khi đọc khi không có sẵn dữ liệu: Chờ cho đến khi có sẵn dữ liệu hoặc hết thời gian chờ. Ngoài ra, khi đọc x số byte và chỉ có một lượng byte là y, trả về sau khi đọc tất cả các byte. Tôi hi vọng cái này giúp được!

public class SlidingStream : Stream 
{ 
    #region Other stream member implementations 

    ... 

    #endregion Other stream member implementations 

    public SlidingStream() 
    { 
     ReadTimeout = -1; 
    } 

    private readonly object _writeSyncRoot = new object(); 
    private readonly object _readSyncRoot = new object(); 
    private readonly LinkedList<ArraySegment<byte>> _pendingSegments = new LinkedList<ArraySegment<byte>>(); 
    private readonly ManualResetEventSlim _dataAvailableResetEvent = new ManualResetEventSlim(); 

    public int ReadTimeout { get; set; } 

    public override int Read(byte[] buffer, int offset, int count) 
    { 
     if (_dataAvailableResetEvent.Wait(ReadTimeout)) 
      throw new TimeoutException("No data available"); 

     lock (_readSyncRoot) 
     { 
      int currentCount = 0; 
      int currentOffset = 0; 

      while (currentCount != count) 
      { 
       ArraySegment<byte> segment = _pendingSegments.First.Value; 
       _pendingSegments.RemoveFirst(); 

       int index = segment.Offset; 
       for (; index < segment.Count; index++) 
       { 
        if (currentOffset < offset) 
        { 
         currentOffset++; 
        } 
        else 
        { 
         buffer[currentCount] = segment.Array[index]; 
         currentCount++; 
        } 
       } 

       if (currentCount == count) 
       { 
        if (index < segment.Offset + segment.Count) 
        { 
         _pendingSegments.AddFirst(new ArraySegment<byte>(segment.Array, index, segment.Offset + segment.Count - index)); 
        } 
       } 

       if (_pendingSegments.Count == 0) 
       { 
        _dataAvailableResetEvent.Reset(); 

        return currentCount; 
       } 
      } 

      return currentCount; 
     } 
    } 

    public override void Write(byte[] buffer, int offset, int count) 
    { 
     lock (_writeSyncRoot) 
     { 
      byte[] copy = new byte[count]; 
      Array.Copy(buffer, offset, copy, 0, count); 

      _pendingSegments.AddLast(new ArraySegment<byte>(copy)); 

      _dataAvailableResetEvent.Set(); 
     } 
    } 
} 

Answer 2

Mã có thể đơn giản hơn trong câu trả lời được chấp nhận. Không cần sử dụng vòng lặp for .:

/// <summary> 
/// This class is a very fast and threadsafe FIFO buffer 
/// </summary> 
public class FastFifo 
{ 
    private List<Byte> mi_FifoData = new List<Byte>(); 

    /// <summary> 
    /// Get the count of bytes in the Fifo buffer 
    /// </summary> 
    public int Count 
    { 
     get 
     { 
      lock (mi_FifoData) 
      { 
       return mi_FifoData.Count; 
      } 
     } 
    } 

    /// <summary> 
    /// Clears the Fifo buffer 
    /// </summary> 
    public void Clear() 
    { 
     lock (mi_FifoData) 
     { 
      mi_FifoData.Clear(); 
     } 
    } 

    /// <summary> 
    /// Append data to the end of the fifo 
    /// </summary> 
    public void Push(Byte[] u8_Data) 
    { 
     lock (mi_FifoData) 
     { 
      // Internally the .NET framework uses Array.Copy() which is extremely fast 
      mi_FifoData.AddRange(u8_Data); 
     } 
    } 

    /// <summary> 
    /// Get data from the beginning of the fifo. 
    /// returns null if s32_Count bytes are not yet available. 
    /// </summary> 
    public Byte[] Pop(int s32_Count) 
    { 
     lock (mi_FifoData) 
     { 
      if (mi_FifoData.Count < s32_Count) 
       return null; 

      // Internally the .NET framework uses Array.Copy() which is extremely fast 
      Byte[] u8_PopData = new Byte[s32_Count]; 
      mi_FifoData.CopyTo(0, u8_PopData, 0, s32_Count); 
      mi_FifoData.RemoveRange(0, s32_Count); 
      return u8_PopData; 
     } 
    } 

    /// <summary> 
    /// Gets a byte without removing it from the Fifo buffer 
    /// returns -1 if the index is invalid 
    /// </summary> 
    public int PeekAt(int s32_Index) 
    { 
     lock (mi_FifoData) 
     { 
      if (s32_Index < 0 || s32_Index >= mi_FifoData.Count) 
       return -1; 

      return mi_FifoData[s32_Index]; 
     } 
    } 
}