lib update to v88 (the Kestrel edition); fix MovedProfiling test

StackExchange.Redis.Tests/Cluster.cs

@@ -740,7 +740,7 @@ public void MovedProfiling()
                     Assert.True(msg.CreationToEnqueued > TimeSpan.Zero);
                     Assert.True(msg.EnqueuedToSending > TimeSpan.Zero);
                     Assert.True(msg.SentToResponse > TimeSpan.Zero);
-                    Assert.True(msg.ResponseToCompletion > TimeSpan.Zero);
+                    Assert.True(msg.ResponseToCompletion >= TimeSpan.Zero); // this can be immeasurably fast
                     Assert.True(msg.ElapsedTime > TimeSpan.Zero);
 
                     if (msg.RetransmissionOf != null)

StackExchange.Redis.Tests/Sockets.cs

@@ -23,7 +23,10 @@ public void CheckForSocketLeaks()
             // Force GC before memory dump in debug below...
             CollectGarbage();
 
-            Debugger.Break();
+            if (Debugger.IsAttached)
+            {
+                Debugger.Break();
+            }
         }
     }
 }

StackExchange.Redis/PhysicalConnection.cs

@@ -90,83 +90,80 @@ internal async void BeginConnectAsync(TextWriter log)
             CancellationTokenSource timeoutSource = null;
             try
             {
-                var awaitable = new SocketAwaitable();
-
-                using (var _socketArgs = new SocketAsyncEventArgs
+                using (var args = new SocketAwaitableEventArgs
                 {
-                    UserToken = awaitable,
                     RemoteEndPoint = endpoint,
                 })
                 {
-                    _socketArgs.Completed += SocketAwaitable.Callback;
-
                     var x = VolatileSocket;
                     if (x == null)
                     {
-                        awaitable.TryComplete(0, SocketError.ConnectionAborted);
+                        args.Abort();
                     }
-                    else if (x.ConnectAsync(_socketArgs))
+                    else if (x.ConnectAsync(args))
                     {   // asynchronous operation is pending
-                        timeoutSource = ConfigureTimeout(_socketArgs, bridge.Multiplexer.RawConfig.ConnectTimeout);
+                        timeoutSource = ConfigureTimeout(args, bridge.Multiplexer.RawConfig.ConnectTimeout);
                     }
                     else
                     {   // completed synchronously
-                        SocketAwaitable.OnCompleted(_socketArgs);
+                        args.Complete();
                     }
-                }
-                // Complete connection
-                try
-                {
-                    // If we're told to ignore connect, abort here
-                    if (BridgeCouldBeNull?.Multiplexer?.IgnoreConnect ?? false) return;
 
-                    await awaitable; // wait for the connect to complete or fail (will throw)
-                    if (timeoutSource != null)
-                    {
-                        timeoutSource.Cancel();
-                        timeoutSource.Dispose();
-                    }
-                    var x = VolatileSocket;
-                    if (x == null)
-                    {
-                        ConnectionMultiplexer.TraceWithoutContext("Socket was already aborted");
-                    }
-                    else if (await ConnectedAsync(x, log, bridge.Multiplexer.SocketManager).ForAwait())
+                    // Complete connection
+                    try
                     {
-                        bridge.Multiplexer.LogLocked(log, "Starting read");
-                        try
+                        // If we're told to ignore connect, abort here
+                        if (BridgeCouldBeNull?.Multiplexer?.IgnoreConnect ?? false) return;
+
+                        await args; // wait for the connect to complete or fail (will throw)
+                        if (timeoutSource != null)
+                        {
+                            timeoutSource.Cancel();
+                            timeoutSource.Dispose();
+                        }
+
+                        x = VolatileSocket;
+                        if (x == null)
+                        {
+                            ConnectionMultiplexer.TraceWithoutContext("Socket was already aborted");
+                        }
+                        else if (await ConnectedAsync(x, log, bridge.Multiplexer.SocketManager).ForAwait())
                         {
-                            StartReading();
-                            // Normal return
+                            bridge.Multiplexer.LogLocked(log, "Starting read");
+                            try
+                            {
+                                StartReading();
+                                // Normal return
+                            }
+                            catch (Exception ex)
+                            {
+                                ConnectionMultiplexer.TraceWithoutContext(ex.Message);
+                                Shutdown();
+                            }
                         }
-                        catch (Exception ex)
+                        else
                         {
-                            ConnectionMultiplexer.TraceWithoutContext(ex.Message);
+                            ConnectionMultiplexer.TraceWithoutContext("Aborting socket");
                             Shutdown();
                         }
                     }
-                    else
-                    {
-                        ConnectionMultiplexer.TraceWithoutContext("Aborting socket");
-                        Shutdown();
-                    }
-                }
-                catch (ObjectDisposedException)
-                {
-                    bridge.Multiplexer.LogLocked(log, "(socket shutdown)");
-                    try { RecordConnectionFailed(ConnectionFailureType.UnableToConnect, isInitialConnect: true); }
-                    catch (Exception inner)
+                    catch (ObjectDisposedException)
                     {
-                        ConnectionMultiplexer.TraceWithoutContext(inner.Message);
+                        bridge.Multiplexer.LogLocked(log, "(socket shutdown)");
+                        try { RecordConnectionFailed(ConnectionFailureType.UnableToConnect, isInitialConnect: true); }
+                        catch (Exception inner)
+                        {
+                            ConnectionMultiplexer.TraceWithoutContext(inner.Message);
+                        }
                     }
-                }
-                catch (Exception outer)
-                {
-                    ConnectionMultiplexer.TraceWithoutContext(outer.Message);
-                    try { RecordConnectionFailed(ConnectionFailureType.UnableToConnect, isInitialConnect: true); }
-                    catch (Exception inner)
+                    catch (Exception outer)
                     {
-                        ConnectionMultiplexer.TraceWithoutContext(inner.Message);
+                        ConnectionMultiplexer.TraceWithoutContext(outer.Message);
+                        try { RecordConnectionFailed(ConnectionFailureType.UnableToConnect, isInitialConnect: true); }
+                        catch (Exception inner)
+                        {
+                            ConnectionMultiplexer.TraceWithoutContext(inner.Message);
+                        }
                     }
                 }
             }
@@ -184,7 +181,7 @@ internal async void BeginConnectAsync(TextWriter log)
             }
         }
 
