buffered I/O

Benchmarking/CapnpProfile/CapnpProfile.csproj

@@ -8,10 +8,11 @@
   <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">
     <DebugType>full</DebugType>
     <DebugSymbols>true</DebugSymbols>
+    <PlatformTarget>x64</PlatformTarget>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.29-g6d711b8579" />
+    <PackageReference Include="Capnp.Net.Runtime" Version="1.3.30-g1649067ef6" />
     <PackageReference Include="CapnpC.CSharp.MsBuild.Generation" Version="1.3.29-g6d711b8579" />
   </ItemGroup>
 

Capnp.Net.Runtime/Capnp.Net.Runtime.csproj

@@ -29,7 +29,7 @@
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)|$(TargetFramework)|$(Platform)'=='Release|netstandard2.0|AnyCPU'">
-    <DebugType>full</DebugType>
+    <DebugType>portable</DebugType>
     <DebugSymbols>true</DebugSymbols>
   </PropertyGroup>
 

Capnp.Net.Runtime/FramePump.cs

@@ -116,6 +116,8 @@ namespace Capnp
 #endif
                     _writer.Write(bytes);
                 }
+
+                _writer.Flush();
             }
         }
 

Capnp.Net.Runtime/Rpc/TcpRpcClient.cs

@@ -128,6 +128,8 @@ namespace Capnp.Rpc
             _rpcEngine = new RpcEngine();
             _client = new TcpClient();
             _client.ExclusiveAddressUse = false;
+
+            InjectMidlayer(s => new Util.DuplexBufferedStream(s));
         }
 
         /// <summary>

Capnp.Net.Runtime/Rpc/TcpRpcServer.cs

@@ -68,6 +68,7 @@ namespace Capnp.Rpc
                 _server = server;
                 Client = client;
                 _stream = client.GetStream();
+                InjectMidlayer(s => new Util.DuplexBufferedStream(s));
             }
 
             public void Start()

Capnp.Net.Runtime/Util/DuplexBufferedStream.cs

@@ -0,0 +1,83 @@
+using System;
+using System.IO;
+
+namespace Capnp.Util
+{
+    internal class DuplexBufferedStream : Stream
+    {
+        readonly BufferedStream _readStream;
+        readonly BufferedStream _writeStream;
+        readonly object _reentrancyBlocker = new object();
+
+        public DuplexBufferedStream(Stream stream)
+        {
+            _readStream = new BufferedStream(stream);
+            _writeStream = new BufferedStream(stream);
+        }
+
+        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)
+        {
+            _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);
+        }
+    }
+}