现代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;
}
```
















不可忽视的cacheline问题

编写高性能的C程序的时候经常需要设计状态机，通过设计状态机可以完成复杂系统的业务逻辑编程实现，状态机可以管理状态和转换，使代码更容易理解和维护，不过很多时候，大家编写的状态机程序都存在一些性能问题，本文聚焦高性能C语言状态机的实现进行说明
# 什么是状态机
状态机也称有限状态机（Finite State Machine, FSM），其最初是一种数学模型，用于描述系统在不同状态之间根据输入进行转换的行为。

假设需要设计一个开关灯的程序，可以通过如下方式来说明状态机
```
        +------------------+
        |                  |
        v      Toggle      |
     +------+          +------+
     | Off  | -------->| On   |
     +------+ <--------+------+
        ^                  |
        |      Toggle     |
        +------------------+
```
根据上面这个最简单的状态机可以总结如下几个必要的要素
* 状态集和： off/on
* 初始状态： off
* 输入： toggle
* 状态转移： off->on

针对这些必要条件，下面列出高性能状态机的几个细节，基于C语言
# 通过枚举实现状态集和
高性能的状态机的必要添加之一就是枚举，通过枚举来实现状态集和相比于代码中的幻数而言是最轻松最容易管理的。
一个简单的状态集和可以如下
```
typedef enum {
    STATE_INIT,
    STATE_RUNNING,
    STATE_STOPPED,
    STATE_ERROR
} State;

State currentState = STATE_INIT;
```
# 高效的状态转移
大部分不关注性能的状态机喜欢使用if来进行判断，例如
```
if(state == STATE_INIT) {
   # TODO
} else if(state == STATE_RUNNING) {
   # TODO
}
```
但是if通常对于cpu的预测而言并不友好，这样在处理状态转换时候会有极大性能损失。可以考虑使用switch-case结构。这种方法通常更快更高效。编译器在处理switch语句上具有最高的性能
```
void handleState(State state) {
    switch (state) {
        case STATE_INIT:
            // Initialization logic
            break;
        case STATE_RUNNING:
            // Running logic
            break;
        case STATE_STOPPED:
            // Stopped logic
            break;
        case STATE_ERROR:
            // Error handling
            break;
        default:
            // Handle unexpected state
            break;
    }
}
```
额外信息：有兴趣的可以了解 达夫设备：https://www.gnu.org/software/c-intro-and-ref/manual/html_node/Duffs-Device.html
# 更高效的状态转移
switch已经满足绝大部分情况下的状态转移，如果追求极高的性能，可以直接避免编译器做任何的分支预测，直接使用函数指针代替分支预测形式的语法。例如
```
void initState() {
    // Initialization logic
}

void runningState() {
    // Running logic
}

void stoppedState() {
    // Stopped logic
}

void errorState() {
    // Error handling
}

void (*stateHandlers[])() = {
    initState,
    runningState,
    stoppedState,
    errorState
};

void handleState(State state) {
    stateHandlers[state]();
}
```
通过 `handleState`函数直接调用函数指针，代码在运行的时候，直接可以通过state的局部变量参数进行函数指针的偏移，从而直接定位到确定的函数调用上。

不过需要注意的是，绝大部分情况下，switch-case已经是最高效的状态机设计方案了。如果状态机状态集和过多，函数指针的优势才能体现。所以选择switch-case还是函数指针并没有什么约束，看个人爱好
# 最小化状态转移
状态机内部的状态转移经常容易出现重复状态的转移，这种很容易变成不轻易被发现的性能问题，当任务从`STATE_RUNNING`切换到`STATE_RUNNING`时，状态转移函数应该准确的识别，避免不必要的性能消耗，如下
```
void transitionTo(State newState) {
    if (currentState != newState) {
        currentState = newState;
        // Additional logic for state change
    }
}
```
# 极致内存占用
状态集和的状态，尽可能的时候位处理的方式实现，通过一个整数来表达多个状态，是内存友好型的行为。linux内核经常这么使用，如下
```
#define STATE_INIT    (1 << 0)
#define STATE_RUNNING (1 << 1)
#define STATE_STOPPED (1 << 2)
#define STATE_ERROR   (1 << 3)

int currentStateFlags = 0;

void setState(int state) {
    currentStateFlags |= state;
}

void clearState(int state) {
    currentStateFlags &= ~state;
}

