在开发高性能的C程序的时候，经常使用环形缓冲区，我们生活中常见的使用环形缓冲区的例子就是声卡。本文介绍如何开发环形缓冲区
# Circular buffer
首先我们看个动画

![Circular_Buffer_Animation.gif](/static/img/846878d6005f23f9f15afde8fe0ae028.Circular_Buffer_Animation.gif)

读写指针一直在环中打转，如果写指针追上读指针的时候，相当于环形缓冲区的内容消耗完毕，那么程序将等待。

在生活的例子中就是我们听到的声音卡在这里了，如果体现到声卡的行为上，如果是存在bug，那就是underrun，也就是pcm没办法从应用层获取数据，使得驱动的ringbuffer空了，声卡“饿晕”了。

如果以生活的成语来说，就是赛跑游戏

# 原理
首先启动的时候，buffer是空的，如下

![image.png](/static/img/70381cd78d5af42b1f99212630ee15f5.image.webp)

当put第一个数据的时候，可以是buffer的任意位置

![image.png](/static/img/4706d466635eaf18f5c22b8be27c4fd7.image.webp)

当继续put数据的时候，那么会跟在第一个数据的位置后面，因为tail指针会移动

![image.png](/static/img/30e630c0d555e157b4a156a9a4bc04ce.image.webp)

如果get数据的时候，buffer默认构造成FIFO结构，所以先来先服务，head指针会移动

![image.png](/static/img/d06ba360242733f769cfd143a2c81e77.image.webp)

长此以往，buffer可能一直在赛跑状态，也有可能是buffer 满的状态，如下

![image.png](/static/img/aa79384182964451054169d8416abbee.image.webp)

如果buffer满的时候，还是不断的put数据进去，那么tail指针会覆盖原来的位置的数据，它们是形成环的。
假设在buffer满的时候，get指针不动的前提上，put了A和B，那么A和B就会覆盖原来的数据3和4

![image.png](/static/img/d624913e937bb260b5927e39cce1340b.image.webp)

在这种情况下，如果get指针获取了两个元素，因为get指针来自head，所以实际上获取的数据是5和6，因为此时5和6是最老的数据

![image.png](/static/img/03079b9669890116ecdef06b12c9c604.image.webp)

# 测试程序
实现环形缓冲区有四大因素，如下
* buffer start
* buffer size
* write buffer index
* read buffer index

例如如下
```
struct ring_buffer {
  void *addr;
  uint32_t size;
  uint32_t head;
  uint32_t tail;
}
```
这里有一个最简单的ringbuffer示例，如下
```
#include <stdio.h>

enum { N = 10 };  // size of circular buffer

int buffer [N]; // note: only (N - 1) elements can be stored at a given time
int writeIndx = 0;
int readIndx  = 0;

int put (int item) 
{
  if ((writeIndx + 1) % N == readIndx)
  {
     // buffer is full, avoid overflow
     return 0;
  }
  buffer[writeIndx] = item;
  writeIndx = (writeIndx + 1) % N;
  return 1;
}

int get (int * value) 
{
  if (readIndx == writeIndx)
  {
     // buffer is empty
     return 0;
  }

  *value = buffer[readIndx];
  readIndx = (readIndx + 1) % N;
  return 1;
}

int main ()
{
  // test circular buffer
  int value = 1001;
  while (put (value ++));
  while (get (& value))
     printf ("read %d\n", value);
  return 0;
}
```
编译运行如下
```
read 1001
read 1002
read 1003
read 1004
read 1005
read 1006
read 1007
read 1008
read 1009
```

# 高性能优化
其实根据上面的测试代码的`writeIndx = (writeIndx + 1) % N;`可以看到，每次判断index的时候用到了取余操作，这在C语言世界是非常低效的。因为CPU对于取余来说指令非常耗时。所以这是第一个优化点

其次，普通实现的环形缓冲区会出现多次拷贝，如下
```
uint32_t circular_buffer_write(struct ring_buffer *rb, void *buf, uint32_t count)
    memcpy(rb->addr + rb->tail, buf, size);
    if(size < count){
        memcpy(rb->addr, buf + size, count - size);
    }
    rb->tail = (rb->tail + count) % rb->size;
    return count;
}
```
可以看到，如果按照普通的实现，ringbuffer在边界上需要处理两次memcpy，这是第二个优化点

