Шардированный кэш на базе Memcached

Привет! Меня зовут Андрей Барболин, я Senior Software Engineer в команде Order Management System. Сегодня я расскажу вам, как мы сделали шардированный кэш и под стресс-тестами добились 30 миллионов операций в секунду, а также про первую open source библиотеку от AliExpress Россия.

Вводные

• Необходимо держать в кэше 200-300GB данных;

• Целевая нагрузка на сервис 30 миллионов key values в секунду.

И это профиль нагрузки только одного из сервисов. Также бывают периоды распродаж, когда нагрузка может увеличиваться в несколько раз. Поэтому нам нужно уметь горизонтально масштабировать инстансы кэшей.

Классическое решение этой проблемы — шардирование на клиенте. Что будет стоять в плане кэша на конечной машине, уже не так важно. Это может быть Redis, Memcached или любое другое решение.

Подбор решения

Для того чтобы выбрать конкретное решение по кэшированию, нужно знать как минимум предполагаемую нагрузку и объем данных. Если у вас нет тысяч RPS, а объем данных небольшой, стоит выбрать зрелое решение с большим комьюнити и низким уровнем входа. На данный момент это Redis. Тем более, его можно рассматривать как этакий швейцарский нож, ведь это уже не просто кэш. Даже если вам не нужна его богатая функциональность прямо сейчас, он дает пространство для дальнейшего развития проекта.

Дальше нужно смотреть на конкретную задачу. Нам в AliExpress необходимо держать в кэше достаточно много данных и использовать CPU по максимуму. Redis однопоточный — многопоточку завезли только в 6-й версии в мае 2020, да и то только на сетевой I/O, то есть на чтение из сокетов и запись в них. Стоит посмотреть в сторону других решений. Энтузиасты форкнули Redis, сделали его многопоточным, и так появился KeyDb. Еще есть Dragonfly, который активно развивается и ругает Redis в своем блоге за слабый фундамент на высоких нагрузках. Тут стоит подумать, насколько вы готовы использовать менее зрелые технологии в продакшене: возможно, придется постоянно держать руку на пульсе и отслеживать, что пофиксили и завезли в новых релизах.

Мы выбрали Memcached, потому что он проверен годами, максимально предсказуем и надежен. Это топор, который просто кэш, и всё. К тому же, даже новомодный Dragonfly по их же бенчмаркам не смог полностью обойти Memcached:

Из-за своей простоты Memcached:

• Лучше использует оперативную память;

• Изначально использует многопоточный подход, который позволяет выжать максимум из одной машины.

Это накладывает определенные ограничения на использование Memcached:

• 250 bytes на ключ. Это никак не правится в настройках. Чтобы это изменить, нужно будет пересобирать Memcached;

• 1MB на значение. Эта величина настраивается;

• Eviction policy только LRU (Last recently used).

Ограничения обоснованы изначальной архитектурой Memcached с slub-классами и LRU. Подробнее можно почитать в их Wiki.

Шардирование на клиенте

Итак, мы взяли максимально простой инструмент для кэша. Пора научиться его готовить и шардировать. Мы можем взять уже готовое решение от Facebook, одного из самых известных пользователей Memcached. Они написали к нему mc router, у которого широкая функциональность — на нем можно сделать вообще любой роутинг, какой пожелаешь. Но все-таки это еще один черный ящик на пути к распределенному кэшу. А нам нужен максимально простой, управляемый и тонкий клиент.

На входе используем consistent hashing aka HashRing. Это подход позволит нам добавлять и удалять инстансы Memcached в реальном времени без редеплоя сервиса, который его использует.

Представим, что A, B, C — инстансы Memcached. Мы можем высчитать их расположение на круге, например, вычислением hash по IP-адресам подов. Дальше мы вычисляем hash по ключу, находим ближайший инстанс по часовой стрелке (или против) и отправляем запрос на него:

Если один из подов откажет, мы должны убрать его из HashRing. Но теперь у нас есть перекос по нагрузке — все значения, которые должны были идти на инстанс B, теперь уходят на инстанс C:

Нам нужно, чтобы нагрузка распределилась максимально равномерно между инстансами A и C. Для этого необходимо добавить на круг «виртуальные» инстансы, которые будут ассоциированы с физическими. Таким образом, если один из инстансов откажет, вся нагрузка должна будет равномерно распределиться между остальными инстансами:

