React to scheduler changes (#1846)

src/Microsoft.AspNetCore.Server.Kestrel.Core/Internal/Infrastructure/InlineLoggingThreadPool.cs

@@ -33,9 +33,16 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure
             action(state);
         }
 
-        public void Schedule(Action action)
+        public void Schedule(Action<object> action, object state)
         {
-            Run(action);
+            try
+            {
+                action(state);
+            }
+            catch (Exception e)
+            {
+                _log.LogError(0, e, "InlineLoggingThreadPool.Schedule");
+            }
         }
     }
 }

src/Microsoft.AspNetCore.Server.Kestrel.Core/Internal/Infrastructure/LoggingThreadPool.cs

@@ -18,7 +18,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure
             _log = log;
 
             // Curry and capture log in closures once
-            // The currying is done in functions of the same name to improve the 
+            // The currying is done in functions of the same name to improve the
             // call stack for exceptions and profiling else it shows up as LoggingThreadPool.ctor>b__4_0
             // and you aren't sure which of the 3 functions was called.
             RunAction();
@@ -50,9 +50,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure
             ThreadPool.QueueUserWorkItem(action, state);
         }
 
-        public void Schedule(Action action)
+        public void Schedule(Action<object> action, object state)
         {
-            Run(action);
+            Run(() => action(state));
         }
     }
 }

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv/Internal/LibuvThread.cs

@@ -383,9 +383,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             return await Task.WhenAny(task, Task.Delay(timeout)).ConfigureAwait(false) == task;
         }
 
-        public void Schedule(Action action)
+        public void Schedule(Action<object> action, object state)
         {
-            Post(state => state(), action);
+            Post(action, state);
         }
 
         private struct Work

test/Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests/LibuvConnectionTests.cs

@@ -114,9 +114,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
             var thread = new LibuvThread(transport);
             var mockScheduler = new Mock<IScheduler>();
             Action backPressure = null;
-            mockScheduler.Setup(m => m.Schedule(It.IsAny<Action>())).Callback<Action>(a =>
+            mockScheduler.Setup(m => m.Schedule(It.IsAny<Action<object>>(), It.IsAny<object>())).Callback<Action<object>, object>((a, o) =>
             {
-                backPressure = a;
+                backPressure = () => a(o);
             });
             mockConnectionHandler.InputOptions.WriterScheduler = mockScheduler.Object;
             mockConnectionHandler.OutputOptions.ReaderScheduler = thread;