解决这两个优化点有什么办法呢？

我们回到现代CPU设计的 虚拟内存和物理内存的概念上，现代CPU设计的时候，可以将一个物理地址分别由两个虚拟地址映射，也就意味着，如果我实际操作的是一个物理地址 `0x1000-0x2000`，但是我申请了两个连续的虚拟地址 `0xffff1000-0xffff2000`和`0xffff2000-0xffff3000`。这里`0xffff1000-0xffff2000`和`0xffff2000-0xffff3000`都是映射在同一个物理地址范围`0x1000-0x2000`。

那么ringbuffer就可以同时解决上述两个点的优化项

1. 不需要通过取模指令计算边界，直接index超越边界即可，超越边界实际上物理地址已经成环，我们只需要时候通过减法调整回边界，如`head -= BUFFER_SIZE;`或者`tail -= BUFFER_SIZE;`
2. 不需要两次memcpy调用，直接省掉一次memcpy的调用可以搞定。

# 总结
本文介绍环形缓冲区的设计和优化方法，这在开发高性能程序过程中至关重要。这种优化项利用了现代操作系统特性做的取巧，可以获得不错的程序性能提升，而且方案成熟。











高性能C程序-环形缓冲区

很多C语言开发者经常错误的使用 volatile 关键字，本文用于矫正 volatile 的错误进行说明
# volatile 的作用
volatile加在变量前面，可以告诉编译器，此变量可能被意外的修改，所以要求编译器每次获取改值的时候，总是从内存拿最新的那份。

可以看到，volatile 本身就是在抑制编译器的正向优化，这就使得加上volatile关键字的程序性能就会偏低，这还是其次，更重要的事情是 volatile 并不能保证CPU不会重排它，它和内存栅栏是完全不同的概念。 

也就是说，如果CPU乱序重排，那么程序就会带来意想不到的问题。本文就是介绍 volatile 带来的程序问题

# 示例
《高性能无锁编程-Dekker算法》介绍了高性能的无锁编程的一种思路，这种办法在业界已经广泛得到认可。它通过设置了flag和turn来实施“孔融让梨”的思想。

那是不是意味着，flag和turn变量用 volatile 让编译器每次从内存拿最新的数据就能有效避免flag和turn数据不一致的情况呢？

关于flag和turn数据可能不一致的情况说明，可以进一步看《不可忽视的cacheline问题》文章

先说结论： volatile 只能保证获取内存最新值，但是如果load/store 可以被CPU重排，那么每次load操作读取到的仍是旧值，这样代码就会出现问题。 

所以千万要注意，volatile 和 内存栅栏 不是一回事。内存栅栏能够有效的隔绝CPU和内存的介限（DMB/DSB/ISB），而 volatile 只能保证读取最新内存的值。