Такой подход позволит нам делать запросы на несколько инстансов одновременно, что Redis из коробки не умеет:

Redis Cluster supports multiple key operations as long as all of the keys involved in a single command execution (or whole transaction, or Lua script execution) belong to the same hash slot. The user can force multiple keys to be part of the same hash slot by using a feature called hash tags.

То есть придется поупражняться, чтобы запустить команду на несколько ключей и не получить:

(error) ERR CROSSSLOT Keys in request don't hash to the same slot

Hash tags используются для объединения ключей по некоему тэгу, по которому будет считаться hash, чтобы отправить ключи в определенный hash slot (пример из документации):

MSET {user:1000}.name Angela {user:1000}.surname White

Возможно, сработает пайплайнинг нескольких команд get/set, но я сходу не нашел примеров без hash tag’ов. Здесь говорится, что:

Redis Enterprise has a few workarounds for simple commands, notably MGET and MSET.

Примеров или документации я найти не смог. Но даже если эти обходные пути существуют, Redis Enterprise все еще стоит денюжку.

Если знаете примеры запросов на несколько ключей в несколько инстансов Redis, напишите в комментарии.

Немного кода по HashRing

С помощью бинарного поиска добиваемся O(log(n)):

private TNode GetNodeInternal(string key)
{
    var keyHash = GetHash(key);

    var index = Array.BinarySearch(_sortedNodeHashKeys, keyHash);
    if (index < 0) // no exact match
    {
        // If the Array does not contain the specified value, the method returns a negative integer.
        // You can apply the bitwise complement operator to the negative result to produce an index.
        // If this index is one greater than the upper bound of the array, there are no elements larger than value in the array.
        // Otherwise, it is the index of the first element that is larger than value.
        index = ~index;

        if (index >= _sortedNodeHashKeys.Length)
        {
            index = 0;
        }
    }

    var hashNodeKey = _sortedNodeHashKeys[index];

    return _hashToNodeMap[hashNodeKey];
}

Выполняем поиск нод параллельно:

public IDictionary<TNode, ConcurrentBag<string>> GetNodes(IEnumerable<string> keys)
{
    var result = new ConcurrentDictionary<TNode, ConcurrentBag<string>>(Comparer);

    try
    {
        _locker.EnterReadLock();
        
        if (_sortedNodeHashKeys == null || _sortedNodeHashKeys.Length == 0)
        {
            return result;
        }

        Parallel.ForEach(keys, new ParallelOptions { MaxDegreeOfParallelism = 16 },key =>
        {
            var node = GetNodeInternal(key);

            var bag = result.GetOrAdd(node, (Func<TNode, ConcurrentBag<string>>) ValueFactory);
            bag.Add(key);
        });
    }
    finally
    {
        _locker.ExitReadLock();
    }

    return result;
}

Бенчмарки при 256 виртуальных нодах на одну физическую:

BenchmarkDotNet=v0.13.1, OS=macOS Monterey 12.3.1 (21E258)

[Darwin 21.4.0]
Apple M1, 1 CPU, 8 logical and 8 physical cores
.NET SDK=6.0.400

[Host]
: .NET 6.0.8 (6.0.822.36306), Arm64 RyuJIT DefaultJob
: .NET 6.0.8 (6.0.822.36306), Arm64 RyuJIT

