使用.NET简单实现一个Redis的高性能克隆版（三）

译者注

该原文是Ayende Rahien大佬业余自己在使用C# 和 .NET构建一个简单、高性能兼容Redis协议的数据库的经历。

首先这个"Redis"是非常简单的实现，但是他在优化这个简单"Redis"路程很有趣，也能给我们在从事性能优化工作时带来一些启示。

原作者：Ayende Rahien

原链接：https://ayende.com/blog/197473-C/high-performance-net-building-a-redis-clone-architecture

构建Redis克隆版-架构

在之前的文章中，我们尝试用最简单的方式来完成一个Redis克隆版。打开一个套接字来监听，为每个客户端单独分配一个Task来从网络读取数据，解析命名并执行它。虽然在流水线上有一些小的改进，但也只仅此而已。

让我们退一步来构建一个与Redis架构更为接近的Redis克隆版。为此，我们需要在一个线程中完成所有工作。这在C#中是比较难实现的，没有用于执行Redis那样工作类型的API。更确切的来说是有 Socket.Select() 方法，但是需要我们自己在此基础上构建一切（比如我们必须写代码处理缓冲、字符串等等）。

考虑到这是通往最终建议的架构的一个中途站，我决定完全跳过这个。相反，我将首先专注于消除系统中的主要瓶颈，即 ConcurrentDictionary 。

分析器的结果表明，我们这最大的开销就是 ConcurrentDictionary 的可伸缩性。即使我使用了1024个分片的锁，它仍然占用50%的时间开销。问题是，我们能做得更好吗？我们可以尝试一个更好的选择，就是我们不再使用 ConcurrentDictionary ，而是直接使用单独的 Dictionary 来分片，这样的话每个 Dictionary 都不需要并发就可以访问。

我的想法是这样的，我们将为客户端提供常规的读写操作。但是，我们不会直接在I/O上处理这些命令，而是将其路由到一个专用的线程（使用它自己的 Dictionary ）来完成这项工作。因为我是16核的机器，我将创建10个这样的线程（假设它们每个都能分配到1个核心），并且我能够将I/O处理放到其余的6个核心上。

以下是更改后的结果：



请注意，我们现在跑分的数据是125w/s，比上一次几乎增长了25%。

下面是这一次新代码的分析器结果：



因此在本例中，花费了大量的时间来处理各种各样的字符串，等待GC（大约占30%）。集合的成本下降了很多。

还有一些其它的开销出现在我眼前，看看这里：



对于“简单”属性查找来说，这个开销非常惊人。另外 SubString 函数的调用开销也很大，超过整个系统开销的6%。

在研究系统其它部分时，看到了这个：



这真的很有趣，因为我们花了很多的时间在等待队列中是否有新的元素，其实我们可以做更多的事情，而不是就在那干等着。

我还尝试了其它的线程数量，如果只运行一个ExecWorker，我们的运行速度是40w/s，两个线程，我们的运行速度是70w/s。当使用4个专用于处理请求的线程时，我们的运行速度是106w/s。

因此，很明显，我们需要重新考虑这种方案，我们不能够正确地扩展到合适的数值。

注意，这种方法也不利用流水线。我们分别处理每个命令和其他命令。我的下一步是添加对使用这种方法的流水线的支持，并测量这种影响。

从另一方面来说，我们现在的性能还是100w/s，考虑到我只花了很少的时间来实现方案，从这个方案可以获得25w/s的性能提升，这是令人激动人心的。从侧面说，我们还有更多的事情可以做，但我想把重点放在修复我们第一个方案上。

下面是当前的状态，因此您可以与原始代码比较。

using System.Collections.Concurrent;

using System.Net.Sockets;

using System.Threading.Channels;

var listener = new TcpListener(System.Net.IPAddress.Any, 6379);

listener.Start();

var redisClone = new RedisClone();

while (true)

{

    var client = listener.AcceptTcpClient();

    var _ = redisClone.HandleConnection(client); // run async

}

public class RedisClone

{

    ShardedDictionary _state = new(Environment.ProcessorCount / 2);

    public async Task HandleConnection(TcpClient tcp)