bool isStateActive(int state) {
    return (currentStateFlags & state) != 0;
}
```
# 禁用动态内存
使用动态内存申请的方式获取数据结构会导致任务的高性能依赖操作系统的内存管理，如果需要开发极高性能的程序状态机，此时建议直接禁止动态内存申请。全部改用静态内存分配。

通过静态分配内存，可以有效的避免动态分配的开销和抖动。

但是另一方面又需要注意的是，一些非状态机外的场景，使用动态内存可以有效的节省内存使用。此情况额外分析。这里只谈涉及高性能状态机内的场景
```
typedef struct {
    State state;
    // Other state-related data
} StateMachine;

StateMachine sm = {STATE_INIT};
```
# 总结
通过使用枚举、优化过渡逻辑、利用函数指针、最小化状态变化以及有效管理内存，可以显著提升状态机的性能。

如果自己在编写状态机时需要抄作业，下面提供了一个开源的高效状态机的仓库，可以直接搬运。
> https://github.com/endurodave/C_StateMachine






高性能C的状态机优化

Arm默认提供硬件crc的指令集，这样在操作系统上的很多校验操作都可以使用crc指令完成，它对于高性能程序开发和校验来说不可或缺。例如网络，媒体，存储，文件系统等等。
# crc是什么
循环冗余校验(Cyclic Redundancy Check)可以检查数据是否出现意外错误。它将一串数据看作一个二进制多项式，然后用一个预定义的“生成多项式”去除它，得到的余数就是 CRC 的校验值。

如果通过软件crc完成，那么cpu会突然暴涨，而arm cpu默认就带crc的硬件功能，我们直接使用就行。

我们在例如操纵系统打包，内存冗余校验都可以用crc完成。
# 查询crc功能
内核cpuinfo就能查询crc32的功能，如下
```
root@ubuntu:~# cat /proc/cpuinfo | grep crc32
Features        : fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm lrcpc dcpop asimddp uscat ilrcpc flagm paca pacg
Features        : fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm lrcpc dcpop asimddp uscat ilrcpc flagm paca pacg
Features        : fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm lrcpc dcpop asimddp uscat ilrcpc flagm paca pacg
Features        : fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm lrcpc dcpop asimddp uscat ilrcpc flagm paca pacg
Features        : fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm lrcpc dcpop asimddp uscat ilrcpc flagm paca pacg
Features        : fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm lrcpc dcpop asimddp uscat ilrcpc flagm paca pacg
```
编译器已经默认打开crc32的选项，无需重新构建编译器
```
armv8-a+crc
```
# 验证完整性
例如操作系统校验，我们可以取系统镜像的前1M，后面随机再取1M进行crc校验，这样能够有效的在不浪费内存加载的前提上进程完整性校验。所以直接提供api实现即可。

软件实现如下
```
static uint32_t crc32_software(const uint8_t *data, size_t len)
{
    uint32_t crc = 0xFFFFFFFFU;
    for (size_t i = 0; i < len; ++i) {
        crc = crc32_table[(crc ^ data[i]) & 0xFF] ^ (crc >> 8);
    }
    return crc ^ 0xFFFFFFFFU;
}
```
硬件实现如下
```
static uint32_t crc32_hardware(const uint8_t *buf, size_t len)
{
    uint32_t crc = 0xFFFFFFFFU;

    while (len >= 8) {
        uint64_t v = *(const uint64_t *)buf;
        crc = __crc32d(crc, v);
        buf += 8;
        len -= 8;
    }

    while (len--) {
        crc = __crc32b(crc, *buf++);
    }

    return crc ^ 0xFFFFFFFFU;
}
```
对应的查表如下
```
static const uint32_t crc32_table[256] = {
    0x00000000, 0x77073096, 0xEE0E612C, 0x990951BA, 0x076DC419, 0x706AF48F, 0xE963A535, 0x9E6495A3,
    0x0EDB8832, 0x79DCB8A4, 0xE0D5E91E, 0x97D2D988, 0x09B64C2B, 0x7EB17CBD, 0xE7B82D07, 0x90BF1D91,
    0x1DB71064, 0x6AB020F2, 0xF3B97148, 0x84BE41DE, 0x1ADAD47D, 0x6DDDE4EB, 0xF4D4B551, 0x83D385C7,
    0x136C9856, 0x646BA8C0, 0xFD62F97A, 0x8A65C9EC, 0x14015C4F, 0x63066CD9, 0xFA0F3D63, 0x8D080DF5,
    0x3B6E20C8, 0x4C69105E, 0xD56041E4, 0xA2677172, 0x3C03E4D1, 0x4B04D447, 0xD20D85FD, 0xA50AB56B,
    0x35B5A8FA, 0x42B2986C, 0xDBBBC9D6, 0xACBCF940, 0x32D86CE3, 0x45DF5C75, 0xDCD60DCF, 0xABD13D59,
    0x26D930AC, 0x51DE003A, 0xC8D75180, 0xBFD06116, 0x21B4F4B5, 0x56B3C423, 0xCFBA9599, 0xB8BDA50F,
    0x2802B89E, 0x5F058808, 0xC60CD9B2, 0xB10BE924, 0x2F6F7C87, 0x58684C11, 0xC1611DAB, 0xB6662D3D,
    0x76DC4190, 0x01DB7106, 0x98D220BC, 0xEFD5102A, 0x71B18589, 0x06B6B51F, 0x9FBFE4A5, 0xE8B8D433,
    0x7807C9A2, 0x0F00F934, 0x9609A88E, 0xE10E9818, 0x7F6A0DBB, 0x086D3D2D, 0x91646C97, 0xE6635C01,
    0x6B6B51F4, 0x1C6C6162, 0x856530D8, 0xF262004E, 0x6C0695ED, 0x1B01A57B, 0x8208F4C1, 0xF50FC457,
    0x65B0D9C6, 0x12B7E950, 0x8BBEB8EA, 0xFCB9887C, 0x62DD1DDF, 0x15DA2D49, 0x8CD37CF3, 0xFBD44C65,
    0x4DB26158, 0x3AB551CE, 0xA3BC0074, 0xD4BB30E2, 0x4ADFA541, 0x3DD895D7, 0xA4D1C46D, 0xD3D6F4FB,
    0x4369E96A, 0x346ED9FC, 0xAD678846, 0xDA60B8D0, 0x44042D73, 0x33031DE5, 0xAA0A4C5F, 0xDD0D7CC9,
    0x5005713C, 0x270241AA, 0xBE0B1010, 0xC90C2086, 0x5768B525, 0x206F85B3, 0xB966D409, 0xCE61E49F,
    0x5EDEF90E, 0x29D9C998, 0xB0D09822, 0xC7D7A8B4, 0x59B33D17, 0x2EB40D81, 0xB7BD5C3B, 0xC0BA6CAD,
    0xEDB88320, 0x9ABFB3B6, 0x03B6E20C, 0x74B1D29A, 0xEAD54739, 0x9DD277AF, 0x04DB2615, 0x73DC1683,
    0xE3630B12, 0x94643B84, 0x0D6D6A3E, 0x7A6A5AA8, 0xE40ECF0B, 0x9309FF9D, 0x0A00AE27, 0x7D079EB1,
    0xF00F9344, 0x8708A3D2, 0x1E01F268, 0x6906C2FE, 0xF762575D, 0x806567CB, 0x196C3671, 0x6E6B06E7,
    0xFED41B76, 0x89D32BE0, 0x10DA7A5A, 0x67DD4ACC, 0xF9B9DF6F, 0x8EBEEFF9, 0x17B7BE43, 0x60B08ED5,
    0xD6D6A3E8, 0xA1D1937E, 0x38D8C2C4, 0x4FDFF252, 0xD1BB67F1, 0xA6BC5767, 0x3FB506DD, 0x48B2364B,
    0xD80D2BDA, 0xAF0A1B4C, 0x36034AF6, 0x41047A60, 0xDF60EFC3, 0xA867DF55, 0x316E8EEF, 0x4669BE79,
    0xCB61B38C, 0xBC66831A, 0x256FD2A0, 0x5268E236, 0xCC0C7795, 0xBB0B4703, 0x220216B9, 0x5505262F,
    0xC5BA3BBE, 0xB2BD0B28, 0x2BB45A92, 0x5CB36A04, 0xC2D7FFA7, 0xB5D0CF31, 0x2CD99E8B, 0x5BDEAE1D,
    0x9B64C2B0, 0xEC63F226, 0x756AA39C, 0x026D930A, 0x9C0906A9, 0xEB0E363F, 0x72076785, 0x05005713,
    0x95BF4A82, 0xE2B87A14, 0x7BB12BAE, 0x0CB61B38, 0x92D28E9B, 0xE5D5BE0D, 0x7CDCEFB7, 0x0BDBDF21,
    0x86D3D2D4, 0xF1D4E242, 0x68DDB3F8, 0x1FDA836E, 0x81BE16CD, 0xF6B9265B, 0x6FB077E1, 0x18B74777,
    0x88085AE6, 0xFF0F6A70, 0x66063BCA, 0x11010B5C, 0x8F659EFF, 0xF862AE69, 0x616BFFD3, 0x166CCF45,
    0xA00AE278, 0xD70DD2EE, 0x4E048354, 0x3903B3C2, 0xA7672661, 0xD06016F7, 0x4969474D, 0x3E6E77DB,
    0xAED16A4A, 0xD9D65ADC, 0x40DF0B66, 0x37D83BF0, 0xA9BCAE53, 0xDEBB9EC5, 0x47B2CF7F, 0x30B5FFE9,
    0xBDBDF21C, 0xCABAC28A, 0x53B39330, 0x24B4A3A6, 0xBAD03605, 0xCDD70693, 0x54DE5729, 0x23D967BF,
    0xB3667A2E, 0xC4614AB8, 0x5D681B02, 0x2A6F2B94, 0xB40BBE37, 0xC30C8EA1, 0x5A05DF1B, 0x2D02EF8D
};
```
这样直接调用函数crc32_software或crc32_hardware即可。
# 测试
为了满足测试要求，可以编写测试程序验证，并编译，编译如下
```
gcc -O3 -march=armv8-a+crc -o test test.c
```
以测试操作系统完整性为例，为了方便测试，这里直接加载全部操作系统，当然，也可以选择部分片段来做完整性校验
```
# ls -lh rootfs.img
-rw-r--r-- 1 root root 2.0G 12月  8 23:26 rootfs.img
# ./test rootfs.img
Software  : 8.740870 s  →  0.23 GB/s   CRC32 = 0xfd94b9a9
Hardware  : 0.236808 s  →  8.45 GB/s   CRC32 = 0xfd94b9a9
Speedup   : 36.9x
Verify    : PASS
```
可以看到，相比于软件crc校验，crc计算的时间提高了36.9倍。
# 测试代码
如下是测试代码，可以直接使用
```
#include <stdio.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <time.h>
#include <arm_acle.h>