Method	KeysNumber	NodesNumber	Mean	Error	StdDev	Median
GetNodes	1	1	5.683 us	0.5613 us	1.655 us	4.692 us
GetNodes	1	16	5.485 us	0.5255 us	1.549 us	4.639 us
GetNodes	1	32	6.239 us	0.7027 us	2.072 us	5.060 us
GetNodes	128	1	33.824 us	2.7571 us	8.086 us	29.377 us
GetNodes	128	16	118.482 us	7.0747 us	20.860 us	114.546 us
GetNodes	128	32	188.920 us	12.1676 us	35.877 us	181.387 us
GetNodes	512	1	73.147 us	4.7427 us	13.984 us	66.924 us
GetNodes	512	16	175.545 us	10.4275 us	30.746 us	168.805 us
GetNodes	512	32	293.666 us	13.3944 us	39.494 us	277.418 us
GetNodes	2048	1	193.951 us	8.5778 us	24.749 us	189.355 us
GetNodes	2048	16	326.530 us	15.2840 us	44.825 us	309.335 us
GetNodes	2048	32	466.940 us	18.1174 us	52.849 us	456.591 us
GetNodes	5000	1	427.750 us	16.5527 us	48.806 us	420.915 us
GetNodes	5000	16	574.372 us	25.4257 us	74.569 us	564.302 us
GetNodes	5000	32	688.616 us	26.3884 us	76.558 us	663.938 us
GetNodes	10000	1	814.684 us	27.5884 us	80.039 us	807.244 us
GetNodes	10000	16	1,020.214 us	36.8499 us	108.074 us	1,021.344 us
GetNodes	10000	32	1,269.259 us	35.2069 us	103.256 us	1,288.021 us
GetNodes	20000	1	1,617.165 us	44.5917 us	131.480 us	1,629.595 us
GetNodes	20000	16	1,899.443 us	63.8206 us	188.176 us	1,828.317 us
GetNodes	20000	32	2,059.760 us	60.0584 us	174.240 us	2,015.047 us

1 us : 1 Microsecond (0.000001 sec)

Изначально в качестве hash-алгоритма использовался MurMurHash3. Затем мы перешли на xxHash, что дало четырехкратный выигрыш в скорости. У xxHash есть реализация на C# без дополнительных аллокаций. В своей библиотеке мы тоже встали на путь zero allocation и активно используем ArrayPool и Span'ы. Но нам еще есть над чем поработать.

Почитать про Array Pool.

Почитать и посмотреть про Span'ы.

Распределение нод на HashRing

Добиться идеального распределения нод невозможно в силу самого подхода. Мы можем добавить достаточное количество виртуальных нод, чтобы разница находилась в пределах 1–2% по нагрузке.

Нагрузочное тестирование при 64 виртуальных нодах показало до 5% разницы в нагрузке. Проводим тесты в консоли и получаем оптимальный результат при 256 виртуальных нодах со средним отклонением в 1–2% между самой нагруженной и минимально нагруженной нодами:

var keysNumber = 2000000;
var nodesNumber = new[] {1, 2, 4, 8, 16, 32};
var virtualNodesNumber = new[] {16, 32, 64, 128, 256, 512};

foreach (var nodeNumber in nodesNumber)
{
    foreach (var virtualNodeNumber in virtualNodesNumber)
    {
        var hashRing = new HashRing<Pod>(new HashCalculator(), virtualNodeNumber);
        
        var pods = Enumerable.Range(0, nodeNumber).Select(n => new Pod
        {
            IpAddress = Guid.NewGuid().ToString()
        });
        
        hashRing.AddNodes(pods);
        var keys = Enumerable.Range(0, keysNumber).Select(n => Guid.NewGuid().ToString()).ToArray();
        
        var nodes = hashRing.GetNodes(keys);
        
        Console.WriteLine($"Nodes number: {nodeNumber}, Virtual nodes number: {virtualNodeNumber}");
        var percentages = new List<decimal>();
        foreach (var node in nodes)
        {
            percentages.Add((decimal) node.Value.Count / keysNumber);
        }

        var max = percentages.Max();
        var min = percentages.Min();
        var diff = max - min;
        
        Console.WriteLine($"Max: {max}, Min: {min}, Diff: {diff}");
    }
}

Headless service

Возникает вопрос, как нам в реальном времени получать все активные инстансы Memcached. Мы деплоимся в Kubernetes и можем воспользоваться теми базовыми решениями, которые он предлагает. Подробнее можно почитать тут.

IP-адрес для сервиса не аллоцируется. Используя селекторы, сервис знает про все поды, которые задеплоены под ним. Сделав DNS lookup на сервис, мы можем узнать адреса всех инстансов Memcached:

// HeadlessServiceAddress: my-memcached-headless.<namespace>.svc.cluster.local

IPAddress[] ipAddresses = Dns.GetHostAddresses(_config.HeadlessServiceAddress);
return ipAddresses.Select(i => new Pod
{
    IpAddress = i.ToString()
});

