写这篇博文主要源于几个月前的一条微博，大概讲的是“在一些数据结构中，需要修改某个数据，对整个数据结构加锁实现，然而大部分时间是在查找，这里把写锁分为seek和write两部分，拆开...”，然后带着好奇心翻外去找了相关的资料和论文，以及代码实现，在进行分析和测试上面的例子后，早就想记录下却发现肚子里硬货不够，知识不够统一全面，故查了好些资料学习并记录关键点在多线程相关一，加之之前对dpdk锁的分析DPDK中的同步与互斥，也只停留在代码实现表面，不知道为什么这么写以及如何更好的应用到项目中，所以这里深入分析原理并记录相关联的知识，比较综合一些。

本篇内容主要如下：
1) CACHE MESI一致性协议；
2) 内存屏障实现原理；内存可见性；
3) C++中volatile关键字的作用；
4) Acquire/Release语意相关；
5) 互斥锁，自旋锁，读写锁以及CAS的实现源码及优化；
6) 多读一写模型下的优化方向(brpc/rcu)；
7) 原微博的具体内容(替换成中文标点了)；

这里不会解释cache的工作原理，但是空间和时间局部性这块还是要在编程中注意些，可以参考下《深入理解计算机系统》相关的章节。

对于一些源码实现中如dpdk，都会看到cache对齐的要求，或者把读写分在不同的cache line上。这一方面防止cache为了保证一致性不断invalid其它core的cache，形成广播风暴等性能问题；二是诸如false sharing的问题；三是加载cache line时读内存的次数，避免跨多个cache line加载多次；还有cache miss等局部性原理。

对于多线程开发中，如果有读和修改共享数据，这里尽量把读和写分在不同的cache line上，且每个线程修改的数据字段也同其他线程分在不同的cache line上，防止a线程修改a[1]，b线程修改a[2]数据这样的情况。

一) CACHE MESI一致性协议

cache mesi用于保证各core间的cache数据一致性，由于现代cpu架构等其他原因，会有L1/L2/L3 cache用于加速cpu存取数据，减少内存的访问，根据局部性原理缓存一部分数据，当同一cache line被加载到不同的core上的cache中时，就会出现不一致。一般L1(L2)为各core私有，而该协议是由硬件替我们处理的，对mesi协议也有一定的扩展，这里不作说明，可以参考相关资料。

MESI 协议有四种状态分别是：
M(Modified) 这行数据有效，数据被修改了，和内存中的数据不一致，数据只存在于本Cache中。
E(Exclusive) 这行数据有效，数据和内存中的数据一致，数据只存在于本Cache中。
S(Shared) 这行数据有效，数据和内存中的数据一致，数据存在于很多Cache中。
I(Invalid) 这行数据无效。

以下引用连接中的状态转换过程：

MESI状态转换图

然后一些行为操作和转换过程可以参考大话处理器 Cache一致性协议之MESI中的分析。

在项目开发过程中，稍微不注意的时候就可能会引起性能问题，可能很难发现。如上图，当core 0写本地cache line中某个int数据时，若此cache line状态为s则表示其他core也缓存了该cache line，那么控制器就需要invalid其他core上的cache line，这可能发多条消息，并且等待其他core的回应，再写cache，此时就会出现性能问题。cpu不可能等，所以为了优化，分别出现了store buffer和invalidate queues，但另一方面却引入其他问题，会在第二小节分析。

以上是硬件自动处理的，然后我们在开发的时候，还是要注意其他的问题，比如在DPDK中无锁环形队列实现中，当考虑的场景是两个线程，分别是生产者和消费者，那么如何使用无锁去实现高效的任务队列呢？

简化那个无锁环形队列的生产者如下：
根据读索引计算，放数据；rte_smp_wmb()；//a 更新写索引；

消费者如下：
根据写索引计算，取数据；rte_smp_rmb()；//b 更新读索引；

全程无加锁解锁的代码实现，那么为什么可以保证正确性？我们来一起研究下上面a/b这一简单的代码到底隐藏了什么？

二) 内存屏障实现原理

#define rte_smp_wmb() rte_wmb()
#define rte_wmb() do { asm volatile ("dmb st" : : : "memory"); } while (0)  //c
#define rte_smp_rmb() rte_rmb()
#define rte_rmb() __sync_synchronize() //full barrier

