Skip to content

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

Try to make sure that QUIT doesn't get stuck to one side with no chance of... 

Try to make sure that QUIT doesn't get stuck to one side with no chance of completion; do a better job of completing messages that we never send; stamp time of ambient exceptions
parent 5afb8a5d
Hide whitespace changes
Inline Side-by-side
Showing with 62 additions and 40 deletions
NRediSearch.Test/RediSearchTestBase.cs
View file @ 2c0707d0
......@@ -52,7 +52,7 @@ internal static ConnectionMultiplexer GetWithFT(ITestOutputHelper output)
SyncTimeout = 15000,
};
var conn = ConnectionMultiplexer.Connect(options);
conn.MessageFaulted += (msg, ex, origin) => output.WriteLine($"Faulted from '{origin}': '{msg}' - {ex.Message}");
conn.MessageFaulted += (msg, ex, origin) => output.WriteLine($"Faulted from '{origin}': '{msg}' - '{(ex == null ? "(null)" : ex.Message)}'");
var server = conn.GetServer(ep);
var arr = (RedisResult[])server.Execute("module", "list");
bool found = false;
......
StackExchange.Redis.Tests/TestBase.cs
View file @ 2c0707d0
......@@ -83,13 +83,13 @@ static TestBase()
}
};
}
private string Time() => DateTime.UtcNow.ToString("T");
protected void OnConnectionFailed(object sender, ConnectionFailedEventArgs e)
{
Interlocked.Increment(ref privateFailCount);
lock (privateExceptions)
{
privateExceptions.Add($"Connection failed ({e.FailureType}): {EndPointCollection.ToString(e.EndPoint)}/{e.ConnectionType}: {e.Exception}");
privateExceptions.Add($"{Time()}: Connection failed ({e.FailureType}): {EndPointCollection.ToString(e.EndPoint)}/{e.ConnectionType}: {e.Exception}");
}
}
......@@ -98,7 +98,7 @@ protected void OnInternalError(object sender, InternalErrorEventArgs e)
Interlocked.Increment(ref privateFailCount);
lock (privateExceptions)
{
privateExceptions.Add("Internal error: " + e.Origin + ", " + EndPointCollection.ToString(e.EndPoint) + "/" + e.ConnectionType);
privateExceptions.Add(Time() + ": Internal error: " + e.Origin + ", " + EndPointCollection.ToString(e.EndPoint) + "/" + e.ConnectionType);
}
}
......
StackExchange.Redis/StackExchange/Redis/ConnectionMultiplexer.cs
View file @ 2c0707d0
......@@ -128,7 +128,7 @@ internal static string TryGetAzureRoleInstanceIdNoThrow()
internal void OnConnectionFailed(EndPoint endpoint, ConnectionType connectionType, ConnectionFailureType failureType, Exception exception, bool reconfigure)
{
if (isDisposed) return;
if (_isDisposed) return;
var handler = ConnectionFailed;
if (handler != null)
{
......@@ -147,7 +147,7 @@ internal void OnInternalError(Exception exception, EndPoint endpoint = null, Con
try
{
Trace("Internal error: " + origin + ", " + exception == null ? "unknown" : exception.Message);
if (isDisposed) return;
if (_isDisposed) return;
var handler = InternalError;
if (handler != null)
{
......@@ -163,7 +163,7 @@ internal void OnInternalError(Exception exception, EndPoint endpoint = null, Con
internal void OnConnectionRestored(EndPoint endpoint, ConnectionType connectionType)
{
if (isDisposed) return;
if (_isDisposed) return;
var handler = ConnectionRestored;
if (handler != null)
{
......@@ -176,7 +176,7 @@ internal void OnConnectionRestored(EndPoint endpoint, ConnectionType connectionT
private void OnEndpointChanged(EndPoint endpoint, EventHandler<EndPointEventArgs> handler)
{
if (isDisposed) return;
if (_isDisposed) return;
if (handler != null)
{
UnprocessableCompletionManager.CompleteSyncOrAsync(
......@@ -194,7 +194,7 @@ private void OnEndpointChanged(EndPoint endpoint, EventHandler<EndPointEventArgs
public event EventHandler<RedisErrorEventArgs> ErrorMessage;
internal void OnErrorMessage(EndPoint endpoint, string message)
{
if (isDisposed) return;
if (_isDisposed) return;
var handler = ErrorMessage;
if (handler != null)
{
......@@ -755,8 +755,8 @@ internal ServerEndPoint AnyConnected(ServerType serverType, uint startOffset, Re
return fallback;
}
private volatile bool isDisposed;
internal bool IsDisposed => isDisposed;
private volatile bool _isDisposed;
internal bool IsDisposed => _isDisposed;
/// <summary>
/// Create a new ConnectionMultiplexer instance
......@@ -944,7 +944,7 @@ internal ServerEndPoint GetServerEndPoint(EndPoint endpoint, TextWriter log = nu
server = (ServerEndPoint)servers[endpoint];
if (server == null)
{
if (isDisposed) throw new ObjectDisposedException(ToString());
if (_isDisposed) throw new ObjectDisposedException(ToString());
server = new ServerEndPoint(this, endpoint);
servers.Add(endpoint, server);
......@@ -1314,7 +1314,7 @@ private void ActivateAllServers(TextWriter log)
}
internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, TextWriter log, EndPoint blame, string cause, bool publishReconfigure = false, CommandFlags publishReconfigureFlags = CommandFlags.None)
{
if (isDisposed) throw new ObjectDisposedException(ToString());
if (_isDisposed) throw new ObjectDisposedException(ToString());
bool showStats = true;
if (log == null)
......@@ -1949,7 +1949,7 @@ public bool IsConnecting
/// <param name="allowCommandsToComplete">Whether to allow all in-queue commands to complete first.</param>
public void Close(bool allowCommandsToComplete = true)
{
isDisposed = true;
_isDisposed = true;
_profilingSessionProvider = null;
using (var tmp = pulse)
{
......@@ -2004,7 +2004,7 @@ private Task[] QuitAllServers()
/// <param name="allowCommandsToComplete">Whether to allow all in-queue commands to complete first.</param>
public async Task CloseAsync(bool allowCommandsToComplete = true)
{
isDisposed = true;
_isDisposed = true;
using (var tmp = pulse)
{
pulse = null;
......@@ -2025,12 +2025,12 @@ public async Task CloseAsync(bool allowCommandsToComplete = true)
public void Dispose()
{
GC.SuppressFinalize(this);
Close(!isDisposed);
Close(!_isDisposed);
}
internal Task<T> ExecuteAsyncImpl<T>(Message message, ResultProcessor<T> processor, object state, ServerEndPoint server)
{
if (isDisposed) throw new ObjectDisposedException(ToString());
if (_isDisposed) throw new ObjectDisposedException(ToString());
if (message == null)
{
......@@ -2099,7 +2099,7 @@ internal static void ThrowFailed<T>(TaskCompletionSource<T> source, Exception un
internal T ExecuteSyncImpl<T>(Message message, ResultProcessor<T> processor, ServerEndPoint server)
{
if (isDisposed) throw new ObjectDisposedException(ToString());
if (_isDisposed) throw new ObjectDisposedException(ToString());
if (message == null) // fire-and forget could involve a no-op, represented by null - for example Increment by 0
{
......
StackExchange.Redis/StackExchange/Redis/PhysicalBridge.cs
View file @ 2c0707d0
......@@ -126,7 +126,7 @@ public WriteResult TryWrite(Message message, bool isSlave)
if (isDisposed) throw new ObjectDisposedException(Name);
if (!IsConnected)
{
if (message.IsInternalCall)
if (message.IsInternalCall && message.Command != RedisCommand.QUIT)
{
// you can go in the queue, but we won't be starting
// a worker, because the handshake has not completed
......@@ -141,12 +141,21 @@ public WriteResult TryWrite(Message message, bool isSlave)
else
{
// sorry, we're just not ready for you yet;
message.Cancel();
Multiplexer?.OnMessageFaulted(message, null);
this.CompleteSyncOrAsync(message);
return WriteResult.NoConnectionAvailable;
}
}
var physical = this.physical;
if (physical == null) return WriteResult.NoConnectionAvailable;
if (physical == null)
{
message.Cancel();
Multiplexer?.OnMessageFaulted(message, null);
this.CompleteSyncOrAsync(message);
return WriteResult.NoConnectionAvailable;
}
var result = WriteMessageTakingWriteLock(physical, message);
LogNonPreferred(message.Flags, isSlave);
......@@ -519,23 +528,29 @@ internal bool TryEnqueue(List<Message> messages, bool isSlave)
/// This writes a message to the output stream
/// </summary>
/// <param name="physical">The phsyical connection to write to.</param>
/// <param name="next">The message to be written.</param>
internal WriteResult WriteMessageTakingWriteLock(PhysicalConnection physical, Message next)
/// <param name="message">The message to be written.</param>
internal WriteResult WriteMessageTakingWriteLock(PhysicalConnection physical, Message message)
{
Trace("Writing: " + next);
next.SetEnqueued();
Trace("Writing: " + message);
message.SetEnqueued();
WriteResult result;
bool haveLock = false;
Monitor.TryEnter(WriteLock, TimeoutMilliseconds, ref haveLock);
if (!haveLock) return WriteResult.TimeoutBeforeWrite;
if (!haveLock)
{
message.Cancel();
Multiplexer?.OnMessageFaulted(message, null);
this.CompleteSyncOrAsync(message);
return WriteResult.TimeoutBeforeWrite;
}
try
{
var messageIsSent = false;
if (next is IMultiMessage)
if (message is IMultiMessage)
{
SelectDatabaseInsideWriteLock(physical, next); // need to switch database *before* the transaction
foreach (var subCommand in ((IMultiMessage)next).GetMessages(physical))
SelectDatabaseInsideWriteLock(physical, message); // need to switch database *before* the transaction
foreach (var subCommand in ((IMultiMessage)message).GetMessages(physical))
{
result = WriteMessageToServerInsideWriteLock(physical, subCommand);
if (result != WriteResult.Success)
......@@ -543,24 +558,24 @@ internal WriteResult WriteMessageTakingWriteLock(PhysicalConnection physical, Me
// we screwed up; abort; note that WriteMessageToServer already
// killed the underlying connection
Trace("Unable to write to server");
next.Fail(ConnectionFailureType.ProtocolFailure, null, "failure before write: " + result.ToString());
this.CompleteSyncOrAsync(next);
message.Fail(ConnectionFailureType.ProtocolFailure, null, "failure before write: " + result.ToString());
this.CompleteSyncOrAsync(message);
return result;
}
//The parent message (next) may be returned from GetMessages
//and should not be marked as sent again below
messageIsSent = messageIsSent || subCommand == next;
messageIsSent = messageIsSent || subCommand == message;
}
if (!messageIsSent)
{
next.SetRequestSent(); // well, it was attempted, at least...
message.SetRequestSent(); // well, it was attempted, at least...
}
result = WriteResult.Success;
}
else
{
result = WriteMessageToServerInsideWriteLock(physical, next);
result = WriteMessageToServerInsideWriteLock(physical, message);
}
result = physical.WakeWriterAndCheckForThrottle();
}
......
StackExchange.Redis/StackExchange/Redis/RedisSubscriber.cs
View file @ 2c0707d0
......@@ -259,10 +259,10 @@ internal bool Validate(ConnectionMultiplexer multiplexer, RedisChannel channel)
#endif
[Conditional("TEST")]
internal void OnMessageFaulted(Message msg, Exception fault, [CallerMemberName] string origin = null)
internal void OnMessageFaulted(Message msg, Exception fault, [CallerMemberName] string origin = default, [CallerFilePath] string path = default, [CallerLineNumber] int lineNumber = default)
{
#if TEST
MessageFaulted?.Invoke(msg?.CommandAndKey, fault, origin);
MessageFaulted?.Invoke(msg?.CommandAndKey, fault, $"{origin} ({path}#{lineNumber})");
#endif
}
}
......
StackExchange.Redis/StackExchange/Redis/ServerEndPoint.cs
View file @ 2c0707d0
......@@ -322,14 +322,21 @@ internal void AutoConfigure(PhysicalConnection connection)
internal Task Close(ConnectionType connectionType)
{
var tmp = GetBridge(connectionType, create: false);
if (tmp == null || !tmp.IsConnected || !Multiplexer.CommandMap.IsAvailable(RedisCommand.QUIT))
try
{
return CompletedTask<bool>.Default(null);
var tmp = GetBridge(connectionType, create: false);
if (tmp == null || !tmp.IsConnected || !Multiplexer.CommandMap.IsAvailable(RedisCommand.QUIT))
{
return Task.CompletedTask;
}
else
{
return WriteDirectAsync(Message.Create(-1, CommandFlags.None, RedisCommand.QUIT), ResultProcessor.DemandOK, bridge: tmp);
}
}
else
catch (Exception ex)
{
return WriteDirectAsync(Message.Create(-1, CommandFlags.None, RedisCommand.QUIT), ResultProcessor.DemandOK, bridge: tmp);
return Task.FromException(ex);
}
}
......
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