Socket pool

Ни один похожий клиент не может обойтись без пулинга подключений. Концепция тоже базовая — нам нужно открыть n подключений в зависимости от нагрузки и переиспользовать их:

Делаем простую реализацию через семафор и ConcurrentStack:

private readonly ConcurrentStack<PooledSocket> _availableSockets;
...
  
public SocketPool(MemcachedConfiguration.SocketPoolConfiguration config, ILogger logger)
{
  ...
  _semaphore = new SemaphoreSlim(_config.MaxPoolSize, _config.MaxPoolSize);
  _availableSockets = new ConcurrentStack<PooledSocket>();
}
...
  
if (!await _semaphore.WaitAsync(_config.SocketPoolingTimeout, token))
{
  _logger.LogWarning("Pool is run out of sockets");
  return result;
}

// Get available socket
_availableSockets.TryPop(out var pooledSocket)
  
// or create one
...

Поддержание актуального состояния

Для этого запускается фоновый процесс, который занимается обслуживанием HashRing и Socket Pool. Раз в n секунд этот процесс запрашивает все доступные инстансы через headless service, добавляет новые ноды в HashRing и удаляет те, что пропали.

Другой процесс отвечает за опрос текущих инстансов на предмет того, живы ли они, и как только инстанс перестает отвечать, помечает его как умерший. После этого инстанс выбрасывается из HashRing до момента, когда он снова оживет.

Еще один процесс постепенно уничтожает по n сокетов из socket pool’a, чтобы иметь возможность после спада нагрузки убрать лишние.

Итоговая схема

1. Приходит запрос с n key values. Вычисляем по всем ключам hash и по ним выбираем ноды из HashRing

2. Для каждой ноды создается свой Socket Pool, если он еще не создан. Если создан, берем уже существующий

3. В каждом пуле берем доступный сокет из уже созданных либо создаем новый, если пул не переполнен

4. В каждый сокет пишем пайплайном несколько команд get или set для Memcached. Протокол Memcached позволяет сделать несколько операций за раз только таким образом

5. Вычитываем последовательно из сокета ответ от Memcached и отдаем на клиент

Протокол общения с Memcached и наша библиотека

Если посмотреть доступные библиотеки под .NET Core, становится немного грустно. Есть портированная с .NET Framework на .NET Core библиотека с очень старым стилем кодирования и кучей ненужных абстракций. Но если нужно сразу начать использовать Memcached, вполне можно ею воспользоваться.

Мы же взяли ее за основу, чтобы не писать весь протокол общения с нуля. Затюнили ее под динамическое изменение количества инстансов Memcached, что-то переписали и оптимизировали. Если хочется разобраться с нуля, то вам сюда: Memcached binary protocol.

Мы выкатили библиотеку в open source. Ее можно настраивать как через headless service, так и указав напрямую IP-адреса Memcached. То есть библиотека умеет работать с любым количеством инстансов.

Профили нагрузки

Ставим один инстанс Memcached и нагружаем его на SET. Ключ и значение — гуиды, то есть объем данных в несколько байт. Прогон в 21.5k RPS на каждый из кластеров (у нас их три), общая нагрузка примерно 65k RPS, 1 ключ-значение на запрос.

На уровне Memcached видим ровные графики по ресурсам — занимаем меньше 1 CPU. Смотрим графики на одном из кластеров:

Берем 20 ключей на запрос, то есть нагрузка на Memcached вырастает в 20 раз. На инстанс Memcached приходится около 18k RPS * 20 key values = 360k операций SET в секунду до того момента, как начинает расти RT. На графиках видим, что начинаем использовать в разы больше CPU:

Увеличиваем количество ключей до 50. На инстанс Memcached приходится около 10k RPS * 50 key values = 500k операций SET в секунду до того момента, как начинает расти RT. Потребление CPU растет незначительно, но можем наблюдать, что RT выросло гораздо сильнее по сравнению со случаем в 20 key values:

Приходим к выводу, что количество операций в пайплайне команды однозначно влияет и на CPU, и на RT. Лучше придерживаться адекватных цифр и в одной команде посылать ~10 операций на Memcached. Иначе время отклика начнет расти в геометрической прогрессии из-за дополнительного времени, которое требуется Memcached для сброса данных в сокет.

