如何无锁并行化含共享数据结构的二叉树构建程序？

问题核心 #

• 串行逻辑：从根节点出发，生成左子节点并将右子节点入任务栈，递归遍历左分支，完成后从栈取任务继续执行
• 并行需求：4线程执行，T2/T3/T4需等待T1填充任务栈，采用无锁栈避免锁开销
• 编译错误：crossbeam::deque::Worker未实现Sync trait，无法跨线程安全共享

解决方案：替换为线程安全的无锁结构 #

crossbeam::deque的Worker设计仅适用于单线程生产/消费，不支持多线程共享。推荐以下两种无锁方案：

方案1：使用crossbeam::queue::SegQueue（无锁队列模拟栈） #

SegQueue是线程安全的无锁队列，实现了Send和Sync，可通过pop()从尾部取元素模拟栈的后进先出行为。

核心实现代码

use crossbeam::queue::SegQueue;
use std::sync::Arc;
use std::thread;
use std::sync::atomic::{AtomicUsize, Ordering};

// 二叉树节点定义
#[derive(Debug, Clone)]
struct Node {
    value: i32,
    left: Option<Arc<Node>>,
    right: Option<Arc<Node>>,
}

fn main() {
    // 初始化根节点
    let root = Arc::new(Node {
        value: 0,
        left: None,
        right: None,
    });

    // 无锁任务栈（SegQueue模拟）
    let task_stack = Arc::new(SegQueue::new());
    // 任务计数器：跟踪待处理任务数，用于线程退出判断
    let task_count = Arc::new(AtomicUsize::new(1)); // 初始1个根节点任务

    // T1线程：初始任务执行
    let t1_stack = Arc::clone(&task_stack);
    let t1_count = Arc::clone(&task_count);
    let t1_handle = thread::spawn(move || {
        process_node(t1_stack, t1_count, root);
    });

    // T2-T4线程：等待任务并处理
    let mut worker_handles = Vec::with_capacity(3);
    for tid in 2..=4 {
        let stack = Arc::clone(&task_stack);
        let count = Arc::clone(&task_count);
        worker_handles.push(thread::spawn(move || {
            loop {
                // 无任务时短暂休眠，避免空轮询
                if count.load(Ordering::Acquire) == 0 {
                    break;
                }
                match stack.pop() {
                    Ok(node) => {
                        println!("Thread {} processing node {}", tid, node.value);
                        process_node(stack.clone(), count.clone(), node);
                        count.fetch_sub(1, Ordering::Release);
                    }
                    Err(_) => {
                        std::thread::sleep(std::time::Duration::from_micros(10));
                    }
                }
            }
        }));
    }

    // 等待T1完成，再递减计数器触发其他线程退出
    t1_handle.join().unwrap();
    task_count.fetch_sub(1, Ordering::Release);

    // 等待所有工作线程结束
    for handle in worker_handles {
        handle.join().unwrap();
    }
}

fn process_node(
    task_stack: Arc<SegQueue<Arc<Node>>>,
    task_count: Arc<AtomicUsize>,
    node: Arc<Node>
) {
    // 生成左、右子节点
    let left_node = Arc::new(Node {
        value: node.value * 2 + 1,
        left: None,
        right: None,
    });
    let right_node = Arc::new(Node {
        value: node.value * 2 + 2,
        left: None,
        right: None,
    });

    // 右子节点入任务栈，增加任务计数
    task_stack.push(right_node);
    task_count.fetch_add(1, Ordering::Release);

    // 递归处理左分支（保持原串行逻辑的左优先，也可将左节点入栈实现全并行）
    process_node(task_stack, task_count, left_node);
}

方案2：使用lockfree crate的无锁栈 #

如果需要严格的栈结构，可引入lockfree crate，其Stack类型原生支持无锁多线程读写：

# Cargo.toml添加依赖
lockfree = "0.5.1"

核心代码片段

use lockfree::stack::Stack;
// 其余逻辑类似，仅替换任务栈类型
let task_stack = Arc::new(Stack::new());
task_stack.push(root.clone());

// 取任务时使用pop()
match task_stack.pop() {
    Some(node) => { /* 处理节点 */ },
    None => { /* 休眠等待 */ },
}

关键注意事项 #

• 退出条件：必须用原子计数器跟踪任务数量，避免线程无限空轮询或卡死
• 无锁结构特性：SegQueue是生产者-消费者模式的无锁队列，适合多线程生产/消费场景；lockfree::Stack是严格后进先出的无锁栈
• 性能优化：空轮询时使用微秒级休眠，避免占用过多CPU资源

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