Building a pipline structure using c ++ threads

user_12646761 0
Viewed 60

I have three algorithms that I would like to process in a kind of pipline. All algorithms are to be executed endlessly, with continuous new data. The algorithm A creates a data set from a sensor algorithm B and C use this data set to derive events from it (B and C are independent of each other). However, all three algorithms take different lengths of time.

|AAAA AAAA AAAA AAAA
|     BBBBBB BBBBBB BBBBBB
|     C    C    C    C
--------------------------- 
                          time

it is clear to me that algorithm B does not get every value of A and therefore only processes every second one sometimes. But that's not bad...

My idea would be something like this:

std::thread A_Thread;
std::thread B_Thread;
std::thread C_Thread;

while(true)
{
    if (!A_Thread.joinable())
    {
        A_Thread = std::thread(&A::a, this);                
    }

    if (!B_Thread.joinable())
    {
        B_Thread = std::thread(&B::b, this);                
    }

    if (!C_Thread.joinable())
    {
        C_Thread = std::thread(&C::c, this);                
    }
}

My idea is: I check in an endless loop whether the thread is still working, if not I restart it. Within the individual threads I use a timestamp to check whether the currently available data has already been processed. A, B and C are methods that are located in the instance in which the threads are called. I guess I misunderstood joinable, because the threads unfortunately do not end themselves. So my first question would be how can I check if a thread grade is still working? I thought a thread ended itself when it reached the last one, but apparently not. Do I need a join for this? But then my endless loop no longer works ...

Another question is the variable exchange. i use instances within the threads where i save the results. the instance is in the instance (this) that calls the thread. is this possible? It only ever writes one thread to a variable, but everyone should be able to read.

is the approach reasonable or grossly negligent?

multithreading c++
2 Answers

I suggest you invert the problem: have thread A notify threads B and C whenever there is new data available, and have threads B and C wait for this data.

// Shared variables for all three threads
std::shared_ptr<sensor_data> content;
std::condition_variable content_cv;
std::mutex content_mutex;

void thread_A() {
    while (true) { 
        shared_ptr<sensor_data> local_data = generate_sensor_data();
        std::lock_guard<std::mutex> lock(content_mutex);
        // Replace content with fresh sensor data and notify threads.
        std::swap(local_data, content);
        content_cv.notify_all();
    }
}

void thread_B() { // Same for thread C
    int last_seen = 0;
    while (true) {
        shared_ptr<sensor_data> local_content;
        {
            std::unique_lock<std::mutex> lock(content_mutex);
            content_cv.wait(lock, []() { return content && content->version > last_seen; });
            local_content = content;
            last_seen = local_content->version;
        }
        // Do something with local_content
    }
}

Note that there is a slight possibility that thread B and C encounter a "spurious wakeup", ie the condition variable will wake up without thread A signaling. This is countered by keeping track of content->version and only proceeding if a higher version is seen. You could use timestamps as well, of course.

user_15484680

May be you should use task-based parallelism when you specify tasks and dependencies between them and let the system execute them in parallel. This is what Intel TBB Task Scheduler does for you.

user_4120800
Related