具有pthread的优先级队列同步

Question

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

我尝试了以下操作：

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);
        }

您要么根本不锁定（在这种情况下是个bug），要么在“内部” exists，insert和push进行锁定。

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

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