【问题标题】：Filter items from boost::lockfree::spsc_queue从 boost::lockfree::spsc_queue 过滤项目
【发布时间】：2016-01-13 12:06:10
【问题描述】：

我使用 boost::lockfree::spsc_queue 将流数据从线程发送到工作线程。

这些是项目的结构：

struct spsc_queue_item
{
    uint8_t ID;
    void *data;
};

数据由 spsc_queue.push 插入，其他线程由 spsc_queue.pop 读取。

但我也有一些用于工作线程的“命令”。像 ID 0 是“开始过滤器”， ID 1 是“停止过滤器”， ID 2 id "数据"...

因此，如果大量“数据”被推送到队列中，“停止过滤器”之类的命令将被延迟，因为首先处理“数据”项。但是如果命令“停止过滤器”进入“数据”项是无用的，可以丢弃。

现在我知道还有成员函数“consume_one”和“consume_all”。

但我没有找到如何将仿函数与这些函数一起使用的示例。我的想法是使用consume_one 作为示例，首先检查ID==0 或ID==1 的项目是否在队列中，然后再继续处理ID==2 的数据项。

有人有一个小例子，如何使用函子过滤掉与请求的 ID 相关的项目？

或者有没有其他快速的方法可以通过“优先级”标志从队列中获取项目？

更新针对sehe的回答，提供更多信息：

感谢您提供此信息。

知道如何让它变得更好吗？

我需要向工作线程发出信号，例如“启动过滤器”、“停止过滤器”……

我正在考虑使用事件：

SetEvent(hStartFilter);

但在这里我必须为每个命令使用、创建和关闭单个事件。

“数据”也可以有不同的 ID。就像工作线程接收：

"start filter" with ID=0
"start filter" with ID=1

然后 ID0 和 ID1 的“数据”进入队列。现在线程收到 ID0 的“停止过滤器”。因此，ID0 的数据队列中的所有项目都已过时，可以删除。

我的第一个测试是从队列中获取所有项目。检查每个匹配的 ID 并删除该项目。之后剩下的其他项目将被推回队列。但是，如果有大量具有不同 ID（最多 32 个）的数据，那么这将占用大量 CPU 资源和时间。队列的最大大小为 2048 个项目。

还有更好的方法吗？

【问题讨论】：

你用这个做什么？如果您正在多路复用异步网络事件，请考虑使用 Boost Asio。它内置了所有的队列和调度逻辑

标签： boost queue

【解决方案1】：

队列具有 FIFO 语义。没有办法过滤，除非您想丢弃不符合您的条件的元素。（只需包装pop 函数）

典型的解决方案是使用

优先级队列
两个独立的队列，一个用于数据，一个用于命令。

如果数据队列中的元素不能超过 x 个，请考虑使用循环缓冲区。 spsc_queue 使用环形缓冲区作为底层存储。

更新为了响应question edit，我决定使用带外信号的每个 ID 的过滤状态创建一个演示。

让我从通常的定义开始：

static constexpr uint8_t NUM_SOURCES = 32;

现在，消费者和生产者共享的通用定义：

namespace queueing {
    using data_t = std::vector<char>; // just for demo

    struct spsc_queue_item {
        uint8_t ID;
        data_t  data;
    };

    // process control
    boost::atomic_bool shutdown_flag { false };

    namespace statistics {
        namespace events {
            boost::atomic_size_t occurred  { 0 };
        }

        namespace packets {
            boost::atomic_size_t queued    { 0 };
            boost::atomic_size_t dropped   { 0 };
            boost::atomic_size_t processed { 0 };
            boost::atomic_size_t skipped[NUM_SOURCES] = {};
        }

        boost::atomic_size_t idle_cycles   { 0 };

        void report();
    }

    // business logic
    boost::atomic_bool   source_enabled  [NUM_SOURCES] = {}; // true:started (process) / false:stopped (skip)
    boost::lockfree::spsc_queue<spsc_queue_item, boost::lockfree::capacity<2048> > shared_queue;
}

如您所见，我更改了数据（因为没有void* 更容易演示）。另外，我添加了一些有用的统计数据，可以在你的休息运行结束时report()-ed。

