C++11 Thread Safety of Atomic Containers

Question

I am trying to implement a thread safe STL vector without mutexes. So I followed through this post and implemented a wrapper for the atomic primitives.

However when I ran the code below, it displayed out Failed!twice from the below code (only two instances of race conditions) so it doesn't seem to be thread safe. I'm wondering how can I fix that?

Wrapper Class

template<typename T>
struct AtomicVariable
{
    std::atomic<T> atomic;

    AtomicVariable() : atomic(T()) {}

    explicit AtomicVariable(T const& v) : atomic(v) {}
    explicit AtomicVariable(std::atomic<T> const& a) : atomic(a.load()) {}

    AtomicVariable(AtomicVariable const&other) : 
        atomic(other.atomic.load()) {}

    inline AtomicVariable& operator=(AtomicVariable const &rhs) {
        atomic.store(rhs.atomic.load());
        return *this;
    }

    inline AtomicVariable& operator+=(AtomicVariable const &rhs) {
        atomic.store(rhs.atomic.load() + atomic.load());
        return *this;
    }

    inline bool operator!=(AtomicVariable const &rhs) {
        return !(atomic.load() == rhs.atomic.load());
    }
};

typedef AtomicVariable<int>    AtomicInt;

Functions and Testing

// Vector of 100 elements.
vector<AtomicInt> common(100, AtomicInt(0));

void add10(vector<AtomicInt> &param){
    for (vector<AtomicInt>::iterator it = param.begin();
        it != param.end(); ++it){
        *it += AtomicInt(10);
    }
}

void add100(vector<AtomicInt> &param){
    for (vector<AtomicInt>::iterator it = param.begin();
        it != param.end(); ++it){
        *it += AtomicInt(100);
    }
}

void doParallelProcessing(){

    // Create threads
    std::thread t1(add10, std::ref(common));
    std::thread t2(add100, std::ref(common));

    // Join 'em
    t1.join();
    t2.join();

    // Print vector again
    for (vector<AtomicInt>::iterator it = common.begin();
        it != common.end(); ++it){
        if (*it != AtomicInt(110)){
            cout << "Failed!" << endl;
        }
    }
}


int main(int argc, char *argv[]) {

    // Just for testing purposes
    for (int i = 0; i < 100000; i++){
        // Reset vector
        common.clear();
        common.resize(100, AtomicInt(0));
        doParallelProcessing();
    }
}

Is there such a thing as an atomic container? I've also tested this with a regular vector<int> it didn't have any Failed output but that might just be a coincidence.

Answer 1

Just write operator += as:

    inline AtomicVariable& operator+=(AtomicVariable const &rhs) {
        atomic += rhs.atomic;
        return *this;
    }

In documentation: http://en.cppreference.com/w/cpp/atomic/atomic operator += is atomic.

Your example fails because below scenario of execution is possible:

1. Thread1 - rhs.atomic.load() - returns 10 ; Thread2 - rhs.atomic.load() - returns 100
2. Thread1 - atomic.load() - returns 0 ; Thread2 - atomic.load - returns 0
3. Thread1 - add values (0 + 10 = 10) ; Thread2 - add values (0 + 100)
4. Thread1 - atomic.store(10) ; Thread2 - atomic.store(100)

Finally in this case in atomic value might be 10 or 100, depends of which thread first execute atomic.store.

Answer 2

please note that

           atomic.store(rhs.atomic.load() + atomic.load());

is not atomic

You have two options to solve it. memoery 1) Use a mutex.

EDIT as T.C mentioned in the comments this is irrelevant since the operation here will be load() then load() then store() (not relaxed mode) - so memory order is not related here.

2) Use memory order http://bartoszmilewski.com/2008/12/01/c-atomics-and-memory-ordering/

memory_order_acquire: guarantees that subsequent loads are not moved before the current load or any preceding loads. memory_order_release: preceding stores are not moved past the current store or any subsequent stores.

I'm still not sure about 2, but I think if the stores will not be on parallel, it will work.