#define ITERATIONS 10

static const uint32_t crc32_table[256] = {
    0x00000000, 0x77073096, 0xEE0E612C, 0x990951BA, 0x076DC419, 0x706AF48F, 0xE963A535, 0x9E6495A3,
    0x0EDB8832, 0x79DCB8A4, 0xE0D5E91E, 0x97D2D988, 0x09B64C2B, 0x7EB17CBD, 0xE7B82D07, 0x90BF1D91,
    0x1DB71064, 0x6AB020F2, 0xF3B97148, 0x84BE41DE, 0x1ADAD47D, 0x6DDDE4EB, 0xF4D4B551, 0x83D385C7,
    0x136C9856, 0x646BA8C0, 0xFD62F97A, 0x8A65C9EC, 0x14015C4F, 0x63066CD9, 0xFA0F3D63, 0x8D080DF5,
    0x3B6E20C8, 0x4C69105E, 0xD56041E4, 0xA2677172, 0x3C03E4D1, 0x4B04D447, 0xD20D85FD, 0xA50AB56B,
    0x35B5A8FA, 0x42B2986C, 0xDBBBC9D6, 0xACBCF940, 0x32D86CE3, 0x45DF5C75, 0xDCD60DCF, 0xABD13D59,
    0x26D930AC, 0x51DE003A, 0xC8D75180, 0xBFD06116, 0x21B4F4B5, 0x56B3C423, 0xCFBA9599, 0xB8BDA50F,
    0x2802B89E, 0x5F058808, 0xC60CD9B2, 0xB10BE924, 0x2F6F7C87, 0x58684C11, 0xC1611DAB, 0xB6662D3D,
    0x76DC4190, 0x01DB7106, 0x98D220BC, 0xEFD5102A, 0x71B18589, 0x06B6B51F, 0x9FBFE4A5, 0xE8B8D433,
    0x7807C9A2, 0x0F00F934, 0x9609A88E, 0xE10E9818, 0x7F6A0DBB, 0x086D3D2D, 0x91646C97, 0xE6635C01,
    0x6B6B51F4, 0x1C6C6162, 0x856530D8, 0xF262004E, 0x6C0695ED, 0x1B01A57B, 0x8208F4C1, 0xF50FC457,
    0x65B0D9C6, 0x12B7E950, 0x8BBEB8EA, 0xFCB9887C, 0x62DD1DDF, 0x15DA2D49, 0x8CD37CF3, 0xFBD44C65,
    0x4DB26158, 0x3AB551CE, 0xA3BC0074, 0xD4BB30E2, 0x4ADFA541, 0x3DD895D7, 0xA4D1C46D, 0xD3D6F4FB,
    0x4369E96A, 0x346ED9FC, 0xAD678846, 0xDA60B8D0, 0x44042D73, 0x33031DE5, 0xAA0A4C5F, 0xDD0D7CC9,
    0x5005713C, 0x270241AA, 0xBE0B1010, 0xC90C2086, 0x5768B525, 0x206F85B3, 0xB966D409, 0xCE61E49F,
    0x5EDEF90E, 0x29D9C998, 0xB0D09822, 0xC7D7A8B4, 0x59B33D17, 0x2EB40D81, 0xB7BD5C3B, 0xC0BA6CAD,
    0xEDB88320, 0x9ABFB3B6, 0x03B6E20C, 0x74B1D29A, 0xEAD54739, 0x9DD277AF, 0x04DB2615, 0x73DC1683,
    0xE3630B12, 0x94643B84, 0x0D6D6A3E, 0x7A6A5AA8, 0xE40ECF0B, 0x9309FF9D, 0x0A00AE27, 0x7D079EB1,
    0xF00F9344, 0x8708A3D2, 0x1E01F268, 0x6906C2FE, 0xF762575D, 0x806567CB, 0x196C3671, 0x6E6B06E7,
    0xFED41B76, 0x89D32BE0, 0x10DA7A5A, 0x67DD4ACC, 0xF9B9DF6F, 0x8EBEEFF9, 0x17B7BE43, 0x60B08ED5,
    0xD6D6A3E8, 0xA1D1937E, 0x38D8C2C4, 0x4FDFF252, 0xD1BB67F1, 0xA6BC5767, 0x3FB506DD, 0x48B2364B,
    0xD80D2BDA, 0xAF0A1B4C, 0x36034AF6, 0x41047A60, 0xDF60EFC3, 0xA867DF55, 0x316E8EEF, 0x4669BE79,
    0xCB61B38C, 0xBC66831A, 0x256FD2A0, 0x5268E236, 0xCC0C7795, 0xBB0B4703, 0x220216B9, 0x5505262F,
    0xC5BA3BBE, 0xB2BD0B28, 0x2BB45A92, 0x5CB36A04, 0xC2D7FFA7, 0xB5D0CF31, 0x2CD99E8B, 0x5BDEAE1D,
    0x9B64C2B0, 0xEC63F226, 0x756AA39C, 0x026D930A, 0x9C0906A9, 0xEB0E363F, 0x72076785, 0x05005713,
    0x95BF4A82, 0xE2B87A14, 0x7BB12BAE, 0x0CB61B38, 0x92D28E9B, 0xE5D5BE0D, 0x7CDCEFB7, 0x0BDBDF21,
    0x86D3D2D4, 0xF1D4E242, 0x68DDB3F8, 0x1FDA836E, 0x81BE16CD, 0xF6B9265B, 0x6FB077E1, 0x18B74777,
    0x88085AE6, 0xFF0F6A70, 0x66063BCA, 0x11010B5C, 0x8F659EFF, 0xF862AE69, 0x616BFFD3, 0x166CCF45,
    0xA00AE278, 0xD70DD2EE, 0x4E048354, 0x3903B3C2, 0xA7672661, 0xD06016F7, 0x4969474D, 0x3E6E77DB,
    0xAED16A4A, 0xD9D65ADC, 0x40DF0B66, 0x37D83BF0, 0xA9BCAE53, 0xDEBB9EC5, 0x47B2CF7F, 0x30B5FFE9,
    0xBDBDF21C, 0xCABAC28A, 0x53B39330, 0x24B4A3A6, 0xBAD03605, 0xCDD70693, 0x54DE5729, 0x23D967BF,
    0xB3667A2E, 0xC4614AB8, 0x5D681B02, 0x2A6F2B94, 0xB40BBE37, 0xC30C8EA1, 0x5A05DF1B, 0x2D02EF8D
};

