多线程访问相同的双端队列 C++

Question

在我的多线程应用程序中，我有两个线程使用相同的std::deque。其中一个写入它，而另一个读取它（用于数据分析）。

我收到此错误：

Deque 迭代器不可取消引用

编辑： 这是我用于从双端队列读取的代码。该错误在 if 条件的某个更深的地方引发（我正在使用 at 访问双端队列）。

for (int i = 0; i <myDeque.size(); i++){
    try{
        if (myDeque.at(i) > 10){
            //do stufff
        }
    }
    catch (...){
        cout << "ERROR" << endl;
    }
}

我认为，这是由于双端队列的这种多线程访问而发生的。 我无法通过 try-catch 块捕获错误。我不能这样做吗，因为它被扔进了一个更深的“平面”？ 有没有可能修复这个错误？

Answer 1

这是一个非常简单的读取器/写入器对通过队列进行通信的示例。

注意使用 condition_variable 来同步有关是否有工作要做以及写入器是否完成的通信（表明读取器在清空队列时可以停止）：

#include <iostream>
#include <thread>
#include <deque>
#include <condition_variable>

std::mutex m;
std::condition_variable reader_action;
bool all_written = false;

std::deque<int> buffer;

// note: this function is called with the mutex unlocked
// we have popped i from the buffer
void handle_read_event(int i)
{
    if (i % 100000 == 0)
        std::cout << i << std::endl;
}

int main()
{
    std::thread writer([]{
        for (int i = 0 ; i < 1000000 ; ++i)
        {
            {
                auto lock = std::unique_lock<std::mutex>(m);
                buffer.push_back(i);
                lock.unlock();
                reader_action.notify_one();
            }
        }

        auto lock = std::unique_lock<std::mutex>(m);
        all_written = true;
        lock.unlock();
        reader_action.notify_one();
    });

    std::thread reader([]{
        while(1)
        {
            auto lock = std::unique_lock<std::mutex>(m);
            reader_action.wait(lock, [] { return all_written || (!buffer.empty()); });
            if (!buffer.empty()) {
                int i = buffer.front();
                buffer.pop_front();
                lock.unlock();
                handle_read_event(i);
            }
            else if(all_written)
            {
                break;
            }
        }
    });

    writer.join();
    reader.join();
    return 0;
}

预期输出：

0
100000
200000
300000
400000
500000
600000
700000
800000
900000

如果我们决定不介意在排空队列时让写入器等待，我们可以这样实现第二个线程：

std::thread reader([]{
    while(1)
    {
        auto lock = std::unique_lock<std::mutex>(m);
        reader_action.wait(lock, [] { return all_written || (!buffer.empty()); });
        while (!buffer.empty()) {
            int i = buffer.front();
            buffer.pop_front();
            handle_read_event(i);
        }

        if(all_written)
        {
            break;
        }
    }
});

...或任何其他适合我们目的的策略。

Answer 2

使用锁定？在一些常用的位置，声明一个mutex：

std::mutex deque_lock;

然后在blocks that acquire it中包装对deque的读写：

... nondeque stuff ...
{
    std::lock_guard<std::mutex> lock(deque_lock);
    mydeque.push_back(...);
}
... more nondeque stuff...

阅读时：

... nondeque stuff ...
{
    std::lock_guard<std::mutex> lock(deque_lock);
    for (const auto& elem : mydeque) {
        ... do stuff with each element, ideally cheap things to avoid blocking writer ...
    }
}
... more nondeque stuff...

尽量减少锁定块中的工作；如果这是一项昂贵的工作，可能值得复制出锁下的值，然后在没有锁的情况下使用它以避免阻塞其他线程。