From 1f0acddc4661a7b91a9b1b640f0a862d41ac2c79 Mon Sep 17 00:00:00 2001 From: Kamron Batman <3953314+kamronbatman@users.noreply.github.com> Date: Tue, 3 Oct 2023 00:42:49 -0700 Subject: [PATCH] fix: Fixes serialization threading, moves world save to end of loop, and eliminates Parallel.ForEach. (#1530) --- Projects/Server/Guild.cs | 4 +- Projects/Server/IEntity.cs | 6 +- Projects/Server/Items/Item.cs | 4 +- Projects/Server/Main.cs | 13 ++ Projects/Server/Mobiles/Mobile.cs | 4 +- Projects/Server/Network/PingServer.cs | 2 +- .../Serialization/GenericEntityPersistence.cs | 15 +- .../Serialization/GenericPersistence.cs | 22 ++- .../Server/Serialization/ISerializable.cs | 4 +- Projects/Server/Serialization/Persistence.cs | 6 +- Projects/Server/World/IGenericSerializable.cs | 24 +++ Projects/Server/World/World.cs | 182 +++++++++++++++--- 12 files changed, 228 insertions(+), 58 deletions(-) create mode 100644 Projects/Server/World/IGenericSerializable.cs diff --git a/Projects/Server/Guild.cs b/Projects/Server/Guild.cs index dcbca3d9f..46aac0f74 100644 --- a/Projects/Server/Guild.cs +++ b/Projects/Server/Guild.cs @@ -50,9 +50,9 @@ public abstract class BaseGuild : ISerializable [CommandProperty(AccessLevel.GameMaster, readOnly: true)] public DateTime Created { get; set; } = Core.Now; - long ISerializable.SavePosition { get; set; } = -1; + public long SavePosition { get; set; } = -1; - BufferWriter ISerializable.SaveBuffer { get; set; } + public BufferWriter SaveBuffer { get; set; } public int TypeRef { get; private set; } diff --git a/Projects/Server/IEntity.cs b/Projects/Server/IEntity.cs index a35bdaa75..d34fdaa14 100644 --- a/Projects/Server/IEntity.cs +++ b/Projects/Server/IEntity.cs @@ -43,11 +43,11 @@ public class Entity : IEntity public Entity(Serial serial) => Serial = serial; - DateTime ISerializable.Created { get; set; } = Core.Now; + public DateTime Created { get; set; } = Core.Now; - long ISerializable.SavePosition { get; set; } = -1; + public long SavePosition { get; set; } = -1; - BufferWriter ISerializable.SaveBuffer { get; set; } + public BufferWriter SaveBuffer { get; set; } public int TypeRef => -1; diff --git a/Projects/Server/Items/Item.cs b/Projects/Server/Items/Item.cs index 7144fa79f..e4d9cdcf7 100644 --- a/Projects/Server/Items/Item.cs +++ b/Projects/Server/Items/Item.cs @@ -760,9 +760,9 @@ public class Item : IHued, IComparable, ISpawnable, IObjectPropertyListEnt [CommandProperty(AccessLevel.GameMaster, readOnly: true)] public DateTime Created { get; set; } = Core.Now; - long ISerializable.SavePosition { get; set; } = -1; + public long SavePosition { get; set; } = -1; - BufferWriter ISerializable.SaveBuffer { get; set; } + public BufferWriter SaveBuffer { get; set; } [CommandProperty(AccessLevel.Counselor)] public Serial Serial { get; } diff --git a/Projects/Server/Main.cs b/Projects/Server/Main.cs index 216c8ab6c..d167da164 100644 --- a/Projects/Server/Main.cs +++ b/Projects/Server/Main.cs @@ -36,6 +36,7 @@ public static class Core { private static readonly ILogger logger = LogFactory.GetLogger(typeof(Core)); + private static bool _performSnapshot; private static bool _crashed; private static string _baseDirectory; @@ -556,6 +557,12 @@ public static class Core Timer.CheckTimerPool(); // Check for pool depletion so we can async refill it. + if (_performSnapshot) + { + // Return value is the offset that can be used to fix timers that should drift + World.Snapshot(); + } + _tickCount = 0; _now = DateTime.MinValue; @@ -578,8 +585,14 @@ public static class Core { CurrentDomain_UnhandledException(null, new UnhandledExceptionEventArgs(e, true)); } + finally + { + World.SleepSerializationThreads(); + } } + internal static void RequestSnapshot() => _performSnapshot = true; + public static void VerifySerialization() { _itemCount = 0; diff --git a/Projects/Server/Mobiles/Mobile.cs b/Projects/Server/Mobiles/Mobile.cs index abe126576..f8e3489fc 100644 --- a/Projects/Server/Mobiles/Mobile.cs +++ b/Projects/Server/Mobiles/Mobile.cs @@ -2258,9 +2258,9 @@ public partial class Mobile : IHued, IComparable, ISpawnable, IObjectPro [CommandProperty(AccessLevel.GameMaster, readOnly: true)] public DateTime Created { get; set; } = Core.Now; - long ISerializable.SavePosition { get; set; } = -1; + public long SavePosition { get; set; } = -1; - BufferWriter ISerializable.SaveBuffer { get; set; } + public BufferWriter SaveBuffer { get; set; } [CommandProperty(AccessLevel.Counselor)] public Serial Serial { get; } diff --git a/Projects/Server/Network/PingServer.cs b/Projects/Server/Network/PingServer.cs index ba28c1dc7..bfe36bb09 100644 --- a/Projects/Server/Network/PingServer.cs +++ b/Projects/Server/Network/PingServer.cs @@ -75,7 +75,7 @@ public static class PingServer public static void Slice() { - if (!Enabled || Core.Closing) + if (!Enabled) { return; } diff --git a/Projects/Server/Serialization/GenericEntityPersistence.cs b/Projects/Server/Serialization/GenericEntityPersistence.cs index ce8a7add1..fa8fa9857 100644 --- a/Projects/Server/Serialization/GenericEntityPersistence.cs +++ b/Projects/Server/Serialization/GenericEntityPersistence.cs @@ -20,7 +20,6 @@ using System.IO; using System.Linq; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; -using System.Threading.Tasks; using Server.Logging; namespace Server; @@ -55,13 +54,10 @@ public class GenericEntityPersistence : Persistence, IGenericEntityPersistenc public override void Serialize() { - // TODO: Hand off to a scheduler instead - Parallel.ForEach(EntitiesBySerial.Values, SerializeEntity); - } - - protected virtual void SerializeEntity(T entity) - { - entity.Serialize(World.SerializedTypes); + foreach (var entity in EntitiesBySerial.Values) + { + World.PushToCache(entity); + } } public override void WriteSnapshot(string basePath) @@ -160,6 +156,7 @@ public class GenericEntityPersistence : Persistence, IGenericEntityPersistenc _pendingAdd[entity.Serial] = entity; break; } + case WorldState.PendingSave: case WorldState.Running: { ref var entityEntry = ref CollectionsMarshal.GetValueRefOrAddDefault(EntitiesBySerial, entity.Serial, out bool exists); @@ -216,6 +213,7 @@ public class GenericEntityPersistence : Persistence, IGenericEntityPersistenc _pendingDelete[entity.Serial] = entity; break; } + case WorldState.PendingSave: case WorldState.Running: { EntitiesBySerial.Remove(entity.Serial); @@ -299,6 +297,7 @@ public class GenericEntityPersistence : Persistence, IGenericEntityPersistenc return null; } + case WorldState.PendingSave: case WorldState.Running: { return EntitiesBySerial.TryGetValue(serial, out var entity) ? entity as R : null; diff --git a/Projects/Server/Serialization/GenericPersistence.cs b/Projects/Server/Serialization/GenericPersistence.cs index 2b9ab5941..4a193ed2d 100644 --- a/Projects/Server/Serialization/GenericPersistence.cs +++ b/Projects/Server/Serialization/GenericPersistence.cs @@ -14,25 +14,33 @@ *************************************************************************/ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.IO; namespace Server; -public abstract class GenericPersistence : Persistence +public abstract class GenericPersistence : Persistence, IGenericSerializable { - private BufferWriter _saveBuffer; - public string Name { get; } public GenericPersistence(string name, int priority) : base(priority) => Name = name; public override void Serialize() { - _saveBuffer ??= new BufferWriter(true, World.SerializedTypes); - _saveBuffer.Seek(0, SeekOrigin.Begin); + World.PushToCache(this); + } - Serialize(_saveBuffer); + public long SavePosition { get; set; } + + public BufferWriter SaveBuffer { get; set; } + + public void Serialize(ConcurrentQueue types) + { + SaveBuffer ??= new BufferWriter(true, types); + + SaveBuffer.Seek(0, SeekOrigin.Begin); + Serialize(SaveBuffer); } public abstract void Serialize(IGenericWriter writer); @@ -40,7 +48,7 @@ public abstract class GenericPersistence : Persistence public override void WriteSnapshot(string basePath) { string binPath = Path.Combine(basePath, Name, $"{Name}.bin"); - var buffer = _saveBuffer!.Buffer.AsSpan(0, (int)_saveBuffer.Position); + var buffer = SaveBuffer!.Buffer.AsSpan(0, (int)SaveBuffer.Position); AdhocPersistence.WriteSnapshot(new FileInfo(binPath), buffer); } diff --git a/Projects/Server/Serialization/ISerializable.cs b/Projects/Server/Serialization/ISerializable.cs index ca7b01e3a..119111a28 100644 --- a/Projects/Server/Serialization/ISerializable.cs +++ b/Projects/Server/Serialization/ISerializable.cs @@ -19,7 +19,7 @@ using System.IO; namespace Server; -public interface ISerializable +public interface ISerializable : IGenericSerializable { // Should be serialized/deserialized with the index so it can be referenced by IGenericReader DateTime Created { get; set; } @@ -48,7 +48,7 @@ public interface ISerializable } } - public void Serialize(ConcurrentQueue types) + void IGenericSerializable.Serialize(ConcurrentQueue types) { SaveBuffer ??= new BufferWriter(true, types); diff --git a/Projects/Server/Serialization/Persistence.cs b/Projects/Server/Serialization/Persistence.cs index 37c240b9d..57223d86d 100644 --- a/Projects/Server/Serialization/Persistence.cs +++ b/Projects/Server/Serialization/Persistence.cs @@ -81,8 +81,10 @@ public abstract class Persistence internal static void SerializeAll() { - // TODO: Hand off to a scheduler - Parallel.ForEach(_registry, p => p.Serialize()); + foreach (var p in _registry) + { + p.Serialize(); + } } internal static void PostSerializeAll() diff --git a/Projects/Server/World/IGenericSerializable.cs b/Projects/Server/World/IGenericSerializable.cs new file mode 100644 index 000000000..40b5c24bc --- /dev/null +++ b/Projects/Server/World/IGenericSerializable.cs @@ -0,0 +1,24 @@ +/************************************************************************* + * ModernUO * + * Copyright 2019-2023 - ModernUO Development Team * + * Email: hi@modernuo.com * + * File: IWorldSerializable.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; + +namespace Server; + +public interface IGenericSerializable +{ + void Serialize(ConcurrentQueue types); +} diff --git a/Projects/Server/World/World.cs b/Projects/Server/World/World.cs index 7d3be696a..dc3ab9988 100644 --- a/Projects/Server/World/World.cs +++ b/Projects/Server/World/World.cs @@ -17,6 +17,7 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; +using System.IO; using System.Runtime.CompilerServices; using System.Threading; using Server.Guilds; @@ -30,6 +31,7 @@ public enum WorldState Initial, Loading, Running, + PendingSave, Saving, WritingSave } @@ -38,12 +40,14 @@ public static class World { private static ILogger logger = LogFactory.GetLogger(typeof(World)); - private static ItemPersistence _itemPersistence = new(); - private static MobilePersistence _mobilePersistence = new(); - private static GenericEntityPersistence _guildPersistence = new("Guilds", 3, 1, 0x7FFFFFFF); + private static readonly ItemPersistence _itemPersistence = new(); + private static readonly MobilePersistence _mobilePersistence = new(); + private static readonly GenericEntityPersistence _guildPersistence = new("Guilds", 3, 1, 0x7FFFFFFF); - private static ManualResetEvent m_DiskWriteHandle = new(true); - private static ConcurrentQueue _decayQueue = new(); + private static int _threadId; + private static readonly SerializationThreadWorker[] _threadWorkers = new SerializationThreadWorker[Environment.ProcessorCount - 1]; + private static readonly ManualResetEvent _diskWriteHandle = new(true); + private static readonly ConcurrentQueue _decayQueue = new(); private static string _tempSavePath; // Path to the temporary folder for the save @@ -78,7 +82,7 @@ public static class World [MethodImpl(MethodImplOptions.AggressiveInlining)] public static void WaitForWriteCompletion() { - m_DiskWriteHandle.WaitOne(); + _diskWriteHandle.WaitOne(); } [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -177,14 +181,18 @@ public static class World Mobiles.Count, watch.Elapsed.TotalSeconds ); + + // Create the serialization threads. + for (var i = 0; i < _threadWorkers.Length; i++) + { + _threadWorkers[i] = new SerializationThreadWorker(); + } } private static void FinishWorldSave() { WorldState = WorldState.Running; - ProcessDecay(); - Persistence.PostSerializeAll(); // Process safety queues } @@ -223,6 +231,7 @@ public static class World { EventSink.InvokeWorldSavePostSnapshot(SavePath, tempPath); PathUtility.MoveDirectory(tempPath, SavePath); + Directory.SetLastWriteTimeUtc(SavePath, Core.Now); } catch (Exception ex) { @@ -233,7 +242,7 @@ public static class World // Clear types SerializedTypes.Clear(); - m_DiskWriteHandle.Set(); + _diskWriteHandle.Set(); Core.LoopContext.Post(FinishWorldSave); } @@ -294,9 +303,25 @@ public static class World WaitForWriteCompletion(); // Blocks Save until current disk flush is done. - WorldState = WorldState.Saving; + _diskWriteHandle.Reset(); - m_DiskWriteHandle.Reset(); + // Start our serialization threads + for (var i = 0; i < _threadWorkers.Length; i++) + { + _threadWorkers[i].Wake(); + } + + WorldState = WorldState.PendingSave; + + Core.RequestSnapshot(); + } + + internal static TimeSpan Snapshot() + { + if (WorldState != WorldState.PendingSave) + { + return TimeSpan.Zero; + } Broadcast(0x35, true, "The world is saving, please wait."); @@ -310,6 +335,13 @@ public static class World { _serializationStart = Core.Now; Persistence.SerializeAll(); + + // Pause the workers + foreach (var worker in _threadWorkers) + { + worker.Sleep(); + } + EventSink.InvokeWorldSave(); } catch (Exception ex) @@ -319,28 +351,44 @@ public static class World WorldState = WorldState.WritingSave; - watch.Stop(); - if (exception == null) { var duration = watch.Elapsed.TotalSeconds; logger.Information("Saving world {Status} ({Duration:F2} seconds)", "done", duration); - // Only broadcast if it took at least 150ms - if (duration >= 0.15) - { - Broadcast(0x35, true, $"World Save completed in {duration:F2} seconds."); - } + Broadcast(0x35, true, $"World save completed in {duration:F2} seconds."); } else { logger.Error(exception, "Saving world {Status}", "failed"); Persistence.TraceException(exception); - BroadcastStaff(0x35, true, "World save failed."); + BroadcastStaff(0x35, true, "World save failed! Check the logs!"); } ThreadPool.QueueUserWorkItem(WriteFiles); + + watch.Stop(); + + return watch.Elapsed; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + internal static void PushToCache(IGenericSerializable e) + { + _threadWorkers[_threadId++].Push(e); + if (_threadId == _threadWorkers.Length) + { + _threadId = 0; + } + } + + internal static void SleepSerializationThreads() + { + for (var i = 0; i < _threadWorkers.Length; i++) + { + _threadWorkers[i].Sleep(); + } } [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -414,16 +462,6 @@ public static class World { } - protected override void SerializeEntity(Item item) - { - if (item.CanDecay() && item.LastMoved + item.DecayTime <= _serializationStart) - { - EnqueueForDecay(item); - } - - ((ISerializable)item).Serialize(SerializedTypes); - } - public override void PostDeserialize() { base.PostDeserialize(); @@ -438,6 +476,26 @@ public static class World item.ClearProperties(); } } + + public override void WriteSnapshot(string basePath) + { + base.WriteSnapshot(basePath); + + foreach (var item in EntitiesBySerial.Values) + { + if (item.CanDecay() && item.LastMoved + item.DecayTime <= _serializationStart) + { + EnqueueForDecay(item); + } + } + } + + public override void PostSerialize() + { + ProcessDecay(); // Run this before the safety queue + + base.PostSerialize(); + } } private class MobilePersistence : GenericEntityPersistence @@ -459,4 +517,70 @@ public static class World } } } + + private 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 bool _pause; + private readonly ConcurrentQueue _entities; + + public SerializationThreadWorker() + { + _startEvent = new AutoResetEvent(false); + _stopEvent = new AutoResetEvent(false); + _entities = new ConcurrentQueue(); + _thread = new Thread(Execute); + _thread.Start(this); + } + + public void Wake() + { + _startEvent.Set(); + } + + public void Sleep() + { + Volatile.Write(ref _pause, true); + _stopEvent.WaitOne(); + } + + public void Push(IGenericSerializable entity) + { + _entities.Enqueue(entity); + } + + private static void Execute(object obj) + { + var serializedTypes = SerializedTypes; + SerializationThreadWorker worker = (SerializationThreadWorker)obj; + + var reader = worker._entities; + + while (worker._startEvent.WaitOne()) + { + while (true) + { + bool pauseRequested = Volatile.Read(ref worker._pause); + if (reader.TryDequeue(out var entity)) + { + entity.Serialize(serializedTypes); + } + else if (pauseRequested) // Break when finished + { + break; + } + } + + worker._stopEvent.Set(); // Allow the main thread to continue now that we are finished + worker._pause = false; + + if (Core.Closing) + { + return; + } + } + } + } }