Также стоит заметить, что здесь мы попадаем в один и тот же slab-класс, так как размер значений всегда одинаковый. Внутри Memcached каждый slab-класс обслуживает только один поток, поэтому здесь мы можем упереться в ограничение. Так что ~10 операций на команду — это базовая рекомендация, которую нужно проверять именно на вашем профиле нагрузки.

А теперь берем 8 инстансов Memcached в каждом кластере, 100 key values на запрос и наблюдаем, что легко держим нагрузку, которая равномерно распределяется между инстансами. На каждый инстанс приходится примерно по 12–13 key values на запрос:

Помещаемся в 1–2 CPU на инстанс Memcached:

Рекомендуем брать по 10–20 операций SET на команду и 2 CPU на инстанс Memcached. При этом пропускная способность операций GET может быть примерно в два раза больше, чем у операций SET, так как они легче. Но все равно лучше ограничивать количество операций в пайплайне, чтобы меньше грузить Memcached и не увеличивать время отклика. Если текущее количество инстансов не вывозит нагрузку, добавляем еще инстансов.

Профиль нагрузки одного из сервисов. 6 инстансов Memcached, ожидаем прирост нагрузки. Сейчас имеем около 2k RPS в среднем по 15–20 key values на запрос:

На стороне Memcached получаем 40k RPS:

Количество команд = RPS сервиса * количество инстансов Memcached. RT составляет 0,5ms:

Большая часть метрик экспортируется из Memcached в Прометей. Данные с последней картинки пишет наша клиентская библиотека.

Самый нагруженный вариант. Мы взяли 40 инстансов Memcached с 1 CPU и добили нагрузку до 30 миллионов key values в секунду. 2,5KB каждое значение, 3 кластера, получается ~24GB/s сетевого трафика на один кластер и 72GB/s всего:

При самой высокой нагрузке держимся в районе 1ms на одну операцию в Memcached:

Тут мы разбили каждый входящий запрос на пачки. Нам нужно за один запрос получить одновременно 3000 key values. Мы не можем запихнуть всё сразу в один пайплайн — нужно держать примерно 10–20 операций на пайплайн. Без разбития на каждую ноду приходилось бы 3000 / 40 = 75 key values. Разбиваем их примерно на 4 разные части и отправляем параллельно, чтобы избежать роста RT.

Настройка Memcached

При запуске стоит обратить внимание на параметры. Полезно по ним пройтись и посмотреть, что из этого вам понадобится подтюнить. Как минимум, проверьте следующие параметры:

# MaxMemoryLimit, this should be less than the resources.limits.memory, or memcached will crash.  Default is 64MB
- -m 8192 
# Specify the maximum number of simultaneous connections to the memcached service. The default is 1024. 
- -c 20000

Память сразу стоит поднять — редко можно влезть в дефолтное значение. А количество подключений надо рассчитывать так, чтобы хватило на редеплой сервисов, которые используют Memcached. Дело в том, что в момент редеплоя количество подключений подскочит.

Из неочевидного есть, например, такой параметр:

- -R 40
# The command-line parameter -R is in charge of the maximum number of requests per 
# network IO event (default value is 20). The application should adopt its batch 
# size according to this parameter. Please note that the requests limit does not 
# affect multi-key reads, or the number of keys per get request.

И если неаккуратно пользоваться пайплайнингом операций, то можно нарваться на такую вот неприятную стату:

STAT conn_yields 126672162
Number of times any connection yielded to another due to hitting the -R limit

Итог

• Имеем горизонтально масштабируемый кэш, который можем развернуть для любого сервиса

• Умеем динамически добавлять и убирать инстансы Memcached

• Путем нагрузочных тестов выведено оптимальное количество ресурсов на один инстанс: 2.5 CPU и 8GB RAM. Больше CPU брать нет смысла из-за специфики работы LRU — лучше развернуть дополнительный инстанс. Как только текущее количество инстансов перестает справляться с нагрузкой, накидываем еще. Выбор 8GB RAM продиктован тем, что часть данных не жалко потерять

• 0.5 миллисекунды RT из сервиса в Memcached при нагрузке с умеренным количеством ключей на команду;

• Выкатили библиотеку в open source.