static uint32_t crc32_software(const uint8_t *data, size_t len)
{
    uint32_t crc = 0xFFFFFFFFU;
    for (size_t i = 0; i < len; ++i) {
        crc = crc32_table[(crc ^ data[i]) & 0xFF] ^ (crc >> 8);
    }
    return crc ^ 0xFFFFFFFFU;
}

static uint32_t crc32_hardware(const uint8_t *buf, size_t len)
{
    uint32_t crc = 0xFFFFFFFFU;

    while (len >= 8) {
        uint64_t v = *(const uint64_t *)buf;
        crc = __crc32d(crc, v);
        buf += 8;
        len -= 8;
    }

    while (len--) {
        crc = __crc32b(crc, *buf++);
    }

    return crc ^ 0xFFFFFFFFU;
}

static uint64_t get_ns(void)
{
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return ts.tv_sec * 1000000000ULL + ts.tv_nsec;
}

int main(int argc, char **argv)
{
    if (argc != 2) {
        fprintf(stderr, "Usage: %s <file>\n", argv[0]);
        return 1;
    }

    const char *path = argv[1];

    int fd = open(path, O_RDONLY);
    if (fd == -1) { perror("open"); return 1; }

    struct stat st;
    if (fstat(fd, &st) == -1) { perror("fstat"); close(fd); return 1; }
    off_t filesize = st.st_size;
    if (filesize == 0) {
        printf("File is empty\n");
        close(fd);
        return 0;
    }

    void *map = mmap(NULL, filesize, PROT_READ, MAP_PRIVATE, fd, 0);
    if (map == MAP_FAILED) { perror("mmap"); close(fd); return 1; }
    close(fd);

    const uint8_t *data = (const uint8_t *)map;

    crc32_software(data, filesize);
    crc32_hardware(data, filesize);

    uint64_t best_sw = -1ULL, best_hw = -1ULL;
    uint32_t crc_sw = 0, crc_hw = 0;

    for (int i = 0; i < ITERATIONS; ++i) {
        uint64_t t0 = get_ns();
        crc_sw = crc32_software(data, filesize);
        uint64_t t1 = get_ns();
        uint64_t dt = t1 - t0;
        if (dt < best_sw) best_sw = dt;
    }

    for (int i = 0; i < ITERATIONS; ++i) {
        uint64_t t0 = get_ns();
        crc_hw = crc32_hardware(data, filesize);
        uint64_t t1 = get_ns();
        uint64_t dt = t1 - t0;
        if (dt < best_hw) best_hw = dt;
    }

    double sec_sw = best_sw * 1e-9;
    double sec_hw = best_hw * 1e-9;
    double gb_sw  = filesize / (sec_sw * 1073741824.0);
    double gb_hw  = filesize / (sec_hw * 1073741824.0);

    printf("Software  : %.6f s  →  %.2f GB/s   CRC32 = 0x%08x\n", sec_sw, gb_sw, crc_sw);
    printf("Hardware  : %.6f s  →  %.2f GB/s   CRC32 = 0x%08x\n", sec_hw, gb_hw, crc_hw);
    printf("Speedup   : %.1fx\n", sec_sw / sec_hw);
    printf("Verify    : %s\n", (crc_sw == crc_hw) ? "PASS" : "FAIL");

    munmap(map, filesize);
    return 0;
}

