在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语言的协程，协程对于高性能应用开发来说必不可少。



C语言协程

早期的操作系统只需要关心CPU的调度策略，从而提高通用程序的运行效率，但是随着NPU和其他ASIC的普及，每个SOC下都支持了多个NPU和ASIC芯片。面对这样芯片的设计趋势，操作系统调度变得至关重要。调度不再只关心CPU内部，而是需要对所有资源进行有效调度。本文先简单讨论XPU调度的几个关注点
# XPU调度的必要性
在没有XPU调度的环境下，其资源需要长期待命，一方面功耗无法降低，另一方面性能无法完全利用。
## 功耗问题
假设CPU下发一个需要CPU，DSP，GPU，NPU同时运行的任务，当CPU开始运行任务的时候，需要同时将DSP，GPU，NPU都上去启动到就绪状态，等待CPU将任务下发给不同的执行单元。这种情况下，假设SOC环境具备两个DSP单元，那么两个DSP单元都会同时启动，那么整个SOC环境下的功耗一直是以最大运行状态运行。    
那么是不是可以让CPU在使用DSP的时候，尽可能调度在同一个DSP上，这样可以让另一个空闲的DSP得到关闭，降低功耗

## 性能问题
假设CPU下发多个需要CPU，DSP，GPU，NPU同时运行的任务，当CPU开始运行多个任务的时候，因为在DSP/GPU/NPU的视角是谁来就运行谁的任务处于一种无序状态，所以无法有效的将DSP，GPU，NPU的利用率提升到100%。

# XPU调度测量指标
为了有效的开发和设计XPU调度策略，需要统一的从如下几个维度进行分析和决策，如下
## 利用率
XPU完成任务的工作量。利用率根据管理的计算任务的数量和性质而变化。通过利用率可以看到设备的运行时间，空闲时间。 而调度策略的目的是降低设备的空闲时间。如果可降低的空闲时间为100%，那么将节省一个设备的核心
## 吞吐量
它是指每单位时间内完成的进程数量。通过有效地优先级排序和调度进程来提高吞吐量，从而在给定的时间范围内完成更多的任务。   
假设XPU的利用率是100%，但是无序争用和有序的编排不同的是，有序的编排可以让某一段时间内的XPU吞吐量很高（因为运行短任务），同时让某一段时间内的吞吐量降低（此时可以通过降低设备频率获取功耗收益）
## 周转时间
这指从进程开始到完成所经过的时间。周转时间是总等待时间，它包括进程在CPU，DSP,NPU,GPU，以及内存，IO，和调度算法内等待的总时间。   
一方面，调度策略内应该保证所有任务在就绪队列中等待的时间都得到减少   
另一方面，良好的调度策略可以降低单进程维度下的任务等待时间，从而满足具备实时特征的任务能够长期的保证其响应效果
## 等待时间
在周转时间内已经提到，一个进程在就绪队列中花费的总时间称为等待时间。因此，调度算法决定了进程在获得XPU设备资源前的所有停留时间的长短
## 响应时间
它是提交请求和收到第一个响应之间的时间。调度策略确保高优先级进程能够及时获得XPU设备资源，有助于减少响应时间并提升整体用户体验。

# 常规调度策略
如果仅从调度本身来看，其实XPU的调度策略和通用CPU的调度策略是相通的。因为都是按照一定的指标选出最合适的任务。只不过CPU的调度面向的资源设备是CPU本身，而XPU的调度面向的资源设备是加入调度管理的多种多样的设备。   
目前市场上XPU的调度策略都比较简单，本文先从简单调度策略说起

## FCFS
FCFS（先来先服务）算法根据进程就绪顺序，即基于任务的到达时间

## SJF
SJF（最短作业优先）算法根据进程的预算时间分配资源，也就是说，运行时间短的进程会先获得访问权