-        private static CancellationTokenSource ConfigureTimeout(SocketAsyncEventArgs args, int timeoutMilliseconds)
+        private static CancellationTokenSource ConfigureTimeout(SocketAwaitableEventArgs args, int timeoutMilliseconds)
         {
             var cts = new CancellationTokenSource();
             var timeout = Task.Delay(timeoutMilliseconds, cts.Token);
@@ -192,11 +189,9 @@ private static CancellationTokenSource ConfigureTimeout(SocketAsyncEventArgs arg
             {
                 try
                 {
-                    var a = (SocketAsyncEventArgs)state;
-                    if (((SocketAwaitable)a.UserToken).TryComplete(0, SocketError.TimedOut))
-                    {
-                        Socket.CancelConnectAsync(a);
-                    }
+                    var a = (SocketAwaitableEventArgs)state;
+                    a.Abort(SocketError.TimedOut);
+                    Socket.CancelConnectAsync(a);
                 }
                 catch { }
             }, args);

StackExchange.Redis/RedisValue.cs

@@ -32,8 +32,10 @@ internal RedisValue(object obj, long val)
             _objectOrSentinel = obj;
             _memory = default;
         }
+#pragma warning disable RCS1085 // Use auto-implemented property.
         internal object DirectObject => _objectOrSentinel;
         internal long DirectInt64 => _overlappedValue64;
+#pragma warning restore RCS1085 // Use auto-implemented property.
 
         private readonly static object Sentinel_Integer = new object();
         private readonly static object Sentinel_Raw = new object();

StackExchange.Redis/StackExchange.Redis.csproj

@@ -19,7 +19,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Pipelines.Sockets.Unofficial" Version="0.2.1-alpha.87" />
+    <PackageReference Include="Pipelines.Sockets.Unofficial" Version="0.2.1-alpha.88" />
     <PackageReference Include="System.Diagnostics.PerformanceCounter" Version="4.5.0" />
     <PackageReference Include="System.IO.Pipelines" Version="4.5.0" />
     <PackageReference Include="System.Threading.Channels" Version="4.5.0" />

TestConsole/Program.cs

 using System;
-using System.Collections.Generic;
 using System.Diagnostics;
 using System.Runtime.CompilerServices;
 using System.Runtime.InteropServices;
@@ -22,6 +21,7 @@ private static void Main()
                 Console.WriteLine(i++);
                 RunCompetingBatchesOnSameMuxer();
             } while (DateTime.UtcNow < stop);
+            Console.WriteLine($"Completed {i} iterations, {2 * i * IterationCount * InnerCount} operations");
         }
         private static ConnectionMultiplexer Create()
         {