下面查看演示代码
```
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

static volatile int flag0 = 0, flag1 = 0, turn = 1;
static volatile int counter = 0;
int test_case = 0;
int loop_cnt;

static void dekker0(void)
{
    flag0 = 1;
    turn = 1;
    if (test_case == 1)
        __atomic_thread_fence(__ATOMIC_SEQ_CST);
    if (test_case == 2)
        __asm__ __volatile__("dmb ish" ::: "memory");

    while ((flag1 == 1) && (turn == 1));

    counter++;
    flag0 = 0;
}

static void dekker1(void)
{
    flag1 = 1;
    turn = 0;
    if (test_case == 1)
         __atomic_thread_fence(__ATOMIC_SEQ_CST);
    if (test_case == 2)
        __asm__ __volatile__("dmb ish" ::: "memory");

    while ((flag0 == 1) && (turn == 0));

    counter++;
    flag1 = 0;
}

static void *task0(void *arg)
{
    printf("Starting %s\n", __func__);
    for (int i = loop_cnt; i > 0; i--)
        dekker0();
    return NULL;
}

static void *task1(void *arg)
{
    printf("Starting %s\n", __func__);
    for (int i = loop_cnt; i > 0; i--)
        dekker1();
    return NULL;
}

int main(int argc, char **argv)
{
    pthread_t thread1, thread2;

    if (argc != 3) {
        fprintf(stderr, "Usage: %s <loopcount> <0(none)/1(c11_fence)/2(arm64_dmb)>\n", argv[0]);
        exit(1);
    }

    loop_cnt = atoi(argv[1]);
    test_case = atoi(argv[2]);
    int expected_sum = 2 * loop_cnt;

    (void) pthread_create(&thread1, NULL, task0, NULL);
    (void) pthread_create(&thread2, NULL, task1, NULL);

    void *ret;
    (void) pthread_join(thread1, &ret);
    (void) pthread_join(thread2, &ret);
    printf("Both threads terminated\n");

    /* Check result */
    if (counter != expected_sum) {
        printf("[-] Dekker did not work, sum %d rather than %d.\n", counter,
               expected_sum);
        printf("%d missed updates due to memory consistency races.\n",
               (expected_sum - counter));
        return 1;
    }
    printf("[+] Dekker worked.\n");
    return 0;
}

```
编译
```
gcc volatile_bug.c -o volatile_bug
```
运行
```
# ./volatile_bug 1000000 0
Starting task0
Starting task1
Both threads terminated
[-] Dekker did not work, sum 1999986 rather than 2000000.
14 missed updates due to memory consistency races.
```
可以看到，此代码运行多次，dekker算法好像直接失效了。其主要原因是volatile关键字的错误理解

解决此问题可以使用c11的thread fence进行内存同步，它相当于构造了内存栅栏，强制同步了内存。
```
__atomic_thread_fence(__ATOMIC_SEQ_CST);
```
此时能够保证CPU不再乱序（因为内存栅栏的存在），所以代码运行如下
```
# ./volatile_bug 1000000 1
Starting task1
Starting task0
Both threads terminated
[+] Dekker worked.
```

如果在arm平台使用，应该使用
```
dmb + ish
```
dmb构造了内存栅栏，而ish让所有CPU的视角上内存表现一致。
所以其运行结果如下
```
# ./volatile_bug 1000000 2
Starting task0
Starting task1
Both threads terminated
[+] Dekker worked.
```
# 总结
本文介绍了 volatile 关键字非常注意的两个点
1. 性能低下
2. 不能保证CPU乱序

所以在开发程序的时候，尽可能的避免volatile关键字。

同样的，内核也在极力的纠正大部分开发者错误的使用 volatile 关键字，详情可参考下面的连接

# 参考
内核并不建议使用 volatile 关键字： 
> https://www.kernel.org/doc/Documentation/process/volatile-considered-harmful.rst

dmb： 
> https://www.scs.stanford.edu/~zyedidia/arm64/dmb.html

高性能程序volatile的错误使用

并发编程过程中的同步操作往往是通过锁来实现，实际上在业界可以通过Dekker这种无锁同步算法来实现高性能程序的通信问题。  

Dekker是并发编程中仅通过共享内存通信的互斥操作来解决通信的重要算法，它有效的避免了互斥，死锁，饿死三种情况。

# 原理
中文有个词语能够很精妙的解释这个算法
* 孔融让梨

什么意思呢，当两个进程/线程尝试访问临界区域的时候，它们在已经获取到资源的时候，尝试先让给对立的资源，如果对立的资源此时想要运行，那就让对立的资源运行，如果对立的资源没有想要运行，那么自己再运行。

可以看看，这种思想是不是完全的孔融让梨。当获取到梨的时候，并不想着里面吃掉，而是让给大哥吃。

说回dekker，其工作流程如下

![image.png](/static/img/4f31c5ea91789296802fd4ccdf700b4c.image.webp)

# 测试程序
实现dekker算法的原型非常简单，我们只需要为每个线程提供一个flag，flag置位就代表自己获取了资源，再为两个线程提供一个全局的让梨标志位turn，这样在获取资源的时候先让给别人
```
static void dekker0(void)
{
    flag0 = 1;
    turn = 1;
   
    while ((flag1 == 1) && (turn == 1));
    counter++;

    flag0 = 0;
}
static void dekker1(void)
{
    flag1 = 1;
    turn = 0;

    while ((flag0 == 1) && (turn == 0));
    counter++;
    flag1 = 0;
}

```
上面代码可以看到，当dekker0得到运行的时候，它在开始和结束将flag0置位和清除，然后将梨(turn) 设置为别人的1，自己等待，如果别人不吃（while循环），自己在做counter++的操作。

