Streams TCP Transport with Hand-Written Protobuf Deserialization

src/Akka.sln

@@ -61,9 +61,6 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Protobuf", "Protobuf", "{98
 	EndProjectSection
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Build", "Build", "{10C5B1E8-15B5-4EB3-81AE-1FC054FE1305}"
-	ProjectSection(SolutionItems) = preProject
-		..\build.fsx = ..\build.fsx
-	EndProjectSection
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Contrib", "Contrib", "{588C1513-FAB6-42C3-B6FC-3485F13620CF}"
 EndProject

src/benchmark/RemotePingPong/Program.cs

@@ -55,7 +55,11 @@ public static Config CreateActorSystemConfig(string actorSystemName, string ipOr
 
                 dot-netty.tcp {
                     port = 0
-                    hostname = ""localhost""
+                    hostname = """"
+                    batching {
+                        enabled = true
+                        flush-interval = 40ms
+                    }
                 }
               }
             }");
@@ -69,7 +73,14 @@ public static Config CreateActorSystemConfig(string actorSystemName, string ipOr
 
         private static void Main(params string[] args)
         {
-            Process.GetCurrentProcess().PriorityClass = ProcessPriorityClass.High;
+            AppDomain.CurrentDomain.UnhandledException += (sender, eventArgs) =>
+            {
+                Console.WriteLine(eventArgs.ExceptionObject as Exception);
+                Console.WriteLine("STACKOVERFLOW");
+                Console.WriteLine(sender);
+                Console.WriteLine(eventArgs.ExceptionObject as Exception);
+            };
+            Process.GetCurrentProcess().PriorityClass = ProcessPriorityClass.AboveNormal;
             uint timesToRun;
             if (args.Length == 0 || !uint.TryParse(args[0], out timesToRun))
             {
@@ -82,7 +93,7 @@ private static void Main(params string[] args)
 
         private static async void Start(uint timesToRun)
         {
-            const long repeat = 100000L;
+            const long repeat = 50000L;
 
             var processorCount = Environment.ProcessorCount;
             if (processorCount == 0)
@@ -116,9 +127,18 @@ private static async void Start(uint timesToRun)
                 var bestThroughput = 0L;
                 foreach (var throughput in GetClientSettings())
                 {
-                    var result1 = await Benchmark(throughput, repeat, bestThroughput, redCount);
-                    bestThroughput = result1.Item2;
-                    redCount = result1.Item3;
+                    try
+                    {
+                        var result1 = await Benchmark(throughput, repeat, bestThroughput, redCount);
+                        bestThroughput = result1.Item2;
+                        redCount = result1.Item3;
+                    }
+                    catch (Exception e)
+                    {
+                        Console.WriteLine(e);
+                        throw;
+                    }
+
                 }
             }
 
@@ -182,16 +202,18 @@ private static long GetTotalMessagesReceived(int numberOfClients, long numberOfR
                 throw new Exception("Received report that 1 or more remote actor is unable to begin the test. Aborting run.");
             }
 
+            var rng = new Random();
+            var rand = new byte[2048];
+            rng.NextBytes(rand);
             var sw = Stopwatch.StartNew();
             receivers.ForEach(c =>
             {
                 for (var i = 0; i < 50; i++) // prime the pump so EndpointWriters can take advantage of their batching model
-                    c.Tell("hit");
+                    c.Tell("hi");
             });
             var waiting = Task.WhenAll(tasks);
             await Task.WhenAll(waiting);
             sw.Stop();
-
             // force clean termination
             var termination = Task.WhenAll(new[] { system1.Terminate(), system2.Terminate() }).Wait(TimeSpan.FromSeconds(10));
 

src/contrib/transports/Akka.Remote.Transport.Streams.Tcp/Akka.Remote.Transport.Streams.Tcp.csproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>netstandard2.0</TargetFramework>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\..\..\core\Akka.Remote\Akka.Remote.csproj" />
+      <ProjectReference Include="..\..\..\core\Akka.Streams\Akka.Streams.csproj" />
+    </ItemGroup>
+
+</Project>