Ориентированные на многопотоковое исполнение библиотеки кэша для.NET

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

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

Если вы действительно обеспокоены проблемой несовместимости кэша, вы можете использовать существующую инфраструктуру кэша и логически разделить ее на регионы. Затем синхронизировать доступ к каждому региону независимо.

Пример стратегии, если ваш набор данных состоит из элементов, которые имеют числовые идентификаторы, и вы хотите разделить кэш на 10 регионов, вы можете (mod 10) по идентификатору определить, в каком регионе они находятся. Вы будете хранить массив из 10 объектов для блокировки. Весь код может быть написан для переменного числа регионов, которое может быть задано через конфигурацию или определено при запуске приложения в зависимости от общего числа объектов, которые вы прогнозируете/намерены кэшировать.

Если количество просмотров кэша определяется ненормальным образом, вам придется придумать собственную эвристику для разделения кэша.

Обновление (по комментарию): Ну, это было весело. Я думаю, что нижеприведенный вариант - это примерно настолько тонкая блокировка, на которую можно надеяться, не сходя с ума (или поддерживая/синхронизируя словарь блокировок для каждого ключа кэша). Я не тестировал это, так что, возможно, есть ошибки, но идея должна быть наглядной. Отслеживать список запрошенных идентификаторов, а затем использовать его, чтобы решить, нужно ли вам получить элемент самостоятельно, или нужно просто подождать, пока завершится предыдущий запрос. Ожидание (и вставка в кэш) синхронизировано с блокировкой и сигнализацией узкоспециализированных потоков с помощью Wait и PulseAll. Доступ к запрашиваемому списку идентификаторов синхронизируется с помощью тесно скопированного ReaderWriterLockSlim.

Это кэш только для чтения. Если вы выполняете создание/обновление/удаление, вам придется убедиться, что вы удаляете идентификаторы из requestedIds после их получения (перед вызовом Monitor.PulseAll(_cache) вы захотите добавить еще одну try..finally и получить блокировку записи _requestedIdsLock). Кроме того, при создании/обновлении/удалении самым простым способом управления кэшем будет простое удаление существующего элемента из _cache, если/когда основная операция создания/обновления/удаления завершится успешно.

(Упс, см. обновление 2 ниже.)

public class Item 
{
    public int ID { get; set; }
}

public class AsyncCache
{
    protected static readonly Dictionary<int, Item> _externalDataStoreProxy = new Dictionary<int, Item>();

    protected static readonly Dictionary<int, Item> _cache = new Dictionary<int, Item>();

    protected static readonly HashSet<int> _requestedIds = new HashSet<int>();
    protected static readonly ReaderWriterLockSlim _requestedIdsLock = new ReaderWriterLockSlim();

    public Item Get(int id)
    {
        // if item does not exist in cache
        if (!_cache.ContainsKey(id))
        {
            _requestedIdsLock.EnterUpgradeableReadLock();
            try
            {
                // if item was already requested by another thread
                if (_requestedIds.Contains(id))
                {
                    _requestedIdsLock.ExitUpgradeableReadLock();
                    lock (_cache)
                    {
                        while (!_cache.ContainsKey(id))
                            Monitor.Wait(_cache);

                        // once we get here, _cache has our item
                    }
                }
                // else, item has not yet been requested by a thread
                else
                {
                    _requestedIdsLock.EnterWriteLock();
                    try
                    {
                        // record the current request
                        _requestedIds.Add(id);
                        _requestedIdsLock.ExitWriteLock();
                        _requestedIdsLock.ExitUpgradeableReadLock();

                        // get the data from the external resource
                        #region fake implementation - replace with real code
                        var item = _externalDataStoreProxy[id];
                        Thread.Sleep(10000);
                        #endregion

                        lock (_cache)
                        {
                            _cache.Add(id, item);
                            Monitor.PulseAll(_cache);
                        }
                    }
                    finally
                    {
                        // let go of any held locks
                        if (_requestedIdsLock.IsWriteLockHeld)
                            _requestedIdsLock.ExitWriteLock();
                    }
                }
            }
            finally
            {
                // let go of any held locks
                if (_requestedIdsLock.IsUpgradeableReadLockHeld)
                    _requestedIdsLock.ExitReadLock();
            }
        }

        return _cache[id];
    }

