前言
Pistache，不愧是一个开源软件，其组件真的存在各种问题，今天又被其PollableQueue给坑了，之前修复了其中存在的一个内存泄漏Bug，见《记一次Bug 调试 —— 内存泄漏&&内存越界》。今天又发现了一个关于多线程的Bug。
PollableQueue是基于父类Pistache::Queue实现的，Pistache::Queue是一种无锁的MPSC设计，即多生产者单消费者模型。因此，其设计已经保证了多线程安全，而然PollableQueue为了实现 Pollable 属性，添加了一个eventfd，从而为多线程埋下了隐患。
关于eventfd可以参考我翻译的linux手册《linux手册翻译——eventfd(2)》，手册告诉我们，eventfd也是线程安全的，因此PollableQueue在大部分场景下都没有太大问题，但是偏偏被我遇到了有问题的场景。
源代码
#include <pistache/mailbox.h>
#include <thread>
#include <atomic>
using namespace Pistache;
using namespace std;
int main() {
Polling::Epoll poller;
PollableQueue<int> queue;
queue.bind(poller);
atomic<int> count = 0;
thread t([&] {
for (;;) {
vector<Pistache::Polling::Event> events;
int ready_fds = poller.poll(events);
for (const auto &e:events)
if (e.tag == queue.tag())
for (;;) {
auto t = queue.popSafe();
if (!t)
break;
printf("%d ", *t);
count--;
}
}
});
sleep(1);
for (int i = 0;;) {
if (count == 0) {
count++;
queue.push(i++);
}
}
}
代码逻辑
逻辑非常简单，就一个生产者，一个消费者，并且使用了一个atomic<int> count = 0;用于描述队列的最大容量，在这里最大容量为1。即生产者产生一个数据，只有在消费者消费之后，才会继续生产。
理论上来说，这个模型没有什么线程安全问题，程序会一直运行。
运行结果
程序运行一段时间就会停止(停止的位置不确定)，查看函数栈会发现，消费者者停留在poller.poll(events)，且此时count的值为1,生产者一直空循环。
也就是说，生产者产生了数据(因为count值为1)，但是消费者却没有感知到，因为消费者是通过eventfd感知数据到达的，为什么生产者执行了push操作，消费者却感知不到呢？
源码分析
我们需要查看pop的源码(稍微i简化)：
Entry* pop() override
{
auto ret = Queue<T>::pop();
if (isBound())
{
uint64_t val;
ssize_t bytes = read(event_fd, &val, sizeof val);
}
return ret;
}
我们可以看到，每次执行pop，无论是否有结果返回，都会清空eventfd，而问题正是出现在这里：
如果在执行了auto ret = Queue<T>::pop();之后，发现ret为空，但是此时刚好有新数据到达，那么这个时候清空eventfd，就会导致新数据的eventfd也被清除，从而导致epoll无法检测到新的数据到达了
我们结合我们自己的测试用例分析上述过程：
- 生产者判断count为0，使count++，并执行push放入数据，此后在count被消费者置为0之前都不会继续push
- 消费者的epoll检测到eventfd事件，报告queue中产生了新的数据
- 执行以下循环处理数据：
for (;;) { auto t = queue.popSafe(); if (!t) break; printf("%d ", *t); count--; } - 第一次循环，消费了生产者的数据，count--，此时生产者已经具备了可执行的条件
- 然后消费者再次执行pop操作，当执行完
auto ret = Queue<T>::pop();后生产者被调度，产生了数据并将count++，注意此时ret的值是null - 消费者继续执行，清除eventfd，在这里将抹除生产者刚刚产生的eventfd记录
- 消费因为pop返回null导致for循环推出
- 消费者，继续执行
int ready_fds = poller.poll(events);等待事件到达，但是我们在第6步清除了eventfd，导致此时无法检测到新数据到达 - 如果此时生产者还能继续产生消息，那么这个问题将被解决，但是生产者此时由于消费者没有将count--，也无法产生数据，从而导致程序停止输出！！！
Bug解决
解决方案也很简单：
Entry* pop() override
{
auto ret = Queue<T>::pop();
if (isBound() && ret != nullptr)
// if (isBound())
{
uint64_t val;
ssize_t bytes = read(event_fd, &val, sizeof val);
}
return ret;
}
加一个判断即可，如果ret返回是null，那么就不要再清除eventfd了！
思考
其实如果生产者可以源源不断的产生数据，那么其实程序就不会产生这种类似于死锁的问题，在Pistache中使用PollableQueue的确不会像我这样限制queue的大小，因此不会导致严重的后果，但是也会导致请求无法被及时处理，因此这个改进是应该的！
标题给的是多线程，但这好像不是严格意义上的多线程访问共享资源导致的安全问题，反而更类似与死锁问题！
