Skip to content

S
StackExchange.Redis
Commit 16363206 authored by Marc Gravell's avatar Marc Gravell
Browse files
Options

make RunCompetingBatchesOnSameMuxer the console runner

parent dad32414
Hide whitespace changes
Inline Side-by-side
Showing with 80 additions and 86 deletions
StackExchange.Redis/SocketManager.cs
View file @ 16363206
......@@ -28,7 +28,10 @@ public sealed partial class SocketManager : IDisposable
public SocketManager(string name = null)
: this(name, false, DEFAULT_MIN_THREADS, DEFAULT_MAX_THREADS) { }
internal static SocketManager Shared
/// <summary>
/// Default / shared socket manager
/// </summary>
public static SocketManager Shared
{
get
{
......@@ -46,6 +49,16 @@ internal static SocketManager Shared
}
}
/// <summary>Returns a string that represents the current object.</summary>
/// <returns>A string that represents the current object.</returns>
public override string ToString()
{
var scheduler = SchedulerPool;
var comp = CompletionPool;
return $"scheduler - queue: {scheduler?.TotalServicedByQueue}, pool: {scheduler?.TotalServicedByPool}; completion - queue: {comp ?.TotalServicedByQueue}, pool: {comp?.TotalServicedByPool}";
}
private static SocketManager _shared;
/// <summary>
......
TestConsole/Program.cs
View file @ 16363206
using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using StackExchange.Redis;
namespace TestConsole
......@@ -7,97 +11,74 @@ internal static class Program
{
private static void Main()
{
using (var muxer = ConnectionMultiplexer.Connect("localhost:6379", Console.Out))
{
muxer.GetDatabase().Ping();
}
RunCompetingBatchesOnSameMuxer();
}
//private static async Task Main2()
//{
// const int ClientCount = 150, ConnectionCount = 10;
// CancellationTokenSource cancel = new CancellationTokenSource();
// var config = new ConfigurationOptions
// {
// EndPoints = { new IPEndPoint(IPAddress.Loopback, 6379) }
// };
// var muxers = new ConnectionMultiplexer[ConnectionCount];
// try
// {
// for (int i = 0; i < muxers.Length; i++)
// {
// muxers[i] = await ConnectionMultiplexer.ConnectAsync(config);
// }
// var tasks = new Task[ClientCount + 1];
// tasks[0] = Task.Run(() => ShowState(cancel.Token));
// for (int i = 1; i < tasks.Length; i++)
// {
// var db = muxers[i % muxers.Length].GetDatabase();
// int seed = i;
// var key = "test_client_" + i;
// tasks[i] = Task.Run(() => RunClient(key, seed, db, cancel.Token));
// }
// Console.ReadLine();
// cancel.Cancel();
// await Task.WhenAll(tasks);
// }
// finally
// {
// for (int i = 0; i < muxers.Length; i++)
// {
// try { muxers[i]?.Dispose(); } catch { }
// }
// }
//}
static ConnectionMultiplexer Create()
{
var muxer = ConnectionMultiplexer.Connect("localhost:6379");
muxer.GetDatabase().Ping();
return muxer;
}
private const int IterationCount = 5000, InnerCount = 20;
public static void RunCompetingBatchesOnSameMuxer()
{
using (var muxer = Create())
{
var db = muxer.GetDatabase();
//private static int clients;
//private static long totalPings, pings, lastTicks;
//private static async Task ShowState(CancellationToken cancellation)
//{
// while (!cancellation.IsCancellationRequested)
// {
// await Task.Delay(2000);
// var nowTicks = DateTime.UtcNow.Ticks;
// var thenTicks = Interlocked.Exchange(ref lastTicks, nowTicks);
// long pingsInInterval = Interlocked.Exchange(ref pings, 0);
// var newTotalPings = Interlocked.Add(ref totalPings, pingsInInterval);
Thread x = new Thread(state => BatchRunPings((IDatabase)state));
x.Name = nameof(BatchRunPings);
Thread y = new Thread(state => BatchRunIntegers((IDatabase)state));
y.Name = nameof(BatchRunIntegers);
// var deltaTicks = nowTicks - thenTicks;
var watch = Stopwatch.StartNew();
x.Start(db);
y.Start(db);
x.Join();
y.Join();
watch.Stop();
Console.WriteLine($"{watch.ElapsedMilliseconds}ms");
Console.WriteLine(muxer.GetCounters().Interactive);
Console.WriteLine($"Service Counts: {SocketManager.Shared}");
}
}
// Console.WriteLine($"[{Thread.VolatileRead(ref clients)}], Pings: {newTotalPings} ({pingsInInterval}, {Rate(pingsInInterval, deltaTicks)}/s)");
// }
//}
static RedisKey Me([CallerMemberName]string caller = null) => caller;
//private static string Rate(long pingsInInterval, long deltaTicks)
//{
// if (deltaTicks == 0) return "n/a";
// if (pingsInInterval == 0) return "0";
private static void BatchRunIntegers(IDatabase db)
{
var key = Me();
db.KeyDelete(key);
db.StringSet(key, 1);
Task[] tasks = new Task[InnerCount];
for (int i = 0; i < IterationCount; i++)
{
var batch = db.CreateBatch();
for (int j = 0; j < tasks.Length; j++)
{
tasks[j] = batch.StringIncrementAsync(key);
}
batch.Execute();
db.Multiplexer.WaitAll(tasks);
}
// var seconds = ((decimal)deltaTicks) / TimeSpan.TicksPerSecond;
// return (pingsInInterval / seconds).ToString("0.0");
//}
var count = (long)db.StringGet(key);
Console.WriteLine($"tally: {count}");
}
//private static async Task RunClient(RedisKey key, int seed, IDatabase db, CancellationToken cancellation)
//{
// Interlocked.Increment(ref clients);
// try
// {
// while (!cancellation.IsCancellationRequested)
// {
// await db.PingAsync();
// Interlocked.Increment(ref pings);
// }
// }
// catch (Exception ex)
// {
// Console.Error.WriteLine(ex.Message);
// }
// finally
// {
// Interlocked.Decrement(ref clients);
// }
//}
private static void BatchRunPings(IDatabase db)
{
Task[] tasks = new Task[InnerCount];
for (int i = 0; i < IterationCount; i++)
{
var batch = db.CreateBatch();
for (int j = 0; j < tasks.Length; j++)
{
tasks[j] = batch.PingAsync();
}
batch.Execute();
db.Multiplexer.WaitAll(tasks);
}
}
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment