Merge pull request #11 from Lercher/inputhandler-refactoring

Inputhandler refactoring for issue #4 and #8

Author: david-driscoll

src/JsonRpc/IScheduler.cs

@@ -0,0 +1,11 @@
+using System;
+using System.Threading.Tasks;
+
+namespace JsonRpc
+{
+    public interface IScheduler : IDisposable
+    {
+        void Start();
+        void Add(RequestProcessType type, Func<Task> request);
+    }
+}

src/JsonRpc/InputHandler.cs

@@ -1,20 +1,15 @@
 using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
 using System.IO;
 using System.Linq;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
-using JsonRpc.Server;
 using JsonRpc.Server.Messages;
 using Newtonsoft.Json.Linq;
 
 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,8 +23,7 @@ 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 Thread _queueThread;
+        private readonly IScheduler _scheduler;
 
         public InputHandler(
             TextReader input,
@@ -46,31 +40,16 @@ IResponseRouter responseRouter
             _requestProcessIdentifier = requestProcessIdentifier;
             _requestRouter = requestRouter;
             _responseRouter = responseRouter;
-            _queue = new ConcurrentQueue<(RequestProcessType type, Func<Task> request)>();
 
-            _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;
-        }
+            _scheduler = new ProcessScheduler();
+            _inputThread = new Thread(ProcessInputStream) { IsBackground = true, Name = "ProcessInputStream" };
+            }
 
         public void Start()
         {
             _outputHandler.Start();
             _inputThread.Start();
-            _queueThread.Start();
+            _scheduler.Start();
         }
 
         private async void ProcessInputStream()
@@ -81,10 +60,14 @@ private async void ProcessInputStream()
 
                 var buffer = new char[300];
                 var current = await _input.ReadBlockAsync(buffer, 0, MinBuffer);
+                if (current == 0) return; // no more _input
+
                 while (current < MinBuffer || buffer[current - 4] != CR || buffer[current - 3] != LF ||
                        buffer[current - 2] != CR || buffer[current - 1] != LF)
                 {
-                    current += await _input.ReadBlockAsync(buffer, current, 1);
+                    var n = await _input.ReadBlockAsync(buffer, current, 1);
+                    if (n == 0) return; // no more _input, mitigates endless loop here.
+                    current += n;
                 }
 
                 var headersContent = new string(buffer, 0, current);
@@ -97,17 +80,28 @@ private async void ProcessInputStream()
                     var value = headers[i].Trim();
                     if (header.Equals("Content-Length", StringComparison.OrdinalIgnoreCase))
                     {
-                        length = long.Parse(value);
+                        length = 0;
+                        long.TryParse(value, out length);
                     }
                 }
 
-                var requestBuffer = new char[length];
-
-                await _input.ReadBlockAsync(requestBuffer, 0, requestBuffer.Length);
-
-                var payload = new string(requestBuffer);
-
-                HandleRequest(payload);
+                if (length == 0 || length >= int.MaxValue)
+                {
+                    HandleRequest(string.Empty);
+                }
+                else
+                {
+                    var requestBuffer = new char[length];
+                    var received = 0;
+                    while (received < length)
+                    {
+                        var n = await _input.ReadBlockAsync(requestBuffer, received, requestBuffer.Length - received);
+                        if (n == 0) return; // no more _input
+                        received += n;
+                    }
+                    var payload = new string(requestBuffer);
+                    HandleRequest(payload);
+                }
             }
         }
 
@@ -158,24 +152,23 @@ private void HandleRequest(string request)
             {
                 if (item.IsRequest)
                 {
-                    _queue.Enqueue((
+                    _scheduler.Add(
                         type,
                         async () => {
                             var result = await _requestRouter.RouteRequest(item.Request);
-
                             _outputHandler.Send(result.Value);
                         }
-                    ));
+                    );
                 }
                 else if (item.IsNotification)
                 {
-                    _queue.Enqueue((
+                    _scheduler.Add(
                         type,
                         () => {
                             _requestRouter.RouteNotification(item.Notification);
                             return Task.CompletedTask;
                         }
-                    ));
+                    );
                 }
                 else if (item.IsError)
                 {
@@ -185,42 +178,12 @@ private void HandleRequest(string request)
             }
         }
 
-        private bool IsNextSerial()
-        {
-            return _queue.TryPeek(out var queueResult) && queueResult.type == RequestProcessType.Serial;
-        }
-
-        private async void ProcessRequestQueue()
-        {
-            while (true)
-            {
-                if (_queueThread == null) return;
-                var items = new List<Func<Task>>();
-                while (!_queue.IsEmpty)
-                {
-                    if (IsNextSerial() && items.Count > 0)
-                    {
-                        break;
-                    }
-
-                    if (_queue.TryDequeue(out var queueResult))
-                        items.Add(queueResult.request);
-                }
-
-                await Task.WhenAll(items.Select(x => x()));
-
-                if (_queue.IsEmpty)
-                {
-                    await Task.Delay(_sleepTime);
-                }
-            }
-        }
 
         public void Dispose()
         {
             _outputHandler.Dispose();
             _inputThread = null;
-            _queueThread = null;
+            _scheduler.Dispose();
         }
     }
 }

src/JsonRpc/OutputHandler.cs

