C++ 11原子内存排序 - 这是放宽(释放 - 消耗)排序的正确用法吗？

Question

C++ 11原子内存排序 - 这是放宽(释放 - 消耗)排序的正确用法吗？

And*_*ves 38 c++ memory atomic compare-and-swap c++11

我最近使用std :: atomic三重缓冲区为C++ 11创建了一个端口,用作并发同步机制.这种线程同步方法背后的想法是,对于生产者 - 消费者情况,你有一个运行速度更快的生产者,消费者,三重缓冲可以带来一些好处,因为生产者线程不会因为必须等待而"减慢"速度对于消费者.在我的例子中,我有一个物理线程,在~120fps时更新,以及一个以~60fps运行的渲染线程.显然,我希望渲染线程始终能够获得最新状态,但我也知道我将从物理线程中跳过很多帧,因为速率不同.另一方面,我希望我的物理线程保持其不变的更新速率,而不受锁定我的数据的较慢渲染线程的限制.

最初的C代码是由remis-ideas制作的,完整的解释在他的博客中.我鼓励任何有兴趣阅读它的人进一步了解原始实现.

我的实现可以在这里找到.

基本思想是使一个具有3个位置(缓冲区)的数组和一个原子标志进行比较和交换,以定义在任何给定时间哪些数组元素对应于什么状态.这样,只有一个原子变量用于模拟数组的所有3个索引和三重缓冲背后的逻辑.缓冲区的3个位置被命名为Dirty,Clean和Snap.该生产商始终写入脏指数,以及可翻转作家交换肮脏与当前清洁指数.该消费者可以要求一个新的管理单元,其交换与清洁指数当前捕捉指数以获得最新的缓冲区.该消费者总是读取对齐位置的缓冲区.

该标志由8位无符号整数组成,这些位对应于:

(未使用)(新写入)(2x脏)(2x清洁)(2x快照)

newWrite extra bit标志由写入器设置并由读取器清除.读者可以使用它来检查自上次捕捉以来是否有任何写入,如果不是,则不会再次捕捉.可以使用简单的按位运算获得标志和索引.

现在好了代码:

template <typename T>
class TripleBuffer
{

public:

  TripleBuffer<T>();
  TripleBuffer<T>(const T& init);

  // non-copyable behavior
  TripleBuffer<T>(const TripleBuffer<T>&) = delete;
  TripleBuffer<T>& operator=(const TripleBuffer<T>&) = delete;

  T snap() const; // get the current snap to read
  void write(const T newT); // write a new value
  bool newSnap(); // swap to the latest value, if any
  void flipWriter(); // flip writer positions dirty / clean

  T readLast(); // wrapper to read the last available element (newSnap + snap)
  void update(T newT); // wrapper to update with a new element (write + flipWriter)

private:

  bool isNewWrite(uint_fast8_t flags); // check if the newWrite bit is 1
  uint_fast8_t swapSnapWithClean(uint_fast8_t flags); // swap Snap and Clean indexes
  uint_fast8_t newWriteSwapCleanWithDirty(uint_fast8_t flags); // set newWrite to 1 and swap Clean and Dirty indexes

  // 8 bit flags are (unused) (new write) (2x dirty) (2x clean) (2x snap)
  // newWrite   = (flags & 0x40)
  // dirtyIndex = (flags & 0x30) >> 4
  // cleanIndex = (flags & 0xC) >> 2
  // snapIndex  = (flags & 0x3)
  mutable atomic_uint_fast8_t flags;

  T buffer[3];
};

Run Code Online (Sandbox Code Playgroud)

执行:

template <typename T>
TripleBuffer<T>::TripleBuffer(){

  T dummy = T();

  buffer[0] = dummy;
  buffer[1] = dummy;
  buffer[2] = dummy;

  flags.store(0x6, std::memory_order_relaxed); // initially dirty = 0, clean = 1 and snap = 2
}

template <typename T>
TripleBuffer<T>::TripleBuffer(const T& init){

  buffer[0] = init;
  buffer[1] = init;
  buffer[2] = init;

  flags.store(0x6, std::memory_order_relaxed); // initially dirty = 0, clean = 1 and snap = 2
}

template <typename T>
T TripleBuffer<T>::snap() const{

  return buffer[flags.load(std::memory_order_consume) & 0x3]; // read snap index
}

template <typename T>
void TripleBuffer<T>::write(const T newT){

  buffer[(flags.load(std::memory_order_consume) & 0x30) >> 4] = newT; // write into dirty index
}

template <typename T>
bool TripleBuffer<T>::newSnap(){

  uint_fast8_t flagsNow(flags.load(std::memory_order_consume));
  do {
    if( !isNewWrite(flagsNow) ) // nothing new, no need to swap
      return false;
  } while(!flags.compare_exchange_weak(flagsNow,
                                       swapSnapWithClean(flagsNow),
                                       memory_order_release,
                                       memory_order_consume));
  return true;
}

template <typename T>
void TripleBuffer<T>::flipWriter(){

  uint_fast8_t flagsNow(flags.load(std::memory_order_consume));
  while(!flags.compare_exchange_weak(flagsNow,
                                     newWriteSwapCleanWithDirty(flagsNow),
                                     memory_order_release,
                                     memory_order_consume));
}