这两个分别是写屏障和读屏障，如实现注释所言：
47 * General memory barrier.
49 * Guarantees that the LOAD and STORE operations generated before the
50 * barrier occur before the LOAD and STORE operations generated after.

55 * Write memory barrier.
57 * Guarantees that the STORE operations generated before the barrier
58 * occur before the STORE operations generated after.

而c语句实现则非常关键，引用网上资料作用如下：

1. asm用于指示编译器在此插入汇编语句；
2. volatile 相当于C语言的volatile(见后面的分析)；
3. memory强制gcc编译器假设RAM所有内存单元均被汇编指令修改，这样cpu中的registers和cache中已缓存的内存单元中的数据 将作废。cpu将不得不在需要的时候重新读取内存中的数据。这就阻止了cpu又将registers，cache中的数据用于去优化指令，而避 免去访问内存；
4. "":::表示这是个空指令。barrier()不用在此插入一条串行化汇编指令；

一般的实现如上所示__asm__ __volatile__("":::"memory")，但dpdk中的有点区别dmb st，查了相关资料表示的意思是：
“数据内存屏障可作为内存屏障使用。 它可确保会先检测到程序中位于 DMB 指令前的所有显式内存访问指令，然后再检测到程序中位于DMB 指令后的显式内存访问指令。它不影响其他指令在处理器上的执行顺序。”
wiki Memory ordering
asm volatile("": : :"memory")

回到第一小节的最后出现的store buffer和invalidate queues作用，这里先引用一张图：

CPU 架构

还有张有点区别：

CPU 架构

即load buffer所在位置不同，得好好查些官方资料证实一。

由于编译器的优化导致指令重排和处理器的乱序执行，有的时候咱们写的程序执行流程，最后可能的结果跟原来的不一样。cpu的乱序执行意味着什么？比如当在core 0上执行a = 1, b = 1时，如果a不在core 0的cache line上，而b在core 0上且状态是e，当a = 1时，则会发invalid消息给其他有该cache line的core，core invalid自己的cache line后给core 0回应，但此时core 0需要等待全部回应(Invalidate Ack)，再进行a = 1操作，此时会有性能问题，所以先把a = 1写入store buffer，就执行b = 1，这样导致a = 1晚于b = 1执行，所以可能跟我们写的流程不一样。

当然这里举的例子没有什么依赖关系，拿连接中有依赖关系的例子说明下，考虑a和b默认都为0：
core 0：
a = 1; b = 1;
c = a + 1;
core 1：
while (b != 1); assert(a == 1);
当a不在core 0的cache line中，b不在core 1的cache line中时，就会出现问题，究其原因还是因为store buffer和invalidate queues。这里当core 1执行assert(a == 1)时出现失败，即可能a=0。什么情况下会出现？当core 0执行a = 1时，会invalid有cache a的所有core，此时收到invalid后直接回invalid ack并且把这条消息放在了invalidate queues中，并没有及时执行，而此时cache line中的a还是0，所以取到的是无效值。如果在取a时先执行invalidate queues中的指令或去先看一下有没有无效的a，有的话则重新read。那如何解决呢？

这类似于可见性，看见的是旧值，当请求的最新值时，另外的core 可能还没完成写，即可能这一步的写操作还停留在store buffer，还没同步到cache中。所以这里我们需要加上内存屏障相关的指令，来告诉core怎么做。包括读取a的值也是，可能使用到a的值是其他cache同步过来的a，而不是store buffer中最新的a。

