Akka.Remote: improved write performance with DotNetty flush-batching

src/core/Akka.Remote/Transport/DotNetty/AkkaLoggingHandler.cs

@@ -8,14 +8,93 @@
 using System;
 using System.Net;
 using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using Akka.Util;
 using DotNetty.Buffers;
+using DotNetty.Common.Concurrency;
 using DotNetty.Transport.Channels;
 using ILoggingAdapter = Akka.Event.ILoggingAdapter;
 
 namespace Akka.Remote.Transport.DotNetty
 {
+ internal class BatchWriter : ChannelHandlerAdapter
+ {
+ private readonly int _maxPendingWrites;
+ private readonly long _maxPendingMillis;
+
+ public BatchWriter(int maxPendingWrites = 30, long maxPendingMillis = 40l)
+ {
+ _maxPendingWrites = maxPendingWrites;
+ _maxPendingMillis = maxPendingMillis;
+ }
+
+ private int _currentPendingWrites = 0;
+
+ public bool HasPendingWrites => _currentPendingWrites > 0;
+
+ public override void HandlerAdded(IChannelHandlerContext context)
+ {
+ ScheduleFlush(context);
+ base.HandlerAdded(context);
+ }
+
+ public override Task WriteAsync(IChannelHandlerContext context, object message)
+ {
+ var write = base.WriteAsync(context, message);
+ if (++_currentPendingWrites == _maxPendingWrites)
+ {
+ context.Flush();
+ Reset();
+ }
+
+ return write;
+ }
+
+ void ScheduleFlush(IChannelHandlerContext context)
+ {
+ // Schedule a recurring flush - only fires when there's writable data
+ var time = TimeSpan.FromMilliseconds(_maxPendingMillis);
+ var task = new FlushTask(context, time, this);
+ context.Executor.Schedule(task, time);
+ }
+
+ public void Reset()
+ {
+ _currentPendingWrites = 0;
+ }
+
+ class FlushTask : IRunnable
+ {
+ private readonly IChannelHandlerContext _context;
+ private readonly TimeSpan _interval;
+ private readonly BatchWriter _writer;
+
+ public FlushTask(IChannelHandlerContext context, TimeSpan interval, BatchWriter writer)
+ {
+ _context = context;
+ _interval = interval;
+ _writer = writer;
+ }
+
+ public void Run()
+ {
+ if (_writer.HasPendingWrites)
+ {
+ // execute a flush operation
+ _context.Flush();
+ _writer.Reset();
+ } 
+
+ // channel is still writing
+ if (_context.Channel.Active)
+ { 
+ _context.Executor.Schedule(this, _interval); // reschedule
+ }
+ }
+ }
+ }
+
  /// <summary>
  /// INTERNAL API
  /// 

src/core/Akka.Remote/Transport/DotNetty/DotNettyTransport.cs

@@ -327,6 +327,8 @@ private void SetInitialChannelPipeline(IChannel channel)
  pipeline.AddLast("FrameEncoder", new LengthFieldPrepender(Settings.ByteOrder, 4, 0, false));
  }
  }
+
+ pipeline.AddLast("BatchWriter", new BatchWriter());
  }
 
  private void SetClientPipeline(IChannel channel, Address remoteAddress)

src/core/Akka.Remote/Transport/DotNetty/TcpTransport.cs

@@ -169,10 +169,10 @@ public TcpAssociationHandle(Address localAddress, Address remoteAddress, DotNett
 
  public override bool Write(ByteString payload)
  {
- if (_channel.Open && _channel.IsWritable)
+ if (_channel.Open)
  {
  var data = ToByteBuffer(payload);
- _channel.WriteAndFlushAsync(data);
+ _channel.WriteAsync(data);
  return true;
  }
  return false;