Pipelines race (#857)

* really nasty commit - incomplete investigating thread-race

* fixed race - was incomplete messages (packet fragmentation)

BasicTest/Program.cs

@@ -10,32 +10,47 @@ public static async Task Main()
         {
             using (var conn = await ConnectionMultiplexer.ConnectAsync("127.0.0.1:6379,syncTimeout=2000"))
             {
-                conn.ConnectionFailed += (sender, e) => Console.WriteLine($"{e.ConnectionType}, {e.FailureType}: {e.Exception.Message}");
-                var db = conn.GetDatabase(3);
+                int expected = 0;
+
+                try
+                {
 
-                var batch = db.CreateBatch();
-                var del = batch.KeyDeleteAsync("abc");
-                var set = batch.StringSetAsync("abc", "Does SE.Redis work on System.IO.Pipelines?");
-                var s = batch.StringGetAsync("abc");
-                batch.Execute();
+                    conn.ConnectionFailed += (sender, e) => Console.WriteLine($"{e.ConnectionType}, {e.FailureType}: {e.Exception.Message}");
+                    var db = conn.GetDatabase(3);
 
-                await del;
-                await set;
-                Console.WriteLine(await s);
+                    var batch = db.CreateBatch();
+                    var del = batch.KeyDeleteAsync("abc");
+                    var set = batch.StringSetAsync("abc", "Does SE.Redis work on System.IO.Pipelines?");
+                    var s = batch.StringGetAsync("abc");
+                    batch.Execute();
 
-                var rand = new Random(12345);
-                RedisKey counter = "counter";
-                db.KeyDelete(counter, CommandFlags.FireAndForget);
-                int expected = 0;
-                for (int i = 0; i < 1000; i++)
+                    await del;
+                    await set;
+                    Console.WriteLine(await s);
+
+                    var rand = new Random(12345);
+                    RedisKey counter = "counter";
+                    db.KeyDelete(counter, CommandFlags.FireAndForget);
+                    for (int i = 0; i < 1000; i++)
+                    {
+                        int x = rand.Next(50);
+                        //Console.WriteLine($"{i}:{x}");
+                        expected += x;
+                        db.StringIncrement(counter, x); //, CommandFlags.FireAndForget);
+                    }
+                    int actual = (int)await db.StringGetAsync(counter);
+                    Console.WriteLine($"{expected} vs {actual}");
+                }
+                catch (Exception ex)
+                {
+                    Console.WriteLine($"expected when fail: {expected}");
+                    Console.WriteLine(ex.Message);
+                }
+                finally
                 {
-                    int x = rand.Next(50);
-                    Console.WriteLine($"{i}:{x}");
-                    expected += x;
-                    db.StringIncrement(counter, x); //, CommandFlags.FireAndForget);
+                    Console.WriteLine("Press any key");
+                    Console.ReadKey();
                 }
-                int actual = (int)await db.StringGetAsync(counter);
-                Console.WriteLine($"{expected} vs {actual}");
             }
         }
     }

StackExchange.Redis/StackExchange/Redis/PhysicalConnection.cs

@@ -982,17 +982,22 @@ void ISocketCallback.OnHeartbeat()
             {
                 while (true)
                 {
-                    var input = _ioPipe.Input;
+                    var input = _ioPipe?.Input;
+                    if (input == null) break;
+
                     var readResult = await input.ReadAsync();
                     if (readResult.IsCompleted && readResult.Buffer.IsEmpty)
                     {
                         break; // we're all done
                     }
                     var buffer = readResult.Buffer;
-
-                    int handled = ProcessBuffer(in buffer, out var consumed);
+    
+                    var s = new RawResult(ResultType.BulkString, buffer, false).GetString().Replace("\r","\\r").Replace("\n","\\n");
+                    
+                    int handled = ProcessBuffer(ref buffer); // updates buffer.Start
+                    
                     Multiplexer.Trace($"Processed {handled} messages", physicalName);
-                    input.AdvanceTo(buffer.GetPosition(consumed), buffer.End);
+                    input.AdvanceTo(buffer.Start, buffer.End);
                 }
                 Multiplexer.Trace("EOF", physicalName);
                 RecordConnectionFailed(ConnectionFailureType.SocketClosed);
@@ -1003,27 +1008,28 @@ void ISocketCallback.OnHeartbeat()
                 RecordConnectionFailed(ConnectionFailureType.InternalFailure, ex);
             }
         }
-        private int ProcessBuffer(in ReadOnlySequence<byte> entireBuffer, out long consumed)
+
+        private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
         {
             int messageCount = 0;
-            var remainingBuffer = entireBuffer; // create a snapshot so we can trim it after each decoded message
-                                                // (so that slicing later doesn't require us to keep skipping segments)
-
-            consumed = 0;
-            while (!remainingBuffer.IsEmpty)
+            
+            while (!buffer.IsEmpty)
             {
-                var reader = new BufferReader(remainingBuffer);
-                var result = TryParseResult(in remainingBuffer, ref reader);
+                var reader = new BufferReader(buffer);
+                var result = TryParseResult(in buffer, ref reader);
 
                 if (result.HasValue)
                 {
-                    consumed += reader.TotalConsumed;
-                    remainingBuffer = remainingBuffer.Slice(reader.TotalConsumed);
+                    buffer = buffer.Slice(reader.TotalConsumed);
 
                     messageCount++;
                     Multiplexer.Trace(result.ToString(), physicalName);
                     MatchResult(result);
                 }
+                else
+                {
+                    break; // remaining buffer isn't enough; give up
+                }
             }
             return messageCount;
         }