Не удается безопасно заблокировать значение ConcurrentDictionary

Question

У меня проблемы с блокировкой элемента внутри Коллекции, в частности ConcurrentDictionary.

Мне нужно принять сообщение, найти это сообщение в Словаре и затем выполнить длительное сканирование по нему.Поскольку программа занимает много памяти, после проверки объекты возвращают true , если считают, что сейчас самое время удалить его (что я и делаю, удаляя его из словаря).Однако другой поток может прийти в то же время и попытаться получить доступ к этому же объекту сразу после удаления.Это моя первая попытка:

string dictionaryKey = myMessage.someValue;

DictionaryObject currentObject = myConcurrentDictionary.GetOrAdd(dictionaryKey, new DictionaryObject());
// we can be interrupted here
lock (currentObject)
{
    //KeyNotFoundException is possible on line below
    if (myConcurrentDictionary[dictonaryKey].scan(myMessage)) // Scans the message - returns true if the object says its OK to remove it from the dictionary
    {
      DictionaryObject temp;                      //   It's OK to delete it
      if (!queuedMessages.TryRemove(ric, out temp))   // Did delete work?
       throw new Exception("Was unable to delete a DictionaryObject that just reported it was ok to delete it");
    }
}

Однако вышеприведенное не работает - один поток может удалить объект из Словаря, прежде чем другой попытается получить доступ к этому объекту в Словаре.,Прочитав, что блокировка является сокращением для Monitor.Enter и Monitor.Exit , я попытался сделать следующее:

string dictionaryKey = myMessage.someValue;
Monitor.Enter(GetDictionaryLocker);
DictionaryObject currentObject = myConcurrentDictionary.GetOrAdd(dictionaryKey, new DictionaryObject());
// we can be interrupted here
lock (currentObject)
{
    Monitor.Exit(GetDictionaryLocker);
    //KeyNotFoundException is still possible on line below
    if (myConcurrentDictionary[dictonaryKey].scan(myMessage)) // Scans the message - returns true if the object says its OK to remove it from the dictionary
    {
      DictionaryObject temp;                   //   It's OK to delete it
      if (!queuedMessages.TryRemove(ric, out temp))   // Did delete work?
       throw new Exception("Was unable to delete a DictionaryObject that just reported it was ok to delete it");
    }
}

Оба способа могут привести к KeyNotFoundException при попыткеищите объект в Словаре.

Кто-нибудь знает, как я могу найти объект, который хочу заблокировать, и затем заблокировать его без прерывания?Извините - я новичок в параллелизме и чувствую себя сбитым с толку!

Спасибо,

Фредерик

Answer 1

Вы должны удалить объект из словаря перед началом сканирования, чтобы другие потоки не пытались использовать его одновременно.Вы всегда можете добавить его обратно, если потребуется позже, после сбоя в scan().Как удаление, так и добавление гарантированно поточно-ориентированы в этой параллельной коллекции.

Это должно сделать возможным то, что вы хотите, без использования lock s или Monitor.

string dictionaryKey = myMessage.someValue;

DictionaryObject currentObject = null;
if (myConcurrentDictionary.TryRemove(dictionaryKey, out currentObject))
{
    //KeyNotFoundException is possible on line below
    if (!currentObject.scan(myMessage)) // Scans the message - returns true if the object says its OK to remove it from the dictionary
    {
      if (!myConcurrentDictionary.TryAdd(dictionaryKey, currentObject))
       throw new Exception("Was unable to re-insert a DictionaryObject that is not OK for deletion");
    }
} 

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

Проблема с вашей текущей моделью заключается в том, что, хотя коллекция является поточно-ориентированной, вам действительно нужно сделать, если вы хотите оставить «сканируемые» элементыв коллекции нужно выполнить следующую комбинацию операций атомарно : 1. найти свободный предмет и 2. пометить его как «используемый».

1. может быть сделано благодаря тому, что коллекция является поточно-ориентированной, но
2. должна выполняться отдельно, поэтому вы открываете окно для нескольких scan() s на одном и том же объекте.

Answer 2

Как насчет включения поля в объект, который указывает, какому потоку принадлежит он, и использования Threading.Interlocked.CompareExchange для попытки его получения? Представьте, что если объект используется, код не должен блокироваться, а просто отказаться от операции.

Answer 3

Это предположение ...

Проблема в том, что concurrentDictionary возвращает уникальные объекты каждому потоку, чтобы они не мешали друг другу.

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

Любое решение зависит от того, что вы можете сделать со словарем.

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

Таким образом, каждый поток никогда не будет бороться за одно и то же сообщение.

Answer 4

Я думаю, что есть фундаментальное неправильное понимание "блокировки" объекта здесь.

Когда вы удерживаете блокировку объекта (Monitor.Enter), это фактически не мешает другим потокам использовать этот объект,Он просто предотвращает захват других потоков для этого конкретного объекта .Что вам нужно сделать, это добавить вторичный объект и заблокировать его - и убедиться, что каждый поток также блокируется на нем.

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