fix: Fixes serialization threading, moves world save to end of loop, and eliminates Parallel.ForEach. (#1530)

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; }
 

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;
 

Projects/Server/Items/Item.cs

@@ -760,9 +760,9 @@ public class Item : IHued, IComparable<Item>, 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; }

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;

Projects/Server/Mobiles/Mobile.cs

@@ -2258,9 +2258,9 @@ public partial class Mobile : IHued, IComparable<Mobile>, 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; }

Projects/Server/Network/PingServer.cs

@@ -75,7 +75,7 @@ public static class PingServer
 
     public static void Slice()
     {
-        if (!Enabled || Core.Closing)
+        if (!Enabled)
         {
             return;
         }

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<T> : 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<T> : 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<T> : Persistence, IGenericEntityPersistenc
                     _pendingDelete[entity.Serial] = entity;
                     break;
                 }
+            case WorldState.PendingSave:
             case WorldState.Running:
                 {
                     EntitiesBySerial.Remove(entity.Serial);
@@ -299,6 +297,7 @@ public class GenericEntityPersistence<T> : Persistence, IGenericEntityPersistenc
 
                     return null;
                 }
+            case WorldState.PendingSave:
             case WorldState.Running:
                 {
                     return EntitiesBySerial.TryGetValue(serial, out var entity) ? entity as R : null;

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<Type> 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);
     }
 

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<Type> types)
+    void IGenericSerializable.Serialize(ConcurrentQueue<Type> types)
     {
         SaveBuffer ??= new BufferWriter(true, types);
 

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()

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 <http://www.gnu.org/licenses/>. *
+ *************************************************************************/
+
+using System;
+using System.Collections.Concurrent;
+
+namespace Server;
+
+public interface IGenericSerializable
+{
+    void Serialize(ConcurrentQueue<Type> types);
+}

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<BaseGuild> _guildPersistence = new("Guilds", 3, 1, 0x7FFFFFFF);
+    private static readonly ItemPersistence _itemPersistence = new();
+    private static readonly MobilePersistence _mobilePersistence = new();
+    private static readonly GenericEntityPersistence<BaseGuild> _guildPersistence = new("Guilds", 3, 1, 0x7FFFFFFF);
 
-    private static ManualResetEvent m_DiskWriteHandle = new(true);
-    private static ConcurrentQueue<Item> _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<Item> _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<Mobile>
@@ -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<IGenericSerializable> _entities;
+
+        public SerializationThreadWorker()
+        {
+            _startEvent = new AutoResetEvent(false);
+            _stopEvent = new AutoResetEvent(false);
+            _entities = new ConcurrentQueue<IGenericSerializable>();
+            _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;
+                }
+            }
+        }
+    }
 }