为什么std::lock能避免死锁

最近在研读《C++ Concurrency in Action – 2rd》,在讲到死锁的问题描述和解决方案时,给出了std::lock能够实现同时锁住多个mutex而不出现死锁,但是没有讲为啥用std::lock就能避免死锁。google搜了一下全网也没有人回答这个问题,于是打算翻下STL代码自己看看,参考的VS2019 MSVC 14.29.30037 <mutex>文件的代码。

先来看一下std::lock函数,std::lock函数能接收多个(至少需要两个,_Lock0和_Lock1是必须的参数)互斥量,而且允许不同类型的互斥量(每个都是独立的互斥量模板,只要实现了lock、unlock、try_lock函数的类都能当做这里的互斥量,用户可以自定义互斥量类型)。std::lock函数调的内部的_Lock_nonmember1函数。

template <class _Lock0, class _Lock1, class... _LockN>
void lock(_Lock0& _Lk0, _Lock1& _Lk1, _LockN&... _LkN) { // lock multiple locks, without deadlock
    _Lock_nonmember1(_Lk0, _Lk1, _LkN...);
}

接下来看下_Lock_nonmember1函数,这个模板函数有一个针对两个以上互斥量的泛化实现,并实现了一个针对两个互斥量的偏特化。两个以上互斥量的泛化实现和针对两个互斥量的偏特化实现,从原理上看是完全一致的。因此我们在这里用针对两个互斥量的偏特化实现来研究“为什么std::lock能避免死锁”这个问题,聚焦问题的本身。

_Lock_nonmember1函数内循环执行两个_Lock_attempt_small函数,_Lk0和_Lk1两个互斥量在这两个函数中交换了顺序,可以知道只要有一个_Lock_attempt_small函数返回false时就会退出循环,即加锁成功了。

template <class _Lock0, class _Lock1>
void _Lock_nonmember1(_Lock0& _Lk0, _Lock1& _Lk1) {
    // lock 2 locks, without deadlock, special case for better codegen and reduced metaprogramming for common case
    while (_Lock_attempt_small(_Lk0, _Lk1) && _Lock_attempt_small(_Lk1, _Lk0)) { // keep trying
    }
}

再进去看_Lock_attempt_small的实现。_Lock_attempt_small先对入参中的_Lk0加锁[1],然后对_Lk1尝试加锁[2]:如果try_lock返回true说明_Lk1也加锁成功,_Lock_attempt_small返回false让_Lock_nonmember1的while循环退出,_Lk0和_Lk1都成功锁住,std::lock执行结束;如果出现异常会对_Lk0解锁并将异常往上抛出[3],这一步异常保护是保证了异常安全的,因为异常处理中恢复了之前_Lk0没加锁的状态;如果try_lock返回false说明尝试对_Lk1加锁失败,其他线程占有着_Lk1,此时会对之前已经上锁的_Lk0解锁[4],然后让渡当前的CPU时间片给其他线程,最后返回true,_Lock_nonmember1的while循环继续。

template <class _Lock0, class _Lock1>
bool _Lock_attempt_small(_Lock0& _Lk0, _Lock1& _Lk1) {
    // attempt to lock 2 locks, by first locking _Lk0, and then trying to lock _Lk1 returns whether to try again
    _Lk0.lock(); // [1]
    _TRY_BEGIN
    if (_Lk1.try_lock()) { // [2]
        return false;
    }
    _CATCH_ALL
    _Lk0.unlock(); // [3]
    _RERAISE;
    _CATCH_END

    _Lk0.unlock(); // [4]
    _STD this_thread::yield();
    return true;
}

我们再回到_Lock_nonmember1(_Lock0& _Lk0, _Lock1& _Lk1)函数中去看(此时下文中描述的_Lk0和_Lk1是基于_Lock_nonmember1的_Lk0和_Lk1,不是_Lock_attempt_small内的_Lk0和_Lk1),_Lock_nonmember1在调用_Lock_attempt_small时是交替地给进去_Lk0和_Lk1的,即这次先lock住_Lk0再对_Lk1做try_lock,下一次则会交换,先lock住_Lk1再对_Lk0做try_lock,同时当lock住一个锁而try_lock另一个锁失败的时候,会解掉已经lock住的锁,让渡时间片后再重新进入下一次lock和try_lock。这样,使用std::lock的开发者就不必自己去确保两个互斥量始终以相同的顺序去上锁(事实上有些情况下开发者也没法保证,比如开发者在实现一个线程安全的容器的swap函数的时候),而lock住一个锁并try_lock另一个锁失败时解锁并让渡的操作保证了std::lock不会有死锁风险。

搞懂两个互斥量时std::lock的实现了,扩展一下,再来看看两个以上互斥量时_Lock_nonmember1的实现:

// FUNCTION TEMPLATE lock
template <class _Lock0, class _Lock1, class _Lock2, class... _LockN>
void _Lock_nonmember1(_Lock0& _Lk0, _Lock1& _Lk1, _Lock2& _Lk2, _LockN&... _LkN) {
    // lock 3 or more locks, without deadlock
    int _Hard_lock = 0;
    while (_Hard_lock != -1) {
        _Hard_lock = _Lock_attempt(_Hard_lock, _Lk0, _Lk1, _Lk2, _LkN...);
    }
}

template <class... _LockN>
int _Lock_attempt(const int _Hard_lock, _LockN&... _LkN) {
    // attempt to lock 3 or more locks, starting by locking _LkN[_Hard_lock] and trying to lock the rest
    using _Indices = index_sequence_for<_LockN...>;
    _Lock_from_locks(_Hard_lock, _Indices{}, _LkN...);
    int _Failed        = -1;
    int _Backout_start = _Hard_lock; // that is, unlock _Hard_lock

    _TRY_BEGIN
    _Failed = _Try_lock_range(0, _Hard_lock, _LkN...);
    if (_Failed == -1) {
        _Backout_start = 0; // that is, unlock [0, _Hard_lock] if the next throws
        _Failed        = _Try_lock_range(_Hard_lock + 1, sizeof...(_LockN), _LkN...);
        if (_Failed == -1) { // we got all the locks
            return -1;
        }
    }
    _CATCH_ALL
    _Unlock_locks(_Backout_start, _Hard_lock + 1, _Indices{}, _LkN...);
    _RERAISE;
    _CATCH_END

    // we didn't get all the locks, backout
    _Unlock_locks(_Backout_start, _Hard_lock + 1, _Indices{}, _LkN...);
    _STD this_thread::yield();
    return _Failed;
}

实现上其实还是类似的,_Lock_attempt_small变成了_Lock_attempt,而_Lock_attempt函数实际上也是先将_Hard_lock锁住(_Lock_from_locks),然后依次对其他的互斥量做try_lock操作(_Try_lock_range),如果所有的锁都锁定了,_Lock_attempt返回-1结束掉_Lock_nonmember1里的while循环,如果失败了,所有已加的锁全部解掉,然后让渡时间片,_Failed是当前的失败位置,返回回去给_Hard_lock,下一次循环时会在这个位置开始lock,再依次try_lock其他的锁。如果出现异常,处理的方式也是类似的,先解掉所有已经加了锁的互斥量上的锁,再往上throw出去。


本文为原创内容,遵循CC BY-ND 4.0协议,署名-禁止演绎。
转载请注明出处:https://tis.ac.cn/blog/kongdeyou/why_lock_avoids_deadlock/

发表评论

您的电子邮箱地址不会被公开。 必填项已用*标注