    public Collection<Item> Get(Collection<int> ids)
    {
        var notInCache = ids.Except(_cache.Keys);

        // if some items don't exist in cache
        if (notInCache.Count() > 0)
        {
            _requestedIdsLock.EnterUpgradeableReadLock();
            try
            {
                var needToGet = notInCache.Except(_requestedIds);

                // if any items have not yet been requested by other threads
                if (needToGet.Count() > 0)
                {
                    _requestedIdsLock.EnterWriteLock();
                    try
                    {
                        // record the current request
                        foreach (var id in ids)
                            _requestedIds.Add(id);

                        _requestedIdsLock.ExitWriteLock();
                        _requestedIdsLock.ExitUpgradeableReadLock();

                        // get the data from the external resource
                        #region fake implementation - replace with real code
                        var data = new Collection<Item>();
                        foreach (var id in needToGet)
                        {
                            var item = _externalDataStoreProxy[id];
                            data.Add(item);
                        }
                        Thread.Sleep(10000);
                        #endregion

                        lock (_cache)
                        {
                            foreach (var item in data)
                                _cache.Add(item.ID, item);

                            Monitor.PulseAll(_cache);
                        }
                    }
                    finally
                    {
                        // let go of any held locks
                        if (_requestedIdsLock.IsWriteLockHeld)
                            _requestedIdsLock.ExitWriteLock();
                    }
                }

                if (requestedIdsLock.IsUpgradeableReadLockHeld)
                    _requestedIdsLock.ExitUpgradeableReadLock();

                var waitingFor = notInCache.Except(needToGet);
                // if any remaining items were already requested by other threads
                if (waitingFor.Count() > 0)
                {
                    lock (_cache)
                    {
                        while (waitingFor.Count() > 0)
                        {
                            Monitor.Wait(_cache);
                            waitingFor = waitingFor.Except(_cache.Keys);
                        }

                        // once we get here, _cache has all our items
                    }
                }
            }
            finally
            {
                // let go of any held locks
                if (_requestedIdsLock.IsUpgradeableReadLockHeld)
                    _requestedIdsLock.ExitReadLock();
            }
        }

        return new Collection<Item>(ids.Select(id => _cache[id]).ToList());
    }
}

Обновление 2:

Я неправильно понял поведение UpgradeableReadLock... только один поток в одно время может держать UpgradeableReadLock. Поэтому вышеприведенный пример должен быть доработан, чтобы изначально захватывать только блокировки чтения, а при добавлении элементов в _requestedIds полностью отказываться от них и приобретать полноценную блокировку записи.

Наконец, благодаря диалогу в комментариях, мы нашли работоспособное решение этой проблемы. Я создал оболочку, которая представляет собой частично реализованный абстрактный базовый класс, который использует любую стандартную библиотеку кеша в качестве резервного кеша (просто необходимо реализовать Contains , Get , Положите методы и Удалить ). В настоящий момент я использую для этого блок приложения EntLib Caching Application, и потребовалось время, чтобы его запустить и запустить, потому что некоторые аспекты этой библиотеки ... ну ... не настолько хорошо продуманы.

В любом случае, общий код сейчас близок к 1 КБ строк, поэтому я не собираюсь публиковать здесь все, но основная идея такова:

1. Перехватить все вызовы к Get , Положить / добавить и Удалить методы.

2. Вместо добавления исходного элемента добавьте элемент «запись», который содержит ManualResetEvent в дополнение к свойству Value . Согласно некоторым советам, данным мне по ранее заданному сегодня вопросу, запись реализует защелку обратного отсчета, которая увеличивается при каждом получении записи и уменьшается при ее освобождении. И загрузчик, и все последующие операции поиска участвуют в защелке обратного отсчета, поэтому, когда счетчик достигает нуля, данные гарантированно доступны, а событие ManualResetEvent уничтожается в целях экономии ресурсов.

3. Когда требуется отложенная загрузка записи, она сразу же создается и добавляется в резервный кэш, при этом событие находится в несигнальном состоянии. Последующие вызовы либо нового метода GetOrAdd , либо перехваченных методов Get найдут эту запись и либо дождутся события (если событие существует), либо немедленно вернут связанное значение (если событие не существует).

4. Метод Put добавляет запись без события; они выглядят так же, как записи, для которых отложенная загрузка уже завершена.

5. Поскольку GetOrAdd по-прежнему реализует Get , за которым следует необязательный Put ], этот метод синхронизируется (сериализуется) с Put и Удалить методы , но только для добавления неполной записи, а не на все время отложенной загрузки. Методы Get не сериализованы; фактически весь интерфейс работает как автоматическая блокировка чтения и записи.

Это все еще незавершенная работа, но я прогнал ее через дюжину юнит-тестов, и, похоже, она работает. Он работает правильно для обоих сценариев, описанных в вопросе. Другими словами:

• Вызов длительной отложенной загрузки ( GetOrAdd ) для ключа X (моделируется с помощью Thread.Sleep ), который занимает 10 секунд, за которым следует еще один GetOrAdd для того же ключа X в другом потоке ровно через 9 секунд, в результате оба потока получают правильные данные одновременно (10 секунд от T ₀ ). Нагрузки не дублируются.

• Немедленная загрузка значения для ключа X , затем запуск длительной отложенной загрузки для ключа Y , затем запрос ключа X в другом потоке ( до завершения Y ) немедленно возвращает значение для X . Блокирующие вызовы привязаны к соответствующему ключу.

Это также дает то, что я считаю наиболее интуитивно понятным, когда вы начинаете отложенную загрузку, а затем немедленно удаляете ключ из кеша; поток, который первоначально запросил значение, получит реальное значение, но любые другие потоки, запрашивающие тот же ключ в любое время после удаления, ничего не получат обратно ( null ) и немедленно вернутся.

В целом я очень доволен этим. Я все еще хотел бы, чтобы была библиотека, которая делала бы это для меня, но я полагаю, если вы хотите, чтобы что-то было сделано правильно ... ну, вы знаете.