Safe handling of variables in multi-threading application with shared resources

Question

I've been developing this application as my first C++ project and am at a point where it is functional and would like to take a step back and critique/improve my code from a standpoint of manageability and better coding practices. Additionally, it has some stability issues which I need to address, so I thought instead of dumping a ton of code out there and asking for feedback we'd start at the beginning.

This application has multiple threads with shared resources, and sharing these resources across threads was particularly tricky for a novice programmer. So please see the below 'main.cpp' of my application that creates and waits for the destruction of all threads. Additionally these notes might be helpful:

• The threads are meant to be asynchronous and operate at different rates. For example, the thread that updates the display operates a lower frame-rate than the thread that actually captures the image. I particularly wanted to avoid the scenario where the threads were locked together and the slowest thread set the pace for all others.
• All threads are self exiting based on the 'exitsignal' variable. Essentially this code runs on raspberry pi in an automobile and when 12V signal is lost (car shut-off), a 3.3V input goes low and the GPIO thread writes 'exitsignal' true and all the other threads exit on their next iteration of their for-loop.
• The GPIO thread is finally exited by the 'shutdownsignal' written at the end.

Thanks in advance, and all (constructive) input is appreciated.

int main()
{
    std::cout << "Program launched, starting log file..." << '\n';
    //Quick and dirty log file
    std::ofstream out("/log.txt", std::ios_base::app | std::ios_base::out);
    std::streambuf *coutbuf = std::cout.rdbuf();
    std::cout.rdbuf(out.rdbuf());

    //Create log header
    PrintHeader();

    //Check XML Properties
    if (settings::kreadsuccess < 0) {
        std::cout << "XML reading failed, using defaults." << '\n';
    } else {
        std::cout << "XML reading successful!" << '\n';
    }

    //Create shared resources
    std::atomic<bool> exitsignal{ false };
    std::atomic<bool> shutdownsignal{ false };
    cv::Mat captureimage;
    std::mutex capturemutex;
    cv::Mat displayimage;
    std::mutex displaymutex;
    ProcessValues processvalues;

    //Start threads

    //Start image capture thread
    std::thread t_imagecapture( CaptureImageThread,
                                &captureimage,
                                &capturemutex,
                                &exitsignal );
    //Start image processor thread
    std::thread t_imageprocessor( ProcessImageThread,
                                  &captureimage,
                                  &capturemutex,
                                  &processvalues,
                                  &exitsignal );
    //Start display thread
    std::thread t_displayupdate( DisplayUpdateThread,
                                 &displayimage,
                                 &displaymutex,
                                 &exitsignal );
    //Start image editor thread
    std::thread t_imeageeditor( ImageEditorThread,
                                &captureimage,
                                &capturemutex,
                                &displayimage,
                                &displaymutex,
                                &processvalues,
                                &exitsignal );
    //Start video writer thread
    std::thread t_videowriter( VideoWriterThread,
                               &captureimage,
                               &capturemutex,
                               &displayimage,
                               &displaymutex,
                               &exitsignal );
    //Start GPS poling thread
    std::thread t_gpspolling( GpsPollingThread,
                              &processvalues,
                              &exitsignal );
    //Start LIDAR polling thread
    std::thread t_lidarpolling( LidarPolingThread,
                                &processvalues,
                                &exitsignal );
    //Start GPIO thread
    std::thread t_gpiohandler( GpioHandlerThread,
                               &processvalues,
                               &exitsignal,
                               &shutdownsignal );

    //Set pace setter class!
    PaceSetter mypacesetter( 2, "main" );

    while( !exitsignal ){
        //Should check all threads still running
        mypacesetter.SetPace();
    }

    //Handle all the threads
    t_videowriter.join();
    t_imageprocessor.join();
    t_lidarpolling.join();
    t_gpspolling.join();
    t_imeageeditor.join();
    t_imagecapture.join();
    t_displayupdate.join();
    shutdownsignal = true;
    t_gpiohandler.join();
    std::cout.rdbuf(coutbuf);
    std::cout << "Program exited gracefully!"  << '\n';

    return 0;
}

---

This is what the code looks like now with a common polling thread for the 'IO':

