Optimize memory usage

README.md

@@ -16,6 +16,43 @@ Currently, the supported buffer type is only memory buffers, but more types of b
 
 Scenarios that require concurrent batch processing of data when the speed between producers and consumers is inconsistent.
 
+## Comparison with Common In-Memory Queues
+
+**BufferQueue's core advantage is high performance in batch consumption scenarios.**
+
+The project includes BenchmarkDotNet benchmarks that compare `MemoryBufferQueue<T>` in BufferQueue's Memory mode with `Channel<T>` and `BlockingCollection<T>` for concurrent producing and consuming. The table below summarizes the `Channel<T>` comparison results.
+
+Summary:
+
+- Producing: `Channel<T>` is faster; `MemoryBufferQueue<T>` allocates less memory in Bounded producing, while Unbounded producing is not its advantage.
+- Consuming: `MemoryBufferQueue<T>` is mainly optimized for batch consumption. In this benchmark set, larger batches usually show a clearer advantage, up to about `84x` under the current parameters.
+- Memory allocation: `MemoryBufferQueue<T>` allocates less in producing scenarios; `Channel<T>` allocates less in consuming scenarios.
+
+Representative results:
+
+| Type | Scenario | Parameters | `Channel<T>` | `MemoryBufferQueue<T>` | Result |
+| --- | --- | --- | ---: | ---: | --- |
+| Producing | Unbounded | `MessageSize = 8192` | `336.8 μs` | `689.2 μs` | `Channel<T>` is faster |
+| Producing | Bounded | `MessageSize = 8192` | `360.2 μs` | `8,402.0 μs` | `Channel<T>` is faster |
+| Consuming | Unbounded | `MessageSize = 8192`, `BatchSize = 1000` | `3,461.03 μs` | `41.30 μs` | About `84x` faster under current parameters |
+| Consuming | Bounded | `MessageSize = 8192`, `BatchSize = 1000` | `2,214.21 μs` | `41.68 μs` | About `53x` faster under current parameters |
+
+Test platform:
+
+| Item | Value |
+| --- | --- |
+| OS | macOS `15.7.7` (`24G720`) |
+| RID | `osx-arm64` |
+| .NET SDK | `10.0.100` |
+| .NET Host | `10.0.0`, `arm64` |
+| Benchmark target | `net8.0` |
+
+Run the full benchmark with:
+
+```shell
+dotnet run -c Release --project tests/BufferQueue.Benchmarks/BufferQueue.Benchmarks.csproj
+```
+
 ## Functional Design
 
 1. Supports creating multiple topics, each of which can have multiple data types. Each pair of topics and data types corresponds to an independent buffer.

README.zh-CN.md

@@ -15,6 +15,43 @@ BufferQueue 是一个用 .NET 编写的高性能的缓冲队列实现，支持
 ## 适用场景
 生产者和消费者之间的速度不一致，需要并发批量处理数据的场景。
 
+## 和其他常用内存缓存 Queue 的对比
+
+**BufferQueue 的核心优势在于批量消费时的高性能表现。**
+
+项目内置 BenchmarkDotNet 基准测试，对比 `BufferQueue` 启用 Memory 模式后的 `MemoryBufferQueue<T>`、`Channel<T>` 和 `BlockingCollection<T>` 的并发生产、并发消费表现。下表展示 `Channel<T>` 对比结果摘要。
+
+结果摘要：
+
+- 生产：`Channel<T>` 更快；`MemoryBufferQueue<T>` Bounded 生产内存分配更少，Unbounded 生产不是优势场景。
+- 消费：`MemoryBufferQueue<T>` 的主要优势在批量消费；在本组测试中，批量越大优势通常越明显，当前参数下最高约快 `84x`。
+- 内存分配：生产场景 `MemoryBufferQueue<T>` 分配较少；消费场景 `Channel<T>` 分配较少。
+
+代表性数据：
+
+| 类型 | 场景 | 参数 | `Channel<T>` | `MemoryBufferQueue<T>` | 结论 |
+| --- | --- | --- | ---: | ---: | --- |
+| 生产 | Unbounded | `MessageSize = 8192` | `336.8 μs` | `689.2 μs` | `Channel<T>` 更快 |
+| 生产 | Bounded | `MessageSize = 8192` | `360.2 μs` | `8,402.0 μs` | `Channel<T>` 更快 |
+| 消费 | Unbounded | `MessageSize = 8192`, `BatchSize = 1000` | `3,461.03 μs` | `41.30 μs` | 当前参数下约快 `84x` |
+| 消费 | Bounded | `MessageSize = 8192`, `BatchSize = 1000` | `2,214.21 μs` | `41.68 μs` | 当前参数下约快 `53x` |
+
+测试平台：
+
+| 项目 | 信息 |
+| --- | --- |
+| 操作系统 | macOS `15.7.7` (`24G720`) |
+| 运行时标识 | `osx-arm64` |
+| .NET SDK | `10.0.100` |
+| .NET Host | `10.0.0`, `arm64` |
+| Benchmark 运行目标 | `net8.0` |
+
+可以通过以下命令运行完整 benchmark：
+
+```shell
+dotnet run -c Release --project tests/BufferQueue.Benchmarks/BufferQueue.Benchmarks.csproj
+```
+
 ## 功能设计：
 1. 支持创建多个 Topic，每个 Topic 可以有多种数据类型。每一对 Topic 和数据类型对应一个独立的缓冲区。
 
