Scheduler improvements

src/System.IO.Pipelines.Networking.Sockets/SocketConnection.cs

@@ -86,8 +86,8 @@ internal SocketConnection(Socket socket, MemoryPool<byte> pool)
 
             // TODO: Make this configurable
             // Dispatch to avoid deadlocks
-            _input = new Pipe(new PipeOptions(pool, Scheduler.TaskRun, Scheduler.TaskRun));
-            _output = new Pipe(new PipeOptions(pool, Scheduler.TaskRun, Scheduler.TaskRun));
+            _input = new Pipe(new PipeOptions(pool, Scheduler.ThreadPool, Scheduler.ThreadPool));
+            _output = new Pipe(new PipeOptions(pool, Scheduler.ThreadPool, Scheduler.ThreadPool));
 
             _receiveTask = ReceiveFromSocketAndPushToWriterAsync();
             _sendTask = ReadFromReaderAndWriteToSocketAsync();

src/System.IO.Pipelines/System.IO.Pipelines.csproj

@@ -2,7 +2,7 @@
   <Import Project="..\..\tools\common.props" />
   <PropertyGroup>
     <Description>An abstraction for doing efficient asynchronous IO</Description>
-    <TargetFramework>netstandard1.1</TargetFramework>
+    <TargetFrameworks>netstandard1.1;netstandard2.0;netcoreapp2.1</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     <PackageIconUrl>http://go.microsoft.com/fwlink/?linkid=833199</PackageIconUrl>
     <!--<DefineConstants Condition="'$(Configuration)' == 'Debug'">$(DefineConstants);BLOCK_LEASE_TRACKING</DefineConstants>-->

src/System.IO.Pipelines/System/Threading/InlineScheduler.cs

@@ -4,7 +4,7 @@
 
 namespace System.Threading
 {
-    internal class InlineScheduler : Scheduler
+    internal sealed class InlineScheduler : Scheduler
     {
         public override void Schedule(Action action)
         {

src/System.IO.Pipelines/System/Threading/Scheduler.cs

@@ -5,8 +5,11 @@ namespace System.Threading
 {
     public abstract class Scheduler
     {
-        public static Scheduler TaskRun { get; } = new TaskRunScheduler();
-        public static Scheduler Inline { get; } = new InlineScheduler();
+        private static ThreadPoolScheduler _threadPoolScheduler = new ThreadPoolScheduler();
+        private static InlineScheduler _inlineScheduler = new InlineScheduler();
+
+        public static Scheduler ThreadPool => _threadPoolScheduler;
+        public static Scheduler Inline => _inlineScheduler;
 
         public abstract void Schedule(Action action);
         public abstract void Schedule(Action<object> action, object state);

src/System.IO.Pipelines/System/Threading/ThreadPoolScheduler.cs

@@ -0,0 +1,83 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Threading.Tasks;
+
+namespace System.Threading
+{
+    internal sealed class ThreadPoolScheduler : Scheduler
+    {
+        public override void Schedule(Action action)
+        {
+#if NETCOREAPP2_1
+            // Queue to low contention local ThreadPool queue; rather than global queue as per Task
+            Threading.ThreadPool.QueueUserWorkItem(_actionAsTask, action, preferLocal: true);
+#elif NETSTANDARD2_0
+            Threading.ThreadPool.QueueUserWorkItem(_actionAsTask, action);
+#else
+            Task.Factory.StartNew(action);
+#endif
+        }
+
+        public override void Schedule(Action<object> action, object state)
+        {
+#if NETCOREAPP2_1
+            // Queue to low contention local ThreadPool queue; rather than global queue as per Task
+            Threading.ThreadPool.QueueUserWorkItem(_actionObjectAsTask, new ActionObjectAsTask(action, state), preferLocal: true);
+#elif NETSTANDARD2_0
+            Threading.ThreadPool.QueueUserWorkItem(_actionObjectAsTask, new ActionObjectAsTask(action, state));
+#else
+            Task.Factory.StartNew(action, state);
+#endif
+        }
+
+#if NETCOREAPP2_1 || NETSTANDARD2_0
+        // Catches only the exception into a failed Task, so the fire-and-forget action 
+        // can be queued directly to the ThreadPool without the extra overhead of as Task
+        private readonly static WaitCallback _actionAsTask = state =>
+        {
+            try
+            {
+                ((Action)state)();
+            }
+            catch (Exception ex)
+            {
+                // Create faulted Task for the TaskScheulder to handle exception
+                // rather than letting it escape onto the ThreadPool and crashing the process
+                Task.FromException(ex);
+            }
+        };
+
+        private readonly static WaitCallback _actionObjectAsTask = state => ((ActionObjectAsTask)state).Run();
+
+        private sealed class ActionObjectAsTask
+        {
+            private Action<object> _action;
+            private object _state;
+
+            public ActionObjectAsTask(Action<object> action, object state)
+            {
+                _action = action;
+                _state = state;
+            }
+
+            // Catches only the exception into a failed Task, so the fire-and-forget action 
+            // can be queued directly to the ThreadPool without the extra overhead of as Task
+            public void Run()
+            {
+                try
+                {
+                    _action(_state);
+                }
+                catch (Exception ex)
+                {
+                    // Create faulted Task for the TaskScheulder to handle exception
+                    // rather than letting it escape onto the ThreadPool and crashing the process
+                    Task.FromException(ex);
+                }
+            }
+        }
+#endif
+    }
+}