引用wiki上对store buffer和invalidate queues的解释说明：
Store Buffer
A store buffer is used when writing to an invalid cache line. Since the write will proceed anyway, the CPU issues a read-invalid message (hence the cache line in question and all other CPUs' cache lines which store that memory address are invalidated) and then pushes the write into the store buffer, to be executed when the cache line finally arrives in the cache.

A direct consequence of the store buffer's existence is that when a CPU commits a write, that write is not immediately written in the cache. Therefore, whenever a CPU needs to read a cache line, it first has to scan its own store buffer for the existence of the same line, as there is a possibility that the same line was written by the same CPU before but hasn't yet been written in the cache (the preceding write is still waiting in the store buffer). Note that while a CPU can read its own previous writes in its store buffer, other CPUs cannot see those writes before they are flushed from the store buffer to the cache - a CPU cannot scan the store buffer of other CPUs.

Invalidate Queues

With regard to invalidation messages, CPUs implement invalidate queues, whereby incoming invalidate requests are instantly acknowledged but not in fact acted upon. Instead, invalidation messages simply enter an invalidation queue and their processing occurs as soon as possible (but not necessarily instantly). Consequently, a CPU can be oblivious to the fact that a cache line in its cache is actually invalid, as the invalidation queue contains invalidations which have been received but haven't yet been applied. Note that, unlike the store buffer, the CPU can't scan the invalidation queue, as that CPU and the invalidation queue are physically located on opposite sides of the cache.

As a result, memory barriers are required. A store barrier will flush the store buffer, ensuring all writes have been applied to that CPU's cache. A read barrier will flush the invalidation queue, thus ensuring that all writes by other CPUs become visible to the flushing CPU. Furthermore, memory management units do not scan the store buffer, causing similar problems. This effect is already visible in single threaded processors.

对应到上面的写屏障和读屏障，即store memory fence和load memory fence，他们作用分别是：
“A store barrier will flush the store-buffer (ensuring all writes have entered that CPUs cache). A read barrier will flush the invalidation queue (ensuring all writes by other CPUs become visible to the flushing CPU)”
其中sfence和lfence自带release和acquire语意。见第四小节。

那volatile能不能用于两个线程间的同步？

三) C++中volatile关键字的作用

此volatile非Java中的volatile，两者截然不同，这里不分析Java中的作用。
连接中C/C++ Volatile关键词深度剖析讲的很清楚，这里简单列一下几个特别，并为什么不能用于同步及原子操作。有兴趣的可以写几个demo反汇编看下。
volatile有三个特点：易变的，即cpu每次需要从内存中加载；不可优化的，即不能被编译器优化掉；顺序的，即相对于volatile间的变量定义，顺序是不能交换的，但对于非volatile和volatile之间顺序则可能会被编译器指令重排，那基于这一点，volatile用于线程间同步可能出现问题。

这里摘取dpdk中自旋锁的实现来说明volatile和原子操作的实现，以及为什么这么做是正确的。

 58 typedef struct {
 59     volatile int locked; /**< lock status 0 = unlocked, 1 = locked */
 60 } rte_spinlock_t;

 89 static inline void
 90 rte_spinlock_lock(rte_spinlock_t *sl)
 91 {
 92     while (__sync_lock_test_and_set(&sl->locked, 1))
 93         while(sl->locked)
 94             rte_pause();
 95 }
 96 #endif

290 /**         
291  * PAUSE instruction for tight loops (avoid busy waiting)
292  */
293 static inline void
294 rte_pause (void)
295 {
296     _mm_pause();
297 }

上面修饰locked的volatile关键字是必须的，__sync_lock_test_and_set是个原子操作，相关源码没有找到，还是看一下glibc-2.27中的spin lock实现：

 21 int
 22 pthread_spin_lock (pthread_spinlock_t *lock)
 23 {
 24   unsigned int __tmp;
 25 
 26   asm volatile (
 27        "1:  lwarx   %0,0,%1" MUTEX_HINT_ACQ "\n"
 28        "    cmpwi   0,%0,0\n"
 29        "    bne-    2f\n"
 30        "    stwcx.  %2,0,%1\n"
 31        "    bne-    2f\n"
 32                 __ARCH_ACQ_INSTR "\n"
 33        "    .subsection 1\n"
 34        "2:  lwzx    %0,0,%1\n"
 35        "    cmpwi   0,%0,0\n"
 36        "    bne 2b\n"
 37        "    b   1b\n"
 38        "    .previous"
 39        : "=&r" (__tmp)
 40        : "r" (lock), "r" (1)
 41        : "cr0", "memory");
 42   return 0;
 43 }

基本上会锁总线，这一部分比较复杂，我看的也不是很明白，类似的实现可以参考下后面的连接，是否正确不大好判断。spin lock用户态加锁源码分析

以上，为什么使用锁后，就能保证正确性，而不需要考虑到处理器的乱序执行以及内存屏障等相关的事情？

