如何用TaskFlow在C++中并行化自上而下BFS？求正确及优化实现

一、现有代码的错误点 #

你的代码存在多处问题，导致无法正确运行：

1. 变量名错误：rectp未定义（应为rtp），rects应为rts，null不是C++合法标识符，初始化空Task应使用tf::Task{}。
2. 并发数据竞争：闭包[&]捕获了p、rtp等局部变量，TaskFlow的任务是异步执行的，主线程和任务线程同时操作rtp会导致未定义行为；且rtp.pop_front()放在任务中执行，会打乱主线程基于队列初始大小的遍历逻辑。
3. 任务依赖逻辑错误：当前代码中taskA是生成子节点数据的任务，但子节点的后续任务并未正确关联——你只是把taskA作为父任务存入队列，却没有为子节点创建对应的处理任务。
4. 无限循环：top函数中rts.front()调用后未执行rts.pop_front()，会导致循环无法终止。
5. Executor获取问题：GetGlobalExecutor()并非TaskFlow标准API，应直接创建tf::Executor实例。

二、修正后的正确实现 #

以下是修复后的并行BFS实现，确保逻辑正确且无并发问题：

#include <iostream>
#include <deque>
#include <taskflow.hpp>

using namespace std;

void bfs(deque<pair<int, tf::Task>> rtp) {
    tf::Taskflow taskflow;
    tf::Executor executor;

    while (!rtp.empty()) {
        size_t sz = rtp.size();
        vector<tf::Task> current_layer_tasks;

        for (size_t i = 0; i < sz; ++i) {
            auto [data, parent_task] = rtp.front();
            rtp.pop_front();

            // 创建当前节点的处理任务
            auto process_task = taskflow.emplace([data]() {
                cout << "Processing node: " << data << endl;
            });

            // 设置与父任务的依赖
            if (parent_task.valid()) {
                parent_task.precede(process_task);
            }

            // 创建生成子节点的任务
            auto generate_children = taskflow.emplace([data, &rtp, &taskflow, process_task]() {
                int child1 = data + 1;
                int child2 = data + 2;

                // 为子节点创建任务并设置依赖
                auto child_task1 = taskflow.emplace([child1]() {
                    cout << "Processing node: " << child1 << endl;
                });
                auto child_task2 = taskflow.emplace([child2]() {
                    cout << "Processing node: " << child2 << endl;
                });

                process_task.precede(child_task1, child_task2);

                // 将子任务加入队列用于下一层处理
                rtp.emplace_back(child1, child_task1);
                rtp.emplace_back(child2, child_task2);
            });

            process_task.precede(generate_children);
            current_layer_tasks.push_back(generate_children);
        }

        // 可选：层级同步屏障，确保下一层等当前层全部完成再开始
        if (!current_layer_tasks.empty()) {
            auto barrier = taskflow.emplace([](){});
            for (auto& task : current_layer_tasks) {
                task.precede(barrier);
            }
        }
    }

    executor.run(taskflow).wait();
}

void top(deque<int>& rts) {
    while (!rts.empty()) {
        int root = rts.front();
        rts.pop_front();
        deque<pair<int, tf::Task>> pq;
        pq.emplace_back(root, tf::Task{});
        bfs(pq);
    }
}

int main() {
    deque<int> rts;
    rts.push_front(1);
    top(rts);
    return 0;
}

三、无需存储父节点的更优实现 #

可以利用TaskFlow的动态任务依赖构建，直接在任务中创建子任务并设置依赖，无需维护存储父Task的队列。

方案1：简化队列存储 #

仅存储节点数据和对应任务，无需额外维护父任务关联逻辑：

#include <iostream>
#include <deque>
#include <taskflow.hpp>

using namespace std;

void parallel_bfs(int root) {
    tf::Taskflow taskflow;
    tf::Executor executor;
    deque<int> node_queue;
    deque<tf::Task> task_queue;

    // 初始化根节点
    auto root_task = taskflow.emplace([root]() {
        cout << "Processing node: " << root << endl;
    });
    node_queue.push_back(root);
    task_queue.push_back(root_task);

    while (!node_queue.empty()) {
        size_t sz = node_queue.size();
        for (size_t i = 0; i < sz; ++i) {
            int data = node_queue.front();
            tf::Task parent_task = task_queue.front();
            node_queue.pop_front();
            task_queue.pop_front();

            int child1 = data + 1;
            int child2 = data + 2;

            // 创建子任务并直接绑定依赖
            auto child_task1 = taskflow.emplace([child1]() {
                cout << "Processing node: " << child1 << endl;
            });
            auto child_task2 = taskflow.emplace([child2]() {
                cout << "Processing node: " << child2 << endl;
            });

            parent_task.precede(child_task1, child_task2);

            node_queue.push_back(child1);
            node_queue.push_back(child2);
            task_queue.push_back(child_task1);
            task_queue.push_back(child_task2);
        }
    }

    executor.run(taskflow).wait();
}

int main() {
    parallel_bfs(1);
    return 0;
}

方案2：动态递归提交任务 #

完全不需要队列存储，通过递归调用直接构建任务依赖树，TaskFlow自动处理并行调度：

#include <iostream>
#include <taskflow.hpp>

using namespace std;

void bfs_task(int data, tf::Taskflow& taskflow, tf::Task parent_task) {
    auto current_task = taskflow.emplace([data]() {
        cout << "Processing node: " << data << endl;
    });

    if (parent_task.valid()) {
        parent_task.precede(current_task);
    }

    // 限制深度避免无限递归
    if (data < 5) {
        int child1 = data + 1;
        int child2 = data + 2;
        // 递归创建子任务并绑定依赖
        bfs_task(child1, taskflow, current_task);
        bfs_task(child2, taskflow, current_task);
    }
}

int main() {
    tf::Taskflow taskflow;
    tf::Executor executor;

    bfs_task(1, taskflow, tf::Task{});

    executor.run(taskflow).wait();
    return 0;
}

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