From 0caf1846d55a149f99f8a95195de79263164fa22 Mon Sep 17 00:00:00 2001 From: asayre Date: Sat, 18 Dec 2010 11:00:57 +0000 Subject: [PATCH] Preprocessor directive for .NET 4.0: Framework_4_0. (Removed the 3.5 directive as it wasn't a new framework runtime anyways) Implemented working Parallel Save strategy based on the TPL: DynamicSaveStrategy. --- Server/Persistence/DynamicSaveStrategy.cs | 313 ++++++++++++++++++++++ Server/Persistence/QueuedMemoryWriter.cs | 124 +++++++++ Server/Persistence/SaveMetrics.cs | 4 +- Server/Persistence/SaveStrategy.cs | 38 +-- Server/ScriptCompiler.cs | 13 +- 5 files changed, 464 insertions(+), 28 deletions(-) create mode 100644 Server/Persistence/DynamicSaveStrategy.cs create mode 100644 Server/Persistence/QueuedMemoryWriter.cs diff --git a/Server/Persistence/DynamicSaveStrategy.cs b/Server/Persistence/DynamicSaveStrategy.cs new file mode 100644 index 000000000..52dfbe4a2 --- /dev/null +++ b/Server/Persistence/DynamicSaveStrategy.cs @@ -0,0 +1,313 @@ +/*************************************************************************** + * DynamicSaveStrategy.cs + * ------------------- + * begin : December 16, 2010 + * copyright : (C) The RunUO Software Team + * email : info@runuo.com + * + * $Id: DynamicSaveStrategy.cs 164 2007-04-20 22:35:41Z asayre $ + * + ***************************************************************************/ + +/*************************************************************************** + * + * 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 2 of the License, or + * (at your option) any later version. + * + ***************************************************************************/ + +#if Framework_4_0 + +using System; +using System.Collections.Generic; +using System.Text; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using System.Diagnostics; +using System.Collections.Concurrent; +using System.Linq; + +using Server; +using Server.Guilds; + +namespace Server +{ + public sealed class DynamicSaveStrategy : SaveStrategy + { + public override string Name { get { return "Dynamic"; } } + + private SaveMetrics metrics; + + private SequentialFileWriter _itemData, _itemIndex; + private SequentialFileWriter _mobileData, _mobileIndex; + private SequentialFileWriter _guildData, _guildIndex; + + private ConcurrentBag _decayBag; + + private BlockingCollection _itemThreadWriters; + private BlockingCollection _mobileThreadWriters; + private BlockingCollection _guildThreadWriters; + + public DynamicSaveStrategy() + { + _decayBag = new ConcurrentBag(); + _itemThreadWriters = new BlockingCollection(); + _mobileThreadWriters = new BlockingCollection(); + _guildThreadWriters = new BlockingCollection(); + } + + public override void Save(SaveMetrics metrics) + { + this.metrics = metrics; + + OpenFiles(); + + Task[] saveTasks = new Task[3]; + + saveTasks[0] = SaveItems( metrics ); + saveTasks[1] = SaveMobiles( metrics ); + saveTasks[2] = SaveGuilds( metrics ); + + SaveTypeDatabases(); + + + Task.WaitAll(saveTasks); //Waits for the completion of all of the tasks(committing to disk) + CloseFiles(); + + //This option makes it finish the writing to disk in the background, continuing even after Save() returns. VERY DANGEROUS. + //TODO: Make this less dangerous, by notifying rest of the server that this can and does happen + //Task.Factory.ContinueWhenAll(saveTasks, _ => CloseFiles()); + } + + private Task StartCommitTask(BlockingCollection threadWriter, SequentialFileWriter data, SequentialFileWriter index) + { + Task commitTask = Task.Factory.StartNew(() => + { + while (!(threadWriter.IsCompleted)) + { + QueuedMemoryWriter writer; + + try + { + writer = threadWriter.Take(); + } + catch (InvalidOperationException) + { + //Per MSDN, it's fine if we're here, successful completion of adding can rarely put us into this state. + break; + } + + int bytesWritten = writer.CommitTo(data, index); + + if (metrics != null) + { + metrics.OnFileWritten(bytesWritten); + } + } + }); + + return commitTask; + } + + private Task SaveItems(SaveMetrics metrics) + { + //Start the blocking consumer; this runs in background. + Task commitTask = StartCommitTask(_itemThreadWriters, _itemData, _itemIndex); + + IEnumerable items = World.Items.Values; + + //Start the producer. + Parallel.ForEach(items, () => new QueuedMemoryWriter(), + (Item item, ParallelLoopState state, QueuedMemoryWriter writer) => + { + long startPosition = writer.Position; + + item.Serialize(writer); + + int size = (int)(writer.Position - startPosition); + + writer.QueueForIndex(item, size); + + if (item.Decays && item.Parent == null && item.Map != Map.Internal && DateTime.Now > (item.LastMoved + item.DecayTime)) + { + _decayBag.Add(item); + } + + if (metrics != null) + { + metrics.OnItemSaved(size); + } + + return writer; + }, + (writer) => + { + writer.Flush(); + + _itemThreadWriters.Add(writer); + }); + + _itemThreadWriters.CompleteAdding(); //We only get here after the Parallel.ForEach completes. Lets our task + + return commitTask; + } + + + + private Task SaveMobiles(SaveMetrics metrics) + { + //Start the blocking consumer; this runs in background. + Task commitTask = StartCommitTask( _mobileThreadWriters, _mobileData, _mobileIndex ); + + IEnumerable mobiles = World.Mobiles.Values; + + //Start the producer. + Parallel.ForEach(mobiles, () => new QueuedMemoryWriter(), + (Mobile mobile, ParallelLoopState state, QueuedMemoryWriter writer) => + { + long startPosition = writer.Position; + + mobile.Serialize(writer); + + int size = (int)(writer.Position - startPosition); + + writer.QueueForIndex(mobile, size); + + if (metrics != null) + { + metrics.OnMobileSaved(size); + } + + return writer; + }, + (writer) => + { + writer.Flush(); + + _mobileThreadWriters.Add(writer); + }); + + _mobileThreadWriters.CompleteAdding(); //We only get here after the Parallel.ForEach completes. Lets our task tell the consumer that we're done + + return commitTask; + } + + private Task SaveGuilds(SaveMetrics metrics) + { + //Start the blocking consumer; this runs in background. + Task commitTask = StartCommitTask(_guildThreadWriters, _guildData, _guildIndex); + + IEnumerable guilds = BaseGuild.List.Values; + + //Start the producer. + Parallel.ForEach(guilds, () => new QueuedMemoryWriter(), + (BaseGuild guild, ParallelLoopState state, QueuedMemoryWriter writer) => + { + long startPosition = writer.Position; + + guild.Serialize(writer); + + int size = (int)(writer.Position - startPosition ); + + writer.QueueForIndex(guild, size); + + if (metrics != null) + { + metrics.OnGuildSaved(size); + } + + return writer; + }, + (writer) => + { + writer.Flush(); + + _guildThreadWriters.Add(writer); + }); + + _guildThreadWriters.CompleteAdding(); //We only get here after the Parallel.ForEach completes. Lets our task + + return commitTask; + } + + public override void ProcessDecay() + { + Item item; + + while( _decayBag.TryTake( out item ) ) + { + item.Delete(); + } + } + + private void OpenFiles() + { + _itemData = new SequentialFileWriter(World.ItemDataPath, metrics); + _itemIndex = new SequentialFileWriter(World.ItemIndexPath, metrics); + + _mobileData = new SequentialFileWriter(World.MobileDataPath, metrics); + _mobileIndex = new SequentialFileWriter(World.MobileIndexPath, metrics); + + _guildData = new SequentialFileWriter(World.GuildDataPath, metrics); + _guildIndex = new SequentialFileWriter(World.GuildIndexPath, metrics); + + WriteCount(_itemIndex, World.Items.Count); + WriteCount(_mobileIndex, World.Mobiles.Count); + WriteCount(_guildIndex, BaseGuild.List.Count); + } + + private void CloseFiles() + { + _itemData.Close(); + _itemIndex.Close(); + + _mobileData.Close(); + _mobileIndex.Close(); + + _guildData.Close(); + _guildIndex.Close(); + + Console.WriteLine("Closing files"); + } + + private void WriteCount(SequentialFileWriter indexFile, int count) + { + //Equiv to GenericWriter.Write( (int)count ); + byte[] buffer = new byte[4]; + + buffer[0] = (byte)(count); + buffer[1] = (byte)(count >> 8); + buffer[2] = (byte)(count >> 16); + buffer[3] = (byte)(count >> 24); + + indexFile.Write(buffer, 0, buffer.Length); + } + + private void SaveTypeDatabases() + { + SaveTypeDatabase(World.ItemTypesPath, World.m_ItemTypes); + SaveTypeDatabase(World.MobileTypesPath, World.m_MobileTypes); + } + + private void SaveTypeDatabase(string path, List types) + { + BinaryFileWriter bfw = new BinaryFileWriter(path, false); + + bfw.Write(types.Count); + + foreach (Type type in types) + { + bfw.Write(type.FullName); + } + + bfw.Flush(); + + bfw.Close(); + } + } + +} +#endif \ No newline at end of file diff --git a/Server/Persistence/QueuedMemoryWriter.cs b/Server/Persistence/QueuedMemoryWriter.cs new file mode 100644 index 000000000..09e79f06d --- /dev/null +++ b/Server/Persistence/QueuedMemoryWriter.cs @@ -0,0 +1,124 @@ +/*************************************************************************** + * QueuedMemoryWriter.cs + * ------------------- + * begin : December 16, 2010 + * copyright : (C) The RunUO Software Team + * email : info@runuo.com + * + * $Id: QueuedMemoryWriter.cs 37 2006-06-19 17:28:24Z asayre $ + * + ***************************************************************************/ + +/*************************************************************************** + * + * 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 2 of the License, or + * (at your option) any later version. + * + ***************************************************************************/ + +using System; +using System.IO; +using System.Collections.Generic; +using System.Text; + +namespace Server +{ + + public sealed class QueuedMemoryWriter : BinaryFileWriter + { + private struct IndexInfo + { + public int size; + public int typeCode; + public int serial; + } + + private MemoryStream _memStream; + private Queue _indexQueue = new Queue(); //TODO: Pick a more optimal starting size + + protected override int BufferSize + { + get { return 4096; } + } + + public QueuedMemoryWriter() + : base(new MemoryStream(1024 * 1024), true) + { + this._memStream = this.UnderlyingStream as MemoryStream; + } + + public void QueueForIndex(ISerializable serializable, int size) + { + IndexInfo info; + + info.size = size; + + info.typeCode = serializable.TypeReference; //For guilds, this will automagically be zero. + info.serial = serializable.SerialIdentity; + + _indexQueue.Enqueue(info); + } + + public int CommitTo(SequentialFileWriter dataFile, SequentialFileWriter indexFile) + { + this.Flush(); + + byte[] memBuffer = _memStream.GetBuffer(); + int memLength = (int)_memStream.Position; + + long actualPosition = dataFile.Position; + + dataFile.Write(memBuffer, 0, memLength); //The buffer contains the data from many items. + + //Console.WriteLine("Writing {0} bytes starting at {1}", memLength, actualPosition); + + byte[] indexBuffer = new byte[20]; + + int indexWritten = _indexQueue.Count * indexBuffer.Length; + + while (_indexQueue.Count > 0) + { + IndexInfo info = _indexQueue.Dequeue(); + + int typeCode = info.typeCode; + int serial = info.serial; + int length = info.size; + + + indexBuffer[0] = (byte)(info.typeCode); + indexBuffer[1] = (byte)(info.typeCode >> 8); + indexBuffer[2] = (byte)(info.typeCode >> 16); + indexBuffer[3] = (byte)(info.typeCode >> 24); + + indexBuffer[4] = (byte)(info.serial); + indexBuffer[5] = (byte)(info.serial >> 8); + indexBuffer[6] = (byte)(info.serial >> 16); + indexBuffer[7] = (byte)(info.serial >> 24); + + indexBuffer[8] = (byte)(actualPosition); + indexBuffer[9] = (byte)(actualPosition >> 8); + indexBuffer[10] = (byte)(actualPosition >> 16); + indexBuffer[11] = (byte)(actualPosition >> 24); + indexBuffer[12] = (byte)(actualPosition >> 32); + indexBuffer[13] = (byte)(actualPosition >> 40); + indexBuffer[14] = (byte)(actualPosition >> 48); + indexBuffer[15] = (byte)(actualPosition >> 56); + + indexBuffer[16] = (byte)(info.size); + indexBuffer[17] = (byte)(info.size >> 8); + indexBuffer[18] = (byte)(info.size >> 16); + indexBuffer[19] = (byte)(info.size >> 24); + + indexFile.Write(indexBuffer, 0, indexBuffer.Length); + + actualPosition += info.size; + } + + this.Close(); //We're donezo with this writer. + + return memLength + indexWritten; + } + } +} \ No newline at end of file diff --git a/Server/Persistence/SaveMetrics.cs b/Server/Persistence/SaveMetrics.cs index 167d270c1..4f49eaaa6 100644 --- a/Server/Persistence/SaveMetrics.cs +++ b/Server/Persistence/SaveMetrics.cs @@ -25,8 +25,8 @@ using System.Text; namespace Server { public sealed class SaveMetrics : IDisposable { - private const string PerformanceCategoryName = "RunUO 2.0"; - private const string PerformanceCategoryDesc = "Performance counters for RunUO 2.0."; + private const string PerformanceCategoryName = "RunUO 2.1"; + private const string PerformanceCategoryDesc = "Performance counters for RunUO 2.1."; private PerformanceCounter numberOfWorldSaves; diff --git a/Server/Persistence/SaveStrategy.cs b/Server/Persistence/SaveStrategy.cs index 311e621db..3feb803ab 100644 --- a/Server/Persistence/SaveStrategy.cs +++ b/Server/Persistence/SaveStrategy.cs @@ -19,33 +19,39 @@ ***************************************************************************/ using System; -using System.Collections.Generic; -using System.Text; -using System.IO; -using System.Threading; -using System.Diagnostics; - using Server; -using Server.Guilds; -namespace Server { - public abstract class SaveStrategy { - public static SaveStrategy Acquire() { - if ( Core.MultiProcessor ) { +namespace Server +{ + public abstract class SaveStrategy + { + public static SaveStrategy Acquire() + { + if (Core.MultiProcessor) + { int processorCount = Core.ProcessorCount; - if ( processorCount > 16 ) { - return new ParallelSaveStrategy( processorCount ); - } else { + if (processorCount > 16) + { +#if Framework_4_0 + return new DynamicSaveStrategy(); +#else + return new ParallelSaveStrategy(processorCount); +#endif + } + else + { return new DualSaveStrategy(); } - } else { + } + else + { return new StandardSaveStrategy(); } } public abstract string Name { get; } - public abstract void Save( SaveMetrics metrics ); + public abstract void Save(SaveMetrics metrics); public abstract void ProcessDecay(); } diff --git a/Server/ScriptCompiler.cs b/Server/ScriptCompiler.cs index fb5663cdd..ff4938a8b 100644 --- a/Server/ScriptCompiler.cs +++ b/Server/ScriptCompiler.cs @@ -91,8 +91,8 @@ namespace Server AppendDefine( ref sb, "/d:Framework_2_0" ); -#if Framework_3_5 - AppendDefine( ref sb, "/d:Framework_3_5" ); +#if Framework_4_0 + AppendDefine( ref sb, "/d:Framework_4_0" ); #endif return (sb == null ? null : sb.ToString()); @@ -212,11 +212,7 @@ namespace Server DeleteFiles( "Scripts.CS*.dll" ); -#if Framework_3_5 - using ( CSharpCodeProvider provider = new CSharpCodeProvider( new Dictionary() { { "CompilerVersion", "v3.5" } } ) ) -#else using ( CSharpCodeProvider provider = new CSharpCodeProvider() ) -#endif { string path = GetUnusedPath( "Scripts.CS" ); @@ -355,11 +351,8 @@ namespace Server } DeleteFiles( "Scripts.VB*.dll" ); -#if Framework_3_5 - using ( VBCodeProvider provider = new VBCodeProvider( new Dictionary() { { "CompilerVersion", "v3.5" } } ) ) -#else + using ( VBCodeProvider provider = new VBCodeProvider() ) -#endif { string path = GetUnusedPath( "Scripts.VB" );