stale connection detection changed to use first unanswered write timestamp...

stale connection detection changed to use first unanswered write timestamp instead of lastWrite - lastRead diff

StackExchange.Redis/StackExchange/Redis/DebuggingAids.cs

@@ -273,11 +273,11 @@ public static bool EmulateStaleConnection
             }
         }
 
-        partial void DebugEmulateStaleConnection(ref int lastRead)
+        partial void DebugEmulateStaleConnection(ref int firstUnansweredWrite)
         {
             if (emulateStaleConnection)
             {
-                lastRead -= 100500;
+                firstUnansweredWrite = Environment.TickCount - 100000;
             }
         }
     }

StackExchange.Redis/StackExchange/Redis/PhysicalConnection.cs

@@ -64,6 +64,7 @@ private static readonly Message
         int ioBufferBytes = 0;
 
         int lastWriteTickCount, lastReadTickCount, lastBeatTickCount;
+        int firstUnansweredWriteTickCount;
 
         private Stream netStream, outStream;
 
@@ -85,6 +86,7 @@ public PhysicalConnection(PhysicalBridge bridge)
 
         public void BeginConnect()
         {
+            Thread.VolatileWrite(ref firstUnansweredWriteTickCount, 0);
             var endpoint = this.bridge.ServerEndPoint.EndPoint;
 
             multiplexer.Trace("Connecting...", physicalName);
@@ -166,10 +168,13 @@ public void RecordConnectionFailed(ConnectionFailureType failureType, Exception
             {
                 int now = Environment.TickCount, lastRead = Thread.VolatileRead(ref lastReadTickCount), lastWrite = Thread.VolatileRead(ref lastWriteTickCount),
                     lastBeat = Thread.VolatileRead(ref lastBeatTickCount);
+                int unansweredRead = Thread.VolatileRead(ref firstUnansweredWriteTickCount);
 
                 string message = failureType + " on " + Format.ToString(bridge.ServerEndPoint.EndPoint) + "/" + connectionType
                     + ", input-buffer: " + ioBufferBytes + ", outstanding: " + GetSentAwaitingResponseCount()
-                    + ", last-read: " + unchecked(now - lastRead) / 1000 + "s ago, last-write: " + unchecked(now - lastWrite) / 1000 + "s ago, keep-alive: " + bridge.ServerEndPoint.WriteEverySeconds + "s, pending: "
+                    + ", last-read: " + unchecked(now - lastRead) / 1000 + "s ago, last-write: " + unchecked(now - lastWrite) / 1000 + "s ago"
+                    + ", unanswered-write: " + unchecked(now - unansweredRead) / 1000 + "s ago"
+                    + ", keep-alive: " + bridge.ServerEndPoint.WriteEverySeconds + "s, pending: "
                     + bridge.GetPendingCount() + ", state: " + oldState + ", last-heartbeat: " + (lastBeat == 0 ? "never" : (unchecked(now - lastBeat) / 1000 + "s ago"))
                     + (bridge.IsBeating ? " (mid-beat)" : "") + ", last-mbeat: " + multiplexer.LastHeartbeatSecondsAgo + "s ago, global: "
                     + ConnectionMultiplexer.LastGlobalHeartbeatSecondsAgo + "s ago";
@@ -381,6 +386,10 @@ internal void WriteHeader(RedisCommand command, int arguments)
                 throw ExceptionFactory.CommandDisabled(multiplexer.IncludeDetailInExceptions, command, null, bridge.ServerEndPoint);
             }
             outStream.WriteByte((byte)'*');
+
+            // remember the time of the first write that still not followed by read
+            Interlocked.CompareExchange(ref firstUnansweredWriteTickCount, Environment.TickCount, 0);
+
             WriteRaw(outStream, arguments + 1);
             WriteUnified(outStream, commandBytes);
         }
@@ -816,6 +825,10 @@ private bool ProcessReadBytes(int bytesRead)
             }
 
             Interlocked.Exchange(ref lastReadTickCount, Environment.TickCount);
+
+            // reset unanswered write timestamp
+            Thread.VolatileWrite(ref firstUnansweredWriteTickCount, 0);
+
             ioBufferBytes += bytesRead;
             multiplexer.Trace("More bytes available: " + bytesRead + " (" + ioBufferBytes + ")", physicalName);
             int offset = 0, count = ioBufferBytes;
@@ -954,17 +967,18 @@ RawResult TryParseResult(byte[] buffer, ref int offset, ref int count)
             }
         }
 
-        partial void DebugEmulateStaleConnection(ref int lastWrite);
+        partial void DebugEmulateStaleConnection(ref int firstUnansweredWrite);
 
         public void CheckForStaleConnection()
         {
-            int lastRead, lastWrite;
-            lastRead = Thread.VolatileRead(ref this.lastReadTickCount);
-            lastWrite = Thread.VolatileRead(ref this.lastWriteTickCount);
+            int firstUnansweredWrite;
+            firstUnansweredWrite = Thread.VolatileRead(ref firstUnansweredWriteTickCount);
+
+            DebugEmulateStaleConnection(ref firstUnansweredWrite);
 
-            DebugEmulateStaleConnection(ref lastRead);
+            int now = Environment.TickCount;
 
-            if ((lastWrite - lastRead) > this.multiplexer.RawConfig.SyncTimeout)
+            if (firstUnansweredWrite != 0 && (now - firstUnansweredWrite) > this.multiplexer.RawConfig.SyncTimeout)
             {
                 this.RecordConnectionFailed(ConnectionFailureType.SocketFailure, origin: "CheckForStaleConnection");
             }