Removed busy waiting in the processing scheduler thread ProcessRequestQueue()

Author: Lercher

src/JsonRpc/InputHandler.cs

@@ -14,7 +14,6 @@ namespace JsonRpc
 {
     public class InputHandler : IInputHandler
     {
-        private readonly TimeSpan _sleepTime = TimeSpan.FromMilliseconds(50);
         public const char CR = '\r';
         public const char LF = '\n';
         public static char[] CRLF = { CR, LF };
@@ -28,7 +27,8 @@ public class InputHandler : IInputHandler
         private Thread _inputThread;
         private readonly IRequestRouter _requestRouter;
         private readonly IResponseRouter _responseRouter;
-        private readonly ConcurrentQueue<(RequestProcessType type, Func<Task> request)> _queue;
+        private readonly BlockingCollection<(RequestProcessType type, Func<Task> request)> _queue;
+        private readonly CancellationTokenSource _cancelQueue;
         private Thread _queueThread;
 
         public InputHandler(
@@ -46,26 +46,14 @@ IResponseRouter responseRouter
             _requestProcessIdentifier = requestProcessIdentifier;
             _requestRouter = requestRouter;
             _responseRouter = responseRouter;
-            _queue = new ConcurrentQueue<(RequestProcessType type, Func<Task> request)>();
+            _queue = new BlockingCollection<(RequestProcessType type, Func<Task> request)>();
+            _cancelQueue = new CancellationTokenSource();
 
             _inputThread = new Thread(ProcessInputStream) { IsBackground = true };
 
             _queueThread = new Thread(ProcessRequestQueue) { IsBackground = true };
         }
 
-        internal InputHandler(
-            TextReader input,
-            IOutputHandler outputHandler,
-            IReciever reciever,
-            IRequestProcessIdentifier requestProcessIdentifier,
-            IRequestRouter requestRouter,
-            IResponseRouter responseRouter,
-            TimeSpan sleepTime
-        ) : this(input, outputHandler, reciever, requestProcessIdentifier, requestRouter, responseRouter)
-        {
-            _sleepTime = sleepTime;
-        }
-
         public void Start()
         {
             _outputHandler.Start();
@@ -157,7 +145,7 @@ private void HandleRequest(string request)
             {
                 if (item.IsRequest)
                 {
-                    _queue.Enqueue((
+                    _queue.Add((
                         type,
                         async () => {
                             var result = await _requestRouter.RouteRequest(item.Request);
@@ -168,7 +156,7 @@ private void HandleRequest(string request)
                 }
                 else if (item.IsNotification)
                 {
-                    _queue.Enqueue((
+                    _queue.Add((
                         type,
                         () => {
                             _requestRouter.RouteNotification(item.Notification);
@@ -184,33 +172,32 @@ private void HandleRequest(string request)
             }
         }
 
-        private bool IsNextSerial()
-        {
-            return _queue.TryPeek(out var queueResult) && queueResult.type == RequestProcessType.Serial;
-        }
-
         private async void ProcessRequestQueue()
         {
-            while (true)
+            // see https://github.com/OmniSharp/csharp-language-server-protocol/issues/4
+            var token = _cancelQueue.Token;
+            var waitables = new List<Task>();
+            while(true)
             {
                 if (_queueThread == null) return;
-                var items = new List<Func<Task>>();
-                while (!_queue.IsEmpty)
+                if (_queue.TryTake(out var item, Timeout.Infinite, token))
                 {
-                    if (IsNextSerial() && items.Count > 0)
+                    var (type, request) = item;
+                    var task = request();
+                    if (type == RequestProcessType.Serial)
                     {
-                        break;
+                        await Task.WhenAll(waitables);
+                        waitables.Clear();
+                        task.Start();
+                        await task;
                     }
-
-                    if (_queue.TryDequeue(out var queueResult))
-                        items.Add(queueResult.request);
-                }
-
-                await Task.WhenAll(items.Select(x => x()));
-
-                if (_queue.IsEmpty)
-                {
-                    await Task.Delay(_sleepTime);
+                    else if (type == RequestProcessType.Parallel)
+                    {
+                        task.Start();
+                        waitables.Add(task);
+                    }
+                    else
+                        throw new NotImplementedException("Only Serial and Parallel execution types can be handled currently");
                 }
             }
         }
@@ -220,6 +207,7 @@ public void Dispose()
             _outputHandler.Dispose();
             _inputThread = null;
             _queueThread = null;
+            _cancelQueue.Cancel();
         }
     }
 }

test/JsonRpc.Tests/InputHandlerTests.cs

@@ -38,8 +38,7 @@ private static InputHandler NewHandler(
                 reciever,
                 requestProcessIdentifier,
                 requestRouter,
-                responseRouter,
-                TimeSpan.Zero);
+                responseRouter);
             handler.Start();
             cts.Wait();
             return handler;