如果为所有程序设计的是静态优先级，那么天生的弊端就是低优先级任务饿死。所以在设计调度策略的时候，考虑动态优先级是必不可少的。本文简单说明一下动态优先级
# 如何设计动态优先级
通常一个程序包含两种情况
* CPU密集型
* IO密集型

而这两种情况并不代表着程序一直都是，而是在运行过程中切换的。也就是一个任务一会儿是CPU密集型，一会儿是IO密集型。我们应该在不同情况下为其设置不同的优先级情况。通常来说，对于IO密集型的任务，应该将其优先级设置较高，让其立马能够将任务推向IO去处理，而对于CPU密集型任务，应该将其优先级稍微降低，从而保证IO型的任务不会出现饿死的情况。

这样，我们可以构造一个公式如下

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

因为在100-140优先级范围内，越高代表越优先级越低，此公式的关键点在于bonus值，这个值是等待IO的时间。也就是说，一个任务等待IO时间越长，那么其整体优先级会更高。

# 负载均衡
在多核系统上，为了避免大小脚的问题，也就是某个核心任务过重，而其他核心任务太轻，所以需要进行负载均衡。首先，我们知道现在runqueue是per-CPU的，所以当任务从一个CPU迁移到另一个CPU的时候，那么此任务相当于也是在做动态优先级调整。这种调整的目的是为了CPU的负载更均衡

在Linux上通过PELT计算负载，其思路是在一个任务开始运行的时候，先判断CPU的负载情况，如果当前CPU负载较高，则将其迁移到负载低的那个CPU队列上运行。

PELT的量化负载公式如下

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

实际算力公式：

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

在安卓平台，可能会采用WALT

# 参考
> https://static.linaro.org/connect/sfo17/Presentations/SFO17-307%20WALT%20vs%20PELT.pdf

调度(2)-动态优先级的必要性

早期的Linux调度策略设计的比较简单，逐步的随着linux的越来越火热，调度问题受到了很多人的关注，本文基于历史和一些数据总结一下linux2.4和linux2.6的调度状态。为什么从这两个版本说呢，因为现在linux内核的调度策略的雏形是从2.6开始形成的

# Linux 2.4
在linux 2.4时，内核使用全局队列来管理任务，当任务到来的时候，此队列在content switch的时候会遍历整个runqueue，找到优先级最高的任务来运行。这样的设计有如下几个问题
* O(n)负载度：因为没有考虑设计问题，所以调度器本身的效率低下
* 锁争用：所有CPU都在同一个全局runqueue上争抢锁，这使得锁竞争巨大

# Linux 2.6 
2.6版本的调度器为了设计O(1)的调度器，对齐进行了整体的设计，首先为了解决锁的问题，调度队列是通过per-cpu上维护两个队列（active和expired），0-99作为实时任务，值越高优先级越高，100-140作为普通任务。下面来介绍2.6上调度是如何设计的。

其实翻看我之前rtems的分析就可以提前知道，调度器通过bitmap+拉链来设计O(1)的代码。同样的Linux 2.6就是这么设计的。它有两个关键元素
* bitmask: 以优先级作为bitmask，这样在寻找优先级最高的任务的时候，只是高效的bitmap就能知道哪个优先级内的队列非空
* queue： 每个优先级的bit下对应一个拉链，拉链本身是fifo结构的，当找到最高优先级的非空队列的时候，只需要在队列里面pop最先入队的任务即可
* active和expired： 在per-CPU上设计两个队列轮转，active是需要立即调度的任务队列，expire将要调度的任务队列。当active的任务运行完毕之后，active和expired做一次轮转

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

通过图可以演示这两个队列

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

此时OS从100上选择了一个任务来运行，当运行完成之后，该任务被放置在expired队列上，等待轮转后运行

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

当active runqueues的任务都运行完成了，则active和expired做一次轮转。

## 关于timeslice
调度器的分片是在时钟tick中计算的。默认会分配一个值，每次tick到期，会对当前正在运行的任务的timeslice值自减1。当tick完成的时候，也就是timeslice完成的时候。就是将该任务从active迁移到expired的时机。

# 总结
本文介绍了基本的调度概念，可以强化理解O(1)调度器的历史由来。

# 参考
> https://zhuanlan.zhihu.com/p/33461281











调度(1)-基本调度概念

在开发高性能的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



