c++ - 避免并发等待对象中的死锁

Question

我已经实现了一个“Ticket”类，它在多个线程之间作为 shared_ptr 共享。

程序流程是这样的:

1. 调用 parallelQuery() 以启动新的查询作业。创建 Ticket 的共享实例。
2. 查询被分成多个任务，每个任务都在一个工作线程上排队(这部分很重要，否则我会加入线程并完成)。每个任务都获得共享票。
3. 调用 ticket.wait() 等待作业的所有任务完成。
4. 完成一项任务后，它会调用工单上的 done() 方法。
5. 当所有任务完成后，票证被解锁，任务的结果数据聚合并从 parallelQuery() 返回

在伪代码中:

     std::vector<T> parallelQuery(std::string str) {
         auto ticket = std::make_shared<Ticket>(2);
         auto task1 = std::make_unique<Query>(ticket, str+"a");
         addTaskToWorker(task1);
         auto task2 = std::make_unique<Query>(ticket, str+"b");
         addTaskToWorker(task2);
         ticket->waitUntilDone();
         auto result = aggregateData(task1, task2);
         return result;
     }

我的代码有效。但我想知道，如果在等待线程调用 waitUntilDone() 再次锁定互斥锁之前执行解锁互斥锁，理论上是否可能导致死锁。

这是否可能，如何避免这个陷阱？

这里是完整的Ticket类，注意上面问题描述相关的执行顺序示例注释:

#include <mutex>
#include <atomic>

    class Ticket {
    public:
        Ticket(int numTasks = 1) : _numTasks(numTasks), _done(0), _canceled(false) {
            _mutex.lock();
        }

        void waitUntilDone() {
            _doneLock.lock();
            if (_done != _numTasks) {
                _doneLock.unlock(); // Execution order 1: "waiter" thread is here
                _mutex.lock(); // Execution order 3: "waiter" thread is now in a dealock?
            }
            else {
                _doneLock.unlock();
            }
        }

        void done() {
            _doneLock.lock();
            _done++;
            if (_done == _numTasks) {
                _mutex.unlock(); // Execution order 2: "task1" thread unlocks the mutex
            }
            _doneLock.unlock();
        }

        void cancel() {
            _canceled = true;
            _mutex.unlock();
        }

        bool wasCanceled() {
            return _canceled;
        }

        bool isDone() {
            return _done >= _numTasks;
        }

        int getNumTasks() {
            return _numTasks;
        }

    private:
        std::atomic<int> _numTasks;
        std::atomic<int> _done;
        std::atomic<bool> _canceled;
        // mutex used for caller wait state
        std::mutex _mutex;
        // mutex used to safeguard done counter with lock condition in waitUntilDone
        std::mutex _doneLock;
    };

编辑问题时我想到的一个可能的解决方案是我可以把 _done++;在 _doneLock() 之前。最终，这应该就足够了吗？

更新

我已经根据 Tomer 和 Phil1970 提供的建议更新了 Ticket 类。以下实现是否避免了上述陷阱？

class Ticket {
public:
    Ticket(int numTasks = 1) : _numTasks(numTasks), _done(0), _canceled(false) { }

    void waitUntilDone() {
        std::unique_lock<std::mutex> lock(_mutex);
        // loop to avoid spurious wakeups
        while (_done != _numTasks && !_canceled) {
            _condVar.wait(lock);
        }
    }

    void done() {
        std::unique_lock<std::mutex> lock(_mutex);
        // just bail out in case we call done more often than needed
        if (_done == _numTasks) {
            return;
        }
        _done++;
        _condVar.notify_one();
    }

    void cancel() {
        std::unique_lock<std::mutex> lock(_mutex);
        _canceled = true;
        _condVar.notify_one();
    }

    const bool wasCanceled() const {
        return _canceled;
    }

    const bool isDone() const {
        return _done >= _numTasks;
    }

    const int getNumTasks() const {
        return _numTasks;
    }

private:
    std::atomic<int> _numTasks;
    std::atomic<int> _done;
    std::atomic<bool> _canceled;
    std::mutex _mutex;
    std::condition_variable _condVar;
};

Answer 1

不要编写自己的等待方法，而是使用 std::condition_variable。

https://en.cppreference.com/w/cpp/thread/condition_variable .

互斥锁的使用

通常，互斥锁 应该保护给定的代码区域。也就是说，它应该锁定，完成它的工作然后解锁。在您的类(class)中，您有多种方法，其中一些锁定 _mutex 而另一些解锁它。这是非常容易出错的，就像你以错误的顺序调用方法一样，你很可能处于不一致的状态。如果互斥量被锁定两次会怎样？还是在已经解锁时解锁？

关于互斥量需要注意的另一件事是，如果您有多个互斥量，如果您需要锁定两个互斥量但又没有按一致的顺序进行操作，则很容易出现死锁。假设线程 A 先锁定互斥锁 1 和互斥锁 2，线程 B 以相反的顺序锁定它们(先锁定互斥锁 2)。有可能会发生这样的事情:

• 线程A锁互斥体1
• 线程 B 锁互斥锁 2
• 线程 A 想锁定互斥量 2 但不能，因为它已经被锁定。
• 线程 B 想锁定互斥量 1 但不能，因为它已经被锁定了。
• 两个线程将永远等待

因此在您的代码中，您至少应该进行一些检查以确保正确使用。例如，您应该在解锁互斥锁之前验证 _canceled 以确保仅调用一次 cancel。

解决方案

我只是给出一些想法

声明一个 mutux 和一个 condition_variable 来管理类中的done 条件。

std::mutex doneMutex;
std::condition_variable done_condition;

然后 waitUntilDone 看起来像:

void waitUntilDone()
{
    std::unique_lock<std::mutex> lk(doneMutex);
    done_condition.wait(lk, []{ return isDone() || wasCancelled();});
}

done 函数看起来像:

void done() 
{
    std::lock_guard<std::mutex> lk(doneMutex);
    _done++;
    if (_done == _numTasks) 
    {
        doneCondition.notify_one();
    }
}

cancel 函数会变成

void done() 
{
    std::lock_guard<std::mutex> lk(doneMutex);
    _cancelled = true;
    doneCondition.notify_one();
}

如您所见，您现在只有一个互斥锁，因此基本上消除了死锁的可能性。

变量命名

我建议您不要在互斥体的名称中使用lock，因为它会造成混淆。

std::mutex someMutex;
std::guard_lock<std::mutex> someLock(someMutex); // std::unique_lock when needed

这样一来，就更容易知道哪个变量引用了互斥量，哪个变量引用了互斥量的锁。

好书

如果你对多线程很认真，那么你应该买那本书:

C++ Concurrency in Action
Practical Multithreading
Anthony Williams

代码审查 (添加的部分)

本质上相同的代码已发布到 CODE REVIEW:https://codereview.stackexchange.com/questions/225863/multithreading-ticket-class-to-wait-for-parallel-task-completion/225901#225901 .

我已经在那里添加了一些额外的答案。