四) Acquire/Release语意相关

其中锁自带了Lock Acquire和Unlock Release语义，即：
Acquire语义禁止read-acquire之后所有的内存读写操作，被提前到read-acquire操作之前进行。
Release语义禁止write-release之前所有的内存读写操作，被推迟到write-release操作之后进行，如图一：

图一

五) 互斥锁，自旋锁，读写锁以及CAS的实现源码及优化

如上小节解释的，锁自带了acquire和release，所以用起来的时候不需要考虑上面说的情况，但也还是会造成死锁，或者性能问题。这小节列出几外常用锁的实现，主要是参考dpdk中的。另个提一点，用自旋锁时不能sleep，否则造成死锁，可以参考这里：
Linux内核spin_lock与spin_lock_irq分析
中断上下文使用spin_lock使导致死锁案例分析
why-cant-you-sleep-while-holding-spinlock
spinlock中不允许休眠调

一些锁的实现源码：

 61 typedef struct {
 62     volatile int32_t cnt; /**< -1 when W lock held, > 0 when R locks held. */
 63 } rte_rwlock_t;

 88 static inline void
 89 rte_rwlock_read_lock(rte_rwlock_t *rwl)
 90 {
 91     int32_t x;
 92     int success = 0;
 93 
 94     while (success == 0) {
 95         x = rwl->cnt;
 96         /* write lock is held */
 97         if (x < 0) {
 98             rte_pause();
 99             continue;
100         }
101         success = rte_atomic32_cmpset((volatile uint32_t *)&rwl->cnt,
102                           x, x + 1);
103     }
104 }

124 static inline void
125 rte_rwlock_write_lock(rte_rwlock_t *rwl)
126 {
127     int32_t x;
128     int success = 0;
129 
130     while (success == 0) {
131         x = rwl->cnt;
132         /* a lock is held */
133         if (x != 0) {
134             rte_pause();
135             continue;
136         }
137         success = rte_atomic32_cmpset((volatile uint32_t *)&rwl->cnt,
138                           0, -1);
139     }
140 }

390 rte_atomic32_cmpset(volatile uint32_t *dst, uint32_t exp, uint32_t src)
391 {                         
392     return __sync_bool_compare_and_swap(dst, exp, src);
393 }

139 static inline int
140 rte_atomic32_cmpset(volatile uint32_t *dst, uint32_t exp, uint32_t src)
141 {                         
142     uint8_t res;
143 
144     asm volatile(
145             MPLOCKED
146             "cmpxchgl %[src], %[dst];"
147             "sete %[res];"
148             : [res] "=a" (res),     /* output */
149               [dst] "=m" (*dst)
150             : [src] "r" (src),      /* input */
151               "a" (exp),
152               "m" (*dst)
153             : "memory");            /* no-clobber list */
154     return res;
155 }

以上操作会锁总线，“汇编指令是 cmpxchg， 这是一条原子操作指令，也就是说CPU 会保证这条指令执行的原子性。亦即，在特定的一个时间点，当多个cpu core 同时执行这条指令(参数相同)，只会有一个core 成功的执行swap的操作。”dpdk 无锁队列详细分解——Lock与Cache，到底有没有锁？

290 /**
291  * PAUSE instruction for tight loops (avoid busy waiting)
292  */
293 static inline void
294 rte_pause (void)
295 {
296     _mm_pause();
297 }
298 #else
299 static inline void
300 rte_pause(void) {}
301 #endif

其中rte_pause指令是一种优化，作用就是减少并行load的数量，从而减少reorder时所耗时间，原理参考这里提升自旋锁spinlock的性能-pause指令

自己经历过的业务场景中，自旋锁和互斥锁用的多一些。
待解决的问题：没有考虑到读写锁的优先关系，防止某一方饿死的情况。

六) 多读一写模型下的优化方向(brpc/rcu)

我上一个项目做的是软件防火墙，专门用于抗DDoS攻击流量的。记得当时项目中有个功能，就是主线程通过后台命令终端(前端界面还没开发完成)，添加/删除过滤ip配置黑白名单等信息。然后由于是多线程那种框架，启动的时候是主线程加工作线程那种，配置信息只添加一份，多个工作线程读配置。