同样的，对于dekker1也是如此

# 解释
可以看到，这种设计原则上，没有使用任何的锁同步机制，其原理上可以保证互斥，死锁，饿死。下面逐一解释
## 避免互斥
因为每次线程得以运行的时候，都是先尝试让给对方，而不是自己里面运行，所以天然就避免了互斥
## 死锁
这种设计里面没有锁，每次进入临界区的时候（counter++）即使turn被错误的置位给自己，下次无论轮到自己还是别人运行，都会尝试把turn让给别人，所以也不会发生死锁
## 饿死
假设dekker0在某种情况下被卡在 `while ((flag1 == 1) && (turn == 1));`上，那么等dekker1得到资源运行的时候，dekker1会将turn设置为0 ，为dekker0解围，所以dekker0永远不会饿死。

# 运行
运行这个代码很简单，创建两个任务，直接运行即可
```
    (void) pthread_create(&thread1, NULL, task0, NULL);
    (void) pthread_create(&thread2, NULL, task1, NULL);
```

# 总结
Dekker算法就是计算机领域的孔融让梨，它是一种wait-free的设计哲学，它的优势在于不需要任何特殊的原子读/修改/写的操作，所以基于此算法的代码行为非常高效。

注意，你可能看到当资源来到的时候让给对立资源的短暂收益是低效的(梨让给别人吃)，但是其代码全局确实高效的(无需原子操作)。但是当CAS指令诞生之后，高性能程序可以转向CAS指令型设计。

# 参考
> https://en.wikipedia.org/wiki/Dekker%27s_algorithm





高性能无锁编程-Dekker算法

本文简单汇总MESI和MOESI协议理解。

# MESI
* M： Modified

在当前CPU上是属于被修改过的，其他CPU尚未同步，内存的值已经过时

* E：Exclusive

独占的，其他CPU尚未同步，内存的值和当前独占的这份相同，如果其他CPU在bus上发 snooping，则同步给其他CPU，自己转成 Shared

* S：Shared

和其他CPU上的内容共享，并且内存的值和当前CPU已经其他CPU上的cache值一致

* I：Invalid

无效的，自己这份cache已经无效了，snooping自动会更新


MESI的状态机如下：
1. 自己的cache更新状态机

![image.png](/static/img/4f0a1ef7fc394566b1d590d3a5a678e6.image.webp)

2. snooping下的cache更新状态机

![image.png](/static/img/90f8cecfbfbc9860ea47544da0d7ba4a.image.webp)

arm上有几个bit来查看cache状态，如下

![image.png](/static/img/2f6313d3407c69ced696b488e1c3a156.image.webp)

# MOESI
相比于MESI多了个O，是从S状态的一种分化，O指的是Owner。所以S和O的概念会有刷新
* O：Owner

和Share类似，其他CPU上也是共享的，但是这份数据的Owner是当前CPU的cache，与Share不一样的是，O标志的更新，所以内存的值可能是旧的

* S：Shared

完全的共享状态，所以CPU都是当前这份内容，内存中的值是否是cache的值看snooping的情况

1. 自己的cache更新状态机

![image.png](/static/img/6fd8da813ef7cd9c184ff3e455f64cf5.image.webp)

2. snooping下的cache更新状态机

![image.png](/static/img/71a7ec92c43361f0e160cc9e0a20d6e9.image.webp)

arm上有几个bit来查看cache状态，如下

![image.png](/static/img/5feded5a202915e104a79947b94fd3ec.image.webp)

# 参考
> https://en.wikipedia.org/wiki/MESI_protocol
> https://en.wikipedia.org/wiki/MOESI_protocol

MESI和MOESI协议简介

cacheline导致的bug可以参考《不可忽视的cacheline问题》，本文介绍一个不是属于运行错误的例子，其属于性能损失，cache的伪共享，这类问题比cacheline导致的bug而言更容易出现，也是必须要解决的。

