Unverified Commit 988d6aef authored by Marc Gravell's avatar Marc Gravell Committed by GitHub

Arenas (#1075)

* remove the entire concept of the completion manager; it just doesn't make sense any more; everything async and external facing should be via the TP

* use "in" with ForAwait to avoid some extra copies

* experimental "backlog queue" approach

* Cleanup and de-dupe timeout exception data

* WriteMessageTakingWriteLockSync should consider backlog

* don't allocate all those strings

* this compiles, but nothing else; not tested yet

* fix the by-ref indexer

* record that CompletedSynchronously has changed - impacts a test

* update benchmarks and rev unofficial

* update to  pipelines 1.1.2 and simplify

* in debug mode, use a small page size in the arena

* rev pipelines to 1.1.3

* Update src/StackExchange.Redis/CommandTrace.cs
Co-Authored-By: 's avatarmgravell <marc.gravell@gmail.com>

* Perf regression (#1076)

* compare baseline to 1.2.6

* remove the entire concept of the completion manager; it just doesn't make sense any more; everything async and external facing should be via the TP

* use "in" with ForAwait to avoid some extra copies

* experimental "backlog queue" approach

* Cleanup and de-dupe timeout exception data

* WriteMessageTakingWriteLockSync should consider backlog

* don't allocate all those strings
parent cfe491ee
...@@ -78,12 +78,13 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes ...@@ -78,12 +78,13 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
case ResultType.MultiBulk: case ResultType.MultiBulk:
var parts = result.GetItems(); var parts = result.GetItems();
CommandTrace[] arr = new CommandTrace[parts.Length]; CommandTrace[] arr = new CommandTrace[parts.Length];
for (int i = 0; i < parts.Length; i++) int i = 0;
foreach(var item in parts)
{ {
var subParts = parts[i].GetItems(); var subParts = item.GetItems();
if (!subParts[0].TryGetInt64(out long uniqueid) || !subParts[1].TryGetInt64(out long time) || !subParts[2].TryGetInt64(out long duration)) if (!subParts[0].TryGetInt64(out long uniqueid) || !subParts[1].TryGetInt64(out long time) || !subParts[2].TryGetInt64(out long duration))
return false; return false;
arr[i] = new CommandTrace(uniqueid, time, duration, subParts[3].GetItemsAsValues()); arr[i++] = new CommandTrace(uniqueid, time, duration, subParts[3].GetItemsAsValues());
} }
SetResult(message, arr); SetResult(message, arr);
return true; return true;
......
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
using System.Security.Authentication; using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates; using System.Security.Cryptography.X509Certificates;
using System.Text; using System.Text;
using Pipelines.Sockets.Unofficial.Arenas;
namespace StackExchange.Redis namespace StackExchange.Redis
{ {
...@@ -271,5 +272,13 @@ internal static int VectorSafeIndexOfCRLF(this ReadOnlySpan<byte> span) ...@@ -271,5 +272,13 @@ internal static int VectorSafeIndexOfCRLF(this ReadOnlySpan<byte> span)
return -1; return -1;
} }
#endif #endif
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal static T[] ToArray<T>(in this RawResult result, Projection<RawResult, T> selector)
=> result.IsNull ? null : result.GetItems().ToArray(selector);
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal static TTo[] ToArray<TTo, TState>(in this RawResult result, Projection<RawResult, TState, TTo> selector, in TState state)
=> result.IsNull ? null : result.GetItems().ToArray(selector, in state);
} }
} }
...@@ -17,6 +17,7 @@ ...@@ -17,6 +17,7 @@
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Pipelines.Sockets.Unofficial; using Pipelines.Sockets.Unofficial;
using Pipelines.Sockets.Unofficial.Arenas;
namespace StackExchange.Redis namespace StackExchange.Redis
{ {
...@@ -55,7 +56,7 @@ private static readonly Message ...@@ -55,7 +56,7 @@ private static readonly Message
internal void GetBytes(out long sent, out long received) internal void GetBytes(out long sent, out long received)
{ {
if(_ioPipe is IMeasuredDuplexPipe sc) if (_ioPipe is IMeasuredDuplexPipe sc)
{ {
sent = sc.TotalBytesSent; sent = sc.TotalBytesSent;
received = sc.TotalBytesReceived; received = sc.TotalBytesReceived;
...@@ -268,6 +269,7 @@ public void Dispose() ...@@ -268,6 +269,7 @@ public void Dispose()
RecordConnectionFailed(ConnectionFailureType.ConnectionDisposed); RecordConnectionFailed(ConnectionFailureType.ConnectionDisposed);
} }
OnCloseEcho(); OnCloseEcho();
_arena.Dispose();
GC.SuppressFinalize(this); GC.SuppressFinalize(this);
} }
...@@ -1411,6 +1413,12 @@ private async Task ReadFromPipe() ...@@ -1411,6 +1413,12 @@ private async Task ReadFromPipe()
} }
} }
private static readonly ArenaOptions s_arenaOptions = new ArenaOptions(
#if DEBUG
blockSizeBytes: Unsafe.SizeOf<RawResult>() * 8 // force an absurdly small page size to trigger bugs
#endif
);
private readonly Arena<RawResult> _arena = new Arena<RawResult>(s_arenaOptions);
private int ProcessBuffer(ref ReadOnlySequence<byte> buffer) private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
{ {
int messageCount = 0; int messageCount = 0;
...@@ -1418,7 +1426,7 @@ private int ProcessBuffer(ref ReadOnlySequence<byte> buffer) ...@@ -1418,7 +1426,7 @@ private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
while (!buffer.IsEmpty) while (!buffer.IsEmpty)
{ {
var reader = new BufferReader(buffer); var reader = new BufferReader(buffer);
var result = TryParseResult(in buffer, ref reader, IncludeDetailInExceptions, BridgeCouldBeNull?.ServerEndPoint); var result = TryParseResult(_arena, in buffer, ref reader, IncludeDetailInExceptions, BridgeCouldBeNull?.ServerEndPoint);
try try
{ {
if (result.HasValue) if (result.HasValue)
...@@ -1436,7 +1444,7 @@ private int ProcessBuffer(ref ReadOnlySequence<byte> buffer) ...@@ -1436,7 +1444,7 @@ private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
} }
finally finally
{ {
result.Recycle(); _arena.Reset();
} }
} }
return messageCount; return messageCount;
...@@ -1466,7 +1474,7 @@ private int ProcessBuffer(ref ReadOnlySequence<byte> buffer) ...@@ -1466,7 +1474,7 @@ private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
// } // }
//} //}
private static RawResult ReadArray(in ReadOnlySequence<byte> buffer, ref BufferReader reader, bool includeDetailInExceptions, ServerEndPoint server) private static RawResult ReadArray(Arena<RawResult> arena, in ReadOnlySequence<byte> buffer, ref BufferReader reader, bool includeDetailInExceptions, ServerEndPoint server)
{ {
var itemCount = ReadLineTerminatedString(ResultType.Integer, ref reader); var itemCount = ReadLineTerminatedString(ResultType.Integer, ref reader);
if (itemCount.HasValue) if (itemCount.HasValue)
...@@ -1485,16 +1493,33 @@ private static RawResult ReadArray(in ReadOnlySequence<byte> buffer, ref BufferR ...@@ -1485,16 +1493,33 @@ private static RawResult ReadArray(in ReadOnlySequence<byte> buffer, ref BufferR
return RawResult.EmptyMultiBulk; return RawResult.EmptyMultiBulk;
} }
var oversized = ArrayPool<RawResult>.Shared.Rent(itemCountActual); var oversized = arena.Allocate(itemCountActual);
var result = new RawResult(oversized, itemCountActual); var result = new RawResult(oversized, false);
for (int i = 0; i < itemCountActual; i++)
if (oversized.IsSingleSegment)
{
var span = oversized.FirstSpan;
for(int i = 0; i < span.Length; i++)
{ {
if (!(oversized[i] = TryParseResult(in buffer, ref reader, includeDetailInExceptions, server)).HasValue) if (!(span[i] = TryParseResult(arena, in buffer, ref reader, includeDetailInExceptions, server)).HasValue)
{ {
result.Recycle(i); // passing index here means we don't need to "Array.Clear" before-hand
return RawResult.Nil; return RawResult.Nil;
} }
} }
}
else
{
foreach(var span in oversized.Spans)
{
for (int i = 0; i < span.Length; i++)
{
if (!(span[i] = TryParseResult(arena, in buffer, ref reader, includeDetailInExceptions, server)).HasValue)
{
return RawResult.Nil;
}
}
}
}
return result; return result;
} }
return RawResult.Nil; return RawResult.Nil;
...@@ -1541,7 +1566,7 @@ private static RawResult ReadLineTerminatedString(ResultType type, ref BufferRea ...@@ -1541,7 +1566,7 @@ private static RawResult ReadLineTerminatedString(ResultType type, ref BufferRea
internal void StartReading() => ReadFromPipe().RedisFireAndForget(); internal void StartReading() => ReadFromPipe().RedisFireAndForget();
internal static RawResult TryParseResult(in ReadOnlySequence<byte> buffer, ref BufferReader reader, internal static RawResult TryParseResult(Arena<RawResult> arena, in ReadOnlySequence<byte> buffer, ref BufferReader reader,
bool includeDetilInExceptions, ServerEndPoint server, bool allowInlineProtocol = false) bool includeDetilInExceptions, ServerEndPoint server, bool allowInlineProtocol = false)
{ {
var prefix = reader.PeekByte(); var prefix = reader.PeekByte();
...@@ -1562,15 +1587,15 @@ private static RawResult ReadLineTerminatedString(ResultType type, ref BufferRea ...@@ -1562,15 +1587,15 @@ private static RawResult ReadLineTerminatedString(ResultType type, ref BufferRea
return ReadBulkString(ref reader, includeDetilInExceptions, server); return ReadBulkString(ref reader, includeDetilInExceptions, server);
case '*': // array case '*': // array
reader.Consume(1); reader.Consume(1);
return ReadArray(in buffer, ref reader, includeDetilInExceptions, server); return ReadArray(arena, in buffer, ref reader, includeDetilInExceptions, server);
default: default:
// string s = Format.GetString(buffer); // string s = Format.GetString(buffer);
if (allowInlineProtocol) return ParseInlineProtocol(ReadLineTerminatedString(ResultType.SimpleString, ref reader)); if (allowInlineProtocol) return ParseInlineProtocol(arena, ReadLineTerminatedString(ResultType.SimpleString, ref reader));
throw new InvalidOperationException("Unexpected response prefix: " + (char)prefix); throw new InvalidOperationException("Unexpected response prefix: " + (char)prefix);
} }
} }
private static RawResult ParseInlineProtocol(in RawResult line) private static RawResult ParseInlineProtocol(Arena<RawResult> arena, in RawResult line)
{ {
if (!line.HasValue) return RawResult.Nil; // incomplete line if (!line.HasValue) return RawResult.Nil; // incomplete line
...@@ -1578,13 +1603,14 @@ private static RawResult ParseInlineProtocol(in RawResult line) ...@@ -1578,13 +1603,14 @@ private static RawResult ParseInlineProtocol(in RawResult line)
#pragma warning disable IDE0059 #pragma warning disable IDE0059
foreach (var _ in line.GetInlineTokenizer()) count++; foreach (var _ in line.GetInlineTokenizer()) count++;
#pragma warning restore IDE0059 #pragma warning restore IDE0059
var oversized = ArrayPool<RawResult>.Shared.Rent(count); var block = arena.Allocate(count);
count = 0;
var iter = block.GetEnumerator();
foreach (var token in line.GetInlineTokenizer()) foreach (var token in line.GetInlineTokenizer())
{ { // this assigns *via a reference*, returned via the iterator; just... sweet
oversized[count++] = new RawResult(line.Type, token, false); iter.GetNext() = new RawResult(line.Type, token, false);
} }
return new RawResult(oversized, count); return new RawResult(block, false);
} }
} }
} }
using System; using System;
using System.Buffers; using System.Buffers;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices; using System.Runtime.InteropServices;
using System.Text; using System.Text;
using Pipelines.Sockets.Unofficial.Arenas;
namespace StackExchange.Redis namespace StackExchange.Redis
{ {
internal readonly struct RawResult internal readonly struct RawResult
{ {
internal RawResult this[int index] internal ref RawResult this[int index] => ref GetItems()[index];
{
get internal int ItemsCount => (int)_items.Length;
{
if (index >= ItemsCount) throw new IndexOutOfRangeException();
return _itemsOversized[index];
}
}
internal int ItemsCount { get; }
internal ReadOnlySequence<byte> Payload { get; } internal ReadOnlySequence<byte> Payload { get; }
internal static readonly RawResult NullMultiBulk = new RawResult(null, 0); internal static readonly RawResult NullMultiBulk = new RawResult(default(Sequence<RawResult>), isNull: true);
internal static readonly RawResult EmptyMultiBulk = new RawResult(Array.Empty<RawResult>(), 0); internal static readonly RawResult EmptyMultiBulk = new RawResult(default(Sequence<RawResult>), isNull: false);
internal static readonly RawResult Nil = default; internal static readonly RawResult Nil = default;
// note: can't use Memory<RawResult> here - struct recursion breaks runtimr // note: can't use Memory<RawResult> here - struct recursion breaks runtimr
private readonly RawResult[] _itemsOversized; private readonly Sequence _items;
private readonly ResultType _type; private readonly ResultType _type;
private const ResultType NonNullFlag = (ResultType)128; private const ResultType NonNullFlag = (ResultType)128;
...@@ -42,17 +38,14 @@ public RawResult(ResultType resultType, in ReadOnlySequence<byte> payload, bool ...@@ -42,17 +38,14 @@ public RawResult(ResultType resultType, in ReadOnlySequence<byte> payload, bool
if (!isNull) resultType |= NonNullFlag; if (!isNull) resultType |= NonNullFlag;
_type = resultType; _type = resultType;
Payload = payload; Payload = payload;
_itemsOversized = default; _items = default;
ItemsCount = default;
} }
public RawResult(RawResult[] itemsOversized, int itemCount) public RawResult(Sequence<RawResult> items, bool isNull)
{ {
_type = ResultType.MultiBulk; _type = isNull ? ResultType.MultiBulk : (ResultType.MultiBulk | NonNullFlag);
if (itemsOversized != null) _type |= NonNullFlag;
Payload = default; Payload = default;
_itemsOversized = itemsOversized; _items = items.Untyped();
ItemsCount = itemCount;
} }
public bool IsError => Type == ResultType.Error; public bool IsError => Type == ResultType.Error;
...@@ -195,20 +188,6 @@ internal Lease<byte> AsLease() ...@@ -195,20 +188,6 @@ internal Lease<byte> AsLease()
throw new InvalidCastException("Cannot convert to Lease: " + Type); throw new InvalidCastException("Cannot convert to Lease: " + Type);
} }
internal void Recycle(int limit = -1)
{
var arr = _itemsOversized;
if (arr != null)
{
if (limit < 0) limit = ItemsCount;
for (int i = 0; i < limit; i++)
{
arr[i].Recycle();
}
ArrayPool<RawResult>.Shared.Return(arr, clearArray: false);
}
}
internal bool IsEqual(in CommandBytes expected) internal bool IsEqual(in CommandBytes expected)
{ {
if (expected.Length != Payload.Length) return false; if (expected.Length != Payload.Length) return false;
...@@ -284,83 +263,17 @@ internal bool GetBoolean() ...@@ -284,83 +263,17 @@ internal bool GetBoolean()
} }
} }
internal ReadOnlySpan<RawResult> GetItems() [MethodImpl(MethodImplOptions.AggressiveInlining)]
{ internal Sequence<RawResult> GetItems() => _items.Cast<RawResult>();
if (Type == ResultType.MultiBulk)
return new ReadOnlySpan<RawResult>(_itemsOversized, 0, ItemsCount);
throw new InvalidOperationException();
}
internal ReadOnlyMemory<RawResult> GetItemsMemory()
{
if (Type == ResultType.MultiBulk)
return new ReadOnlyMemory<RawResult>(_itemsOversized, 0, ItemsCount);
throw new InvalidOperationException();
}
internal RedisKey[] GetItemsAsKeys() [MethodImpl(MethodImplOptions.AggressiveInlining)]
{ internal RedisKey[] GetItemsAsKeys() => this.ToArray<RedisKey>((in RawResult x) => x.AsRedisKey());
var items = GetItems();
if (IsNull)
{
return null;
}
else if (items.Length == 0)
{
return Array.Empty<RedisKey>();
}
else
{
var arr = new RedisKey[items.Length];
for (int i = 0; i < arr.Length; i++)
{
arr[i] = items[i].AsRedisKey();
}
return arr;
}
}
internal RedisValue[] GetItemsAsValues() [MethodImpl(MethodImplOptions.AggressiveInlining)]
{ internal RedisValue[] GetItemsAsValues() => this.ToArray<RedisValue>((in RawResult x) => x.AsRedisValue());
var items = GetItems();
if (IsNull) [MethodImpl(MethodImplOptions.AggressiveInlining)]
{ internal string[] GetItemsAsStrings() => this.ToArray<string>((in RawResult x) => (string)x.AsRedisValue());
return null;
}
else if (items.Length == 0)
{
return RedisValue.EmptyArray;
}
else
{
var arr = new RedisValue[items.Length];
for (int i = 0; i < arr.Length; i++)
{
arr[i] = items[i].AsRedisValue();
}
return arr;
}
}
internal string[] GetItemsAsStrings()
{
var items = GetItems();
if (IsNull)
{
return null;
}
else if (items.Length == 0)
{
return Array.Empty<string>();
}
else
{
var arr = new string[items.Length];
for (int i = 0; i < arr.Length; i++)
{
arr[i] = (string)(items[i].AsRedisValue());
}
return arr;
}
}
internal GeoPosition? GetItemsAsGeoPosition() internal GeoPosition? GetItemsAsGeoPosition()
{ {
...@@ -370,43 +283,36 @@ internal string[] GetItemsAsStrings() ...@@ -370,43 +283,36 @@ internal string[] GetItemsAsStrings()
return null; return null;
} }
var coords = items[0].GetItems(); ref RawResult root = ref items[0];
if (items[0].IsNull) if (root.IsNull)
{ {
return null; return null;
} }
return new GeoPosition((double)coords[0].AsRedisValue(), (double)coords[1].AsRedisValue()); return AsGeoPosition(root.GetItems());
} }
internal GeoPosition?[] GetItemsAsGeoPositionArray() static GeoPosition AsGeoPosition(Sequence<RawResult> coords)
{
var items = GetItems();
if (IsNull)
{ {
return null; double longitude, latitude;
} if (coords.IsSingleSegment)
else if (items.Length == 0)
{ {
return Array.Empty<GeoPosition?>(); var span = coords.FirstSpan;
longitude = (double)span[0].AsRedisValue();
latitude = (double)span[1].AsRedisValue();
} }
else else
{ {
var arr = new GeoPosition?[items.Length]; var iter = coords.GetEnumerator();
for (int i = 0; i < arr.Length; i++) longitude = (double)iter.GetNext().AsRedisValue();
{ latitude = (double)iter.GetNext().AsRedisValue();
var item = items[i].GetItems();
if (items[i].IsNull)
{
arr[i] = null;
}
else
{
arr[i] = new GeoPosition((double)item[0].AsRedisValue(), (double)item[1].AsRedisValue());
}
}
return arr;
} }
return new GeoPosition(longitude, latitude);
} }
internal GeoPosition?[] GetItemsAsGeoPositionArray()
=> this.ToArray<GeoPosition?>((in RawResult item) => item.IsNull ? (GeoPosition?)null : AsGeoPosition(item.GetItems()));
internal unsafe string GetString() internal unsafe string GetString()
{ {
if (IsNull) return null; if (IsNull) return null;
......
...@@ -3494,13 +3494,16 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes ...@@ -3494,13 +3494,16 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
{ {
case ResultType.MultiBulk: case ResultType.MultiBulk:
var arr = result.GetItems(); var arr = result.GetItems();
long i64; if (arr.Length == 2)
if (arr.Length == 2 && arr[1].Type == ResultType.MultiBulk && arr[0].TryGetInt64(out i64))
{ {
var sscanResult = new ScanIterator<T>.ScanResult(i64, Parse(arr[1])); ref RawResult inner = ref arr[1];
if (inner.Type == ResultType.MultiBulk && arr[0].TryGetInt64(out var i64))
{
var sscanResult = new ScanIterator<T>.ScanResult(i64, Parse(inner));
SetResult(message, sscanResult); SetResult(message, sscanResult);
return true; return true;
} }
}
break; break;
} }
return false; return false;
......
...@@ -60,11 +60,13 @@ internal static RedisResult TryCreate(PhysicalConnection connection, in RawResul ...@@ -60,11 +60,13 @@ internal static RedisResult TryCreate(PhysicalConnection connection, in RawResul
var items = result.GetItems(); var items = result.GetItems();
if (items.Length == 0) return EmptyArray; if (items.Length == 0) return EmptyArray;
var arr = new RedisResult[items.Length]; var arr = new RedisResult[items.Length];
for (int i = 0; i < arr.Length; i++) int i = 0;
var iter = items.GetEnumerator();
while (iter.MoveNext())
{ {
var next = TryCreate(connection, items[i]); var next = TryCreate(connection, in iter.CurrentReference);
if (next == null) return null; // means we didn't understand if (next == null) return null; // means we didn't understand
arr[i] = next; arr[i++] = next;
} }
return new ArrayRedisResult(arr); return new ArrayRedisResult(arr);
case ResultType.Error: case ResultType.Error:
......
...@@ -755,9 +755,10 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes ...@@ -755,9 +755,10 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
case ResultType.MultiBulk: case ResultType.MultiBulk:
var arr = result.GetItems(); var arr = result.GetItems();
long i64; long i64;
if (arr.Length == 2 && arr[1].Type == ResultType.MultiBulk && arr[0].TryGetInt64(out i64)) RawResult inner;
if (arr.Length == 2 && (inner = arr[1]).Type == ResultType.MultiBulk && arr[0].TryGetInt64(out i64))
{ {
var keysResult = new ScanResult(i64, arr[1].GetItemsAsKeys()); var keysResult = new ScanResult(i64, inner.GetItemsAsKeys());
SetResult(message, keysResult); SetResult(message, keysResult);
return true; return true;
} }
......
...@@ -503,11 +503,14 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes ...@@ -503,11 +503,14 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
{ {
connection.Trace("Server committed; processing nested replies"); connection.Trace("Server committed; processing nested replies");
connection?.BridgeCouldBeNull?.Multiplexer?.OnTransactionLog($"processing {arr.Length} wrapped messages"); connection?.BridgeCouldBeNull?.Multiplexer?.OnTransactionLog($"processing {arr.Length} wrapped messages");
for (int i = 0; i < arr.Length; i++)
int i = 0;
var iter = arr.GetEnumerator();
while(iter.MoveNext())
{ {
var inner = wrapped[i].Wrapped; var inner = wrapped[i++].Wrapped;
connection?.BridgeCouldBeNull?.Multiplexer?.OnTransactionLog($"> got {arr[i]} for {inner.CommandAndKey}"); connection?.BridgeCouldBeNull?.Multiplexer?.OnTransactionLog($"> got {iter.Current} for {inner.CommandAndKey}");
if (inner.ComputeResult(connection, arr[i])) if (inner.ComputeResult(connection, iter.CurrentReference))
{ {
inner.Complete(); inner.Complete();
} }
......
This diff is collapsed.
...@@ -15,7 +15,7 @@ ...@@ -15,7 +15,7 @@
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="Pipelines.Sockets.Unofficial" Version="1.0.26" /> <PackageReference Include="Pipelines.Sockets.Unofficial" Version="1.1.3" />
<PackageReference Include="System.Diagnostics.PerformanceCounter" Version="4.5.0" /> <PackageReference Include="System.Diagnostics.PerformanceCounter" Version="4.5.0" />
<PackageReference Include="System.IO.Pipelines" Version="4.5.1" /> <PackageReference Include="System.IO.Pipelines" Version="4.5.1" />
<PackageReference Include="System.Threading.Channels" Version="4.5.0" /> <PackageReference Include="System.Threading.Channels" Version="4.5.0" />
......
...@@ -12,7 +12,7 @@ ...@@ -12,7 +12,7 @@
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="BenchmarkDotNet" Version="0.11.1" /> <PackageReference Include="BenchmarkDotNet" Version="0.11.4" />
<ProjectReference Include="..\..\src\StackExchange.Redis\StackExchange.Redis.csproj" /> <ProjectReference Include="..\..\src\StackExchange.Redis\StackExchange.Redis.csproj" />
</ItemGroup> </ItemGroup>
......
...@@ -26,7 +26,7 @@ protected virtual Job Configure(Job j) ...@@ -26,7 +26,7 @@ protected virtual Job Configure(Job j)
public CustomConfig() public CustomConfig()
{ {
Add(new MemoryDiagnoser()); Add(MemoryDiagnoser.Default);
Add(StatisticColumn.OperationsPerSecond); Add(StatisticColumn.OperationsPerSecond);
Add(JitOptimizationsValidator.FailOnError); Add(JitOptimizationsValidator.FailOnError);
...@@ -64,11 +64,18 @@ public void Setup() ...@@ -64,11 +64,18 @@ public void Setup()
connection = ConnectionMultiplexer.Connect(options); connection = ConnectionMultiplexer.Connect(options);
db = connection.GetDatabase(3); db = connection.GetDatabase(3);
db.KeyDelete(GeoKey, CommandFlags.FireAndForget); db.KeyDelete(GeoKey);
db.GeoAdd(GeoKey, 13.361389, 38.115556, "Palermo "); db.GeoAdd(GeoKey, 13.361389, 38.115556, "Palermo ");
db.GeoAdd(GeoKey, 15.087269, 37.502669, "Catania"); db.GeoAdd(GeoKey, 15.087269, 37.502669, "Catania");
db.KeyDelete(HashKey);
for (int i = 0; i < 1000; i++)
{
db.HashSet(HashKey, i, i);
}
} }
private static readonly RedisKey GeoKey = "GeoTest", IncrByKey = "counter", StringKey = "string";
private static readonly RedisKey GeoKey = "GeoTest", IncrByKey = "counter", StringKey = "string", HashKey = "hash";
void IDisposable.Dispose() void IDisposable.Dispose()
{ {
mgr?.Dispose(); mgr?.Dispose();
...@@ -78,16 +85,13 @@ void IDisposable.Dispose() ...@@ -78,16 +85,13 @@ void IDisposable.Dispose()
connection = null; connection = null;
} }
private const int COUNT = 500; private const int COUNT = 50;
/// <summary> /// <summary>
/// Run INCRBY lots of times /// Run INCRBY lots of times
/// </summary> /// </summary>
#if TEST_BASELINE // [Benchmark(Description = "INCRBY/s", OperationsPerInvoke = COUNT)]
// [Benchmark(Description = "INCRBY:v1/s", OperationsPerInvoke = COUNT)]
#else
// [Benchmark(Description = "INCRBY:v2/s", OperationsPerInvoke = COUNT)]
#endif
public int ExecuteIncrBy() public int ExecuteIncrBy()
{ {
var rand = new Random(12345); var rand = new Random(12345);
...@@ -108,11 +112,7 @@ public int ExecuteIncrBy() ...@@ -108,11 +112,7 @@ public int ExecuteIncrBy()
/// <summary> /// <summary>
/// Run INCRBY lots of times /// Run INCRBY lots of times
/// </summary> /// </summary>
#if TEST_BASELINE // [Benchmark(Description = "INCRBY/a", OperationsPerInvoke = COUNT)]
// [Benchmark(Description = "INCRBY:v1/a", OperationsPerInvoke = COUNT)]
#else
// [Benchmark(Description = "INCRBY:v2/a", OperationsPerInvoke = COUNT)]
#endif
public async Task<int> ExecuteIncrByAsync() public async Task<int> ExecuteIncrByAsync()
{ {
var rand = new Random(12345); var rand = new Random(12345);
...@@ -133,11 +133,7 @@ public async Task<int> ExecuteIncrByAsync() ...@@ -133,11 +133,7 @@ public async Task<int> ExecuteIncrByAsync()
/// <summary> /// <summary>
/// Run GEORADIUS lots of times /// Run GEORADIUS lots of times
/// </summary> /// </summary>
#if TEST_BASELINE // [Benchmark(Description = "GEORADIUS/s", OperationsPerInvoke = COUNT)]
// [Benchmark(Description = "GEORADIUS:v1/s", OperationsPerInvoke = COUNT)]
#else
// [Benchmark(Description = "GEORADIUS:v2/s", OperationsPerInvoke = COUNT)]
#endif
public int ExecuteGeoRadius() public int ExecuteGeoRadius()
{ {
int total = 0; int total = 0;
...@@ -153,11 +149,7 @@ public int ExecuteGeoRadius() ...@@ -153,11 +149,7 @@ public int ExecuteGeoRadius()
/// <summary> /// <summary>
/// Run GEORADIUS lots of times /// Run GEORADIUS lots of times
/// </summary> /// </summary>
#if TEST_BASELINE // [Benchmark(Description = "GEORADIUS/a", OperationsPerInvoke = COUNT)]
// [Benchmark(Description = "GEORADIUS:v1/a", OperationsPerInvoke = COUNT)]
#else
// [Benchmark(Description = "GEORADIUS:v2/a", OperationsPerInvoke = COUNT)]
#endif
public async Task<int> ExecuteGeoRadiusAsync() public async Task<int> ExecuteGeoRadiusAsync()
{ {
int total = 0; int total = 0;
...@@ -173,11 +165,7 @@ public async Task<int> ExecuteGeoRadiusAsync() ...@@ -173,11 +165,7 @@ public async Task<int> ExecuteGeoRadiusAsync()
/// <summary> /// <summary>
/// Run StringSet lots of times /// Run StringSet lots of times
/// </summary> /// </summary>
#if TEST_BASELINE [Benchmark(Description = "StringSet/s", OperationsPerInvoke = COUNT)]
[Benchmark(Description = "StringSet:v1/a", OperationsPerInvoke = COUNT)]
#else
[Benchmark(Description = "StringSet:v2/a", OperationsPerInvoke = COUNT)]
#endif
public void StringSet() public void StringSet()
{ {
for (int i = 0; i < COUNT; i++) for (int i = 0; i < COUNT; i++)
...@@ -189,11 +177,7 @@ public void StringSet() ...@@ -189,11 +177,7 @@ public void StringSet()
/// <summary> /// <summary>
/// Run StringGet lots of times /// Run StringGet lots of times
/// </summary> /// </summary>
#if TEST_BASELINE [Benchmark(Description = "StringGet/s", OperationsPerInvoke = COUNT)]
[Benchmark(Description = "StringGet:v1/a", OperationsPerInvoke = COUNT)]
#else
[Benchmark(Description = "StringGet:v2/a", OperationsPerInvoke = COUNT)]
#endif
public void StringGet() public void StringGet()
{ {
for (int i = 0; i < COUNT; i++) for (int i = 0; i < COUNT; i++)
...@@ -201,6 +185,33 @@ public void StringGet() ...@@ -201,6 +185,33 @@ public void StringGet()
db.StringGet(StringKey); db.StringGet(StringKey);
} }
} }
/// <summary>
/// Run HashGetAll lots of times
/// </summary>
[Benchmark(Description = "HashGetAll F+F/s", OperationsPerInvoke = COUNT)]
public void HashGetAll_FAF()
{
for (int i = 0; i < COUNT; i++)
{
db.HashGetAll(HashKey, CommandFlags.FireAndForget);
db.Ping(); // to wait for response
}
}
/// <summary>
/// Run HashGetAll lots of times
/// </summary>
[Benchmark(Description = "HashGetAll F+F/a", OperationsPerInvoke = COUNT)]
public async Task HashGetAllAsync_FAF()
{
for (int i = 0; i < COUNT; i++)
{
await db.HashGetAllAsync(HashKey, CommandFlags.FireAndForget);
await db.PingAsync(); // to wait for response
}
}
} }
#pragma warning disable CS1591 #pragma warning disable CS1591
......
...@@ -17,8 +17,8 @@ ...@@ -17,8 +17,8 @@
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="BenchmarkDotNet" Version="0.11.1" /> <PackageReference Include="BenchmarkDotNet" Version="0.11.4" />
<PackageReference Include="StackExchange.Redis" Version="[1.2.7-alpha-00002]" /> <PackageReference Include="StackExchange.Redis" Version="[2.0.519]" />
</ItemGroup> </ItemGroup>
</Project> </Project>
...@@ -2,14 +2,15 @@ ...@@ -2,14 +2,15 @@
using System.Buffers; using System.Buffers;
using System.Collections.Generic; using System.Collections.Generic;
using System.Text; using System.Text;
using Pipelines.Sockets.Unofficial.Arenas;
using Xunit; using Xunit;
using Xunit.Abstractions; using Xunit.Abstractions;
namespace StackExchange.Redis.Tests namespace StackExchange.Redis.Tests
{ {
public class Parse : TestBase public class ParseTests : TestBase
{ {
public Parse(ITestOutputHelper output) : base(output) { } public ParseTests(ITestOutputHelper output) : base(output) { }
public static IEnumerable<object[]> GetTestData() public static IEnumerable<object[]> GetTestData()
{ {
...@@ -34,7 +35,10 @@ public static IEnumerable<object[]> GetTestData() ...@@ -34,7 +35,10 @@ public static IEnumerable<object[]> GetTestData()
public void ParseAsSingleChunk(string ascii, int expected) public void ParseAsSingleChunk(string ascii, int expected)
{ {
var buffer = new ReadOnlySequence<byte>(Encoding.ASCII.GetBytes(ascii)); var buffer = new ReadOnlySequence<byte>(Encoding.ASCII.GetBytes(ascii));
ProcessMessages(buffer, expected); using (var arena = new Arena<RawResult>())
{
ProcessMessages(arena, buffer, expected);
}
} }
[Theory] [Theory]
...@@ -58,16 +62,19 @@ public void ParseAsLotsOfChunks(string ascii, int expected) ...@@ -58,16 +62,19 @@ public void ParseAsLotsOfChunks(string ascii, int expected)
} }
var buffer = new ReadOnlySequence<byte>(chain, 0, tail, 1); var buffer = new ReadOnlySequence<byte>(chain, 0, tail, 1);
Assert.Equal(bytes.Length, buffer.Length); Assert.Equal(bytes.Length, buffer.Length);
ProcessMessages(buffer, expected); using (var arena = new Arena<RawResult>())
{
ProcessMessages(arena, buffer, expected);
}
} }
private void ProcessMessages(ReadOnlySequence<byte> buffer, int expected) private void ProcessMessages(Arena<RawResult> arena, ReadOnlySequence<byte> buffer, int expected)
{ {
Writer.WriteLine($"chain: {buffer.Length}"); Writer.WriteLine($"chain: {buffer.Length}");
var reader = new BufferReader(buffer); var reader = new BufferReader(buffer);
RawResult result; RawResult result;
int found = 0; int found = 0;
while (!(result = PhysicalConnection.TryParseResult(buffer, ref reader, false, null, false)).IsNull) while (!(result = PhysicalConnection.TryParseResult(arena, buffer, ref reader, false, null, false)).IsNull)
{ {
Writer.WriteLine($"{result} - {result.GetString()}"); Writer.WriteLine($"{result} - {result.GetString()}");
found++; found++;
......
...@@ -14,6 +14,8 @@ public class TestInfoReplicationChecks : TestBase ...@@ -14,6 +14,8 @@ public class TestInfoReplicationChecks : TestBase
[Fact] [Fact]
public async Task Exec() public async Task Exec()
{ {
Skip.Inconclusive("need to think about CompletedSynchronously");
using(var conn = Create()) using(var conn = Create())
{ {
var parsed = ConfigurationOptions.Parse(conn.Configuration); var parsed = ConfigurationOptions.Parse(conn.Configuration);
......
...@@ -34,8 +34,6 @@ internal RedisRequest(in RawResult result) ...@@ -34,8 +34,6 @@ internal RedisRequest(in RawResult result)
Count = result.ItemsCount; Count = result.ItemsCount;
} }
internal void Recycle() => _inner.Recycle();
public RedisValue GetValue(int index) public RedisValue GetValue(int index)
=> _inner[index].AsRedisValue(); => _inner[index].AsRedisValue();
......
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Pipelines.Sockets.Unofficial; using Pipelines.Sockets.Unofficial;
using Pipelines.Sockets.Unofficial.Arenas;
namespace StackExchange.Redis.Server namespace StackExchange.Redis.Server
{ {
...@@ -241,6 +242,7 @@ protected void DoShutdown(ShutdownReason reason) ...@@ -241,6 +242,7 @@ protected void DoShutdown(ShutdownReason reason)
public void Dispose() => Dispose(true); public void Dispose() => Dispose(true);
protected virtual void Dispose(bool disposing) protected virtual void Dispose(bool disposing)
{ {
_arena.Dispose();
DoShutdown(ShutdownReason.ServerDisposed); DoShutdown(ShutdownReason.ServerDisposed);
} }
...@@ -363,10 +365,11 @@ void WritePrefix(PipeWriter ooutput, char pprefix) ...@@ -363,10 +365,11 @@ void WritePrefix(PipeWriter ooutput, char pprefix)
} }
await output.FlushAsync().ConfigureAwait(false); await output.FlushAsync().ConfigureAwait(false);
} }
public static bool TryParseRequest(ref ReadOnlySequence<byte> buffer, out RedisRequest request)
private static bool TryParseRequest(Arena<RawResult> arena, ref ReadOnlySequence<byte> buffer, out RedisRequest request)
{ {
var reader = new BufferReader(buffer); var reader = new BufferReader(buffer);
var raw = PhysicalConnection.TryParseResult(in buffer, ref reader, false, null, true); var raw = PhysicalConnection.TryParseResult(arena, in buffer, ref reader, false, null, true);
if (raw.HasValue) if (raw.HasValue)
{ {
buffer = reader.SliceFromCurrent(); buffer = reader.SliceFromCurrent();
...@@ -377,6 +380,9 @@ public static bool TryParseRequest(ref ReadOnlySequence<byte> buffer, out RedisR ...@@ -377,6 +380,9 @@ public static bool TryParseRequest(ref ReadOnlySequence<byte> buffer, out RedisR
return false; return false;
} }
private readonly Arena<RawResult> _arena = new Arena<RawResult>();
public ValueTask<bool> TryProcessRequestAsync(ref ReadOnlySequence<byte> buffer, RedisClient client, PipeWriter output) public ValueTask<bool> TryProcessRequestAsync(ref ReadOnlySequence<byte> buffer, RedisClient client, PipeWriter output)
{ {
async ValueTask<bool> Awaited(ValueTask wwrite, TypedRedisValue rresponse) async ValueTask<bool> Awaited(ValueTask wwrite, TypedRedisValue rresponse)
...@@ -385,11 +391,11 @@ async ValueTask<bool> Awaited(ValueTask wwrite, TypedRedisValue rresponse) ...@@ -385,11 +391,11 @@ async ValueTask<bool> Awaited(ValueTask wwrite, TypedRedisValue rresponse)
rresponse.Recycle(); rresponse.Recycle();
return true; return true;
} }
if (!buffer.IsEmpty && TryParseRequest(ref buffer, out var request)) if (!buffer.IsEmpty && TryParseRequest(_arena, ref buffer, out var request))
{ {
TypedRedisValue response; TypedRedisValue response;
try { response = Execute(client, request); } try { response = Execute(client, request); }
finally { request.Recycle(); } finally { _arena.Reset(); }
var write = WriteResponseAsync(client, output, response); var write = WriteResponseAsync(client, output, response);
if (!write.IsCompletedSuccessfully) return Awaited(write, response); if (!write.IsCompletedSuccessfully) return Awaited(write, response);
......
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