Commit a52077c2 authored by Marc Gravell's avatar Marc Gravell

make roslynator happy; update sockets-unofficial

parent e75d0746
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
<GenerateDocumentationFile>true</GenerateDocumentationFile> <GenerateDocumentationFile>true</GenerateDocumentationFile>
<AssemblyOriginatorKeyFile>../StackExchange.Redis.snk</AssemblyOriginatorKeyFile> <AssemblyOriginatorKeyFile>../StackExchange.Redis.snk</AssemblyOriginatorKeyFile>
<PackageId>$(AssemblyName)</PackageId> <PackageId>$(AssemblyName)</PackageId>
<Features>strict</Features>
<Authors>Stack Exchange, Inc.; marc.gravell</Authors> <Authors>Stack Exchange, Inc.; marc.gravell</Authors>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks> <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
<CodeAnalysisRuleset>$(MSBuildThisFileDirectory)Shared.ruleset</CodeAnalysisRuleset> <CodeAnalysisRuleset>$(MSBuildThisFileDirectory)Shared.ruleset</CodeAnalysisRuleset>
......
...@@ -21,7 +21,9 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "RedisConfigs", "RedisConfig ...@@ -21,7 +21,9 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "RedisConfigs", "RedisConfig
RedisConfigs\cli-secure.cmd = RedisConfigs\cli-secure.cmd RedisConfigs\cli-secure.cmd = RedisConfigs\cli-secure.cmd
RedisConfigs\cli-slave.cmd = RedisConfigs\cli-slave.cmd RedisConfigs\cli-slave.cmd = RedisConfigs\cli-slave.cmd
RedisConfigs\start-all.cmd = RedisConfigs\start-all.cmd RedisConfigs\start-all.cmd = RedisConfigs\start-all.cmd
RedisConfigs\start-all.sh = RedisConfigs\start-all.sh
RedisConfigs\start-basic.cmd = RedisConfigs\start-basic.cmd RedisConfigs\start-basic.cmd = RedisConfigs\start-basic.cmd
RedisConfigs\start-basic.sh = RedisConfigs\start-basic.sh
RedisConfigs\start-cluster.cmd = RedisConfigs\start-cluster.cmd RedisConfigs\start-cluster.cmd = RedisConfigs\start-cluster.cmd
RedisConfigs\start-sentinel.cmd = RedisConfigs\start-sentinel.cmd RedisConfigs\start-sentinel.cmd = RedisConfigs\start-sentinel.cmd
EndProjectSection EndProjectSection
......
...@@ -26,7 +26,7 @@ ...@@ -26,7 +26,7 @@
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="Pipelines.Sockets.Unofficial" Version="0.2.1-alpha.72" /> <PackageReference Include="Pipelines.Sockets.Unofficial" Version="0.2.1-alpha.73" />
<PackageReference Include="System.Threading.Channels" Version="4.5.0" /> <PackageReference Include="System.Threading.Channels" Version="4.5.0" />
</ItemGroup> </ItemGroup>
</Project> </Project>
\ No newline at end of file
...@@ -125,22 +125,21 @@ internal static Exception NoConnectionAvailable(bool includeDetail, bool include ...@@ -125,22 +125,21 @@ internal static Exception NoConnectionAvailable(bool includeDetail, bool include
internal static Exception PopulateInnerExceptions(ReadOnlySpan<ServerEndPoint> serverSnapshot) internal static Exception PopulateInnerExceptions(ReadOnlySpan<ServerEndPoint> serverSnapshot)
{ {
var innerExceptions = new List<Exception>(); var innerExceptions = new List<Exception>();
if (serverSnapshot != null)
if (serverSnapshot.Length > 0 && serverSnapshot[0].Multiplexer.LastException != null)
{ {
if (serverSnapshot.Length > 0 && serverSnapshot[0].Multiplexer.LastException != null) innerExceptions.Add(serverSnapshot[0].Multiplexer.LastException);
{ }
innerExceptions.Add(serverSnapshot[0].Multiplexer.LastException);
}
for (int i = 0; i < serverSnapshot.Length; i++) for (int i = 0; i < serverSnapshot.Length; i++)
{
if (serverSnapshot[i].LastException != null)
{ {
if (serverSnapshot[i].LastException != null) var lastException = serverSnapshot[i].LastException;
{ innerExceptions.Add(lastException);
var lastException = serverSnapshot[i].LastException;
innerExceptions.Add(lastException);
}
} }
} }
if (innerExceptions.Count == 1) if (innerExceptions.Count == 1)
{ {
return innerExceptions[0]; return innerExceptions[0];
......
...@@ -280,6 +280,7 @@ internal void ResetNonConnected() ...@@ -280,6 +280,7 @@ internal void ResetNonConnected()
internal void OnConnectionFailed(PhysicalConnection connection, ConnectionFailureType failureType, Exception innerException) internal void OnConnectionFailed(PhysicalConnection connection, ConnectionFailureType failureType, Exception innerException)
{ {
Trace($"OnConnectionFailed: {connection}");
AbandonPendingBacklog(innerException); AbandonPendingBacklog(innerException);
if (reportNextFailure) if (reportNextFailure)
{ {
...@@ -292,7 +293,7 @@ internal void OnConnectionFailed(PhysicalConnection connection, ConnectionFailur ...@@ -292,7 +293,7 @@ internal void OnConnectionFailed(PhysicalConnection connection, ConnectionFailur
internal void OnDisconnected(ConnectionFailureType failureType, PhysicalConnection connection, out bool isCurrent, out State oldState) internal void OnDisconnected(ConnectionFailureType failureType, PhysicalConnection connection, out bool isCurrent, out State oldState)
{ {
Trace("OnDisconnected"); Trace($"OnDisconnected: {failureType}");
oldState = default(State); // only defined when isCurrent = true oldState = default(State); // only defined when isCurrent = true
if (isCurrent = (physical == connection)) if (isCurrent = (physical == connection))
...@@ -344,7 +345,7 @@ private void AbandonPendingBacklog(Exception ex) ...@@ -344,7 +345,7 @@ private void AbandonPendingBacklog(Exception ex)
if(next != null) if(next != null)
{ {
Multiplexer?.OnMessageFaulted(next, ex); Multiplexer?.OnMessageFaulted(next, ex);
next.SetException(ex); next.SetException(ex);
this.CompleteSyncOrAsync(next); this.CompleteSyncOrAsync(next);
} }
} while (next != null); } while (next != null);
......
...@@ -277,7 +277,6 @@ public Task FlushAsync() ...@@ -277,7 +277,6 @@ public Task FlushAsync()
public void RecordConnectionFailed(ConnectionFailureType failureType, Exception innerException = null, [CallerMemberName] string origin = null, bool isInitialConnect = false) public void RecordConnectionFailed(ConnectionFailureType failureType, Exception innerException = null, [CallerMemberName] string origin = null, bool isInitialConnect = false)
{ {
Exception outerException = innerException; Exception outerException = innerException;
IdentifyFailureType(innerException, ref failureType); IdentifyFailureType(innerException, ref failureType);
var bridge = BridgeCouldBeNull; var bridge = BridgeCouldBeNull;
...@@ -831,8 +830,7 @@ private static int AppendToSpanCommand(Span<byte> span, CommandBytes value, int ...@@ -831,8 +830,7 @@ private static int AppendToSpanCommand(Span<byte> span, CommandBytes value, int
offset = WriteRaw(span, len, offset: offset); offset = WriteRaw(span, len, offset: offset);
value.CopyTo(span.Slice(offset, len)); value.CopyTo(span.Slice(offset, len));
offset += value.Length; offset += value.Length;
offset = WriteCrlf(span, offset); return WriteCrlf(span, offset);
return offset;
} }
private static int AppendToSpanSpan(Span<byte> span, ReadOnlySpan<byte> value, int offset = 0) private static int AppendToSpanSpan(Span<byte> span, ReadOnlySpan<byte> value, int offset = 0)
...@@ -840,8 +838,7 @@ private static int AppendToSpanSpan(Span<byte> span, ReadOnlySpan<byte> value, i ...@@ -840,8 +838,7 @@ private static int AppendToSpanSpan(Span<byte> span, ReadOnlySpan<byte> value, i
offset = WriteRaw(span, value.Length, offset: offset); offset = WriteRaw(span, value.Length, offset: offset);
value.CopyTo(span.Slice(offset, value.Length)); value.CopyTo(span.Slice(offset, value.Length));
offset += value.Length; offset += value.Length;
offset = WriteCrlf(span, offset); return WriteCrlf(span, offset);
return offset;
} }
internal void WriteSha1AsHex(byte[] value) internal void WriteSha1AsHex(byte[] value)
...@@ -1342,7 +1339,7 @@ private int ProcessBuffer(ref ReadOnlySequence<byte> buffer) ...@@ -1342,7 +1339,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(in ReadOnlySequence<byte> buffer, ref BufferReader reader, bool includeDetailInExceptions, ServerEndPoint server)
{ {
var itemCount = ReadLineTerminatedString(ResultType.Integer, in buffer, ref reader); var itemCount = ReadLineTerminatedString(ResultType.Integer, ref reader);
if (itemCount.HasValue) if (itemCount.HasValue)
{ {
if (!itemCount.TryGetInt64(out long i64)) throw ExceptionFactory.ConnectionFailure(includeDetailInExceptions, ConnectionFailureType.ProtocolFailure, "Invalid array length", server); if (!itemCount.TryGetInt64(out long i64)) throw ExceptionFactory.ConnectionFailure(includeDetailInExceptions, ConnectionFailureType.ProtocolFailure, "Invalid array length", server);
...@@ -1374,9 +1371,9 @@ private static RawResult ReadArray(in ReadOnlySequence<byte> buffer, ref BufferR ...@@ -1374,9 +1371,9 @@ private static RawResult ReadArray(in ReadOnlySequence<byte> buffer, ref BufferR
return RawResult.Nil; return RawResult.Nil;
} }
private static RawResult ReadBulkString(in ReadOnlySequence<byte> buffer, ref BufferReader reader, bool includeDetailInExceptions, ServerEndPoint server) private static RawResult ReadBulkString(ref BufferReader reader, bool includeDetailInExceptions, ServerEndPoint server)
{ {
var prefix = ReadLineTerminatedString(ResultType.Integer, in buffer, ref reader); var prefix = ReadLineTerminatedString(ResultType.Integer, ref reader);
if (prefix.HasValue) if (prefix.HasValue)
{ {
if (!prefix.TryGetInt64(out long i64)) throw ExceptionFactory.ConnectionFailure(includeDetailInExceptions, ConnectionFailureType.ProtocolFailure, "Invalid bulk string length", server); if (!prefix.TryGetInt64(out long i64)) throw ExceptionFactory.ConnectionFailure(includeDetailInExceptions, ConnectionFailureType.ProtocolFailure, "Invalid bulk string length", server);
...@@ -1402,7 +1399,7 @@ private static RawResult ReadBulkString(in ReadOnlySequence<byte> buffer, ref Bu ...@@ -1402,7 +1399,7 @@ private static RawResult ReadBulkString(in ReadOnlySequence<byte> buffer, ref Bu
return RawResult.Nil; return RawResult.Nil;
} }
private static RawResult ReadLineTerminatedString(ResultType type, in ReadOnlySequence<byte> buffer, ref BufferReader reader) private static RawResult ReadLineTerminatedString(ResultType type, ref BufferReader reader)
{ {
int crlfOffsetFromCurrent = BufferReader.FindNextCrLf(reader); int crlfOffsetFromCurrent = BufferReader.FindNextCrLf(reader);
if (crlfOffsetFromCurrent < 0) return RawResult.Nil; if (crlfOffsetFromCurrent < 0) return RawResult.Nil;
...@@ -1424,21 +1421,21 @@ private static RawResult ReadLineTerminatedString(ResultType type, in ReadOnlySe ...@@ -1424,21 +1421,21 @@ private static RawResult ReadLineTerminatedString(ResultType type, in ReadOnlySe
{ {
case '+': // simple string case '+': // simple string
reader.Consume(1); reader.Consume(1);
return ReadLineTerminatedString(ResultType.SimpleString, in buffer, ref reader); return ReadLineTerminatedString(ResultType.SimpleString, ref reader);
case '-': // error case '-': // error
reader.Consume(1); reader.Consume(1);
return ReadLineTerminatedString(ResultType.Error, in buffer, ref reader); return ReadLineTerminatedString(ResultType.Error, ref reader);
case ':': // integer case ':': // integer
reader.Consume(1); reader.Consume(1);
return ReadLineTerminatedString(ResultType.Integer, in buffer, ref reader); return ReadLineTerminatedString(ResultType.Integer, ref reader);
case '$': // bulk string case '$': // bulk string
reader.Consume(1); reader.Consume(1);
return ReadBulkString(in buffer, 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(in buffer, ref reader, includeDetilInExceptions, server);
default: default:
if (allowInlineProtocol) return ParseInlineProtocol(ReadLineTerminatedString(ResultType.SimpleString, in buffer, ref reader)); if (allowInlineProtocol) return ParseInlineProtocol(ReadLineTerminatedString(ResultType.SimpleString, ref reader));
throw new InvalidOperationException("Unexpected response prefix: " + (char)prefix); throw new InvalidOperationException("Unexpected response prefix: " + (char)prefix);
} }
} }
......
...@@ -1731,18 +1731,13 @@ protected NameValueEntry[] ParseStreamEntryValues(RawResult result) ...@@ -1731,18 +1731,13 @@ protected NameValueEntry[] ParseStreamEntryValues(RawResult result)
// 3) "temperature" // 3) "temperature"
// 4) "18.2" // 4) "18.2"
if (result.Type != ResultType.MultiBulk) if (result.Type != ResultType.MultiBulk || result.IsNull)
{ {
return null; return null;
} }
var arr = result.GetItems(); var arr = result.GetItems();
if (arr == null)
{
return null;
}
// Calculate how many name/value pairs are in the stream entry. // Calculate how many name/value pairs are in the stream entry.
int count = arr.Length / 2; int count = arr.Length / 2;
......
using System; using System;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using StackExchange.Redis; using StackExchange.Redis;
internal static class Program namespace TestConsole
{ {
private static void Main() internal static class Program
{ {
using (var muxer = ConnectionMultiplexer.Connect("localhost:6379", Console.Out)) private static void Main()
{ {
muxer.GetDatabase().Ping(); using (var muxer = ConnectionMultiplexer.Connect("localhost:6379", Console.Out))
}
}
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); muxer.GetDatabase().Ping();
} }
var tasks = new Task[ClientCount + 1]; }
tasks[0] = Task.Run(() => ShowState(cancel.Token)); //private static async Task Main2()
//{
// const int ClientCount = 150, ConnectionCount = 10;
// CancellationTokenSource cancel = new CancellationTokenSource();
for (int i = 1; i < tasks.Length; i++) // var config = new ConfigurationOptions
{ // {
var db = muxers[i % muxers.Length].GetDatabase(); // EndPoints = { new IPEndPoint(IPAddress.Loopback, 6379) }
int seed = i; // };
var key = "test_client_" + i; // var muxers = new ConnectionMultiplexer[ConnectionCount];
tasks[i] = Task.Run(() => RunClient(key, seed, db, cancel.Token)); // 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));
Console.ReadLine(); // for (int i = 1; i < tasks.Length; i++)
cancel.Cancel(); // {
await Task.WhenAll(tasks); // var db = muxers[i % muxers.Length].GetDatabase();
} // int seed = i;
finally // var key = "test_client_" + i;
{ // tasks[i] = Task.Run(() => RunClient(key, seed, db, cancel.Token));
for (int i = 0; i < muxers.Length; i++) // }
{
try { muxers[i]?.Dispose(); } catch { }
}
}
}
private static int clients; // Console.ReadLine();
private static long totalPings, pings, lastTicks; // cancel.Cancel();
private static async Task ShowState(CancellationToken cancellation) // await Task.WhenAll(tasks);
{ // }
while (!cancellation.IsCancellationRequested) // finally
{ // {
await Task.Delay(2000); // for (int i = 0; i < muxers.Length; i++)
var nowTicks = DateTime.UtcNow.Ticks; // {
var thenTicks = Interlocked.Exchange(ref lastTicks, nowTicks); // try { muxers[i]?.Dispose(); } catch { }
long pingsInInterval = Interlocked.Exchange(ref pings, 0); // }
var newTotalPings = Interlocked.Add(ref totalPings, pingsInInterval); // }
//}
var deltaTicks = nowTicks - thenTicks; //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);
Console.WriteLine($"[{Thread.VolatileRead(ref clients)}], Pings: {newTotalPings} ({pingsInInterval}, {Rate(pingsInInterval, deltaTicks)}/s)"); // var deltaTicks = nowTicks - thenTicks;
}
}
private static string Rate(long pingsInInterval, long deltaTicks) // Console.WriteLine($"[{Thread.VolatileRead(ref clients)}], Pings: {newTotalPings} ({pingsInInterval}, {Rate(pingsInInterval, deltaTicks)}/s)");
{ // }
if (deltaTicks == 0) return "n/a"; //}
if (pingsInInterval == 0) return "0";
var seconds = ((decimal)deltaTicks) / TimeSpan.TicksPerSecond; //private static string Rate(long pingsInInterval, long deltaTicks)
return (pingsInInterval / seconds).ToString("0.0"); //{
} // if (deltaTicks == 0) return "n/a";
// if (pingsInInterval == 0) return "0";
private static async Task RunClient(RedisKey key, int seed, IDatabase db, CancellationToken cancellation) // var seconds = ((decimal)deltaTicks) / TimeSpan.TicksPerSecond;
{ // return (pingsInInterval / seconds).ToString("0.0");
Interlocked.Increment(ref clients); //}
try
{ //private static async Task RunClient(RedisKey key, int seed, IDatabase db, CancellationToken cancellation)
while (!cancellation.IsCancellationRequested) //{
{ // Interlocked.Increment(ref clients);
await db.PingAsync(); // try
Interlocked.Increment(ref pings); // {
} // while (!cancellation.IsCancellationRequested)
} // {
catch(Exception ex) // await db.PingAsync();
{ // Interlocked.Increment(ref pings);
Console.Error.WriteLine(ex.Message); // }
} // }
finally // catch (Exception ex)
{ // {
Interlocked.Decrement(ref clients); // Console.Error.WriteLine(ex.Message);
} // }
// finally
// {
// Interlocked.Decrement(ref clients);
// }
//}
} }
} }
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