C++并行Bellman-Ford算法原子操作数据竞态问题排查与修复

问题描述 #

我用std::atomic实现C++版本的并行Bellman-Ford算法，核心多线程函数calculateDistances和主函数代码如下。原本想在distances[source] + distance更小时更新distances[destination]，也用了compare_exchange_strong处理可能的线程修改，但代码仍存在数据竞态，导致部分迭代被跳过，算法在部分输入上执行失败。

核心函数calculateDistances #

void calculateDistances(size_t start, size_t end, const Graph& graph, std::vector<std::atomic<double>>& distances, bool& haveChange)
{
    for (size_t source = start; source < end; ++source) {
        for (const auto& connection : graph[source]) {
            const size_t& destination = connection.destination;
            const double& distance = connection.distance;

            double oldDistance = distances[destination];
            while (distances[source] + distance < oldDistance)
            {
                if (distances[destination].compare_exchange_strong(oldDistance, distances[source] + distance)) {
                    haveChange = true;
                    break;
                }
            }
        }
    }
}

主函数上下文 #

std::vector<std::atomic<double>> BellmanFordAsync(const Graph& graph, size_t tasks, size_t start)
{
    size_t nodesCount = graph.getNodesCount();
    std::vector<std::atomic<double>> distances(nodesCount);

    for (auto& atomic : distances)
        atomic = std::numeric_limits<double>::infinity();

    distances[start] = 0;

    size_t chunkSize = (nodesCount + tasks - 1) / tasks;

    std::vector<std::future<void>> return_results;
    bool haveChange = true;

    for (size_t i = 1; i < nodesCount; ++i)
    {
        haveChange = false;

        for (size_t i = 0; i < tasks; ++i) {
            size_t start = i * chunkSize;
            size_t end = std::min((i + 1) * chunkSize, nodesCount);

            std::future<void> future = std::async(calculateDistances,
                start, end, std::cref(graph), std::ref(distances), std::ref(haveChange));

            return_results.push_back(std::move(future));
        }

        for (auto& result : return_results)
            result.wait();

        return_results.clear();
        if (!haveChange) break;
    }

    bool haveNegativeCycles = false;

    for (size_t i = 0; i < tasks; ++i) {
        size_t start = i * chunkSize;
        size_t end = std::min((i + 1) * chunkSize, nodesCount);

        std::future<void> future = std::async(checkNegativeCycles<std::atomic<double>>, start, end, std::cref(graph), std::cref(distances), std::ref(haveNegativeCycles));

        return_results.push_back(std::move(future));
    }

    for (auto& result : return_results) {
        result.wait();
        if (haveNegativeCycles) return std::vector<double>();
    }

    return distances;
}

问题原因 #

1. haveChange的无保护写操作：多个线程同时执行haveChange = true，普通布尔变量的写操作不是原子的，会出现互相覆盖的情况，导致本该触发的迭代被提前终止。

2. 同一轮内的距离值互相干扰：标准Bellman-Ford要求每一轮松弛必须基于上一轮的最终距离值，但直接共享同一个distances数组会让同一轮中某个线程更新的距离被其他线程立即使用，破坏了算法的迭代逻辑，可能计算出错误的最短路径。

3. distances[source]的读取逻辑缺陷：while循环中每次判断distances[source] + distance < oldDistance时，distances[source]可能已经被其他线程修改，导致基于旧值的计算无效，甚至陷入无意义的循环。

修复方案 #

1. 修复haveChange的竞态 #

把普通bool换成std::atomic<bool>，确保写操作的原子性：

• 主函数中声明std::atomic<bool> haveChange = true;
• calculateDistances函数的参数改为std::atomic<bool>& haveChange，写操作改为haveChange.store(true, std::memory_order_relaxed);

2. 改用双缓冲区模型（核心修复） #

并行Bellman-Ford的正确姿势是用双缓冲区：每一轮基于只读的上一轮距离数组计算新值，写入到新数组中，一轮结束后再切换缓冲区。这样避免了同一轮内的线程干扰，符合算法的迭代规则。

