译者注
该原文是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的性能提升,这是令人激动人心的。从侧面说,我们还有更多的事情可以做,但我想把重点放在修复我们第一个方案上。
下面是当前的状态,因此您可以与原始代码比较。
代码语言:javascript复制
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 = "rn"
}
};
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}rn{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);
}
}
公众号
之前一直有朋友让开通公众号,由于一直比较忙没有弄。现在终于抽空弄好了,译者公众号如下,欢迎大家关注。