Stack Overflow question Disruptor.NET exampleからコード例を取得し、時間を「測定」するように変更しました。完全なリストは以下のとおりです。
using System;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Disruptor;
using Disruptor.Dsl;
namespace DisruptorTest
{
public sealed class ValueEntry
{
public long Value { get; set; }
public ValueEntry()
{
Console.WriteLine("New ValueEntry created");
}
}
public class ValueAdditionHandler : IEventHandler<ValueEntry>
{
public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
{
Program.sw.Stop();
long microseconds = Program.sw.ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
Console.WriteLine("elapsed microseconds = " + microseconds);
Console.WriteLine("Event handled: Value = {0} (processed event {1}", data.Value, sequence);
}
}
class Program
{
public static Stopwatch sw = Stopwatch.StartNew();
private static readonly Random _random = new Random();
private static readonly int _ringSize = 16; // Must be multiple of 2
static void Main(string[] args)
{
var disruptor = new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), _ringSize, TaskScheduler.Default);
disruptor.HandleEventsWith(new ValueAdditionHandler());
var ringBuffer = disruptor.Start();
while (true)
{
var valueToSet = _random.Next();
long sequenceNo = ringBuffer.Next();
ValueEntry entry = ringBuffer[sequenceNo];
entry.Value = valueToSet;
sw.Restart();
ringBuffer.Publish(sequenceNo);
Console.WriteLine("Published entry {0}, value {1}", sequenceNo, entry.Value);
Thread.Sleep(1000);
}
}
}
}
出力は次のとおりです。
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
Published entry 0, value 1510145842
elapsed microseconds = 2205
Event handled: Value = 1510145842 (processed event 0
Published entry 1, value 1718075893
elapsed microseconds = 85
Event handled: Value = 1718075893 (processed event 1
Published entry 2, value 1675907645
elapsed microseconds = 32
Event handled: Value = 1675907645 (processed event 2
Published entry 3, value 1563009446
elapsed microseconds = 75
Event handled: Value = 1563009446 (processed event 3
Published entry 4, value 1782914062
elapsed microseconds = 34
Event handled: Value = 1782914062 (processed event 4
Published entry 5, value 1516398244
elapsed microseconds = 50
Event handled: Value = 1516398244 (processed event 5
Published entry 6, value 76829327
elapsed microseconds = 50
Event handled: Value = 76829327 (processed event 6
したがって、あるスレッドから別のスレッドにデータを渡すのに約 50 マイクロ秒かかります。でも全然速くない！「Disruptor の現在のバージョンは、1 秒あたり 100 万メッセージの速度で、スレッド間で最大 50 ns を実行できます。」したがって、私の結果は予想よりも 1000 倍遅くなります。
私の例の何が問題なのですか?どうすれば 50 ns の速度を達成できますか?
上記のプログラムを修正したところ、1 マイクロ秒の遅延が発生し、はるかに改善されました。disruptorただし、パターンの専門家からの回答を待っています。実際に 50 ns でデータを渡すことができることを証明できる例を探しています。
BlockingCollectionまた、平均で 14 マイクロ秒を使用して同じテストを作成し、受信しました。これは、Disruptorより高速であることを証明しています。
BlockingCollection の使用:
average = 14 minimum = 0 0-5 = 890558, 5-10 = 1773781, 10-30 = 6900128, >30 = 435433
ディスラプターの使用:
average = 0 minimum = 0 0-5 = 9908469, 5-10 = 64464, 10-30 = 19902, >30 = 7065
BlockingCollection コード:
using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace DisruptorTest
{
public sealed class ValueEntry
{
public int Value { get; set; }
public ValueEntry()
{
// Console.WriteLine("New ValueEntry created");
}
}
//public class ValueAdditionHandler : IEventHandler<ValueEntry>
//{
// public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
// {
// long microseconds = Program.sw[data.Value].ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
// Program.results[data.Value] = microseconds;
// //Console.WriteLine("elapsed microseconds = " + microseconds);
// //Console.WriteLine("Event handled: Value = {0} (processed event {1}", data.Value, sequence);
// }
//}
class Program
{
public const int length = 10000000;
public static Stopwatch[] sw = new Stopwatch[length];
public static long[] results = new long[length];
static BlockingCollection<ValueEntry> dataItems = new BlockingCollection<ValueEntry>(150);
static void Main(string[] args)
{
for (int i = 0; i < length; i++)
{
sw[i] = Stopwatch.StartNew();
}
// A simple blocking consumer with no cancellation.
Task.Factory.StartNew(() =>
{
while (!dataItems.IsCompleted)
{
ValueEntry ve = null;
try
{
ve = dataItems.Take();
long microseconds = sw[ve.Value].ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
results[ve.Value] = microseconds;
//Console.WriteLine("elapsed microseconds = " + microseconds);
//Console.WriteLine("Event handled: Value = {0} (processed event {1}", ve.Value, ve.Value);
}
catch (InvalidOperationException) { }
}
}, TaskCreationOptions.LongRunning);
for (int i = 0; i < length; i++)
{
var valueToSet = i;
ValueEntry entry = new ValueEntry();
entry.Value = valueToSet;
sw[i].Restart();
dataItems.Add(entry);
//Console.WriteLine("Published entry {0}, value {1}", valueToSet, entry.Value);
//Thread.Sleep(1000);
}
// Wait until all events are delivered
Thread.Sleep(5000);
long average = 0;
long minimum = 10000000000;
int firstFive = 0;
int fiveToTen = 0;
int tenToThirty = 0;
int moreThenThirty = 0;
// Do not count first 100 items because they could be extremely slow
for (int i = 100; i < length; i++)
{
average += results[i];
if (results[i] < minimum)
{
minimum = results[i];
}
if (results[i] < 5)
{
firstFive++;
}
else if (results[i] < 10)
{
fiveToTen++;
}
else if (results[i] < 30)
{
tenToThirty++;
} else
{
moreThenThirty++;
}
}
average /= (length - 100);
Console.WriteLine("average = {0} minimum = {1} 0-5 = {2}, 5-10 = {3}, 10-30 = {4}, >30 = {5}", average, minimum, firstFive, fiveToTen, tenToThirty, moreThenThirty);
}
}
}
ディスラプター コード:
using System;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Disruptor;
using Disruptor.Dsl;
namespace DisruptorTest
{
public sealed class ValueEntry
{
public int Value { get; set; }
public ValueEntry()
{
// Console.WriteLine("New ValueEntry created");
}
}
public class ValueAdditionHandler : IEventHandler<ValueEntry>
{
public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
{
long microseconds = Program.sw[data.Value].ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
Program.results[data.Value] = microseconds;
//Console.WriteLine("elapsed microseconds = " + microseconds);
//Console.WriteLine("Event handled: Value = {0} (processed event {1}", data.Value, sequence);
}
}
class Program
{
public const int length = 10000000;
public static Stopwatch[] sw = new Stopwatch[length];
public static long[] results = new long[length];
private static readonly Random _random = new Random();
private static readonly int _ringSize = 1024; // Must be multiple of 2
static void Main(string[] args)
{
for (int i = 0; i < length; i++)
{
sw[i] = Stopwatch.StartNew();
}
var disruptor = new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), _ringSize, TaskScheduler.Default);
disruptor.HandleEventsWith(new ValueAdditionHandler());
var ringBuffer = disruptor.Start();
for (int i = 0; i < length; i++)
{
var valueToSet = i;
long sequenceNo = ringBuffer.Next();
ValueEntry entry = ringBuffer[sequenceNo];
entry.Value = valueToSet;
sw[i].Restart();
ringBuffer.Publish(sequenceNo);
//Console.WriteLine("Published entry {0}, value {1}", sequenceNo, entry.Value);
//Thread.Sleep(1000);
}
// wait until all events are delivered
Thread.Sleep(5000);
long average = 0;
long minimum = 10000000000;
int firstFive = 0;
int fiveToTen = 0;
int tenToThirty = 0;
int moreThenThirty = 0;
// Do not count first 100 items because they could be extremely slow
for (int i = 100; i < length; i++)
{
average += results[i];
if (results[i] < minimum)
{
minimum = results[i];
}
if (results[i] < 5)
{
firstFive++;
}
else if (results[i] < 10)
{
fiveToTen++;
}
else if (results[i] < 30)
{
tenToThirty++;
}
else
{
moreThenThirty++;
}
}
average /= (length - 100);
Console.WriteLine("average = {0} minimum = {1} 0-5 = {2}, 5-10 = {3}, 10-30 = {4}, >30 = {5}", average, minimum, firstFive, fiveToTen, tenToThirty, moreThenThirty);
}
}
}