@@ -241,4 +278,3 @@ public class TestController(IBufferQueue bufferQueue) : ControllerBase
     }
 }
 ```
-

src/BufferQueue/Memory/MemoryBufferPartition.cs

@@ -2,11 +2,11 @@
 // The .NET Core Community licenses this file to you under the MIT license.
 
 using System;
+using System.Collections;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
-using System.Linq;
 
 namespace BufferQueue.Memory;
 
@@ -176,8 +176,10 @@ public bool TryRead(int batchSize, [NotNullWhen(true)] out IEnumerable<T>? items
         {
             var remainingCount = batchSize;
             var readPosition = _readPosition;
-            var result = Enumerable.Empty<T>();
             var currentSegment = _currentSegment;
+            T[]? result = null;
+            T singleItem = default!;
+            var copiedCount = 0;
 
             while (true)
             {
@@ -194,10 +196,21 @@ public bool TryRead(int batchSize, [NotNullWhen(true)] out IEnumerable<T>? items
                 var retrievalSuccess = currentSegment.TryGet(readPosition, remainingCount, out var segmentItems);
                 if (retrievalSuccess)
                 {
-                    var length = segmentItems!.Length;
+                    var length = segmentItems.Count;
                     readPosition += (ulong)length;
                     remainingCount -= length;
-                    result = result.Concat(segmentItems);
+
+                    if (batchSize == 1)
+                    {
+                        singleItem = segmentItems.Array![segmentItems.Offset];
+                        copiedCount = 1;
+                    }
+                    else
+                    {
+                        result ??= new T[batchSize];
+                        Array.Copy(segmentItems.Array!, segmentItems.Offset, result, copiedCount, length);
+                        copiedCount += length;
+                    }
                 }
                 else
                 {
@@ -228,7 +241,15 @@ public bool TryRead(int batchSize, [NotNullWhen(true)] out IEnumerable<T>? items
             }
 
             _lastReadCount = batchSize - remainingCount;
-            items = result;
+            if (batchSize == 1)
+            {
+                items = new SingleItemBatch<T>(singleItem);
+                return true;
+            }
+
+            items = _lastReadCount == result!.Length
+                ? result
+                : new SnapshotBatch<T>(result, _lastReadCount);
             return true;
         }
 
@@ -242,6 +263,74 @@ public void MoveNext()
         }
     }
 
+    private sealed class SingleItemBatch<TItem>(TItem item) : IReadOnlyList<TItem>, ICollection<TItem>
+    {
+        public int Count => 1;
+
+        public bool IsReadOnly => true;
+
+        public TItem this[int index] => index == 0
+            ? item
+            : throw new ArgumentOutOfRangeException(nameof(index));
+
+        public IEnumerator<TItem> GetEnumerator()
+        {
+            yield return item;
+        }
+
+        IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+
+        public bool Contains(TItem itemToFind) => EqualityComparer<TItem>.Default.Equals(item, itemToFind);
+
+        public void CopyTo(TItem[] array, int arrayIndex) => array[arrayIndex] = item;
+
+        public void Add(TItem itemToAdd) => throw new NotSupportedException();
+
+        public void Clear() => throw new NotSupportedException();
+
+        public bool Remove(TItem itemToRemove) => throw new NotSupportedException();
+    }
+
+    private sealed class SnapshotBatch<TItem>(TItem[] items, int count) : IReadOnlyList<TItem>, ICollection<TItem>
+    {
+        public int Count => count;
+
+        public bool IsReadOnly => true;
+
+        public TItem this[int index]
+        {
+            get
+            {
+                if ((uint)index >= (uint)count)
+                {
+                    throw new ArgumentOutOfRangeException(nameof(index));
+                }
+
+                return items[index];
+            }
+        }
+
+        public IEnumerator<TItem> GetEnumerator()
+        {
+            for (var i = 0; i < count; i++)
+            {
+                yield return items[i];
+            }
+        }
+
+        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() => GetEnumerator();
+
+        public bool Contains(TItem item) => Array.IndexOf(items, item, 0, count) >= 0;
+
+        public void CopyTo(TItem[] array, int arrayIndex) => Array.Copy(items, 0, array, arrayIndex, count);
+
+        public void Add(TItem item) => throw new NotSupportedException();
+
+        public void Clear() => throw new NotSupportedException();
+
+        public bool Remove(TItem item) => throw new NotSupportedException();
+    }
+
     private class DebugView(MemoryBufferPartition<T> partition)
     {
         public int PartitionId => partition.PartitionId;

src/BufferQueue/Memory/MemoryBufferSegment.cs

@@ -3,7 +3,6 @@
 
 using System;
 using System.Diagnostics;
-using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using System.Threading;
 
@@ -99,33 +98,26 @@ public bool TryEnqueue(T item)
         }
     }
 
-    public bool TryGet(MemoryBufferPartitionOffset offset, int count, [NotNullWhen(true)] out T[]? items)
+    public bool TryGet(MemoryBufferPartitionOffset offset, int count, out ArraySegment<T> items)
     {
         if (offset < _startOffset || offset > _endOffset)
         {
-            items = null;
+            items = default;
             return false;
         }
 
         var readPosition = (offset - _startOffset).ToInt32();
 
         if (_publishedWritePosition < 0 || readPosition > _publishedWritePosition)
         {
-            items = null;
+            items = default;
             return false;
         }
 
         var writePosition = Math.Min(_publishedWritePosition, _slots.Length - 1);
         // Number of items actually available to return (bounded by requested count and written items).
         var availableCount = Math.Min(count, writePosition - readPosition + 1);
-        var wholeSegment = readPosition == 0 && availableCount == _slots.Length;
-        if (wholeSegment)
-        {
-            items = _slots;
-            return true;
-        }
-
-        items = _slots[readPosition..(readPosition + availableCount)];
+        items = new ArraySegment<T>(_slots, readPosition, availableCount);
         return true;
     }
 

tests/BufferQueue.Benchmarks/MemoryBufferQueueBoundedChannelConsumeBenchmark.cs

@@ -0,0 +1,102 @@
+// Licensed to the .NET Core Community under one or more agreements.
+// The .NET Core Community licenses this file to you under the MIT license.
+
+using System.Threading.Channels;
+using BenchmarkDotNet.Attributes;
+using BufferQueue.Memory;
+
+namespace BufferQueue.Benchmarks;
+
+public class MemoryBufferQueueBoundedChannelConsumeBenchmark
+{
+    private Channel<int> _boundedChannel = default!;
+    private IEnumerable<IBufferPullConsumer<int>> _boundedMemoryBufferQueueConsumers = default!;
+
+    [Params(4096, 8192)] public int MessageSize { get; set; }
+
+    [Params(1, 10, 100, 1000)] public int BatchSize { get; set; }
+
+    [IterationSetup]
+    public void Setup()
+    {
+        _boundedChannel = Channel.CreateBounded<int>(new BoundedChannelOptions(MessageSize)
+        {
+            SingleReader = false,
+            SingleWriter = false
+        });
+
+        var boundedMemoryBufferQueue = new MemoryBufferQueue<int>(new MemoryBufferQueueOptions
+        {
+            TopicName = "test-bounded",
+            PartitionNumber = Environment.ProcessorCount,
+            BoundedCapacity = (ulong)MessageSize
+        });
+        var boundedMemoryBufferQueueProducer = boundedMemoryBufferQueue.GetProducer();
+
+        for (var i = 0; i < MessageSize; i++)
+        {
+            _boundedChannel.Writer.TryWrite(i);
+            boundedMemoryBufferQueueProducer.ProduceAsync(i);
+        }
+
+        _boundedMemoryBufferQueueConsumers = boundedMemoryBufferQueue.CreateConsumers(
+            new BufferPullConsumerOptions
+            {
+                GroupName = "TestGroup",
+                TopicName = "test-bounded",
+                AutoCommit = true,
+                BatchSize = BatchSize,
+            },
+            Environment.ProcessorCount);
+    }
+
+    [Benchmark(Baseline = true)]
+    public void Channel_BoundedConcurrentConsume()
+    {
+        ConsumeChannel(_boundedChannel);
+    }
+
+    [Benchmark]
+    public void MemoryBufferQueue_BoundedConcurrentConsume()
+    {
+        ConsumeMemoryBufferQueue(_boundedMemoryBufferQueueConsumers);
+    }
+
+    private void ConsumeChannel(Channel<int> channel)
+    {
+        var remaining = MessageSize;
+        var tasks = Enumerable.Range(0, Environment.ProcessorCount).Select(_ => Task.Run(() =>
+        {
+            var reader = channel.Reader;
+            while (Volatile.Read(ref remaining) > 0 && reader.TryRead(out _))
+            {
+                Interlocked.Decrement(ref remaining);
+            }
+        })).ToArray();
+
+        Task.WaitAll(tasks);
+    }
+
+    private void ConsumeMemoryBufferQueue(IEnumerable<IBufferPullConsumer<int>> consumers)
+    {
+        var consumerList = consumers.ToList();
+        var partitionCount = consumerList.Count;
+        var baseMessageCount = MessageSize / partitionCount;
+        var remainder = MessageSize % partitionCount;
+        var tasks = consumerList.Select((consumer, index) => Task.Run(async () =>
+        {
+            var expectedCount = baseMessageCount + (index < remainder ? 1 : 0);
+            var consumedCount = 0;
+            await foreach (var items in consumer.ConsumeAsync())
+            {
+                consumedCount += items.Count();
+                if (consumedCount >= expectedCount)
+                {
+                    break;
+                }
+            }
+        })).ToArray();
+
+        Task.WaitAll(tasks);
+    }
+}