c++ - Sharing Data Among Threads

Question

I am writing a Windows 10 application that collects data from 4 different sensors (encoders and IMUs) via serial. I am writing this in Visual Studio in c++11. I have one thread per sensor that continuously pulls data from it in an infinite loop.

As I am new to this, I am currently struggling to somehow 'collect' the data in a single thread. I have found that I can use conditional variables to signal another thread (in my case I assume I would be signaling the main thread when each sensor is complete). In this scenario, would I have to store the data in a global variable and protect it with a mutex as I write into it (from the sensor thread) and read from it (in the main loop)?

I worry, however, that this process may be too slow for my application (each sensor gets new data every 1-2 ms), and so in the process of locking the data while the main thread reads, I would lose some data. Would it make sense to (within each sensor thread) store the data in a local variable and then copy this variable to a global variable, and have the main thread only read from the 2nd variable?

I apologize if these are stupid questions, but I really need help. Any example code would be extremely appreciated.

Answer 1

I would handle this by using the boost::asio library to set up asynchronous reads on each sensor. When an asynchronous read completes, it calls a read handler to parse the data and then sets up another asynchronous read. All the read handlers run in the same thread - which makes life a lot simpler.

Answer 2

This looks about like what I would implement. I really doubt you're going to have trouble with throughput if you have milliseconds between input (that is a huge amount of time).

If anyone spots any subtle bugs in the below, let me know.

#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();  
}