## PRIO
SP（静态优先级）算法根据特定的数值表示更高的或更低的优先级，高优先级的进程会比低优先级的进程更早获得设备访问权   
注意，一些系统上的设计，可能Priority越高，优先级越高，但是另一些系统上，可能Priority越低，优先级越高。为什么会有后者呢？主要还是因为红黑树的原因，红黑树找最小值的效率最高。所以设计系统时出现了常人认为的反直觉，也就是Priority越低，优先级越高。对红黑树兴趣的可以查看我其他文章《rb-tree实现》系列

## RR
RR（轮转调度）算法根据时间片周期的运行任务，这使得所有进入就绪的任务都能周期性的得到运行，体现了调度的公平性

## 弊端
* FCFS 会遭受“车队效应”的困扰，队列前面的长进程会延迟短进程。
* SJF 的缺点是无法高效处理动态负载。优先级调度可能导致饥饿，如果高优先级进程持续到达，会阻止低优先级进程执行。
* 轮转调度（RR）具有高上下文切换开销，可能对长进程导致性能不佳。
* 基于优先级的算法可能会忽视低优先级进程，造成不公平。

# 下一代XPU调度策略
根据上面可以看到，常规调度目前已经应用在各种XPU的中间件上，首先，他们是单独维护和服务于某一类型的设备，例如GPU，那么对应就是GPU驱动，例如NPU，那么就对于cuda库（以nvidia为例）。   
这里再提供一些新的调度策略可用于XPU上。
## 自适应调度策略
常规的自适应调度策略例如动态优先级，动态时间片，基于负载均衡的调度都属于自适应调度策略。
这种调度策略会动态的修改任务的整体优先级排序情况，从而让任务更具备公平性，不被饿死。

在NPU领域，Shockwave也提出了基于市场调控的调度策略。也属于自适应调度策略的一种。

## 实时调度策略
实时调度算法在通常应用在具备强实时环境中，例如硬实时操作系统中调度策略，或者确保具有严格时间约束的中间件，例如常规的速率单调调度（RMS）、最早截止日期优先（EDF）和截止日期单调调度（DMS）。

在XPU的调度中，我们更关心的是任务的实时性，可以牺牲不具备实时要求的任务的响应效果。

# 抢占
对于XPU环境下的调度策略，抢占是维系任务实时系的重要功能，抢占分两种
* 调度器抢占
* 用户抢占

调度器抢占很好理解，例如实时调度算法EDF根据deadline来实施对任务的抢占，RR调度在时间片完成后，默认的轮转可以任务是抢占点

而用户抢占指的是提供了用户的接口，让用户执行任务前决定是否可以抢占设备。通过用户主动实施的任务抢占，可以让设备被抢占后保存上下文，从而运行下一个任务。

不过抢占都是具备开销的，过多的抢占会给系统带来很强的调度器负载，例如低时间片下的RR调度算法。过少的抢占又会影响XPU调度的实时效果。这里面要从实际情况中获取经验值。反哺到XPU调度算法的设计中去。

# 总结
随着AI的逐步普及，目前的SOC设计都会带有NPU核心，同时带有多个DSP单元和其他的ASIC设计。这些设计不再像以前一样，只要用起来就行。我们需要关注整体实时性响应，性能效果，和功耗输出。所以将SOC中以这种XPU维度的调度策略越来越重要。 









异构(XPU)计算中的任务调度

