Thread: Looking for advice on C++ mulithreading

Originally Posted by phantomotap

The `volatile' semantics does not nor has ever been intended to solve any issues related to memory integrity in the context of multiple threads.

and yet Microsoft compilers have supported this usage for a long time as mentioned above: This enables volatile objects to be used for memory locks and releases in multithreaded applications. My guess is that Microsoft's own applications rely on volatile to be thread safe, which is why they continue to support it in current versions of their compilers via a command line switch that defaults to volatile being thread safe on X86 versions of their compilers.

The issue here isn't about volatile, since the code is using an event class with atomic and conditional variables.

Getting back to the original point in my last example was how to deal with the double buffering of the two arrays, where both threads concurrently read one of the arrays while one of the threads is updating the other array. The usage of an atomic conditional variable to implement the event signal and wait functions should guarantee that any previous writes in the code are performed before the atomic operations, but I'm not sure on this. Cache coherency will take care of the updated array in the caches for both cores, so my concern is if a memory barrier is needed to ensure the array update takes place before the atomic operations that occur afterwards in the code, but could potentially be performed out of order due to compiler or hardware optimization.

What MS did with their compiler was a mistake by their own admission. They certainly don't use it in their own code, and neither should anyone else.

Just because a single compiler vender screwed up by adding extra semantics to volatile is no reason to advocate or use it.

https://msdn.microsoft.com/en-us/lib...v=vs.110).aspx

For those who don't like to click/read links:

Originally Posted by MS

we strongly recommend that you specify /volatile:iso, and use explicit synchronization primitives and compiler intrinsics when you are dealing with memory that is shared across threads.

gg

and yet Microsoft compilers have supported this usage for a long time as mentioned above:

O_o

So? Microsoft does not set the standard for the C or C++ languages.

The issue here isn't about volatile, since the code is using an event class with atomic and conditional variables.

My issue with you is actually about `volatile' semantics because you keep making misleading statements about behavior offered as an extension.

Soma

Originally Posted by rcgldr

Getting back to the original point in my last example was how to deal with the double buffering of the two arrays, where both threads concurrently read one of the arrays while one of the threads is updating the other array. The usage of an atomic conditional variable to implement the event signal and wait functions should guarantee that any previous writes in the code are performed before the atomic operations, but I'm not sure on this. Cache coherency will take care of the updated array in the caches for both cores, so my concern is if a memory barrier is needed to ensure the array update takes place before the atomic operations that occur afterwards in the code, but could potentially be performed out of order due to compiler or hardware optimization.

A conditional variable may be a memory barrier. I don't know. Nevertheless, the mutex definitely is a memory barrier, so any call to signal or wait should guarantee that all writes are flushed.

Now, as far as the code goes... post #20 has no visible data races* that I can see, but I might be missing it due to an insufficiently long execution history. But it's still very confusing. Swapping the buffers from two threads is asking for trouble. Limit it to one. Let one thread be the master and one be the slave. Let print decide when to swap the buffers and only run change when it knows it has swapped the buffers. Otherwise you might get races.

Post #22 has a data race. I didn't look into why, but again, the possibility because you're swapping the buffer on two threads may be the cause. Again, limit it to one thread.

*) I used Intel Inspector.

Swapping the buffers from two threads is asking for trouble.

O_o

If it ain't one...

*shrug*