@@ -10,7 +10,7 @@ namespace JsonRpc
     public class OutputHandler : IOutputHandler
     {
         private readonly TextWriter _output;
-        private Thread _thread;
+        private readonly Thread _thread;
         private readonly BlockingCollection<object> _queue;
         private readonly CancellationTokenSource _cancel;
 
@@ -19,9 +19,7 @@ public OutputHandler(TextWriter output)
             _output = output;
             _queue = new BlockingCollection<object>();
             _cancel = new CancellationTokenSource();
-            _thread = new Thread(ProcessOutputQueue) {
-                IsBackground = true
-            };
+            _thread = new Thread(ProcessOutputQueue) { IsBackground = true, Name = "ProcessOutputQueue" };
         }
 
         public void Start()
@@ -37,10 +35,9 @@ public void Send(object value)
         private void ProcessOutputQueue()
         {
             var token = _cancel.Token;
-            while (true)
+            try
             {
-                if (_thread == null) return;
-                try
+                while (true)
                 {
                     if (_queue.TryTake(out var value, Timeout.Infinite, token))
                     {
@@ -51,19 +48,24 @@ private void ProcessOutputQueue()
                         sb.Append($"Content-Length: {content.Length}\r\n");
                         sb.Append($"\r\n");
                         sb.Append(content);
-
                         _output.Write(sb.ToString());
                     }
                 }
-                catch (OperationCanceledException) { }
+            }
+            catch (OperationCanceledException ex)
+            {
+                if (ex.CancellationToken != token)
+                    throw;
+                // else ignore. Exceptions: OperationCanceledException - The CancellationToken has been canceled.
             }
         }
 
         public void Dispose()
         {
-            _output?.Dispose();
-            _thread = null;
             _cancel.Cancel();
+            _thread.Join();
+            _cancel.Dispose();
+            _output.Dispose();
         }
     }
 }

src/JsonRpc/ProcessScheduler.cs

@@ -0,0 +1,115 @@
+using System;
+using System.Collections.Generic;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace JsonRpc
+{
+    public class ProcessScheduler : IScheduler
+    {
+        private readonly BlockingCollection<(RequestProcessType type, Func<Task> request)> _queue;
+        private readonly CancellationTokenSource _cancel;
+        private readonly Thread _thread;
+
+        public ProcessScheduler()
+        {
+            _queue = new BlockingCollection<(RequestProcessType type, Func<Task> request)>();
+            _cancel = new CancellationTokenSource();
+            _thread = new Thread(ProcessRequestQueue) { IsBackground = true, Name = "ProcessRequestQueue" };
+        }
+
+        public void Start()
+        {
+            _thread.Start();
+        }
+
+        public void Add(RequestProcessType type, Func<Task> request)
+        {
+            _queue.Add((type, request));
+        }
+
+        private Task Start(Func<Task> request)
+        {
+            var t = request();
+            if (t.Status == TaskStatus.Created) // || t.Status = TaskStatus.WaitingForActivation ?
+                t.Start();
+            return t;
+        }
+
+        private List<Task> RemoveCompleteTasks(List<Task> list)
+        {
+            if (list.Count == 0) return list;
+
+            var result = new List<Task>();
+            foreach(var t in list)
+            {
+                if (t.IsFaulted)
+                {
+                    // TODO: Handle Fault
+                }
+                else if (!t.IsCompleted)
+                {
+                    result.Add(t);
+                }
+            }
+            return result;
+        }
+
+        public long _TestOnly_NonCompleteTaskCount = 0;
+        private void ProcessRequestQueue()
+        {
+            // see https://github.com/OmniSharp/csharp-language-server-protocol/issues/4
+            // no need to be async, because this thing already allocated a thread on it's own.
+            var token = _cancel.Token;
+            var waitables = new List<Task>();
+            try
+            {
+                while (true)
+                {
+                    if (_queue.TryTake(out var item, Timeout.Infinite, token))
+                    {
+                        var (type, request) = item;
+                        if (type == RequestProcessType.Serial)
+                        {
+                            Task.WaitAll(waitables.ToArray(), token);
+                            Start(request).Wait(token);
+                        }
+                        else if (type == RequestProcessType.Parallel)
+                        {
+                            waitables.Add(Start(request));
+                        }
+                        else
+                            throw new NotImplementedException("Only Serial and Parallel execution types can be handled currently");
+                        waitables = RemoveCompleteTasks(waitables);
+                        Interlocked.Exchange(ref _TestOnly_NonCompleteTaskCount, waitables.Count);
+                    }
+                }
+            }
+            catch (OperationCanceledException ex)
+            {
+                if (ex.CancellationToken != token)
+                    throw;
+                // OperationCanceledException - The CancellationToken has been canceled.
+                Task.WaitAll(waitables.ToArray(), TimeSpan.FromMilliseconds(1000));
+                var keeponrunning = RemoveCompleteTasks(waitables);
+                Interlocked.Exchange(ref _TestOnly_NonCompleteTaskCount, keeponrunning.Count);
+                keeponrunning.ForEach((t) =>
+                {
+                    // TODO: There is no way to abort a Task. As we don't construct the tasks, we can do nothing here
+                    // Option is: change the task factory "Func<Task> request" to a "Func<CancellationToken, Task> request"
+                });
+            }
+        }
+
+        private bool _disposed = false;
+        public void Dispose()
+        {
+            if (_disposed) return;
+            _disposed = true;
+            _cancel.Cancel();
+            _thread.Join();
+            _cancel.Dispose();
+        }
+    }
+}