void producer_thread() {
    using namespace boost;
    namespace stats = queueing::statistics;
    // helpers to generate random data packets or start/stop filter events

    enum kind_t { knd_data, knd_start, knd_stop };
    queueing::data_t const empty {};
    struct event_t { kind_t kind; spsc_queue_item item; };
    // ...

    // now generate queue items in a loop
    while (!queueing::shutdown_flag) {
        auto evt = gen_event();

        std::this_thread::sleep_for(std::chrono::nanoseconds(engine()%102400));

        switch(evt.kind) {
            case knd_data:
                stats::events::occurred++;

                if (queueing::shared_queue.push(evt.item)) {
                    stats::packets::queued++;
                } else {
                    stats::packets::dropped++;
                }
                break;
            case knd_start: {
                    bool expected = false;
                    if (queueing::source_enabled[evt.item.ID].compare_exchange_weak(expected, true))
                        std::cout << "+";// << static_cast<int>(evt.item.ID);
                }
                break;
            case knd_stop: {
                    bool expected = true;
                    if (queueing::source_enabled[evt.item.ID].compare_exchange_weak(expected, false))
                        std::cout << "-";// << static_cast<int>(evt.item.ID);
                }
                break;
        }
    }
}

线程函数的主体非常简单，但值得注意的是start 和stop 事件不会通过队列进行通信。

生产者更简单。它所做的只是耗尽队列，更新一些统计计数器。在处理一个项目之前，检查相应的过滤状态（source_enabled）：

void consumer_thread() {
    namespace stats = queueing::statistics;

    queueing::spsc_queue_item item;

    auto consume_pending = [&] {
        while (queueing::shared_queue.pop(item)) {
            if (queueing::source_enabled[item.ID])
                fake_process(item); // if filtering started, process 
            else
                stats::packets::skipped[item.ID]++; // if filtering stopped, skip 
        }
    };

    while (!queueing::shutdown_flag) {
        consume_pending();
        stats::idle_cycles++;
    }

    consume_pending(); // drain any remaining queued items, to avoid race with shutdown_flag
}

现在，一切都应该是不言自明的，所以，拼凑一个 main() 函数：

int main() {
    using namespace std;

    // check no source_enabled flags are set at start
    assert(0 == count(begin(queueing::source_enabled), end(queueing::source_enabled), true));

    auto producer = thread(producer_thread);
    auto consumer = thread(consumer_thread);

    this_thread::sleep_for(chrono::seconds(1));
    queueing::shutdown_flag = true;

    if (producer.joinable()) producer.join();
    if (consumer.joinable()) consumer.join();

    queueing::statistics::report();
}

我们的程序运行两个线程大约 1 秒，然后等待它们加入。然后它会报告统计数据，在我的系统上看起来像：