使用vscode来进行服务器ssh连接在win10上会出现权限问题，需要几条命令设置一下。
# 报错
```
[13:14:15.845] > Bad permissions. Try removing permissions for user: \\Everyone (S-1-1-0) on file C:/Users/Administrator/.ssh/config.
> Bad owner or permissions on C:\\Users\\Administrator/.ssh/config
> 过程试图写入的管道不存在。
> ]0;C:\WINDOWS\System32\cmd.exe
[13:14:15.845] Got some output, clearing connection timeout
[13:14:17.122] "install" terminal command done
[13:14:17.122] Install terminal quit with output: ]0;C:\WINDOWS\System32\cmd.exe
[13:14:17.122] Received install output: ]0;C:\WINDOWS\System32\cmd.exe
[13:14:17.123] Failed to parse remote port from server output
[13:14:17.124] Resolver error: Error: 
	at g.Create (c:\Users\Administrator\.vscode-insiders\extensions\ms-vscode-remote.remote-ssh-0.111.2024040515\out\extension.js:2:499181)
	at t.handleInstallOutput (c:\Users\Administrator\.vscode-insiders\extensions\ms-vscode-remote.remote-ssh-0.111.2024040515\out\extension.js:2:496503)
	at t.tryInstall (c:\Users\Administrator\.vscode-insiders\extensions\ms-vscode-remote.remote-ssh-0.111.2024040515\out\extension.js:2:620043)
	at async c:\Users\Administrator\.vscode-insiders\extensions\ms-vscode-remote.remote-ssh-0.111.2024040515\out\extension.js:2:579901
	at async t.withShowDetailsEvent (c:\Users\Administrator\.vscode-insiders\extensions\ms-vscode-remote.remote-ssh-0.111.2024040515\out\extension.js:2:583207)
	at async k (c:\Users\Administrator\.vscode-insiders\extensions\ms-vscode-remote.remote-ssh-0.111.2024040515\out\extension.js:2:576866)
	at async t.resolve (c:\Users\Administrator\.vscode-insiders\extensions\ms-vscode-remote.remote-ssh-0.111.2024040515\out\extension.js:2:580578)
	at async c:\Users\Administrator\.vscode-insiders\extensions\ms-vscode-remote.remote-ssh-0.111.2024040515\out\extension.js:2:846696
[13:14:17.126] ------
```
# 解决方案
直接命令解决即可
```
icacls "C:\Users\Administrator\.ssh\config" /remove "Everyone"
icacls "C:\Users\Administrator\.ssh\config" /inheritance:r
icacls "C:\Users\Administrator\.ssh\config" /grant "Everyone:R"
```
为windows下的config文件移除everyone用户的权限，并禁用权限继承后，并设置everyone的读权限


vscode remote-ssh的权限异常

zram的算法种类
LZ77 LZO LZ4 LZ4HC LZO-RLE 842 deflate zstd

压缩的任务是找出重复的字符。
关于LZ算法
分区--找匹配---编码---滑动窗口

lz4 匹配优化hash  编码优化，顺序访问，字节对齐    压缩率有牺牲，压缩速度快
lz4hc 匹配更深入   压缩速度慢
lzo  hash，变长编码   压缩率好
lzo-rle  连续重复字节优化  
deflate  常用于gzip
zstd  天生多线程  可加速

lz4m：基于lz4的内存数据压缩优化， 压缩速度提高，但是压缩率减少

lz4 benchmark: zram-perf

使用zram-perf测试，不是用户数据，没办法模拟用户场景

构造zram-dump来进行效率评估

使用zram-dump下的测试数据：
lz4m 速度快，压缩率劣化

压缩算法优劣：
压缩率：可用内存  压缩速度：slowpath/kswapd 解压速度: page fault 不卡

腾内存的速度，单位时间腾内存的速度，lz4m腾内存速度最快

zram_multi_comp：可以支持zram使用多种不同的压缩算法，取长补短
6.2合入
1. 配置

zram结构体从1个变成4个，1个主压缩算法，3个次压缩算法

目标：让zram重新压缩指定数据，减少zmalloc使用量，关注在压缩率上
遍历zram slot，根据echo的算法，先对slot解压，在按顺序用新算法压缩，直到新的数据大小比原数据小，则更新zram slot

也就是通过配置来实现重压缩

问题是：速度太慢，不可接受，压缩算法仅考虑了压缩率，不考虑压缩速度。不够完善

目标是：重载情况下，腾内存更快。

配置：将lz4作为主压缩算法，lz4m作为第二压缩算法
场景关联：kswapd和绝大部分slowpath，则lz4。 前台进程 lz4m，更快腾出内存

新增精准重压，增加线程，闲时主动压缩


展望：zram使用异构方向优化。


是不是按照不同场景来进行压缩