对应到这里的主题是一个写线程会更新配置，多个读线程只读配置，而且写操作很少进行，那么怎么安全的性能更好些的修改配置呢？

场景如上面所述，多读一写，不考虑读和写的优先级，性能要高且安全。这里列出几个类似的方案，一种是来自百度开源的brpc中的，一种是书上《Linux 多线程服务端编程》(copy-on-other-reading)，还有一种是RCU(read-copy-update)，思路都是差不多的。总体是在保证正确性的前提下，减少锁竞争，提高性能。

copy-on-other-reading做法
引用书上的例子作为参考，大致实现代码如下：

  5 class CustomerData : boost::noncopyable {
  6 public:
  7     int query(const std::string &customer, const std::string &stock) const;
  8 
  9     void update(const std::string &customer, const EntryList &entries);
 10  
 11     MapPtr getData() const {
 12         MutexLockGuard lock(mutex_); //保护读操作取到的data_是有效的
 13         return data_;
 14     }
 15  
 16 private:
 17     typedef std::map<std::string> Map;
 18     typedef boost::shared_ptr<Map> MapPtr; //shared_ptr自带原子计数器
 19     MutexLockGuard mutex_;       
 20     MapPtr data_;
 21 };

 23 int CustomerData::query(const std::string &customer, const std::string &stock) const {
 24     MapPtr data = getData();
 25    
 26     //do read
 27     return 0;
 28 }
 29 
 30 void CustomerData::update(const std::string &customer, const EntryList &entries) {
 31     MutexLockGuard lock(mutex_);
 32     if (!data_.unique()) {
 33         MapPtr newData(new Map(*data_));
 34         data_.swap(newData);
 35     }           
 36                 
 37     assert(data_.unique());
 38     //do update
 39 }

以上，确保读的时候，用到的data_始终有效，虽然数据可以有一定延迟的旧数据。当多读到来时，可能在getData内部会竞争同一把锁，最后计数是读的个数，如果不考虑写，其实锁可以去掉，因为shared_ptr内部用于计数的内存已经是原子的，多读是线程安全的。当然，这里是为了防止写破坏data_，并且有其他读线程的时候，阻止了写线程直接修改。可以参考shared_ptr线程安全性分析和为什么多线程读写 shared_ptr 要加锁？

对于写线程，在执行update时，是需要加锁整个过程的，这里判断是否有其他的读线程，如果没有则直接更新。否则拷贝一份，并修改，再交换。

百度的brpc做法
实现数据结构是DoublyBufferedData，引用连接处的一段描述：
我们需要写以某种形式和读同步，但读之间相互没竞争。一种解法是，读拿一把thread-local锁，写需要拿到所有的thread-local锁。具体过程如下：

数据分前台和后台。
读拿到自己所在线程的thread-local锁，执行查询逻辑后释放锁。
同时只有一个写：修改后台数据，切换前后台，挨个获得所有thread-local锁并立刻释放，结束后再改一遍新后台（老前台）。
我们来分析下这个方法的基本原理：

当一个读正在发生时，它会拿着所在线程的thread-local锁，这把锁会挡住同时进行的写，从而保证前台数据不会被修改。
在大部分时候thread-local锁都没有竞争，对性能影响很小。
逐个获取thread-local锁并立刻释放是为了确保对应的读线程看到了切换后的新前台。如果所有的读线程都看到了新前台，写线程便可以安全地修改老前台（新后台）了。

其他特点：
不同的读之间没有竞争，高度并发。
如果没有写，读总是能无竞争地获取和释放thread-local锁，一般小于25ns，对延时基本无影响。如果有写，由于其临界区极小（拿到立刻释放），读在大部分时候仍能快速地获得锁，少数时候释放锁时可能有唤醒写线程的代价。由于写本身就是少数情况，读整体上几乎不会碰到竞争锁。

以下是截取的部分源码实现，详细的可以参考这里。

