#include <boost/intrusive_ptr.hpp>
#include <boost/atomic.hpp>
class X {
public:
  typedef boost::intrusive_ptr<X> pointer;
  X() : refcount_(0) {}
private:
  mutable boost::atomic<int> refcount_;
  friend void intrusive_ptr_add_ref(const X * x)
  {
    x->refcount_.fetch_add(1, boost::memory_order_relaxed);
  }
  friend void intrusive_ptr_release(const X * x)
  {
    if (x->refcount_.fetch_sub(1, boost::memory_order_release) == 1) {
      boost::atomic_thread_fence(boost::memory_order_acquire);
      delete x;
    }
  }
};

X::pointer x = new X;

Увеличение счетчика ссылок всегда можно сделать с помощью<memory_order_relaxed>: Новые ссылки на объект могут быть сформированы только из существующей ссылки, и передача существующей ссылки из одной нити в другую уже должна обеспечивать любую требуемую синхронизацию.

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

Можно было бы использовать<memory_order_acq_rel>для операции<fetch_sub>, но это приводит к ненужным операциям «приобретения», когда эталонный счетчик еще не достигает нуля и может налагать штраф за производительность.

#include <boost/atomic.hpp>
class spinlock {
private:
  typedef enum {Locked, Unlocked} LockState;
  boost::atomic<LockState> state_;
public:
  spinlock() : state_(Unlocked) {}
  void lock()
  {
    while (state_.exchange(Locked, boost::memory_order_acquire) == Locked) {
      /* busy-wait */
    }
  }
  void unlock()
  {
    state_.store(Unlocked, boost::memory_order_release);
  }
};

spinlock s;
s.lock();
// access data structure here
s.unlock();

Цель блокировки — убедиться, что один доступ к общей структуре данных всегда строго «случается раньше» другого. Использование приобретения / выпуска в блокировке / разблокировке требуется и достаточно, чтобы гарантировать этот заказ.

Правильно было бы написать операцию «замок» следующим образом:

lock()
{
  while (state_.exchange(Locked, boost::memory_order_relaxed) == Locked) {
    /* busy-wait */
  }
  atomic_thread_fence(boost::memory_order_acquire);
}

Однако эта «оптимизация» а) бесполезна, а b) может на самом деле навредить: а) Поскольку поток будет активно вращаться на блокировке, не имеет значения, будет ли он тратить циклы процессора только на операции «обмена» или на бесполезные операции «обмена» и «приобретения». b) Жесткий цикл «обмена» без какой-либо команды синхронизации памяти, введенной посредством операции «приобретение», монополизирует подсистему памяти и ухудшает производительность других компонентов системы.

#include <boost/atomic.hpp>
#include <boost/thread/mutex.hpp>
class X {
public:
  static X * instance()
  {
    X * tmp = instance_.load(boost::memory_order_consume);
    if (!tmp) {
      boost::mutex::scoped_lock guard(instantiation_mutex);
      tmp = instance_.load(boost::memory_order_consume);
      if (!tmp) {
        tmp = new X;
        instance_.store(tmp, boost::memory_order_release);
      }
    }
    return tmp;
  }
private:
  static boost::atomic<X *> instance_;
  static boost::mutex instantiation_mutex;
};
boost::atomic<X *> X::instance_(0);

X * x = X::instance();
// dereference x

Мутекс гарантирует, что только один экземпляр объекта когда-либо создан. Метод<instance>должен гарантировать, что любое отклонение объекта строго «происходит после» создания экземпляра в другой нити. Использование<memory_order_release>после создания и инициализации объекта и<memory_order_consume>до отсылки к объекту обеспечивает эту гарантию.

Было бы допустимо использовать<memory_order_acquire>вместо<memory_order_consume>, но это обеспечивает более сильную гарантию, чем требуется, поскольку заказывать нужно только операции в зависимости от значения указателя.

#include <boost/atomic.hpp>
template<typename T, size_t Size>
class ringbuffer {
public:
  ringbuffer() : head_(0), tail_(0) {}
  bool push(const T & value)
  {
    size_t head = head_.load(boost::memory_order_relaxed);
    size_t next_head = next(head);
    if (next_head == tail_.load(boost::memory_order_acquire))
      return false;
    ring_[head] = value;
    head_.store(next_head, boost::memory_order_release);
    return true;
  }
  bool pop(T & value)
  {
    size_t tail = tail_.load(boost::memory_order_relaxed);
    if (tail == head_.load(boost::memory_order_acquire))
      return false;
    value = ring_[tail];
    tail_.store(next(tail), boost::memory_order_release);
    return true;
  }
private:
  size_t next(size_t current)
  {
    return (current + 1) % Size;
  }
  T ring_[Size];
  boost::atomic<size_t> head_, tail_;
};

ringbuffer<int, 32> r;
// try to insert an element
if (r.push(42)) { /* succeeded */ }
else { /* buffer full */ }
// try to retrieve an element
int value;
if (r.pop(value)) { /* succeeded */ }
else { /* buffer empty */ }

Реализация гарантирует, что кольцевые индексы не «обходят» друг друга, чтобы гарантировать, что никакие элементы не будут потеряны или прочитаны дважды.

Кроме того, он должен гарантировать, что доступ к чтению для конкретного объекта в<pop>"происходит после" он был написан в<push>. Это достигается написанием<head_>с «освобождением» и чтением его с «приобретением». И наоборот, реализация также гарантирует, что доступ считывания к определенному кольцевому элементу «происходит раньше», прежде чем переписывать этот элемент с новым значением, получая доступ<tail_>с соответствующими ограничениями упорядочения.

template<typename T>
class waitfree_queue {
public:
  struct node {
    T data;
    node * next;
  };
  void push(const T &data)
  {
    node * n = new node;
    n->data = data;
    node * stale_head = head_.load(boost::memory_order_relaxed);
    do {
      n->next = stale_head;
    } while (!head_.compare_exchange_weak(stale_head, n, boost::memory_order_release));
  }
  node * pop_all(void)
  {
    T * last = pop_all_reverse(), * first = 0;
    while(last) {
      T * tmp = last;
      last = last->next;
      tmp->next = first;
      first = tmp;
    }
    return first;
  }
  waitfree_queue() : head_(0) {}
  // alternative interface if ordering is of no importance
  node * pop_all_reverse(void)
  {
    return head_.exchange(0, boost::memory_order_consume);
  }
private:
  boost::atomic<node *> head_;
};

waitfree_queue<int> q;
// insert elements
q.push(42);
q.push(2);
// pop elements
waitfree_queue<int>::node * x = q.pop_all()
while(x) {
  X * tmp = x;
  x = x->next;
  // process tmp->data, probably delete it afterwards
  delete tmp;
}

Реализация гарантирует, что все объекты, стоящие в очереди, обрабатываются в том порядке, в котором они были поставлены в очередь, создавая единый список объектов в порядке обратной обработки. Очередь атомарно опорожняется потребителем и приводится в правильный порядок.

Необходимо гарантировать, что любой доступ к объекту, который будет поставлен в очередь производителем, «происходит раньше» любого доступа потребителя. Это обеспечивается путем вставки объектов в список свысвобождениеми очерчивания их спотреблениемпорядка памяти. Необязательно использовать, чтобы получитьпорядок памяти в<waitfree_queue::pop_all>, потому что все задействованные операции зависят от значения атомного указателя посредством отсчета.