Атомарный обмен двумя объектами std::atomic‹T*› без блокировки в С++ 11?

Следующий код представляет собой скелет класса атомарных указателей, взятый из приложения для имитации отжига в наборе тестов PARSEC для многопроцессорных систем с общей памятью< /а>.

В этом приложении центральной структурой данных является граф (точнее, список соединений интегральной схемы). Каждый узел в графе имеет атрибут, указывающий его физическое местоположение. Алгоритм порождает множество потоков, и каждый поток многократно и случайным образом выбирает два узла и меняет их физическое местоположение, если это приводит к лучшей стоимости маршрутизации для чипа.

Поскольку граф огромен и любая пара узлов может быть выбрана каждым потоком, единственное приемлемое решение — неблокируемая параллельная структура данных (CDS). Вот почему следующий класс AtomicPtr имеет решающее значение (он используется для атомарного обмена указателями на два объекта физического местоположения без блокировок). Функция atomic_load_acq_ptr() определена в ассемблере и близко соответствует std::atomic<T*>::load(memory_order_acquire).

Я хочу реализовать этот CDS, используя атомарность С++ 11.

template <typename T>
class AtomicPtr {
  private:
    typedef long unsigned int ATOMIC_TYPE;
    T *p __attribute__ ((aligned (8)));
    static const T *ATOMIC_NULL;
    inline T *Get() const {
        T *val;
        do {
            val = (T *)atomic_load_acq_ptr((ATOMIC_TYPE *)&p);
        } while(val == ATOMIC_NULL);
        return val;
    }
    inline void Swap(AtomicPtr<T> &X) {
        // Define partial order in which to acquire elements to prevent deadlocks
        AtomicPtr<T> *first;
        AtomicPtr<T> *last;
        // Always process elements from lower to higher memory addresses
        if (this < &X) {
            first = this;
            last  = &X;
        } else {
            first = &X;
            last  = this;
        }
        // Acquire and update elements in correct order
        T *valFirst = first->Checkout(); // This sets p to ATOMIC_NULL so all Get() calls will spin.
        T *valLast  =  last->PrivateSet(valFirst);
        first->Checkin(valLast); // This restores p to valLast
    }
};

Метод std::atomic<T*>::exchange() можно использовать только для замены голого указателя T* на объект std::atomic<T*>. Как сделать обмен двумя объектами std::atomic<T*> без блокировок?

Я могу предположить, что приведенный ниже класс AtomicPtr сам может быть основан на std::atomic<T*>, объявив:

std::atomic<T*> p;

и заменить все вызовы atomic_load_acq_ptr() на std::atomic<T*>::load(memory_order_acquire) и заменить все вызовы atomic_store_rel_ptr() на std::atomic<T*>::store(memory_order_release). Но моей первой мыслью было, что std::atomic<T*> должен заменить сам AtomicPtr, и может быть умный способ напрямую обмениваться объектами std::atomic<T*>. Какие-нибудь мысли?


person Ahmed Nassar    schedule 21.08.2013    source источник
comment
Невозможно атомарно поменять местами содержимое двух std::atomic в C++11.   -  person Casey    schedule 21.08.2013
comment
вообще-то я не думаю, что такое вообще возможно на x86/x64   -  person Tobias Langner    schedule 21.08.2013
comment
Это невозможно напрямую. Но класс AtomicPtr уже делает это, следуя дисциплине check-out\check-in: 1- Любой поток, который хочет выполнить обмен, сначала проверяет указатель (путем записи контрольного значения, называемого ATOMIC_NULL выше), и, когда это делается, проверяет 2- Любой поток, который хочет прочитать (например, Get()) значение указателя, должен продолжать вращаться, если указатель был проверен.   -  person Ahmed Nassar    schedule 21.08.2013
comment
@AhmedNassar: почему бы не повторить эту логику с std::atomic<T*>?   -  person Matthieu M.    schedule 21.08.2013
comment
@Matthieu: это то, что я уже предлагаю как единственное решение, которое я мог придумать. Я искал лучшие решения :)   -  person Ahmed Nassar    schedule 21.08.2013