The array (#20) is just getting in your way. You should understand better if you remove the array. (You can change the array to a single element for convenience.) The approach works because of a very simple "page flip".

o_O

Actually, you shouldn't do away with the array. I've thought of a better way to visualize what is happening.

Code:

struct MyData
{
    std::array<double, 10> m;
};

Code:

struct test
{
    MyData x, y; // MyData m[2]; x = m[0]; y = m[1]
    MyData *cxpx = &x;               
    MyData *cxpy = &y;
    MyData *pxpx = &x;              
    MyData *pxpy = &y;
    std::atomic<bool> stop = {0};
};

You should see now that you don't really have two queues as you imagined.

You just have a single queue containing two elements which just happens to store multiple results.

The "page flip" changes which thread owns which element.

The synchronization looks a little unusual, but I believe is correctly functioning.

Code:

Print (`x'): Tells `Change' to do some work. Prints current results. Swaps result buffer. Waits until `Change' does work.
Change (`y'): Stores the result of some work. Swaps result buffer. Tells `Print' about generated results. Waits until `Print' has consumed generated results.
Print (`y'): Tells `Change' to do some work. Prints current results. Swaps result buffer. Waits until `Change' does work.
Change (`x'): Stores the result of some work. Swaps result buffer. Tells `Print' about generated results. Waits until `Print' has consumed generated results.

Soma

Originally Posted by phantomotap

The array (#20) is just getting in your way.

I used the arrays to closely match the original example which had arrays named x[] and print[].

Originally Posted by phantomotap

You should see now that you don't really have two queues as you imagined. You just have a single queue containing two elements which just happens to store multiple results.

I didn't consider the arrays to be queues, and instead I referred to the code I posted as an example of double buffering, which is similar to a single circular queue containing two elements as you mentioned.

My concern was about the updates to an array being completed when the event (mutex, conditional variable) gets signaled to indicate the array has been updated. Based on the wiki article about memory barriers (link below), mutexes and semaphores are implemented with the required memory barriers, so the example code should be ok.

Memory barrier - Wikipedia, the free encyclopedia

I didn't consider the arrays to be queues, and instead I referred to the code I posted as an example of double buffering, which is similar to a single circular queue containing two elements as you mentioned.

O_o

1): I don't care in the slightest what you think a result queue should like because it literally doesn't matter so long as correctly implemented.

2): I was simplifying the expression of the approach so that Elsyia might appreciate that the mechanism is not "asking for trouble" so to what you were referring isn't relevant to the comment you quoted.

My concern was about the updates to an array being completed when the event (mutex, conditional variable) gets signaled to indicate the array has been updated.

You should watch the series Codeplug was kind enough to link.

The story, in brief, is that the C++11 memory model requires hardware to exhibit acquire/release semantics for the C++11 synchronized/synchronizing primitives so you should be fine.

Soma

Originally Posted by Elysia

Post #22 has a data race. I didn't look into why, but again, the possibility because you're swapping the buffer on two threads may be the cause. Again, limit it to one thread.

Yes, it does, I didn't pay proper attention to the swapping: The reason is simply that the buffering can be "synchronized in the wrong way", i.e. that reading and printing takes place on the same array. This happens because the swapping is interrupted when not printing.

PS: In my actual implementation I have two classes, one for the viewing, one for the acquisition, so separating them is not so opaque as in the example. However, I picked up the buffer idea by rcgldr and it works fine.

I"m not sure what the goal is at this point, but here is an example with pause and continue from main():

Code:

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>

struct test
{
    double x[10], y[10];
    double *cxpx = &x[0];               // ptrs for change_x
    double *cxpy = &y[0];
    double *pxpx = &x[0];               // ptrs for print_x
    double *pxpy = &y[0];
    std::atomic<bool> stop = false;
    std::atomic<bool> pause = false;
};

class handle_Event
{
private:
    bool signalled;
    std::condition_variable_any c_v;
    std::mutex mtx;

public:
    handle_Event()
    {
        signalled = false;
    }
    ~handle_Event()
    {
    }
    void signal(void)
    {
        std::lock_guard<std::mutex> lck(mtx);
        signalled = true;
        c_v.notify_one();
    }
    void reset(void)
    {
        std::lock_guard<std::mutex> lck(mtx);
        signalled = false;
    }
    bool wait(void)
    {
        std::lock_guard<std::mutex> lck(mtx);
        while (!signalled)
            c_v.wait(mtx);
        return signalled;
    }
};

handle_Event hndl_change_x, hndl_print_x, hndl_continue;

void change_x(test &t)
{
    while (1){
        hndl_change_x.wait();           // wait
        hndl_change_x.reset();
        if (t.stop.load())              // exit if stop flag set
            break;
        const double f = t.cxpx[0];
        for (uint16_t i = 0; i < 10; i++)
            t.cxpy[i] = ((i + 1) % 10) ? t.cxpx[i + 1] : f;
        std::swap(t.cxpx, t.cxpy);
        hndl_print_x.signal();          // ready for print
    }
}

void print_x(test &t)
{
    while (1){
        hndl_print_x.wait();            // wait
        hndl_print_x.reset();
        if (t.pause.load()){            // pause if pause set
            hndl_continue.wait();
            hndl_continue.reset();
        }
        if (t.stop.load())              // exit if stop flag set
            break;
        hndl_change_x.signal();         // ready for change
        for (uint16_t j = 0; j < 10; j++)
            std::cout << t.pxpx[j] << " ";
        std::cout << std::endl;
        std::swap(t.pxpx, t.pxpy);
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }
}

int main(void)
{
    double x0[] = { 0., 1., 2., 3., 4., 5., 6., 7., 8., 9. };
    test var;
    for (uint16_t i = 0; i < 10; i++) var.x[i] = x0[i];
    char hit = 0;

    std::thread thread_change_x(change_x, std::ref(var));
    std::thread thread_print_x(print_x,   std::ref(var));
    hndl_print_x.signal();          // start with print

    while (1){                      // wait for input
        if ((hit = std::cin.get()) == 'x')
            break;
        if ((hit == 'p') && (var.pause.load() == false))
            var.pause.store(true);
        if ((hit == 'c') && (var.pause.load() == true)){
            var.pause.store(false);
            hndl_continue.signal();
        }
    }
    var.stop.store(true);           // stop threads
    hndl_change_x.signal();
    hndl_print_x.signal();
    hndl_continue.signal();
    thread_change_x.join();         // wait for child threads to stop
    thread_print_x.join();
    return 0;
}

Originally Posted by rcgldr

I don't understand this statement. No one mentioned queues before. I now realize that you interpreted Elsyia's concern about both threads doing swaps to mean that Elsyia was thinking this meant two queues were involved, while it wasn't clear to me what Elsyia's issue was, other than a generic potential synchronization problem as opposed to anything specific.

My concern is that the design is too complicated and therefore prone to bugs.

>> My concern is that the design is too complicated and therefore prone to bugs.
Indeed. The code could be cleaned up a bit by supporting both a manual-reset and auto-reset Event (example in link.)
Semaphore review

gg

Originally Posted by Elysia

My concern is that the design is too complicated and therefore prone to bugs.

Since events and semaphores and functions like Windows WaitForMultipleObjects() aren't part of the C++ standard, someone has to create the functional equivalent using what's available with the standard, and the code to implement this stuff gets complicated. My guess is that WaitForMultipleObjects() would be one of the more complicated functions to impement.

It wasn't clear to me if you had issues with the swapping of pointers to implement double buffering.

As Codeplug mentioned, an auto-reset event would have eliminated the need to manually reset the event handles in the threads.

A common method I've seen since the 1970's on multi-threading min-computers to current (2015) embedded systems and some PC applications is a message queue interface for synchronization and communication between threads (each thread has one or more queues), with function names like SendMesage() and ReceiveMessage(). Sometimes a copy interface is used, sometimes the actual internal structures, like linked list nodes are used.

Originally Posted by rcgldr

It wasn't clear to me if you had issues with the swapping of pointers to implement double buffering.

As Codeplug mentioned, an auto-reset event would have eliminated the need to manually reset the event handles in the threads.

To be honest, I don't find auto-reset events to be very useful. They just add more fuel to the fire. Suddenly I have to worry if the events auto-reset themselves or not.
But anyway, what I have problems with is your swapping of the buffers.

If each thread only touches local data, then I can reason about the threads without looking at the other thread. If both threads use data from each other, but with a lock, then I have to reason about both threads at once, but only in a limited scope. I only really have to reason about the parts where they touch the shared data with the locks. If the threads read and write shared data without any locks, then I have to look at every line in one thread, then interleave that with the entire other thread and reason about it. And vice versa for the other thread. Extremely complicated. Very prone to bugs.

And that is precisely what you're doing with your double swaps. Limit scope of interaction!

Originally Posted by Elysia

what you're doing with your double swaps. Limit scope of interaction!

The double swap method isn't that uncommon, as mentioned by phantomotap, it's like a single circular queue with just two elements. Expand this to a larger circular queue where a producer thread advances it's head pointer/index while the comsumer thread advances it's tail pointer/index. Neither thread touches the others pointer/index. Instead of an event, each thread could use a semaphore, the producer thread uses a semaphore for a count of empty elements in the queue, only blocking if the queue is full, while the consumer thread uses a semaphore for a count of pending elements in the queue, only blocking if the queue is empty.