Auto-tuning DotNetty batching + removing scheduler from batching system (#4685)

* implement FlushConsolidationHandler

* removed previous BatchWriter settings + handle

* removed unused batching settings from Akka.Remote

Author: Aaronontheweb

src/common.props

@@ -2,7 +2,7 @@
   <PropertyGroup>
     <Copyright>Copyright © 2013-2020 Akka.NET Team</Copyright>
     <Authors>Akka.NET Team</Authors>
-    <VersionPrefix>1.4.13</VersionPrefix>
+    <VersionPrefix>1.4.14</VersionPrefix>
     <PackageIconUrl>https://getakka.net/images/akkalogo.png</PackageIconUrl>
     <PackageProjectUrl>https://github.com/akkadotnet/akka.net</PackageProjectUrl>
     <PackageLicenseUrl>https://github.com/akkadotnet/akka.net/blob/master/LICENSE</PackageLicenseUrl>
@@ -28,33 +28,7 @@
     <CopyLocalLockFileAssemblies>true</CopyLocalLockFileAssemblies>
   </PropertyGroup>
   <PropertyGroup>
-    <PackageReleaseNotes>Maintenance Release for Akka.NET 1.4**
-Akka.NET v1.4.13 includes a number of bug fixes and enhancements:
-`AppVersion` now uses Assembly Version by Default**
-The new `AppVersion` setting, which is used to communicate application version numbers throughout Akka.Cluster and is used in scenarios such as Akka.Cluster.Sharding to help determine which nodes receive new shard allocations and which ones do not, now uses the following default HOCON setting:
-```
-akka.cluster.app-version = assembly-version
-```
-By default now the `AppVersion` communicated inside Akka.Cluster `Member` events uses the `Major.Minor.BuildNumber` from the `Assembly.GetEntryssembly()` or `Assembly.GetExecutingAssembly()` (in case the `EntryAssembly` is `null`). That way any updates made to your executable's (i.e. the .dll that hosts `Program.cs`) version number will be automatically reflected in the cluster now without Akka.NET developers having to set an additional configuration value during deployments.
-Other bug fixes and improvements:
-[Akka.IO: UdpExt.Manager: OverflowException when sending UDP packets to terminated clients](https://github.com/akkadotnet/akka.net/issues/4641)
-[Akka.Configuration / Akka.Streams: Memory Leak when using many short lived instances of ActorMaterializer](https://github.com/akkadotnet/akka.net/issues/4659)
-[Akka: Deprecate `PatternMatch`](https://github.com/akkadotnet/akka.net/issues/4658)
-[Akka: FSM: exception in LogTermination changes stopEvent.Reason to Shutdown](https://github.com/akkadotnet/akka.net/issues/3723)
-[Akka.Cluster.Tools: ClusterSingleton - Ignore possible state change in start](https://github.com/akkadotnet/akka.net/pull/4646)
-[Akka.Cluster.Tools: DistributedPubSub - new setting and small fixes](https://github.com/akkadotnet/akka.net/pull/4649)
-[Akka.DistributedData: `KeyNotFoundException` thrown periodically](https://github.com/akkadotnet/akka.net/issues/4639)
-To see the [full set of fixes in Akka.NET v1.4.13, please see the milestone on Github](https://github.com/akkadotnet/akka.net/milestone/44).
-| COMMITS | LOC+ | LOC- | AUTHOR |
-| --- | --- | --- | --- |
-| 5 | 316 | 29 | Aaron Stannard |
-| 2 | 53 | 8 | Gregorius Soedharmo |
-| 2 | 223 | 197 | zbynek001 |
-| 2 | 2 | 2 | dependabot-preview[bot] |
-| 2 | 11 | 3 | Ebere Abanonu |
-| 1 | 37 | 27 | Razvan Goga |
-| 1 | 217 | 11 | motmot80 |
-| 1 | 2 | 0 | Ismael Hamed |</PackageReleaseNotes>
+    <PackageReleaseNotes>Placeholder for nightlies**</PackageReleaseNotes>
   </PropertyGroup>
   <!-- SourceLink support for all Akka.NET projects -->
   <ItemGroup>

src/core/Akka.Remote.Tests/RemoteConfigSpec.cs

@@ -174,9 +174,7 @@ public void Remoting_should_contain_correct_BatchWriter_settings_in_ReferenceCon
             var s = DotNettyTransportSettings.Create(c);
 
             s.BatchWriterSettings.EnableBatching.Should().BeTrue();
-            s.BatchWriterSettings.FlushInterval.Should().Be(BatchWriterSettings.DefaultFlushInterval);
-            s.BatchWriterSettings.MaxPendingBytes.Should().Be(BatchWriterSettings.DefaultMaxPendingBytes);
-            s.BatchWriterSettings.MaxPendingWrites.Should().Be(BatchWriterSettings.DefaultMaxPendingWrites);
+            s.BatchWriterSettings.MaxExplicitFlushes.Should().Be(BatchWriterSettings.DefaultMaxPendingWrites);
         }
    }
 }

src/core/Akka.Remote.Tests/Transport/DotNettyBatchWriterSpecs.cs

@@ -66,17 +66,13 @@ public void Bugfix4434_should_overwrite_default_BatchWriterSettings()
                     batching{
                         enabled = false
                         max-pending-writes = 50
-                        max-pending-bytes = 32k
-                        flush-interval = 10ms
                     }
                 }
             ";
             var s = DotNettyTransportSettings.Create(c.GetConfig("akka.remote.dot-netty.tcp"));
 
             s.BatchWriterSettings.EnableBatching.Should().BeFalse();
-            s.BatchWriterSettings.FlushInterval.Should().NotBe(BatchWriterSettings.DefaultFlushInterval);
-            s.BatchWriterSettings.MaxPendingBytes.Should().NotBe(BatchWriterSettings.DefaultMaxPendingBytes);
-            s.BatchWriterSettings.MaxPendingWrites.Should().NotBe(BatchWriterSettings.DefaultMaxPendingWrites);
+            s.BatchWriterSettings.MaxExplicitFlushes.Should().NotBe(BatchWriterSettings.DefaultMaxPendingWrites);
         }
 
         /// <summary>
@@ -85,7 +81,7 @@ public void Bugfix4434_should_overwrite_default_BatchWriterSettings()
         [Fact]
         public async Task BatchWriter_should_succeed_with_timer()
         {
-            var writer = new BatchWriter(new BatchWriterSettings(), Sys.Scheduler);
+            var writer = new FlushConsolidationHandler();
             var ch = new EmbeddedChannel(Flush, writer);
 
             await Flush.Activated;
@@ -98,14 +94,13 @@ public async Task BatchWriter_should_succeed_with_timer()
                 var ints = Enumerable.Range(0, 4).ToArray();
                 foreach (var i in ints)
                 {
-                    _ = ch.WriteAsync(Unpooled.Buffer(1).WriteInt(i));
+                    _ = ch.WriteAndFlushAsync(Unpooled.Buffer(1).WriteInt(i));
                 }
 
                 // force write tasks to run
                 ch.RunPendingTasks();
 
-                ch.Unsafe.OutboundBuffer.TotalPendingWriteBytes().Should().Be(ints.Length * 4);
-                ch.OutboundMessages.Count.Should().Be(0);
+                ch.OutboundMessages.Count.Should().Be(ints.Length);
 
                 await AwaitAssertAsync(() =>
                 {
@@ -124,7 +119,7 @@ await AwaitAssertAsync(() =>
         [Fact]
         public async Task BatchWriter_should_flush_messages_during_shutdown()
         {
-            var writer = new BatchWriter(new BatchWriterSettings(), Sys.Scheduler);
+            var writer = new FlushConsolidationHandler();
             var ch = new EmbeddedChannel(Flush, writer);
 
             await Flush.Activated;
@@ -136,14 +131,13 @@ public async Task BatchWriter_should_flush_messages_during_shutdown()
             var ints = Enumerable.Range(0, 10).ToArray();
             foreach (var i in ints)
             {
-                _ = ch.WriteAsync(Unpooled.Buffer(1).WriteInt(i));
+                _ = ch.WriteAndFlushAsync(Unpooled.Buffer(1).WriteInt(i));
             }
 
             // force write tasks to run
             ch.RunPendingTasks();
 
-            ch.Unsafe.OutboundBuffer.TotalPendingWriteBytes().Should().Be(ints.Length * 4);
-            ch.OutboundMessages.Count.Should().Be(0);
+            ch.OutboundMessages.Count.Should().Be(ints.Length);
 
             // close channels
             _ = ch.CloseAsync();

src/core/Akka.Remote/Configuration/Remote.conf

@@ -444,28 +444,9 @@ akka {
         # fast (< 40ms) acknowledgement for all periodic messages.
         enabled = true
 
-        # The max write threshold based on the number of logical messages regardless of their size. 
-        # This is a safe default value - decrease it if you have a small number of remote actors
-        # who engage in frequent request->response communication which requires low latency (< 40ms).
+        # The maximum number of explicit flushes that will be allowed before being batched.
+        # Batching normally happens when the channel switches from writes to reads and self-tunes automatically.
         max-pending-writes = 30
-
-        # The max write threshold based on the byte size of all buffered messages. If there are 4 messages
-        # waiting to be written (with batching.max-pending-writes = 30) but their total size is greater than
-        # batching.max-pending-bytes, a flush will be triggered immediately.
-        #
-        # Increase this value is you have larger message sizes and watch to take advantage of batching, but
-        # otherwise leave it as-is.
-        #
-        # NOTE: this value should always be smaller than dot-netty.tcp.maximum-frame-size.
-        max-pending-bytes = 16k
-
-        # In the event that neither the batching.max-pending-writes or batching.max-pending-bytes
-        # is hit we guarantee that all pending writes will be flushed within this interval.
-        #
-        # This setting, realistically, can't be enforced any lower than the OS' clock resolution (~20ms).
-        # If you have a very low-traffic system, either disable pooling altogether or lower the batching.max-pending-writes
-        # threshold to maximize throughput. Otherwise, leave this setting as-is.
-        flush-interval = 40ms
       }
 
       # If set to "<id.of.dispatcher>" then the specified dispatcher