修改后的核心函数：

void calculateDistances(size_t start, size_t end, const Graph& graph, const std::vector<double>& old_distances, std::vector<double>& new_distances, std::atomic<bool>& haveChange)
{
    for (size_t source = start; source < end; ++source) {
        if (old_distances[source] == std::numeric_limits<double>::infinity()) {
            continue; // 跳过不可达的源节点
        }
        for (const auto& connection : graph[source]) {
            const size_t dest = connection.destination;
            const double new_dist = old_distances[source] + connection.distance;
            
            // 原子更新new_distances[dest]，确保多个线程更新同一个节点时取最小值
            double current_val = new_distances[dest];
            while (new_dist < current_val) {
                if (std::atomic_compare_exchange_strong(reinterpret_cast<std::atomic<double>*>(&new_distances[dest]), &current_val, new_dist)) {
                    haveChange.store(true, std::memory_order_relaxed);
                    break;
                }
            }
        }
    }
}

修改后的主函数关键部分：

std::vector<double> BellmanFordAsync(const Graph& graph, size_t tasks, size_t start)
{
    size_t nodesCount = graph.getNodesCount();
    std::vector<double> old_distances(nodesCount, std::numeric_limits<double>::infinity());
    old_distances[start] = 0;
    std::vector<double> new_distances = old_distances;

    size_t chunkSize = (nodesCount + tasks - 1) / tasks;
    std::vector<std::future<void>> return_results;
    std::atomic<bool> haveChange = true;

    for (size_t i = 1; i < nodesCount; ++i) {
        haveChange.store(false, std::memory_order_relaxed);
        new_distances = old_distances; // 初始化新缓冲区为上一轮结果

        for (size_t task_idx = 0; task_idx < tasks; ++task_idx) {
            size_t task_start = task_idx * chunkSize;
            size_t task_end = std::min((task_idx + 1) * chunkSize, nodesCount);

            // 强制创建新线程，避免std::async默认策略的同步执行
            return_results.emplace_back(std::async(std::launch::async, calculateDistances,
                task_start, task_end, std::cref(graph), std::cref(old_distances), std::ref(new_distances), std::ref(haveChange)));
        }

        // 等待所有线程完成本轮松弛
        for (auto& fut : return_results) {
            fut.wait();
        }
        return_results.clear();

        if (!haveChange.load(std::memory_order_relaxed)) {
            break; // 无更新，提前终止迭代
        }
        std::swap(old_distances, new_distances); // 切换缓冲区，准备下一轮
    }

    // 负权环检测逻辑（略作修改，基于最终的old_distances）
    std::atomic<bool> cycle_found = false;
    for (size_t task_idx = 0; task_idx < tasks; ++task_idx) {
        size_t task_start = task_idx * chunkSize;
        size_t task_end = std::min((task_idx + 1) * chunkSize, nodesCount);

        return_results.emplace_back(std::async(std::launch::async, [&](size_t s, size_t e) {
            for (size_t source = s; source < e; ++source) {
                if (cycle_found.load(std::memory_order_relaxed) || old_distances[source] == std::numeric_limits<double>::infinity()) {
                    return;
                }
                for (const auto& connection : graph[source]) {
                    if (old_distances[source] + connection.distance < old_distances[connection.destination]) {
                        cycle_found.store(true, std::memory_order_relaxed);
                        return;
                    }
                }
            }
        }, task_start, task_end));
    }

    for (auto& fut : return_results) {
        fut.wait();
    }
    return_results.clear();

    if (cycle_found.load(std::memory_order_relaxed)) {
        return {}; // 存在负权环，返回空数组
    }

    return old_distances;
}

3. 额外优化 #

• 使用std::launch::async强制std::async创建新线程，避免默认策略下可能的同步执行，确保真正并行。
• 负权环检测时，一旦发现环就通过std::atomic<bool>快速终止所有线程，减少不必要的计算。

内容的提问来源于stack exchange，提问作者Givikap120