zram异构压缩技术 基于GPU加速的内存回收
lz4执行时间太长，占整个回收过程的30-40%，这是导致重载情况下CPU争抢和体验劣化的关键瓶颈。
社区patch zsmalloc/zram: there be preemption




cuda已经有lz4的并发实现































zram多压缩算法的效率

在某车载项目上，因为其中多域使用了can总线通信，系统使用麒麟，在大负载的情况下，低概率会触发死机问题。死机伴随着rcu stall信息。很明显，这是一个典型的死核的问题，本文主要总结和分享此问题。
# 复现问题
根据描述的现象，首先就是需要针对现场问题寻找复现办法，可以了解到问题是can通信导致的rcu，所以可以简单通过脚本测试，如下
```
while true
do
cansend can0 1f602001#00000000D0570068
done
```
# 日志
根据上面的测试脚本，内核跑一段时间就出来rcu stall了，其日志如下
```
[ 815.850352] rcu: INFO: rcu_sched self-detected stall on CPU
[ 815.855942] rcu: 0-....: (17844 ticks this GP) idle=d1a/1/0x4000000000000004 softirq=29552/29558 fqs=6000
[ 815.865680] (t=18000 jiffies g=51353 q=2202)
[ 815.870039] Task dump for CPU 0:
[ 815.873269] task:kworker/0:0 state:R running task stack: 0 pid: 1131 ppid: 2 flags:0x0000000a
[ 815.883192] Workqueue: events_freezable input_dev_poller_work
[ 815.888937] Call trace:
[ 815.891389] dump_backtrace+0x0/0x1a8
[ 815.895052] show_stack+0x2c/0x38
[ 815.898370] sched_show_task+0x12c/0x148
[ 815.902299] dump_cpu_task+0x48/0x54
[ 815.905879] rcu_dump_cpu_stacks+0xec/0xf8
[ 815.909977] rcu_sched_clock_irq+0x274/0x6c8
[ 815.914254] update_process_times+0x68/0x90
[ 815.918437] tick_sched_handle.isra.16+0x50/0x64
[ 815.923050] tick_sched_timer+0x50/0x90
[ 815.926893] __hrtimer_run_queues+0x170/0x23c
[ 815.931254] hrtimer_interrupt+0xb4/0x1b8
[ 815.935267] arch_timer_handler_phys+0x38/0x50
[ 815.939713] handle_percpu_devid_irq+0x6c/0x124
[ 815.944243] generic_handle_irq_desc+0x10/0x18
[ 815.948688] __handle_domain_irq+0xbc/0xc4
[ 815.952785] gic_handle_irq+0x2c0/0x30c
[ 815.956627] el1_irq+0xcc/0x180
[ 815.959775] slab_free_freelist_hook+0x0/0xf0
[ 815.964136] skb_free_head+0x38/0x3c
[ 815.967716] skb_release_data+0x64/0xb4
[ 815.971558] skb_release_all+0x30/0x3c
[ 815.975306] __kfree_skb+0x20/0x34
[ 815.978705] consume_skb+0x38/0x44
[ 815.982117] can_receive+0x90/0xf8
[ 815.985526] can_rcv+0xac/0xbc
[ 815.988589] __netif_receive_skb_one_core+0x6c/0x94
[ 815.993466] __netif_receive_skb+0x78/0x80
[ 815.997562] process_backlog+0xac/0x15c
[ 816.001405] net_rx_action+0x148/0x328
[ 816.005152] __do_softirq+0x228/0x2d4
[ 816.008816] __irq_exit_rcu+0x48/0x94
[ 816.012479] irq_exit+0x14/0x20
[ 816.015626] __handle_domain_irq+0xc0/0xc4
[ 816.019721] gic_handle_irq+0x2c0/0x30c
[ 816.023563] el1_irq+0xcc/0x180
[ 816.026710] __schedule+0x40c/0x5d0
[ 816.030204] preempt_schedule_common+0x10/0x24
[ 816.034648] _cond_resched+0x20/0x38
[ 816.038227] mutex_lock+0x1c/0x44
[ 816.041544] iio_read_channel_processed+0x2c/0xd0
[ 816.046253] adc_keys_poll+0x40/0x130
[ 816.049917] input_dev_poller_work+0x28/0x3c
[ 816.054194] process_one_work+0x1d4/0x28c
[ 816.058205] worker_thread+0x1d0/0x260
[ 816.061954] kthread+0xf0/0x100
[ 816.065101] ret_from_fork+0x10/0x18
[ 900.973670] rcu: INFO: rcu_sched self-detected stall on CPU
[ 900.979258] rcu: 0-....: (17996 ticks this GP) idle=d1a/1/0x4000000000000004 softirq=29563/29765 fqs=6000
[ 900.988996] (t=18000 jiffies g=51361 q=715)
[ 900.993271] Task dump for CPU 0:
[ 900.996500] task:kworker/0:0H state:R running task stack: 0 pid: 6 ppid: 2 flags:0x0000000a
[ 901.006426] Workqueue: 0x0 (mmc_complete)
[ 901.010528] Call trace:
[ 901.012980] dump_backtrace+0x0/0x1a8
[ 901.016644] show_stack+0x2c/0x38
[ 901.019961] sched_show_task+0x12c/0x148
[ 901.023889] dump_cpu_task+0x48/0x54
[ 901.027468] rcu_dump_cpu_stacks+0xec/0xf8
[ 901.031565] rcu_sched_clock_irq+0x274/0x6c8
[ 901.035842] update_process_times+0x68/0x90
[ 901.040023] tick_sched_handle.isra.16+0x50/0x64
[ 901.044637] tick_sched_timer+0x50/0x90
[ 901.048480] __hrtimer_run_queues+0x170/0x23c
[ 901.052841] hrtimer_interrupt+0xb4/0x1b8
[ 901.056854] arch_timer_handler_phys+0x38/0x50
[ 901.061299] handle_percpu_devid_irq+0x6c/0x124
[ 901.065830] generic_handle_irq_desc+0x10/0x18
[ 901.070275] __handle_domain_irq+0xbc/0xc4
[ 901.074370] gic_handle_irq+0x2c0/0x30c
[ 901.078212] el1_irq+0xcc/0x180
[ 901.081359] trace_consume_skb+0x10/0x7c
[ 901.085286] consume_skb+0x30/0x44
[ 901.088687] can_receive+0x90/0xf8
[ 901.092086] can_rcv+0xac/0xbc
[ 901.095148] __netif_receive_skb_one_core+0x6c/0x94
[ 901.100025] __netif_receive_skb+0x78/0x80
[ 901.104122] process_backlog+0xac/0x15c
[ 901.107964] net_rx_action+0x148/0x328
[ 901.111712] __do_softirq+0x228/0x2d4
[ 901.115376] __irq_exit_rcu+0x48/0x94
[ 901.119039] irq_exit+0x14/0x20
[ 901.122186] __handle_domain_irq+0xc0/0xc4
[ 901.126281] gic_handle_irq+0x2c0/0x30c
[ 901.130123] el1_irq+0xcc/0x180
[ 901.133269] _cond_resched+0x10/0x38
[ 901.136848] mutex_lock+0x1c/0x44
[ 901.140163] set_pf_worker+0x24/0x5c
[ 901.143743] worker_thread+0x6c/0x260
[ 901.147406] kthread+0xf0/0x100
[ 901.150552] ret_from_fork+0x10/0x18
[ 1056.446969] rcu: INFO: rcu_sched self-detected stall on CPU
[ 1056.452557] rcu: 0-....: (17835 ticks this GP) idle=d1a/1/0x4000000000000004 softirq=29774/29878 fqs=6000
[ 1056.462294] (t=18003 jiffies g=51369 q=1254)
[ 1056.466655] Task dump for CPU 0:
[ 1056.469884] task:kworker/0:0H state:R running task stack: 0 pid: 6 ppid: 2 flags:0x0000000a
[ 1056.479806] Workqueue: 0x0 (mmc_complete)
[ 1056.483908] Call trace:
[ 1056.486360] dump_backtrace+0x0/0x1a8
[ 1056.490024] show_stack+0x2c/0x38
[ 1056.493342] sched_show_task+0x12c/0x148
[ 1056.497269] dump_cpu_task+0x48/0x54
[ 1056.500848] rcu_dump_cpu_stacks+0xec/0xf8
[ 1056.504945] rcu_sched_clock_irq+0x274/0x6c8
[ 1056.509222] update_process_times+0x68/0x90
[ 1056.513404] tick_sched_handle.isra.16+0x50/0x64
[ 1056.518017] tick_sched_timer+0x50/0x90
[ 1056.521860] __hrtimer_run_queues+0x170/0x23c
[ 1056.526221] hrtimer_interrupt+0xb4/0x1b8
[ 1056.530233] arch_timer_handler_phys+0x38/0x50
[ 1056.534679] handle_percpu_devid_irq+0x6c/0x124
[ 1056.539209] generic_handle_irq_desc+0x10/0x18
[ 1056.543654] __handle_domain_irq+0xbc/0xc4
[ 1056.547750] gic_handle_irq+0x2c0/0x30c
[ 1056.551592] el1_irq+0xcc/0x180
[ 1056.554740] __slab_free+0x124/0x2a0
[ 1056.558319] kfree+0x218/0x238
[ 1056.561380] skb_free_head+0x38/0x3c
[ 1056.564960] skb_release_data+0x64/0xb4
[ 1056.568803] skb_release_all+0x30/0x3c
[ 1056.572550] __kfree_skb+0x20/0x34
[ 1056.575949] consume_skb+0x38/0x44
[ 1056.579349] can_receive+0x90/0xf8
[ 1056.582759] can_rcv+0xac/0xbc
[ 1056.585821] __netif_receive_skb_one_core+0x6c/0x94
[ 1056.590698] __netif_receive_skb+0x78/0x80
[ 1056.594795] process_backlog+0xac/0x15c
[ 1056.598637] net_rx_action+0x148/0x328
[ 1056.602384] __do_softirq+0x228/0x2d4
[ 1056.606049] __irq_exit_rcu+0x48/0x94
[ 1056.609712] irq_exit+0x14/0x20
[ 1056.612858] __handle_domain_irq+0xc0/0xc4
[ 1056.616954] gic_handle_irq+0x2c0/0x30c
[ 1056.620796] el1_irq+0xcc/0x180
[ 1056.623942] __schedule+0x108/0x5d0
[ 1056.627436] preempt_schedule_common+0x10/0x24
[ 1056.631880] _cond_resched+0x20/0x38
[ 1056.635459] mutex_lock+0x1c/0x44
[ 1056.638774] set_pf_worker+0x24/0x5c
[ 1056.642353] worker_thread+0x6c/0x260
[ 1056.646016] kthread+0xf0/0x100
[ 1056.649162] ret_from_fork+0x10/0x18
```
# rcu_sched
这里借助内核的文档可以理解
    * https://www.kernel.org/doc/html/v5.10/RCU/stallwarn.html

