很多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



伪共享问题

现代CPU通常通过cache来加速内存的访问效果，例如arm 的MESI协议，但是我们不能轻易的忽视cacheline导致的代码问题。本文介绍一下cacheline导致的代码问题。
# 原理
arm系列cpu的cacheline通常是64字节，如果结构体的数据跨cacheline，MESI协议只能够有效的从Poc和Pou视角保证变量的一致性。跨cacheline的那个数据成员其实并没办法很好的保证其值在每时每刻的正确性。参考：《cache的策略》，但是这种跨cacheline的行为（跨CPU）还是有可能出现数据和理论情况不一致的情况。

为了构造上面的问题，我需要构造一个60字节+8字节的结构体，如下
```
#pragma pack(1)
struct data {
    int32_t pad[15];
    int64_t v;
};
static struct data value __attribute__((aligned(64)));
```
这种情况下，pad默认填充了cacheline的前60字节，而v变量正好处于cacheline的前4字节和后4字节

为了构造cacheline导致的数据异常的案例，我还需要将其放在线程里面运行，这样data结构体的实例化必须是全局的。

注意：这里 pack(1) 是强制按照1字节对齐，通过aligned(64)保证了value结构体一直在cacheline的开头

到这里基本上我的用意很明显了，我需要有一个线程运行函数，如下
```
static void worker(int *cnt) {
    for (int64_t i = 0; i < loop_count; ++i) {
        const int64_t t = value.v;

        *cnt += 1;

        value.v = ~t;
        __asm__ volatile("" ::: "memory");
    }
}
```
值得注意的是，为了避免编译器给我指令重排，我需要为其添加内存栅栏
```
__asm__ volatile("" ::: "memory");
```
这里的输入部和输出部都是空，所以此汇编只是简单的声明内存栅栏


# 运行
测试程序需要两个输入参数，一个是线程的个数，为了让这个任务在多个CPU上运行，另一个是线程将value.v反转的次数。
```
    int64_t n = min(atol(argv[1]), N_THREADS);
    loop_count = atol(argv[2]);
```

根据worker函数是一个非常简单的逻辑，就是简单的将value.v不断反转。这个代码运行理论上的值应该是
```
FFFFFFFFFFFFFFFF
0000000000000000
```

那实际上呢？ 我们运行看看
```
# ./cacheline_bug 8 10000
iteration: 80000
data size: 68
final: FFFFFFFFFFFFFFFF
# ./cacheline_bug 8 10000
iteration: 80000
data size: 68
final: 0000000000000000
# ./cacheline_bug 8 10000
iteration: 80000
data size: 68
final: 00000000FFFFFFFF
# ./cacheline_bug 8 10000
iteration: 80000
data size: 68
final: FFFFFFFF00000000
```
可以看到，v的值有四种情况，这里
```
00000000FFFFFFFF
FFFFFFFF00000000
```
就是cache带来的隐藏问题。
# 总结
在开发高性能的程序的时候，一方面需要利用好cache的特性，另一方面在存在数据竞争的场景下也需要考虑周全，否则就容易出现cache带来的隐藏bug。

# 源码
下面分享一下测试的源码，本文分享基于arm V8 上运行
```
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

#ifndef min
#define min(x, y) (x) < (y) ? (x) : (y)
#endif

#define N_THREADS 128
static int64_t loop_count = 0;

#pragma pack(1)
struct data {
    int32_t pad[15];
    int64_t v;
};

static struct data value __attribute__((aligned(64)));
static int64_t counter[N_THREADS];

static void worker(int *cnt) {
    for (int64_t i = 0; i < loop_count; ++i) {
        const int64_t t = value.v;
        *cnt += 1;

        value.v = ~t;
        /* creates a compiler level memory barrier
         * forcing optimizer to not re-order memory
         * accesses across the barrier. */
        __asm__ volatile("" ::: "memory");
    }
}

int main(int argc, char *argv[]) {
    pthread_t threads[N_THREADS];

    if (argc != 3) {
        fprintf(stderr, "USAGE: %s <threads> <loopcount>\n", argv[0]);
        return 1;
    }

    int64_t n = min(atol(argv[1]), N_THREADS);
    loop_count = atol(argv[2]);

    for (int64_t i = 0L; i < n; ++i)
        pthread_create(&threads[i], NULL, (void *(*) (void *) ) worker,
                       &counter[i]);

    int64_t count = 0L;
    for (int64_t i = 0L; i < n; ++i) {
        pthread_join(threads[i], NULL);
        count += counter[i];
    }

    printf("iteration: %lu\n", count);
    printf("data size: %lu\n", sizeof(value));
    printf("final: %016lX\n", value.v);

    return 0;
}
```














