introduce inlineHandlers setting

src/csharp/Grpc.Core/GrpcEnvironment.cs

@@ -39,6 +39,7 @@ namespace Grpc.Core
         static int refCount;
         static int? customThreadPoolSize;
         static int? customCompletionQueueCount;
+        static bool inlineHandlers;
         static readonly HashSet<Channel> registeredChannels = new HashSet<Channel>();
         static readonly HashSet<Server> registeredServers = new HashSet<Server>();
 
@@ -217,13 +218,32 @@ namespace Grpc.Core
             }
         }
 
+        /// <summary>
+        /// By default, gRPC's internal event handlers get offloaded to .NET default thread pool thread (<c>inlineHandlers=false</c>).
+        /// Setting <c>inlineHandlers</c> to <c>true</c> will allow scheduling the event handlers directly to
+        /// <c>GrpcThreadPool</c> internal threads. That can lead to significant performance gains in some situations,
+        /// but requires user to never block in async code (incorrectly written code can easily lead to deadlocks).
+        /// Inlining handlers is an advanced setting and you should only use it if you know what you are doing.
+        /// Most users should rely on the default value provided by gRPC library.
+        /// Note: this method is part of an experimental API that can change or be removed without any prior notice.
+        /// Note: <c>inlineHandlers=true</c> was the default in gRPC C# v1.4.x and earlier.
+        /// </summary>
+        public static void SetHandlerInlining(bool inlineHandlers)
+        {
+            lock (staticLock)
+            {
+                GrpcPreconditions.CheckState(instance == null, "Can only be set before GrpcEnvironment is initialized");
+                GrpcEnvironment.inlineHandlers = inlineHandlers;
+            }
+        }
+
         /// <summary>
         /// Creates gRPC environment.
         /// </summary>
         private GrpcEnvironment()
         {
             GrpcNativeInit();
-            threadPool = new GrpcThreadPool(this, GetThreadPoolSizeOrDefault(), GetCompletionQueueCountOrDefault());
+            threadPool = new GrpcThreadPool(this, GetThreadPoolSizeOrDefault(), GetCompletionQueueCountOrDefault(), inlineHandlers);
             threadPool.Start();
         }
 

src/csharp/Grpc.Core/Internal/GrpcThreadPool.cs

@@ -41,6 +41,7 @@ namespace Grpc.Core.Internal
         readonly List<Thread> threads = new List<Thread>();
         readonly int poolSize;
         readonly int completionQueueCount;
+        readonly bool inlineHandlers;
 
         readonly List<BasicProfiler> threadProfilers = new List<BasicProfiler>();  // profilers assigned to threadpool threads
 
@@ -54,11 +55,13 @@ namespace Grpc.Core.Internal
         /// <param name="environment">Environment.</param>
         /// <param name="poolSize">Pool size.</param>
         /// <param name="completionQueueCount">Completion queue count.</param>
-        public GrpcThreadPool(GrpcEnvironment environment, int poolSize, int completionQueueCount)
+        /// <param name="inlineHandlers">Handler inlining.</param>
+        public GrpcThreadPool(GrpcEnvironment environment, int poolSize, int completionQueueCount, bool inlineHandlers)
         {
             this.environment = environment;
             this.poolSize = poolSize;
             this.completionQueueCount = completionQueueCount;
+            this.inlineHandlers = inlineHandlers;
             GrpcPreconditions.CheckArgument(poolSize >= completionQueueCount,
                 "Thread pool size cannot be smaller than the number of completion queues used.");
         }
@@ -168,7 +171,14 @@ namespace Grpc.Core.Internal
                     {
                         var callback = cq.CompletionRegistry.Extract(tag);
                         // Use cached delegates to avoid unnecessary allocations
-                        ThreadPool.QueueUserWorkItem(success ? RunCompletionQueueEventCallbackSuccess : RunCompletionQueueEventCallbackFailure, callback);
+                        if (!inlineHandlers)
+                        {
+                            ThreadPool.QueueUserWorkItem(success ? RunCompletionQueueEventCallbackSuccess : RunCompletionQueueEventCallbackFailure, callback);
+                        }
+                        else
+                        {
+                            RunCompletionQueueEventCallback(callback, success);
+                        }
                     }
                     catch (Exception e)
                     {