diff --git a/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs b/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs new file mode 100644 index 000000000..7a7609657 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs @@ -0,0 +1,118 @@ +using System; +using System.Collections.Generic; +using Xunit; + +namespace Server.Tests; + +public class ShadowDictionaryEntriesTests +{ + private class TestEntity : ISerializable + { + public TestEntity(Serial serial) => Serial = serial; + + public Serial Serial { get; } + public DateTime Created { get; set; } + public bool Deleted => false; + public byte SerializedThread { get; set; } + public int SerializedPosition { get; set; } + public int SerializedLength { get; set; } + + public void Delete() + { + } + + public void Serialize(IGenericWriter writer) + { + writer.Write(Serial); + writer.Write(0xC0FFEE); + } + + public void Deserialize(IGenericReader reader) + { + } + } + + [Fact] + public void RuntimeLayoutIsSupported() + { + // If this fails on a runtime upgrade, saves still work via the fallback path, + // but the parallel iteration fast path is silently lost — this test makes it loud. + Assert.True(ShadowDictionaryEntries.Supported); + } + + [Fact] + public void SerializeRangeCoversExactlyTheLiveEntities() + { + var persistence = new GenericEntityPersistence("ShadowTest", 1000, 1, 0x7FFFFFFF); + + try + { + var rng = new System.Random(0xBEEF); + var dict = persistence.EntitiesBySerial; + + // Heavy churn: adds, removes, and re-adds to exercise freelist reuse and resizes, + // leaving free slots scattered through the entries array. + var serials = new List(); + for (var i = 0; i < 50_000; i++) + { + var serial = (Serial)(uint)rng.Next(1, int.MaxValue); + if (dict.TryAdd(serial, new TestEntity(serial))) + { + serials.Add(serial); + } + + if (i % 4 == 3) + { + var index = rng.Next(serials.Count); + dict.Remove(serials[index]); + serials.RemoveAt(index); + } + } + + Assert.True(persistence.TrySnapshotEntries(out var slotCount)); + Assert.True(slotCount >= dict.Count); + + var source = (ISlotRangeSource)persistence; + var writer = new BufferWriter(new byte[dict.Count * 16], true); + + // Serialize in worker-sized slices, like the drain does. + var serialized = 0; + for (var offset = 0; offset < slotCount; offset += 4096) + { + serialized += source.SerializeRange(writer, 3, offset, Math.Min(4096, slotCount - offset)); + } + + Assert.Equal(dict.Count, serialized); + + // Every live entity was stamped exactly once with a coherent span. + foreach (var (serial, entity) in dict) + { + Assert.Equal(3, entity.SerializedThread); + Assert.Equal(8, entity.SerializedLength); // serial + int + + var span = writer.Buffer.AsSpan(entity.SerializedPosition, entity.SerializedLength); + Assert.Equal(serial, (Serial)System.Buffers.Binary.BinaryPrimitives.ReadUInt32LittleEndian(span)); + } + } + finally + { + persistence.Unregister(); + } + } + + [Fact] + public void SnapshotFailsGracefullyOnEmptyDictionary() + { + var persistence = new GenericEntityPersistence("ShadowTestEmpty", 1001, 1, 0x7FFFFFFF); + + try + { + Assert.False(persistence.TrySnapshotEntries(out var slotCount)); + Assert.Equal(0, slotCount); + } + finally + { + persistence.Unregister(); + } + } +} diff --git a/Projects/Server/Serialization/GenericEntityPersistence.cs b/Projects/Server/Serialization/GenericEntityPersistence.cs index c28fd6a0c..2b592d35e 100644 --- a/Projects/Server/Serialization/GenericEntityPersistence.cs +++ b/Projects/Server/Serialization/GenericEntityPersistence.cs @@ -31,10 +31,20 @@ public interface IGenericEntityPersistence public void DeserializeIndexes(string savePath, Dictionary typesDb); } -public class GenericEntityPersistence : GenericPersistence, IGenericEntityPersistence where T : class, ISerializable +public class GenericEntityPersistence : GenericPersistence, IGenericEntityPersistence, ISlotRangeSource + where T : class, ISerializable { private static readonly ILogger logger = LogFactory.GetLogger(typeof(GenericEntityPersistence)); + // Layout-validated direct access to EntitiesBySerial's entries array, letting workers + // iterate the dictionary in parallel during saves. Null when unsupported on this runtime. + private static readonly FieldInfo _entriesField = + ShadowDictionaryEntries.Supported ? ShadowDictionaryEntries.GetEntriesField() : null; + + // The entries array captured at freeze time. The dictionary cannot mutate while saving + // (adds/removes divert to the pending queues), so the array is stable until released. + private object _entriesSnapshot; + // Support legacy split file serialization private static Dictionary>> _entities; @@ -155,6 +165,15 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer // Self-payload first so a large one overlaps the entity stream instead of ending it. World.PushSingleToCache(this); + // Fast path: publish slot ranges of the dictionary's entries array so the workers + // iterate it directly in parallel — the main thread never touches the entities. + if (TrySnapshotEntries(out var slotCount)) + { + World.PushSlotRangesToCache(this, slotCount); + return; + } + + // Fallback: enumerate and hand off every entity from the main thread. foreach (var entity in EntitiesBySerial.Values) { // Previous save's length is the cost estimate; 0 (new entity or first save) takes @@ -171,6 +190,44 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer } } + internal bool TrySnapshotEntries(out int slotCount) + { + if (_entriesField != null && EntitiesBySerial.Count > 0 && + _entriesField.GetValue(EntitiesBySerial) is Array entries) + { + _entriesSnapshot = entries; + slotCount = entries.Length; + return true; + } + + slotCount = 0; + return false; + } + + int ISlotRangeSource.SerializeRange(BufferWriter writer, byte threadIndex, int offset, int count) + { + // Layout proven at startup by ShadowDictionaryEntries.Supported; ranges are produced + // from the same array's length, so every read is in-bounds. + var entries = Unsafe.As[]>(_entriesSnapshot); + var serialized = 0; + + ref var entry = ref Unsafe.Add(ref MemoryMarshal.GetArrayDataReference(entries), offset); + + for (var i = 0; i < count; i++, entry = ref Unsafe.Add(ref entry, 1)) + { + // Occupied slots are exactly the non-null values: Dictionary clears reference + // values on remove, and never-used capacity is zero-initialized. + var entity = entry.Value; + if (entity != null) + { + SerializationThreadWorker.Serialize(entity, writer, threadIndex); + serialized++; + } + } + + return serialized; + } + private static ConstructorInfo GetConstructorFor(string typeName, Type t, Type[] constructorTypes) { if (t?.IsAbstract != false) @@ -495,6 +552,8 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer public override void PostWorldSave() { + // Release the snapshot so a between-saves resize doesn't pin the old array. + _entriesSnapshot = null; ProcessSafetyQueues(); } diff --git a/Projects/Server/Serialization/SerializationChunkSource.cs b/Projects/Server/Serialization/SerializationChunkSource.cs index 0b0af73cd..141200519 100644 --- a/Projects/Server/Serialization/SerializationChunkSource.cs +++ b/Projects/Server/Serialization/SerializationChunkSource.cs @@ -20,6 +20,21 @@ using System.Runtime.InteropServices; namespace Server; +/// +/// A range of backing-store slots that a serialization worker can serialize directly, +/// letting workers iterate a persistence's storage in parallel instead of the main thread +/// enumerating and handing off every entity. Implemented by +/// over its dictionary's entries array. +/// +public interface ISlotRangeSource +{ + /// + /// Serializes every occupied slot in [offset, offset + count) into the writer, + /// stamping each entity's thread/position/length. Returns the number serialized. + /// + int SerializeRange(BufferWriter writer, byte threadIndex, int offset, int count); +} + /// /// Single-producer/multi-consumer handoff between the game loop and the serialization /// thread workers during a world save. The producer batches entities into pooled chunks @@ -29,6 +44,8 @@ namespace Server; /// Entities whose previous serialized size exceeds are /// published as dedicated single-entity chunks so multi-megabyte payloads spread across /// workers instead of riding inside one chunk. +/// Persistences that support direct parallel iteration publish slot ranges instead of +/// filled chunks, removing the per-entity handoff from the freeze entirely. /// public sealed class SerializationChunkSource { @@ -47,21 +64,28 @@ public sealed class SerializationChunkSource { public readonly IGenericSerializable Single; public readonly IGenericSerializable[] Buffer; + public readonly ISlotRangeSource Source; + public readonly int Offset; public readonly int Count; public Chunk(IGenericSerializable single) { Single = single; - Buffer = null; Count = 1; } public Chunk(IGenericSerializable[] buffer, int count) { - Single = null; Buffer = buffer; Count = count; } + + public Chunk(ISlotRangeSource source, int offset, int count) + { + Source = source; + Offset = offset; + Count = count; + } } private readonly ConcurrentQueue _chunks = new(); @@ -96,6 +120,19 @@ public sealed class SerializationChunkSource /// public void PushSingle(IGenericSerializable entity) => _chunks.Enqueue(new Chunk(entity)); + /// + /// Publishes slot ranges covering [0, slotCount) of a directly-iterable persistence. + /// Workers claim ranges like any other chunk, so the per-entity handoff cost disappears + /// and load balancing is unchanged. + /// + public void PushSlotRanges(ISlotRangeSource source, int slotCount) + { + for (var offset = 0; offset < slotCount; offset += ChunkCapacity) + { + _chunks.Enqueue(new Chunk(source, offset, Math.Min(ChunkCapacity, slotCount - offset))); + } + } + /// /// Publishes the partial chunk, if any. Must be called on the producer thread before /// the workers are told to finish draining, or the tail of the stream is not serialized. diff --git a/Projects/Server/Serialization/SerializationThreadWorker.cs b/Projects/Server/Serialization/SerializationThreadWorker.cs index 4559bfbdb..c4c6ed339 100644 --- a/Projects/Server/Serialization/SerializationThreadWorker.cs +++ b/Projects/Server/Serialization/SerializationThreadWorker.cs @@ -36,16 +36,33 @@ public class SerializationThreadWorker private long _bytesSerialized; public SerializationThreadWorker(int index, SerializationChunkSource chunkSource, int heapSizeHint = 0) + : this(index, chunkSource, heapSizeHint, inline: false) + { + } + + private SerializationThreadWorker(int index, SerializationChunkSource chunkSource, int heapSizeHint, bool inline) { _index = index; _chunkSource = chunkSource; _heapSizeHint = heapSizeHint; - _startEvent = new AutoResetEvent(false); - _stopEvent = new AutoResetEvent(false); - _thread = new Thread(Execute); - _thread.Start(this); + + if (!inline) + { + _startEvent = new AutoResetEvent(false); + _stopEvent = new AutoResetEvent(false); + _thread = new Thread(Execute); + _thread.Start(this); + } } + /// + /// Creates a worker with no thread of its own. The owner drains chunks inline via + /// — used by the main thread to join the drain instead of + /// idling while the thread workers finish. + /// + public static SerializationThreadWorker CreateInline(int index, SerializationChunkSource chunkSource, int heapSizeHint = 0) => + new(index, chunkSource, heapSizeHint, inline: true); + // Stats from the most recent save, for diagnosing load balance. public long EntitiesSerialized => _entitiesSerialized; public long BytesSerialized => _bytesSerialized; @@ -82,7 +99,7 @@ public class SerializationThreadWorker public ReadOnlySpan GetHeap(int start, int length) => _heap.AsSpan(start, length); [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static void Serialize(IGenericSerializable e, BufferWriter writer, byte threadIndex) + internal static void Serialize(IGenericSerializable e, BufferWriter writer, byte threadIndex) { e.SerializedThread = threadIndex; var start = e.SerializedPosition = (int)writer.Position; @@ -90,6 +107,56 @@ public class SerializationThreadWorker e.SerializedLength = (int)(writer.Position - start); } + private static long ProcessChunk( + in SerializationChunkSource.Chunk chunk, SerializationChunkSource chunkSource, + BufferWriter writer, byte threadIndex + ) + { + if (chunk.Single != null) + { + Serialize(chunk.Single, writer, threadIndex); + return 1; + } + + if (chunk.Source != null) + { + return chunk.Source.SerializeRange(writer, threadIndex, chunk.Offset, chunk.Count); + } + + var buffer = chunk.Buffer; + var count = chunk.Count; + for (var i = 0; i < count; i++) + { + Serialize(buffer[i], writer, threadIndex); + } + + chunkSource.Return(buffer, count); + return count; + } + + /// + /// Drains chunks on the calling thread until the queue is empty, then returns. + /// Only valid on inline workers; the main thread calls this after publishing all work + /// so it contributes drain throughput instead of idling. + /// + public void DrainInline() + { + var writer = new BufferWriter(_heap, true, World.SerializedTypes); + var threadIndex = (byte)_index; + var entities = 0L; + + while (_chunkSource.TryTake(out var chunk)) + { + entities += ProcessChunk(in chunk, _chunkSource, writer, threadIndex); + } + + _heap = writer.Buffer; + _entitiesSerialized = entities; + _bytesSerialized = writer.Position; + + writer.Close(); + } + private static void Execute(object obj) { var worker = (SerializationThreadWorker)obj; @@ -110,24 +177,7 @@ public class SerializationThreadWorker if (chunkSource.TryTake(out var chunk)) { spinner.Reset(); - - if (chunk.Single != null) - { - Serialize(chunk.Single, writer, threadIndex); - entities++; - } - else - { - var buffer = chunk.Buffer; - var count = chunk.Count; - for (var i = 0; i < count; i++) - { - Serialize(buffer[i], writer, threadIndex); - } - - entities += count; - chunkSource.Return(buffer, count); - } + entities += ProcessChunk(in chunk, chunkSource, writer, threadIndex); } else if (pauseRequested) // Break when finished { diff --git a/Projects/Server/Serialization/ShadowDictionaryEntries.cs b/Projects/Server/Serialization/ShadowDictionaryEntries.cs new file mode 100644 index 000000000..170349f2a --- /dev/null +++ b/Projects/Server/Serialization/ShadowDictionaryEntries.cs @@ -0,0 +1,197 @@ +/************************************************************************* + * ModernUO * + * Copyright 2019-2026 - ModernUO Development Team * + * Email: hi@modernuo.com * + * File: ShadowDictionaryEntries.cs * + * * + * This program is free software: you can redistribute it and/or modify * + * it under the terms of the GNU General Public License as published by * + * the Free Software Foundation, either version 3 of the License, or * + * (at your option) any later version. * + * * + * You should have received a copy of the GNU General Public License * + * along with this program. If not, see . * + *************************************************************************/ + +using System; +using System.Collections.Generic; +using System.Reflection; +using System.Runtime.CompilerServices; +using System.Runtime.InteropServices; +using Server.Logging; + +namespace Server; + +/// +/// Mirrors the field layout of Dictionary<Serial, TValue>.Entry so serialization +/// workers can scan a dictionary's backing entries array directly, in parallel, without the +/// main thread enumerating and handing off every entity. +/// The CLR's auto-layout algorithm is deterministic for identical field sequences, so this +/// struct lays out identically to the runtime's private Entry struct — and +/// proves that empirically at startup before +/// any code reads through it. If validation fails on a future runtime, callers fall back to +/// the enumerate-and-push path. +/// A free or never-used slot always has a null value (Dictionary clears values of +/// reference-type TValue on remove to release references), so occupancy is exactly +/// Value != null. +/// +internal struct ShadowEntry +{ + // Field order must mirror S.P.CoreLib Dictionary.Entry: hashCode, next, key, value + public uint HashCode; + public int Next; + public Serial Key; + public TValue Value; +} + +internal static class ShadowDictionaryEntries +{ + private static readonly ILogger logger = LogFactory.GetLogger(typeof(ShadowDictionaryEntries)); + + /// + /// True when this runtime's Dictionary entry layout matches , + /// proven by validation at startup. All reference-type TValue instantiations share one + /// canonical layout, so a single validation covers every entity dictionary. + /// + internal static readonly bool Supported = Validate(); + + internal static FieldInfo GetEntriesField() where TValue : class => + typeof(Dictionary).GetField("_entries", BindingFlags.NonPublic | BindingFlags.Instance); + + private sealed class ValidationValue + { + public int Id; + } + + private static bool Validate() + { + try + { + var field = GetEntriesField(); + if (field == null) + { + logger.Warning("Dictionary._entries not found; parallel save iteration disabled."); + return false; + } + + // A compacting GC between snapshotting reference bits and scanning them can only + // produce a false negative, so retry a few times before falling back. + for (var attempt = 0; attempt < 3; attempt++) + { + if (RunValidationPass(field)) + { + return true; + } + } + + logger.Warning("Dictionary entry layout mismatch; parallel save iteration disabled."); + return false; + } + catch (Exception e) + { + logger.Warning(e, "Dictionary entry layout validation failed; parallel save iteration disabled."); + return false; + } + } + + /// + /// Measures the true element stride of the runtime's Entry struct by allocating a large + /// array of it and reading the precise allocated byte count. Verifying the stride first + /// guarantees every subsequent shadow read is in-bounds of the entries array even if the + /// field layout were wrong. + /// + private static bool StrideMatches(Type entryType) + { + const int probeLength = 64 * 1024; + const int arrayHeaderSize = 24; // 64-bit: sync block + method table + length + padding + + // Warm the reflection/allocation path so the measured delta contains only the probe. + Array.CreateInstance(entryType, 1); + + var before = GC.GetAllocatedBytesForCurrentThread(); + var probe = Array.CreateInstance(entryType, probeLength); + var delta = GC.GetAllocatedBytesForCurrentThread() - before; + GC.KeepAlive(probe); + + // Alignment slack is < 8 bytes on a 512KB+ allocation, so integer division is exact. + var actualStride = (delta - arrayHeaderSize) / probeLength; + + return actualStride == Unsafe.SizeOf>(); + } + + /// + /// Proves the layout without ever materializing a managed reference through the shadow + /// view: value slots are compared as raw pointer bits (nint). Only after the stride and + /// every key and value-pointer of a churned, resized, freelist-exercised dictionary match + /// is the layout trusted for typed reads. + /// + private static bool RunValidationPass(FieldInfo field) + { + const int count = 1000; + + var dict = new Dictionary(); + var rng = new System.Random(0x5EED); + var inserted = new List(count); + + // Adds with interleaved removes and re-adds: exercises resizes and freelist reuse so + // freed slots (which production skips via null values) are present in the array. + for (var i = 0; i < count; i++) + { + var serial = (Serial)(uint)rng.Next(1, int.MaxValue); + if (dict.TryAdd(serial, new ValidationValue { Id = i })) + { + inserted.Add(serial); + } + + if (i % 3 == 2) + { + var victim = inserted[rng.Next(inserted.Count)]; + if (dict.Remove(victim)) + { + inserted.Remove(victim); + } + } + } + + if (field.GetValue(dict) is not Array entriesObj || !StrideMatches(entriesObj.GetType().GetElementType()!)) + { + return false; + } + + // Snapshot expected key -> value-pointer-bits pairs before scanning, so no allocation + // happens between reading the real references and reading the shadow view. + var expected = new Dictionary(dict.Count); + foreach (var (key, value) in dict) + { + var v = value; + expected[key] = Unsafe.As(ref v); + } + + var entries = Unsafe.As[]>(entriesObj); + var length = entriesObj.Length; + var matched = 0; + + ref var entry = ref MemoryMarshal.GetArrayDataReference(entries); + + for (var i = 0; i < length; i++, entry = ref Unsafe.Add(ref entry, 1)) + { + // Read the value slot as pointer bits only — never as a reference — so a wrong + // field offset cannot fabricate a managed reference for the GC to trip over. + var bits = Unsafe.As(ref entry.Value); + + if (bits == 0) + { + continue; // free or never-used slot + } + + if (!expected.TryGetValue(entry.Key, out var expectedBits) || bits != expectedBits) + { + return false; + } + + matched++; + } + + return matched == dict.Count; + } +} diff --git a/Projects/Server/World/World.cs b/Projects/Server/World/World.cs index 9df761fa4..4ded72fb1 100644 --- a/Projects/Server/World/World.cs +++ b/Projects/Server/World/World.cs @@ -44,7 +44,10 @@ public static class World private static readonly MobilePersistence _mobilePersistence = new(); private static readonly GenericEntityPersistence _guildPersistence = new("Guilds", 3, 1, 0x7FFFFFFF); + // All workers including the main thread's inline worker (last index), for heap lookups. internal static SerializationThreadWorker[] _threadWorkers; + // How many of those own a thread (wake/sleep/exit applies only to these). + private static int _realWorkerCount; private static readonly SerializationChunkSource _chunkSource = new(); private static readonly ManualResetEvent _diskWriteHandle = new(true); @@ -195,18 +198,22 @@ public static class World watch.Elapsed.TotalSeconds ); - // Create the serialization threads. - var threadCount = UseMultiThreadedSaves ? Math.Max(Environment.ProcessorCount - 1, 1) : 1; - _threadWorkers = new SerializationThreadWorker[threadCount]; + // Create the serialization threads, plus an inline worker the main thread uses to + // join the drain after publishing work instead of idling until the workers finish. + _realWorkerCount = UseMultiThreadedSaves ? Math.Max(Environment.ProcessorCount - 1, 1) : 1; + _threadWorkers = new SerializationThreadWorker[_realWorkerCount + 1]; // The save we just loaded tells us how big the heaps need to be, so the first save // doesn't pay copy-on-grow inside the freeze window. 25% headroom for world growth. - var heapSizeHint = (int)Math.Min(GetLoadedSaveSize() / threadCount * 5 / 4, 1024 * 1024 * 1024); + var heapSizeHint = (int)Math.Min(GetLoadedSaveSize() / _threadWorkers.Length * 5 / 4, 1024 * 1024 * 1024); - for (var i = 0; i < _threadWorkers.Length; i++) + for (var i = 0; i < _realWorkerCount; i++) { _threadWorkers[i] = new SerializationThreadWorker(i, _chunkSource, heapSizeHint); } + + _threadWorkers[_realWorkerCount] = + SerializationThreadWorker.CreateInline(_realWorkerCount, _chunkSource, heapSizeHint); } private static long GetLoadedSaveSize() @@ -446,7 +453,7 @@ public static class World [MethodImpl(MethodImplOptions.AggressiveInlining)] internal static void WakeSerializationThreads() { - for (var i = 0; i < _threadWorkers.Length; i++) + for (var i = 0; i < _realWorkerCount; i++) { _threadWorkers[i].Wake(); } @@ -457,7 +464,10 @@ public static class World // Publish the partial chunk before the workers are told to finish draining. _chunkSource.Flush(); - for (var i = 0; i < _threadWorkers.Length; i++) + // Join the drain: the main thread would otherwise idle here while workers finish. + _threadWorkers[_realWorkerCount].DrainInline(); + + for (var i = 0; i < _realWorkerCount; i++) { _threadWorkers[i].Sleep(); } @@ -469,9 +479,13 @@ public static class World [MethodImpl(MethodImplOptions.AggressiveInlining)] internal static void PushSingleToCache(IGenericSerializable e) => _chunkSource.PushSingle(e); + [MethodImpl(MethodImplOptions.AggressiveInlining)] + internal static void PushSlotRangesToCache(ISlotRangeSource source, int slotCount) => + _chunkSource.PushSlotRanges(source, slotCount); + public static void ExitSerializationThreads() { - for (var i = 0; i < _threadWorkers.Length; i++) + for (var i = 0; i < _realWorkerCount; i++) { _threadWorkers[i]?.Exit(); }