From 2e7c3040f4d947dc1fd7f877279a1b4f0d1b2344 Mon Sep 17 00:00:00 2001 From: Youmu Liu <105042599+Nitload-NSI@users.noreply.github.com> Date: Thu, 8 Oct 2026 18:15:05 +0800 Subject: [PATCH] rewrite of swiss, swiss will not be public but l4 specific --- src/Nitload.Benchmark/SwissBenchmark.cs | 18 +- .../Collections/SwissTable.X86.cs | 130 ----------- src/Nitload.Common/Collections/SwissTable.cs | 203 +++++------------- .../HighPerformance/Memory/L4.Debug.cs | 3 +- .../HighPerformance/Memory/L4.Evict.cs | 4 +- .../HighPerformance/Memory/L4.Prefetch.cs | 7 +- .../HighPerformance/Memory/L4.cs | 2 +- .../ProcessorAbstraction/LogicalCore.cs | 19 +- .../ProcessorAbstraction/Numa.cs | 66 ++++-- .../ProcessorAbstraction/NumaNode.cs | 14 +- src/Nitload.Common/Threading/ThreadHelper.cs | 6 +- src/Nitload.Common/Timing/StgSharpTime.cs | 2 +- .../Timing/TimeSourceProvider.cs | 8 +- .../nif-head/NitloadFrameWork.h | 8 +- src/Nitload.Native/nif-src/thread.linux.c | 72 +++++-- src/Nitload.Native/nif-src/thread.windows.c | 171 +++++++++++---- test/STgSharpTest/UnitTest1.cs | 21 +- test/StgSharpDebug/TestSwiss.cs | 4 +- 18 files changed, 367 insertions(+), 391 deletions(-) delete mode 100644 src/Nitload.Common/Collections/SwissTable.X86.cs diff --git a/src/Nitload.Benchmark/SwissBenchmark.cs b/src/Nitload.Benchmark/SwissBenchmark.cs index 02596b8d..c72087cf 100644 --- a/src/Nitload.Benchmark/SwissBenchmark.cs +++ b/src/Nitload.Benchmark/SwissBenchmark.cs @@ -231,7 +231,7 @@ namespace Nitload.Benchmark { nuint key = _residentKeys[i]; nuint val = key ^ unchecked((nuint)0xDEAD_BEEF); - _freshSwiss.TryAddOrSet(key, val, out _); + SwissTable.Writer.TrySet(_freshSwiss, key, val); _freshDict[key] = val; } } @@ -273,7 +273,7 @@ namespace Nitload.Benchmark { nuint key = _residentKeys[i]; nuint val = key ^ unchecked((nuint)0xDEAD_BEEF); // simulated cache pointer - _swissLookup.TryAddOrSet(key, val, out _); + SwissTable.Writer.TrySet(_swissLookup, key, val); _dictLookup[key] = val; } } @@ -284,7 +284,7 @@ namespace Nitload.Benchmark SwissTable swiss = _freshSwiss; nuint[] keys = _residentKeys; for (int i = 0; i < keys.Length; i++) { - swiss.TryAddOrSet(keys[i], keys[i] ^ unchecked((nuint)0xDEAD_BEEF), out _); + SwissTable.Writer.TrySet(swiss, keys[i], keys[i] ^ unchecked((nuint)0xDEAD_BEEF)); } } @@ -296,7 +296,7 @@ namespace Nitload.Benchmark SwissTable swiss = _swissLookup; for (int i = 0; i < keys.Length; i++) { - if (swiss.TryGet(keys[i], out nuint value)) { + if (SwissTable.Reader.TryGet(swiss, keys[i], out nuint value)) { sum += value; } } @@ -311,7 +311,7 @@ namespace Nitload.Benchmark SwissTable swiss = _swissLookup; for (int i = 0; i < keys.Length; i++) { - if (swiss.TryGet(keys[i], out nuint value)) { + if (SwissTable.Reader.TryGet(swiss, keys[i], out nuint value)) { sum += value; } } @@ -336,7 +336,7 @@ namespace Nitload.Benchmark if (op >= 0) { // Cache hit: lookup existing entry - if (swiss.TryGet(resident[op], out nuint value)) + if (SwissTable.Reader.TryGet(swiss, resident[op], out nuint value)) { sum += value; } @@ -345,15 +345,15 @@ namespace Nitload.Benchmark // Cache miss: evict a cold entry, insert a new one, then lookup int evictIdx = evict[evictCursor % evict.Length]; evictCursor++; - swiss.TryRemove(resident[evictIdx], out _); + SwissTable.Writer.TryRemove(swiss, resident[evictIdx], out _); int extIdx = externalCursor % external.Length; externalCursor++; nuint newKey = external[extIdx]; nuint newVal = newKey ^ unchecked((nuint)0xDEAD_BEEF); - swiss.TryAddOrSet(newKey, newVal, out _); + SwissTable.Writer.TryAdd(swiss, newKey, newVal); - if (swiss.TryGet(newKey, out nuint v)) { + if (SwissTable.Reader.TryGet(swiss, newKey, out nuint v)) { sum += v; } } diff --git a/src/Nitload.Common/Collections/SwissTable.X86.cs b/src/Nitload.Common/Collections/SwissTable.X86.cs deleted file mode 100644 index a2e01eac..00000000 --- a/src/Nitload.Common/Collections/SwissTable.X86.cs +++ /dev/null @@ -1,130 +0,0 @@ -// ----------------------------------------------------------------------------- -// file="SwissTable.X86" -// Project: StgSharp -// Copyright (c) Nitload. -// SPDX-License-Identifier: MIT -// ----------------------------------------------------------------------------- - -using System; -using System.Numerics; -using System.Runtime.Intrinsics; -using System.Runtime.Intrinsics.X86; - -namespace Nitload.Common.Collections -{ - public partial class SwissTable - { - - private unsafe class BucketScannerX86 : IBucketScanner - { - - public bool TryFind( - nuint key, - uint hash, - Bucket* bucket, - out nuint value - ) - { - byte ctrl = (byte)((hash >> EntryMaskWidth) & 0x7F); - Vector128 ctrlVector = Vector128.Create(ctrl); - Vector128 bucketCtrlVector = Sse2.LoadVector128((byte*)&bucket->Control); - Vector128 cmpResult = Sse2.CompareEqual(ctrlVector, bucketCtrlVector); - int mask = Sse2.MoveMask(cmpResult); - while (mask != 0) - { - int bitPos = BitOperations.TrailingZeroCount(mask); - int index = bitPos; - if (bucket->Keys[index] == key) - { - value = bucket->Values[index]; - return true; - } - mask &= mask - 1; - } - value = default; - return false; - } - - public bool TryRemove( - nuint key, - uint hash, - Bucket* bucket, - out nuint value - ) - { - byte ctrl = (byte)((hash >> EntryMaskWidth) & 0x7F); - Vector128 ctrlVector = Vector128.Create(ctrl); - - Vector128 bucketCtrl = Sse2.LoadVector128((byte*)&bucket->Control); - Vector128 matchMask = Sse2.CompareEqual(ctrlVector, bucketCtrl); - int matches = Sse2.MoveMask(matchMask); - - while (matches != 0) - { - int bitPos = BitOperations.TrailingZeroCount(matches); - int index = bitPos; - if (bucket->Keys[index] == key) - { - value = bucket->Values[index]; - bucket->Control[index] = 0xFE; // mark as DELETED - // no need to remove Key/Value - return true; - } - matches &= matches - 1; - } - value = default; - return false; - } - - public bool TrySet( - nuint key, - uint hash, - Bucket* bucket, - nuint value, - out bool isNewKey - ) - { - byte ctrl = (byte)((hash >> EntryMaskWidth) & 0x7F); // H2: upper 8 bits (highest bit is naturally 0) - Vector128 ctrlVector = Vector128.Create(ctrl); - - Vector128 bucketCtrl = Sse2.LoadVector128( - (byte*)&bucket->Control); - - // === Phase 1: Look up an existing key === - Vector128 matchMask = Sse2.CompareEqual(ctrlVector, bucketCtrl); - int matches = Sse2.MoveMask(matchMask); - - while (matches != 0) - { - int bitPos = BitOperations.TrailingZeroCount(matches); - int index = bitPos; - if (bucket->Keys[index] == key) - { - // Key already exists -> update value - bucket->Values[index] = value; - isNewKey = false; - return true; - } - matches &= matches - 1; - } - - // === Phase 2: Track an available insertion slot === - int emptyOrDeleted = Sse2.MoveMask(bucketCtrl); - if (emptyOrDeleted == 0) - { - isNewKey = false; - return false; // bucket full - } - - int slot = BitOperations.TrailingZeroCount(emptyOrDeleted); - bucket->Control[slot] = ctrl; - bucket->Keys[slot] = key; - bucket->Values[slot] = value; - isNewKey = true; - return true; - } - - } - - } -} diff --git a/src/Nitload.Common/Collections/SwissTable.cs b/src/Nitload.Common/Collections/SwissTable.cs index 67bbc0a1..3f9c3eb6 100644 --- a/src/Nitload.Common/Collections/SwissTable.cs +++ b/src/Nitload.Common/Collections/SwissTable.cs @@ -5,143 +5,41 @@ // SPDX-License-Identifier: MIT // ----------------------------------------------------------------------------- +using Nitload.Common.HighPerformance.ProcessorAbstraction; using System; -using System.Numerics; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; -using System.Runtime.Intrinsics.Arm; -using System.Runtime.Intrinsics.X86; namespace Nitload.Common.Collections { - /// - /// - /// - public sealed unsafe partial class SwissTable : IDisposable + public sealed unsafe partial class SwissTable { - private const int BucketCount = 1 << EntryMaskWidth; private const int BucketSize = 16; private const uint CrcMagic = 0xffff9961; // Magic number for CRC32C, used in hash computation private const int EntryMaskWidth = 10; // 1024 entries total, 16 per bucket - public const int MaxCapacity = BucketCount * BucketSize; // 256 entries per bucket, 1024 buckets - public static readonly nuint RequestedNativeMemorySize = (nuint)Unsafe.SizeOf(); + private const int GroupCount = 1 << EntryMaskWidth; + public const int MaxCapacity = GroupCount * BucketSize; // 256 entries per bucket, 1024 buckets - private readonly Entry* _entries; - private readonly Action _freeHandle; - private bool _disposed; + public static readonly nuint RequestedNativeMemorySize = (nuint)Unsafe.SizeOf() + (nuint)Unsafe.SizeOf(); + + private byte* _controlBytes; private SwissTable( - Entry* buffer, - Action freeHandle + byte* buffer ) { - _entries = buffer; - _freeHandle = freeHandle; + _controlBytes = buffer; } - public static SwissTable Create() - { - Entry* buffer = (Entry*)NativeMemory.AlignedAlloc((nuint)Unsafe.SizeOf(), 16); - for (int i = 0; i < BucketCount; i++) - { - // Set all control bytes to 0xFF (empty) - Unsafe.InitBlock(&(*buffer)[i].Control, 0xFF, BucketSize); - } - return new SwissTable(buffer, handle => NativeMemory.AlignedFree((void*)handle)); - } + private M128* KvpArray => (M128*)(_controlBytes + MaxCapacity); - public static SwissTable Create( - Func allocator, - Action freeHandle + private static void Initialize( + byte* buffer ) { - Entry* buffer = (Entry*)allocator(RequestedNativeMemorySize); - for (int i = 0; i < BucketCount; i++) - { - // Set all control bytes to 0xFF (empty) - Unsafe.InitBlock(&(*buffer)[i].Control, 0xFF, BucketSize); - } - return new SwissTable(buffer, freeHandle); - } - - public void Dispose() - { - if (_disposed) { - return; - } - _disposed = true; - _freeHandle((nuint)_entries); - GC.SuppressFinalize(this); - } - - public bool TryAddOrSet( - nuint key, - nuint value, - out bool isNewKey - ) - { - uint hash = BitOperations.Crc32C(CrcMagic, (ulong)key); - ushort entry = (ushort)(hash & (BucketCount - 1)); - Bucket* bucket = &(*_entries)[entry]; - return IBucketScanner.Current.TrySet(key, hash, bucket, value, out isNewKey); - } - - public bool TryGet( - nuint key, - out nuint value - ) - { - uint hash = BitOperations.Crc32C(CrcMagic, key); - ushort entry = (ushort)(hash & (BucketCount - 1)); - Bucket* bucket = &(*_entries)[entry]; - return IBucketScanner.Current.TryFind(key, hash, bucket, out value); - /* - * hash = CRC32 of key - * entry = low 8 of hash - * ctrl = 8-15 of hash - * index = rest of hash - * find value of hash - */ - } - - public bool TryRemove( - nuint key, - out nuint value - ) - { - uint hash = BitOperations.Crc32C(CrcMagic, key); - ushort entry = (ushort)(hash & (BucketCount - 1)); - Bucket* bucket = &(*_entries)[entry]; - return IBucketScanner.Current.TryRemove(key, hash, bucket, out value); - } - - internal static SwissTable Create( - nuint bufferPtr, - Action FreeHandle - ) - { - Entry* buffer = (Entry*)bufferPtr; - for (int i = 0; i < BucketCount; i++) - { - // Set all control bytes to 0xFF (empty) - Unsafe.InitBlock(&(*buffer)[i].Control, 0xFF, BucketSize); - } - return new SwissTable(buffer, FreeHandle); - } - - ~SwissTable() - { - Dispose(); - } - - [InlineArray(BucketCount)] - private struct Entry - { - - public Bucket Bucket; - + Unsafe.InitBlock(buffer, 0b10000000, BucketSize); } private unsafe struct Bucket @@ -153,45 +51,48 @@ namespace Nitload.Common.Collections } - private interface IBucketScanner + public static class Reader { - static IBucketScanner Current { get; } = Create(); - - bool TryFind( - nuint key, - uint Hash, - Bucket* bucket, - out nuint value - ); - - unsafe bool TryRemove( - nuint key, - uint hash, - Bucket* bucket, - out nuint value - ); - - bool TrySet( - nuint key, - uint hash, - Bucket* bucket, - nuint value, - out bool isNewKey - ); - - private static IBucketScanner Create() + public static bool TryGet( + SwissTable table, + nuint key, + out nuint value + ) { - if (Sse42.IsSupported) - { - return new BucketScannerX86(); - } else if (ArmBase.IsSupported) - { - throw new NotSupportedException("ARM architecture is currently not supported for bucket scanning. Please use an x86 platform."); - } else - { - throw new NotSupportedException("No suitable bucket scanner implementation found for the current platform."); - } + throw new NotImplementedException("SwissTable reader is awaiting a redesigned storage and lookup implementation."); + } + + } + + public static class Writer + { + + public static bool TryAdd( + SwissTable table, + nuint key, + nuint value + ) + { + throw new NotImplementedException("SwissTable writer is awaiting a redesigned storage and insertion implementation."); + } + + public static bool TryRemove( + SwissTable table, + nuint key, + out nuint value + ) + { + throw new NotImplementedException("SwissTable writer is awaiting a redesigned storage and removal implementation."); + } + + public static bool TrySet( + SwissTable table, + nuint key, + nuint value + ) + { + throw new NotImplementedException("SwissTable writer is awaiting a redesigned storage and update implementation."); } } diff --git a/src/Nitload.Common/HighPerformance/Memory/L4.Debug.cs b/src/Nitload.Common/HighPerformance/Memory/L4.Debug.cs index 53a40633..fff7dc56 100644 --- a/src/Nitload.Common/HighPerformance/Memory/L4.Debug.cs +++ b/src/Nitload.Common/HighPerformance/Memory/L4.Debug.cs @@ -5,6 +5,7 @@ // SPDX-License-Identifier: MIT // ----------------------------------------------------------------------------- +using Nitload.Common.Collections; using System; using System.Collections.Generic; using System.Linq; @@ -26,7 +27,7 @@ namespace Nitload.Common.HighPerformance.Memory nuint origin ) { - return _map.TryGet(origin, out _); + return SwissTable.Reader.TryGet(_map, origin, out _); } public int DebugEvict( diff --git a/src/Nitload.Common/HighPerformance/Memory/L4.Evict.cs b/src/Nitload.Common/HighPerformance/Memory/L4.Evict.cs index f53be977..82cf0c26 100644 --- a/src/Nitload.Common/HighPerformance/Memory/L4.Evict.cs +++ b/src/Nitload.Common/HighPerformance/Memory/L4.Evict.cs @@ -5,6 +5,8 @@ // SPDX-License-Identifier: MIT // ----------------------------------------------------------------------------- +using Nitload.Common.Collections; + namespace Nitload.Common.HighPerformance.Memory { public partial class L4 @@ -49,7 +51,7 @@ namespace Nitload.Common.HighPerformance.Memory if (Interlocked.CompareExchange(ref head->RefCount, int.MinValue, 0) == 0) { // EvictLine(i + index); - _ = _map.TryRemove(head->Origin, out _); + _ = SwissTable.Writer.TryRemove(_map, head->Origin, out _); IL4Predict predict = CurrentPredict; predict.WriteBack(head->Origin, head->CustomizeProfile, new ReadOnlySpan((byte*)(ulong)(head->PositionMask + _baseAddress), (int)(ulong)head->SizeMask)); _bufferAllocator.Free(index); diff --git a/src/Nitload.Common/HighPerformance/Memory/L4.Prefetch.cs b/src/Nitload.Common/HighPerformance/Memory/L4.Prefetch.cs index 73c7694e..0a87f428 100644 --- a/src/Nitload.Common/HighPerformance/Memory/L4.Prefetch.cs +++ b/src/Nitload.Common/HighPerformance/Memory/L4.Prefetch.cs @@ -5,6 +5,7 @@ // SPDX-License-Identifier: MIT // ----------------------------------------------------------------------------- +using Nitload.Common.Collections; using System.ComponentModel; using System.Numerics; @@ -32,7 +33,7 @@ namespace Nitload.Common.HighPerformance.Memory { ref CacheLineDescription handle = ref span[i]; nuint address = handle.Address; - if (!_map.TryGet(address, out _)) + if (!SwissTable.Reader.TryGet(_map, address, out _)) { // miss and record span[missIndex] = handle; @@ -47,7 +48,7 @@ namespace Nitload.Common.HighPerformance.Memory ref CacheLineDescription pHandle = ref span[i]; nuint address = pHandle.Address; int index; - if (_map.TryGet(address, out nuint handle)) + if (SwissTable.Reader.TryGet(_map, address, out nuint handle)) { head = (CacheLineMetadata*)handle; index = (int)(head - _head); @@ -93,7 +94,7 @@ namespace Nitload.Common.HighPerformance.Memory // pre fetch them here predict.Prefetch(head->Origin, head->CustomizeProfile, new Span((byte*)(head->PositionMask + _baseAddress), (int)head->SizeMask)); - _ = _map.TryAddOrSet(address, (nuint)head, out _); + _ = SwissTable.Writer.TrySet(_map, address, (nuint)head); } _ = Interlocked.Add(ref _aheadCount, actualCount); } diff --git a/src/Nitload.Common/HighPerformance/Memory/L4.cs b/src/Nitload.Common/HighPerformance/Memory/L4.cs index cd7e7a48..e91f65ef 100644 --- a/src/Nitload.Common/HighPerformance/Memory/L4.cs +++ b/src/Nitload.Common/HighPerformance/Memory/L4.cs @@ -162,7 +162,7 @@ namespace Nitload.Common.HighPerformance.Memory } while (true) { - if (_map.TryGet(prediction.GetHashCodeLong(), out nuint entryHandle)) + if (SwissTable.Reader.TryGet(_map, prediction.GetHashCodeLong(), out nuint entryHandle)) { // cache hit CacheLineMetadata* entry = (CacheLineMetadata*)entryHandle; diff --git a/src/Nitload.Common/HighPerformance/ProcessorAbstraction/LogicalCore.cs b/src/Nitload.Common/HighPerformance/ProcessorAbstraction/LogicalCore.cs index b4ada2af..a15550cf 100644 --- a/src/Nitload.Common/HighPerformance/ProcessorAbstraction/LogicalCore.cs +++ b/src/Nitload.Common/HighPerformance/ProcessorAbstraction/LogicalCore.cs @@ -16,22 +16,21 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction public readonly struct LogicalCore { - public readonly uint GlobalCpuId; - public readonly uint NumaGroupId; - - /// - /// Stays 0 on linux. - /// + public readonly uint ProcessorGroupId; + public readonly uint SocketId; + public readonly uint CpuIndex; public readonly uint NumaNodeId; internal LogicalCore( - uint globalCpuId, - uint numaGroupId, + uint processorGroupId, + uint socketId, + uint cpuIndex, uint numaNodeId ) { - GlobalCpuId = globalCpuId; - NumaGroupId = numaGroupId; + ProcessorGroupId = processorGroupId; + SocketId = socketId; + CpuIndex = cpuIndex; NumaNodeId = numaNodeId; } diff --git a/src/Nitload.Common/HighPerformance/ProcessorAbstraction/Numa.cs b/src/Nitload.Common/HighPerformance/ProcessorAbstraction/Numa.cs index 4215ea23..be342a30 100644 --- a/src/Nitload.Common/HighPerformance/ProcessorAbstraction/Numa.cs +++ b/src/Nitload.Common/HighPerformance/ProcessorAbstraction/Numa.cs @@ -33,7 +33,7 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction if (_numaNodes.Length == 0) { throw new InvalidOperationException("Numa info has not been loaded."); } - if ((numaId < 0) || (numaId >= _numaNodes.Length)) { + if ((numaId < 0) || !ContainsNode((uint)numaId)) { throw new ArgumentOutOfRangeException(nameof(numaId), "Invalid NUMA node ID."); } return NumaAllocNative(size, numaId); @@ -69,7 +69,7 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction if (_numaNodes.Length == 0) { throw new InvalidOperationException("Numa info has not been loaded."); } - if ((numaId < 0) || (numaId >= _numaNodes.Length)) { + if ((numaId < 0) || !ContainsNode((uint)numaId)) { throw new ArgumentOutOfRangeException(nameof(numaId), "Invalid NUMA node ID."); } NumaFreeNative(address, size, numaId); @@ -80,8 +80,10 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction [DefaultDllImportSearchPaths(DllImportSearchPath.AssemblyDirectory)] public static partial int GetNumaEach( uint numaId, - Span CpuIndexInNuma, - Span globalCpuIdInNuma + Span processorGroupIds, + Span socketIds, + Span cpuIndices, + Span numaNodeIds ); [LibraryImport(Native.LibName, EntryPoint = "nif_get_numa_overall")] @@ -97,10 +99,10 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction if (_numaNodes.Length == 0) { throw new InvalidOperationException("Numa info has not been loaded."); } - L4[] buffers = new L4[NumaNodeCount]; - for (int i = 0; i < NumaNodeCount; i++) + L4[] buffers = new L4[_numaNodes.Length]; + for (int i = 0; i < _numaNodes.Length; i++) { - L4 buffer = new L4(i); + L4 buffer = new L4((int)_numaNodes[i].NumaNodeId); buffers[i] = buffer; } return buffers; @@ -108,14 +110,29 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction internal static void Initialize() { - Span coreCountInNuma = stackalloc int[(Environment.ProcessorCount / 32) + 1]; int numaNodeCount = 0; - if (0 == GetNumaOverall(ref numaNodeCount, coreCountInNuma)) { + if (0 == GetNumaOverall(ref numaNodeCount, default)) { throw new InvalidOperationException("Failed to get NUMA overall information."); } - _numaNodes = new NumaNode[numaNodeCount]; - for (uint i = 0; i < numaNodeCount; i++) { - _numaNodes[i] = InitNumaNode(i, coreCountInNuma[(int)i]); + int[] coreCountInNuma = new int[numaNodeCount]; + if (0 == GetNumaOverall(ref numaNodeCount, coreCountInNuma)) { + throw new InvalidOperationException("Failed to get NUMA node processor counts."); + } + int activeNodeCount = 0; + for (int i = 0; i < coreCountInNuma.Length; i++) { + if (coreCountInNuma[i] > 0) { + activeNodeCount++; + } + } + NumaNodeCount = activeNodeCount; + _numaNodes = new NumaNode[activeNodeCount]; + int nodeIndex = 0; + for (uint numaId = 0; numaId < numaNodeCount; numaId++) { + int coreCount = coreCountInNuma[(int)numaId]; + if (coreCount == 0) { + continue; + } + _numaNodes[nodeIndex++] = InitNumaNode(numaId, coreCount); } } @@ -123,8 +140,8 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction [UnmanagedCallConv(CallConvs = [typeof(CallConvCdecl)])] [DefaultDllImportSearchPaths(DllImportSearchPath.AssemblyDirectory)] internal static partial void SetThreadAffinity( - int coreId, - int numaId + int cpuIndex, + int processorGroupId ); [MethodImpl(MethodImplOptions.NoInlining)] @@ -133,9 +150,24 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction int count ) { - Span globalIndex = stackalloc uint[count]; - _ = GetNumaEach(numaId, null, globalIndex); - return new(numaId, count, globalIndex); + Span processorGroupIds = stackalloc uint[count]; + Span socketIds = stackalloc uint[count]; + Span cpuIndices = stackalloc uint[count]; + Span numaNodeIds = stackalloc uint[count]; + if (0 == GetNumaEach(numaId, processorGroupIds, socketIds, cpuIndices, numaNodeIds)) { + throw new InvalidOperationException($"Failed to enumerate processors for NUMA node {numaId}."); + } + return new(numaId, count, processorGroupIds, socketIds, cpuIndices, numaNodeIds); + } + + private static bool ContainsNode(uint numaId) + { + foreach (NumaNode node in _numaNodes) { + if (node.NumaNodeId == numaId) { + return true; + } + } + return false; } [LibraryImport(Native.LibName, EntryPoint = "nif_numa_alloc")] diff --git a/src/Nitload.Common/HighPerformance/ProcessorAbstraction/NumaNode.cs b/src/Nitload.Common/HighPerformance/ProcessorAbstraction/NumaNode.cs index 441880a7..af4971ee 100644 --- a/src/Nitload.Common/HighPerformance/ProcessorAbstraction/NumaNode.cs +++ b/src/Nitload.Common/HighPerformance/ProcessorAbstraction/NumaNode.cs @@ -22,16 +22,26 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction internal NumaNode( uint numaId, int count, - Span logicalCoreIds + Span processorGroupIds, + Span socketIds, + Span cpuIndices, + Span numaNodeIds ) { + NumaNodeId = numaId; _logicalCores = new LogicalCore[count]; for (int i = 0; i < count; i++) { - _logicalCores[i] = new LogicalCore((uint)i, numaId, logicalCoreIds[i]); + _logicalCores[i] = new LogicalCore( + processorGroupIds[i], + socketIds[i], + cpuIndices[i], + numaNodeIds[i]); } } + public uint NumaNodeId { get; } + public int LogicalCoreCount => _logicalCores.Length; public ReadOnlySpan LogicalCores => _logicalCores; diff --git a/src/Nitload.Common/Threading/ThreadHelper.cs b/src/Nitload.Common/Threading/ThreadHelper.cs index 597f9e19..756612fa 100644 --- a/src/Nitload.Common/Threading/ThreadHelper.cs +++ b/src/Nitload.Common/Threading/ThreadHelper.cs @@ -37,7 +37,7 @@ namespace Nitload.Threading LogicalCore core ) { - SetThreadAffinity(core.GlobalCpuId, core.NumaNodeId); + SetThreadAffinity(core.CpuIndex, core.ProcessorGroupId); } [Conditional("DEBUG")] @@ -55,8 +55,8 @@ namespace Nitload.Threading [UnmanagedCallConv(CallConvs = [typeof(CallConvCdecl)])] [DefaultDllImportSearchPaths(DllImportSearchPath.AssemblyDirectory)] private static partial void SetThreadAffinity( - uint globalCoreId, - uint numaNodeId + uint cpuIndex, + uint processorGroupId ); #pragma warning restore CA5393 diff --git a/src/Nitload.Common/Timing/StgSharpTime.cs b/src/Nitload.Common/Timing/StgSharpTime.cs index f4c8422e..d8451b65 100644 --- a/src/Nitload.Common/Timing/StgSharpTime.cs +++ b/src/Nitload.Common/Timing/StgSharpTime.cs @@ -97,7 +97,7 @@ namespace Nitload } Thread.Sleep(10); - Numa.SetThreadAffinity((int)threadConfig.BindedCoreIndex, (int)threadConfig.BindedNumaNode); + Numa.SetThreadAffinity((int)threadConfig.CpuIndex, (int)threadConfig.ProcessorGroupId); while (mainProvider.IsRunning) { long elapsedTicks; diff --git a/src/Nitload.Common/Timing/TimeSourceProvider.cs b/src/Nitload.Common/Timing/TimeSourceProvider.cs index 73bd5776..395c26ba 100644 --- a/src/Nitload.Common/Timing/TimeSourceProvider.cs +++ b/src/Nitload.Common/Timing/TimeSourceProvider.cs @@ -14,15 +14,15 @@ using System.Threading; namespace Nitload.Common.Timing { public class TimeSourceStartupConfiguration( - nuint bindedNumaNode = 0, - nuint bindedCoreIndex = 0, + nuint processorGroupId = 0, + nuint cpuIndex = 0, ThreadPriority threadPriority = ThreadPriority.Highest ) { - public nuint BindedNumaNode { get; } = bindedNumaNode; + public nuint ProcessorGroupId { get; } = processorGroupId; - public nuint BindedCoreIndex { get; } = bindedCoreIndex; + public nuint CpuIndex { get; } = cpuIndex; public ThreadPriority Priority { get; } = threadPriority; diff --git a/src/Nitload.Native/nif-head/NitloadFrameWork.h b/src/Nitload.Native/nif-head/NitloadFrameWork.h index 01665b13..df1f430f 100644 --- a/src/Nitload.Native/nif-head/NitloadFrameWork.h +++ b/src/Nitload.Native/nif-head/NitloadFrameWork.h @@ -27,9 +27,13 @@ NIF_API SIMDID NIF_DECL nif_get_simd_level_global(); NIF_API SIMDID NIF_DECL nif_get_simd_level_local(); NIF_API SIMDID NIF_DECL nif_get_unite_simd(SIMDID left, SIMDID right); NIF_API int NIF_DECL nif_compare_simd(SIMDID left, SIMDID right); -NIF_API void NIF_DECL nif_set_thread_affinity(int core_id, int numa_id); +NIF_API void NIF_DECL nif_set_thread_affinity(int cpu_index, int processor_group_id); NIF_API void* NIF_DECL nif_numa_alloc(size_t size, int numa_id); NIF_API void NIF_DECL nif_numa_free(void *ptr, size_t size, int numa_id); -NIF_API int NIF_DECL nif_get_numa_each(int numa_id, uint64_t *internal, uint32_t *global); +NIF_API int NIF_DECL nif_get_numa_each(int numa_id, + uint32_t *processor_group_ids, + uint32_t *socket_ids, + uint32_t *cpu_indices, + uint32_t *numa_node_ids); NIF_API int NIF_DECL nif_get_numa_overall(int *out_numa_count, int *count_of_each_numa); #endif diff --git a/src/Nitload.Native/nif-src/thread.linux.c b/src/Nitload.Native/nif-src/thread.linux.c index b29bf47e..f97575c0 100644 --- a/src/Nitload.Native/nif-src/thread.linux.c +++ b/src/Nitload.Native/nif-src/thread.linux.c @@ -27,8 +27,11 @@ typedef struct nif_linux_max_context { } nif_linux_max_context; typedef struct nif_linux_fill_cpu_context { - uint64_t *internal; - uint32_t *global; + uint32_t *processor_group_ids; + uint32_t *socket_ids; + uint32_t *cpu_indices; + uint32_t *numa_node_ids; + int numa_id; int index; } nif_linux_fill_cpu_context; @@ -118,14 +121,37 @@ static int nif_linux_track_max_id(int value, void *context) return 1; } +static int nif_linux_read_socket_id(int cpu_id, uint32_t *socket_id) +{ + char path[160]; + char buffer[64]; + char *end = NULL; + long parsed_value; + snprintf(path, sizeof(path), + "/sys/devices/system/cpu/cpu%d/topology/physical_package_id", cpu_id); + if (!nif_linux_read_first_line(path, buffer, sizeof(buffer))) + return 0; + + errno = 0; + parsed_value = strtol(buffer, &end, 10); + if (end == buffer || errno == ERANGE || parsed_value < 0 || parsed_value > UINT32_MAX) + return 0; + + *socket_id = (uint32_t)parsed_value; + return 1; +} + static int nif_linux_fill_cpu_id(int value, void *context) { nif_linux_fill_cpu_context *fill_context = (nif_linux_fill_cpu_context *)context; + uint32_t socket_id = 0; + if (!nif_linux_read_socket_id(value, &socket_id)) + return 0; - if (fill_context->internal != NULL) - fill_context->internal[fill_context->index] = (uint64_t)fill_context->index; - if (fill_context->global != NULL) - fill_context->global[fill_context->index] = (uint32_t)value; + fill_context->processor_group_ids[fill_context->index] = 0u; + fill_context->socket_ids[fill_context->index] = socket_id; + fill_context->cpu_indices[fill_context->index] = (uint32_t)value; + fill_context->numa_node_ids[fill_context->index] = (uint32_t)fill_context->numa_id; fill_context->index++; return 1; @@ -211,12 +237,14 @@ static int nif_linux_populate_writable_mapping(void *address, size_t size) #endif } -/* Pin the calling thread to a single logical processor. - * core_id >= 0 : bind to that logical processor (global index, across groups). - * core_id < 0 : clear affinity constraint within the current group. */ -NIF_API void NIF_DECL nif_set_thread_affinity(int core_id, int numa_id) +/* Linux has no Windows-style processor groups; processor_group_id is ignored. */ +NIF_API void NIF_DECL nif_set_thread_affinity(int cpu_index, int processor_group_id) { - int cpu_count = (int)sysconf(_SC_NPROCESSORS_ONLN); + (void)processor_group_id; + + int cpu_count = (int)sysconf(_SC_NPROCESSORS_CONF); + if (cpu_index >= cpu_count) + cpu_count = cpu_index + 1; if (cpu_count <= 0) return; @@ -227,11 +255,11 @@ NIF_API void NIF_DECL nif_set_thread_affinity(int core_id, int numa_id) CPU_ZERO_S(size, set); - if (core_id < 0) { + if (cpu_index < 0) { for (int i = 0; i < cpu_count; i++) CPU_SET_S((int)i, size, set); } else { - CPU_SET_S(core_id, size, set); + CPU_SET_S(cpu_index, size, set); } pthread_setaffinity_np(pthread_self(), size, set); @@ -264,9 +292,14 @@ NIF_API int NIF_DECL nif_get_numa_overall(int *out_numa_count, int *count_of_eac return 1; } -NIF_API int NIF_DECL nif_get_numa_each(int numa_id, uint64_t *internal, uint32_t *global) +NIF_API int NIF_DECL nif_get_numa_each(int numa_id, + uint32_t *processor_group_ids, + uint32_t *socket_ids, + uint32_t *cpu_indices, + uint32_t *numa_node_ids) { - if (numa_id < 0) + if (numa_id < 0 || processor_group_ids == NULL || socket_ids == NULL || + cpu_indices == NULL || numa_node_ids == NULL) return 0; char path[128]; @@ -275,7 +308,14 @@ NIF_API int NIF_DECL nif_get_numa_each(int numa_id, uint64_t *internal, uint32_t if (!nif_linux_read_first_line(path, buffer, sizeof(buffer))) return 0; - nif_linux_fill_cpu_context fill_context = { internal, global, 0 }; + nif_linux_fill_cpu_context fill_context = { + processor_group_ids, + socket_ids, + cpu_indices, + numa_node_ids, + numa_id, + 0 + }; return nif_linux_visit_id_list(buffer, nif_linux_fill_cpu_id, &fill_context); } diff --git a/src/Nitload.Native/nif-src/thread.windows.c b/src/Nitload.Native/nif-src/thread.windows.c index 74f8fda2..fd3f2eef 100644 --- a/src/Nitload.Native/nif-src/thread.windows.c +++ b/src/Nitload.Native/nif-src/thread.windows.c @@ -3,33 +3,102 @@ #if NIF_IS_OS(NIF_OS_WINDOWS) -/* Pin the calling thread to a single logical processor. - * core_id >= 0 : bind to that logical processor (global index, across groups). - * core_id < 0 : clear affinity constraint within the current group. */ -NIF_API void NIF_DECL nif_set_thread_affinity(int core_id, int numa_id) +static int nif_windows_get_socket_map(uint32_t **out_socket_map, DWORD *out_group_count) { - if (core_id < 0 || numa_id < 0) { - /* Query which group the thread currently belongs to */ + DWORD group_count = GetActiveProcessorGroupCount(); + DWORD buffer_size = 0; + uint32_t *socket_map; + SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX *information; + BYTE *cursor; + uint32_t socket_id = 0; + + if (group_count == 0) + return 0; + + socket_map = (uint32_t *)malloc((size_t)group_count * 64u * sizeof(uint32_t)); + if (socket_map == NULL) + return 0; + for (size_t i = 0; i < (size_t)group_count * 64u; i++) + socket_map[i] = UINT32_MAX; + + GetLogicalProcessorInformationEx(RelationProcessorPackage, NULL, &buffer_size); + if (GetLastError() != ERROR_INSUFFICIENT_BUFFER || buffer_size == 0) { + free(socket_map); + return 0; + } + + information = (SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX *)malloc(buffer_size); + if (information == NULL || + !GetLogicalProcessorInformationEx(RelationProcessorPackage, information, &buffer_size)) { + free(information); + free(socket_map); + return 0; + } + + cursor = (BYTE *)information; + while (cursor < (BYTE *)information + buffer_size) { + SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX *entry = + (SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX *)cursor; + if (entry->Relationship == RelationProcessorPackage) { + PROCESSOR_RELATIONSHIP *package = &entry->Processor; + for (WORD i = 0; i < package->GroupCount; i++) { + WORD group = package->GroupMask[i].Group; + KAFFINITY mask = package->GroupMask[i].Mask; + if (group >= group_count) + continue; + for (DWORD cpu = 0; cpu < 64; cpu++) { + if ((mask & ((KAFFINITY)1 << cpu)) != 0) + socket_map[(size_t)group * 64u + cpu] = socket_id; + } + } + socket_id++; + } + if (entry->Size == 0) + break; + cursor += entry->Size; + } + + free(information); + *out_socket_map = socket_map; + *out_group_count = group_count; + return 1; +} + +static int nif_windows_get_processor_node(WORD group, BYTE cpu, USHORT *node_id) +{ + PROCESSOR_NUMBER processor; + memset(&processor, 0, sizeof(processor)); + processor.Group = group; + processor.Number = cpu; + return GetNumaProcessorNodeEx(&processor, node_id) != 0; +} + +/* Pin the calling thread using its processor group and group-local CPU index. */ +NIF_API void NIF_DECL nif_set_thread_affinity(int cpu_index, int processor_group_id) +{ + if (cpu_index < 0 || processor_group_id < 0) { GROUP_AFFINITY current; GetThreadGroupAffinity(GetCurrentThread(), ¤t); - /* Build a full mask for that group's actual core count */ DWORD count = GetActiveProcessorCount(current.Group); - GROUP_AFFINITY ga; - memset(&ga, 0, sizeof(ga)); - ga.Group = current.Group; - ga.Mask = (count >= 64) ? ~(KAFFINITY)0 : ((KAFFINITY)1 << count) - 1; - SetThreadGroupAffinity(GetCurrentThread(), &ga, NULL); + GROUP_AFFINITY affinity; + memset(&affinity, 0, sizeof(affinity)); + affinity.Group = current.Group; + affinity.Mask = (count >= 64) ? ~(KAFFINITY)0 : ((KAFFINITY)1 << count) - 1; + SetThreadGroupAffinity(GetCurrentThread(), &affinity, NULL); return; } - int index = core_id % 64; - DWORD count = GetActiveProcessorCount(numa_id); - if ((DWORD)index >= count) + + if ((DWORD)processor_group_id >= GetActiveProcessorGroupCount() || cpu_index >= 64) return; - GROUP_AFFINITY ga; - memset(&ga, 0, sizeof(ga)); - ga.Group = (WORD)numa_id; - ga.Mask = (KAFFINITY)1 << index; - SetThreadGroupAffinity(GetCurrentThread(), &ga, NULL); + USHORT node_id = 0; + if (!nif_windows_get_processor_node((WORD)processor_group_id, (BYTE)cpu_index, &node_id)) + return; + + GROUP_AFFINITY affinity; + memset(&affinity, 0, sizeof(affinity)); + affinity.Group = (WORD)processor_group_id; + affinity.Mask = (KAFFINITY)1 << cpu_index; + SetThreadGroupAffinity(GetCurrentThread(), &affinity, NULL); } NIF_API int NIF_DECL nif_get_numa_overall(int *out_numa_count, int *count_of_each_numa) @@ -39,30 +108,60 @@ NIF_API int NIF_DECL nif_get_numa_overall(int *out_numa_count, int *count_of_eac return 0; *out_numa_count = (int)highest + 1; - for (UCHAR n = 0; n <= (UCHAR)highest; n++) { - ULONGLONG mask = 0; - GetNumaNodeProcessorMask(n, &mask); - count_of_each_numa[n] = (int)__popcnt64(mask); + if (count_of_each_numa != NULL) { + for (ULONG n = 0; n <= highest; n++) + count_of_each_numa[n] = 0; + } + + DWORD group_count = GetActiveProcessorGroupCount(); + for (DWORD group = 0; group < group_count; group++) { + for (DWORD cpu = 0; cpu < 64; cpu++) { + USHORT node_id = 0; + if (nif_windows_get_processor_node((WORD)group, (BYTE)cpu, &node_id) && + count_of_each_numa != NULL && node_id <= highest) + count_of_each_numa[node_id]++; + } } return 1; } -NIF_API int NIF_DECL nif_get_numa_each(int numa_id, uint64_t *internal, uint32_t *global) +NIF_API int NIF_DECL nif_get_numa_each(int numa_id, + uint32_t *processor_group_ids, + uint32_t *socket_ids, + uint32_t *cpu_indices, + uint32_t *numa_node_ids) { - ULONGLONG mask = 0; - if (!GetNumaNodeProcessorMask((UCHAR)numa_id, &mask)) + if (numa_id < 0 || processor_group_ids == NULL || socket_ids == NULL || + cpu_indices == NULL || numa_node_ids == NULL) return 0; - int intra = 0; - while (mask) { - DWORD idx; - _BitScanForward64(&idx, mask); - if (internal != NULL) - internal[intra] = (uint64_t)intra; - global[intra] = (uint32_t)idx; - intra++; - mask &= mask - 1; + uint32_t *socket_map = NULL; + DWORD group_count = 0; + if (!nif_windows_get_socket_map(&socket_map, &group_count)) + return 0; + + int index = 0; + for (DWORD group = 0; group < group_count; group++) { + for (DWORD cpu = 0; cpu < 64; cpu++) { + USHORT node_id = 0; + if (!nif_windows_get_processor_node((WORD)group, (BYTE)cpu, &node_id) || + node_id != (USHORT)numa_id) + continue; + + uint32_t socket_id = socket_map[(size_t)group * 64u + cpu]; + if (socket_id == UINT32_MAX) { + free(socket_map); + return 0; + } + processor_group_ids[index] = (uint32_t)group; + socket_ids[index] = socket_id; + cpu_indices[index] = cpu; + numa_node_ids[index] = node_id; + index++; + } } + + free(socket_map); return 1; } diff --git a/test/STgSharpTest/UnitTest1.cs b/test/STgSharpTest/UnitTest1.cs index 3b842f5b..998f444c 100644 --- a/test/STgSharpTest/UnitTest1.cs +++ b/test/STgSharpTest/UnitTest1.cs @@ -7,7 +7,7 @@ // // Permission is hereby granted, free of charge, to any person // obtaining a copy of this software and associated documentation -// files (the ¡°Software¡±), to deal in the Software without restriction, +// files (the ��Software��), to deal in the Software without restriction, // including without limitation the rights to use, copy, modify, merge, // publish, distribute, sublicense, and/or sell copies of the Software, // and to permit persons to whom the Software is furnished to do so, @@ -17,7 +17,7 @@ // this permission notice shall be included in all copies // or substantial portions of the Software. // -// THE SOFTWARE IS PROVIDED ¡°AS IS¡±, +// THE SOFTWARE IS PROVIDED ��AS IS��, // WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, // INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. @@ -32,6 +32,7 @@ using StgSharp; using StgSharp.Collections; using StgSharp.Script; using StgSharp.Script.Express; +using Nitload.Common.Collections; //using StgSharp.Model; @@ -46,6 +47,22 @@ namespace STgSharpTest { string testExpPath = @"D:\Programmer\stepcode\data\ap203\ap203.exp"; + [TestMethod] + public void SwissTableOperationsAreNotImplemented() + { + using SwissTable table = SwissTable.Create(); + nuint key = 17; + + Assert.ThrowsException( + () => SwissTable.Reader.TryGet(table, key, out _)); + Assert.ThrowsException( + () => SwissTable.Writer.TryAdd(table, key, 100)); + Assert.ThrowsException( + () => SwissTable.Writer.TrySet(table, key, 100)); + Assert.ThrowsException( + () => SwissTable.Writer.TryRemove(table, key, out _)); + } + [TestMethod] public void RegexTest() { diff --git a/test/StgSharpDebug/TestSwiss.cs b/test/StgSharpDebug/TestSwiss.cs index 9b9fd7e2..3997e1ff 100644 --- a/test/StgSharpDebug/TestSwiss.cs +++ b/test/StgSharpDebug/TestSwiss.cs @@ -24,7 +24,7 @@ namespace Nitload for (nuint i = 0; i < 5000; i++) { Console.Write($"Set {i}: "); - _ = swiss.TryAddOrSet(i, i * 2, out _); + _ = SwissTable.Writer.TrySet(swiss, i, i * 2); Console.WriteLine($"Set a kvp of ({i},{i * 2})"); } Random rand = new Random(); @@ -32,7 +32,7 @@ namespace Nitload { Console.Write($"Get {i}: "); nuint key = (nuint)rand.Next(4000, 5000); - if (swiss.TryGet(key, out nuint value)) + if (SwissTable.Reader.TryGet(swiss, key, out nuint value)) { Console.WriteLine($"Success in get a kvp of ({key},{value})"); } else