简单理解就是： 内核在每个cpu上的任务，没有任何的机会调用schedule切出，一定时间后，抛出日志以及每个cpu上监听rcu的结构体的值。

那么所有的点都可以围绕到没有办法schedule的代码语义中，这种代码语义有如下场景：
1. 禁用中断
2. 禁用buttom half
3. 禁用抢占
4. 禁用timer
5. 中断风暴

当然，文中还提到一下corner cases，其中有一个如下
* 一些rt程序的优先级被强制设置，它会干扰了rcu stall的检测线程，从而导致问题出现

本文的问题在于中断风暴。
# 分析
根据日志，初步判断堆栈可以知道如下
1. 每次都是在netrx过程的堆栈
2. softirq的计数会变
3. cpu核心不变
4. 宽限期gp序号g会变

可以知道，这并不是完全在cpu0上停滞了，而是不断的有softirq在处理，并且softirq的数量特别多，也就意味着很可能有中断风暴。 结合日志，中断风暴来自can驱动。

# 结合现象
初步定位到是中断风暴，那么简单确认系统现象可以知道复现问题的时候，can的硬件中断异常升高，百万级到千万级，ksoftirq 100% kworker 100%。所以可以直接分析can驱动的中断判断问题了。

# 代码
分析到这，直接从代码找答案即可。关键函数在netif_receive_skb，其代码路径如下
```
devm_request_irq(rockchip_can_interrupt)
    rockchip_can_err
        netif_receive_skb
```
这些就比较明朗了，问题来自can总线发送了大量的中断，中断指向了can的错误帧，接下来解析错误帧内容即可。比较重点的寄存器如下
```
#define BUS_ERR_INT     BIT(6)
#define TX_LOSTARB_INT      BIT(5)
err_code = rockchip_canfd_read(rcan, CAN_ERR_CODE);
```
这里一个值是0x20，一个是0x40。

