TcpRpcServer,StartAccepting

Benchmarking/Benchmark/Benchmark.csproj

@@ -7,7 +7,7 @@
 
   <ItemGroup>
     <PackageReference Include="BenchmarkDotNet" Version="0.12.0" />
-    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.32-g49c5e80436" />
+    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.33-g596a97a362" />
     <PackageReference Include="CapnpC.CSharp.MsBuild.Generation" Version="1.2.138" />
     <PackageReference Include="Google.Protobuf" Version="3.11.3" />
     <PackageReference Include="Grpc.Net.Client" Version="2.27.0" />

Benchmarking/Benchmark/CapnpBenchmark.cs

@@ -13,6 +13,9 @@ namespace Benchmark
         [Params(20, 200, 2000, 20000, 200000, 2000000)]
         public int PayloadBytes;
 
+        [Params(0, 256, 1024, 4096)]
+        public int BufferSize;
+
         TcpRpcClient _client;
         IEchoer _echoer;
         byte[] _payload;
@@ -21,6 +24,8 @@ namespace Benchmark
         public void Setup()
         {
             _client = new TcpRpcClient("localhost", 5002);
+            if (BufferSize > 0)
+                _client.AddBuffering(BufferSize);
             _client.WhenConnected.Wait();
             _echoer = _client.GetMain<IEchoer>();
             _payload = new byte[PayloadBytes];

Benchmarking/EchoServiceCapnp/EchoServiceCapnp.csproj

@@ -6,7 +6,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.32-g49c5e80436" />
+    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.33-g596a97a362" />
     <PackageReference Include="CapnpC.CSharp.MsBuild.Generation" Version="1.2.138" />
   </ItemGroup>
 

Benchmarking/EchoServiceCapnp/Program.cs

@@ -11,6 +11,7 @@ namespace EchoServiceCapnp
         {
             using (var server = new TcpRpcServer(IPAddress.Any, 5002))
             {
+                server.AddBuffering();
                 server.Main = new CapnpEchoService();
                 Console.WriteLine("Press RETURN to stop listening");
                 Console.ReadLine();

Capnp.Net.Runtime/Rpc/TcpRpcServer.cs

@@ -153,17 +153,17 @@ namespace Capnp.Rpc
         }
 
         readonly RpcEngine _rpcEngine;
-        readonly TcpListener _listener;
         readonly object _reentrancyBlocker = new object();
-        readonly Thread _acceptorThread;
         readonly List<Connection> _connections = new List<Connection>();
+        Thread? _acceptorThread;
+        TcpListener? _listener;
 
         /// <summary>
         /// Gets the number of currently active inbound TCP connections.
         /// </summary>
         public int ConnectionCount { get; private set; }
 
-        void AcceptClients()
+        void AcceptClients(TcpListener listener)
         {
             try
             {
@@ -172,7 +172,7 @@ namespace Capnp.Rpc
 
                 while (true)
                 {
-                    var client = _listener.AcceptTcpClient();
+                    var client = listener.AcceptTcpClient();
                     var connection = new Connection(this, client);
 
                     lock (_reentrancyBlocker)
@@ -236,7 +236,10 @@ namespace Capnp.Rpc
         /// </summary>
         public void Dispose()
         {
-            StopListening();
+            if (_listener != null)
+            {
+                StopListening();
+            }
 
             var connections = new List<Connection>();
 
@@ -252,8 +255,6 @@ namespace Capnp.Rpc
                 SafeJoin(connection.PumpRunner);
             }
 
-            SafeJoin(_acceptorThread);
-
             GC.SuppressFinalize(this);
         }
 
@@ -262,6 +263,9 @@ namespace Capnp.Rpc
         /// </summary>
         public void StopListening()
         {
+            if (_listener == null)
+                throw new InvalidOperationException("Listening was never started");
+
             try
             {
                 _listener.Stop();
@@ -269,6 +273,12 @@ namespace Capnp.Rpc
             catch (SocketException)
             {
             }
+            finally
+            {
+                _listener = null;
+                SafeJoin(_acceptorThread);
+                _acceptorThread = null;
+            }
         }
 
         /// <summary>
@@ -292,15 +302,46 @@ namespace Capnp.Rpc
         /// <summary>
         /// Constructs an instance.
         /// </summary>
+        public TcpRpcServer()
+        {
+            _rpcEngine = new RpcEngine();
+
+        }
+
+        /// <summary>
+        /// Constructs an instance, starts listening to the specified TCP/IP endpoint and accepting clients.
+        /// If you intend configuring a midlayer or consuming the <see cref="OnConnectionChanged"/> event, 
+        /// you should not use this constructor, since it may lead to an early-client race condition.
+        /// Instead, use the parameterless constructor, configure, then call <see cref="StartAccepting(IPAddress, int)"/>.
+        /// </summary>
         /// <param name="localAddr">An System.Net.IPAddress that represents the local IP address.</param>
         /// <param name="port">The port on which to listen for incoming connection attempts.</param>
         /// <exception cref="ArgumentNullException"><paramref name="localAddr"/> is null.</exception>
         /// <exception cref="ArgumentOutOfRangeException"><paramref name="port"/> is not between System.Net.IPEndPoint.MinPort and System.Net.IPEndPoint.MaxPort.</exception>
-        public TcpRpcServer(IPAddress localAddr, int port)
+        /// <exception cref="SocketException">The underlying <see cref="TcpListener"/> detected an error condition, such as the desired endpoint is already occupied.</exception>
+        public TcpRpcServer(IPAddress localAddr, int port): this()
         {
-            _rpcEngine = new RpcEngine();
+            StartAccepting(localAddr, port);
-            _listener = new TcpListener(localAddr, port);
+        }
-            _listener.ExclusiveAddressUse = false;
+
+        /// <summary>
+        /// Starts listening to the specified TCP/IP endpoint and accepting clients. 
+        /// </summary>
+        /// <param name="localAddr">An System.Net.IPAddress that represents the local IP address.</param>
+        /// <param name="port">The port on which to listen for incoming connection attempts.</param>
+        /// <exception cref="ArgumentNullException"><paramref name="localAddr"/> is null.</exception>
+        /// <exception cref="ArgumentOutOfRangeException"><paramref name="port"/> is not between System.Net.IPEndPoint.MinPort and System.Net.IPEndPoint.MaxPort.</exception>
+        /// <exception cref="InvalidOperationException">Listening activity was already started</exception>
+        /// <exception cref="SocketException">The underlying <see cref="TcpListener"/> detected an error condition, such as the desired endpoint is already occupied.</exception>
+        public void StartAccepting(IPAddress localAddr, int port)
+        {
+            if (_listener != null)
+                throw new InvalidOperationException("Listening activity was already started");
+
+            var listener = new TcpListener(localAddr, port)
+            {
+                ExclusiveAddressUse = false
+            };
 
             int attempt = 0;
 
@@ -308,32 +349,33 @@ namespace Capnp.Rpc
             {
                 try
                 {
-                    _listener.Start();
+                    listener.Start();
                     break;
                 }
                 catch (SocketException socketException)
                 {
                     if (attempt == 5)
                         throw;
-                    
+
                     Logger.LogWarning($"Failed to listen on port {port}, attempt {attempt}: {socketException}");
                 }
 
                 ++attempt;
+                Thread.Sleep(10);
             }
 
-            _acceptorThread = new Thread(AcceptClients);
+            _acceptorThread = new Thread(() => AcceptClients(listener));
-
+            _listener = listener;
             _acceptorThread.Start();
         }
 
         /// <summary>
         /// Whether the thread which is responsible for acception incoming attempts is still alive.
-        /// The thread will die upon disposal, but also in case of a socket error condition.
+        /// The thread will die after calling <see cref="StopListening"/>, upon disposal, but also in case of a socket error condition.
         /// Errors which occur on a particular connection will just close that connection and won't interfere
         /// with the acceptor thread.
         /// </summary>
-        public bool IsAlive => _acceptorThread.IsAlive;
+        public bool IsAlive => _acceptorThread?.IsAlive ?? false;
 
         /// <summary>
         /// Sets the bootstrap capability. It must be an object which implements a valid capability interface