C++ 多线程 条件变量

使用了条件变量的多线程程序, 在某个线程中调用了wait操作后, 当其他线程未调用notify时, 原来的wait线程就自动重新启动. 真是如此的不合理.

问题出现

最初接触Spurious wakeup是在陈硕的Linux多线程服务端编程中, 后来帮姐姐修改服务端时, 也遇到了条件变量的使用问题, 毫无疑问, 一知半解无法写出优秀的程序.

以下是我的程序想要实现的功能: 是一种类似垃圾回收的机制.

  • 主线程中需要知道被清理的变量(使用vector<int>来存储), 之后主线程会遍历该vector来进行 清理工作.
  • 回收线程总是在进行遍历全部或是部分对象, 将清理变量存储到vector中, 之后进行挂起, 直到 主线程清理完毕再进行下一步的操作.

为什么不将遍历对象也放在主线程中呢?

答: 最主要的原因就是遍历全部对象需要的时间太久了, 主线程不能接受这样的时间要求

为什么不将回收活动也放在子线程中呢?

答: 当需要被清理的对象总是很多, 并且对主线程的执行有了阻塞时, 会考虑将其放在子线程中进行. 但如果这样设计, 程序的复杂度会进一步提高. 如果有这种需要, 我会再开一帖来分析解决方案.

程序初版

  • 共享的变量
1
2
3
4
5
6
7
class Server {
    vector<int> forDel;
    mutex delMtx;
    condition delCond;

    map<int, int> needClean;    /**< 需要被清理的对象, 当前不考虑其线程安全性 */
}
  • 主线程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
while(1) {
    if(!s.forDel.empty()) {
        vector<int> rmVec;
        {
            Lock l(&delMtx);
            rmVec.swap(s.forDel);
        }

        /** 遍历rmVec, 进行删除 */

        delCond.notify();   /** 解除休眠状态 */
    }

    // do other job.
}
  • 回收线程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
while(1) {
    vector<int> rmVec;
    auto it = needClean.begin();

    while(it != needClean.end()) {
        if(/**< *it invalid */) rmVec.push_back(it->first);
    }

    if(!rmVec.empty()) {
        {
            Lock l(&delMtx);
            s.forDel.swap(rmVec);
        }
        delCond.wait(); /** 此时期望的行为: 进程进入休眠状态 */
    }
}

资料收集与分析

man手册

man 3 pthread_cond_wait

Spurious wakeups from the pthread_cond_timedwait() or pthread_cond_wait() functions may occur. Since the return from pthread_cond_timedwait() or pthread_cond_wait() does not imply anything about the value of this predicate, the predicate should be re-evaluated upon such return.

我们调用了cond_wait函数, 他总是会返回的, 但是返回并不能表明谓词(bool值)发生了改变, 因此 需要不断的衡量返回时的真正状态

为什么明知有这个问题, 却只将其列在说明文档里而不去修改程序?

(POSIX Thread Architect)邮件中

  • 宗教般的使用循环来检测谓词, 确实是一种良好的编码实践
  • 不难想象, 如果用户的代码增加了检测功能, 底层 可以优化同步机制 来提高性能

Spurious wakeup真的发生过吗?

stackoverflow Q&A

回答中的许多博客已经打不开了, 有朋友说有**20%**的wakeup都是Spurious wakeup. 未来 如果有机会, 我会将大量使用线上代码进行修改, 使其能够检测Spurious wakeup.

测试程序

上面的程序犯了很严重的错误, 属于教科书般的失误, 感谢姐姐的认真指出

条件变量必须和mutex一起使用, 也就是在加锁后进行wait/signal操作

最终实现效果

  • 共享的变量
1
2
3
4
5
6
7
8
class Server {
    vector<int> forDel;
    bool isDel; /**< bool值, 也就是之前所说的谓词 */
    mutex delMtx;
    condition delCond;

    map<int, int> needClean;    /**< 需要被清理的对象, 当前不考虑其线程安全性 */
}
  • 主线程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
while(1) {
    
    {   /** 整个状态读写全部加锁 */
        Lock l(&delMtx);
        if(!s.forDel.empty()) {
            vector<int> rmVec;
                rmVec.swap(s.forDel);

            /** 遍历rmVec, 进行删除 */

            s.isDel = true;
            delCond.notify();   /** 解除休眠状态 */
        }
    }

    // do other job.
}
  • 回收线程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
while(1) {
    vector<int> rmVec;
    auto it = needClean.begin();

    while(it != needClean.end()) {
        if(/**< *it invalid */) rmVec.push_back(it->first);
    }

    {
        Lock l(&delMtx);
        if(!rmVec.empty()) {
            s.isDel = false;
            s.forDel.swap(rmVec);
            while(s.isDel == false) {
                delCond.wait(); /** 此时期望的行为: 进程进入休眠状态 */

                if (s.isDel == false) { err; /** 检测到Spurious wakeup */ }
            }
        }
    }
}

很不幸, 我将程序跑过多次, 依旧没有能看到Spurious wakeup现象的出现, 未来工作后, 我会 努力实践一下. 下方是测试时的代码.

Github gist

未来注意事项

教科书般的, 宗教信仰般的记住三句话(针对wait端)

  1. 必须和mutex一起使用, 该布尔表达式的读写需要受mutex保护
  2. 在mutex已上锁的时候才能调用wait
  3. 把判断布尔条件和wait放到while循环中