```










硬件CRC开发

在操作系统中，如果一个新的进程就绪的时候的优先级比正在运行的进程的优先级更高，那么就会发生抢占，通过抢占，打断正在执行的进程，让高优先级的进程得到机会运行。

# 抢占图示
对于早期的linux内核，默认是不开启抢占模式，当一个进程高优先级进程需要运行的时候，必须等到当前正在运行的较低优先级的进程运行完毕后才能运行，在这里，等待低优先级进程的时间也会算在高优先级进程的响应时间中去。这对一些具有实时性要求的业务来说是不可接受的。（排队的时间为什么要算成自己运行的时间）

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

现代的linux版本内核默认都是开启抢占的，也就意味着，一个高优先级进程进来的时候，系统会在预设的抢占点上，尽可能的让高优先级任务得到运行，这样就最小化了高优先级任务的排队时间，从而提高了高优先级进程的响应能力

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

从整体的运行视角上来看，假设正在运行的任务是A，高优先级任务是B，那么状态就是B抢占了A，如下

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

# 何时抢占
操作系统的操作并不是代表着高优先级任务B就绪就立马发生的抢占行为，而是正在运行的任务A主动提供了被抢占时机。通俗的说法是任务A让出(yield)了cpu。在linux中，实际上是置为了`TIF_NEED_RESCHED`标志
## tick
操作系统在调度器分配给自己的时间片运行完毕之后，主动由定时器检查并调用`scheduler_tick`函数，然后调用`task_tick`，`entity_tick`。`entity_tick`会调用`check_preempt_tick`来触发抢占，简单代码流程如下
```
update_process_times
  scheduler_tick
    task_tick
      task_tick_fair
        entity_tick
          update_curr
            curr->vruntime += calc_delta_fair(delta_exec, curr);
          check_preempt_tick
            resched_curr
              set_tsk_need_resched
                set_tsk_thread_flag(tsk,TIF_NEED_RESCHED);
