kill the unsent queue when connection fail; this is far more reasonable...

kill the unsent queue when connection fail; this is far more reasonable (expected), and ensures async works correctly

StackExchange.Redis.Tests/AsyncTests.cs

+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using NUnit.Framework;
+
+namespace StackExchange.Redis.Tests
+{
+    [TestFixture]
+    public class AsyncTests : TestBase
+    {
+        protected override string GetConfiguration()
+        {
+            return PrimaryServer + ":" + PrimaryPortString;
+        }
+
+#if DEBUG // IRedisServerDebug and AllowConnect are only available if DEBUG is defined
+        [Test]
+        public void AsyncTasksReportFailureIfServerUnavailable()
+        {
+            SetExpectedAmbientFailureCount(-1); // this will get messy
+
+            using(var conn = Create(allowAdmin: true))
+            {
+                var server = (IRedisServerDebug)conn.GetServer(PrimaryServer, PrimaryPort);
+
+                RedisKey key = Me();
+                var db = conn.GetDatabase();
+                db.KeyDelete(key);
+                var a = db.SetAddAsync(key, "a");
+                var b = db.SetAddAsync(key, "b");
+
+                Assert.AreEqual(true, conn.Wait(a));
+                Assert.AreEqual(true, conn.Wait(b));
+
+                conn.AllowConnect = false;
+                server.SimulateConnectionFailure();
+                var c = db.SetAddAsync(key, "c");
+
+                Assert.IsTrue(c.IsFaulted, "faulted");
+                var ex = c.Exception.InnerExceptions.Single();
+                Assert.IsInstanceOf<RedisConnectionException>(ex);
+                Assert.AreEqual("No connection is available to service this operation: SADD", ex.Message);
+            }
+        }
+#endif
+    }
+}

StackExchange.Redis.Tests/StackExchange.Redis.Tests.csproj

@@ -60,6 +60,7 @@
     <Reference Include="Microsoft.CSharp" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="AsyncTests.cs" />
     <Compile Include="BasicOps.cs" />
     <Compile Include="Cluster.cs" />
     <Compile Include="Commands.cs" />

StackExchange.Redis/StackExchange/Redis/ConnectionCounters.cs

@@ -68,6 +68,17 @@ internal void Add(ConnectionCounters other)
             this.NonPreferredEndpointCount += other.NonPreferredEndpointCount;
         }
 
+        internal bool Any()
+        {
+            return CompletedAsynchronously != 0 || CompletedSynchronously != 0
+                || FailedAsynchronously != 0 || OperationCount != 0
+                || PendingUnsentItems != 0 || ResponsesAwaitingAsyncCompletion != 0
+                || SentItemsAwaitingResponse != 0 || SocketCount != 0
+                || Subscriptions != 0 || WriterCount != 0
+                || NonPreferredEndpointCount != 0;
+        }
+
+
         /// <summary>
         /// Operations that have been requested, but which have not yet been sent to the server
         /// </summary>

StackExchange.Redis/StackExchange/Redis/ConnectionMultiplexer.cs

@@ -59,6 +59,7 @@ public ServerCounters GetCounters()
             {
                 counters.Add(snapshot[i].GetCounters());
             }
+            unprocessableCompletionManager.GetCounters(counters.Other);
             return counters;
         }
 
