Multi-Threaded доступ к тому же deque C++

Question

В моем многопоточном приложении у меня есть два потока, которые используют один и тот же std::deque. Один из них пишет ему, а другой читает от него (для анализа данных).

Я получаю эту ошибку:

Deque iterator not dereferencable

EDIT: Это код, который я использую для чтения из дека. Ошибка где-то глубже в if-состоянии (где я получаю доступ к deque с помощью at).

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

Я предполагаю, что это происходит из-за этого многопоточного доступа deque. Я не могу поймать ошибку с блоком try-catch. Разве я не могу этого сделать, потому что он брошен в более глубокий «самолет»? Есть ли возможность исправить эту ошибку?

Answer 1

Использовать блокировку? В некоторых часто доступном месте, объявить mutex:

std::mutex deque_lock; 

затем обертывание считывает и записывает в deque в blocks that acquire it:

... 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... 

Старайтесь держать работайте с минимумом в заблокированном блоке; если это дорогостоящая работа, может стоить копировать значение под замком, а затем использовать его без блокировки, чтобы избежать блокировки другой нити.

Answer 2

Вот очень простой пример пары читатель/писатель, проходящей через очередь.

Примечание Использование 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; 
     } 
    } 
}); 

... или любая другая стратегия, которая подходит для наших целей.