Skip to content

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

remove the concept of the unsent queue; the pipeline **is** that queue - no... 

remove the concept of the unsent queue; the pipeline **is** that queue - no need to duplicate (and significantly simplifies); if you've successfully pushed a message to the bridge: *it is in the pipe*; also removes "ar", "wr", "wq", "qu" and "pending" counters
parent 60f90766
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 88 additions and 383 deletions
StackExchange.Redis/StackExchange/Redis/ConnectionMultiplexer.ReaderWriter.cs
View file @ 795379da
......@@ -18,18 +18,6 @@ public partial class ConnectionMultiplexer
if (ownsSocketManager) socketManager?.Dispose();
socketManager = null;
}
internal void RequestWrite(PhysicalBridge bridge, bool forced)
{
if (bridge == null) return;
var tmp = SocketManager;
if (tmp != null)
{
Trace("Requesting write: " + bridge.Name);
tmp.RequestWrite(bridge, forced);
}
}
partial void OnWriterCreated();
}
}
StackExchange.Redis/StackExchange/Redis/ConnectionMultiplexer.cs
View file @ 795379da
......@@ -2007,10 +2007,6 @@ internal T ExecuteSyncImpl<T>(Message message, ResultProcessor<T> processor, Ser
}
else
{
#if FEATURE_SOCKET_MODE_POLL
var mgrState = socketManager.State;
var lastError = socketManager.LastErrorTimeRelative();
#endif
var sb = new StringBuilder("Timeout performing ").Append(message.CommandAndKey);
data = new List<Tuple<string, string>> { Tuple.Create("Message", message.CommandAndKey) };
void add(string lk, string sk, string v)
......@@ -2019,20 +2015,12 @@ void add(string lk, string sk, string v)
sb.Append(", ").Append(sk).Append(": ").Append(v);
}
int queue = server.GetOutstandingCount(message.Command, out int inst, out int qu, out int qs, out int qc, out int wr, out int wq, out int @in, out int ar);
int queue = server.GetOutstandingCount(message.Command, out int inst, out int qs, out int qc, out int @in);
add("Instantaneous", "inst", inst.ToString());
#if FEATURE_SOCKET_MODE_POLL
add("Manager-State", "mgr", mgrState.ToString());
add("Last-Error", "err", lastError);
#endif
add("Queue-Length", "queue", queue.ToString());
add("Queue-Outstanding", "qu", qu.ToString());
add("Queue-Awaiting-Response", "qs", qs.ToString());
add("Queue-Completion-Outstanding", "qc", qc.ToString());
add("Active-Writers", "wr", wr.ToString());
add("Write-Queue", "wq", wq.ToString());
add("Inbound-Bytes", "in", @in.ToString());
add("Active-Readers", "ar", ar.ToString());
add("Client-Name", "clientName", ClientName);
add("Server-Endpoint", "serverEndpoint", server.EndPoint.ToString());
......
StackExchange.Redis/StackExchange/Redis/DebuggingAids.cs
View file @ 795379da
......@@ -17,12 +17,6 @@ internal partial class ResultBox
public partial interface IServer
{
/// <summary>
/// Show what is in the pending (unsent) queue
/// </summary>
/// <param name="maxCount">The maximum count to list.</param>
string ListPending(int maxCount);
/// <summary>
/// Get the value of key. If the key does not exist the special value nil is returned. An error is returned if the value stored at key is not a string, because GET only handles string values.
/// </summary>
......@@ -114,21 +108,11 @@ internal void SimulateConnectionFailure()
interactive?.SimulateConnectionFailure();
subscription?.SimulateConnectionFailure();
}
internal string ListPending(int maxCount)
{
var sb = new StringBuilder();
interactive?.ListPending(sb, maxCount);
subscription?.ListPending(sb, maxCount);
return sb.ToString();
}
}
internal partial class RedisServer
{
void IServer.SimulateConnectionFailure() => server.SimulateConnectionFailure();
string IServer.ListPending(int maxCount) => server.ListPending(maxCount);
void IServer.Crash()
{
// using DB-0 because we also use "DEBUG OBJECT", which is db-centric
......@@ -225,11 +209,6 @@ internal void SimulateConnectionFailure()
}
physical?.RecordConnectionFailed(ConnectionFailureType.SocketFailure);
}
internal void ListPending(StringBuilder sb, int maxCount)
{
queue.ListPending(sb, maxCount);
}
}
internal partial class PhysicalConnection
......
StackExchange.Redis/StackExchange/Redis/PhysicalConnection.cs
View file @ 795379da
......@@ -169,14 +169,14 @@ public void RecordConnectionFailed(ConnectionFailureType failureType, ref Socket
// stop anything new coming in...
Bridge.Trace("Failed: " + failureType);
int @in = -1, ar = -1;
int @in = -1;
managerState = SocketManager.ManagerState.RecordConnectionFailed_OnDisconnected;
Bridge.OnDisconnected(failureType, this, out bool isCurrent, out PhysicalBridge.State oldState);
if (oldState == PhysicalBridge.State.ConnectedEstablished)
{
try
{
@in = GetAvailableInboundBytes(out ar);
@in = GetAvailableInboundBytes();
}
catch { /* best effort only */ }
}
......@@ -211,23 +211,16 @@ void add(string lk, string sk, string v)
add("Last-Write", "last-write", (unchecked(now - lastWrite) / 1000) + "s ago");
add("Unanswered-Write", "unanswered-write", (unchecked(now - unansweredRead) / 1000) + "s ago");
add("Keep-Alive", "keep-alive", Bridge.ServerEndPoint.WriteEverySeconds + "s");
add("Pending", "pending", Bridge.GetPendingCount().ToString());
add("Previous-Physical-State", "state", oldState.ToString());
if (@in >= 0)
{
add("Inbound-Bytes", "in", @in.ToString());
add("Active-Readers", "ar", ar.ToString());
}
add("Last-Heartbeat", "last-heartbeat", (lastBeat == 0 ? "never" : ((unchecked(now - lastBeat) / 1000) + "s ago")) + (Bridge.IsBeating ? " (mid-beat)" : ""));
add("Last-Multiplexer-Heartbeat", "last-mbeat", Multiplexer.LastHeartbeatSecondsAgo + "s ago");
add("Last-Global-Heartbeat", "global", ConnectionMultiplexer.LastGlobalHeartbeatSecondsAgo + "s ago");
#if FEATURE_SOCKET_MODE_POLL
var mgr = Bridge.Multiplexer.SocketManager;
add("SocketManager-State", "mgr", mgr.State.ToString());
add("Last-Error", "err", mgr.LastErrorTimeRelative());
#endif
}
var ex = innerException == null
......@@ -592,6 +585,12 @@ private static int WriteRaw(Span<byte> span, long value, bool withLengthPrefix =
return WriteCrlf(span, offset);
}
internal void WakeWriterAndCheckForThrottle()
{
var flush = _ioPipe.Output.FlushAsync();
if (!flush.IsCompletedSuccessfully) flush.AsTask().Wait();
}
static readonly byte[] NullBulkString = Encoding.ASCII.GetBytes("$-1\r\n"), EmptyBulkString = Encoding.ASCII.GetBytes("$0\r\n\r\n");
private static void WriteUnified(PipeWriter writer, byte[] value)
{
......@@ -801,15 +800,8 @@ private static void WriteUnified(PipeWriter writer, long value)
var bytes = WriteRaw(span, value, withLengthPrefix: true, offset: 1);
writer.Advance(bytes);
}
private int haveReader;
internal int GetAvailableInboundBytes(out int activeReaders)
{
activeReaders = Interlocked.CompareExchange(ref haveReader, 0, 0);
return socketToken.Available;
}
internal int GetAvailableInboundBytes() => socketToken.Available;
private static LocalCertificateSelectionCallback GetAmbientCertificateCallback()
{
......
StackExchange.Redis/StackExchange/Redis/ServerEndPoint.cs
View file @ 795379da
......@@ -367,15 +367,15 @@ internal ServerCounters GetCounters()
return counters;
}
internal int GetOutstandingCount(RedisCommand command, out int inst, out int qu, out int qs, out int qc, out int wr, out int wq, out int @in, out int ar)
internal int GetOutstandingCount(RedisCommand command, out int inst, out int qs, out int qc, out int @in)
{
var bridge = GetBridge(command, false);
if (bridge == null)
{
return inst = qu = qs = qc = wr = wq = @in = ar = 0;
return inst = qs = qc = @in = 0;
}
return bridge.GetOutstandingCount(out inst, out qu, out qs, out qc, out wr, out wq, out @in, out ar);
return bridge.GetOutstandingCount(out inst, out qs, out qc, out @in);
}
internal string GetProfile()
......
StackExchange.Redis/StackExchange/Redis/SocketManager.cs
View file @ 795379da
using System;
using System.Collections.Generic;
using System.IO;
using System.IO.Pipelines;
using System.Net;
......@@ -13,9 +12,7 @@ namespace StackExchange.Redis
internal enum SocketMode
{
Abort,
[Obsolete("just don't", error: true)]
Poll,
Async
Async,
}
/// <summary>
......@@ -108,11 +105,7 @@ internal enum ManagerState
ProcessReadQueue,
ProcessErrorQueue,
}
private readonly Queue<PhysicalBridge> writeQueue = new Queue<PhysicalBridge>();
private bool isDisposed;
private readonly bool useHighPrioritySocketThreads = true;
/// <summary>
/// Gets the name of this SocketManager instance
/// </summary>
......@@ -133,12 +126,7 @@ public SocketManager(string name, bool useHighPrioritySocketThreads)
{
if (string.IsNullOrWhiteSpace(name)) name = GetType().Name;
Name = name;
this.useHighPrioritySocketThreads = useHighPrioritySocketThreads;
_writeOneQueueAsync = () => WriteOneQueueAsync();
Task.Run(() => WriteAllQueuesAsync());
const int Receive_PauseWriterThreshold = 1024 * 1024 * 1024; // let's give it up to 1GiB of buffer for now
var defaultPipeOptions = PipeOptions.Default;
......@@ -159,9 +147,6 @@ public SocketManager(string name, bool useHighPrioritySocketThreads)
readonly DedicatedThreadPoolPipeScheduler _scheduler;
internal readonly PipeOptions SendPipeOptions, ReceivePipeOptions;
private readonly Func<Task> _writeOneQueueAsync;
private enum CallbackOperation
{
Read,
......@@ -174,12 +159,6 @@ private enum CallbackOperation
public void Dispose()
{
_scheduler?.Dispose();
lock (writeQueue)
{
// make sure writer threads know to exit
isDisposed = true;
Monitor.PulseAll(writeQueue);
}
OnDispose();
}
internal SocketToken BeginConnect(EndPoint endpoint, ISocketCallback callback, ConnectionMultiplexer multiplexer, TextWriter log)
......@@ -247,27 +226,6 @@ void proxyCallback(IAsyncResult ar)
return token;
}
internal void RequestWrite(PhysicalBridge bridge, bool forced)
{
if (Interlocked.CompareExchange(ref bridge.inWriteQueue, 1, 0) == 0 || forced)
{
lock (writeQueue)
{
writeQueue.Enqueue(bridge);
if (writeQueue.Count == 1)
{
Monitor.PulseAll(writeQueue);
}
else if (writeQueue.Count >= 2)
{ // struggling are we? let's have some help dealing with the backlog
Task.Run(_writeOneQueueAsync);
}
}
}
}
internal void Shutdown(SocketToken token)
{
Shutdown(token.Socket);
......@@ -345,88 +303,5 @@ private void Shutdown(Socket socket)
try { socket.Dispose(); } catch { }
}
}
private async Task WriteAllQueuesAsync()
{
while (true)
{
PhysicalBridge bridge;
lock (writeQueue)
{
if (writeQueue.Count == 0)
{
if (isDisposed) break; // <========= exit point
Monitor.Wait(writeQueue);
if (isDisposed) break; // (woken by Dispose)
if (writeQueue.Count == 0) continue; // still nothing...
}
bridge = writeQueue.Dequeue();
}
switch (await bridge.WriteQueueAsync(200))
{
case WriteResult.MoreWork:
case WriteResult.QueueEmptyAfterWrite:
// back of the line!
lock (writeQueue)
{
writeQueue.Enqueue(bridge);
}
break;
case WriteResult.CompetingWriter:
break;
case WriteResult.NoConnection:
Interlocked.Exchange(ref bridge.inWriteQueue, 0);
break;
case WriteResult.NothingToDo:
if (!bridge.ConfirmRemoveFromWriteQueue())
{ // more snuck in; back of the line!
lock (writeQueue)
{
writeQueue.Enqueue(bridge);
}
}
break;
}
}
}
private Task WriteOneQueueAsync()
{
PhysicalBridge bridge;
lock (writeQueue)
{
bridge = writeQueue.Count == 0 ? null : writeQueue.Dequeue();
}
if (bridge == null) return Task.CompletedTask;
return WriteOneQueueAsyncImpl(bridge);
}
private async Task WriteOneQueueAsyncImpl(PhysicalBridge bridge)
{
bool keepGoing;
do
{
switch (await bridge.WriteQueueAsync(-1))
{
case WriteResult.MoreWork:
case WriteResult.QueueEmptyAfterWrite:
keepGoing = true;
break;
case WriteResult.NothingToDo:
keepGoing = !bridge.ConfirmRemoveFromWriteQueue();
break;
case WriteResult.CompetingWriter:
keepGoing = false;
break;
case WriteResult.NoConnection:
Interlocked.Exchange(ref bridge.inWriteQueue, 0);
keepGoing = false;
break;
default:
keepGoing = false;
break;
}
} while (keepGoing);
}
}
}
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