优先队列与 pthread 同步

Question

我正在完成一项大学作业，我们将在其中实施并行 A* 搜索 15 puzzle。对于这一部分，我们将只使用一个优先级队列（我想看到多个线程的争用会限制加速）。我面临的一个问题是正确同步从优先级队列中弹出下一个“候选人”。

我尝试了以下方法：

while(1) {
  // The board I'm trying to pop.
  Board current_board;

  pthread_mutex_lock(&priority_queue_lock);
  // If the heap is empty, wait till another thread adds new candidates.
  if (pq->heap_size == 0)
  {
    printf("Waiting...\n");
    pthread_mutex_unlock(&priority_queue_lock);
    continue;
  }
  current_board = top(pq);
  pthread_mutex_unlock(&priority_queue_lock);

  // Generate the new boards from the current one and add to the heap...
}

我尝试过同一想法的不同变体，但由于某种原因，有时线程会卡在“等待”中。该代码可以串行（或使用两个线程）正常工作，因此让我相信这是代码的违规部分。如有必要，我可以发布整个内容。不过，我觉得这是我对互斥锁的理解的问题。提前感谢您的帮助。

编辑： 我在下面添加了并行线程的完整代码：

// h and p are global pointers initialized in main()
void* parallelThread(void* arg)
{
    int thread_id = (int)(long long)(arg);
    while(1)
    {
        Board current_board;

        pthread_mutex_lock(&priority_queue_lock);
        current_board = top(p);
        pthread_mutex_unlock(&priority_queue_lock);

        // Move blank up.
        if (current_board.blank_x > 0)
        {
            int newpos = current_board.blank_x - 1;
            Board new_board = current_board;
            new_board.board[current_board.blank_x][current_board.blank_y] = new_board.board[newpos][current_board.blank_y];
            new_board.board[newpos][current_board.blank_y] = BLANK;
            new_board.blank_x = newpos;

            new_board.goodness = get_goodness(new_board.board);
            new_board.turncount++;

            if (check_solved(new_board))
            {
                printf("Solved in %d turns",new_board.turncount);
                exit(0);
            }

            if (!exists(h,new_board))
            {
                insert(h,new_board);
                push(p,new_board);
            }
        }

        // Move blank down.
        if (current_board.blank_x < 3)
        {
            int newpos = current_board.blank_x + 1;
            Board new_board = current_board;
            new_board.board[current_board.blank_x][current_board.blank_y] = new_board.board[newpos][current_board.blank_y];
            new_board.board[newpos][current_board.blank_y] = BLANK;
            new_board.blank_x = newpos;

            new_board.goodness = get_goodness(new_board.board);
            new_board.turncount++;

            if (check_solved(new_board))
            {
                printf("Solved in %d turns",new_board.turncount);
                exit(0);
            }

            if (!exists(h,new_board))
            {
                insert(h,new_board);
                push(p,new_board);
            }
        }

        // Move blank right.
        if (current_board.blank_y < 3)
        {
            int newpos = current_board.blank_y + 1;
            Board new_board = current_board;
            new_board.board[current_board.blank_x][current_board.blank_y] = new_board.board[current_board.blank_x][newpos];
            new_board.board[current_board.blank_x][newpos] = BLANK;
            new_board.blank_y = newpos;

            new_board.goodness = get_goodness(new_board.board);
            new_board.turncount++;

            if (check_solved(new_board))
            {
                printf("Solved in %d turns",new_board.turncount);
                exit(0);
            }

            if (!exists(h,new_board))
            {
                insert(h,new_board);
                push(p,new_board);
            }
        }

        // Move blank left.
        if (current_board.blank_y > 0)
        {
            int newpos = current_board.blank_y - 1;
            Board new_board = current_board;
            new_board.board[current_board.blank_x][current_board.blank_y] = new_board.board[current_board.blank_x][newpos];
            new_board.board[current_board.blank_x][newpos] = BLANK;
            new_board.blank_y = newpos;

            new_board.goodness = get_goodness(new_board.board);
            new_board.turncount++;

            if (check_solved(new_board))
            {
                printf("Solved in %d turns",new_board.turncount);
                exit(0);
            }

            if (!exists(h,new_board))
            {
                insert(h,new_board);
                push(p,new_board);
            }
        }
    }

    return NULL;
}

Answer 1

我尝试了以下方法：

假设top 也将板从队列中删除，我看不出后面的代码有什么问题。这很浪费（如果队列为空，它将旋转锁定和解锁互斥锁），但不会出错。

我已经添加了完整的代码

如果没有 exists、insert 和 push 的代码，这将毫无用处。

一般观察：

    pthread_mutex_lock(&priority_queue_lock);
    current_board = top(p);
    pthread_mutex_unlock(&priority_queue_lock);

在上面的代码中，您的锁定是 top 函数的“外部”。但在这里：

        if (!exists(h,new_board))
        {
            insert(h,new_board);
            push(p,new_board);
        }

您要么根本不加锁（在这种情况下这是一个错误），要么加锁“内部”exists、insert 和 push。

您不应混用“内部”和“外部”锁定。选择其中一个并坚持下去。

如果您实际上没有将队列锁定在exists、insert 等内部，那么您就有了数据竞争并且错误地考虑了互斥锁：它们保护不变量，您可以'不检查队列是否为空并行与另一个线程执行“删除顶部元素” - 这些操作需要序列化，因此必须在锁下完成。