C++11 Потокобезопасность атомных контейнеров

Question

Я пытаюсь реализовать потокобезопасный вектор STL без мьютексов. Так что я прошел через this выложить и реализовать обертку для атомарных примитивов.

Однако, когда я запустил код ниже, он отображал Failed! дважды из приведенного ниже кода (только два случая условий гонки), поэтому он не выглядит потокобезопасным. Мне интересно, как я могу это исправить?

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;

Функции и тестирование

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

Есть ли такая вещь, как атомный контейнер? Я также проверил это с регулярным vector<int> не было никакого Failed выходной, но это может быть просто совпадением.

Answer 1

Просто напишите оператор += как:

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

В документации: http://en.cppreference.com/w/cpp/atomic/atomic operator += atomic.

Ваш пример терпит неудачу, потому что ниже сценарий выполнения возможен:

1. Thread1 - rhs.atomic.load () - возвращает 10; Thread2 - rhs.atomic.load () - возвращает 100
2. Thread1 - atomic.load () - возвращает 0; Thread2 - atomic.load - возвращает 0
3. Thread1 - добавить значения (0 + 10 = 10); Thread2 - добавить значения (0 + 100)
4. Thread1 - магазин atomic.store (10); Thread2 - atomic.store (100)

Наконец, в этом случае атомарное значение может быть 10 или 100, в зависимости от того, какой поток сначала выполняет atomic.store.

Answer 2

Пожалуйста, обратите внимание, что

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

не атомный

У вас есть два варианта решения. Памятка 1) Используйте мьютекс.

РЕДАКТИРОВАТЬ, как TC упомянул в комментариях, это не имеет значения, так как здесь будут операции load(), затем load(), затем store() (не расслабленный режим) - так что порядок памяти здесь не связан.

2) Используйте порядок памяти http://bartoszmilewski.com/2008/12/01/c-atomics-and-memory-ordering/

memory_order_acquire: гарантирует, что последующие загрузки не будут перемещены до текущей загрузки или любых предыдущих загрузок. memory_order_release: предыдущие хранилища не перемещаются за текущее хранилище или любые последующие хранилища.

Я все еще не уверен насчет 2, но я думаю, что если магазины не будут параллельны, это будет работать.