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.
This commit is contained in:
asayre
parent
5ff9459102
commit
0caf1846d5
5 changed files with 464 additions and 28 deletions
313
Server/Persistence/DynamicSaveStrategy.cs
Normal file
313
Server/Persistence/DynamicSaveStrategy.cs
Normal file
|
|
@ -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<Item> _decayBag;
|
||||
|
||||
private BlockingCollection<QueuedMemoryWriter> _itemThreadWriters;
|
||||
private BlockingCollection<QueuedMemoryWriter> _mobileThreadWriters;
|
||||
private BlockingCollection<QueuedMemoryWriter> _guildThreadWriters;
|
||||
|
||||
public DynamicSaveStrategy()
|
||||
{
|
||||
_decayBag = new ConcurrentBag<Item>();
|
||||
_itemThreadWriters = new BlockingCollection<QueuedMemoryWriter>();
|
||||
_mobileThreadWriters = new BlockingCollection<QueuedMemoryWriter>();
|
||||
_guildThreadWriters = new BlockingCollection<QueuedMemoryWriter>();
|
||||
}
|
||||
|
||||
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<QueuedMemoryWriter> 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<Item> 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<Mobile> 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<BaseGuild> 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<Type> 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
|
||||
124
Server/Persistence/QueuedMemoryWriter.cs
Normal file
124
Server/Persistence/QueuedMemoryWriter.cs
Normal file
|
|
@ -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<IndexInfo> _indexQueue = new Queue<IndexInfo>(); //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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<string, string>() { { "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<string, string>() { { "CompilerVersion", "v3.5" } } ) )
|
||||
#else
|
||||
|
||||
using ( VBCodeProvider provider = new VBCodeProvider() )
|
||||
#endif
|
||||
{
|
||||
string path = GetUnusedPath( "Scripts.VB" );
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue