C++ - обмен данными между потоками

Question

Я пишу приложение для Windows 10, которое собирает данные от 4 различных датчиков (кодировщиков и IMU) через последовательный порт. Я пишу это в Visual Studio в C++11. У меня есть один поток на датчик, который непрерывно извлекает данные из него в бесконечном цикле.

Поскольку я новичок в этом, я сейчас пытаюсь каким-то образом "собрать" данные в один поток. Я обнаружил, что могу использовать условные переменные для сигнализации о другом потоке (в моем случае я предполагаю, что буду сигнализировать о основном потоке, когда каждый датчик будет завершен). В этом сценарии я должен был бы хранить данные в глобальной переменной и защищать их мьютексом, когда я пишу в него (из потока датчика) и читаю из него (в основном цикле)?

Я беспокоюсь, однако, что этот процесс может быть слишком медленным для моего приложения (каждый датчик получает новые данные каждые 1-2 мс), и поэтому в процессе блокировки данных, пока основной поток читает, я потерял бы некоторые данные. Имеет ли смысл (в каждом потоке датчика) сохранять данные в локальной переменной, а затем копировать эту переменную в глобальную переменную, и чтобы основной поток считывал только из 2-й переменной?

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

Answer 1

Это выглядит примерно так, как я бы реализовал. Я действительно сомневаюсь, что у вас будут проблемы с пропускной способностью, если у вас есть миллисекунды между вводом (это огромное количество времени).

Если кто-то обнаружит какие-либо тонкие ошибки в приведенном ниже, дайте мне знать.

#include <thread>
#include <iostream>
#include <chrono>
#include <queue>
#include <mutex>
#include <vector>
#include <condition_variable>

using namespace std::chrono_literals;
using std::vector;
using std::thread;
using std::unique_lock;
using std::mutex;
using std::condition_variable;
using std::queue;

class WorkQueue
{
  condition_variable work_available;
  mutex work_mutex;
  queue<int> work;

public:
  void push_work(int item)
  {
    unique_lock<mutex> lock(work_mutex);

    bool was_empty = work.empty();
    work.push(item);

    lock.unlock();

    if (was_empty)
    {
      work_available.notify_one();
    }    
  }

  int wait_and_pop()
  {
    unique_lock<mutex> lock(work_mutex);
    while (work.empty())
    {
      work_available.wait(lock);
    }

    int tmp = work.front();
    work.pop();
    return tmp;
  }
};

int main() {
  WorkQueue work_queue;

  auto producer = [&]() {
    while (true) {
      work_queue.push_work(10);
      std::this_thread::sleep_for(2ms);
    }
  };

  vector<thread> producers;
  producers.push_back(std::thread(producer));
  producers.push_back(std::thread(producer));
  producers.push_back(std::thread(producer));
  producers.push_back(std::thread(producer));

  std::thread consumer([&]() {        
    while (true)
    {
      int work_to_do = work_queue.wait_and_pop();
      std::cout << "Got some work: " << work_to_do << std::endl;
    }
  });

  std::for_each(producers.begin(), producers.end(), [](thread &p) {
    p.join();
  });    

  consumer.join();  
}

Answer 2

Я бы справился с этим, используя библиотеку boost::asio для настройки асинхронного чтения на каждом датчике. Когда асинхронное чтение завершается, он вызывает обработчик чтения для анализа данных, а затем устанавливает другое асинхронное чтение. Все обработчики чтения выполняются в одном потоке, что значительно упрощает жизнь.