++-+++++++--+++-++++-++-+++---+-+-+-+++++-+--+---+++-++---+-++-++-+-+++---++--+++-++---+----+-+-+-+--+++-++--+--+--++--+-+-+-+--+--+++--++-+-++-++-+--+--+++-++-+---+----++-+++-+-++-+----+--+-+-+--+++--+++++-+-+--++-+--++++-+-+---++-+---+-+--++---++++----+-+---+-+-+-+--+-++--+-+++--+++-+----+-+-+-+++-+++--+-++-++++++---++--+-++-++---+-+-++--+-+-----++---+-+-+--+++--++---++--+-+++-++++-+++-+-+--+++-+-+----+-++++-+--+++----+++-------+-++-+-+-++++-++++---++-+---+-++-----+-++++----+++-++++--+--+-----+-++++----++++-+++-+---+---+-+-++++-++---+-++-+-+-+++-+-+--+-----++-+++---+-++---+++-++-+--+++++------++---+-++++-+-+-+--++++-++++-+--+++-++---+-----++-+-++-+-+++--++-+-+-++-++-----+-++--+--+--+-------++--+-++-+--++-++-++--+-+-++-+-+++-++++-+---+--+++--++--+-+++++-+-----++--++--+++--++-+---++----+--+-+--++-++---+++++++-+--+-++---+----+-+-+--+-+-+--++++-++--+--+-+---+++-+++++++-++-+-----+--++------+-++++++--++-++-+---+-++---++-++------+-++--+-++-+++--+++-+++-+-+--+-+--+--+---+-+-+-+--+-++-+-++---+++-+-+-++--+-++-+---++--+-+--++-+++-+--+++---+----+--++-++++++-++-+----+++-+-+--+++-----+---+--++-+--+-++++++-+-+++--+++---+-+-++++-++-+-+----++++----+++-++----+---++-+---++-+-+-++--+++---+--+++----++-++-+++--+--+---+++--+--+--+--+--++++-++++---+-+-+--+-+-+--++++--+-+--++--++++----++-++++++-+--+-+------+-----+++----++-+++++-+--+--+---++-+-++-+--++++-+++---+++-+----+--+++++-+-+--+++--++-+++-+-++---++-++-+-+-+--+-++--+---+-+++--+++++-----+-++-+-+++-+-+-------++++---+-+-++-+--+++++---+--++-+-++-+++----+++-++++---++------+-+---++++--+-+---+++------++++++---++-+----+-+++-+--++-+-+-+-----+-++-++-++--++-+-+-++++++--++---+-+-+-+-+-+-++-++-++----++--++-+++-++---+++--+++---+++--+-+++----++--+-+-+++---++---++-+--+++++-+---++----++--+++-+--+-+++++++-+--+---+--+---+----+-++-++-+--++--+--++-++---+++++--+-+---+-++-+-+----+++-++-+-+--+---+-++-+-----++---++++--+++++-+---+-++--+-+-+----+--++++-+-----++++--++-+-+++++----+++---+++++++--+---+--+--++++--+++-----+-++--+-+-----+++++----+-++++---+-++--+-++-+++--+++-+-+++++--+----++--+--+-+-++-----++-+--++--++++++-+-+++----++++---++-+--+-+------+-+--+++++--+++--++-----+--++-++-+++++-++-------+----++-++--+--++--++++-++---+-+++++----+-++-++---+++---+-++-++----++--++--+++++-+--+-----+-+-+-+-+++-+--++-+-+++--+-+-+++-+-++--+-+-+-+--+-+-+++++---+---+-++-+---++-+-++-+-+++-++-++-+-++-------++---+-++-++++-++--++--+-++-+++---++++--+----+---+-++-+++--+-+++---+-++-++----+--+--+-++--+-++-++++++--+-++-+--+---+-+--+-+--++---+--+-++--++--+--++-++++----+--+--+++-+++-+-+-++--++-+-+---+-+-------+--++++++-++++++-++-+-++-+---+--+-+-++--+++---+----+--+--+-++----+-+-++-++-++-+++--++---++-------+++++--+-+++++++--+--+-+--++--++--++-+--+--+++----+++++-++-------++---+-+--++-++--+++-+-+-+-+------+-+--+++++-+-+--++-++-++--+++++++---+-++--+++-+++--++++-++--+-+---+----+----+---+--+-+++-+-+++++---+--++--+-+++-+++++--+---+-+++++-+---++++--+-++----+---++----+++---+++++-+-++--+--+-++-++----+---++-++-+-+-+---+++-++-+++-+---+++--+-+-----++-+---++-+---++---+-++--++++-+--++-+-++----+-+-+--++--++++--+--++--+--+-+-+++++++--++-+-+-+++--+---+++--++++++--+-+-----+---++-+++--+++--++---+++--+--+-++++-----+++-----++++--++--+-+--
Events occurred:   3061
Queued packets:    3061
Dropped packets:   0
Processed packets: 1464
Filtered (per source) 58 48 53 51 47 39 45 42 53 52 57 50 63 43 49 57 45 58 40 42 56 54 58 52 44 53 61 41 50 33 51 52 
Total filtered:   1597
Idle cycles:      26408166

第一行 (++-+++++++--+++-++++-++-+++---+...) 是一个简写符号，显示source_enabled[] 标志中有效更改的次数。

可以看到，在这个速率下，队列并没有饱和，消费者线程有相当多的空闲周期。

演示Live On Coliru

完整列表供参考：

#include <boost/lockfree/spsc_queue.hpp>
#include <boost/atomic.hpp>
#include <boost/random.hpp>
#include <boost/bind.hpp>
#include <thread>
static constexpr uint8_t NUM_SOURCES = 32;

namespace queueing {
    using data_t = std::vector<char>; // just for demo

    struct spsc_queue_item {
        uint8_t ID;
        data_t  data;
    };

    // process control
    boost::atomic_bool shutdown_flag { false };

    namespace statistics {
        namespace events {
            boost::atomic_size_t occurred  { 0 };
        }

        namespace packets {
            boost::atomic_size_t queued    { 0 };
            boost::atomic_size_t dropped   { 0 };
            boost::atomic_size_t processed { 0 };
            boost::atomic_size_t skipped[NUM_SOURCES] = {};
        }

        boost::atomic_size_t idle_cycles   { 0 };