对于TX_LOSTARB_INT其原因是总线竞争失败错误，会重新竞争。这个不是问题的原因，只能代表当前总线上负载太高了。   
对于 BUS_ERR_INT 是rx接收crc校验错误，然后发送的bus err中断，这个中断特别多，问题出现在这。

# 根本问题
通过查阅寄存器手册，和不断的打印can的state寄存器，发现芯片存在如下现象：

1. 这个can的ip在发送扩展帧时候，内部寄存器状态值在特定组合条件下，触发load失败，从而最后按标准帧而非扩展帧的格式来发送id和dlc段。
2. 这个can的ip概率性出现CRC校验错误和ID段填充位错误，导致接收方会往总线发送错误帧，由发送方进行重发。

上面两个条件一组合，并且还是在车载多域融合的情况下使用的can通信，那么整个can的通信环境就会出现大量的错误帧，设备接受到错误帧，由控制器主动发送中断，驱动获取中断状态，进而产生中断风暴。

# 解决办法
到这里，已经知道问题的根本点在于can的控制器在特定情况下触发的中断风暴，所以解决也很简单，芯片问题，驱动只能workaround，也就是屏蔽，操作如下
```
rockchip_canfd_write(rcan, CAN_INT_MASK, BUS_ERR_INT);
```
当恢复到正常传输时，再打开 BUS_ERR_INT 中断，这样能够有效的解决此问题，代码如下
```
val &= ~BUS_ERR_INT;
rockchip_canfd_write(rcan, CAN_INT_MASK, val);
```