# False sharing
简单来说就是MESI和MOESI协议下，如果一个数据反复的在访问同一个cacheline，那么MOESI就会对bus上的数据进行invalidate，并在bus上进行snooping到每个CPU上。这种行为对于CPU的视角来看是低效的。相当于无意义的吃掉了内存的带宽，浪费了系统资源。

# 测试代码
测试代码来自参考里面，如下
```
#include <iostream>
#include <thread>
#include <new>
#include <atomic>
#include <chrono>
#include <latch>
#include <vector>

using namespace std;
using namespace chrono;

#if defined(__cpp_lib_hardware_interference_size)
// default cacheline size from runtime
constexpr size_t CL_SIZE = hardware_constructive_interference_size;
#else
// most common cacheline size otherwise
constexpr size_t CL_SIZE = 64;
#endif

int main()
{
    vector<jthread> threads;
    int hc = jthread::hardware_concurrency();
    hc = hc <= CL_SIZE ? hc : CL_SIZE;
    for (int nThreads = 1; nThreads <= hc; ++nThreads)
    {
        // synchronize beginning of threads coarse on kernel level
        latch coarseSync(nThreads);
        // fine synch via atomic in userspace
        atomic_uint fineSync(nThreads);
        // as much chars as would fit into a cacheline
        struct alignas(CL_SIZE) { char shareds[CL_SIZE]; } cacheLine;
        // sum of all threads execution times
        atomic_int64_t nsSum(0);
        for (int t = 0; t != nThreads; ++t)
            threads.emplace_back(
                [&](char volatile &c)
                {
                    coarseSync.arrive_and_wait(); // synch beginning of thread execution on kernel-level
                    if (fineSync.fetch_sub(1, memory_order::relaxed) != 1) // fine-synch on user-level
                        while (fineSync.load(memory_order::relaxed));
                    auto start = high_resolution_clock::now();
                    for (size_t r = 10'000'000; r--;)
                        c = c + 1;
                    nsSum += duration_cast<nanoseconds>(high_resolution_clock::now() - start).count();
                }, ref(cacheLine.shareds[t]));
        threads.resize(0); // join all threads
        cout << nThreads << ": " << (int)(nsSum / (1.0e7 * nThreads) + 0.5) << endl;
    }
}
```
其思路就是构造一个cacheline大小的数组
```
struct alignas(CL_SIZE) { char shareds[CL_SIZE]; } cacheLine;
```
然后对齐自加
```
            threads.emplace_back(
                [&](char volatile &c)
                {
                    for (size_t r = 10'000'000; r--;)
                        c = c + 1;
                }, ref(cacheLine.shareds[t]));
```
通过不断的增加线程来查看伪共享的负面效果
```
 for (int nThreads = 1; nThreads <= hc; ++nThreads)
```
# 测试
编译运行
```
g++ -std=c++20 falsesharing.cpp -o falsesharing
```
运行
```
# ./falsesharing
1: 3
2: 7
3: 12
4: 13
5: 14
6: 19
7: 21
8: 29
9: 30
10: 34
11: 37
12: 39
13: 41
14: 47
15: 49
16: 54
```
可以看到，随着多线程的加剧，这种伪共享带来的性能损失来到了 54 / 3 = 18倍。

# 如何解决
官方的解决方案如下：
1. 重新排序变量或在变量之间添加填充（未使用的字节）来防止 CPU 缓存中的虚假共享
2. 编译时数据转换还可以减少虚假共享

其实最简单直接的办法就是每个线程使用不同的数据，这样就规避了多个线程访问同一个cacheline的问题。
按照我的思路，直接让所有线程访问不同的cacheline，运行结果如下
```
# ./fixfalsesharing
1: 3
2: 3
3: 3
4: 3
5: 3
6: 2
7: 4
8: 3
9: 3
10: 3
11: 3
12: 3
13: 3
14: 3
15: 3
16: 3
```
# 总结
在编写多线程或者并发程序的时候，相比于cachelinebug《不可忽视的cacheline问题》，伪共享其实是更容易遇到和需要避免的问题，只要编程时有这样的印象，只需要很简单的让多线程不竞争同一个cacheline即可。
# 参考
> https://en.wikipedia.org/wiki/False_sharing

