DotNetty發送請求的最佳實踐

長鏈接發送request/response時, 絕大部分包都是小包, 而每個小包都要消耗一個IP包, 成本大約是20-30us, 普通千兆網卡的pps大約是60Wpps, 所以想要提高長鏈接密集IO的應用性能, 需要做包的合並, 也稱為了scatter/gather io或者vector io.

 

在linux下有readv/writev就是對應這個需求的, 減少系統調用, 減少pps, 提高網卡的吞吐量. 關於readv提高讀的速度, 可以看看陳碩muduo里面對於readv的使用, 思路是就是在棧上面弄一個64KB的數組, 組成readv的第二塊buffer, 從而盡可能一次性把socket緩沖區的內容全部出來(參見5). 這里不再贅述, 重點描述DotNetty下面怎么做Gathering Write.

 

首先得有一個Channel<IMessage>, 用來做寫的緩沖, 讓業務關心業務, 網絡關心網絡, 否則每個業務都WriteAndFlushAsync, 那是不太可能有合並發送的.

 

然后就是SendingLoop的主循環, 里面不停的從Channel里面TryRead包, 然后WriteAsync, 隔幾個包Flush一次. 類似的思想在Orleans Network里面也存在.

 

public void RunSendLoopAsync(IChannel channel)
{
    var allocator = channel.Allocator;
    var reader = this.queue.Reader;
    Task.Run(async () => 
    {
        while (!this.stop) 
        {
            var more = await reader.WaitToReadAsync();
            if (!more) 
            {
                break;
            }

            IOutboundMessage message = default;
            var number = 0;
            try 
            {
                while (number < 4 && reader.TryRead(out message) && message != null) 
                {
                    Interlocked.Decrement(ref this.queueCount);
                    var msg = message.Inner as IMessage;
                    var buffer = msg.ToByteBuffer(allocator);
                    channel.WriteAsync(buffer);
                    number++;
                }
                channel.Flush();
                number = 0;
            }
            catch (Exception e)  when(message != default)
            {
                logger.LogError("SendOutboundMessage Fail, SessionID:{0}, Exception:{1}",
                    this.sessionID, e.Message);
                this.messageCenter.OnMessageFail(message);
            }
        }
        this.logger.LogInformation("SessionID:{0}, SendingLoop Exit", this.sessionID);
    });
}

第19-27行是關鍵, 這邊每4個包做一下flush, 然后flush會觸發DotNetty的DoWrite:

protected override void DoWrite(ChannelOutboundBuffer input)
{
    List<ArraySegment<byte>> sharedBufferList = null;
    try
    {
        while (true)
        {
            int size = input.Size;
            if (size == 0)
            {
                // All written
                break;
            }
            long writtenBytes = 0;
            bool done = false;

            // Ensure the pending writes are made of ByteBufs only.
            int maxBytesPerGatheringWrite = ((TcpSocketChannelConfig)this.config).GetMaxBytesPerGatheringWrite();
            sharedBufferList = input.GetSharedBufferList(1024, maxBytesPerGatheringWrite);
            int nioBufferCnt = sharedBufferList.Count;
            long expectedWrittenBytes = input.NioBufferSize;
            Socket socket = this.Socket;

            List<ArraySegment<byte>> bufferList = sharedBufferList;
            // Always us nioBuffers() to workaround data-corruption.
            // See https://github.com/netty/netty/issues/2761
            switch (nioBufferCnt)
            {
                case 0:
                    // We have something else beside ByteBuffers to write so fallback to normal writes.
                    base.DoWrite(input);
                    return;
                default:
                    for (int i = this.Configuration.WriteSpinCount - 1; i >= 0; i--)
                    {
                        long localWrittenBytes = socket.Send(bufferList, SocketFlags.None, out SocketError errorCode);
                        if (errorCode != SocketError.Success && errorCode != SocketError.WouldBlock)
                        {
                            throw new SocketException((int)errorCode);
                        }

DotNetty TcpSocketChannel類的DoWrite函數, 19行獲取當前ChannelOutboundBuffer的Segment<byte>數組, 然后在36行調用Socket.Send一次性發出去, 這個是Gathering Write的關鍵. 有了這個, 就可以不在業務層用CompositeByteBuffer.

DotNetty Libuv Transport的實現可以看6, 思想是類似的.

 

實際上Orleans 3.x做的網絡優化, 也有類似的思想:

private async Task ProcessOutgoing()
{
    await Task.Yield();

    Exception error = default;   
    PipeWriter output = default;
    var serializer = this.serviceProvider.GetRequiredService<IMessageSerializer>();
    try
    {
        output = this.Context.Transport.Output;
        var reader = this.outgoingMessages.Reader;
        if (this.Log.IsEnabled(LogLevel.Information))
        {
            this.Log.LogInformation(
                "Starting to process messages from local endpoint {Local} to remote endpoint {Remote}",
                this.LocalEndPoint,
                this.RemoteEndPoint);
        }

        while (true)
        {
            var more = await reader.WaitToReadAsync();
            if (!more)
            {
                break;
            }

            Message message = default;
            try
            {
                while (inflight.Count < inflight.Capacity && reader.TryRead(out message) && this.PrepareMessageForSend(message))
                {
                    inflight.Add(message);
                    var (headerLength, bodyLength) = serializer.Write(ref output, message);
                    MessagingStatisticsGroup.OnMessageSend(this.MessageSentCounter, message, headerLength + bodyLength, headerLength, this.ConnectionDirection);
                }
            }
            catch (Exception exception) when (message != default)
            {
                this.OnMessageSerializationFailure(message, exception);
            }

            var flushResult = await output.FlushAsync();
            if (flushResult.IsCompleted || flushResult.IsCanceled)
            {
                break;
            }

            inflight.Clear();
        }

核心在31行, 開始寫, 43行開始flush, 只不過Orleans用的pipelines io, DotNetty是傳統模型.

 

這樣做, 可以在有限的pps下, 支撐更高的吞吐量.

 

個人感覺DotNetty更好用一些.

 

參考:

1. https://github.com/Azure/DotNetty/blob/dev/src/DotNetty.Transport/Channels/Sockets/TcpSocketChannel.cs#L271-L288

2. https://github.com/dotnet/orleans/blob/master/src/Orleans.Core/Networking/Connection.cs#L282-L294

3. https://docs.microsoft.com/zh-cn/windows/win32/winsock/scatter-gather-i-o-2

4. https://linux.die.net/man/2/writev

5. https://github.com/chenshuo/muduo/blob/d980315dc054b122612f423ee2e1316cb14bd3b5/muduo/net/Buffer.cc#L28-L38

6. https://github.com/Azure/DotNetty/blob/dev/src/DotNetty.Transport.Libuv/Native/WriteRequest.cs#L106-L128