int main()
{
    std::cout << "Program launched, starting log file..." << '\n';
    //Quick and dirty log file
    std::ofstream out("/log.txt", std::ios_base::app | std::ios_base::out);
    std::streambuf *coutbuf = std::cout.rdbuf();
    std::cout.rdbuf(out.rdbuf());

    //Create log header
    PrintHeader();

    //Check XML Properties
    if (settings::kreadsuccess < 0) {
        std::cout << "XML reading failed, using defaults." << '\n';
    } else {
        std::cout << "XML reading successful!" << '\n';
    }

    //Create shared resources
    std::atomic<bool> exitsignal{ false };
    cv::Mat captureimage;
    std::mutex capturemutex;
    cv::Mat displayimage;
    std::mutex displaymutex;
    ProcessValues processvalues;

    //Start threads

    //Start image capture thread
    std::thread t_imagecapture( CaptureImageThread,
                                &captureimage,
                                &capturemutex,
                                &exitsignal );
    //Start image processor thread
    std::thread t_imageprocessor( ProcessImageThread,
                                  &captureimage,
                                  &capturemutex,
                                  &processvalues,
                                  &exitsignal );
    //Start display thread
    std::thread t_displayupdate( DisplayUpdateThread,
                                 &displayimage,
                                 &displaymutex,
                                 &exitsignal );
    //Start image editor thread
    std::thread t_imeageeditor( ImageEditorThread,
                                &captureimage,
                                &capturemutex,
                                &displayimage,
                                &displaymutex,
                                &processvalues,
                                &exitsignal );
    //Start video writer thread
    std::thread t_videowriter( VideoWriterThread,
                               &captureimage,
                               &capturemutex,
                               &displayimage,
                               &displaymutex,
                               &exitsignal );

    //Set pace setter class!
    int pollrate { std::max(std::max(settings::comm::kpollrategps,
                                     settings::comm::kpollratelidar),
                            settings::comm::kpollrategpio) };
    int gpspollinterval{ pollrate / settings::comm::kpollrategps };
    int gpiopollinterval{ pollrate / settings::comm::kpollrategpio };
    int fcwpollinterval{ pollrate / settings::comm::kpollratelidar };
    PaceSetter mypacesetter( pollrate, "Main" );

    //Setup polling

    //GPIO
    bool gpiopoll{ false };
    if ( settings::gpio::kenabled ) {
        try {
            gpiopoll = GpioHandlerSetup();
        } catch ( const std::exception& ex ) {
            std::cout << "GPIO handler setup threw exception: "<< ex.what() << '\n';
            gpiopoll = false;
        } catch ( const std::string& str ) {
            std::cout << "GPIO handler setup threw exception: "<< str << '\n';
            gpiopoll = false;
        } catch (...) {
            std::cout << "GPIO handler setup threw exception of unknown type!" << '\n';
            gpiopoll = false;
        }
    }
    //GPS
    gpsmm* gpsrecv{ NULL };
    bool gpspoll{ false };
    bool timeset{ false };
    if ( settings::gps::kenabled ) {
        try {
            gpsrecv = new gpsmm("localhost", DEFAULT_GPSD_PORT);
            gpspoll = GpsPollingSetup( gpsrecv );
        } catch ( const std::exception& ex ) {
            std::cout << "GPS polling setup threw exception: "<< ex.what() << '\n';
            gpspoll = false;
        } catch ( const std::string& str ) {
            std::cout << "GPS polling setup threw exception: "<< str << '\n';
            gpspoll = false;
        } catch (...) {
            std::cout << "GPS polling setup threw exception of unknown type!" << '\n';
            gpspoll = false;
        }
    } else {
        gpsrecv = NULL;
    }
    //FCW
    FcwTracker* fcwtracker{ NULL };
    bool fcwpoll{ false };
    int dacmodule{ -1 };
    if ( settings::fcw::kenabled ) {
        try {
            fcwtracker = new FcwTracker( settings::fcw::ksamplestoaverage );
            dacmodule = LidarPollingSetup();
            if ( dacmodule >= 0 ) fcwpoll = true;
        } catch ( const std::exception& ex ) {
            std::cout << "FCW setup threw exception: "<< ex.what() << '\n';
            fcwpoll = false;
        } catch ( const std::string& str ) {
            std::cout << "FCW setup threw exception: "<< str << '\n';
            fcwpoll = false;
        } catch (...) {
            std::cout << "FCW setup threw exception of unknown type!" << '\n';
            fcwpoll = false;
        }
    } else {
        fcwtracker = NULL;
    }

    //Loop
    int i{ 0 };
    do {
        //Increment
        i++;
        //Flush cout buffer every second
        if ( i % gpspollinterval == 0 ) std::cout << std::flush;
        if ( (gpspoll) &&
             (i % gpspollinterval == 0) ) GpsPolling( processvalues,
                                                      gpsrecv,
                                                      timeset );
        if ( (gpiopoll) &&
             (i % gpiopollinterval == 0) ) GpioHandler( processvalues,
                                                        exitsignal );
        if ( (fcwpoll) &&
             (i % fcwpollinterval == 0) ) LidarPolling( processvalues,
                                                        dacmodule,
                                                        fcwtracker );

        //Set Pace
        mypacesetter.SetPace();
    } while( !exitsignal );

    //Cleanup variables
    delete gpsrecv;
    gpsrecv = NULL;
    delete fcwtracker;
    fcwtracker = NULL;

    //Handle all the threads
    t_videowriter.join();
    t_imageprocessor.join();
    t_imeageeditor.join();
    t_imagecapture.join();
    t_displayupdate.join();
    std::cout.rdbuf(coutbuf);
    std::cout << "Program exited gracefully!"  << '\n';

    return 0;
}

Probably is the polling of the GPS takes much longer than the LIDAR and GPIO, so it's pretty limiting of the poll rate of those devices. However, there is now a noticeable reduction in CPU utilization.

Answer 1

Don't use threads to scatter your timing schedule of independent actions

In most cases timing schedules can be solved with a simple main loop and time measurement like such:

 bool exit = false; // In case you have concurrency here a `std::atomic<bool>`
                    // is fine of course
 auto least = std::chrono::high_resolution_clock::now();
 do {
      auto current = std::chrono::high_resolution_clock::now(); 
      auto timeDiff = current - least;
      if(timeDiff >= 20ms)  {
          // do the action every 20 milliseconds
      }
      if(timeDiff >= 1s)  {
          // do the action every second
      }
      // aso.
 } while(!exit);

Threads and synchronization mechanisms come with some cost by means of performance and memory overhead you may want to avoid.

Minimize thread interdependencies

It seems a number of processing steps in your example could be serialized trivially by calling these one after another (has finishied).

You're passing cv::Mat captureimage; as parameter all of the time, but the processing steps are separate threads that have to wait for each others processing step.

Such could be easier solved with a simple function calling sequence.

Reduce the number of threads to the parts that could be run independently

It seems your scenario can be reduced to 3 threads. One that takes the video capture input, one that does the image processing (depends if another CPU core is available), and the last one that feeds the video output.

You likely can synchronize all these using a (std::mutex protected) processing queue of std::queue<cv::Mat> frames.

In that mentioned scenario a std::condition_variable may come in handy to trigger the concurrently waiting std::threads execution, as soon a new captured frame is available, or it's readily processed for output.