l4 evict and swiss api improve

This commit is contained in:
Youmu Liu
2026-10-20 18:47:31 +08:00
parent e349bcc4dd
commit 3951dd6686
7 changed files with 140 additions and 40 deletions
@@ -71,7 +71,7 @@ namespace Nitload.Common.HighPerformance.Memory
#region buffer #region buffer
[FieldOffset(8)] public ulong Position; [FieldOffset(8)] public nuint Position;
[FieldOffset(16)] public uint SizeMask; [FieldOffset(16)] public uint SizeMask;
#endregion #endregion
@@ -16,6 +16,10 @@ namespace Nitload.Common.HighPerformance.Memory
public unsafe partial class L4 public unsafe partial class L4
{ {
private int _active_line_count;
private int _last_evict_pos;
internal partial int Predict() internal partial int Predict()
{ {
if (!TryGetCurrentPredictor(out IL4Predictor? p)) { if (!TryGetCurrentPredictor(out IL4Predictor? p)) {
@@ -23,12 +27,13 @@ namespace Nitload.Common.HighPerformance.Memory
} }
int total_count = 0, count; int total_count = 0, count;
Span<L4.AddressSource> as_sp = stackalloc L4.AddressSource[4]; Span<L4.AddressSource> as_sp = stackalloc L4.AddressSource[4];
SwissEnumerator e = _addressMap.GetWriteBufferEnumerator();
Predict: Predict:
while ((count = p.PredictMission(in as_sp)) > 0) while ((count = p.PredictMission(in as_sp)) > 0)
{ {
for (int i = 0; i < count; i++) for (int i = 0; i < count; i++)
{ {
if (_addressMap.TryGet(as_sp[i], out ulong map_handle_adr)) if (_addressMap.TryGetWrite(as_sp[i], out ulong map_handle_adr))
{ {
L4.CacheLineMetadata* map_handle = (L4.CacheLineMetadata*)map_handle_adr; L4.CacheLineMetadata* map_handle = (L4.CacheLineMetadata*)map_handle_adr;
map_handle->IncreasePredictCount(); map_handle->IncreasePredictCount();
@@ -38,13 +43,31 @@ namespace Nitload.Common.HighPerformance.Memory
uint size = as_sp[i].Size; uint size = as_sp[i].Size;
int handle_idx = _bufferAllocator.Alloc(size); int handle_idx = _bufferAllocator.Alloc(size);
L4.CacheLineMetadata* handle = _bufferAllocator.MetadataAt(handle_idx); L4.CacheLineMetadata* handle = _bufferAllocator.MetadataAt(handle_idx);
p.Prefetch(as_sp[i], handle->Position, handle->SizeMask);
if (!_addressMap.TryAddOrSet(as_sp[i], (ulong)handle)) { if (!_addressMap.TryAddOrSet(as_sp[i], (ulong)handle)) {
throw new InvalidOperationException(); throw new InvalidOperationException();
} }
_active_line_count++;
if (_active_line_count >= MaxActiveCacheLineCount)
{
int retry = 0;
const int max_retry = 3;
for (; retry < max_retry && EvictOnce(ref e); retry++) {
e = _addressMap.GetWriteBufferEnumerator(_last_evict_pos);
}
if (retry >= max_retry) {
throw new OverflowException("Evict failed, maybe all cache lines are in use.");
}
_active_line_count--;
}
} }
} }
total_count += count; total_count += count;
if (total_count >= 2048) { if (total_count >= 2048)
{
_ = Interlocked.Add(ref _predict_count, total_count);
_last_evict_pos = e.CurrentOffset;
return total_count; return total_count;
} }
} }
@@ -53,22 +76,36 @@ namespace Nitload.Common.HighPerformance.Memory
_ = TryGetCurrentPredictor(out p); _ = TryGetCurrentPredictor(out p);
goto Predict; goto Predict;
} }
_ = Interlocked.Add(ref _predict_count, total_count);
_last_evict_pos = e.CurrentOffset;
return total_count; return total_count;
} }
private bool EvictOnce(
ref SwissEnumerator e
)
{
while (e.MoveNext())
{
(AddressSource source, ulong cache_head_ptr) = e.Current;
L4.CacheLineMetadata* handle = (L4.CacheLineMetadata*)cache_head_ptr;
if (Volatile.Read(ref handle->ConsumeCount) == Volatile.Read(ref handle->ConsumeCount))
{
// evict
_ = _addressMap.TryRemove(source, out _);
_bufferAllocator.Free(handle);
return true;
}
}
return false;
}
} }
} }
/* // Eviction follow-up:
goto 本身没问题,但当前控制流有几个需要先确认的点: // 1. Compare ConsumeCount with PredictCount; reset both on TLSF slot reuse and count new predictions.
// 2. Fix retry/result handling; decrement _active_line_count only after a successful eviction.
高:换 predictor 的分支看起来反了。 MoveNextPredictor() 在 _predict 到达 _enter 时返回 true。因此 L4.Predict.cs:51 现在会在队列耗尽时再次取 predictor;TryGetCurrentPredictor 失败后仍 goto Predict,接着会用空的 p 调 PredictMission。反过来,有下一个 predictor 时它会提前返回,不再处理后续 predictor。 // 3. Correct TLSF reverse-slab pointer validation before freeing metadata pointers.
// 4. Defer TLSF Free until retired published buffers and outstanding handles can no longer reference the line.
高:新分配的 metadata 没有增加预测计数。 现有映射分支会调用 IncreasePredictCount(),但新分配分支只分配并写入 Swiss。若该计数表示未消费预测数,新条目也需要相同的计数处理;还要确认 TLSF 重用 metadata 时会重置旧计数。 // 5. Add Swiss TryRemoveAt(offset) to tombstone the enumerated slot without hashing and probing again.
高:同一轮内重复的 AddressSource 可能重复分配。 SwissTable.TryGet 只查 published buffer,而新条目先写入 inactive buffer。后续再次预测相同 key 时会查不到,重新分配 metadata,再由 TryAddOrSet 覆盖先前 value,旧分配就失去映射。
中:需要校验 PredictMission 的返回值范围。 Span 容量是 4;count > 4 会让 as_sp[i] 越界,负值则被当作本轮结束。建议契约明确要求 0..4,并在 L4 边界验证。
本轮未改代码。L4.Predict.cs 的文件诊断通过,但尚未运行完整构建或运行时测试。
*/
@@ -44,8 +44,7 @@ namespace Nitload.Common.HighPerformance.Memory
if (ReferenceEquals(predictor, _ring[idx])) if (ReferenceEquals(predictor, _ring[idx]))
{ {
int end = _predict; int end = _predict;
for (int move = i; move < end - 1; move++) for (int move = i; move < end - 1; move++) {
{
_ring[(byte)move] = _ring[(byte)(move + 1)]; _ring[(byte)move] = _ring[(byte)(move + 1)];
} }
_ring[(byte)(end - 1)] = default!; _ring[(byte)(end - 1)] = default!;
@@ -64,7 +63,7 @@ namespace Nitload.Common.HighPerformance.Memory
internal bool MoveNextPredictor() internal bool MoveNextPredictor()
{ {
int cur = Interlocked.Increment(ref _predict); int cur = Interlocked.Increment(ref _predict);
return cur == _enter; return cur < _enter;
} }
internal void RegistPredictor( internal void RegistPredictor(
@@ -92,8 +91,8 @@ namespace Nitload.Common.HighPerformance.Memory
{ {
unchecked unchecked
{ {
byte idx = (byte)_consume; byte idx = (byte)_enter;
_consume++; _enter++;
_ring[idx] = predictor; _ring[idx] = predictor;
} }
} }
@@ -97,16 +97,18 @@ namespace Nitload.Common.HighPerformance.Memory
} }
internal unsafe struct SwissEnumerator( internal unsafe struct SwissEnumerator(
byte* buffer byte* buffer,
int begins_at = 0
) )
{ {
private readonly SwissBucket* _buckets = (SwissBucket*)(buffer + IdealCount); private readonly SwissBucket* _buckets = (SwissBucket*)(buffer + IdealCount);
private readonly Vector128<byte>* _control_bytes = (Vector128<byte>*)buffer; private readonly Vector128<byte>* _control_bytes = (Vector128<byte>*)buffer;
private bool _group_loaded; private bool _group_loaded = false;
private int _current_group; private int _visited_group = 0;
private int _current_mask; private int _current_group = (begins_at / BucketCapacity) & (GroupCount - 1);
private int _current_offset; private int _current_mask = 0;
private int _current_offset = 0;
public (AddressSource key, ulong value) Current public (AddressSource key, ulong value) Current
{ {
@@ -119,9 +121,11 @@ namespace Nitload.Common.HighPerformance.Memory
} }
} }
public readonly int CurrentOffset => (_current_group * BucketCapacity) + _current_offset;
public bool MoveNext() public bool MoveNext()
{ {
while (_current_group < GroupCount) while (_visited_group < GroupCount)
{ {
if (!_group_loaded) if (!_group_loaded)
{ {
@@ -142,7 +146,8 @@ namespace Nitload.Common.HighPerformance.Memory
return true; return true;
} }
_current_group++; _current_group = (_current_group + 1) & (GroupCount - 1);
_visited_group++;
_group_loaded = false; _group_loaded = false;
} }
return false; return false;
@@ -218,6 +223,22 @@ namespace Nitload.Common.HighPerformance.Memory
throw new NotImplementedException("The Swiss backend is not implemented for this ISA."); throw new NotImplementedException("The Swiss backend is not implemented for this ISA.");
} }
/// <summary>
/// Reads the inactive write buffer. The caller must serialize this query with all
/// writes.
/// </summary>
public bool TryGetWrite(
AddressSource key,
out ulong value
)
{
if (Sse42.IsSupported) {
return SwissReaderX86.TryGet(_buffer_write, key, out value);
}
throw new NotImplementedException("The Swiss backend is not implemented for this ISA.");
}
public void TryRehash() public void TryRehash()
{ {
if (_delete_count < 2048) if (_delete_count < 2048)
@@ -278,6 +299,14 @@ namespace Nitload.Common.HighPerformance.Memory
return new SwissTable(buffers); return new SwissTable(buffers);
} }
// Enumerates the write buffer; the caller must serialize writers and prevent swaps.
internal readonly SwissEnumerator GetWriteBufferEnumerator(
int begins_at = 0
)
{
return new SwissEnumerator(_buffer_write, begins_at);
}
private static void InitializeBuffer( private static void InitializeBuffer(
byte* buffer byte* buffer
) )
@@ -6,6 +6,7 @@
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
using System.Numerics; using System.Numerics;
using System.Reflection.Metadata;
namespace Nitload.Common.HighPerformance.Memory namespace Nitload.Common.HighPerformance.Memory
{ {
@@ -23,10 +24,10 @@ namespace Nitload.Common.HighPerformance.Memory
private unsafe struct TLSF private unsafe struct TLSF
{ {
public const int MaxCacheLineCount = 8192; public const int MaxCacheLineCount = MaxActiveCacheLineCount;
public static readonly nuint RequestedAllocation = public static readonly nuint RequestedAllocation =
((nuint)MaxCacheLineCount + 1u) * CacheLineMetadata.SizeOf + CacheLineMetadata.SizeOf - 1u; (((nuint)MaxCacheLineCount + 1u) * CacheLineMetadata.SizeOf) + CacheLineMetadata.SizeOf - 1u;
private readonly CacheLineMetadata* _spareMemory; private readonly CacheLineMetadata* _spareMemory;
private readonly int _align; private readonly int _align;
private readonly int _maxLevel; private readonly int _maxLevel;
@@ -62,7 +63,7 @@ namespace Nitload.Common.HighPerformance.Memory
// init spare memory node // init spare memory node
_spareMemory->Position = _basePtr; _spareMemory->Position = (nuint)(ulong)_basePtr;
_spareMemory->SizeMask = _totalSize; _spareMemory->SizeMask = _totalSize;
_spareMemory->NextLevel = 0; _spareMemory->NextLevel = 0;
_spareMemory->NextNear = 0; _spareMemory->NextNear = 0;
@@ -119,7 +120,7 @@ namespace Nitload.Common.HighPerformance.Memory
node->PreviousNear = 0; node->PreviousNear = 0;
node->NextNear = 0; node->NextNear = 0;
node->Position = _basePtr; node->Position = (nuint)(ulong)_basePtr;
node->SizeMask = 0; node->SizeMask = 0;
node->State = AllocationState.SpareOrCollect; node->State = AllocationState.SpareOrCollect;
node->PreviousLevel = 0; node->PreviousLevel = 0;
@@ -405,11 +406,13 @@ namespace Nitload.Common.HighPerformance.Memory
} }
public void Free( public void Free(
int handle, CacheLineMetadata* head,
int collectScanWindow = 4 int collectScanWindow = 4
) )
{ {
CacheLineMetadata* head = MetadataAt(handle); if (head < _spareMemory || head >= _spareMemory + MaxCacheLineCount) {
throw new ArgumentOutOfRangeException(nameof(head), "Handle is out of range.");
}
short headIdx = (short)(_spareMemory - head); short headIdx = (short)(_spareMemory - head);
uint size = head->SizeMask; uint size = head->SizeMask;
for (int i = 0; i < collectScanWindow; i++) for (int i = 0; i < collectScanWindow; i++)
@@ -463,6 +466,27 @@ namespace Nitload.Common.HighPerformance.Memory
return; return;
} }
public void Free(
ulong handle,
int collectScanWindow = 4
)
{
if (handle < (ulong)_basePtr || handle >= (ulong)_basePtr + _totalSize) {
throw new ArgumentOutOfRangeException(nameof(handle), "Handle is out of range.");
}
CacheLineMetadata* head = (CacheLineMetadata*)handle;
Free(head, collectScanWindow);
}
public void Free(
int handle,
int collectScanWindow = 4
)
{
CacheLineMetadata* head = MetadataAt(handle);
Free(head, collectScanWindow);
}
public bool TryAlloc( public bool TryAlloc(
uint size, uint size,
out int h out int h
@@ -12,13 +12,16 @@ namespace Nitload.Common.HighPerformance.Memory
public sealed unsafe partial class L4 : IDisposable public sealed unsafe partial class L4 : IDisposable
{ {
private const int MaxActiveCacheLineCount = 8192;
private const nuint PageSize = 4096; private const nuint PageSize = 4096;
private const int PredictorRingCapacity = 256; private const int PredictorRingCapacity = 256;
private readonly IL4Predictor[] _ring;
private bool _disposed; private bool _disposed;
private readonly int _bindedNuma; private readonly int _bindedNuma;
private readonly nuint _baseAddress; private readonly nuint _baseAddress;
private readonly IL4Predictor[] _ring;
private SwissTable _addressMap; private SwissTable _addressMap;
private TLSF _bufferAllocator; private TLSF _bufferAllocator;
@@ -76,6 +79,8 @@ namespace Nitload.Common.HighPerformance.Memory
_bufferAllocator.Free(handle); _bufferAllocator.Free(handle);
} }
internal partial int Predict();
internal bool TryAllocate( internal bool TryAllocate(
uint size, uint size,
out int handle out int handle
@@ -85,14 +90,15 @@ namespace Nitload.Common.HighPerformance.Memory
return _bufferAllocator.TryAlloc(size, out handle); return _bufferAllocator.TryAlloc(size, out handle);
} }
internal partial int Predict(); private static int GetPredictSlowPathCount(
int bindedNuma
private static int GetPredictSlowPathCount(int bindedNuma) )
{ {
foreach (NumaNode node in Numa.AllNumaNode) foreach (NumaNode node in Numa.AllNumaNode)
{ {
if (node.NumaNodeId == (uint)bindedNuma) if (node.NumaNodeId == (uint)bindedNuma) {
return checked(node.LogicalCoreCount * 2); return checked(node.LogicalCoreCount * 2);
}
} }
throw new ArgumentOutOfRangeException(nameof(bindedNuma), "No topology information exists for the bound NUMA node."); throw new ArgumentOutOfRangeException(nameof(bindedNuma), "No topology information exists for the bound NUMA node.");
@@ -22,6 +22,12 @@ namespace Nitload.Common.HighPerformance.Memory
in Span<L4.AddressSource> RequiredBuffer in Span<L4.AddressSource> RequiredBuffer
); );
void Prefetch(
L4.AddressSource RequiredBuffer,
nuint allocBuffer,
nuint bufferSize
);
} }
public interface IL4Predictor<TMission> : IL4Predictor public interface IL4Predictor<TMission> : IL4Predictor
@@ -32,5 +38,4 @@ namespace Nitload.Common.HighPerformance.Memory
); );
} }
} }