我正在尝试使用 openmp 并行化老鼠迷宫问题，但它花费了太多时间，有人可以帮助我吗？

Question

与原始代码相比，上面的并行代码花费更多的时间。我已经使用了BFS方法来解决问题。我得到了正确的输出，但它需要太多时间。

（x，y）表示矩阵单元格坐标，并且 dist表示其与源

struct Node
{
    int x, y, dist;
};

 \\Below arrays detail all four possible movements from a cell
int row[] = { -1, 0, 0, 1 };
int col[] = { 0, -1, 1, 0 };

功能的最小距离，以检查是否可以转到位置（行，col） 从当前位置。该函数返回false如果（行，col） 不是有效的位置，也不是值0或已经访问过的。

bool isValid(vector<vector<int>> const &mat, vector<vector<bool>> &visited, int row, int col) 
{
    return (row >= 0 && row < mat.size()) && (col >= 0 && col < mat[0].size())
        && mat[row][col] && !visited[row][col];
}

从源中找到矩阵中的最短路由 电池（i，j）到目标单元格（x，y）

int findShortestPathLength(vector<vector<int>> const &mat, pair<int, int> &src,
                    pair<int, int> &dest)
{
    if (mat.size() == 0 || mat[src.first][src.second] == 0 ||
            mat[dest.first][dest.second] == 0) {
        return -1;
    }
 
    // `M × N` matrix
    int M = mat.size();
    int N = mat[0].size();
 
    // construct a `M × N` matrix to keep track of visited cells
    vector<vector<bool>> visited;
    visited.resize(M, vector<bool>(N));
 
    // create an empty queue
    queue<Node> q;
    
    // get source cell (i, j)
    int i = src.first;
    int j = src.second;
 
    // mark the source cell as visited and enqueue the source node
    visited[i][j] = true;
    q.push({i, j, 0});
 
    // stores length of the longest path from source to destination
    int min_dist = INT_MAX;
    // loop till queue is empty
    while (!q.empty())
    {
        // dequeue front node and process it
        Node node = q.front();
        q.pop();
 
        // (i, j) represents a current cell, and `dist` stores its
        // minimum distance from the source
        int i = node.x, j = node.y, dist = node.dist;

        // if the destination is found, update `min_dist` and stop
        if (i == dest.first && j == dest.second)
        {
            min_dist = dist;
            break;
        }
 
        // check for all four possible movements from the current cell
        // and enqueue each valid movement
       #pragma omp parallel for 
        for (int k = 0; k < 4; k++)
        {
            // check if it is possible to go to position
            // (i + row[k], j + col[k]) from current position
           #pragma omp task shared(i,visited,j)
            {
            if (isValid(mat, visited, i + row[k], j + col[k]))
            {
                // mark next cell as visited and enqueue it
                visited[i + row[k]][j + col[k]] = true;
                q.push({ i + row[k], j + col[k], dist + 1 });
            }
            }
        }
    }
 
    if (min_dist != INT_MAX) {
        return min_dist;
    }
 
    return -1;
}

代码的主要部分仅包含一个矩阵，源和目标坐标，

int main()
{
  vector<vector<int>> mat =
    {
        { 1, 1, 1, 1, 1, 0, 0, 1, 1, 1 },
        { 0, 1, 1, 1, 1, 1, 0, 1, 0, 1 },
        { 0, 0, 1, 0, 1, 1, 1, 0, 0, 1 },
        { 1, 0, 1, 1, 1, 0, 1, 1, 0, 1 },
        { 0, 0, 0, 1, 0, 0, 0, 1, 0, 1 },
        { 1, 0, 1, 1, 1, 0, 0, 1, 1, 0 },
        { 0, 0, 0, 0, 1, 0, 0, 1, 0, 1 },
        { 0, 1, 1, 1, 1, 1, 1, 1, 0, 0 },
        { 1, 1, 1, 1, 1, 0, 0, 1, 1, 1 },
        { 0, 0, 1, 0, 0, 1, 1, 0, 0, 1 },
    };
    pair<int, int> src = make_pair(0, 0);
    pair<int, int> dest = make_pair(7, 5);
    int min_dist = findShortestPathLength(mat, src, dest);
    if (min_dist != -1)
    {
        cout << min_dist<<endl;
    }
    else {
        cout << "Destination cannot be reached from a given source"<<endl;
    }
    return 0;
}

我使用了共享变量，但花费了太多时间。 有人可以帮我吗？

Answer 1

正如人们所说，你只能在 4 个方向上获得并行性；更好的方法是保留一组点：首先是起点，然后是从那里一步可以到达的所有点，然后是两步可以到达的点。

通过这种方法，您可以获得更多的并行性：您正在构建所谓的“波前”，并且可以同时处理所有点。

auto last_level = reachable.back();
vector<int> newly_reachable;
for ( auto n : last_level ) {
  const auto& row = graph.row(n);
  for ( auto j : row ) {
    if ( not reachable.has(j)
         and not any_of
         ( newly_reachable.begin(),newly_reachable.end(),
           [j] (int i) { return i==j; } ) )
      newly_reachable.push_back(j);
  }
}
if (newly_reachable.size()>0)
  reachable.push_back(newly_reachable);

（我是为一般的 DAG 编写的；将迷宫写为 DAG 对读者来说是一种练习。）

但是，这种方法仍然存在很大的问题：如果当前波前的两个点决定添加相同的新点，那么您必须解决这个问题。

对于非常并行的方法，您需要放弃完全添加新点的“推”模型，并采用“拉”模型：在每次“距离”迭代中，您循环遍历所有点并询问，如果我无法到达，是我的一位邻居可以联系到吗？如果是这样，请将我标记为比该邻居多一步可到达。

如果您考虑一下最后一种方法，您会发现您实际上是在使用邻接矩阵和当前可达集进行一系列矩阵向量乘积。只不过您将标量“+”运算替换为“min”，将标量“*”替换为“+1”。阅读有关将图运算解释为线性代数的任何教程。只是它并不是真正的线性。