template <typename T>
T TripleBuffer<T>::readLast(){
    newSnap(); // get most recent value
    return snap(); // return it
}

template <typename T>
void TripleBuffer<T>::update(T newT){
    write(newT); // write new value
    flipWriter(); // change dirty/clean buffer positions for the next update
}

template <typename T>
bool TripleBuffer<T>::isNewWrite(uint_fast8_t flags){
    // check if the newWrite bit is 1
    return ((flags & 0x40) != 0);
}

template <typename T>
uint_fast8_t TripleBuffer<T>::swapSnapWithClean(uint_fast8_t flags){
    // swap snap with clean
    return (flags & 0x30) | ((flags & 0x3) << 2) | ((flags & 0xC) >> 2);
}

template <typename T>
uint_fast8_t TripleBuffer<T>::newWriteSwapCleanWithDirty(uint_fast8_t flags){
    // set newWrite bit to 1 and swap clean with dirty 
    return 0x40 | ((flags & 0xC) << 2) | ((flags & 0x30) >> 2) | (flags & 0x3);
}

Run Code Online (Sandbox Code Playgroud)

如您所见,我决定使用Release-Consume模式进行内存排序.该版本的商店(memory_order_release)确保在当前线程没有写入可以重新排序后的商店.另一方面,Consume确保当前线程中的读取不依赖于当前加载的值,可以在此加载之前重新排序.这确保了在当前线程中可以看到对释放相同原子变量的其他线程中的因变量的写入.

如果我的理解是正确的,因为我只需要原子设置标志,对其他不直接影响标志的变量的操作可以由编译器自由重新排序,允许更多的优化.通过阅读新内存模型上的一些文档,我也意识到这些轻松的原子只会对ARM和POWER等平台产生明显的影响(主要是因为它们而引入).由于我的目标是ARM,我相信我可以从这些操作中受益,并能够将性能提高一点.

现在提问:

我是否正确使用了Release-Consume这个特定问题的轻松排序？

谢谢,

安德烈

PS:对于长篇文章感到抱歉,但我认为需要一些不错的背景来更好地了解问题.

编辑: 实施@Yakk的建议:

修复了flags读取newSnap()和flipWriter()使用直接赋值的问题,因此使用默认值load(std::memory_order_seq_cst).
为清晰起见,将位操作移动到专用功能.
添加了bool返回类型newSnap(),现在在没有新内容时返回false,否则返回true.
使用= delete习惯用法将类定义为不可复制的,因为如果使用复制和赋值构造函数,则它们TripleBuffer是不安全的.

编辑2: 修正了描述,这是不正确的(谢谢@Useless).这是消费者请求新的捕捉,并从捕捉指数(不是"作家")的读取.抱歉分心,感谢Useless指出来.

编辑3:根据@Display Name的建议优化newSnap()和flipriter()运行,有效地消除load()每个循环周期2个冗余.

Answer 1

Ale*_*tin 1

是的，这是memory_order_acquire和memory_order_consume之间的区别，但是当你每秒使用180次左右时你不会注意到它。如果您想知道数字答案，可以使用 m2 = memory_order_consume 运行我的测试。只需将 Producer_or_consumer_Thread 更改为类似的内容：

TripleBuffer <int> tb;

void producer_or_consumer_Thread(void *arg)
{
    struct Arg * a = (struct Arg *) arg;
    bool succeeded = false;
    int i = 0, k, kold = -1, kcur;

    while (a->run)
    {
        while (a->wait) a->is_waiting = true; // busy wait
        if (a->producer)
        {
            i++;
            tb.update(i);
            a->counter[0]++;
        }
        else
        {
            kcur = tb.snap();
            if (kold != -1 && kcur != kold) a->counter[1]++;
            succeeded = tb0.newSnap();
            if (succeeded)
            {
                k = tb.readLast();
                if (kold == -1)
                    kold = k;
                else if (kold = k + 1)
                    kold = k;
                else
                    succeeded = false;
            }
            if (succeeded) a->counter[0]++;   
        }
    }
    a->is_waiting =  true;
}

Run Code Online (Sandbox Code Playgroud)

测试结果：

_#_  __Produced __Consumed _____Total
  1    39258150   19509292   58767442
  2    24598892   14730385   39329277
  3    10615129   10016276   20631405
  4    10617349   10026637   20643986
  5    10600334    9976625   20576959
  6    10624009   10069984   20693993
  7    10609040   10016174   20625214
  8    25864915   15136263   41001178
  9    39847163   19809974   59657137
 10    29981232   16139823   46121055
 11    10555174    9870567   20425741
 12    25975381   15171559   41146940
 13    24311523   14490089   38801612
 14    10512252    9686540   20198792
 15    10520211    9693305   20213516
 16    10523458    9720930   20244388
 17    10576840    9917756   20494596
 18    11048180    9528808   20576988
 19    11500654    9530853   21031507
 20    11264789    9746040   21010829

Run Code Online (Sandbox Code Playgroud)

归档时间：	12 年，7 月前
查看次数：	2843 次
最近记录：	11 年，6 月前