如何优化n叉树lock_it函数的多线程并行执行效率？

问题描述 #

我现在在给N叉树的lock_it()函数实现多线程支持，目前已经解决了死锁问题，但没法最大化并行运行效率。lock_it()的功能是锁定一个节点，要求该节点本身未被锁定、所有祖先未被锁定，且没有任何后代被锁定；锁定成功后，需要更新所有祖先的cnt_locked_desc计数。现有代码用了全局互斥锁，但这样所有lock_it()调用都串行执行了，效率很低，求优化方案。

现有代码：

#include <bits/stdc++.h>
#include <thread>
#include <mutex>
using namespace std;
class Node {
public:
    char key;
    bool locked;
    int cnt_locked_desc;
    Node* parent;
    vector<Node*> child;
};
Node* root = NULL;
mutex mtx;
class LockUnlock {
public:
    Node* newNode(char node_key, Node* prt) {
        Node* tmp = new Node;
        (*tmp).key = node_key;
        (*tmp).locked = (*tmp).cnt_locked_desc = 0;
        (*tmp).parent = prt;
        return tmp;
    }
    bool lock_it(Node* node) {
        //O(h)
        if ((*node).cnt_locked_desc > 0 || node == NULL) return 0;
        for (Node* curr = node; curr != NULL; curr = (*curr).parent)
            if ((*curr).locked) return 0;
        mtx.lock();
        for (Node* curr = node; curr != NULL; curr = (*curr).parent) {
            (*curr).cnt_locked_desc++;
        }
        mtx.unlock();
        (*node).locked = 1;
        return 1;
    }
};

优化方案 #

你的核心问题在于使用了全局互斥锁，导致所有lock_it()调用完全串行，完全无法利用多线程的并行性。下面是针对并行效率的关键优化点：

1. 给每个节点添加独立的互斥锁

把全局mtx替换为每个Node实例自己的mutex，这样只有操作同一节点或其祖先/后代的线程才会互斥，不同分支的节点锁定操作可以并行执行。修改Node类：

class Node {
public:
    char key;
    bool locked;
    int cnt_locked_desc;
    Node* parent;
    vector<Node*> child;
    mutex node_mtx; // 每个节点专属的锁
};

2. 按固定顺序获取锁，避免死锁

为了防止死锁，必须按照从根节点到目标节点的顺序获取所有祖先节点的锁（包括目标节点自己）。这样所有线程都遵循相同的锁获取顺序，不会出现循环等待的情况。

3. 在持有锁的情况下完成检查与更新

原来的代码在加锁前做的检查（cnt_locked_desc、祖先是否锁定）是不安全的，因为检查和加锁之间可能有其他线程修改状态。必须在持有所有相关节点的锁之后，再进行合法性检查，确保状态的一致性。

优化后的lock_it()实现

bool lock_it(Node* node) {
    if (node == nullptr) return false;

    // 1. 收集从根到当前节点的完整路径（按根到节点的顺序）
    vector<Node*> path;
    for (Node* curr = node; curr != nullptr; curr = curr->parent) {
        path.push_back(curr);
    }
    reverse(path.begin(), path.end()); // 调整为根 -> 父节点 -> ... -> 目标节点的顺序

    // 2. 按顺序获取所有节点的锁，用unique_lock自动管理锁的生命周期
    vector<unique_lock<mutex>> locks;
    try {
        for (Node* n : path) {
            locks.emplace_back(n->node_mtx); // 自动加锁，异常时会自动释放已获取的锁
        }
    } catch (...) {
        return false;
    }

    // 3. 持有所有锁的前提下，检查锁定合法性
    // 检查目标节点是否有已锁定的后代
    if (node->cnt_locked_desc > 0) {
        return false;
    }
    // 检查所有祖先是否已被锁定
    for (Node* curr = node->parent; curr != nullptr; curr = curr->parent) {
        if (curr->locked) {
            return false;
        }
    }
    // 检查目标节点本身是否已被锁定
    if (node->locked) {
        return false;
    }

    // 4. 执行锁定操作
    node->locked = true;
    // 更新所有祖先的已锁定后代计数
    for (Node* curr = node->parent; curr != nullptr; curr = curr->parent) {
        curr->cnt_locked_desc++;
    }

    return true;
    // 5. locks离开作用域时，unique_lock会自动释放所有持有的锁
}

优化效果说明 #

• 并行性提升：不同分支的节点锁定操作可以同时进行，只有当两个操作涉及同一祖先链时才会互斥，大幅提高多线程场景下的效率。
• 线程安全：所有状态检查和修改都在持有锁的情况下完成，避免了竞态条件；锁获取顺序严格遵循根到节点，彻底避免死锁。
• 资源安全：使用unique_lock自动管理锁的释放，即使发生异常也不会出现锁泄漏的问题。

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