        void report() {
            namespace stats = queueing::statistics;
            std::cout << "\n";
            std::cout << "Events occurred:   " << stats::events::occurred   << "\n";
            std::cout << "Queued packets:    " << stats::packets::queued    << "\n";
            std::cout << "Dropped packets:   " << stats::packets::dropped   << "\n";
            std::cout << "Processed packets: " << stats::packets::processed << "\n";

            std::cout << "Filtered (per source) ";
            std::copy(std::begin(stats::packets::skipped), std::end(stats::packets::skipped), 
                    std::ostream_iterator<size_t>(std::cout, " "));
            std::cout << "\n";

            auto total_filtered = std::accumulate(std::begin(stats::packets::skipped), std::end(stats::packets::skipped), 0ul);
            std::cout << "Total filtered:   " << total_filtered << "\n";
            std::cout << "Idle cycles:      " << stats::idle_cycles      << "\n";
        }
    }

    // business logic
    boost::atomic_bool   source_enabled  [NUM_SOURCES] = {}; // true:started (process) / false:stopped (skip)
    boost::lockfree::spsc_queue<spsc_queue_item, boost::lockfree::capacity<2048> > shared_queue;
}

void producer_thread() {
    using namespace boost;
    namespace stats = queueing::statistics;
    // generate random data packets or start/stop filter events

    using queueing::spsc_queue_item;

    mt19937 engine;
    auto gen_srce = bind(uniform_int<uint8_t>(0, NUM_SOURCES-1), ref(engine));
    auto gen_data = [&] {
        std::vector<char> v; 
        std::generate_n(back_inserter(v), engine()%1024, bind(uniform_int<uint8_t>{}, ref(engine)));
        return v;
    };
    enum kind_t { knd_data, knd_start, knd_stop };
    auto gen_kind = bind(uniform_int<uint8_t>(knd_data, knd_stop), ref(engine));

    queueing::data_t const empty {};

    // 
    struct event_t { kind_t kind; spsc_queue_item item; };
    auto gen_event = [&] {
        auto kind = static_cast<kind_t>(gen_kind());
        return event_t {
            kind,
            spsc_queue_item {
                gen_srce(),
                kind == knd_data? gen_data() : empty
            }
        };
    };

    // now that we can easily generate queue items, let's do so in a loop
    while (!queueing::shutdown_flag) {
        auto evt = gen_event();

        std::this_thread::sleep_for(std::chrono::nanoseconds(engine()%102400));

        switch(evt.kind) {
            case knd_data:
                stats::events::occurred++;

                if (queueing::shared_queue.push(evt.item)) {
                    stats::packets::queued++;
                } else {
                    stats::packets::dropped++;
                }
                break;
            case knd_start:
                {
                    bool expected = false;
                    if (queueing::source_enabled[evt.item.ID].compare_exchange_weak(expected, true))
                        std::cout << "+";// << static_cast<int>(evt.item.ID);
                }
                break;
            case knd_stop:
                {
                    bool expected = true;
                    if (queueing::source_enabled[evt.item.ID].compare_exchange_weak(expected, false))
                        std::cout << "-";// << static_cast<int>(evt.item.ID);
                }
                break;
        }
    }
}

void fake_process(queueing::spsc_queue_item const& item) {
    // pretend it takes time proportional to the amount of data
    std::this_thread::sleep_for(std::chrono::microseconds(item.data.size()));

    queueing::statistics::packets::processed++;
}

void consumer_thread() {
    namespace stats = queueing::statistics;

    queueing::spsc_queue_item item;

    auto consume_pending = [&] {
        while (queueing::shared_queue.pop(item)) {
            if (queueing::source_enabled[item.ID])
                fake_process(item); // if filtering started, process 
            else
                stats::packets::skipped[item.ID]++; // if filtering stopped, skip 
        }
    };

    while (!queueing::shutdown_flag) {
        consume_pending();
        stats::idle_cycles++;
    }

    consume_pending(); // drain any remaining queued items, to avoid race with shutdown_flag
}

#include <cassert>

int main() {
    using namespace std;

    // check no source_enabled flags are set at start
    assert(0 == count(begin(queueing::source_enabled), end(queueing::source_enabled), true));

    auto producer = thread(producer_thread);
    auto consumer = thread(consumer_thread);

    this_thread::sleep_for(chrono::seconds(1));
    queueing::shutdown_flag = true;

    if (producer.joinable()) producer.join();
    if (consumer.joinable()) consumer.join();

    queueing::statistics::report();
}

【讨论】：

更新了答案。现在有了功能性Live Demo
我编码、测试和重构的录制直播在这里：livecoding.tv/video/lockfree-queue-filtering-with-32-channels