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

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

@@ -8,14 +8,17 @@
 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
 {
+
  /// <summary>
  /// INTERNAL API
  /// 

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

@@ -0,0 +1,104 @@
+//-----------------------------------------------------------------------
+// <copyright file="BatchWriter.cs" company="Akka.NET Project">
+// Copyright (C) 2009-2019 Lightbend Inc. <http://www.lightbend.com>
+// Copyright (C) 2013-2019 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System;
+using System.Threading.Tasks;
+using DotNetty.Buffers;
+using DotNetty.Common.Concurrency;
+using DotNetty.Transport.Channels;
+
+namespace Akka.Remote.Transport.DotNetty
+{
+ /// <summary>
+ /// INTERNAL API.
+ /// 
+ /// Responsible for batching socket writes together into fewer sys calls to the socket.
+ /// </summary>
+ internal class BatchWriter : ChannelHandlerAdapter
+ {
+ private readonly int _maxPendingWrites;
+ private readonly int _maxPendingMillis;
+ private readonly int _maxPendingBytes;
+
+ public BatchWriter(int maxPendingWrites = 20, int maxPendingMillis = 40, int maxPendingBytes = 128000)
+ {
+ _maxPendingWrites = maxPendingWrites;
+ _maxPendingMillis = maxPendingMillis;
+ _maxPendingBytes = maxPendingBytes;
+ }
+
+ private int _currentPendingWrites = 0;
+ private long _currentPendingBytes;
+
+ 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);
+ _currentPendingBytes += ((IByteBuffer)message).ReadableBytes;
+ _currentPendingWrites++;
+ if (_currentPendingWrites >= _maxPendingWrites
+ || _currentPendingBytes >= _maxPendingBytes)
+ {
+ 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;
+ _currentPendingBytes = 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
+ }
+ }
+ }
+ }
+}

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

@@ -240,6 +240,8 @@ public override async Task<bool> Shutdown()
  var tasks = new List<Task>();
  foreach (var channel in ConnectionGroup)
  {
+ // flush any pending writes first
+ channel.Flush();
  tasks.Add(channel.CloseAsync());
  }
  var all = Task.WhenAll(tasks);
@@ -327,6 +329,8 @@ private void SetInitialChannelPipeline(IChannel channel)
  pipeline.AddLast("FrameEncoder", new LengthFieldPrepender(Settings.ByteOrder, 4, 0, false));
  }
  }
+
+ pipeline.AddLast("BatchWriter", new BatchWriter(maxPendingBytes:Settings.MaxFrameSize));
  }
 
  private void SetClientPipeline(IChannel channel, Address remoteAddress)

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

@@ -8,6 +8,7 @@
 using System;
 using System.Net;
 using System.Net.Sockets;
+using System.Runtime.CompilerServices;
 using System.Threading.Tasks;
 using Akka.Actor;
 using Akka.Configuration;
@@ -147,6 +148,7 @@ public override void ChannelActive(IChannelHandlerContext context)
  {
  InitOutbound(context.Channel, (IPEndPoint)context.Channel.RemoteAddress, null);
  base.ChannelActive(context);
+
  }
 
  private void InitOutbound(IChannel channel, IPEndPoint socketAddress, object msg)
@@ -169,16 +171,17 @@ 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);
+ var data = ToByteBuffer(_channel, payload);
+ _channel.WriteAsync(data);
  return true;
  }
  return false;
  }
 
- private IByteBuffer ToByteBuffer(ByteString payload)
+ [MethodImpl(MethodImplOptions.AggressiveInlining)]
+ private static IByteBuffer ToByteBuffer(IChannel channel, ByteString payload)
  {
  //TODO: optimize DotNetty byte buffer usage 
  // (maybe custom IByteBuffer working directly on ByteString?)
@@ -188,6 +191,7 @@ private IByteBuffer ToByteBuffer(ByteString payload)
 
  public override void Disassociate()
  {
+ _channel.Flush(); // flush before we close
  _channel.CloseAsync();
  }
  }