diff --git a/Projects/Server.Tests/Tests/Maps/ClientEnumeratorTests.cs b/Projects/Server.Tests/Tests/Maps/ClientEnumeratorTests.cs index 82722e3df..bfbc18e94 100644 --- a/Projects/Server.Tests/Tests/Maps/ClientEnumeratorTests.cs +++ b/Projects/Server.Tests/Tests/Maps/ClientEnumeratorTests.cs @@ -443,9 +443,6 @@ public class ClientEnumeratorTests public DateTime Created { get; set; } public Serial Serial { get; } public void Deserialize(IGenericReader reader) => throw new NotImplementedException(); - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } public void Serialize(IGenericWriter writer) => throw new NotImplementedException(); public bool Deleted { get; } public void Delete() => throw new NotImplementedException(); diff --git a/Projects/Server.Tests/Tests/Network/Packets/Outgoing/AccountPacketTests.cs b/Projects/Server.Tests/Tests/Network/Packets/Outgoing/AccountPacketTests.cs index 3f62c01bf..6d8bd9fd9 100644 --- a/Projects/Server.Tests/Tests/Network/Packets/Outgoing/AccountPacketTests.cs +++ b/Projects/Server.Tests/Tests/Network/Packets/Outgoing/AccountPacketTests.cs @@ -52,10 +52,6 @@ public class AccountPacketTests public Serial Serial { get; } public void Deserialize(IGenericReader reader) => throw new NotImplementedException(); - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - public void Serialize(IGenericWriter writer) => throw new NotImplementedException(); public bool Deleted { get; } diff --git a/Projects/Server.Tests/Tests/Serialization/BufferWriterTests.cs b/Projects/Server.Tests/Tests/Serialization/BufferWriterTests.cs new file mode 100644 index 000000000..bb6e5aa70 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/BufferWriterTests.cs @@ -0,0 +1,310 @@ +using System; +using System.Buffers.Binary; +using System.IO; +using System.Linq; +using System.Text; +using Xunit; + +namespace Server.Tests; + +/// +/// Pins BufferWriter's byte-level output and position semantics so the write path can be +/// optimized without behavioral drift. +/// +public class BufferWriterTests +{ + [Fact] + public void PrimitivesAreLittleEndianAtExpectedOffsets() + { + var writer = new BufferWriter(new byte[256], true); + + writer.Write((byte)0xAB); + writer.Write((sbyte)-5); + writer.Write(true); + writer.Write(false); + writer.Write((short)-12345); + writer.Write((ushort)54321); + writer.Write(-123456789); + writer.Write(3123456789u); + writer.Write(-1234567890123456789L); + writer.Write(12345678901234567890UL); + writer.Write(1234.5678d); + writer.Write(56.75f); + writer.Write((Serial)0x40000001u); + + Assert.Equal(1 + 1 + 1 + 1 + 2 + 2 + 4 + 4 + 8 + 8 + 8 + 4 + 4, writer.Position); + + var b = writer.Buffer; + Assert.Equal(0xAB, b[0]); + Assert.Equal(unchecked((byte)-5), b[1]); + Assert.Equal(1, b[2]); + Assert.Equal(0, b[3]); + Assert.Equal(-12345, BinaryPrimitives.ReadInt16LittleEndian(b.AsSpan(4))); + Assert.Equal(54321, BinaryPrimitives.ReadUInt16LittleEndian(b.AsSpan(6))); + Assert.Equal(-123456789, BinaryPrimitives.ReadInt32LittleEndian(b.AsSpan(8))); + Assert.Equal(3123456789u, BinaryPrimitives.ReadUInt32LittleEndian(b.AsSpan(12))); + Assert.Equal(-1234567890123456789L, BinaryPrimitives.ReadInt64LittleEndian(b.AsSpan(16))); + Assert.Equal(12345678901234567890UL, BinaryPrimitives.ReadUInt64LittleEndian(b.AsSpan(24))); + Assert.Equal(1234.5678d, BinaryPrimitives.ReadDoubleLittleEndian(b.AsSpan(32))); + Assert.Equal(56.75f, BinaryPrimitives.ReadSingleLittleEndian(b.AsSpan(40))); + Assert.Equal(0x40000001u, BinaryPrimitives.ReadUInt32LittleEndian(b.AsSpan(44))); + } + + [Fact] + public void SeekEndUsesHighWaterMarkNotCurrentPosition() + { + var writer = new BufferWriter(new byte[256], true); + + writer.Write(1L); + writer.Write(2L); + writer.Write(3L); // high water = 24 + + writer.Seek(4, SeekOrigin.Begin); + writer.Write(99); // position now 8, high water still 24 + + Assert.Equal(24, writer.Seek(0, SeekOrigin.End)); + Assert.Equal(24, writer.Position); + } + + [Fact] + public void SeekCurrentAndBeginBehave() + { + var writer = new BufferWriter(new byte[64], true); + + writer.Write(0xDEADBEEF); + Assert.Equal(2, writer.Seek(2, SeekOrigin.Begin)); + Assert.Equal(3, writer.Seek(1, SeekOrigin.Current)); + + writer.Write((byte)0x77); + Assert.Equal(0x77, writer.Buffer[3]); + } + + [Fact] + public void GrowthPreservesContentAndPosition() + { + var writer = new BufferWriter(new byte[16], true); + + for (var i = 0; i < 100; i++) + { + writer.Write((long)i); + } + + Assert.Equal(800, writer.Position); + Assert.True(writer.Buffer.Length >= 800); + + for (var i = 0; i < 100; i++) + { + Assert.Equal(i, BinaryPrimitives.ReadInt64LittleEndian(writer.Buffer.AsSpan(i * 8))); + } + } + + [Fact] + public void SpanWriteCrossesGrowthBoundary() + { + var writer = new BufferWriter(new byte[8], true); + + Span payload = stackalloc byte[64]; + for (var i = 0; i < payload.Length; i++) + { + payload[i] = (byte)(i + 1); + } + + writer.Write((ushort)7); + writer.Write(payload); + + Assert.Equal(66, writer.Position); + Assert.Equal(payload.ToArray(), writer.Buffer.AsSpan(2, 64).ToArray()); + } + + [Theory] + [InlineData(0, new byte[] { 0x00 })] + [InlineData(127, new byte[] { 0x7F })] + [InlineData(128, new byte[] { 0x80, 0x01 })] + [InlineData(0x3FFF, new byte[] { 0xFF, 0x7F })] + [InlineData(0x4000, new byte[] { 0x80, 0x80, 0x01 })] + [InlineData(0x1F_FFFF, new byte[] { 0xFF, 0xFF, 0x7F })] + [InlineData(0x20_0000, new byte[] { 0x80, 0x80, 0x80, 0x01 })] + [InlineData(0xFFF_FFFF, new byte[] { 0xFF, 0xFF, 0xFF, 0x7F })] + [InlineData(0x1000_0000, new byte[] { 0x80, 0x80, 0x80, 0x80, 0x01 })] + [InlineData(int.MaxValue, new byte[] { 0xFF, 0xFF, 0xFF, 0xFF, 0x07 })] + [InlineData(-1, new byte[] { 0xFF, 0xFF, 0xFF, 0xFF, 0x0F })] + public void EncodedIntMatchesFormat(int value, byte[] expected) + { + var writer = new BufferWriter(new byte[16], true); + + ((IGenericWriter)writer).WriteEncodedInt(value); + + Assert.Equal(expected.Length, writer.Position); + Assert.Equal(expected, writer.Buffer.AsSpan(0, expected.Length).ToArray()); + } + + [Fact] + public void PrefixedStringsWriteFlagLengthAndUtf8() + { + var writer = new BufferWriter(new byte[256], true); + + writer.Write("héllo Ωorld"); + var utf8 = Encoding.UTF8.GetBytes("héllo Ωorld"); + + var b = writer.Buffer; + Assert.Equal(1, b[0]); // not-null flag + Assert.Equal(utf8.Length, b[1]); // encoded length (small string = 1 byte) + Assert.Equal(utf8, b.AsSpan(2, utf8.Length).ToArray()); + Assert.Equal(2 + utf8.Length, writer.Position); + + writer.Write((string)null); + Assert.Equal(0, b[2 + utf8.Length]); // null flag + } + + [Theory] + [InlineData(84)] // scratch path + [InlineData(85)] // scratch path boundary + [InlineData(86)] // two-pass path + [InlineData(300)] // two-pass, length prefix > 1 byte + public void StringPathsAgreeAcrossTheScratchBoundary(int chars) + { + // Mixed ASCII, 2-byte, 3-byte, and surrogate-pair (4-byte) content + var builder = new StringBuilder(chars); + for (var i = 0; builder.Length < chars; i++) + { + switch (i % 4) + { + case 0: + builder.Append('a'); + break; + case 1: + builder.Append('é'); + break; + case 2: + builder.Append('Ω'); + break; + default: + if (builder.Length + 2 <= chars) + { + builder.Append("𝔘"); // surrogate pair + } + else + { + builder.Append('z'); + } + break; + } + } + + var value = builder.ToString(); + Assert.Equal(chars, value.Length); + + var writer = new BufferWriter(new byte[16], true); // forces growth through both paths + writer.Write(value); + + var utf8 = Encoding.UTF8.GetBytes(value); + var b = writer.Buffer; + Assert.Equal(1, b[0]); + + // decode the 7-bit encoded length prefix + var offset = 1; + var length = 0; + var shift = 0; + byte current; + do + { + current = b[offset++]; + length |= (current & 0x7F) << shift; + shift += 7; + } while ((current & 0x80) != 0); + + Assert.Equal(utf8.Length, length); + Assert.Equal(utf8, b.AsSpan(offset, length).ToArray()); + Assert.Equal(offset + length, writer.Position); + } + + [Fact] + public void DateTimeWritesUtcTicksViaInterface() + { + var writer = new BufferWriter(new byte[64], true); + IGenericWriter iface = writer; + + var utc = new DateTime(2026, 7, 13, 1, 2, 3, DateTimeKind.Utc); + var local = utc.ToLocalTime(); + + iface.Write(utc); + iface.Write(local); // must convert to UTC + + Assert.Equal(utc.Ticks, BinaryPrimitives.ReadInt64LittleEndian(writer.Buffer.AsSpan(0))); + Assert.Equal(utc.Ticks, BinaryPrimitives.ReadInt64LittleEndian(writer.Buffer.AsSpan(8))); + } + + [Fact] + public void Point3DWritesThreeInts() + { + var writer = new BufferWriter(new byte[64], true); + IGenericWriter iface = writer; + + iface.Write(new Point3D(100, -200, 30)); + + Assert.Equal(12, writer.Position); + Assert.Equal(100, BinaryPrimitives.ReadInt32LittleEndian(writer.Buffer.AsSpan(0))); + Assert.Equal(-200, BinaryPrimitives.ReadInt32LittleEndian(writer.Buffer.AsSpan(4))); + Assert.Equal(30, BinaryPrimitives.ReadInt32LittleEndian(writer.Buffer.AsSpan(8))); + } + + [Fact] + public void DecimalRoundTripsThroughReader() + { + var writer = new BufferWriter(new byte[64], true); + writer.Write(1234567.89012m); + + IGenericReader reader = new BufferReader(writer.Buffer); + Assert.Equal(1234567.89012m, reader.ReadDecimal()); + } + + [Fact] + public void LongStringPrefixIsZeroPaddedToWorstCaseWidth() + { + // 100 ASCII chars: worst case 300 bytes -> 2-byte prefix; actual 100 bytes would + // canonically fit in 1. The prefix must be the non-minimal 2-byte form the readers + // decode identically: (100 | 0x80), 0x00. + var value = new string('a', 100); + var writer = new BufferWriter(new byte[1024], false); + writer.WriteRaw(value); + + var b = writer.Buffer; + Assert.Equal((byte)(100 | 0x80), b[0]); + Assert.Equal(0, b[1]); + Assert.Equal(2 + 100, writer.Position); + + IGenericReader reader = new BufferReader(writer.Buffer); + Assert.Equal(value, reader.ReadStringRaw()); + } + + [Theory] + [InlineData(42)] // canonical 1-byte prefix (3 * 42 < 0x80) + [InlineData(100)] // padded prefix + [InlineData(10_000)] // multi-byte prefix, forces growth from a small buffer + public void ThreeBytePerCharContentRoundTrips(int chars) + { + // CJK content encodes at the UTF-8 worst case of 3 bytes per char - the case an + // undersized scratch would truncate or throw on. + var value = new string('二', chars); + var writer = new BufferWriter(new byte[16], true); + writer.Write(value); + + Assert.Equal(Encoding.UTF8.GetByteCount(value), 3 * chars); + + IGenericReader reader = new BufferReader(writer.Buffer); + Assert.Equal(value, reader.ReadString()); + } + + [Fact] + public void SurrogatePairContentRoundTripsThroughRawPath() + { + // Surrogate pairs encode 2 chars into 4 bytes (2 bytes per char) - under the + // 3-bytes-per-char reservation, exercising written < maxLength with a padded prefix. + var value = string.Concat(Enumerable.Repeat("😀", 60)); // 120 chars, 240 bytes + var writer = new BufferWriter(new byte[16], false); + writer.WriteRaw(value); + + IGenericReader reader = new BufferReader(writer.Buffer); + Assert.Equal(value, reader.ReadStringRaw()); + } +} diff --git a/Projects/Server.Tests/Tests/Serialization/FileBufferWriterTests.cs b/Projects/Server.Tests/Tests/Serialization/FileBufferWriterTests.cs new file mode 100644 index 000000000..fcbfb70a0 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/FileBufferWriterTests.cs @@ -0,0 +1,136 @@ +using System; +using System.IO; +using System.Text; +using Xunit; + +namespace Server.Tests; + +public class FileBufferWriterTests +{ + private static string TempFile() => + Path.Combine(Path.GetTempPath(), $"muo-fbw-{Guid.NewGuid():N}.bin"); + + [Fact] + public void DrainsAcrossStagingBoundariesAndPatchesBackwards() + { + var path = TempFile(); + + try + { + // Tiny staging block so every few records cross a drain; mirrors the idx + // pattern: version, count placeholder, records, backwards count patch. + using (var writer = new FileBufferWriter(path, expectedSize: 64)) + { + writer.Write(3); // version + + var countPosition = writer.Position; + writer.Write(0); + + const int records = 1000; + for (var i = 0; i < records; i++) + { + writer.Write((ulong)i * 0x9E3779B97F4A7C15); + writer.Write(i); + } + + var end = writer.Position; + writer.Seek(countPosition, SeekOrigin.Begin); + writer.Write(records); + writer.Seek(0, SeekOrigin.End); + + Assert.Equal(end, writer.Position); + } + + var bytes = File.ReadAllBytes(path); + Assert.Equal(4 + 4 + 1000 * 12, bytes.Length); + + IGenericReader reader = new BufferReader(bytes); + Assert.Equal(3, reader.ReadInt()); + Assert.Equal(1000, reader.ReadInt()); + + for (var i = 0; i < 1000; i++) + { + Assert.Equal((ulong)i * 0x9E3779B97F4A7C15, reader.ReadULong()); + Assert.Equal(i, reader.ReadInt()); + } + } + finally + { + File.Delete(path); + } + } + + [Fact] + public void OversizedSingleItemGrowsTheStagingBlock() + { + var path = TempFile(); + + try + { + // A string whose worst-case reservation exceeds the staging block must grow + // the block instead of deadlocking the drain loop. + var value = new string('二', 500); // 1500 bytes utf8, staging 64 + + using (var writer = new FileBufferWriter(path, expectedSize: 64)) + { + writer.Write(value); + writer.Write(0xC0FFEE); + } + + IGenericReader reader = new BufferReader(File.ReadAllBytes(path)); + Assert.Equal(value, reader.ReadString()); + Assert.Equal(0xC0FFEE, reader.ReadInt()); + } + finally + { + File.Delete(path); + } + } + + [Fact] + public void SpanWritesCrossDrains() + { + var path = TempFile(); + + try + { + var payload = new byte[777]; + new System.Random(0x5EED).NextBytes(payload); + + using (var writer = new FileBufferWriter(path, expectedSize: 64)) + { + writer.Write((ReadOnlySpan)payload); + } + + Assert.Equal(payload, File.ReadAllBytes(path)); + } + finally + { + File.Delete(path); + } + } + + [Fact] + public void TypeWritesUseTagAndHashFormat() + { + var path = TempFile(); + + try + { + using (var writer = new FileBufferWriter(path)) + { + writer.Write(typeof(string)); + writer.Write((Type)null); + } + + var bytes = File.ReadAllBytes(path); + Assert.Equal(1 + 8 + 1, bytes.Length); // flag + hash + null flag + Assert.Equal(2, bytes[0]); + Assert.Equal(0, bytes[^1]); + } + finally + { + File.Delete(path); + } + } +} diff --git a/Projects/Server.Tests/Tests/Serialization/GenericEntityPersistenceRoundTripTests.cs b/Projects/Server.Tests/Tests/Serialization/GenericEntityPersistenceRoundTripTests.cs new file mode 100644 index 000000000..f2114a427 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/GenericEntityPersistenceRoundTripTests.cs @@ -0,0 +1,164 @@ +using System; +using System.Collections.Generic; +using System.IO; +using Xunit; + +namespace Server.Tests; + +internal class RoundTripEntity : ISerializable +{ + public RoundTripEntity(Serial serial) => Serial = serial; + + public Serial Serial { get; } + public DateTime Created { get; set; } = DateTime.UtcNow; + public bool Deleted => false; + + public int Value { get; set; } + public string Name { get; set; } + + public void Delete() + { + } + + public void Serialize(IGenericWriter writer) + { + writer.Write(Value); + writer.Write(Name); + } + + public void Deserialize(IGenericReader reader) + { + Value = reader.ReadInt(); + Name = reader.ReadString(); + } +} + +[Collection("Sequential Server Tests")] +public class GenericEntityPersistenceRoundTripTests +{ + private const uint SelfPayloadMarker = 0xDEADBEEF; + + private class RoundTripPersistence : GenericEntityPersistence + { + public bool SelfPayloadDeserialized { get; private set; } + + public RoundTripPersistence(int priority) : base("RoundTrip", priority, 1, 0x7FFFFFFF) + { + } + + public override void Serialize(IGenericWriter writer) => writer.Write(SelfPayloadMarker); + + public override void Deserialize(IGenericReader reader) + { + Assert.Equal(SelfPayloadMarker, reader.ReadUInt()); + SelfPayloadDeserialized = true; + } + } + + /// + /// Drives the real pipeline end to end: entities serialize on workers via slot-range + /// chunks (plus the persistence self-payload as a single), WriteSnapshot assembles the + /// idx/bin from the per-worker segment logs, and a fresh persistence loads it all back. + /// + [Fact] + public void SnapshotRoundTripsThroughWorkersAndSegmentLogs() + { + // The loader resolves types by hash through AssemblyHandler; make this test + // assembly visible for the duration of the test. + var previousAssemblies = AssemblyHandler.Assemblies; + AssemblyHandler.Assemblies = [.. previousAssemblies ?? [], typeof(RoundTripEntity).Assembly]; + + var source = new SerializationChunkSource(); + var workers = new SerializationThreadWorker[3]; + for (var i = 0; i < workers.Length; i++) + { + workers[i] = new SerializationThreadWorker(i, source); + workers[i].AllocateHeap(); + } + + var previousWorkers = World._threadWorkers; + World._threadWorkers = workers; + + var persistence = new RoundTripPersistence(2000); + RoundTripPersistence loaded = null; + + var dir = Path.Combine(Path.GetTempPath(), $"muo-roundtrip-{Guid.NewGuid():N}"); + Directory.CreateDirectory(dir); + + try + { + const int entityCount = 25_000; + var rng = new System.Random(0x5EED); + + for (var i = 1; i <= entityCount; i++) + { + var serial = (Serial)(uint)i; + persistence.EntitiesBySerial[serial] = new RoundTripEntity(serial) + { + Value = rng.Next(), + Name = i % 5 == 0 ? null : $"entity-{i}" + }; + } + + persistence.RegisterType(typeof(RoundTripEntity)); + + // Freeze: what Persistence.SerializeAll + GenericEntityPersistence.Serialize do, + // against this test's chunk source instead of the world's. + foreach (var worker in workers) + { + worker.Wake(); + } + + source.SetOwner(persistence); + source.PushSingle(persistence); + Assert.True(persistence.TrySnapshotEntries(out var slotCount)); + source.PushSlotRanges(persistence, slotCount); + + // Mirrors World.PauseSerializationThreads + source.Flush(); + foreach (var worker in workers) + { + worker.Sleep(); + } + + // Background write phase: snapshot from the segment logs. v4 embeds the type + // table in the idx, so no SerializedTypes.db is produced or needed. + persistence.WriteSnapshot(dir); + + persistence.PostWorldSave(); // releases the entries snapshot + + // Load into a fresh persistence, like a server boot would. + loaded = new RoundTripPersistence(2001); + loaded.DeserializeIndexes(dir, null); + loaded.Deserialize(dir, null); + + Assert.True(loaded.SelfPayloadDeserialized); + Assert.Equal(persistence.EntitiesBySerial.Count, loaded.EntitiesBySerial.Count); + + foreach (var (serial, original) in persistence.EntitiesBySerial) + { + var entity = loaded.EntitiesBySerial[serial]; + Assert.Equal(original.Value, entity.Value); + Assert.Equal(original.Name, entity.Name); + Assert.Equal(original.Created.Ticks, entity.Created.Ticks); + } + + // Loading must hydrate the type table so the next save can write indexes. + Assert.True(loaded.TryGetTypeIndex(typeof(RoundTripEntity), out _)); + } + finally + { + persistence.Unregister(); + loaded?.Unregister(); + + foreach (var worker in workers) + { + worker.Exit(); + } + + World._threadWorkers = previousWorkers; + AssemblyHandler.Assemblies = previousAssemblies; + Directory.Delete(dir, true); + } + } +} diff --git a/Projects/Server.Tests/Tests/Serialization/GenericEntityPersistenceTypeTableTests.cs b/Projects/Server.Tests/Tests/Serialization/GenericEntityPersistenceTypeTableTests.cs new file mode 100644 index 000000000..a4aceba36 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/GenericEntityPersistenceTypeTableTests.cs @@ -0,0 +1,152 @@ +using System; +using System.Collections.Generic; +using System.IO; +using Xunit; + +namespace Server.Tests; + +[Collection("Sequential Server Tests")] +public class GenericEntityPersistenceTypeTableTests +{ + private class TypeTablePersistence : GenericEntityPersistence + { + public TypeTablePersistence() : base("TypeTable", 3000, 1, 0x7FFFFFFF) + { + } + + public override void Serialize(IGenericWriter writer) + { + } + + public override void Deserialize(IGenericReader reader) + { + } + } + + [Fact] + public void RegisterTypeAssignsStableInsertionOrderedIndexes() + { + var persistence = new TypeTablePersistence(); + + try + { + persistence.RegisterType(typeof(RoundTripEntity)); + persistence.RegisterType(typeof(string)); + persistence.RegisterType(typeof(RoundTripEntity)); // duplicate is a no-op + + Assert.Equal(2, persistence.TypeTable.Count); + Assert.Same(typeof(RoundTripEntity), persistence.TypeTable[0]); + Assert.Same(typeof(string), persistence.TypeTable[1]); + + Assert.True(persistence.TryGetTypeIndex(typeof(RoundTripEntity), out var first)); + Assert.Equal(0, first); + Assert.True(persistence.TryGetTypeIndex(typeof(string), out var second)); + Assert.Equal(1, second); + Assert.False(persistence.TryGetTypeIndex(typeof(int), out _)); + } + finally + { + persistence.Unregister(); + } + } + + [Fact] + public void UnresolvedTableEntrySkipsItsRecordsAfterConfirmation() + { + var previousAssemblies = AssemblyHandler.Assemblies; + AssemblyHandler.Assemblies = [.. previousAssemblies ?? [], typeof(RoundTripEntity).Assembly]; + + var dir = Path.Combine(Path.GetTempPath(), $"muo-typetable-{Guid.NewGuid():N}"); + Directory.CreateDirectory(Path.Combine(dir, "TypeTable")); + + TypeTablePersistence loaded = null; + var previousIn = Console.In; + + try + { + // Hand-write a v4 idx: two table entries (one bogus), one record per type. + using (var idx = new FileBufferWriter(Path.Combine(dir, "TypeTable", "TypeTable.idx"))) + { + idx.Write(4); // version + idx.Write(2); // type table count + idx.WriteRaw(typeof(RoundTripEntity).FullName); // index 0: resolvable + idx.WriteRaw("Server.Tests.DoesNotExistAnymore"); // index 1: bogus + idx.Write(2); // record count + idx.Write((ushort)0); // record 1: real type + idx.Write(1u); // serial + idx.Write(DateTime.UtcNow.Ticks); + idx.Write(0L); // position + idx.Write(4); // length + idx.Write((ushort)1); // record 2: bogus type + idx.Write(2u); + idx.Write(DateTime.UtcNow.Ticks); + idx.Write(4L); + idx.Write(4); + } + + // GetConstructorFor prompts on the console; answer "y" (delete those types). + Console.SetIn(new StringReader("y\n")); + + loaded = new TypeTablePersistence(); + loaded.DeserializeIndexes(dir, null); + + Assert.Single(loaded.EntitiesBySerial); + Assert.True(loaded.EntitiesBySerial.ContainsKey((Serial)1u)); + Assert.True(loaded.TryGetTypeIndex(typeof(RoundTripEntity), out _)); + } + finally + { + Console.SetIn(previousIn); + loaded?.Unregister(); + AssemblyHandler.Assemblies = previousAssemblies; + Directory.Delete(dir, true); + } + } + + [Fact] + public void LegacyV3IndexesStillLoadAndHydrateTheTypeTable() + { + var previousAssemblies = AssemblyHandler.Assemblies; + AssemblyHandler.Assemblies = [.. previousAssemblies ?? [], typeof(RoundTripEntity).Assembly]; + + var dir = Path.Combine(Path.GetTempPath(), $"muo-legacyidx-{Guid.NewGuid():N}"); + Directory.CreateDirectory(Path.Combine(dir, "TypeTable")); + + TypeTablePersistence loaded = null; + + try + { + var hash = AssemblyHandler.GetTypeHash(typeof(RoundTripEntity)); + + // Hand-write a v3 idx: records carry flag + 8-byte hash, resolved via typesDb. + using (var idx = new FileBufferWriter(Path.Combine(dir, "TypeTable", "TypeTable.idx"))) + { + idx.Write(3); // version + idx.Write(1); // record count + idx.Write((byte)2); // xxHash3 flag + idx.Write(hash); + idx.Write(1u); // serial + idx.Write(DateTime.UtcNow.Ticks); + idx.Write(0L); // position + idx.Write(4); // length + } + + var typesDb = new Dictionary { [hash] = typeof(RoundTripEntity).FullName }; + + loaded = new TypeTablePersistence(); + loaded.DeserializeIndexes(dir, typesDb); + + Assert.Single(loaded.EntitiesBySerial); + Assert.True(loaded.EntitiesBySerial.ContainsKey((Serial)1u)); + + // Legacy loads must hydrate the table so the NEXT save can write v4 indexes. + Assert.True(loaded.TryGetTypeIndex(typeof(RoundTripEntity), out _)); + } + finally + { + loaded?.Unregister(); + AssemblyHandler.Assemblies = previousAssemblies; + Directory.Delete(dir, true); + } + } +} diff --git a/Projects/Server.Tests/Tests/Serialization/SerializationChunkSourceTests.cs b/Projects/Server.Tests/Tests/Serialization/SerializationChunkSourceTests.cs new file mode 100644 index 000000000..69387f4e6 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/SerializationChunkSourceTests.cs @@ -0,0 +1,312 @@ +using System; +using System.Collections.Generic; +using Xunit; + +namespace Server.Tests; + +// Constructing a persistence mutates the static registry (an unsynchronized SortedSet); +// tests that do so must share the sequential collection. +[Collection("Sequential Server Tests")] +public class SerializationChunkSourceTests +{ + private class TestEntity : IGenericSerializable + { + public int PayloadSize { get; init; } = 16; + + public void Serialize(IGenericWriter writer) + { + for (var i = 0; i < PayloadSize; i++) + { + writer.Write((byte)(i & 0xFF)); + } + } + } + + private class TestPersistence : GenericPersistence + { + public int PayloadSize { get; init; } = 16; + + public TestPersistence() : base("ChunkSourceTest", 100) + { + } + + public (byte Thread, int Position, int Length) Placement => (_selfThread, _selfPosition, _selfLength); + + public override void Serialize(IGenericWriter writer) + { + for (var i = 0; i < PayloadSize; i++) + { + writer.Write((byte)(i & 0xFF)); + } + } + + public override void Deserialize(IGenericReader reader) + { + } + } + + private static List Drain(SerializationChunkSource source) + { + var drained = new List(); + while (source.TryTake(out var chunk)) + { + for (var i = 0; i < chunk.Count; i++) + { + drained.Add((TestEntity)chunk.Buffer[i]); + } + + source.Return(chunk.Buffer, chunk.Count); + } + + return drained; + } + + [Fact] + public void PartialChunkIsNotVisibleUntilFlush() + { + var source = new SerializationChunkSource(); + var entities = new List(); + + for (var i = 0; i < 100; i++) + { + var e = new TestEntity(); + entities.Add(e); + source.Push(e); + } + + Assert.False(source.TryTake(out _)); + + source.Flush(); + + var drained = Drain(source); + Assert.Equal(entities, drained); + + // Flush again should publish nothing + source.Flush(); + Assert.False(source.TryTake(out _)); + } + + [Fact] + public void FullChunkPublishesWithoutFlush() + { + var source = new SerializationChunkSource(); + + for (var i = 0; i < 4096; i++) + { + source.Push(new TestEntity()); + } + + Assert.True(source.TryTake(out var chunk)); + Assert.Null(chunk.Single); + Assert.Equal(4096, chunk.Count); + source.Return(chunk.Buffer, chunk.Count); + + // Nothing partial left behind + source.Flush(); + Assert.False(source.TryTake(out _)); + } + + [Fact] + public void PushSingleDoesNotDisturbPartialChunk() + { + var source = new SerializationChunkSource(); + var heavy = new TestPersistence(); + + try + { + var small1 = new TestEntity(); + var small2 = new TestEntity(); + + source.Push(small1); + source.PushSingle(heavy); // published immediately as a single, ahead of the partial chunk + source.Push(small2); + + Assert.True(source.TryTake(out var chunk)); + Assert.Same(heavy, chunk.Single); + Assert.Equal(1, chunk.Count); + + Assert.False(source.TryTake(out _)); + source.Flush(); + + var drained = Drain(source); + Assert.Equal([small1, small2], drained); + } + finally + { + heavy.Unregister(); + } + } + + [Fact] + public void SetOwnerPublishesPartialChunkOnChange() + { + var source = new SerializationChunkSource(); + var ownerA = new TestPersistence(); + var ownerB = new TestPersistence(); + + try + { + source.SetOwner(ownerA); + source.Push(new TestEntity()); + source.Push(new TestEntity()); + + // Same owner: partial chunk stays private to the producer. + source.SetOwner(ownerA); + Assert.False(source.TryTake(out _)); + + // Owner change publishes the partial chunk, keeping chunks persistence-homogeneous. + source.SetOwner(ownerB); + Assert.True(source.TryTake(out var chunk)); + Assert.Same(ownerA, chunk.Owner); + Assert.Equal(2, chunk.Count); + source.Return(chunk.Buffer, chunk.Count); + + source.Push(new TestEntity()); + source.Flush(); + Assert.True(source.TryTake(out chunk)); + Assert.Same(ownerB, chunk.Owner); + } + finally + { + ownerA.Unregister(); + ownerB.Unregister(); + } + } + + [Fact] + public void ReturnedBuffersAreClearedAndReused() + { + var source = new SerializationChunkSource(); + + for (var i = 0; i < 4096; i++) + { + source.Push(new TestEntity()); + } + + Assert.True(source.TryTake(out var chunk)); + var buffer = chunk.Buffer; + source.Return(buffer, chunk.Count); + + Assert.All(buffer, Assert.Null); + + // Next fill rents the pooled buffer instead of allocating + source.Push(new TestEntity()); + source.Flush(); + Assert.True(source.TryTake(out var reused)); + Assert.Same(buffer, reused.Buffer); + } + + [Fact] + public void WorkersDrainAllEntitiesAndLogSegments() + { + var source = new SerializationChunkSource(); + var owner = new TestPersistence { PayloadSize = 512 * 1024 }; + var workers = new SerializationThreadWorker[2]; + for (var i = 0; i < workers.Length; i++) + { + workers[i] = new SerializationThreadWorker(i, source); + workers[i].AllocateHeap(); + } + + try + { + var entities = new List(); + for (var i = 0; i < 10_000; i++) + { + entities.Add(new TestEntity { PayloadSize = 16 + i % 64 }); + } + + foreach (var worker in workers) + { + worker.Wake(); + } + + // A large payload published as a dedicated single chunk (a persistence + // self-payload), interleaved with the bare entity stream. + source.SetOwner(owner); + for (var i = 0; i < entities.Count; i++) + { + if (i == 5000) + { + source.PushSingle(owner); + } + + source.Push(entities[i]); + } + + // Mirrors World.PauseSerializationThreads + source.Flush(); + foreach (var worker in workers) + { + worker.Sleep(); + } + + long totalEntities = 0; + long totalBytes = 0; + foreach (var worker in workers) + { + totalEntities += worker.EntitiesSerialized; + totalBytes += worker.BytesSerialized; + } + + Assert.Equal(entities.Count + 1, totalEntities); + + // The self-payload recorded its placement on the persistence itself. + var (selfThread, selfPosition, selfLength) = owner.Placement; + Assert.Equal(owner.PayloadSize, selfLength); + Assert.InRange(selfThread, (byte)0, (byte)(workers.Length - 1)); + var selfHeap = workers[selfThread].GetHeap(selfPosition, selfLength); + Assert.Equal(0, selfHeap[0]); + Assert.Equal((selfLength - 1) & 0xFF, selfHeap[^1]); + + // Every entity appears exactly once in the worker segment logs, with a + // consistent span on that worker's heap: positions are implicit (contiguous + // writes), identity comes from the buffer-entities log. + var seen = new HashSet(); + long expectedBytes = owner.PayloadSize; + foreach (var e in entities) + { + expectedBytes += e.PayloadSize; + } + + foreach (var worker in workers) + { + var lengths = worker.Lengths; + var bufferEntities = worker.BufferEntities; + + foreach (var segment in worker.Segments) + { + Assert.Same(owner, segment.Owner); + Assert.Equal(-1, segment.SlotOffset); + + var heapPos = (int)segment.HeapStart; + for (var i = 0; i < segment.RecordCount; i++) + { + var entity = (TestEntity)bufferEntities[segment.EntitiesStart + i]; + var length = lengths[segment.LengthsStart + i]; + + Assert.True(seen.Add(entity)); + Assert.Equal(entity.PayloadSize, length); + + var heap = workers[Array.IndexOf(workers, worker)].GetHeap(heapPos, length); + Assert.Equal(0, heap[0]); + Assert.Equal((length - 1) & 0xFF, heap[^1]); + + heapPos += length; + } + } + } + + Assert.Equal(entities.Count, seen.Count); + Assert.Equal(expectedBytes, totalBytes); + } + finally + { + owner.Unregister(); + foreach (var worker in workers) + { + worker.Exit(); + } + } + } +} diff --git a/Projects/Server.Tests/Tests/Serialization/SerializationThreadWorkerHandshakeTests.cs b/Projects/Server.Tests/Tests/Serialization/SerializationThreadWorkerHandshakeTests.cs new file mode 100644 index 000000000..5ea281172 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/SerializationThreadWorkerHandshakeTests.cs @@ -0,0 +1,56 @@ +using System; +using System.Threading; +using Xunit; + +namespace Server.Tests; + +[Collection("Sequential Server Tests")] +public class SerializationThreadWorkerHandshakeTests +{ + // The pause handshake must tolerate a new cycle starting the moment _stopEvent is + // set (Exit right after Sleep). The watchdog turns a reintroduced deadlock into a + // failure instead of a hung run. + [Fact] + public void WakeSleepExitChurn_NeverDeadlocks() + { + Exception failure = null; + var done = new ManualResetEventSlim(); + + var churn = new Thread(() => + { + try + { + for (var i = 0; i < 2000; i++) + { + var source = new SerializationChunkSource(); + var worker = new SerializationThreadWorker(0, source); + worker.AllocateHeap(); + + worker.Wake(); + worker.Sleep(); + worker.Exit(); // Immediately after Sleep returns — the racy window. + } + } + catch (Exception e) + { + failure = e; + } + finally + { + done.Set(); + } + }) + { + IsBackground = true, + Name = "Handshake Churn" + }; + + churn.Start(); + + Assert.True( + done.Wait(TimeSpan.FromMinutes(2)), + "Worker pause/exit handshake deadlocked (owner blocked in Sleep or worker spinning)." + ); + Assert.Null(failure); + } +} diff --git a/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs b/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs new file mode 100644 index 000000000..bdee95698 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs @@ -0,0 +1,143 @@ +using System; +using System.Collections.Generic; +using Xunit; + +namespace Server.Tests; + +// Constructing a persistence mutates the static registry (an unsynchronized SortedSet); +// tests that do so must share the sequential collection. +[Collection("Sequential 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 void Delete() + { + } + + public void Serialize(IGenericWriter writer) + { + writer.Write(Serial); + writer.Write(0xC0FFEE); + } + + public void Deserialize(IGenericReader reader) + { + } + } + + private static TestEntity GetSlotValue(Array entries, int slot) => + System.Runtime.CompilerServices.Unsafe.As[]>(entries)[slot].Value; + + [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); + var lengths = new List(); + + // Serialize in worker-sized slices, like the drain does. + var serialized = 0; + for (var offset = 0; offset < slotCount; offset += 4096) + { + serialized += source.SerializeRange(writer, lengths, offset, Math.Min(4096, slotCount - offset)); + } + + Assert.Equal(dict.Count, serialized); + Assert.Equal(dict.Count, lengths.Count); + + // Re-walk the same slots in the same order, pairing each occupied slot with the + // next logged length — exactly how the snapshot writer locates each record. + var entriesField = ShadowDictionaryEntries.GetEntriesField(); + var entries = (Array)entriesField.GetValue(dict); + + var position = 0; + var lengthIndex = 0; + var matched = 0; + + for (var slot = 0; slot < entries.Length; slot++) + { + // Occupancy is exactly the non-null values, same as production. + var entity = GetSlotValue(entries, slot); + if (entity == null) + { + continue; + } + + var length = lengths[lengthIndex++]; + Assert.Equal(8, length); // serial + int + + var span = writer.Buffer.AsSpan(position, length); + Assert.Equal(entity.Serial, (Serial)System.Buffers.Binary.BinaryPrimitives.ReadUInt32LittleEndian(span)); + + position += length; + matched++; + } + + Assert.Equal(dict.Count, matched); + } + 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/Guild.cs b/Projects/Server/Guild.cs index c2fd1f679..1b2b013d1 100644 --- a/Projects/Server/Guild.cs +++ b/Projects/Server/Guild.cs @@ -52,10 +52,6 @@ public abstract class BaseGuild : ISerializable [CommandProperty(AccessLevel.GameMaster, readOnly: true)] public DateTime Created { get; set; } = Core.Now; - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - public abstract void Serialize(IGenericWriter writer); public abstract void Deserialize(IGenericReader reader); diff --git a/Projects/Server/IEntity.cs b/Projects/Server/IEntity.cs index bb1408bd3..63a02fc02 100644 --- a/Projects/Server/IEntity.cs +++ b/Projects/Server/IEntity.cs @@ -115,10 +115,6 @@ public class Entity : IEntity Timer.StartTimer(Delete); } - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - public void Serialize(IGenericWriter writer) { } diff --git a/Projects/Server/Items/Item.cs b/Projects/Server/Items/Item.cs index ee066c657..a3e546bc2 100644 --- a/Projects/Server/Items/Item.cs +++ b/Projects/Server/Items/Item.cs @@ -821,10 +821,6 @@ public partial class Item : IHued, IComparable, ISpawnable, IObjectPropert [CommandProperty(AccessLevel.Counselor)] public Serial Serial { get; } - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - public virtual void Serialize(IGenericWriter writer) { writer.Write(9); // version diff --git a/Projects/Server/Mobiles/Mobile.cs b/Projects/Server/Mobiles/Mobile.cs index 778d5b5a7..3e19548c6 100644 --- a/Projects/Server/Mobiles/Mobile.cs +++ b/Projects/Server/Mobiles/Mobile.cs @@ -2308,10 +2308,6 @@ public partial class Mobile : IHued, IComparable, ISpawnable, IObjectPro [CommandProperty(AccessLevel.Counselor)] public Serial Serial { get; } - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - public virtual void Serialize(IGenericWriter writer) { writer.Write(37); // version diff --git a/Projects/Server/Serialization/AdhocPersistence.cs b/Projects/Server/Serialization/AdhocPersistence.cs index 236f24371..d6c132549 100644 --- a/Projects/Server/Serialization/AdhocPersistence.cs +++ b/Projects/Server/Serialization/AdhocPersistence.cs @@ -14,7 +14,6 @@ *************************************************************************/ using System; -using System.Collections.Concurrent; using System.Collections.Generic; using System.IO; using System.IO.MemoryMappedFiles; @@ -27,9 +26,9 @@ public static class AdhocPersistence /** * Serializes to memory. */ - public static IGenericWriter SerializeToBuffer(Action serializer, ConcurrentQueue types = null) + public static IGenericWriter SerializeToBuffer(Action serializer) { - var saveBuffer = new BufferWriter(true, types); + var saveBuffer = new BufferWriter(true); serializer(saveBuffer); return saveBuffer; } @@ -55,22 +54,11 @@ public static class AdhocPersistence { var fullPath = PathUtility.GetFullPath(filePath, Core.BaseDirectory); PathUtility.EnsureDirectory(Path.GetDirectoryName(fullPath)); - HashSet typesSet = []; - var writer = new MemoryMapFileWriter(new FileStream(filePath, FileMode.Create), sizeHint, typesSet); + + var writer = new FileBufferWriter(fullPath, sizeHint); serializer(writer); - Task.Run( - () => - { - var fs = writer.FileStream; - - writer.Dispose(); - fs.Dispose(); - - Persistence.WriteSerializedTypesSnapshot(Path.GetDirectoryName(fullPath), typesSet); - }, - Core.ClosingTokenSource.Token - ); + Task.Run(() => writer.Dispose(), Core.ClosingTokenSource.Token); } public static unsafe void Deserialize(string filePath, Action deserializer) diff --git a/Projects/Server/Serialization/BufferWriter.cs b/Projects/Server/Serialization/BufferWriter.cs index 45f9ed379..b811d68cb 100644 --- a/Projects/Server/Serialization/BufferWriter.cs +++ b/Projects/Server/Serialization/BufferWriter.cs @@ -14,10 +14,12 @@ *************************************************************************/ using System; +using System.Buffers; using System.Buffers.Binary; -using System.Collections.Concurrent; +using System.Collections; using System.Diagnostics; using System.IO; +using System.Net; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; using System.Text; @@ -27,9 +29,9 @@ namespace Server; public class BufferWriter : IGenericWriter { - private readonly ConcurrentQueue _types; private readonly Encoding _encoding; private readonly bool _prefixStrings; + private long _bytesWritten; private long _index; @@ -56,27 +58,25 @@ public class BufferWriter : IGenericWriter private byte[] _buffer; - public BufferWriter(byte[] buffer, bool prefixStr, ConcurrentQueue types = null) + public BufferWriter(byte[] buffer, bool prefixStr) { _prefixStrings = prefixStr; _encoding = TextEncoding.UTF8; _buffer = buffer; - _types = types; } - public BufferWriter(bool prefixStr, ConcurrentQueue types = null) : this(0, prefixStr, types) + public BufferWriter(bool prefixStr) : this(0, prefixStr) { } - public BufferWriter(int count, bool prefixStr, ConcurrentQueue types = null) + public BufferWriter(int count, bool prefixStr) { _prefixStrings = prefixStr; _encoding = TextEncoding.UTF8; _buffer = GC.AllocateUninitializedArray(count < 1 ? BufferSize : count); - _types = types; } - public virtual long Position => Index; + public virtual long Position => _index; protected virtual int BufferSize => 256; @@ -89,6 +89,8 @@ public class BufferWriter : IGenericWriter [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Resize(int size) { + _bytesWritten = Math.Max(_bytesWritten, _index); + // We shouldn't ever resize to a 0 length buffer. That is dangerous if (size <= 0) { @@ -107,13 +109,23 @@ public class BufferWriter : IGenericWriter public virtual void Flush() => Resize(Math.Clamp(_buffer.Length * 2, BufferSize, _buffer.Length + 1024 * 1024 * 64)); + /// + /// Ensures capacity, returns a ref at the current position, and advances the index. + /// The capacity check proves the caller's unaligned store is in-bounds, and the index + /// only moves forward between Seek calls, so no per-write validation is needed. Growth + /// (Flush -> Resize) always adds at least BufferSize, covering any primitive width. + /// [MethodImpl(MethodImplOptions.AggressiveInlining)] - private void FlushIfNeeded(int amount) + private ref byte Reserve(int bytes) { - if (Index + amount > _buffer.Length) + if ((uint)(_index + bytes) > (uint)_buffer.Length) { Flush(); } + + ref var result = ref Unsafe.Add(ref MemoryMarshal.GetArrayDataReference(_buffer), (nint)_index); + _index += bytes; + return ref result; } public virtual void Write(byte[] bytes) => Write(bytes.AsSpan()); @@ -130,7 +142,7 @@ public class BufferWriter : IGenericWriter } bytes.CopyTo(_buffer.AsSpan((int)_index)); - Index += length; + _index += length; } [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -145,13 +157,15 @@ public class BufferWriter : IGenericWriter "Attempting to seek to an invalid position using SeekOrigin.Begin" ); Debug.Assert( - origin != SeekOrigin.Current || Index + offset >= 0 && Index + offset < _buffer.Length, + origin != SeekOrigin.Current || _index + offset >= 0 && _index + offset < _buffer.Length, "Attempting to seek to an invalid position using SeekOrigin.Current" ); + _bytesWritten = Math.Max(_bytesWritten, _index); + return Index = Math.Max(0, origin switch { - SeekOrigin.Current => Index + offset, + SeekOrigin.Current => _index + offset, SeekOrigin.End => _bytesWritten + offset, _ => offset // Begin }); @@ -169,107 +183,111 @@ public class BufferWriter : IGenericWriter else { Write(true); - InternalWriteString(value); + WriteRaw(value); } } else { - InternalWriteString(value); + WriteRaw(value); } } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(long value) { - FlushIfNeeded(8); + if (!BitConverter.IsLittleEndian) + { + value = BinaryPrimitives.ReverseEndianness(value); + } - BinaryPrimitives.WriteInt64LittleEndian(_buffer.AsSpan((int)_index), value); - Index += 8; + Unsafe.WriteUnaligned(ref Reserve(8), value); } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(ulong value) { - FlushIfNeeded(8); + if (!BitConverter.IsLittleEndian) + { + value = BinaryPrimitives.ReverseEndianness(value); + } - BinaryPrimitives.WriteUInt64LittleEndian(_buffer.AsSpan((int)_index), value); - Index += 8; + Unsafe.WriteUnaligned(ref Reserve(8), value); } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(int value) { - FlushIfNeeded(4); + if (!BitConverter.IsLittleEndian) + { + value = BinaryPrimitives.ReverseEndianness(value); + } - BinaryPrimitives.WriteInt32LittleEndian(_buffer.AsSpan((int)_index), value); - Index += 4; + Unsafe.WriteUnaligned(ref Reserve(4), value); } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(uint value) { - FlushIfNeeded(4); + if (!BitConverter.IsLittleEndian) + { + value = BinaryPrimitives.ReverseEndianness(value); + } - BinaryPrimitives.WriteUInt32LittleEndian(_buffer.AsSpan((int)_index), value); - Index += 4; + Unsafe.WriteUnaligned(ref Reserve(4), value); } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(short value) { - FlushIfNeeded(2); + if (!BitConverter.IsLittleEndian) + { + value = BinaryPrimitives.ReverseEndianness(value); + } - BinaryPrimitives.WriteInt16LittleEndian(_buffer.AsSpan((int)_index), value); - Index += 2; + Unsafe.WriteUnaligned(ref Reserve(2), value); } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(ushort value) { - FlushIfNeeded(2); + if (!BitConverter.IsLittleEndian) + { + value = BinaryPrimitives.ReverseEndianness(value); + } - BinaryPrimitives.WriteUInt16LittleEndian(_buffer.AsSpan((int)_index), value); - Index += 2; + Unsafe.WriteUnaligned(ref Reserve(2), value); } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(double value) { - FlushIfNeeded(8); + if (!BitConverter.IsLittleEndian) + { + value = BitConverter.Int64BitsToDouble(BinaryPrimitives.ReverseEndianness(BitConverter.DoubleToInt64Bits(value))); + } - BinaryPrimitives.WriteDoubleLittleEndian(_buffer.AsSpan((int)_index), value); - Index += 8; + Unsafe.WriteUnaligned(ref Reserve(8), value); } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(float value) { - FlushIfNeeded(4); + if (!BitConverter.IsLittleEndian) + { + value = BitConverter.Int32BitsToSingle(BinaryPrimitives.ReverseEndianness(BitConverter.SingleToInt32Bits(value))); + } - BinaryPrimitives.WriteSingleLittleEndian(_buffer.AsSpan((int)_index), value); - Index += 4; + Unsafe.WriteUnaligned(ref Reserve(4), value); } [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(byte value) - { - FlushIfNeeded(1); - _buffer[Index++] = value; - } + public void Write(byte value) => Reserve(1) = value; [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(sbyte value) - { - FlushIfNeeded(1); - _buffer[Index++] = (byte)value; - } + public void Write(sbyte value) => Reserve(1) = (byte)value; [MethodImpl(MethodImplOptions.AggressiveInlining)] - public unsafe void Write(bool value) - { - FlushIfNeeded(1); - _buffer[Index++] = *(byte*)&value; // up to 30% faster to dereference the raw value on the stack - } + public void Write(bool value) => Reserve(1) = Unsafe.As(ref value); [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Write(Serial serial) => Write(serial.Value); @@ -285,7 +303,6 @@ public class BufferWriter : IGenericWriter { Write((byte)0x2); // xxHash3 64bit Write(AssemblyHandler.GetTypeHash(type)); - _types?.Enqueue(type); } } @@ -299,18 +316,262 @@ public class BufferWriter : IGenericWriter } [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal void InternalWriteString(string value) + public void WriteEncodedInt(int value) { - var length = _encoding.GetByteCount(value); + var v = (uint)value; - ((IGenericWriter)this).WriteEncodedInt(length); + // FAST PATH: 1 byte (0 to 127). + // This keeps the inlined code incredibly tiny at the call site. + if (v < 0x80) + { + Reserve(1) = (byte)v; + } + else + { + // SLOW PATH: Push to a non-inlined method to prevent code bloat. + WriteEncodedIntMultiByte(v); + } + } - while (_buffer.Length - _index < length) + [MethodImpl(MethodImplOptions.NoInlining)] + private void WriteEncodedIntMultiByte(uint v) + { + // We already know v >= 0x80. Unroll the loop entirely based on magnitude. + // This allows us to call Reserve() exactly ONE time. + + if (v < 0x4000) // 2 bytes + { + ref byte ptr = ref Reserve(2); + ptr = (byte)(v | 0x80); + Unsafe.Add(ref ptr, 1) = (byte)(v >> 7); + } + else if (v < 0x200000) // 3 bytes + { + ref byte ptr = ref Reserve(3); + ptr = (byte)(v | 0x80); + Unsafe.Add(ref ptr, 1) = (byte)((v >> 7) | 0x80); + Unsafe.Add(ref ptr, 2) = (byte)(v >> 14); + } + else if (v < 0x10000000) // 4 bytes + { + ref byte ptr = ref Reserve(4); + ptr = (byte)(v | 0x80); + Unsafe.Add(ref ptr, 1) = (byte)((v >> 7) | 0x80); + Unsafe.Add(ref ptr, 2) = (byte)((v >> 14) | 0x80); + Unsafe.Add(ref ptr, 3) = (byte)(v >> 21); + } + else // 5 bytes (including all negative numbers due to logical shift) + { + ref byte ptr = ref Reserve(5); + ptr = (byte)(v | 0x80); + Unsafe.Add(ref ptr, 1) = (byte)((v >> 7) | 0x80); + Unsafe.Add(ref ptr, 2) = (byte)((v >> 14) | 0x80); + Unsafe.Add(ref ptr, 3) = (byte)((v >> 21) | 0x80); + Unsafe.Add(ref ptr, 4) = (byte)(v >> 28); + } + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(DateTime value) + { + // If DateTimeKind is Unspecified, we can't assume it needs to be converted. + if (value.Kind == DateTimeKind.Local) + { + value = value.ToUniversalTime(); + } + + Write(value.Ticks); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void WriteDeltaTime(DateTime value) + { + if (value == DateTime.MinValue) + { + Write(long.MinValue); + return; + } + + if (value == DateTime.MaxValue) + { + Write(long.MaxValue); + return; + } + + if (value.Kind == DateTimeKind.Local) + { + value = value.ToUniversalTime(); + } + + // Technically supports negative deltas for times in the past + Write(value.Ticks - DateTime.UtcNow.Ticks); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(IPAddress value) + { + Span stack = stackalloc byte[16]; + value.TryWriteBytes(stack, out var bytesWritten); + Write((byte)bytesWritten); + Write(stack[..bytesWritten]); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(TimeSpan value) => Write(value.Ticks); + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(Point3D value) + { + Write(value.m_X); + Write(value.m_Y); + Write(value.m_Z); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(Point2D value) + { + Write(value.m_X); + Write(value.m_Y); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(Rectangle2D value) + { + Write(value.Start); + Write(value.End); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(Rectangle3D value) + { + Write(value.Start); + Write(value.End); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(Map value) => Write((byte)(value?.MapIndex ?? 0xFF)); + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(Race value) => Write((byte)(value?.RaceIndex ?? 0xFF)); + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public unsafe void WriteEnum(T value) where T : unmanaged, Enum + { + switch (sizeof(T)) + { + default: + { + throw new ArgumentException($"Argument of type {typeof(T)} is not a normal enum"); + } + case 1: + { + Write(*(byte*)&value); + break; + } + case 2: + { + Write(*(ushort*)&value); + break; + } + case 4: + { + WriteEncodedInt(*(int*)&value); + break; + } + case 8: + { + Write(*(ulong*)&value); + break; + } + } + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(Guid guid) + { + Span stack = stackalloc byte[16]; + guid.TryWriteBytes(stack); + Write(stack); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(BitArray bitArray) + { + var bitLength = bitArray.Length; + var byteLength = (bitLength + 7) / 8; + + WriteEncodedInt(bitLength); + + var arrayBuffer = ArrayPool.Shared.Rent(byteLength); + try + { + bitArray.CopyTo(arrayBuffer, 0); + Write(arrayBuffer.AsSpan(0, byteLength)); + } + finally + { + ArrayPool.Shared.Return(arrayBuffer); + } + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Write(TextDefinition def) + { + if (def == null) + { + WriteEncodedInt(3); + } + else if (def.Number > 0) + { + WriteEncodedInt(1); + WriteEncodedInt(def.Number); + } + else if (def.String != null) + { + WriteEncodedInt(2); + Write(def.String); + } + else + { + WriteEncodedInt(0); // Empty + } + } + + public void WriteRaw(string value) + { + // Single pass, in place: reserve the UTF-8 worst case (3 bytes per char) plus a + // length prefix sized for that worst case, encode directly into the buffer, then + // write the actual byte count into the reserved prefix zero-padded to the same + // width. Readers accumulate 7-bit groups, so non-minimal prefixes decode + // identically — no second pass over the string, no scratch copy, no pooling. + var maxLength = value.Length * 3; + var prefixWidth = EncodedIntWidth(maxLength); + + while (_buffer.Length - _index < prefixWidth + maxLength) { Flush(); } - // We don't use spans here since that incurs extra allocations for safety. - Index += _encoding.GetBytes(value, 0, value.Length, _buffer, (int)_index); + var written = _encoding.GetBytes(value, _buffer.AsSpan((int)(_index + prefixWidth))); + + WriteEncodedIntPadded(written, prefixWidth); + _index += written; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static int EncodedIntWidth(int value) => + value < 0x80 ? 1 : value < 0x4000 ? 2 : value < 0x20_0000 ? 3 : value < 0x1000_0000 ? 4 : 5; + + private void WriteEncodedIntPadded(int value, int width) + { + var v = (uint)value; + + for (var i = 1; i < width; i++) + { + _buffer[_index++] = (byte)(v | 0x80); + v >>= 7; + } + + _buffer[_index++] = (byte)v; // fits in 7 bits because width >= EncodedIntWidth(value) } } diff --git a/Projects/Server/Serialization/FileBufferWriter.cs b/Projects/Server/Serialization/FileBufferWriter.cs new file mode 100644 index 000000000..9f1e9c290 --- /dev/null +++ b/Projects/Server/Serialization/FileBufferWriter.cs @@ -0,0 +1,138 @@ +/************************************************************************* + * ModernUO * + * Copyright 2019-2026 - ModernUO Development Team * + * Email: hi@modernuo.com * + * File: FileBufferWriter.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.Buffers; +using System.IO; +using Microsoft.Win32.SafeHandles; + +namespace Server; + +/// +/// A whose staging block drains to a file when full instead of +/// growing: the full raw write path (unrolled encoded ints, in-place strings) composes into +/// memory, and the file sees large sequential positional writes. Seeks flush the staging +/// block and move the file offset, so backwards patches (e.g. the idx entity count) become +/// small positional writes. Memory-mapped writing pays soft page faults on every composed +/// page and dirty-section teardown stalls at dispose — measured ~4x slower at snapshot sizes. +/// A single item larger than the staging block grows the block via the base resize path, +/// so oversized spans and strings remain correct. +/// +public class FileBufferWriter : BufferWriter, IDisposable +{ + private const int MinStagingSize = 256; + private const int MaxStagingSize = 1024 * 1024; // 1MB write granularity for large files + + private readonly SafeFileHandle _handle; + private readonly byte[] _rentedStaging; + private long _fileOffset; // file position where the staging block begins + private long _fileHighWater; // logical end of file across seeks + + /// Destination file; created/truncated. + /// + /// Expected total file size when known. Files at or under the staging cap never + /// drain until close; larger files stream through a pooled block at the cap. The + /// block comes from ArrayPool so sequential snapshot writers recycle one buffer + /// instead of dropping a large-object allocation per file per save. + /// + public FileBufferWriter(string filePath, long expectedSize = MaxStagingSize) + : base(RentStaging(expectedSize), true) + { + _rentedStaging = Buffer; + _handle = File.OpenHandle(filePath, FileMode.Create, FileAccess.Write, FileShare.None, FileOptions.SequentialScan); + } + + private static byte[] RentStaging(long expectedSize) => + ArrayPool.Shared.Rent((int)Math.Clamp(expectedSize, MinStagingSize, MaxStagingSize)); + + public override long Position => _fileOffset + Index; + + public override void Flush() + { + if (Index > 0) + { + Drain(); + } + else + { + // Nothing staged and still not enough room: a single item larger than the + // staging block. Grow the block so the base write loops always make progress. + base.Flush(); + } + } + + private void Drain() + { + var length = (int)Index; + RandomAccess.Write(_handle, Buffer.AsSpan(0, length), _fileOffset); + _fileOffset += length; + + if (_fileOffset > _fileHighWater) + { + _fileHighWater = _fileOffset; + } + + Index = 0; + } + + public override long Seek(long offset, SeekOrigin origin) + { + var position = Position; + + if (position > _fileHighWater) + { + _fileHighWater = position; + } + + var target = origin switch + { + SeekOrigin.Current => position + offset, + SeekOrigin.End => _fileHighWater + offset, + _ => offset // Begin + }; + + if (target < 0) + { + throw new InvalidOperationException("Seek before start of file"); + } + + if (Index > 0) + { + Drain(); + } + + _fileOffset = target; + return target; + } + + public override void Close() + { + if (!_handle.IsClosed) + { + if (Index > 0) + { + Drain(); + } + + _handle.Dispose(); + + // Safe even if an oversized item grew the staging block: growth replaced the + // base buffer with a fresh array, so the rented one is no longer referenced. + ArrayPool.Shared.Return(_rentedStaging); + } + } + + public void Dispose() => Close(); +} diff --git a/Projects/Server/Serialization/GenericEntityPersistence.cs b/Projects/Server/Serialization/GenericEntityPersistence.cs index 501e84aeb..a49f213eb 100644 --- a/Projects/Server/Serialization/GenericEntityPersistence.cs +++ b/Projects/Server/Serialization/GenericEntityPersistence.cs @@ -28,13 +28,23 @@ namespace Server; public interface IGenericEntityPersistence { - public void DeserializeIndexes(string savePath, Dictionary typesDb); + 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; @@ -44,6 +54,35 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer private readonly Dictionary _pendingAdd = new(); private readonly Dictionary _pendingDelete = new(); + // Insertion-ordered table of every entity type added since boot. idx v4 records + // reference types by table index, so entries are never removed — a type whose + // entities were all deleted keeps its slot until restart. Only mutated on the game + // thread (AddEntity, deserialize); only read on the background writer thread during + // WritingSave, when AddEntity diverts to the pending queues. + private readonly Dictionary _typeIndexes = new(); + private readonly List _typeTable = []; + + internal IReadOnlyList TypeTable => _typeTable; + + internal bool TryGetTypeIndex(Type type, out ushort index) => _typeIndexes.TryGetValue(type, out index); + + internal void RegisterType(Type type) + { + ref var index = ref CollectionsMarshal.GetValueRefOrAddDefault(_typeIndexes, type, out var exists); + if (!exists) + { + if (_typeTable.Count > ushort.MaxValue) + { + throw new InvalidOperationException( + $"{Name} exceeded {ushort.MaxValue + 1} distinct entity types." + ); + } + + index = (ushort)_typeTable.Count; + _typeTable.Add(type); + } + } + public Dictionary EntitiesBySerial { get; } = new(); public GenericEntityPersistence(string name, int priority, uint minSerial, uint maxSerial) : this( @@ -63,81 +102,94 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer typeof(T).RegisterFindEntity(Find); } - public override void WriteSnapshot(string savePath, HashSet typeSet) + public override void WriteSnapshot(string savePath) { var dir = Path.Combine(savePath, Name); PathUtility.EnsureDirectory(dir); var threads = World._threadWorkers; - using var binFs = new FileStream(Path.Combine(dir, $"{Name}.bin"), FileMode.Create, FileAccess.Write, FileShare.None); - using var idxFs = new FileStream(Path.Combine(dir, $"{Name}.idx"), FileMode.Create); - using var idx = new MemoryMapFileWriter(idxFs, 1024 * 1024, typeSet); // 1MB + // 1MB buffer: segments are written as large spans, but idx entries and skip splits + // still benefit on the snapshot thread. + using var binFs = new FileStream( + Path.Combine(dir, $"{Name}.bin"), FileMode.Create, FileAccess.Write, FileShare.None, 1024 * 1024 + ); + // v4 records are fixed-width 26 bytes; the header carries the type table + // (name lengths vary — 64 bytes per entry is a staging hint, not a contract). + var expectedIdxSize = 12 + 26L * EntitiesBySerial.Count + 64L * _typeTable.Count; + using var idx = new FileBufferWriter(Path.Combine(dir, $"{Name}.idx"), expectedIdxSize); var binPosition = 0L; // Support for non-entity generic serialization. - if (SerializedLength > 0) + if (_selfLength > 0) { try { - binFs.Write(threads[SerializedThread].GetHeap(SerializedPosition, SerializedLength)); + binFs.Write(threads[_selfThread].GetHeap(_selfPosition, _selfLength)); } catch (Exception error) { logger.Error( error, - "Error writing entity: (Thread: {Thread} - {Start} {Length})", - SerializedThread, - SerializedPosition, - SerializedLength + "Error writing self-payload: (Thread: {Thread} - {Start} {Length})", + _selfThread, + _selfPosition, + _selfLength ); } - binPosition += SerializedLength; + binPosition += _selfLength; } - idx.Write(3); // Version + idx.Write(4); // Version + + // The type table is fully known at freeze (AddEntity diverts to the pending + // queues while saving) and is written before the records so the loader can + // resolve constructors before reading them. + idx.Write(_typeTable.Count); + for (var i = 0; i < _typeTable.Count; i++) + { + idx.WriteRaw(_typeTable[i].FullName); + } var countPosition = idx.Position; idx.Write(0); - var entityCount = EntitiesBySerial.Count; - foreach (var e in EntitiesBySerial.Values) + // The bin is written in worker-heap order, not dictionary order: each worker logged + // (segment, record lengths) as it serialized, so records pair with their bytes by + // re-walking the same slots in the same order. The idx stores absolute positions, + // so the loader never cares about record order. + var entityCount = 0; + + for (var t = 0; t < threads.Length; t++) { - if (e is Item { SkipSerialization: true } or Mobile { SkipSerialization: true }) + var worker = threads[t]; + var segments = worker.Segments; + + for (var s = 0; s < segments.Count; s++) { - entityCount--; - continue; + var segment = segments[s]; + if (!ReferenceEquals(segment.Owner, this)) + { + continue; + } + + try + { + binPosition = WriteSegmentRecords(worker, in segment, idx, binFs, binPosition, ref entityCount); + } + catch (Exception error) + { + logger.Error( + error, + "Error writing segment: (Thread: {Thread} - {Start}, {Records} records)", + t, + segment.HeapStart, + segment.RecordCount + ); + } } - - var thread = e.SerializedThread; - var heapStart = e.SerializedPosition; - var heapLength = e.SerializedLength; - - idx.Write(e.GetType()); - idx.Write(e.Serial); - idx.Write(e.Created.Ticks); - idx.Write(binPosition); - idx.Write(heapLength); - - try - { - binFs.Write(threads[thread].GetHeap(heapStart, heapLength)); - } - catch (Exception error) - { - logger.Error( - error, - "Error writing entity: {Entity} (Thread: {Thread} - {Start} {Length})", - e, - thread, - heapStart, - heapLength - ); - } - - binPosition += heapLength; } var currentPosition = idx.Position; @@ -146,14 +198,175 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer idx.Seek(currentPosition, SeekOrigin.Begin); } + private long WriteSegmentRecords( + SerializationThreadWorker worker, in SerializedSegment segment, FileBufferWriter idx, FileStream binFs, + long binPosition, ref int entityCount + ) + { + var lengths = worker.Lengths; + var lengthIndex = segment.LengthsStart; + + var heapPos = (int)segment.HeapStart; + var spanStart = heapPos; + + if (segment.SlotOffset >= 0) + { + // Re-walk the same slots the worker serialized; occupancy cannot have changed + // because dictionary mutations divert to the pending queues until PostWorldSave. + var entries = Unsafe.As[]>(_entriesSnapshot); + ref var entry = ref Unsafe.Add(ref MemoryMarshal.GetArrayDataReference(entries), segment.SlotOffset); + + for (var i = 0; i < segment.SlotCount; i++, entry = ref Unsafe.Add(ref entry, 1)) + { + var entity = entry.Value; + if (entity == null) + { + continue; + } + + var length = lengths[lengthIndex++]; + + if (entity is Item { SkipSerialization: true } or Mobile { SkipSerialization: true }) + { + // The bytes exist in the heap but are not part of the save: split the span. + if (heapPos > spanStart) + { + binFs.Write(worker.GetHeap(spanStart, heapPos - spanStart)); + } + + heapPos += length; + spanStart = heapPos; + continue; + } + + idx.Write(GetTypeIndex(entity)); + idx.Write(entity.Serial); + idx.Write(entity.Created.Ticks); + idx.Write(binPosition); + idx.Write(length); + + binPosition += length; + heapPos += length; + entityCount++; + } + } + else + { + var bufferEntities = worker.BufferEntities; + + for (var i = 0; i < segment.RecordCount; i++) + { + var entity = (T)bufferEntities[segment.EntitiesStart + i]; + var length = lengths[lengthIndex++]; + + if (entity is Item { SkipSerialization: true } or Mobile { SkipSerialization: true }) + { + if (heapPos > spanStart) + { + binFs.Write(worker.GetHeap(spanStart, heapPos - spanStart)); + } + + heapPos += length; + spanStart = heapPos; + continue; + } + + idx.Write(GetTypeIndex(entity)); + idx.Write(entity.Serial); + idx.Write(entity.Created.Ticks); + idx.Write(binPosition); + idx.Write(length); + + binPosition += length; + heapPos += length; + entityCount++; + } + } + + if (heapPos > spanStart) + { + binFs.Write(worker.GetHeap(spanStart, heapPos - spanStart)); + } + + return binPosition; + } + + private ushort GetTypeIndex(T entity) + { + // Every path into EntitiesBySerial registers the type first, so this cannot fire. + // If it ever does, the segment-level catch in WriteSnapshot logs it and moves on — + // the failed segment's records are dropped from the idx while binPosition rewinds, + // so treat any occurrence as a serious bug in an insertion path, not a bad entity. + if (!_typeIndexes.TryGetValue(entity.GetType(), out var typeIndex)) + { + throw new InvalidOperationException( + $"{entity.GetType()} was serialized but never registered; entities must enter {Name} through AddEntity." + ); + } + + return typeIndex; + } + public override void Serialize() { + // 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. Kept branch-free: + // a bare loop is ~2.3x faster than one carrying per-entity logic, and multi-megabyte + // entities are rare enough that riding inside a shared chunk is an acceptable tail. foreach (var entity in EntitiesBySerial.Values) { World.PushToCache(entity); } + } - World.PushToCache(this); + 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, List lengths, 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) + { + var start = writer.Position; + entity.Serialize(writer); + lengths.Add((int)(writer.Position - start)); + serialized++; + } + } + + return serialized; } private static ConstructorInfo GetConstructorFor(string typeName, Type t, Type[] constructorTypes) @@ -214,7 +427,16 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer { // Legacy didn't have the null flag check var typeName = dataReader.ReadStringRaw(); - types.Add((ulong)i, GetConstructorFor(typeName, AssemblyHandler.FindTypeByName(typeName), ctorArguments)); + var type = AssemblyHandler.FindTypeByName(typeName); + var ctor = GetConstructorFor(typeName, type, ctorArguments); + + if (ctor != null) + { + // Keep the type table complete so the next (v4) save can index it. + RegisterType(type); + } + + types.Add((ulong)i, ctor); } accessor.SafeMemoryMappedViewHandle.ReleasePointer(); @@ -272,50 +494,121 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer var version = dataReader.ReadInt(); - var ctors = version < 2 ? ReadTypes(Path.GetDirectoryName(filePath)) : []; - - if (typesDb == null && ctors.Count == 0) + if (version >= 4) { - return; + DeserializeIndexesV4(dataReader, entities); + } + else + { + var ctors = version < 2 ? ReadTypes(Path.GetDirectoryName(filePath)) : []; + + if (typesDb == null && ctors.Count == 0) + { + accessor.SafeMemoryMappedViewHandle.ReleasePointer(); + return; + } + + var now = DateTime.UtcNow; + var ctorArgs = new object[1]; + Type[] ctorArguments = [typeof(Serial)]; + + var count = dataReader.ReadInt(); + + for (var i = 0; i < count; ++i) + { + ulong hash; + // Version 2 & 3 with SerializedTypes.db + if (version >= 2) + { + var flag = dataReader.ReadByte(); + if (flag != 2) + { + throw new Exception($"Invalid type flag, expected 2 but received {flag}."); + } + + hash = dataReader.ReadULong(); + } + else + { + hash = (ulong)dataReader.ReadInt(); // Legacy RunUO tdb index + } + + if (!ctors.TryGetValue(hash, out var ctor) && typesDb?.TryGetValue(hash, out var typeName) == true) + { + var type = AssemblyHandler.FindTypeByHash(hash); + ctors[hash] = ctor = GetConstructorFor(typeName, type, ctorArguments); + + if (ctor != null) + { + // Keep the type table complete so the next (v4) save can index it. + RegisterType(type); + } + } + + var serial = (Serial)dataReader.ReadUInt(); + var created = version == 0 ? now : new DateTime(dataReader.ReadLong(), DateTimeKind.Utc); + if (version is > 0 and < 3) + { + dataReader.ReadLong(); // LastSerialized + } + + var pos = dataReader.ReadLong(); + var length = dataReader.ReadInt(); + + if (ctor == null) + { + continue; + } + + ctorArgs[0] = serial; + + if (ctor.Invoke(ctorArgs) is T entity) + { + entity.Created = created; + entities.Add(new EntitySpan(entity, pos, length)); + EntitiesBySerial[serial] = entity; + } + } } - var now = DateTime.UtcNow; - var ctorArgs = new object[1]; + accessor.SafeMemoryMappedViewHandle.ReleasePointer(); + + if (EntitiesBySerial.Count > 0) + { + _lastEntitySerial = EntitiesBySerial.Keys.Max(); + } + } + + private void DeserializeIndexesV4(UnmanagedDataReader dataReader, List> entities) + { Type[] ctorArguments = [typeof(Serial)]; + var typeCount = dataReader.ReadInt(); + var ctors = new ConstructorInfo[typeCount]; + + for (var i = 0; i < typeCount; i++) + { + var typeName = dataReader.ReadStringRaw(); + var type = AssemblyHandler.FindTypeByHash(HashUtility.ComputeHash64(typeName)); + var ctor = GetConstructorFor(typeName, type, ctorArguments); + + if (ctor != null) + { + // Keep the type table complete so the next save can index it. + RegisterType(type); + } + + ctors[i] = ctor; + } + + var ctorArgs = new object[1]; var count = dataReader.ReadInt(); for (var i = 0; i < count; ++i) { - ulong hash; - // Version 2 & 3 with SerializedTypes.db - if (version >= 2) - { - var flag = dataReader.ReadByte(); - if (flag != 2) - { - throw new Exception($"Invalid type flag, expected 2 but received {flag}."); - } - - hash = dataReader.ReadULong(); - } - else - { - hash = (ulong)dataReader.ReadInt(); // Legacy RunUO tdb index - } - - if (!ctors.TryGetValue(hash, out var ctor) && typesDb?.TryGetValue(hash, out var typeName) == true) - { - ctors[hash] = ctor = GetConstructorFor(typeName, AssemblyHandler.FindTypeByHash(hash), ctorArguments); - } - + var ctor = ctors[dataReader.ReadUShort()]; var serial = (Serial)dataReader.ReadUInt(); - var created = version == 0 ? now : new DateTime(dataReader.ReadLong(), DateTimeKind.Utc); - if (version is > 0 and < 3) - { - dataReader.ReadLong(); // LastSerialized - } - + var created = new DateTime(dataReader.ReadLong(), DateTimeKind.Utc); var pos = dataReader.ReadLong(); var length = dataReader.ReadInt(); @@ -333,13 +626,6 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer EntitiesBySerial[serial] = entity; } } - - accessor.SafeMemoryMappedViewHandle.ReleasePointer(); - - if (EntitiesBySerial.Count > 0) - { - _lastEntitySerial = EntitiesBySerial.Keys.Max(); - } } public override void Deserialize(string savePath, Dictionary typesDb) @@ -478,6 +764,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(); } @@ -555,6 +843,7 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer case WorldState.PendingSave: case WorldState.Running: { + RegisterType(entity.GetType()); ref var entityEntry = ref CollectionsMarshal.GetValueRefOrAddDefault(EntitiesBySerial, entity.Serial, out var exists); if (exists) { diff --git a/Projects/Server/Serialization/GenericPersistence.cs b/Projects/Server/Serialization/GenericPersistence.cs index 0afd64210..5b8e79c29 100644 --- a/Projects/Server/Serialization/GenericPersistence.cs +++ b/Projects/Server/Serialization/GenericPersistence.cs @@ -25,9 +25,24 @@ public abstract class GenericPersistence : Persistence, IGenericSerializable public string Name { get; } public string SaveFilePath { get; protected set; } // "/.bin" - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } + // Placement of the self-payload in the worker heaps for the most recent save. Only + // persistences carry placement state — entities are located through the per-worker + // segment logs instead. + private protected byte _selfThread; + private protected int _selfPosition; + private protected int _selfLength; + + internal void SetSelfPlacement(byte thread, int position, int length) + { + _selfThread = thread; + _selfPosition = position; + _selfLength = length; + } + + private long _loadedFileLength; + + // Scheduling estimate only: previous save's payload size, or the loaded file size before the first save. + internal long EstimatedSize => _selfLength > 0 ? _selfLength : _loadedFileLength; public GenericPersistence(string name, int priority) : base(priority) { @@ -37,12 +52,14 @@ public abstract class GenericPersistence : Persistence, IGenericSerializable public override void Serialize() { - World.PushToCache(this); + // Always a dedicated chunk: self-payloads can be arbitrarily large and must not + // ride inside a shared chunk where one worker would serialize them plus the chunk. + World.PushSingleToCache(this); } - public override void WriteSnapshot(string savePath, HashSet typeSet) + public override void WriteSnapshot(string savePath) { - if (SerializedLength == 0) + if (_selfLength == 0) { return; } @@ -55,11 +72,7 @@ public abstract class GenericPersistence : Persistence, IGenericSerializable using var binFs = new FileStream(file, FileMode.Create, FileAccess.Write, FileShare.None); - var thread = SerializedThread; - var heapStart = SerializedPosition; - var heapLength = SerializedLength; - - binFs.Write(threads[thread].GetHeap(heapStart, heapLength)); + binFs.Write(threads[_selfThread].GetHeap(_selfPosition, _selfLength)); } public override unsafe void Deserialize(string savePath, Dictionary typesDb) @@ -74,6 +87,7 @@ public abstract class GenericPersistence : Persistence, IGenericSerializable } var fileLength = file.Length; + _loadedFileLength = fileLength; string error; diff --git a/Projects/Server/Serialization/IGenericSerializable.cs b/Projects/Server/Serialization/IGenericSerializable.cs index c1cef38c7..d53b1c03d 100644 --- a/Projects/Server/Serialization/IGenericSerializable.cs +++ b/Projects/Server/Serialization/IGenericSerializable.cs @@ -17,9 +17,5 @@ namespace Server; public interface IGenericSerializable { - byte SerializedThread { get; set; } - int SerializedPosition { get; set; } - int SerializedLength { get; set; } - void Serialize(IGenericWriter writer); } diff --git a/Projects/Server/Serialization/IGenericWriter.cs b/Projects/Server/Serialization/IGenericWriter.cs index ed1cf80b4..9c9563139 100644 --- a/Projects/Server/Serialization/IGenericWriter.cs +++ b/Projects/Server/Serialization/IGenericWriter.cs @@ -14,7 +14,6 @@ *************************************************************************/ using System; -using System.Buffers; using System.Collections; using System.IO; using System.Net; @@ -39,168 +38,24 @@ public interface IGenericWriter void Write(Serial serial); void Write(Type type); void Write(decimal value); - - void Write(DateTime value) - { - // If DateTimeKind is Unspecified, we can't assume it needs to be converted. - if (value.Kind == DateTimeKind.Local) - { - value = value.ToUniversalTime(); - } - - Write(value.Ticks); - } - void WriteDeltaTime(DateTime value) - { - if (value == DateTime.MinValue) - { - Write(long.MinValue); - return; - } - - if (value == DateTime.MaxValue) - { - Write(long.MaxValue); - return; - } - - if (value.Kind == DateTimeKind.Local) - { - value = value.ToUniversalTime(); - } - - // Technically supports negative deltas for times in the past - Write(value.Ticks - DateTime.UtcNow.Ticks); - } - void Write(IPAddress value) - { - Span stack = stackalloc byte[16]; - value.TryWriteBytes(stack, out var bytesWritten); - Write((byte)bytesWritten); - Write(stack[..bytesWritten]); - } - - void Write(TimeSpan value) - { - Write(value.Ticks); - } - - void WriteEncodedInt(int value) - { - var v = (uint)value; - - while (v >= 0x80) - { - Write((byte)(v | 0x80)); - v >>= 7; - } - - Write((byte)v); - } - - void Write(Point3D value) - { - Write(value.m_X); - Write(value.m_Y); - Write(value.m_Z); - } - void Write(Point2D value) - { - Write(value.m_X); - Write(value.m_Y); - } - void Write(Rectangle2D value) - { - Write(value.Start); - Write(value.End); - } - void Write(Rectangle3D value) - { - Write(value.Start); - Write(value.End); - } - void Write(Map value) => Write((byte)(value?.MapIndex ?? 0xFF)); - void Write(Race value) => Write((byte)(value?.RaceIndex ?? 0xFF)); + void WriteEncodedInt(int value); + void Write(DateTime value); + void WriteDeltaTime(DateTime value); + void Write(IPAddress value); + void Write(TimeSpan value); + void Write(Point3D value); + void Write(Point2D value); + void Write(Rectangle2D value); + void Write(Rectangle3D value); + void Write(Map value); + void Write(Race value); void Write(byte[] bytes); void Write(byte[] bytes, int offset, int count); void Write(ReadOnlySpan bytes); - unsafe void WriteEnum(T value) where T : unmanaged, Enum - { - switch (sizeof(T)) - { - default: - { - throw new ArgumentException($"Argument of type {typeof(T)} is not a normal enum"); - } - case 1: - { - Write(*(byte*)&value); - break; - } - case 2: - { - Write(*(ushort*)&value); - break; - } - case 4: - { - WriteEncodedInt(*(int*)&value); - break; - } - case 8: - { - Write(*(ulong*)&value); - break; - } - } - } - void Write(Guid guid) - { - Span stack = stackalloc byte[16]; - guid.TryWriteBytes(stack); - Write(stack); - } - - public void Write(BitArray bitArray) - { - var bitLength = bitArray.Length; - var byteLength = (bitLength + 7) / 8; - - WriteEncodedInt(bitLength); - - var arrayBuffer = ArrayPool.Shared.Rent(byteLength); - try - { - bitArray.CopyTo(arrayBuffer, 0); - Write(arrayBuffer.AsSpan(0, byteLength)); - } - finally - { - ArrayPool.Shared.Return(arrayBuffer); - } - } - - void Write(TextDefinition def) - { - if (def == null) - { - WriteEncodedInt(3); - } - else if (def.Number > 0) - { - WriteEncodedInt(1); - WriteEncodedInt(def.Number); - } - else if (def.String != null) - { - WriteEncodedInt(2); - Write(def.String); - } - else - { - WriteEncodedInt(0); // Empty - } - } + void WriteEnum(T value) where T : unmanaged, Enum; + void Write(Guid guid); + void Write(BitArray bitArray); + void Write(TextDefinition def); long Seek(long offset, SeekOrigin origin); } diff --git a/Projects/Server/Serialization/MemoryMapFileWriter.cs b/Projects/Server/Serialization/MemoryMapFileWriter.cs deleted file mode 100644 index 55866dd85..000000000 --- a/Projects/Server/Serialization/MemoryMapFileWriter.cs +++ /dev/null @@ -1,304 +0,0 @@ -/************************************************************************* - * ModernUO * - * Copyright 2019-2026 - ModernUO Development Team * - * Email: hi@modernuo.com * - * File: MemoryMapFileWriter.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.Buffers.Binary; -using System.Collections.Generic; -using System.IO; -using System.IO.MemoryMappedFiles; -using System.Runtime.CompilerServices; -using System.Runtime.InteropServices; -using System.Text; -using Server.Text; - -namespace Server; - -public unsafe class MemoryMapFileWriter : IGenericWriter, IDisposable -{ - private readonly Encoding _encoding; - - private readonly HashSet _types; - private readonly FileStream _fileStream; - private MemoryMappedFile _mmf; - private MemoryMappedViewAccessor _accessor; - private byte* _ptr; - private long _position; - private long _size; - - public MemoryMapFileWriter(FileStream fileStream, long initialSize, HashSet types = null) - { - _types = types; - _fileStream = fileStream; - _encoding = TextEncoding.UTF8; - _size = Math.Max(initialSize, 1024); - - ResizeMemoryMappedFile(initialSize); - } - - public long Position => _position; - - public FileStream FileStream => _fileStream; - - private void ResizeMemoryMappedFile(long newSize) - { - _accessor?.SafeMemoryMappedViewHandle.ReleasePointer(); - _accessor?.Dispose(); - _mmf?.Dispose(); - - // Do the actual resizing - _fileStream.SetLength(newSize); - - _mmf = MemoryMappedFile.CreateFromFile(_fileStream, null, newSize, MemoryMappedFileAccess.ReadWrite, HandleInheritability.None, leaveOpen: true); - _accessor = _mmf.CreateViewAccessor(); - _accessor.SafeMemoryMappedViewHandle.AcquirePointer(ref _ptr); - } - - private void EnsureCapacity(long bytesToWrite) - { - var shouldResize = false; - while (_position + bytesToWrite > _size) - { - // Don't double forever, eventually we want to have a maximum, like 256MB at a time or something - _size += Math.Min(_size, 1024 * 1024 * 256); - shouldResize = true; - } - - if (shouldResize) - { - ResizeMemoryMappedFile(_size); - } - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(byte[] bytes) => Write(bytes.AsSpan()); - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(byte[] bytes, int offset, int count) => Write(bytes.AsSpan(offset, count)); - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(ReadOnlySpan bytes) - { - var byteCount = bytes.Length; - EnsureCapacity(byteCount); - - bytes.CopyTo(new Span(_ptr + _position, byteCount)); - _position += byteCount; - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public virtual long Seek(long offset, SeekOrigin origin) - { - switch (origin) - { - case SeekOrigin.Begin: - { - if (offset > _size) - { - EnsureCapacity(offset); - } - - _position = offset; - break; - } - case SeekOrigin.Current: - { - EnsureCapacity(offset); - _position += offset; - break; - } - case SeekOrigin.End: - { - if (_position + offset > _size) - { - EnsureCapacity(offset); - } - - _position = _size + offset; - - if (_position < 0) - { - Dispose(); - throw new InvalidOperationException("Seek before start of file"); - } - break; - } - } - - return _position; - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(string value) - { - if (value == null) - { - Write(false); - } - else - { - Write(true); - WriteStringRaw(value); - } - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(long value) - { - EnsureCapacity(sizeof(long)); - BinaryPrimitives.WriteInt64LittleEndian(new Span(_ptr + _position, sizeof(long)), value); - _position += sizeof(long); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(ulong value) - { - EnsureCapacity(sizeof(ulong)); - BinaryPrimitives.WriteUInt64LittleEndian(new Span(_ptr + _position, sizeof(ulong)), value); - _position += sizeof(ulong); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(int value) - { - EnsureCapacity(sizeof(int)); - BinaryPrimitives.WriteInt32LittleEndian(new Span(_ptr + _position, sizeof(int)), value); - _position += sizeof(int); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(uint value) - { - EnsureCapacity(sizeof(uint)); - BinaryPrimitives.WriteUInt32LittleEndian(new Span(_ptr + _position, sizeof(uint)), value); - _position += sizeof(uint); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(short value) - { - EnsureCapacity(sizeof(short)); - BinaryPrimitives.WriteInt16LittleEndian(new Span(_ptr + _position, sizeof(short)), value); - _position += sizeof(short); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(ushort value) - { - EnsureCapacity(sizeof(ushort)); - BinaryPrimitives.WriteUInt16LittleEndian(new Span(_ptr + _position, sizeof(ushort)), value); - _position += sizeof(ushort); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(double value) - { - EnsureCapacity(sizeof(double)); - BinaryPrimitives.WriteDoubleLittleEndian(new Span(_ptr + _position, sizeof(double)), value); - _position += sizeof(double); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(float value) - { - EnsureCapacity(sizeof(float)); - BinaryPrimitives.WriteSingleLittleEndian(new Span(_ptr + _position, sizeof(float)), value); - _position += sizeof(float); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(byte value) - { - EnsureCapacity(1); - *(_ptr + _position++) = value; - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(sbyte value) => Write((byte)value); - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(bool value) => Write(*(byte*)&value); - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(Serial serial) => Write(serial.Value); - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(Type type) - { - if (type == null) - { - Write((byte)0); - } - else - { - Write((byte)0x2); // xxHash3 64bit - Write(AssemblyHandler.GetTypeHash(type)); - _types.Add(type); - } - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Write(decimal value) - { - Span buffer = stackalloc int[sizeof(decimal) / 4]; - decimal.GetBits(value, buffer); - - Write(MemoryMarshal.Cast(buffer)); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void WriteStringRaw(ReadOnlySpan value) - { - var length = _encoding.GetByteCount(value); - - EnsureCapacity(length + 5); - - // WriteEncodedInt - var v = (uint)length; - - while (v >= 0x80) - { - *(_ptr + _position++) = (byte)(v | 0x80); - v >>= 7; - } - *(_ptr + _position++) = (byte)v; - - _encoding.GetBytes(value, new Span(_ptr + _position, length)); - _position += length; - } - - private void Dispose(bool disposing) - { - if (disposing) - { - _accessor.SafeMemoryMappedViewHandle.ReleasePointer(); - _accessor.Dispose(); - _mmf.Dispose(); - - // Truncate the file - _fileStream.SetLength(_position); - } - } - - public void Dispose() - { - Dispose(true); - GC.SuppressFinalize(this); - } - - ~MemoryMapFileWriter() - { - Dispose(false); - } -} diff --git a/Projects/Server/Serialization/Persistence.cs b/Projects/Server/Serialization/Persistence.cs index 85fbb3c17..35d1d7514 100644 --- a/Projects/Server/Serialization/Persistence.cs +++ b/Projects/Server/Serialization/Persistence.cs @@ -84,41 +84,35 @@ public abstract class Persistence } // Note: This is strictly on a background thread - internal static void WriteSnapshotAll(string path, HashSet typeSet) + internal static void WriteSnapshotAll(string path) { foreach (var p in _registry) { - p.WriteSnapshot(path, typeSet); - } - - WriteSerializedTypesSnapshot(path, typeSet); - } - - public static void WriteSerializedTypesSnapshot(string path, HashSet types) - { - var typesPath = Path.Combine(path, "SerializedTypes.db"); - using var fs = new FileStream(typesPath, FileMode.Create); - using var writer = new MemoryMapFileWriter(fs, 1024 * 1024 * 4); - - writer.Write(0); // version - writer.Write(types.Count); - - foreach (var type in types) - { - var fullName = type.FullName; - writer.Write(HashUtility.ComputeHash64(fullName)); - writer.WriteStringRaw(fullName); + p.WriteSnapshot(path); } } internal static void SerializeAll() { - foreach (var p in _registry) + // Largest known payloads first (LPT scheduling). An indivisible multi-megabyte system + // serialized last extends the freeze by its entire duration; serialized first it overlaps + // the entity stream. Sizes come from the previous save, or the loaded files on first save. + var ordered = new Persistence[_registry.Count]; + _registry.CopyTo(ordered); + Array.Sort(ordered, static (a, b) => GetEstimatedSize(b).CompareTo(GetEstimatedSize(a))); + + foreach (var p in ordered) { + // Chunks are persistence-homogeneous: publishing the partial chunk at each + // boundary lets workers attribute buffer-chunk records to their owner without + // any per-entity state. + World.SetChunkSourceOwner(p); p.Serialize(); } } + private static long GetEstimatedSize(Persistence p) => (p as GenericPersistence)?.EstimatedSize ?? 0; + internal static void PostWorldSaveAll() { foreach (var p in _registry) @@ -136,7 +130,7 @@ public abstract class Persistence } // Note: This should only be run on a background thread - public abstract void WriteSnapshot(string savePath, HashSet typeSet); + public abstract void WriteSnapshot(string savePath); public abstract void Serialize(); diff --git a/Projects/Server/Serialization/SerializationChunkSource.cs b/Projects/Server/Serialization/SerializationChunkSource.cs new file mode 100644 index 000000000..8b43304b3 --- /dev/null +++ b/Projects/Server/Serialization/SerializationChunkSource.cs @@ -0,0 +1,171 @@ +/************************************************************************* + * ModernUO * + * Copyright 2019-2026 - ModernUO Development Team * + * Email: hi@modernuo.com * + * File: SerializationChunkSource.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.Concurrent; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +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, appending + /// each record's byte length to . Returns the number serialized. + /// + int SerializeRange(BufferWriter writer, List lengths, 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 +/// so the per-entity cost is a plain array store instead of a synchronized enqueue, and +/// workers pull whole chunks so they naturally load-balance: a worker busy with a thick +/// entity simply takes fewer chunks. +/// Persistence self-payloads are published as dedicated single-entity chunks so large +/// systems 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 +{ + // 4096 refs (32KB per chunk) keeps producer sync cost at one enqueue per 4096 entities + // while the drain tail stays sub-millisecond. + private const int ChunkCapacity = 4096; + + internal readonly struct Chunk + { + public readonly GenericPersistence Single; + public readonly IGenericSerializable[] Buffer; + public readonly ISlotRangeSource Source; + public readonly Persistence Owner; // buffer chunks only; ranges use Source, singles record their own placement + public readonly int Offset; + public readonly int Count; + + public Chunk(GenericPersistence single) + { + Single = single; + Count = 1; + } + + public Chunk(IGenericSerializable[] buffer, int count, Persistence owner) + { + Buffer = buffer; + Count = count; + Owner = owner; + } + + public Chunk(ISlotRangeSource source, int offset, int count) + { + Source = source; + Offset = offset; + Count = count; + } + } + + private readonly ConcurrentQueue _chunks = new(); + private readonly ConcurrentQueue _pool = new(); + + // Producer state - written only by the game loop thread. + private IGenericSerializable[] _current; + private int _count; + private Persistence _currentOwner; + + /// + /// Declares the owner of subsequently pushed entities. Publishes the partial chunk when + /// the owner changes, keeping buffer chunks persistence-homogeneous so workers can + /// attribute their serialized records to a persistence without any per-entity state. + /// + public void SetOwner(Persistence owner) + { + if (!ReferenceEquals(_currentOwner, owner)) + { + Flush(); + _currentOwner = owner; + } + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Push(IGenericSerializable entity) + { + var current = _current ??= Rent(); + + // Ref store skips the bounds and array-covariance checks. Safe by construction: + // _count is producer-thread-only and always < ChunkCapacity here (reset on publish), + // and the array's element type is exactly IGenericSerializable. + Unsafe.Add(ref MemoryMarshal.GetArrayDataReference(current), _count) = entity; + + if (++_count == ChunkCapacity) + { + _chunks.Enqueue(new Chunk(current, ChunkCapacity, _currentOwner)); + _current = null; + _count = 0; + } + } + + /// + /// Publishes a persistence self-payload as a dedicated chunk regardless of its + /// estimated size — it can be large on the first save before an estimate exists. + /// The worker records the payload's placement on the persistence itself. + /// + public void PushSingle(GenericPersistence persistence) => _chunks.Enqueue(new Chunk(persistence)); + + /// + /// 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. + /// + public void Flush() + { + if (_count > 0) + { + _chunks.Enqueue(new Chunk(_current, _count, _currentOwner)); + _current = null; + _count = 0; + } + } + + internal bool TryTake(out Chunk chunk) => _chunks.TryDequeue(out chunk); + + internal void Return(IGenericSerializable[] buffer, int count) + { + // Clear so pooled chunks don't keep entities reachable between saves. + Array.Clear(buffer, 0, count); + _pool.Enqueue(buffer); + } + + private IGenericSerializable[] Rent() => + _pool.TryDequeue(out var buffer) ? buffer : new IGenericSerializable[ChunkCapacity]; +} diff --git a/Projects/Server/Serialization/SerializationThreadWorker.cs b/Projects/Server/Serialization/SerializationThreadWorker.cs index bb0e67783..5a9b5ed1c 100644 --- a/Projects/Server/Serialization/SerializationThreadWorker.cs +++ b/Projects/Server/Serialization/SerializationThreadWorker.cs @@ -14,12 +14,44 @@ *************************************************************************/ using System; -using System.Collections.Concurrent; +using System.Collections.Generic; using System.Runtime.CompilerServices; using System.Threading; namespace Server; +/// +/// One contiguous run of records a worker serialized into its heap from a single chunk. +/// Together with the worker's lengths log this replaces per-entity placement state: +/// positions are implicit (a worker's writes are contiguous), and identity comes from +/// re-walking the same slots for range segments, or from the entities log for buffer +/// segments. The snapshot writer routes segments to files by . +/// +internal readonly struct SerializedSegment +{ + public readonly object Owner; // ISlotRangeSource for range segments, Persistence for buffer segments + public readonly int SlotOffset; // -1 when the segment came from a buffer chunk + public readonly int SlotCount; + public readonly long HeapStart; + public readonly int LengthsStart; + public readonly int RecordCount; + public readonly int EntitiesStart; // buffer segments only + + public SerializedSegment( + object owner, int slotOffset, int slotCount, long heapStart, int lengthsStart, int recordCount, + int entitiesStart + ) + { + Owner = owner; + SlotOffset = slotOffset; + SlotCount = slotCount; + HeapStart = heapStart; + LengthsStart = lengthsStart; + RecordCount = recordCount; + EntitiesStart = entitiesStart; + } +} + public class SerializationThreadWorker { private const int MinHeapSize = 1024 * 1024; // 1MB @@ -27,23 +59,69 @@ public class SerializationThreadWorker private readonly Thread _thread; private readonly AutoResetEvent _startEvent; // Main thread tells the thread to start working private readonly AutoResetEvent _stopEvent; // Main thread waits for the worker finish draining + private readonly SerializationChunkSource _chunkSource; + private readonly int _heapSizeHint; private bool _pause; private bool _exit; private bool _exited; private byte[] _heap; + private long _entitiesSerialized; + private long _bytesSerialized; - private readonly ConcurrentQueue _entities; + // What this worker serialized where, logged during the drain and consumed by + // WriteSnapshot on the background writer thread. Cleared once the snapshot is on disk. + private readonly List _segments = []; + private readonly List _lengths = []; + private readonly List _bufferEntities = []; - public SerializationThreadWorker(int index) + internal List Segments => _segments; + internal List Lengths => _lengths; + internal List BufferEntities => _bufferEntities; + + /// + /// Releases the write logs after the snapshot is written so serialized entity + /// references don't linger between saves. Capacity is retained: the logs regrow to + /// the same size every save. + /// + internal void ReleaseWriteLogs() + { + _segments.Clear(); + _lengths.Clear(); + _bufferEntities.Clear(); + } + + 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; - _startEvent = new AutoResetEvent(false); - _stopEvent = new AutoResetEvent(false); - _entities = new ConcurrentQueue(); - _thread = new Thread(Execute); - _thread.Start(this); + _chunkSource = chunkSource; + _heapSizeHint = heapSizeHint; + + 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; + public void Wake() { _startEvent.Set(); @@ -68,50 +146,139 @@ public class SerializationThreadWorker Sleep(); } - public void AllocateHeap() => _heap ??= GC.AllocateUninitializedArray(MinHeapSize); // 1MB - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void Push(IGenericSerializable entity) => _entities.Enqueue(entity); + // Sized from the previous world load so the first save doesn't pay copy-on-grow during the freeze. + public void AllocateHeap() => + _heap ??= GC.AllocateUninitializedArray(Math.Max(MinHeapSize, _heapSizeHint)); [MethodImpl(MethodImplOptions.AggressiveInlining)] public ReadOnlySpan GetHeap(int start, int length) => _heap.AsSpan(start, length); + private long ProcessChunk(in SerializationChunkSource.Chunk chunk, BufferWriter writer) + { + if (chunk.Single != null) + { + // Self-payloads are written to their own file, so they record placement on the + // persistence itself instead of the segment logs. + var start = writer.Position; + chunk.Single.Serialize(writer); + chunk.Single.SetSelfPlacement((byte)_index, (int)start, (int)(writer.Position - start)); + return 1; + } + + if (chunk.Source != null) + { + var heapStart = writer.Position; + var lengthsStart = _lengths.Count; + var serialized = chunk.Source.SerializeRange(writer, _lengths, chunk.Offset, chunk.Count); + + if (serialized > 0) + { + _segments.Add( + new SerializedSegment(chunk.Source, chunk.Offset, chunk.Count, heapStart, lengthsStart, serialized, -1) + ); + } + + return serialized; + } + + var buffer = chunk.Buffer; + var count = chunk.Count; + + var bufferHeapStart = writer.Position; + var bufferLengthsStart = _lengths.Count; + var entitiesStart = _bufferEntities.Count; + + for (var i = 0; i < count; i++) + { + var e = buffer[i]; + var start = writer.Position; + e.Serialize(writer); + _lengths.Add((int)(writer.Position - start)); + _bufferEntities.Add(e); + } + + _segments.Add( + new SerializedSegment(chunk.Owner, -1, 0, bufferHeapStart, bufferLengthsStart, count, entitiesStart) + ); + + _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() + { + ReleaseWriteLogs(); + + var writer = new BufferWriter(_heap, true); + var entities = 0L; + + while (_chunkSource.TryTake(out var chunk)) + { + entities += ProcessChunk(in chunk, writer); + } + + _heap = writer.Buffer; + _entitiesSerialized = entities; + _bytesSerialized = writer.Position; + + writer.Close(); + } + private static void Execute(object obj) { var worker = (SerializationThreadWorker)obj; - var threadIndex = (byte)worker._index; - var queue = worker._entities; - var serializedTypes = World.SerializedTypes; + var chunkSource = worker._chunkSource; while (worker._startEvent.WaitOne()) { - var writer = new BufferWriter(worker._heap, true, serializedTypes); + worker.ReleaseWriteLogs(); + + var writer = new BufferWriter(worker._heap, true); + var entities = 0L; + var spinner = new SpinWait(); while (true) { var pauseRequested = Volatile.Read(ref worker._pause); - if (queue.TryDequeue(out var e)) + if (chunkSource.TryTake(out var chunk)) { - e.SerializedThread = threadIndex; - var start = e.SerializedPosition = (int)writer.Position; - e.Serialize(writer); - e.SerializedLength = (int)(writer.Position - start); + spinner.Reset(); + entities += worker.ProcessChunk(in chunk, writer); } else if (pauseRequested) // Break when finished { break; } + else + { + // Idle backoff instead of hammering the queue head while the producer works. + // sleep1Threshold: -1 keeps escalation at Yield/Sleep(0) and never Sleep(1), + // avoiding timer-resolution stalls at the end of the drain. + spinner.SpinOnce(-1); + } } worker._heap = writer.Buffer; + worker._entitiesSerialized = entities; + worker._bytesSerialized = writer.Position; writer.Close(); - worker._stopEvent.Set(); // Allow the main thread to continue now that we are finished - worker._pause = false; + // The owning thread may start another pause cycle the moment _stopEvent is set + // (Exit does exactly that). Clear _pause and sample the exit condition before + // signaling, or the new cycle's pause request is clobbered / its Sleep orphaned. + var exiting = Core.Closing || worker._exit; + Volatile.Write(ref worker._pause, false); - if (Core.Closing || worker._exit) + worker._stopEvent.Set(); // Allow the main thread to continue now that we are finished + + if (exiting) { return; } 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 f25ec163b..2c9e6d216 100644 --- a/Projects/Server/World/World.cs +++ b/Projects/Server/World/World.cs @@ -14,7 +14,6 @@ *************************************************************************/ using System; -using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.IO; @@ -44,8 +43,11 @@ public static class World private static readonly MobilePersistence _mobilePersistence = new(); private static readonly GenericEntityPersistence _guildPersistence = new("Guilds", 3, 1, 0x7FFFFFFF); - private static int _threadId; + // 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); private static string _tempSavePath; // Path to the temporary folder for the save @@ -195,48 +197,46 @@ 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]; - for (var i = 0; i < _threadWorkers.Length; i++) + // 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() / _threadWorkers.Length * 5 / 4, 1024 * 1024 * 1024); + + for (var i = 0; i < _realWorkerCount; i++) { - _threadWorkers[i] = new SerializationThreadWorker(i); + _threadWorkers[i] = new SerializationThreadWorker(i, _chunkSource, heapSizeHint); } + + _threadWorkers[_realWorkerCount] = + SerializationThreadWorker.CreateInline(_realWorkerCount, _chunkSource, heapSizeHint); } - /** - * Duplicates can be weeded out asynchronously while flushing - * If performance becomes a problem, we need to build a dual mode concurrent array. - * - ****************************************************** Proposal ****************************************************** - * The structure is initialized with a large capacity to avoid unnecessary resizing. - * Write Mode: - * - Multiple threads can add a single, or a range of elements concurrently. - * - Elements can be Peeked, but there are no guarantees. - * - To resize the internal array, replaced it with the next size up from an array pool. - * - The structure cannot be cleared in this mode. - * - * Read Mode: - * - The array can be read from multiple threads using a ref struct enumerator. - * - Elements cannot be added or reassigned. - * - Cleared by replacing the internal array with another one from the pool. - * - Note: Upon clearing, the existing array is not sent back to the pool until there are zero enumerators. - * - * Enumeration: - * - Multiple threads can enumerate while in read mode. The enumerator will Interlocked.Increment a read counter. - * - Upon dispose of the enumerator, the read counter will be lowered with an Interlocked.Decrement - * - When the read counter reaches 0, if there is a cleared array, the array is sent back to the pool zeroed. - * - * Notes: - * - Elements can never be removed. - * - * How is this different from ConcurrentQueue? - * The functionality is very similar, except the constraints allow the implementation to be done without locks. - * Since this implementation uses pooled arrays, allocations will approach zero over time. - ********************************************************************************************************************** - */ - public static ConcurrentQueue SerializedTypes { get; } = new(); + private static long GetLoadedSaveSize() + { + try + { + if (!Directory.Exists(SavePath)) + { + return 0; + } + + var total = 0L; + foreach (var file in Directory.EnumerateFiles(SavePath, "*.bin", SearchOption.AllDirectories)) + { + total += new FileInfo(file).Length; + } + + return total; + } + catch + { + return 0; + } + } public static void Save() { @@ -304,6 +304,7 @@ public static class World Persistence.SerializeAll(); PauseSerializationThreads(); + LogWorkerBalance(); EventSink.InvokeWorldSave(); } @@ -332,8 +333,6 @@ public static class World } } - private static readonly HashSet _typesSet = []; - private static void WriteFiles(object state) { var snapshotPath = (string)state; @@ -342,15 +341,7 @@ public static class World var watch = Stopwatch.StartNew(); logger.Information("Writing world save snapshot"); - // Dedupe the types - while (SerializedTypes.TryDequeue(out var type)) - { - _typesSet.Add(type); - } - - Persistence.WriteSnapshotAll(snapshotPath, _typesSet); - - _typesSet.Clear(); + Persistence.WriteSnapshotAll(snapshotPath); try { @@ -374,9 +365,6 @@ public static class World BroadcastStaff(0x35, true, "Writing world save snapshot failed! Check the logs!"); } - // Clear types - SerializedTypes.Clear(); - _diskWriteHandle.Set(); Core.LoopContext.Post(FinishWorldSave); } @@ -386,45 +374,84 @@ public static class World WorldState = WorldState.Running; Persistence.PostWorldSaveAll(); // Process decay and safety queues MovementThrottle.ResetAllMovementTiming(); // Prevent post-save movement rejection bursts + + // The snapshot is on disk; release the per-worker write logs so serialized + // entity references don't linger between saves. + for (var i = 0; i < _threadWorkers.Length; i++) + { + _threadWorkers[i].ReleaseWriteLogs(); + } + } + + // Debug-level output only exists in DEBUG builds (see LogFactory), so Release builds + // should not pay for the stat summing and argument boxing inside the freeze at all. + [Conditional("DEBUG")] + private static void LogWorkerBalance() + { + var totalEntities = 0L; + var totalBytes = 0L; + var minBytes = long.MaxValue; + var maxBytes = 0L; + + for (var i = 0; i < _threadWorkers.Length; i++) + { + var worker = _threadWorkers[i]; + totalEntities += worker.EntitiesSerialized; + var bytes = worker.BytesSerialized; + totalBytes += bytes; + minBytes = Math.Min(minBytes, bytes); + maxBytes = Math.Max(maxBytes, bytes); + } + + logger.Debug( + "Serialized {EntityCount} entities ({ByteCount} bytes) across {WorkerCount} workers (min {MinBytes}, max {MaxBytes} bytes per worker)", + totalEntities, + totalBytes, + _threadWorkers.Length, + minBytes, + maxBytes + ); } [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(); } } - [MethodImpl(MethodImplOptions.AggressiveInlining)] internal static void PauseSerializationThreads() { - for (var i = 0; i < _threadWorkers.Length; i++) + // Publish the partial chunk before the workers are told to finish draining. + _chunkSource.Flush(); + + // 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(); } } [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static int GetThreadWorkerCount() => Math.Max(Environment.ProcessorCount - 1, 1); + internal static void SetChunkSourceOwner(Persistence owner) => _chunkSource.SetOwner(owner); [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static void ResetRoundRobin() => _threadId = 0; + internal static void PushToCache(IGenericSerializable e) => _chunkSource.Push(e); [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static void PushToCache(IGenericSerializable e) - { - _threadWorkers[_threadId++].Push(e); - if (_threadId == _threadWorkers.Length) - { - _threadId = 0; - } - } + internal static void PushSingleToCache(GenericPersistence persistence) => _chunkSource.PushSingle(persistence); + + [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(); } diff --git a/Projects/UOContent/Accounting/Account.cs b/Projects/UOContent/Accounting/Account.cs index d098dc101..d79d70b1a 100644 --- a/Projects/UOContent/Accounting/Account.cs +++ b/Projects/UOContent/Accounting/Account.cs @@ -287,10 +287,6 @@ public partial class Account : IAccount, IComparable public Serial Serial { get; set; } - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - [AfterDeserialization(false)] private void AfterDeserialization() { diff --git a/Projects/UOContent/Engines/Bulk Orders/Books/BaseBOBEntry.cs b/Projects/UOContent/Engines/Bulk Orders/Books/BaseBOBEntry.cs index 163dc3a32..d3bbadb45 100644 --- a/Projects/UOContent/Engines/Bulk Orders/Books/BaseBOBEntry.cs +++ b/Projects/UOContent/Engines/Bulk Orders/Books/BaseBOBEntry.cs @@ -26,10 +26,6 @@ public abstract partial class BaseBOBEntry : IBOBEntry public Serial Serial { get; } - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - public bool Deleted { get; private set; } public BaseBOBEntry() diff --git a/Projects/UOContent/Engines/Ethics/Core/EthicsEntity.cs b/Projects/UOContent/Engines/Ethics/Core/EthicsEntity.cs index aaf5af3be..5ce677cb8 100644 --- a/Projects/UOContent/Engines/Ethics/Core/EthicsEntity.cs +++ b/Projects/UOContent/Engines/Ethics/Core/EthicsEntity.cs @@ -16,10 +16,6 @@ public partial class EthicsEntity : ISerializable public Serial Serial { get; } - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - public bool Deleted { get; private set; } public void Delete()