c++ - pthread_rwlock 可以同时拥有多少个读者？

Question

我有一个多线程应用程序，它创建了 48 个线程，它们都需要访问一个公共(public)属性 (STL::map)。 map 只会在线程启动时写入，其余时间将从中读取 map 。这似乎是 pthread_rw_lock 的完美用例，而且一切似乎都运行良好。

我遇到了一个完全不相关的段错误并开始分析内核。使用 gdb，我执行了命令 info threads 并且对结果感到非常惊讶。我观察到有几个线程实际上正在按预期从映射中读取，但奇怪的是，有几个线程在等待 rw_lock 的 pthread_rwlock_rdlock() 中被阻塞。

这是等待锁的线程的堆栈跟踪:

#0  0xffffe430 in __kernel_vsyscall ()
#1  0xf76fe159 in __lll_lock_wait () from /lib/libpthread.so.0
#2  0xf76fab5d in pthread_rwlock_rdlock () from /lib/libpthread.so.0
#3  0x0804a81a in DiameterServiceSingleton::getDiameterService(void*) ()

有这么多线程，很难说有多少正在读取，有多少被阻塞，但我不明白为什么任何线程会被阻塞等待读取，考虑到其他线程已经在阅读。

所以这是我的问题:为什么有些线程在等待读取 rw_lock 时被阻塞，而其他线程已经在读取它了？似乎可以同时读取的线程数有限制。

我查看了 pthread_rwlock_attr_t 函数，但没有看到任何相关内容。

操作系统为Linux，SUSE 11。

相关代码如下:

{
  pthread_rwlock_init(&serviceMapRwLock_, NULL);
}

// This method is called for each request processed by the threads
Service *ServiceSingleton::getService(void *serviceId)
{
  pthread_rwlock_rdlock(&serviceMapRwLock_);
    ServiceMapType::const_iterator iter = serviceMap_.find(serviceId);
    bool notFound(iter == serviceMap_.end());
  pthread_rwlock_unlock(&serviceMapRwLock_);

  if(notFound)
  {
    return NULL;
  }

  return iter->second;
}

// This method is only called when the app is starting
void ServiceSingleton::addService(void *serviceId, Service *service)
{
  pthread_rwlock_wrlock(&serviceMapRwLock_);
    serviceMap_[serviceId] = service;
  pthread_rwlock_unlock(&serviceMapRwLock_);
}

更新:

正如 MarkB 在评论中提到的，如果我将 pthread_rwlockattr_getkind_np() 设置为优先考虑编写器，并且有一个编写器阻塞等待，那么观察到的行为就有意义了。但是，我使用我认为是优先考虑读者的默认值。我刚刚验证了没有线程阻塞等待写入。我还按照@Shahbaz 在评论中的建议更新了代码并获得了相同的结果。

Answer 1

您只是观察了获取锁所涉及的固有性能问题。这需要一些时间，而你恰好在其中捕获了那些线程。当受锁保护的操作持续时间很短时尤其如此。

Edit: Reading the source, glibc uses lll_lock to protect critical sections within its own pthread library data structures. The pthread_rwlock_rdlock checks several flags and increments counters, so it does those things while holding a lock. Once those are done, the lock is released with lll_unlock.

为了演示，我实现了一个在获取 rwlock 后休眠的简短例程。主线程等待他们完成。但在等待之前，它会打印线程实现的并发数。

enum { CONC = 50 };

pthread_rwlock_t rwlock;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
unsigned count;

void *routine(void *arg)
{
    int *fds = static_cast<int *>(arg);
    pthread_rwlock_rdlock(&rwlock);
    pthread_mutex_lock(&mutex);
    ++count;
    if (count == CONC) pthread_cond_signal(&cond);
    pthread_mutex_unlock(&mutex);
    sleep(5);
    pthread_rwlock_unlock(&rwlock);
    pthread_t self = pthread_self();
    write(fds[1], &self, sizeof(self));
    return 0;
}

然后主线程等待计数器达到 50:

int main()
{
    int fds[2];
    pipe(fds);
    pthread_rwlock_init(&rwlock, 0);
    pthread_mutex_lock(&mutex);
    for (int i = 0; i < CONC; i++) {
        pthread_t tid;
        pthread_create(&tid, NULL, routine, fds);
    }
    while (count < CONC) pthread_cond_wait(&cond, &mutex);
    pthread_mutex_unlock(&mutex);
    std::cout << "count: " << count << std::endl;
    for (int i = 0; i < CONC; i++) {
        pthread_t tid;
        read(fds[0], &tid, sizeof(tid));
        pthread_join(tid, 0);
    }
    pthread_rwlock_destroy(&rwlock);
    pthread_exit(0);
}

编辑:使用 C++11 线程支持简化示例:

enum { CONC = 1000 };
std::vector<std::thread> threads;

pthread_rwlock_t rwlock;
std::mutex mutex;
std::condition_variable cond;
unsigned count;

void *routine(int self)
{
    pthread_rwlock_rdlock(&rwlock);
    { std::unique_lock<std::mutex> lk(mutex);
      if (++count == CONC) cond.notify_one(); }
    sleep(5);
    pthread_rwlock_unlock(&rwlock);
    return 0;
}

int main()
{
    pthread_rwlock_init(&rwlock, 0);
    { std::unique_lock<std::mutex> lk(mutex);
      for (int i = 0; i < CONC; i++) {
          threads.push_back(std::thread(routine, i));
      }
      cond.wait(lk, [](){return count == CONC;}); }
    std::cout << "count: " << count << std::endl;
    for (int i = 0; i < CONC; i++) {
        threads[i].join();
    }
    pthread_rwlock_destroy(&rwlock);
    pthread_exit(0);
}