编写高性能的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








RCU同步简介

在linux系统中，默认可以通过创建进程和线程运行程序，而在内核中，被调度的实体单位是进程(task)，而线程是只是创建的时候没有独立的分配地址空间。但无论如何，进程和线程是被调度的最小单元。  
假设我们要设计一个极高性能的软件程序，能不能通过一些软件的技巧，尽可能少的创建线程或者进程，从而可以减少系统调度器的调度负载，同时如果一个任务不经过系统的上下文切换，自然其整体性能是高于线程和进程的。   
在C语言中，可以通过自己实现协程来做到这一点，协程在操作系统层面上是看不到的，它只能看到进程和线程，但是在代码层面，他是“多任务并发”的。本文简单介绍一下协程
# 无栈协程
开源软件putty就使用了无栈协程，它是基于goto的特性实现，switch case就是一种goto的体现。详细可以如下了解
> https://www.chiark.greenend.org.uk/~sgtatham/coroutines.html

以一个例子说明
```
#define crBegin static int state=0; switch(state) { case 0:
#define crReturn(i,x) do { state=i; return x; case i:; } while (0)
#define crFinish }
int main(void) {
    static int i;
    crBegin;
    for (i = 0; i < 10; i++)
        crReturn(1, i);
    crFinish;
}
```
此函数通过宏展开其实就是简单的0，9的循环体，如下
```
# gcc test.c -E -o test.E
int main(void) {
    static int i;
    static int state=0; switch(state) { case 0:;
    for (i = 0; i < 10; i++)
        do { state=1; return i; case 1:; } while (0);
    };
}
```
接下来事情就简单了，如果实现一个yield函数，此函数在switch里面如果能够动态的修改case的值，那么它就好像是多线程一样运行，yield的实现如下
```
#define cr_yield                 \
    {                            \
        __s = __LINE__;          \
        usleep(THREAD_INTERVAL); \
        return;                  \
    case __LINE__:;              \
    }
```
那么这句yield的语义就是将当前任务让出，由其他协程来继续运行。同操作系统调度的yield语义一致。   
下面提供一个完整的无栈协程的例子
```
#include <stdio.h>
#include <unistd.h>

#define THREAD_INTERVAL 1000 * 500
#define cr_start()      \
    static int __s = 0; \
    switch (__s) {      \
    case 0:

#define cr_yield                 \
    {                            \
        __s = __LINE__;          \
        usleep(THREAD_INTERVAL); \
        return;                  \
    case __LINE__:;              \
    }
#define cr_end() \
    }            \
    __s = 0;

static int condition = 1;

static void coroutine_1() {
    cr_start();

    for (;;) {
        /* do something */
        printf("Run %s\n", __FUNCTION__);
        cr_yield;
    }

    cr_end();
}

static void coroutine_2() {
    cr_start();

    for (;;) {
        if (condition) {
            /* do something conditional */
            printf("Run %s - Action(1)\n", __FUNCTION__);
            cr_yield;
        }

        /* do something */
        printf("Run %s - Action(2)\n", __FUNCTION__);
        condition = !condition;
        cr_yield;
    }

    cr_end();
}

int main() {
    for (;;) {
        coroutine_1();
        coroutine_2();
    }
    return 0;
}
```
编译并运行，日志如下
```
Run coroutine_1
Run coroutine_2 - Action(1)
Run coroutine_1
Run coroutine_2 - Action(2)
Run coroutine_1
Run coroutine_2 - Action(2)
Run coroutine_1
Run coroutine_2 - Action(1)
Run coroutine_1
Run coroutine_2 - Action(2)
Run coroutine_1
Run coroutine_2 - Action(2)
```
可以看到，通过switch和case以及__LINE的动态更新，让代码好像是两个线程在运行一样。但实际上，此段代码是在一个线程内顺序执行的。  
论文 Protothreads 实现了更成熟的无栈协程。下面基于通过例子演示一下基于Protothreads的co-routine
首先需要Protothreads实现的头文件，如下
提供lc-switch.h文件
```
#ifndef LC_SWITCH_H_
#define LC_SWITCH_H_

/** \hideinitializer */
typedef unsigned short lc_t;

#define LC_INIT(s) s = 0;

#define LC_RESUME(s) switch(s) { case 0:

#define LC_SET(s) s = __LINE__; case __LINE__:

#define LC_END(s) }

#endif /* LC_SWITCH_H_ */

/** @} */
```
提供lc.h文件
```
#ifndef LC_H_
#define LC_H_

#ifdef LC_CONF_INCLUDE
#include LC_CONF_INCLUDE
#else /* LC_CONF_INCLUDE */
#include "lc-switch.h"
#endif /* LC_CONF_INCLUDE */

#endif /* LC_H_ */

/** @} */
/** @} */
```
提供pt.h文件
```
#ifndef PT_H_
#define PT_H_

#include "lc.h"

struct pt {
  lc_t lc;
};

#define PT_WAITING 0
#define PT_YIELDED 1
#define PT_EXITED  2
#define PT_ENDED   3

#define PT_INIT(pt)   LC_INIT((pt)->lc)

#define PT_THREAD(name_args) char name_args

#define PT_BEGIN(pt) { char PT_YIELD_FLAG = 1; if (PT_YIELD_FLAG) {;} LC_RESUME((pt)->lc)

#define PT_END(pt) LC_END((pt)->lc); PT_YIELD_FLAG = 0; \
                   PT_INIT(pt); return PT_ENDED; }

#define PT_WAIT_UNTIL(pt, condition)            \
  do {                                          \
    LC_SET((pt)->lc);                           \
    if(!(condition)) {                          \
      return PT_WAITING;                        \
    }                                           \
  } while(0)

#define PT_WAIT_WHILE(pt, cond)  PT_WAIT_UNTIL((pt), !(cond))

#define PT_WAIT_THREAD(pt, thread) PT_WAIT_WHILE((pt), PT_SCHEDULE(thread))

#define PT_SPAWN(pt, child, thread)             \
  do {                                          \
    PT_INIT((child));                           \
    PT_WAIT_THREAD((pt), (thread));             \
  } while(0)

#define PT_RESTART(pt)                          \
  do {                                          \
    PT_INIT(pt);                                \
    return PT_WAITING;                  \
  } while(0)

#define PT_EXIT(pt)                             \
  do {                                          \
    PT_INIT(pt);                                \
    return PT_EXITED;                   \
  } while(0)

/** @} */

#define PT_SCHEDULE(f) ((f) < PT_EXITED)

/** @} */

#define PT_YIELD(pt)                            \
  do {                                          \
    PT_YIELD_FLAG = 0;                          \
    LC_SET((pt)->lc);                           \
    if(PT_YIELD_FLAG == 0) {                    \
      return PT_YIELDED;                        \
    }                                           \
  } while(0)

#define PT_YIELD_UNTIL(pt, cond)                \
  do {                                          \
    PT_YIELD_FLAG = 0;                          \
    LC_SET((pt)->lc);                           \
    if((PT_YIELD_FLAG == 0) || !(cond)) {       \
      return PT_YIELDED;                        \
    }                                           \
  } while(0)

/** @} */

#endif /* PT_H_ */

/**
 * @}
 * @}
 */
```
上面是全部Protothreads的实现了，下面编写测试程序
```
#include <stdio.h>
#include "pt.h"

static struct pt co_pt;

static int coroutine(struct pt *pt) {
    PT_BEGIN(pt);
    printf("[%s]: Start\n", __func__);
    PT_YIELD(pt);
    printf("[%s]: After yield\n", __func__);
    PT_END(pt);
}

int main() {
    PT_INIT(&co_pt);
    coroutine(&co_pt);
    coroutine(&co_pt);
    return 0;
}
```
编译运行日志如下
```
tf@tf:# ./test
[coroutine]: Start
[coroutine]: After yield
```
到这里，无栈协程完全说清楚了。
# 有栈协程
无栈协程的特点是简单，但是正因为没有栈的实现，所以局部变量无法保持。同时，有栈协程的体验会更像线程，对于无栈协程，错误的开发可能导致内存泄漏无法察觉。而开辟栈区的协程，可以清晰的管理这些事情。  

开源的有栈协程库为 neco 。下面基于此简单说明一下
```
git clone https://github.com/tidwall/neco.git
```
根据仓库的README.md的指示，测试程序如下
```
#include <stdio.h>
#include "neco.h"

void ticker(int argc, void *argv[]) {
    while (1) {
        neco_sleep(NECO_SECOND);
        printf("tick\n");
    }
}

void tocker(int argc, void *argv[]) {
    while (1) {
        neco_sleep(NECO_SECOND*2);
        printf("tock\n");
    }
}

int neco_main(int argc, char *argv[]) {
    neco_start(ticker, 0);
    neco_start(tocker, 0);

    // Keep the program alive for an hour.
    neco_sleep(NECO_HOUR);
    return 0;
}
```
编译运行后，日志如下
```
tf@tf:# ./test
tick
tock
tick
tick
tock
tick
tick
tock
tick
```
# 总结
本文简单介绍了C语言的协程，协程对于高性能应用开发来说必不可少。