```
我们关注`check_preempt_tick`函数
```
static void
check_preempt_tick(struct cfs_rq *cfs_rq, struct sched_entity *curr)
{
        ideal_runtime = sched_slice(cfs_rq, curr);
        delta_exec = curr->sum_exec_runtime - curr->prev_sum_exec_runtime;
        if (delta_exec > ideal_runtime) {
                resched_curr(rq_of(cfs_rq));
                return;
        }
        delta = curr->vruntime - se->vruntime;
        if (delta > ideal_runtime)
                resched_curr(rq_of(cfs_rq));
}
```
检查vruntime是不是超过granularity太久，发生抢占
## wake up
新唤醒的进程优先级高于正在运行的进程时，通过`wake_up_new_task`/`wake_up_process`/`wake_up_state`三类接口来唤醒正在运行的任务让它触发抢占。代码流程如下
```
wake_up_new_task--->check_preempt_curr--->resched_curr
wake_up_process
wake_up_state
  try_to_wake_up
    ttwu_queue
      ttwu_do_activate
        ttwu_do_wakeup
          check_preempt_curr
            resched_curr
```
我们关注`check_preempt_curr`函数
```
static void check_preempt_wakeup(struct rq *rq, struct task_struct *p, int wake_flags)
{
        if (unlikely(task_has_idle_policy(curr)) &&
            likely(!task_has_idle_policy(p)))
                goto preempt;
preempt:
        resched_curr(rq);
}
```
如果处于idle则发生抢占
## 其他场景
上面已经是大部分场景了，还有负载均衡，进程优先级修改，中断，系统调用等时机都是调度的时机，这里就不一一举例了。代码很容易查到
# preempt_count 
当抢占发生的时候，`preempt_count`会自加，同时解除抢占的时候`preempt_count`自减。`preempt_count`维持平衡，同时也是调试抢占的关键变量
```
#define preempt_count_add(val)  __preempt_count_add(val)
#define preempt_count_sub(val)  __preempt_count_sub(val)
#define preempt_count_dec_and_test() __preempt_count_dec_and_test()
#endif

