/************************************************************************* * 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]; }