additional write buffer

Benchmarking/Benchmark/Benchmark.csproj

@@ -7,7 +7,7 @@
 
   <ItemGroup>
     <PackageReference Include="BenchmarkDotNet" Version="0.12.0" />
-    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.90-g65e87e5aa9" />
+    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.92-gdb2d6bce2e" />
     <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/Program.cs

@@ -11,8 +11,11 @@ namespace Benchmark
     {
         static void Main(string[] args)
         {
-            BenchmarkRunner.Run<GrpcBenchmark>();
+            if (args.Length == 0 || args[0] == "grpc")
-            BenchmarkRunner.Run<CapnpBenchmark>();
+                BenchmarkRunner.Run<GrpcBenchmark>();
+
+            if (args.Length == 0 || args[0] == "capnp")
+                BenchmarkRunner.Run<CapnpBenchmark>();
         }
     }
 }

Benchmarking/EchoServiceCapnp/EchoServiceCapnp.csproj

@@ -6,8 +6,8 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.90-g65e87e5aa9" />
+    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.92-gdb2d6bce2e" />
-    <PackageReference Include="CapnpC.CSharp.MsBuild.Generation" Version="1.2.138" />
+    <PackageReference Include="CapnpC.CSharp.MsBuild.Generation" Version="1.3.92-gdb2d6bce2e" />
   </ItemGroup>
 
 </Project>

Capnp.Net.Runtime/Rpc/MidlayerExtensions.cs

@@ -17,7 +17,7 @@ namespace Capnp.Rpc
         /// <param name="bufferSize">Buffer size (bytes). You should choose it according to the maximum expected raw capnp frame size</param>
         public static void AddBuffering(this ISupportsMidlayers obj, int bufferSize)
         {
-            obj.InjectMidlayer(s => new Util.BufferedNetworkStreamAdapter(s, bufferSize));
+            obj.InjectMidlayer(s => new Util.WriteBufferedStream(new Util.BufferedNetworkStreamAdapter(s, bufferSize), bufferSize));
         }
 
         /// <summary>
@@ -27,7 +27,7 @@ namespace Capnp.Rpc
         /// <param name="obj"><see cref="TcpRpcServer"/> or <see cref="TcpRpcClient"/></param>
         public static void AddBuffering(this ISupportsMidlayers obj)
         {
-            obj.InjectMidlayer(s => new Util.BufferedNetworkStreamAdapter(s));
+            obj.InjectMidlayer(s => new Util.WriteBufferedStream(new Util.BufferedNetworkStreamAdapter(s)));
         }
     }
 }

Capnp.Net.Runtime/Util/WriteBufferedStream.cs

@@ -0,0 +1,96 @@
+using System;
+using System.IO;
+
+namespace Capnp.Util
+{
+    internal class WriteBufferedStream : Stream
+    {
+        // A buffer size of 1024 bytes seems to be a good comprise, giving good performance 
+        // in TCP/IP-over-localhost scenarios for small to medium (200kiB) frame sizes.
+        const int DefaultBufferSize = 1024;
+
+        readonly Stream _readStream;
+        readonly BufferedStream _writeStream;
+        readonly int _bufferSize;
+        readonly object _reentrancyBlocker = new object();
+
+        public WriteBufferedStream(Stream stream, int bufferSize)
+        {
+            _readStream = stream;
+            _writeStream = new BufferedStream(stream, bufferSize);
+            _bufferSize = bufferSize;
+        }
+
+        public WriteBufferedStream(Stream stream) : this(stream, DefaultBufferSize)
+        {
+        }
+
+        public override bool CanRead => true;
+
+        public override bool CanSeek => false;
+
+        public override bool CanWrite => true;
+
+        public override long Length => 0;
+
+        public override long Position
+        {
+            get => 0;
+            set => throw new NotSupportedException();
+        }
+
+        public override void Flush()
+        {
+            _writeStream.Flush();
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            return _readStream.Read(buffer, offset, count);
+        }
+
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void SetLength(long value)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            if (buffer.Length > _bufferSize) // avoid moiré-like timing effects
+                _writeStream.Flush();
+
+            _writeStream.Write(buffer, offset, count);
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                lock (_reentrancyBlocker)
+                {
+                    try
+                    {
+                        _readStream.Dispose();
+                    }
+                    catch
+                    {
+                    }
+                    try
+                    {
+                        _writeStream.Dispose();
+                    }
+                    catch
+                    {
+                    }
+                }
+            }
+
+            base.Dispose(disposing);
+        }
+    }
+}