#define __preempt_count_inc() __preempt_count_add(1)
#define __preempt_count_dec() __preempt_count_sub(1)

#define preempt_count_inc() preempt_count_add(1)
#define preempt_count_dec() preempt_count_sub(1)

#ifdef CONFIG_PREEMPT_COUNT

#define preempt_disable() \
do { \
        preempt_count_inc(); \
        barrier(); \
} while (0)

#define sched_preempt_enable_no_resched() \
do { \
        barrier(); \
        preempt_count_dec(); \
} while (0)

#define preempt_enable_no_resched() sched_preempt_enable_no_resched()

#define preemptible()   (preempt_count() == 0 && !irqs_disabled())

#ifdef CONFIG_PREEMPTION
#define preempt_enable() \
do { \
        barrier(); \
        if (unlikely(preempt_count_dec_and_test())) \
                __preempt_schedule(); \
} while (0)

#define preempt_enable_notrace() \
do { \
        barrier(); \
        if (unlikely(__preempt_count_dec_and_test())) \
                __preempt_schedule_notrace(); \
} while (0)

#define preempt_check_resched() \
do { \
        if (should_resched(0)) \
                __preempt_schedule(); \
} while (0)
```
在这里`preempt_disable`和`preempt_enable`经常和spinlock搭配，这样可以使得每次spinlock结束的时候，也是一次调度时机
# 执行抢占
执行抢占就是调用schedule函数，下面列出来几个时机
## syscall
系统调用完成之后，主动调用schedule，代码流程如下
```
syscall_exit_to_user_mode
  exit_to_user_mode_prepare
    exit_to_user_mode_loop
      schedule()
```
## 中断结束
中断结束也会主动调用schedule(
```
irqentry_exit
  irqentry_exit_cond_resched
    preempt_schedule_irq
      __schedule(True)
```
## 等待队列
```
wait_event
  __wait_event
    ___wait_event
       schedule()