202 template <typename T, typename TLS>
203 class DoublyBufferedData<T, TLS>::Wrapper
204     : public DoublyBufferedDataWrapperBase<T, TLS> {
205 friend class DoublyBufferedData;
206 public:
207     explicit Wrapper(DoublyBufferedData* c) : _control(c) {
208         pthread_mutex_init(&_mutex, NULL);
209     }
210 
211     ~Wrapper() {
212         if (_control != NULL) {
213             _control->RemoveWrapper(this);
214         }
215         pthread_mutex_destroy(&_mutex);
216     }
217 
218     // _mutex will be locked by the calling pthread and DoublyBufferedData.
219     // Most of the time, no modifications are done, so the mutex is
220     // uncontended and fast.
221     inline void BeginRead() { //读线程加锁
222         pthread_mutex_lock(&_mutex);
223     }
224 
225     inline void EndRead() { //读线程解锁
226         pthread_mutex_unlock(&_mutex);
227     }
228 
229     inline void WaitReadDone() { //写线程加锁，函数结束自动解锁
230         BAIDU_SCOPED_LOCK(_mutex);
231     }
232 
233 private:
234     DoublyBufferedData* _control;
235     pthread_mutex_t _mutex;
236 };

以上是每个读线程拥有的Wrapper，不知如何用中文叫，在这里相当于只有一把钥匙的!潘多拉盒吧。

239 template <typename T, typename TLS>
240 typename DoublyBufferedData<T, TLS>::Wrapper*
241 DoublyBufferedData<T, TLS>::AddWrapper() {
242     std::unique_ptr<Wrapper> w(new (std::nothrow) Wrapper(this));
243     if (NULL == w) {
244         return NULL;
245     }
246     try {
247         BAIDU_SCOPED_LOCK(_wrappers_mutex);
248         _wrappers.push_back(w.get());
249     } catch (std::exception& e) {
250         return NULL;
251     }
252     return w.release();
253 }

创建一个并push到管理器中，需要加锁。
简化后的管理器如下：

 52 template <typename T, typename TLS = Void>
 53 class DoublyBufferedData {
 54     class Wrapper;
 55 public:

 83     int Read(ScopedPtr* ptr);
 93     size_t Modify(Fn& fn, const Arg1&, const Arg2&);
162     const T* UnsafeRead() const
163     { return _data + _index.load(butil::memory_order_acquire); }

167     // Foreground and background void.
168     T _data[2];  //配置数据相关
169 
170     // Index of foreground instance.
171     butil::atomic<int> _index;  //哪个是最新的，即前后台配置
172 
173     // Key to access thread-local wrappers.
174     bool _created_key;
175     pthread_key_t _wrapper_key;
176 
177     // All thread-local instances.
178     std::vector<Wrapper*> _wrappers;
179 
180     // Sequence access to _wrappers.
181     pthread_mutex_t _wrappers_mutex;  //用于互斥修改_wrappers
182 
183     // Sequence modifications.
184     pthread_mutex_t _modify_mutex; //用于互斥多个写线程
185 };

316 template <typename T, typename TLS>
317 int DoublyBufferedData<T, TLS>::Read(
318     typename DoublyBufferedData<T, TLS>::ScopedPtr* ptr) {
319     if (BAIDU_UNLIKELY(!_created_key)) {
320         return -1;
321     }
322     Wrapper* w = static_cast<Wrapper*>(pthread_getspecific(_wrapper_key));
323     if (BAIDU_LIKELY(w != NULL)) {
324         w->BeginRead();
325         ptr->_data = UnsafeRead();
326         ptr->_w = w;
327         return 0;
328     }
329     w = AddWrapper();
330     if (BAIDU_LIKELY(w != NULL)) {
331         const int rc = pthread_setspecific(_wrapper_key, w);
332         if (rc == 0) {
333             w->BeginRead();
334             ptr->_data = UnsafeRead();
335             ptr->_w = w;
336             return 0;
337         }
338     }
339     return -1;
340 }

以上是获取配置，第一次获取时会配置一下该线程自己数据，然后加锁，等读完后在ScopedPtr析构的时候解锁。

