如何在C++中使用任意字符串作为锁

How do I use an arbitrary string as a lock in C++?

本文关键字：字符串 任意 C++      更新时间：2023-10-16

假设我有一个多线程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的调用之间的争用很高，这可能不是问题，但如果争用很低，可能会带来很大的开销。

你可以做一些类似于你在问题中所做的事情，但不是一个global_key_map有几个（可能在数组或向量中）-使用哪一个是由字符串上的一些简单哈希函数决定的。

这样，您就可以将其分散到几个独立的全局锁上，而不是单个全局锁。

这是一种经常在内存分配器中使用的模式（我不知道这个模式是否有名字——它应该有名字）。当一个请求进来时，某些东西决定了分配将来自哪个池（通常是请求的大小，但其他参数也可以考虑在内），然后只需要锁定该池。如果分配请求来自将使用不同池的另一个线程，则不存在锁争用。

这将取决于平台，但我会尝试的两种技术是：

  • 使用命名互斥/同步对象，其中对象名称=键
  • 使用基于文件系统的锁定尝试创建一个不可共享的具有密钥名称的临时文件。如果它已经存在锁定）这将失败，您将必须轮询才能重试

这两种技术都将取决于操作系统的细节。进行实验，看看哪种有效。.

也许std::map<std::string, MutexType>就是您想要的，其中MutexType是您想要的互斥对象的类型。您可能必须将对映射的访问封装在另一个互斥体中，以确保没有其他线程同时插入（记住在互斥体锁定后再次执行检查，以确保另一个线程在等待互斥体时没有添加密钥！）。

同样的原理可以应用于任何其他同步方法，例如关键部分。

提高粒度并锁定整个密钥范围

这是Mike B答案的变体，其中没有几个流体锁映射，而是有一个固定的锁阵列，应用于键范围，而不是单个键。

简化示例：在启动时创建256个锁的数组，然后使用密钥的第一个字节来确定要获取的锁的索引（即，所有以"k"开头的密钥都将由locks[107]保护）。

为了保持最佳吞吐量，您应该分析密钥的分布和争用率。这种方法的好处是零动态分配和简单的清理；您还可以避免两步锁定。不利的一面是，如果密钥分布随着时间的推移而倾斜，则可能出现争用峰值。

经过思考，另一种方法可能会变成这样：

  • 在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();
    }
}

这样做的好处是释放了原本会等待密钥锁的线程，但除此之外，它与我在问题中发布的天真的解决方案几乎相同。

不过，它可以与迈克·B和康斯坦丁给出的答案相结合。

      /**
      * 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;
      };
相关文章：