Ответы (3)


Мне кажется, что более простой способ получить желаемое — это воспроизвести логику, которую вы здесь видели.

Проблема в том, что нет возможности получить атомарные операции над двумя атомарными объектами, поэтому вам нужно следовать процедуре:

  • заказать атомарность (чтобы избежать взаимоблокировок)
  • "заблокировать" все, кроме последнего (в порядке возрастания)
  • выполнить операцию атомарно над последним
  • выполнить операцию и «разблокировать» остальные по одному (в порядке убывания)

Это, конечно, совсем несовершенно:

  • не атомарный: пока вы заняты блокировкой переменной, любая из еще не заблокированных может изменить состояние
  • не свободен от препятствий: если по какой-то причине поток блокируется при блокировке переменной, все остальные ожидающие потоки также блокируются; будьте осторожны, чтобы избежать взаимоблокировок здесь (если у вас есть другие блокировки)
  • хрупкий: сбой после блокировки переменной оставляет вас в затруднительном положении, избегайте операций, которые могут вызвать и / или использовать RAII для «блокировки»

Однако на практике это должно работать относительно хорошо в случае только двух объектов (и, следовательно, одного для блокировки).

Напоследок у меня два замечания:

  • для того, чтобы заблокировать, вы должны иметь возможность определить дозорное значение, 0x01 обычно хорошо работает для указателей.
  • Стандарт C++ не гарантирует, что std::atomic<T*> не будет блокироваться, вы можете проверить это для своей конкретной реализации и платформы с помощью std::atomic<T*>::is_lock_free().
person Matthieu M.    schedule 21.08.2013
comment
Как я уже говорил в вопросе, единственным решением, которое у меня было, было портирование класса AtomicPtr для использования атомарных функций С++ 11 вместо пользовательских функций уровня сборки. Класс AtomicPtr на самом деле создает свою собственную блокировку вращения, вращаясь всякий раз, когда указатель удерживает сигнальное значение. Но большое спасибо за уточнение недостатков предложенного решения. Я использую его в качестве эталона, а не в рабочем коде. Кстати, исходный код уже определял значение дозорного точно так, как вы предложили: template<typename T> const T *AtomicPtr<T>::ATOMIC_NULL((T *)((int)NULL + 1)); - person Ahmed Nassar; 21.08.2013
comment
@AhmedNassar: из соображений переносимости я бы посоветовал вам преобразовать NULL в intptr_t вместо int; int часто составляет всего 32 бита в 64-битном коде. - person Matthieu M.; 21.08.2013

Самое близкое, что вы можете сделать без спин-блокировки:

std::atomic<T> a;
std::atomic<T> b;
a = b.exchange(a);

Что является потокобезопасным для b.

a не может быть одновременно доступен.

person ronag    schedule 21.08.2013
comment
Это не атомарно, потому что std::atomic<T>::exchange() принимает простое значение T, а не объект std::atomic<T>. Следовательно, он сначала вызовет функцию преобразования a в T, а затем передаст ее b.exchange(). - person Ahmed Nassar; 21.08.2013
comment
Я никогда не говорил, что это атомарно для a, это атомарно только для b. Это лучшее, что вы можете достичь без спин-блокировки. - person ronag; 21.08.2013

Вы проверили операцию CAS (сравнение и обмен)?

   std::atomic<T*> v;

      while(!v.compare_exchange_weak(old_value,new_value, std::memory_order_release, memory_order_relaxed))
person bjackfly    schedule 21.08.2013
comment
Но параметр new_value в std::atomic<T*>::compare_exchange_weak() — это голый объект T*, а не объект std::atomic<T*>. Если это голое значение T* ранее было прочитано из объекта std::atomic<T*>, обмен не будет атомарным. Я что-то пропустил здесь? - person Ahmed Nassar; 21.08.2013