342 template <typename T, typename TLS>
343 template <typename Fn>
344 size_t DoublyBufferedData<T, TLS>::Modify(Fn& fn) {
345     // _modify_mutex sequences modifications. Using a separate mutex rather
346     // than _wrappers_mutex is to avoid blocking threads calling
347     // AddWrapper() or RemoveWrapper() too long. Most of the time, modifications
348     // are done by one thread, contention should be negligible.
349     BAIDU_SCOPED_LOCK(_modify_mutex);
350     int bg_index = !_index.load(butil::memory_order_relaxed);
351     // background instance is not accessed by other threads, being safe to
352     // modify.
353     const size_t ret = fn(_data[bg_index]);
354     if (!ret) {
355         return 0;
356     }
357 
358     // Publish, flip background and foreground.
359     // The release fence matches with the acquire fence in UnsafeRead() to
360     // make readers which just begin to read the new foreground instance see
361     // all changes made in fn.
362     _index.store(bg_index, butil::memory_order_release);
363     bg_index = !bg_index;
364 
365     // Wait until all threads finishes current reading. When they begin next
366     // read, they should see updated _index.
367     {   
368         BAIDU_SCOPED_LOCK(_wrappers_mutex);
369         for (size_t i = 0; i < _wrappers.size(); ++i) {
370             _wrappers[i]->WaitReadDone();
371         }
372     }
373 
374     const size_t ret2 = fn(_data[bg_index]);
375     CHECK_EQ(ret2, ret) << "index=" << _index.load(butil::memory_order_relaxed);
376     return ret2;
377 }

以上是写线程更新，由于就一个写线程，先获取后台配置，并进行更新操作。然后交换前后台配置，那么后面如果有其他读线程则会获取到最新的前台配置，如果有读线程正在读旧的配置，那么暂时会有些延迟。此时写线程不能更新旧的配置，所以需要WaitReadDone，等每一个过了一遍后，就能确保再也没有读线程使用旧的配置了，此时写线程再进行旧配置的更新操作。

比如上面两种实现方案，第二种效率更高。因为竞争只存在于读和写之间，且读和写的临界区为空，读和读之间没有竞争。

这里使用到了C++内存模型相关的知识点。

RCU(read-copy-update)做法
由于本篇幅内容过多，就不写了，以下是几个连接资料，有兴趣可以自己研究下。
RCU synchronize原理分析
Linux内核同步机制之（七）：RCU基础
Linux2.6.11版本：classic RCU的实现

七) 原微博的具体内容

问题:

1. 类radix tree以及其他的数据结构，read和write互斥
2. read占绝大多数
3. read比write更快，read L1 cache命中比例很高
4. write的过程中主要时候花在seek (read)上。 这样导致的问题就是一点点write操作影响了大量的read。

解决方法:

1. 把write的过程拆分为两步，第一步seek，第二步write，seek的时候不加write lock，从而可以和read操作并发
2. write阶段升级成write lock
3. 增加seek lock，write操作的seek阶段拿seek lock，write阶段升级成write lock，seek lock可以和read lock并发，和其他lock互斥，所以seek lock可以简单升级成write lock。(seek lock (upgrade to) -> write 优于 read lock -> write)
4. 优化lock时需要的atomic操作，pthread rwlock需要2次atomic，优化后只需要一次。

测试结果：所有情况几乎都有优势，竞争大的情况甚至能比rwlock或者spinlock性能好5-8倍，代码已经opensource，我们可以看看X-Engine的cache是不是有同样的问题。

因为前几节内容有些多，这小节会更多，具体论文和源码以及测试数据我会在后面补上，作为补充。

另外在分析leveldb基础库的时候，里面的cache缓存进行了分片，每个分片都有一把锁，这样降低竞争。后面准备开始完整的leveldb之旅，分析leveldb的实现原理以及不足，看下自己有没有能力找到更合适的方案对这些不足点进行优化，而不是一上来分析rocksdb，虽然后者比前者更好些。

建议多看下连接中引用的资料和wiki，或者是些知名的开源项目，可能这里纯理论性的较多，需要多学习和实践。

参考资料：
C/C++ Volatile关键词深度剖析
并发编程系列之一：锁的意义
wiki MESI protocol
Parallel Labs
cpu 乱序执行与问题
store buffer and invalidate queues
如何理解 C++11 的六种 memory order
Memory Model: 从多处理器到高级语言
Store Forwarding
Memory barrier
Linux中的线程局部存储
提升自旋锁spinlock的性能-pause指令
高性能服务端系列 -- 处理器篇

基数树(radix tree)
查找——图文翔解RadixTree（基数树）