并行 Dijkstra 死锁

Question

我正在使用 pthreads 来尝试并行化 Dijkstra 的寻路算法，但我遇到了我似乎无法弄清楚的死锁场景。它的要点是每个线程都有自己的优先级队列，它可以在其中工作（std::multiset）和一个与该队列相对应的互斥锁，该队列在需要修改时被锁定。

每个节点都有一个所有者线程，它对应于节点 ID 模线程数。如果线程正在查看节点的邻居并将它们的权重（标签）之一更新为低于以前的值，它会锁定其所有者的队列并删除/重新插入（这是强制集合更新其在队列中的位置） .然而，这个实现似乎陷入了僵局。我不知道为什么，因为据我所知，每个线程一次只持有一个锁。

每个线程的初始队列都包含它的所有节点，但是除了源之外的每个节点的权重都被初始化为 ULONG_MAX。如果一个线程不工作（它从队列中获取具有 ULONG_MAX 权重的节点），它只会保持锁定和解锁，直到另一个线程让它工作。

void *Dijkstra_local_owner_worker(void *param){
  struct thread_args *myargs = ((struct thread_args *)param);

  int tid = myargs->tid;
  std::multiset<Node *,cmp_label> *Q = (myargs->Q);

  struct thread_args *allargs = ((struct thread_args *)param)-tid;

  AdjGraph *G = (AdjGraph *)allargs[thread_count].Q;
  struct Node *n, *p;
  int owner;
  std::set<Edge>::iterator it;
  Edge e;

  pthread_mutex_lock(&myargs->mutex);
  while(!Q->empty()){
    n = *Q->begin(); Q->erase(Q->begin());
    pthread_mutex_unlock(&myargs->mutex);

    if(n->label == ULONG_MAX){
      pthread_mutex_lock(&myargs->mutex);
      Q->insert(n);
      continue;
    }
    for( it = n->edges->begin(); it != n->edges->end(); it++){
      e = *it;
      p = G->getNode(e.dst);
      owner = (int)(p->index % thread_count);
      if(p->label > n->label + e.weight){
        pthread_mutex_lock(&(allargs[owner].mutex));
        allargs[owner].Q->erase(p);
        p->label = n->label + e.weight;
        p->prev = n;
        allargs[owner].Q->insert(p);//update p's position in the PQ
        pthread_mutex_unlock(&(allargs[owner].mutex));
      }
    }
    pthread_mutex_lock(&myargs->mutex);
  }
  pthread_mutex_unlock(&myargs->mutex);
  return NULL;
}

这是产生线程的函数。

bool Dijkstra_local_owner(AdjGraph *G, struct Node *src){
  G->setAllNodeLabels(ULONG_MAX);
  struct thread_args args[thread_count+1];
  src->label = 0;
  struct Node *n;
  for(int i=0; i<thread_count; i++){
    args[i].Q = new std::multiset<Node *,cmp_label>;
    args[i].tid = i;
    pthread_mutex_init(&args[i].mutex,NULL);
  }

  for(unsigned long i = 0; i < G->n; i++){
    n = G->getNode(i); //give all threads their workload in advance
    args[(n->index)%thread_count].Q->insert(n);
  }
  args[thread_count].Q = (std::multiset<Node *,cmp_label> *)G;
  //hacky repackaging of a pointer to prevent use of globals
  //please note this works and is not the issue. I know it's horrible.

  pthread_t threads[thread_count];
  for(int i=0; i< thread_count; i++){
    pthread_create(&threads[i],NULL,Dijkstra_local_owner_worker,&args[i]);
  }

  for(int i=0; i< thread_count; i++){
    pthread_join(threads[i],NULL);
  }

  for(int i=0; i< thread_count; i++){
    delete args[i].Q;
  }
}

每个线程参数的结构定义：

struct thread_args{
  std::multiset<Node *,cmp_label> *Q; 
  pthread_mutex_t mutex; 
  int tid;
};

我的问题是，这段代码死锁在哪里？我在这里得到了隧道视野，所以我看不出哪里出错了。我已经确保所有其他逻辑都正常工作，所以指针取消引用等都是正确的。

Answer 1

如果线程不工作（它正在获取具有 ULONG_MAX 权重的节点 从队列中）它只是保持锁定和解锁直到另一个 线程让它工作。

这是一个潜在的问题——一旦线程进入这种状态，它实际上会在其时间片的整个持续时间内保持互斥锁的锁定。 pthreads 互斥锁是轻量级的，这意味着它们不能保证是公平的 - 忙碌等待的线程很有可能（甚至可能）能够在唤醒的等待线程能够获取锁之前重新获取锁。

您应该在这里使用pthread_cond_wait()，并在另一个线程更新队列时发出条件变量信号。然后你的循环开始看起来像：

  pthread_mutex_lock(&myargs->mutex);

  while (!Q->empty())
  {
    n = *Q->begin();
    if (n->label == ULONG_MAX)
    {
        pthread_cond_wait(&myargs->cond, &myargs->mutex);
        continue;  /* Re-check the condition after `pthread_cond_wait()` returns */
    }

    Q->erase(Q->begin());
    pthread_mutex_unlock(&myargs->mutex);

    /* ... */

您更新另一个节点的队列的位置如下所示：

    /* ... */
    allargs[owner].Q->insert(p); //update p's position in the PQ
    pthread_cond_signal(&allargs[owner].cond);
    pthread_mutex_unlock(&allargs[owner].mutex);

Answer 2

您的代码如下所示：

lock()
While(cond)
{
   unlock()
   if (cond1)
   { 
      lock()
   }

  for(...)
  {
   ....
  }

  lock()  
}

unlock()

我认为很容易看出，根据数据路径的不同，这种方法可能会出现问题。

我只会将锁用于关键操作：

lock()
Q->erase(..)
unlock()

OR

lock()
Q->insert(..)
unlock()

尝试简化事情，看看是否有帮助