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/GenericEntityPersistenceRoundTripTests.cs b/Projects/Server.Tests/Tests/Serialization/GenericEntityPersistenceRoundTripTests.cs new file mode 100644 index 000000000..eeedd9241 --- /dev/null +++ b/Projects/Server.Tests/Tests/Serialization/GenericEntityPersistenceRoundTripTests.cs @@ -0,0 +1,165 @@ +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}" + }; + } + + // 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, then the types db. + var typeSet = new HashSet(); + persistence.WriteSnapshot(dir, typeSet); + + var typesDb = new Dictionary(); + foreach (var type in typeSet) + { + typesDb[AssemblyHandler.GetTypeHash(type)] = type.FullName; + } + + persistence.PostWorldSave(); // releases the entries snapshot + + // Load into a fresh persistence, like a server boot would. + loaded = new RoundTripPersistence(2001); + loaded.DeserializeIndexes(dir, typesDb); + loaded.Deserialize(dir, typesDb); + + 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); + } + } + 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/SerializationChunkSourceTests.cs b/Projects/Server.Tests/Tests/Serialization/SerializationChunkSourceTests.cs index 73a9d50eb..3f315e8d9 100644 --- a/Projects/Server.Tests/Tests/Serialization/SerializationChunkSourceTests.cs +++ b/Projects/Server.Tests/Tests/Serialization/SerializationChunkSourceTests.cs @@ -1,3 +1,4 @@ +using System; using System.Collections.Generic; using Xunit; @@ -7,10 +8,6 @@ public class SerializationChunkSourceTests { private class TestEntity : IGenericSerializable { - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } - public int PayloadSize { get; init; } = 16; public void Serialize(IGenericWriter writer) @@ -22,24 +19,40 @@ public class SerializationChunkSourceTests } } + 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)) { - if (chunk.Single != null) + for (var i = 0; i < chunk.Count; i++) { - drained.Add((TestEntity)chunk.Single); + drained.Add((TestEntity)chunk.Buffer[i]); } - else - { - for (var i = 0; i < chunk.Count; i++) - { - drained.Add((TestEntity)chunk.Buffer[i]); - } - source.Return(chunk.Buffer, chunk.Count); - } + source.Return(chunk.Buffer, chunk.Count); } return drained; @@ -94,24 +107,67 @@ public class SerializationChunkSourceTests public void PushSingleDoesNotDisturbPartialChunk() { var source = new SerializationChunkSource(); + var heavy = new TestPersistence(); - var small1 = new TestEntity(); - var heavy = new TestEntity { SerializedLength = 2 * 1024 * 1024 }; - var small2 = new TestEntity(); + 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); + 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.True(source.TryTake(out var chunk)); + Assert.Same(heavy, chunk.Single); + Assert.Equal(1, chunk.Count); - Assert.False(source.TryTake(out _)); - source.Flush(); + Assert.False(source.TryTake(out _)); + source.Flush(); - var drained = Drain(source); - Assert.Equal([small1, small2], drained); + 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] @@ -138,9 +194,10 @@ public class SerializationChunkSourceTests } [Fact] - public void WorkersDrainAllEntitiesAndStampPositions() + 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++) { @@ -161,22 +218,17 @@ public class SerializationChunkSourceTests worker.Wake(); } - // A large payload published as a dedicated single chunk (like persistence - // self-payloads), interleaved with the bare entity stream. - var heavy = new TestEntity { PayloadSize = 512 * 1024 }; - entities.Insert(5000, heavy); - + // 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++) { - var e = entities[i]; - if (e == heavy) + if (i == 5000) { - source.PushSingle(e); - } - else - { - source.Push(e); + source.PushSingle(owner); } + + source.Push(entities[i]); } // Mirrors World.PauseSerializationThreads @@ -194,26 +246,60 @@ public class SerializationChunkSourceTests totalBytes += worker.BytesSerialized; } - Assert.Equal(entities.Count, totalEntities); + Assert.Equal(entities.Count + 1, totalEntities); - long expectedBytes = 0; + // 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; - - // Every entity serialized exactly once with a consistent span on its worker's heap - Assert.Equal(e.PayloadSize, e.SerializedLength); - Assert.InRange(e.SerializedThread, (byte)0, (byte)(workers.Length - 1)); - - var heap = workers[e.SerializedThread].GetHeap(e.SerializedPosition, e.SerializedLength); - Assert.Equal(0, heap[0]); - Assert.Equal((e.SerializedLength - 1) & 0xFF, heap[^1]); } + 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/ShadowDictionaryEntriesTests.cs b/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs index 7a7609657..6f5661f5a 100644 --- a/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs +++ b/Projects/Server.Tests/Tests/Serialization/ShadowDictionaryEntriesTests.cs @@ -13,9 +13,6 @@ public class ShadowDictionaryEntriesTests public Serial Serial { get; } public DateTime Created { get; set; } public bool Deleted => false; - public byte SerializedThread { get; set; } - public int SerializedPosition { get; set; } - public int SerializedLength { get; set; } public void Delete() { @@ -32,6 +29,9 @@ public class ShadowDictionaryEntriesTests } } + private static TestEntity GetSlotValue(Array entries, int slot) => + System.Runtime.CompilerServices.Unsafe.As[]>(entries)[slot].Value; + [Fact] public void RuntimeLayoutIsSupported() { @@ -74,25 +74,47 @@ public class ShadowDictionaryEntriesTests 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, 3, offset, Math.Min(4096, slotCount - offset)); + serialized += source.SerializeRange(writer, lengths, offset, Math.Min(4096, slotCount - offset)); } Assert.Equal(dict.Count, serialized); + Assert.Equal(dict.Count, lengths.Count); - // Every live entity was stamped exactly once with a coherent span. - foreach (var (serial, entity) in dict) + // 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++) { - Assert.Equal(3, entity.SerializedThread); - Assert.Equal(8, entity.SerializedLength); // serial + int + // Occupancy is exactly the non-null values, same as production. + var entity = GetSlotValue(entries, slot); + if (entity == null) + { + continue; + } - var span = writer.Buffer.AsSpan(entity.SerializedPosition, entity.SerializedLength); - Assert.Equal(serial, (Serial)System.Buffers.Binary.BinaryPrimitives.ReadUInt32LittleEndian(span)); + 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 { 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/GenericEntityPersistence.cs b/Projects/Server/Serialization/GenericEntityPersistence.cs index 5e3e35bdf..aa56e630f 100644 --- a/Projects/Server/Serialization/GenericEntityPersistence.cs +++ b/Projects/Server/Serialization/GenericEntityPersistence.cs @@ -80,8 +80,8 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer var threads = World._threadWorkers; - // 1MB buffer: entity spans are written individually, so the default 4KB buffer - // costs a syscall every few entities on the snapshot thread. + // 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 ); @@ -91,24 +91,24 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer 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 @@ -116,42 +116,40 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer 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; @@ -160,6 +158,99 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer idx.Seek(currentPosition, SeekOrigin.Begin); } + private long WriteSegmentRecords( + SerializationThreadWorker worker, in SerializedSegment segment, MemoryMapFileWriter 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(entity.GetType()); + 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(entity.GetType()); + 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; + } + public override void Serialize() { // Self-payload first so a large one overlaps the entity stream instead of ending it. @@ -196,7 +287,7 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer return false; } - int ISlotRangeSource.SerializeRange(BufferWriter writer, byte threadIndex, int offset, int count) + 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. @@ -212,7 +303,9 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer var entity = entry.Value; if (entity != null) { - SerializationThreadWorker.Serialize(entity, writer, threadIndex); + var start = writer.Position; + entity.Serialize(writer); + lengths.Add((int)(writer.Position - start)); serialized++; } } @@ -393,8 +486,6 @@ public class GenericEntityPersistence : GenericPersistence, IGenericEntityPer if (ctor.Invoke(ctorArgs) is T entity) { entity.Created = created; - // Cost estimate for the first save's scheduling; overwritten by every save. - entity.SerializedLength = length; entities.Add(new EntitySpan(entity, pos, length)); EntitiesBySerial[serial] = entity; } diff --git a/Projects/Server/Serialization/GenericPersistence.cs b/Projects/Server/Serialization/GenericPersistence.cs index 1e2af7f00..f1c939729 100644 --- a/Projects/Server/Serialization/GenericPersistence.cs +++ b/Projects/Server/Serialization/GenericPersistence.cs @@ -25,14 +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 => SerializedLength > 0 ? SerializedLength : _loadedFileLength; + internal long EstimatedSize => _selfLength > 0 ? _selfLength : _loadedFileLength; public GenericPersistence(string name, int priority) : base(priority) { @@ -49,7 +59,7 @@ public abstract class GenericPersistence : Persistence, IGenericSerializable public override void WriteSnapshot(string savePath, HashSet typeSet) { - if (SerializedLength == 0) + if (_selfLength == 0) { return; } @@ -62,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) 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/Persistence.cs b/Projects/Server/Serialization/Persistence.cs index 79ca5506e..01a0043fe 100644 --- a/Projects/Server/Serialization/Persistence.cs +++ b/Projects/Server/Serialization/Persistence.cs @@ -122,6 +122,10 @@ public abstract class Persistence 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(); } } diff --git a/Projects/Server/Serialization/SerializationChunkSource.cs b/Projects/Server/Serialization/SerializationChunkSource.cs index 3b498b77f..8b43304b3 100644 --- a/Projects/Server/Serialization/SerializationChunkSource.cs +++ b/Projects/Server/Serialization/SerializationChunkSource.cs @@ -15,6 +15,7 @@ using System; using System.Collections.Concurrent; +using System.Collections.Generic; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; @@ -29,10 +30,10 @@ namespace Server; public interface ISlotRangeSource { /// - /// Serializes every occupied slot in [offset, offset + count) into the writer, - /// stamping each entity's thread/position/length. Returns the number serialized. + /// 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, byte threadIndex, int offset, int count); + int SerializeRange(BufferWriter writer, List lengths, int offset, int count); } /// @@ -54,22 +55,24 @@ public sealed class SerializationChunkSource internal readonly struct Chunk { - public readonly IGenericSerializable Single; + 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(IGenericSerializable single) + public Chunk(GenericPersistence single) { Single = single; Count = 1; } - public Chunk(IGenericSerializable[] buffer, int count) + public Chunk(IGenericSerializable[] buffer, int count, Persistence owner) { Buffer = buffer; Count = count; + Owner = owner; } public Chunk(ISlotRangeSource source, int offset, int count) @@ -86,6 +89,21 @@ public sealed class SerializationChunkSource // 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) @@ -99,18 +117,18 @@ public sealed class SerializationChunkSource if (++_count == ChunkCapacity) { - _chunks.Enqueue(new Chunk(current, ChunkCapacity)); + _chunks.Enqueue(new Chunk(current, ChunkCapacity, _currentOwner)); _current = null; _count = 0; } } /// - /// Publishes the entity as a dedicated chunk regardless of its estimated size. - /// Used for persistence self-payloads, which can be large on the first save - /// before a estimate exists. + /// 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(IGenericSerializable entity) => _chunks.Enqueue(new Chunk(entity)); + public void PushSingle(GenericPersistence persistence) => _chunks.Enqueue(new Chunk(persistence)); /// /// Publishes slot ranges covering [0, slotCount) of a directly-iterable persistence. @@ -133,7 +151,7 @@ public sealed class SerializationChunkSource { if (_count > 0) { - _chunks.Enqueue(new Chunk(_current, _count)); + _chunks.Enqueue(new Chunk(_current, _count, _currentOwner)); _current = null; _count = 0; } diff --git a/Projects/Server/Serialization/SerializationThreadWorker.cs b/Projects/Server/Serialization/SerializationThreadWorker.cs index c4c6ed339..c4aee0750 100644 --- a/Projects/Server/Serialization/SerializationThreadWorker.cs +++ b/Projects/Server/Serialization/SerializationThreadWorker.cs @@ -14,11 +14,44 @@ *************************************************************************/ using System; +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 @@ -35,6 +68,28 @@ public class SerializationThreadWorker private long _entitiesSerialized; private long _bytesSerialized; + // 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 = []; + + 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) { @@ -98,39 +153,55 @@ public class SerializationThreadWorker [MethodImpl(MethodImplOptions.AggressiveInlining)] public ReadOnlySpan GetHeap(int start, int length) => _heap.AsSpan(start, length); - [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static void Serialize(IGenericSerializable e, BufferWriter writer, byte threadIndex) - { - e.SerializedThread = threadIndex; - var start = e.SerializedPosition = (int)writer.Position; - e.Serialize(writer); - e.SerializedLength = (int)(writer.Position - start); - } - - private static long ProcessChunk( - in SerializationChunkSource.Chunk chunk, SerializationChunkSource chunkSource, - BufferWriter writer, byte threadIndex - ) + private long ProcessChunk(in SerializationChunkSource.Chunk chunk, BufferWriter writer) { if (chunk.Single != null) { - Serialize(chunk.Single, writer, threadIndex); + // 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) { - return chunk.Source.SerializeRange(writer, threadIndex, chunk.Offset, chunk.Count); + 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++) { - Serialize(buffer[i], writer, threadIndex); + var e = buffer[i]; + var start = writer.Position; + e.Serialize(writer); + _lengths.Add((int)(writer.Position - start)); + _bufferEntities.Add(e); } - chunkSource.Return(buffer, count); + _segments.Add( + new SerializedSegment(chunk.Owner, -1, 0, bufferHeapStart, bufferLengthsStart, count, entitiesStart) + ); + + _chunkSource.Return(buffer, count); return count; } @@ -141,13 +212,14 @@ public class SerializationThreadWorker /// public void DrainInline() { + ReleaseWriteLogs(); + var writer = new BufferWriter(_heap, true, World.SerializedTypes); - var threadIndex = (byte)_index; var entities = 0L; while (_chunkSource.TryTake(out var chunk)) { - entities += ProcessChunk(in chunk, _chunkSource, writer, threadIndex); + entities += ProcessChunk(in chunk, writer); } _heap = writer.Buffer; @@ -160,13 +232,14 @@ public class SerializationThreadWorker private static void Execute(object obj) { var worker = (SerializationThreadWorker)obj; - var threadIndex = (byte)worker._index; var chunkSource = worker._chunkSource; var serializedTypes = World.SerializedTypes; while (worker._startEvent.WaitOne()) { + worker.ReleaseWriteLogs(); + var writer = new BufferWriter(worker._heap, true, serializedTypes); var entities = 0L; var spinner = new SpinWait(); @@ -177,7 +250,7 @@ public class SerializationThreadWorker if (chunkSource.TryTake(out var chunk)) { spinner.Reset(); - entities += ProcessChunk(in chunk, chunkSource, writer, threadIndex); + entities += worker.ProcessChunk(in chunk, writer); } else if (pauseRequested) // Break when finished { diff --git a/Projects/Server/World/World.cs b/Projects/Server/World/World.cs index 4ded72fb1..401460de1 100644 --- a/Projects/Server/World/World.cs +++ b/Projects/Server/World/World.cs @@ -421,6 +421,13 @@ 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(); + } } private static void LogWorkerBalance() @@ -473,11 +480,14 @@ public static class World } } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + internal static void SetChunkSourceOwner(Persistence owner) => _chunkSource.SetOwner(owner); + [MethodImpl(MethodImplOptions.AggressiveInlining)] internal static void PushToCache(IGenericSerializable e) => _chunkSource.Push(e); [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static void PushSingleToCache(IGenericSerializable e) => _chunkSource.PushSingle(e); + internal static void PushSingleToCache(GenericPersistence persistence) => _chunkSource.PushSingle(persistence); [MethodImpl(MethodImplOptions.AggressiveInlining)] internal static void PushSlotRangesToCache(ISlotRangeSource source, int slotCount) => 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()