@@ -1098,7 +1099,7 @@ internal async Task<bool> ReconfigureAsync(bool first, bool reconfigureAll, Text
                 {
                     Trace("Testing: " + Format.ToString(endpoints[i]));
                     var server = GetServerEndPoint(endpoints[i]);
-                    server.ReportNextFailure();
+                    //server.ReportNextFailure();
                     servers[i] = server;
                     if (reconfigureAll && server.IsConnected)
                     {

StackExchange.Redis/StackExchange/Redis/Message.cs

@@ -59,6 +59,8 @@ public sealed class RedisServerException : RedisException
     
     abstract class Message : ICompletable
     {
+
+        public static readonly Message[] EmptyArray = new Message[0];
         public readonly int Db;
 
         protected RedisCommand command;

StackExchange.Redis/StackExchange/Redis/MessageQueue.cs

@@ -26,6 +26,22 @@ public Message Dequeue()
             return null;
         }
 
+        internal Message[] DequeueAll()
+        {
+            lock (regular)
+            {
+                int count = high.Count + regular.Count;
+                if (count == 0) return Message.EmptyArray;
+
+                var arr = new Message[count];
+                high.CopyTo(arr, 0);
+                regular.CopyTo(arr, high.Count);
+                high.Clear();
+                regular.Clear();
+                return arr;
+            }
+        }
+
         public object SyncLock {  get {  return regular; } }
         public Message PeekPing(out int queueLength)
         {

StackExchange.Redis/StackExchange/Redis/PhysicalBridge.cs

@@ -384,10 +384,18 @@ internal void OnHeartbeat(bool ifConnectedOnly)
                     case (int)State.Disconnected:
                         if (!ifConnectedOnly)
                         {
+                            AbortUnsent();
                             multiplexer.Trace("Resurrecting " + this.ToString());
                             GetConnection();
                         }
                         break;
+                    case (int)State.Connecting:
+                    default:
+                        if (!ifConnectedOnly)
+                        {
+                            AbortUnsent();
+                        }
+                        break;
                 }
             }
             catch (Exception ex)
@@ -469,7 +477,7 @@ internal bool ConfirmRemoveFromWriteQueue()
         /// connected" handshake (when there is no dequeue loop) - otherwise,
         /// you can pretty much assume you're going to destroy the stream
         /// </summary>
-        internal void WriteMessageDirect(PhysicalConnection tmp, Message next)
+        internal bool WriteMessageDirect(PhysicalConnection tmp, Message next)
         {
             Trace("Writing: " + next);
             if (next is IMultiMessage)
@@ -484,13 +492,14 @@ internal void WriteMessageDirect(PhysicalConnection tmp, Message next)
                         Trace("Unable to write to server");
                         next.Fail(ConnectionFailureType.ProtocolFailure, null);
                         CompleteSyncOrAsync(next);
-                        break;
+                        return false;
                     }
                 }
+                return true;
             }
             else
             {
-                WriteMessageToServer(tmp, next);
+                return WriteMessageToServer(tmp, next);
             }
         }
 
@@ -500,10 +509,27 @@ private State ChangeState(State newState)
             if (oldState != newState)
             {
                 multiplexer.Trace(connectionType + " state changed from " + oldState + " to " + newState);
+
+                if (newState == State.Disconnected)
+                {
+                    AbortUnsent();
+                }
             }
             return oldState;
         }
 