# 原厂方案
因为芯片已经量产了，这种问题只能规避。如果需要简单有效的办法，建议可以按照上面我的操作，针对此情况屏蔽BUS_ERR_INT中断，恢复后再开启BUS_ERR_INT中断。

反馈原厂后，原厂最后解决办法是禁用软件实现的can_bus_off，通过配置寄存器的方式，重新开关can控制器，这种方法默认代码已经同步了。代码如下
```
    if (rcan->can.state >= CAN_STATE_BUS_OFF ||
        ((sta_reg & CAN_BUSOFF_FLAG) == CAN_BUSOFF_FLAG)) {
        cancel_delayed_work(&rcan->tx_err_work);
        netif_stop_queue(ndev);
        rockchip_canfd_stop(ndev);
        can_free_echo_skb(ndev, 0);
        rockchip_canfd_start(ndev);
        netif_start_queue(ndev);
    }
```
无论哪种办法，都能解决问题。

# 总结
本文通过分享了一个中断风暴带来的rcu stall日志的具体问题，最后的原因还是芯片内部的已知缺陷，通过修改寄存器屏蔽中断或者重新开关can控制器都能解决问题。

通过这个问题的实战，也可以知道，如果一个驱动/硬件可能有缺陷，根据我个人的建议最好还是想办法绑定到其他核心上，如果问题出现在cpu0，那么带来的是系统死机，而问题出现在非cpu0，一切还有挽救的机会。 
```
irq_set_affinity_hint(irq, get_cpu_mask(num_online_cpus() - 1));
```

