并行自底向上遍历任务依赖树：避免节点重复处理

核心思路 #

避免节点重复处理的关键是原子化节点状态管理+集中式可处理任务队列，确保每个节点只会被一个线程标记为处理中并执行，同时只有当所有子节点处理完成后，父节点才会被加入待处理队列。

具体实现步骤 #

1. 节点状态与计数设计 #

给每个节点添加两个原子变量，从根源上避免竞态：

• std::atomic<int> uncompleted_children：记录未处理完成的子节点数量，初始值为该节点的子节点总数
• std::atomic<NodeState> state：节点状态，枚举值为UNPROCESSED（未处理）、PROCESSING（处理中）、PROCESSED（已完成）

enum class NodeState { UNPROCESSED, PROCESSING, PROCESSED };

struct Node {
    int id;
    std::vector<Node*> parents;
    std::atomic<int> uncompleted_children;
    std::atomic<NodeState> state;

    Node(int id, int child_count) : id(id), uncompleted_children(child_count), state(NodeState::UNPROCESSED) {}
};

2. 初始化与线程启动 #

• 先遍历所有叶子节点（uncompleted_children == 0的节点），将它们加入线程安全的任务队列
• 启动预设数量的工作线程，每个线程循环从队列中取节点处理

3. 线程核心处理逻辑 #

每个线程执行以下循环逻辑，确保无重复、无提前处理：

std::mutex queue_mutex;
std::condition_variable queue_cv;
std::queue<Node*> task_queue;
std::atomic<bool> should_exit = false;

void worker_thread() {
    while (true) {
        Node* node = nullptr;
        // 从队列安全取任务，空队列时阻塞等待
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            queue_cv.wait(lock, []{ return !task_queue.empty() || should_exit; });
            if (should_exit && task_queue.empty()) break;
            node = task_queue.front();
            task_queue.pop();
        }

        // 原子CAS操作抢处理权，避免重复处理
        NodeState expected = NodeState::UNPROCESSED;
        if (!node->state.compare_exchange_strong(expected, NodeState::PROCESSING)) {
            // 已有其他线程在处理该节点，直接跳过
            continue;
        }

        // 执行节点处理逻辑（这里替换成你的process函数）
        process_node(node);

        // 标记节点为已完成
        node->state.store(NodeState::PROCESSED);

        // 更新父节点的未完成子节点计数，符合条件则加入队列
        for (Node* parent : node->parents) {
            // 原子递减计数，确保多线程更新不出错
            int remaining = --parent->uncompleted_children;
            if (remaining == 0) {
                // 父节点所有子节点已处理完成，加入待处理队列
                std::lock_guard<std::mutex> lock(queue_mutex);
                task_queue.push(parent);
                queue_cv.notify_one();
            }
        }
    }
}

4. 关键细节说明 #

• CAS抢处理权：compare_exchange_strong是原子操作，只有第一个线程能将节点状态从UNPROCESSED改为PROCESSING，后续线程尝试时会直接跳过，彻底杜绝重复处理
• 父节点触发逻辑：只有当父节点的uncompleted_children原子递减到0时，才说明所有子节点都处理完成，此时才会将父节点加入队列，完全避免提前处理父节点的问题
• 线程安全队列：用互斥锁和条件变量实现队列的线程安全访问，空队列时线程会阻塞等待，避免无效循环重试，同时保证有新任务时及时唤醒线程

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