Как мне использовать произвольную строку в качестве блокировки в C ++?

Question

Допустим, у меня есть многопоточная программа на C ++, которая обрабатывает запросы в форме вызова функции для handleRequest(string key).Каждый звонок в handleRequest происходит в отдельном потоке, и существует сколь угодно большое количество возможных значений для key.

Я хочу следующее поведение:

• Одновременные вызовы в handleRequest(key) сериализуются, когда они имеют одинаковое значение для key.
• Глобальная сериализация сведена к минимуму.

Тело из handleRequest может выглядеть примерно так:

void handleRequest(string key) {
    KeyLock lock(key);
    // Handle the request.
}

Вопрос: Как бы я реализовал KeyLock чтобы получить требуемое поведение?

Наивная реализация может начинаться примерно так:

KeyLock::KeyLock(string key) {
    global_lock->Lock();
    internal_lock_ = global_key_map[key];
    if (internal_lock_  == NULL) {
        internal_lock_  = new Lock();
        global_key_map[key] = internal_lock_;
    }
    global_lock->Unlock();
    internal_lock_->Lock();
}

KeyLock::~KeyLock() {
    internal_lock_->Unlock();
    // Remove internal_lock_ from global_key_map iff no other threads are waiting for it.
}

...но для этого требуется глобальная блокировка в начале и конце каждого запроса и создание отдельного Lock объект для каждого запроса.Если конкуренция высока между вызовами в handleRequest, возможно, это и не проблема, но это может привести к большим накладным расходам, если конкуренция невелика.

Answer 1

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

Таким образом, вместо одной глобальной блокировки вы распространяете ее на несколько независимых.

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

Answer 2

Это будет зависеть от платформы, но два метода, которые я бы попробовал, были бы:

• Используйте именованный мьютекс / синхронизацию объекты, где имя объекта = ключ
• Используйте блокировку на основе файловой системы, где вы пытаетесь создать недоступный общий доступ временный файл с именем ключа.Если он уже существует (= уже заблокирован), произойдет сбой, и вам придется выполнить опрос, чтобы повторить попытку

Оба метода будут зависеть от детализации вашей операционной системы.Поэкспериментируйте и посмотрите, что работает..

Answer 3

Возможно , std::map<std::string, MutexType> было бы тем, чего ты хочешь, где MutexType это тип мьютекса, который вы хотите.Вероятно, вам придется обернуть доступы к карте в другой мьютекс, чтобы убедиться, что ни один другой поток не вставляет одновременно (и не забудьте выполнить проверку еще раз после блокировки мьютекса, чтобы убедиться, что другой поток не добавил ключ во время ожидания мьютекса!).

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

Answer 4

Повысьте степень детализации и заблокируйте целые диапазоны ключей

Это вариация ответа Майка Б., где вместо нескольких карт плавных блокировок у вас есть один фиксированный массив блокировок, которые применяются к диапазонам ключей вместо отдельных ключей.

Упрощенный пример:создайте массив из 256 блокировок при запуске, затем используйте первый байт ключа для определения индекса блокировки, который будет получен (т.е.все ключи, начинающиеся с буквы "к", будут охраняться locks[107]).

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

Answer 5

Подумав об этом, другой подход мог бы выглядеть примерно так:

• В handleRequest, создать Callback это делает реальную работу.
• Создать multimap<string, Callback*> global_key_map, защищенный мьютексом.
• Если поток увидит, что key уже обрабатывается, он добавляет свои Callback* к тому global_key_map и возвращается.
• В противном случае он немедленно вызывает свой обратный вызов, а затем вызывает обратные вызовы, которые появились за это время для того же ключа.

Реализовано что-то вроде этого:

LockAndCall(string key, Callback* callback) {
    global_lock.Lock();
    if (global_key_map.contains(key)) {
        iterator iter = global_key_map.insert(key, callback);
        while (true) {
            global_lock.Unlock();
            iter->second->Call();
            global_lock.Lock();
            global_key_map.erase(iter);
            iter = global_key_map.find(key);
            if (iter == global_key_map.end()) {
                global_lock.Unlock();
                return;
            }
        }
    } else {
        global_key_map.insert(key, callback);
        global_lock.Unlock();
    }
}

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

Однако это можно было бы объединить с ответами, данными Майком Би и Константином.

Answer 6

      /**
      * StringLock class for string based locking mechanism
      * e.g. usage
      *     StringLock strLock;
      *     strLock.Lock("row1");
      *     strLock.UnLock("row1");
      */
      class StringLock    {
      public:
          /**
           * Constructor
           * Initializes the mutexes
           */
          StringLock()    {
              pthread_mutex_init(&mtxGlobal, NULL);
          }
          /**
           * Lock Function
           * The thread will return immediately if the string is not locked
           * The thread will wait if the string is locked until it gets a turn
           * @param string the string to lock
           */
          void Lock(string lockString)    {
              pthread_mutex_lock(&mtxGlobal);
              TListIds *listId = NULL;
              TWaiter *wtr = new TWaiter;
              wtr->evPtr = NULL;
              wtr->threadId = pthread_self();
              if (lockMap.find(lockString) == lockMap.end())    {
                  listId = new TListIds();
                  listId->insert(listId->end(), wtr);
                  lockMap[lockString] = listId;
                  pthread_mutex_unlock(&mtxGlobal);
              } else    {
                  wtr->evPtr = new Event(false);
                  listId = lockMap[lockString];
                  listId->insert(listId->end(), wtr);
                  pthread_mutex_unlock(&mtxGlobal);
                  wtr->evPtr->Wait();
              }
          }
          /**
          * UnLock Function
          * @param string the string to unlock
          */
          void UnLock(string lockString)    {
              pthread_mutex_lock(&mtxGlobal);
              TListIds *listID = NULL;
              if (lockMap.find(lockString) != lockMap.end())    {
                  lockMap[lockString]->pop_front();
                  listID = lockMap[lockString];
                  if (!(listID->empty()))    {
                      TWaiter *wtr = listID->front();
                      Event *thdEvent = wtr->evPtr;
                      thdEvent->Signal();
                  } else    {
                      lockMap.erase(lockString);
                      delete listID;
                  }
              }
              pthread_mutex_unlock(&mtxGlobal);
          }
      protected:
          struct TWaiter    {
              Event *evPtr;
              long threadId;
          };
          StringLock(StringLock &);
          void operator=(StringLock&);
          typedef list TListIds;
          typedef map TMapLockHolders;
          typedef map TMapLockWaiters;
      private:
          pthread_mutex_t mtxGlobal;
          TMapLockWaiters lockMap;
      };