youmuliu/stgsharp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rewrite of swiss, swiss will not be public but l4 specific

Browse Source
This commit is contained in:
Youmu Liu
2026-10-08 18:15:05 +08:00
parent 1845a00390
commit 2e7c3040f4
18 changed files with 367 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Nitload.Benchmark/SwissBenchmark.cs
+9 -9
View File
@@ -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;
}
}
src/Nitload.Common/Collections/SwissTable.X86.cs
-130
View File
@@ -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<byte> ctrlVector = Vector128.Create(ctrl);
Vector128<byte> bucketCtrlVector = Sse2.LoadVector128((byte*)&bucket->Control);
Vector128<byte> 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<byte> ctrlVector = Vector128.Create(ctrl);
Vector128<byte> bucketCtrl = Sse2.LoadVector128((byte*)&bucket->Control);
Vector128<byte> 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<byte> ctrlVector = Vector128.Create(ctrl);
Vector128<byte> bucketCtrl = Sse2.LoadVector128(
(byte*)&bucket->Control);
// === Phase 1: Look up an existing key ===
Vector128<byte> 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;
}
}
}
}
src/Nitload.Common/Collections/SwissTable.cs
+52 -151
View File
@@ -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
{
/// <summary>
///
/// </summary>
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<Entry>();
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<nuint> _freeHandle;
private bool _disposed;
public static readonly nuint RequestedNativeMemorySize = (nuint)Unsafe.SizeOf<byte>() + (nuint)Unsafe.SizeOf<M128>();
private byte* _controlBytes;
private SwissTable(
Entry* buffer,
Action<nuint> freeHandle
byte* buffer
)
{
_entries = buffer;
_freeHandle = freeHandle;
_controlBytes = buffer;
}
public static SwissTable Create()
{
Entry* buffer = (Entry*)NativeMemory.AlignedAlloc((nuint)Unsafe.SizeOf<Entry>(), 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<nuint, nuint> allocator,
Action<nuint> 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<nuint> 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.");
}
}
src/Nitload.Common/HighPerformance/Memory/L4.Debug.cs
+2 -1
View File
@@ -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(
src/Nitload.Common/HighPerformance/Memory/L4.Evict.cs
+3 -1
View File
@@ -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>((byte*)(ulong)(head->PositionMask + _baseAddress), (int)(ulong)head->SizeMask));
_bufferAllocator.Free(index);
src/Nitload.Common/HighPerformance/Memory/L4.Prefetch.cs
+4 -3
View File
@@ -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>((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);
}
src/Nitload.Common/HighPerformance/Memory/L4.cs
+1 -1
View File
@@ -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;
src/Nitload.Common/HighPerformance/ProcessorAbstraction/LogicalCore.cs
+9 -10
View File
@@ -16,22 +16,21 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction
public readonly struct LogicalCore
{
public readonly uint GlobalCpuId;
public readonly uint NumaGroupId;
/// <summary>
/// Stays 0 on linux.
/// </summary>
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;
}
src/Nitload.Common/HighPerformance/ProcessorAbstraction/Numa.cs
+49 -17
View File
@@ -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<ulong> CpuIndexInNuma,
Span<uint> globalCpuIdInNuma
Span<uint> processorGroupIds,
Span<uint> socketIds,
Span<uint> cpuIndices,
Span<uint> 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<int> 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<uint> globalIndex = stackalloc uint[count];
_ = GetNumaEach(numaId, null, globalIndex);
return new(numaId, count, globalIndex);
Span<uint> processorGroupIds = stackalloc uint[count];
Span<uint> socketIds = stackalloc uint[count];
Span<uint> cpuIndices = stackalloc uint[count];
Span<uint> 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")]
src/Nitload.Common/HighPerformance/ProcessorAbstraction/NumaNode.cs
+12 -2
View File
@@ -22,16 +22,26 @@ namespace Nitload.Common.HighPerformance.ProcessorAbstraction
internal NumaNode(
uint numaId,
int count,
Span<uint> logicalCoreIds
Span<uint> processorGroupIds,
Span<uint> socketIds,
Span<uint> cpuIndices,
Span<uint> 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<LogicalCore> LogicalCores => _logicalCores;
src/Nitload.Common/Threading/ThreadHelper.cs
+3 -3
View File
@@ -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
src/Nitload.Common/Timing/StgSharpTime.cs
+1 -1
View File
@@ -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;
src/Nitload.Common/Timing/TimeSourceProvider.cs
+4 -4
View File
@@ -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;
src/Nitload.Native/nif-head/NitloadFrameWork.h
+6 -2
View File
@@ -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
src/Nitload.Native/nif-src/thread.linux.c
+56 -16
View File
@@ -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);
}
src/Nitload.Native/nif-src/thread.windows.c
+135 -36
View File
@@ -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(), &current);
/* 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;
}
test/STgSharpTest/UnitTest1.cs
+19 -2
View File
@@ -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<NotImplementedException>(
() => SwissTable.Reader.TryGet(table, key, out _));
Assert.ThrowsException<NotImplementedException>(
() => SwissTable.Writer.TryAdd(table, key, 100));
Assert.ThrowsException<NotImplementedException>(
() => SwissTable.Writer.TrySet(table, key, 100));
Assert.ThrowsException<NotImplementedException>(
() => SwissTable.Writer.TryRemove(table, key, out _));
}
[TestMethod]
public void RegexTest()
{
test/StgSharpDebug/TestSwiss.cs
+2 -2
View File
@@ -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