Асинхронное ожидание изменений в подмножестве строк в большой таблице в памяти
ТЛ;ДР:
- У меня есть база данных в памяти, которая поддерживает большое (10–50 миллионов) количество строк.
- Мне нужен эффективный способ, с помощью которого наблюдатель мог бы начать прослушивать любое их подмножество (обычно около 10 000 элементов) и асинхронно ждать, пока какой-либо из них не изменится.
- Система должна быть готова к одновременной активности не менее 50 000 таких наблюдателей.
- Предположим, что будет записываться/обновляться около 1000–10 000 строк в секунду.
Каким будет эффективный способ реализовать этот шаблон на C#?
Чуть подробнее:
В качестве мысленного примера предположим, что мы проектируем базу данных в памяти. Наблюдатели должны иметь возможность ожидать изменений в любом подмножестве строк, которые им интересны. Итак, ради этого вопроса интерфейс может выглядеть так просто, как
interface ITable<TPrimaryKey, TRow>
{
Task WaitUntilAnyRowChanged(IReadOnlySet<TPrimaryKey> keysToListenTo, CancellationToken ct);
}
Пока что мне на ум пришло следующее:
Идея 1: TaskCompletionSource для строк
Свяжите TaskCompletionSource с каждой строкой. При изменении значения система дополняет существующую TaskCompletionSource, связанную с этой строкой, и заменяет ее новой для следующего ожидающего обновления. Тогда потребители (т. е. WaitUntilAnyRowChanged) смогут просто Task.WhenAny(...) на все соответствующие товары.
Недостатки:
- Я бы ожидал, что цепочка
Task.WhenAny(...)из тысячTaskCompletionSourceбудет очень неэффективной. - Не самая большая проблема, но требует относительно большого объема ассигнований.
Идея 2: AsyncAutoResetEvent (или аналогичные структуры синхронизации) для наблюдателей
Меняемся ролями. Каждая строка содержит List<AsyncAutoResetEvent>. Каждый наблюдатель создаст свой собственный AsyncAutoResetEvent и зарегистрирует его во всех интересующих его строках. При записи строка установит сигнал для всех своих слушателей.
Недостатки:
- Это потенциально может значительно замедлить скорость записи в базу данных; то, что в противном случае могло бы быть простой записью в словарь, теперь становится записью + циклом потенциально тысяч наблюдателей, вызывающим их AsyncAutoResentEvent
Идея 3: Наблюдатели следят за всеми изменениями
Наблюдатели могли слушать поток, нажимающий клавиши для любых изменений. Наблюдатели будут нести ответственность за его фильтрацию до интересующего их подмножества ключей и, возможно, за установку локального TaskCompletionSource или чего-то подобного.
Плюсы:
- Просто реализовать
- Относительно предсказуемая производительность (можно выделить x потоков для пересылки этих событий наблюдателям)
Минусы:
- Совершенно неэффективно, если наблюдатель заботится только о небольшом подмножестве строк. Представьте себе, что вы слушаете только одну строку в таблице из 50 миллионов строк.
Идея 4: Опрос
Каждая строка содержит Box<long>, значение которого содержит отметку текущей версии строки.
Наблюдатели периодически опрашивают все интересующие их строки, сохраняют последнюю известную версию и проверяют, обновилась ли какая-либо версия с момента последнего опроса.
Плюсы:
- Простой
- Похоже, что он может хорошо масштабироваться: скорость записи не пострадает. И для того, чтобы наблюдатель перебирал ~ 10 тыс. коробочных длинных значений для чтения и сравнения их значений, он должен быть достаточно быстрым.
- Практически бесплатно GC
Недостатки:
- Никто не любит опросы! Обнаружение изменений задерживается на интервал опроса, а не всегда обнаруживается немедленно.
- Может не так хорошо масштабироваться с увеличением количества наблюдаемых ключей.
В целом я склоняюсь к идее 4 как к наиболее осуществимому на данный момент подходу. Но у меня есть, возможно, иррациональная ненависть к опросам общественного мнения..
Поэтому мне было бы интересно услышать мысли: может ли кто-нибудь придумать лучшее решение? Такое ощущение, что это не такая уж нишевая проблема. Может быть, есть какой-то стандартный подход к этому типу проблем, о котором я не знаю?
Ответы 1
Это интересная проблема. Моя первая мысль — поддерживать статический словарь с подписками:
private static readonly Dictionary<TPrimaryKey, HashSet<Observer>> s_subscriptions;
Observer происходит от TaskCompletionSource:
class Observer : TaskCompletionSource
{
private readonly TPrimaryKey[] _keys;
public void Subscribe();
public void Complete();
}
Наблюдатель подписывается путем добавления всех наблюдаемых им ключей в словарь s_subscriptions:
public void Subscribe()
{
foreach (TPrimaryKey key in _keys)
{
s_subscriptions.GetOrAdd(key, () => new HashSet<Observer>()).Add(this);
}
}
Когда значение меняется, вы переходите к словарю и Complete всем наблюдателям, которые наблюдают за этим ключом:
if (_subscriptions.TryGetValue(key, out List<Observer> observers))
{
foreach (Observer observer in observers)
{
observer.Complete();
}
}
Когда наблюдатель завершен, он удаляется из s_subscriptions:
public void Complete()
{
foreach (TPrimaryKey key in _keys)
{
s_subscriptions[key].Remove(this);
}
base.SetResult(); // Completes the base.Task
}
Возможно, вам придется создать экземпляр каждого Observer с помощью параметра TaskCreationOptions.RunContinuationsAsynchronous, чтобы он завершался в ThreadPool, а не в том же потоке, который изменяет значение.
Чтобы свести к минимуму нагрузку на сборщик мусора, вы можете рассмотреть возможность использования ValueTask вместо TaskCompletionSource, подкрепленных многоразовыми реализациями IValueTaskSource. Но это может быть слишком много работы и слишком мало пользы.