    {

        var _ = tcp;

        var stream = tcp.GetStream();

        var client = new Client

        {

            Tcp = tcp,

            Dic = _state,

            Reader = new StreamReader(stream),

            Writer = new StreamWriter(stream)

            {

                NewLine = "\r\n"

            }

        };

        await client.ReadAsync();

    }

}

class Client

{

    public TcpClient Tcp;

    public StreamReader Reader;

    public StreamWriter Writer;

    public string Key;

    public string? Value;

    public ShardedDictionary Dic;

    List<string> Args = new();

    public async Task ReadAsync()

    {

        try

        {

            Args.Clear();

            var lineTask = Reader.ReadLineAsync();

            if (lineTask.IsCompleted == false)

            {

                await Writer.FlushAsync();

            }

            var line = await lineTask;

            if (line == null)

            {

                using (Tcp)

                {

                    return;

                }

            }

            if (line[0] != '*')

                throw new InvalidDataException("Cannot understand arg batch: " + line);

            var argsv = int.Parse(line.Substring(1));

            for (int i = 0; i < argsv; i++)

            {

                line = await Reader.ReadLineAsync();

                if (line == null || line[0] != '$')

                    throw new InvalidDataException("Cannot understand arg length: " + line);

                var argLen = int.Parse(line.Substring(1));

                line = await Reader.ReadLineAsync();

                if (line == null || line.Length != argLen)

                    throw new InvalidDataException("Wrong arg length expected " + argLen + " got: " + line);

                Args.Add(line);

            }

            switch (Args[0])

            {

                case "GET":

                    Key = Args[1];

                    Value = null;

                    break;

                case "SET":

                    Key = Args[1];

                    Value = Args[2];

                    break;

                default:

                    throw new ArgumentOutOfRangeException("Unknown command: " + Args[0]);

            }

            Dic.Run(this);

        }

        catch (Exception e)

        {

            await HandleError(e);

        }

    }

    public async Task NextAsync()

    {

        try

        {

            if (Value == null)

            {

                await Writer.WriteLineAsync("$-1");

            }

            else

            {

                await Writer.WriteLineAsync($"${Value.Length}\r\n{Value}");

            }

            await ReadAsync();

        }

        catch (Exception e)

        {

            await HandleError(e);

        }

    }

    public async Task HandleError(Exception e)

    {

        using (Tcp)

        {

            try

            {

                string? line;

                var errReader = new StringReader(e.ToString());

                while ((line = errReader.ReadLine()) != null)

                {

                    await Writer.WriteAsync("-");

                    await Writer.WriteLineAsync(line);

                }

                await Writer.FlushAsync();

            }

            catch (Exception)

            {

                // nothing we can do

            }

        }

    }

}

class ShardedDictionary

{

    Dictionary<string, string>[] _dics;

    BlockingCollection<Client>[] _workers;

    public ShardedDictionary(int shardingFactor)

    {

        _dics = new Dictionary<string, string>[shardingFactor];

        _workers = new BlockingCollection<Client>[shardingFactor];

        for (int i = 0; i < shardingFactor; i++)

        {

            var dic = new Dictionary<string, string>();

            var worker = new BlockingCollection<Client>();

            _dics[i] = dic;

            _workers[i] = worker;

            // readers

            new Thread(() =>

            {

                ExecWorker(dic, worker);

            })

            {

                IsBackground = true,

            }.Start();

        }

    }

    private static void ExecWorker(Dictionary<string, string> dic, BlockingCollection<Client> worker)

    {

        while (true)

        {

            var client = worker.Take();

            if (client.Value != null)

            {

                dic[client.Key] = client.Value;

                client.Value = null;

            }

            else

            {

                dic.TryGetValue(client.Key, out client.Value);

            }

            var _ = client.NextAsync();

        }

    }

    public void Run(Client c)

    {

        var reader = _workers[c.GetHashCode() % _workers.Length];

        reader.Add(c);

    }

}

公众号

之前一直有朋友让开通公众号，由于一直比较忙没有弄。

现在终于抽空弄好了，译者公众号如下，欢迎大家关注。

系列链接

使用.NET简单实现一个Redis的高性能克隆版（一）

使用.NET简单实现一个Redis的高性能克隆版（二）