support multi-threading

Author: Jacques Kang

src/IpcServiceSample.ConsoleServer/Program.cs

@@ -14,10 +14,6 @@ static void Main(string[] args)
             IServiceCollection services = ConfigureServices(new ServiceCollection());
             ServiceProvider serviceProvider = services.BuildServiceProvider();
 
-            // configure console logging
-            serviceProvider.GetRequiredService<ILoggerFactory>()
-                .AddConsole(LogLevel.Debug);
-
             // TODO start IPC service host
             IpcServiceHostBuilder
                 .Buid("pipeName", serviceProvider as IServiceProvider)
@@ -28,10 +24,17 @@ static void Main(string[] args)
         private static IServiceCollection ConfigureServices(IServiceCollection services)
         {
             services
-                .AddLogging();
+                .AddLogging(builder =>
+                {
+                    builder.AddConsole();
+                    builder.SetMinimumLevel(LogLevel.Debug);
+                });
 
             services
-                .AddIpc()
+                .AddIpc(options =>
+                {
+                    options.ThreadCount = 2;
+                })
                 .AddService<IComputingService, ComputingService>()
                 ;
 

src/JKang.IpcServiceFramework.Server/IpcServiceHost.cs

@@ -5,6 +5,7 @@
 using System;
 using System.IO.Pipes;
 using System.Reflection;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace JKang.IpcServiceFramework
@@ -14,6 +15,7 @@ public class IpcServiceHost : IIpcServiceHost
         private readonly string _pipeName;
         private readonly IServiceProvider _serviceProvider;
         private readonly ILogger<IpcServiceHost> _logger;
+        private readonly IpcServiceOptions _options;
         private readonly IIpcMessageSerializer _serializer;
         private readonly IValueConverter _converter;
 
@@ -22,45 +24,68 @@ public IpcServiceHost(string pipeName, IServiceProvider serviceProvider)
             _pipeName = pipeName;
             _serviceProvider = serviceProvider;
             _logger = _serviceProvider.GetService<ILogger<IpcServiceHost>>();
+            _options = _serviceProvider.GetRequiredService<IpcServiceOptions>();
             _serializer = _serviceProvider.GetRequiredService<IIpcMessageSerializer>();
             _converter = _serviceProvider.GetRequiredService<IValueConverter>();
         }
 
         public void Start()
         {
-            using (var server = new NamedPipeServerStream(_pipeName, PipeDirection.InOut, 1))
-            using (var writer = new IpcWriter(server, _serializer))
-            using (var reader = new IpcReader(server, _serializer))
+            Thread[] threads = new Thread[_options.ThreadCount];
+            for (int i = 0; i < threads.Length; i++)
             {
-                _logger?.LogInformation("IPC server started.");
-                while (true)
-                {
-                    _logger?.LogDebug("Waiting for connection...");
-                    server.WaitForConnection();
+                threads[i] = new Thread(StartServerThread);
+                threads[i].Start();
+            }
+            _logger?.LogInformation("IPC server started.");
 
-                    try
+            while (true)
+            {
+                Thread.Sleep(100);
+                for (int i = 0; i < threads.Length; i++)
+                {
+                    if (threads[i].Join(250))
                     {
-                        _logger?.LogDebug("client connected, reading request...");
-                        IpcRequest request = reader.ReadIpcRequest();
+                        // thread is finished, starting a new thread
+                        threads[i] = new Thread(StartServerThread);
+                        threads[i].Start();
+                    }
+                }
+            }
+        }
 
-                        _logger?.LogDebug("request received, invoking corresponding method...");
-                        IpcResponse response;
-                        using (IServiceScope scope = _serviceProvider.CreateScope())
-                        {
-                            response = GetReponse(request, scope);
-                        }
+        private void StartServerThread(object obj)
+        {
+            using (var server = new NamedPipeServerStream(_pipeName, PipeDirection.InOut, _options.ThreadCount))
+            using (var writer = new IpcWriter(server, _serializer))
+            using (var reader = new IpcReader(server, _serializer))
+            {
+                server.WaitForConnection();
 
-                        _logger?.LogDebug("sending response...");
-                        writer.Write(response);
+                try
+                {
+                    _logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] client connected, reading request...");
+                    IpcRequest request = reader.ReadIpcRequest();
 
-                        // disconnect client
-                        server.Disconnect();
-                    }
-                    catch (Exception ex)
+                    _logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] request received, invoking corresponding method...");
+                    IpcResponse response;
+                    using (IServiceScope scope = _serviceProvider.CreateScope())
                     {
-                        _logger?.LogError(ex, ex.Message);
-                        server.Disconnect();
+                        response = GetReponse(request, scope);
                     }
+
+                    _logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] sending response...");
+                    writer.Write(response);
+
+                    _logger?.LogDebug($"[thread {Thread.CurrentThread.ManagedThreadId}] done.");
+                }
+                catch (Exception ex)
+                {
+                    _logger?.LogError(ex, ex.Message);
+                }
+                finally
+                {
+                    server.Close();
                 }
             }
         }
@@ -115,7 +140,7 @@ private IpcResponse GetReponse(IpcRequest request, IServiceScope scope)
                 }
                 else
                 {
-                    return IpcResponse.Success(@return); 
+                    return IpcResponse.Success(@return);
                 }
             }
             catch (Exception ex)

src/JKang.IpcServiceFramework.Server/IpcServiceOptions.cs

@@ -0,0 +1,7 @@
+namespace JKang.IpcServiceFramework
+{
+    public class IpcServiceOptions
+    {
+        public int ThreadCount { get; set; } = 4;
+    }
+}

src/JKang.IpcServiceFramework.Server/IpcServiceServiceCollectionExtensions.cs

@@ -1,13 +1,27 @@
 using JKang.IpcServiceFramework.Services;
 using Microsoft.Extensions.DependencyInjection;
+using System;
 
 namespace JKang.IpcServiceFramework
 {
     public static class IpcServiceServiceCollectionExtensions
     {
         public static IIpcServiceCollection AddIpc(this IServiceCollection services)
+        {
+            return services.AddIpc(new IpcServiceOptions());
+        }
+
+        public static IIpcServiceCollection AddIpc(this IServiceCollection services, Action<IpcServiceOptions> configure)
+        {
+            var options = new IpcServiceOptions();
+            configure?.Invoke(options);
+            return services.AddIpc(options);
+        }
+
+        public static IIpcServiceCollection AddIpc(this IServiceCollection services, IpcServiceOptions options)
         {
             services
+                .AddSingleton(options)
                 .AddScoped<IValueConverter, DefaultValueConverter>()
                 .AddScoped<IIpcMessageSerializer, DefaultIpcMessageSerializer>()
                 ;