```
## 其他
其他常见都是少见常见了，可以通过代码找出来，这里就不一一介绍了。


Linux抢占简介

RCU是一种基于无锁的设计， 对于多个读者来说，几乎可以等效为无锁情况，特别适合于像文件系统目录查询，DNS resolution这样的场景。   
使用RCU的场景必须要满足如下两点
* 频繁的读取，很少的写入
* 对资源的数据没有强一致性要求

# 原理介绍
RCU管理的成员通常是一个内核的list或link-list。主要原理如下
## 删除
当对RCU的一个list的成员进行删除，这个删除不是立即删除，而是先标记为释放，等所有的读端执行所有读请求之后，再统一进行实际删除操作。 在这里，RCU将一个删除操作开始，到删除操作真的完成中间耗费的实际叫做宽限期(grace period)

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

## 插入
当对RCU的一个list的成员进行插入的时候，RCU需要保证任何时候对这个list的读取是完整的，非中间态的。也就是其他的读端，要么是未插入的状态，要么是已插入的状态，绝不能是中间状态

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

## 完整性
任何时候对list的操作，无论是删除还是插入，RCU都需要保证任何时候的读端对list的遍历都是正常的，不会出现断链问题。不过同插入一样，rcu不保证一定能够读到新增的元素，或不读到要被删除的元素。

# 宽限期

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

顾名思义，只要任务在宽限期做的任何事情，都相当于有免死金牌，中间发生的错误可以忽略。对于RCU而言，具体表达的意思就是在宽限期内的list的成员，读到旧值也是没有问题的。

换个角度，如果等待宽限期过了之后，一切需要按照原本预期发生。   
所以在linux中，`synchronize_rcu()`执行的这段时间就是宽限期。

所以使用RCU的范式如下：

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

对于读端，引用一个指针,触发rcu read
```
   rcu_read_lock();
   p = rcu_dereference(gptr);
   if (p) {
      printk(KERN_INFO "Reader: val = %d\n", p->val);
   }
   rcu_read_unlock();   
```
对于写端
```
   old = rcu_dereference_protected(gptr, lockdep_is_held(&some_lock)); 
   rcu_assign_pointer(gptr, new);
   if (old) {
        synchronize_rcu();
        kfree(old);
    }
```
关于rcu使用的实际图示如下

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

当写线程要将指针 p 从指向 data1 改为指向 data2 时，需考虑读线程可能仍在使用 data1。因此，写线程会先“发布”新数据（即更新指针），然后进入等待期，直到所有读线程都退出当前的 RCU 临界区。

在此期间，不同读线程可能看到不同版本的数据：有的仍看到旧的 data1（记为 p1），有的已看到新的 data2（记为 p2）。只有当所有读线程都完成本次临界区访问后，才能确保它们全部看到 data2，此时 data1 才可安全释放。

# 删除操作
在linux中，通过`list_del_rcu`删除了一个元素，那么它整个周期的步骤如下   
1. 初始状态

假设想要删除节点5,6,7的位置

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

2. 伪删除操作

为了满足RCU宽限期的要求，所有的读端可能读到两个版本，一个是删除前的版本，一个是删除后的版本，代码发生在list_del_rcu后 synchronize_rcu 前，如下

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

3. 宽限期后

到宽限期后，读端只能看到一个版本，这样代码可以安全的删除节点5,6,7的位置，此时代码在 synchronize_rcu 后


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

# 插入操作
在linux中，通过`list_replace_rcu`替换了一个元素，那么它整个周期的步骤如下   
1. 初始状态

假设想要替换节点5,6,7的位置

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

2. 伪更新

此时q元素携带值5，2，3想要替换p元素的5，6，7。伪更新操作会让代码在宽限期内看到两个版本，一个是123，567,1148。另一个是123,525,1148。对应代码为 list_replace_rcu 后， synchronize_rcu 前

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

3. 宽限期后

等待宽限期完成之后，整个视角都变成一个唯一的版本，元素5，6，7可以被安全的删除了。此时代码在 执行synchronize_rcu 之后。 

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

# 总结
本文简单介绍了rcu，在用户层面上为了支持高性能的开发，可以使用用户态的rcu库，例如liburcu。

rcu在内核中设计的知识比较庞大，这里只是简单将基础知识介绍一下。详细的可以参考 《深入理解RCU》 的系列文章

# 参考
> https://gitee.com/xiebaoyou/documents/blob/master/%E6%B7%B1%E5%85%A5%E7%90%86%E8%A7%A3RCU.zip