+        private void AbortUnsent()
+        {
+            var dead = queue.DequeueAll();
+            Trace(dead.Length != 0, "Aborting " + dead.Length + " messages");
+            for (int i = 0; i < dead.Length; i++)
+            {
+                var msg = dead[i];
+                msg.Fail(ConnectionFailureType.UnableToResolvePhysicalConnection, null);
+                CompleteSyncOrAsync(msg);
+            }
+        }
+
         private bool ChangeState(State oldState, State newState)
         {
             bool result = Interlocked.CompareExchange(ref state, (int)newState, (int)oldState) == (int)oldState;
@@ -659,7 +685,7 @@ private bool WriteMessageToServer(PhysicalConnection connection, Message message
                 message.Fail(ConnectionFailureType.InternalFailure, ex);
                 CompleteSyncOrAsync(message);
 
-                // we're not sure *what* happened here; kill the connection
+                // we're not sure *what* happened here; probably an IOException; kill the connection
                 if(connection != null) connection.RecordConnectionFailed(ConnectionFailureType.InternalFailure, ex);
                 return false;
             }
@@ -683,6 +709,7 @@ internal WriteResult WriteQueue(int maxWork)
                 var conn = GetConnection();
                 if(conn == null)
                 {
+                    AbortUnsent();
                     Trace("Connection not available; exiting");
                     return WriteResult.NoConnection;
                 }
@@ -702,7 +729,12 @@ internal WriteResult WriteQueue(int maxWork)
                     last = next;
                     
                     Trace("Now pending: " + GetPendingCount());
-                    WriteMessageDirect(conn, next);
+                    if(!WriteMessageDirect(conn, next))
+                    {
+                        AbortUnsent();
+                        Trace("write failed; connection is toast; exiting");
+                        return WriteResult.NoConnection;
+                    }
                     count++;
                     if (maxWork > 0 && count >= maxWork)
                     {

StackExchange.Redis/StackExchange/Redis/PhysicalConnection.cs

@@ -160,8 +160,8 @@ public void RecordConnectionFailed(ConnectionFailureType failureType, Exception
             
             if (isCurrent && Interlocked.CompareExchange(ref failureReported, 1, 0) == 0)
             {
-                try
-                {
+                //try
+                //{
                     long now = Environment.TickCount, lastRead = Interlocked.Read(ref lastReadTickCount), lastWrite = Interlocked.Read(ref lastWriteTickCount),
                         lastBeat = Interlocked.Read(ref lastBeatTickCount);
 
@@ -175,12 +175,14 @@ public void RecordConnectionFailed(ConnectionFailureType failureType, Exception
                     var ex = innerException == null
                         ? new RedisConnectionException(failureType, message)
                         : new RedisConnectionException(failureType, message, innerException);
-                    throw ex;
-                }
-                catch (Exception caught)
-                {
-                    bridge.OnConnectionFailed(this, failureType, caught);
-                }
+
+                    bridge.OnConnectionFailed(this, failureType, ex);
+                //    throw ex;
+                //}
+                //catch (Exception caught)
+                //{
+                //    bridge.OnConnectionFailed(this, failureType, caught);
+                //}
 
             }
 

StackExchange.Redis/StackExchange/Redis/RedisBatch.cs

@@ -96,9 +96,12 @@ internal override T ExecuteSync<T>(Message message, ResultProcessor<T> processor
         private void FailNoServer(List<Message> messages)
         {
             if (messages == null) return;
+            var completion = multiplexer.UnprocessableCompletionManager;
             foreach(var msg in messages)
             {
                 msg.Fail(ConnectionFailureType.UnableToResolvePhysicalConnection, null);
+                completion.CompleteSyncOrAsync(msg);
+
             }
         }
     }

StackExchange.Redis/StackExchange/Redis/ResultBox.cs

@@ -11,14 +11,15 @@ abstract partial class ResultBox
 
         public void SetException(Exception exception)
         {
-            try
-            {
-                throw exception;
-            }
-            catch (Exception caught)
-            { // stacktrace etc
-                this.exception = caught;
-            }
+            this.exception = exception;
+            //try
+            //{
+            //    throw exception;
+            //}
+            //catch (Exception caught)
+            //{ // stacktrace etc
+            //    this.exception = caught;
+            //}
         }
         public abstract bool TryComplete(bool isAsync);
 

StackExchange.Redis/StackExchange/Redis/ServerCounters.cs

@@ -14,6 +14,7 @@ internal ServerCounters(EndPoint endpoint)
             this.EndPoint = endpoint;
             this.Interactive = new ConnectionCounters(ConnectionType.Interactive);
             this.Subscription = new ConnectionCounters(ConnectionType.Subscription);
+            this.Other = new ConnectionCounters(ConnectionType.None);
         }
 
         /// <summary>
@@ -29,14 +30,18 @@ internal ServerCounters(EndPoint endpoint)
         /// <summary>
         /// Counters associated with the subscription (pub-sub) connection
         /// </summary>
-
         public ConnectionCounters Subscription { get; private set; }
 
+        /// <summary>
+        /// Counters associated with other ambient activity
+        /// </summary>
+        public ConnectionCounters Other { get; private set; }
+
 
         /// <summary>
         /// Indicates the total number of outstanding items against this server
         /// </summary>
-        public long TotalOutstanding { get { return Interactive.TotalOutstanding + Subscription.TotalOutstanding; } }
+        public long TotalOutstanding { get { return Interactive.TotalOutstanding + Subscription.TotalOutstanding + Other.TotalOutstanding; } }
 
         /// <summary>
         /// See Object.ToString();
@@ -48,6 +53,11 @@ public override string ToString()
             Interactive.Append(sb);
             sb.Append("; sub ");
             Subscription.Append(sb);
+            if (Other.Any())
+            {
+                sb.Append("; other ");
+                Other.Append(sb);
+            }
             return sb.ToString();
         }