并发

从队列到信号量：计数、二值与中断里的 Give

前文：RTOS多任务传数据：全局变量、环形buffer和队列、队列集和队列句柄注册表：多对一与一对多。

队列传递的是具体的数据。信号量不传数据，只维护一个整数。

信号量有两个操作：Take 和 Give。

  • Take 从信号量中拿走一个计数，也就是 cnt--，可以阻塞等到 cnt 非 0；
  • Give 放一个计数到信号量中，也就是 cnt++，到计数上限会立刻返回不等待。

信号量和队列的差别在于，队列还会把数据拷入环形buffer，信号量只会改计数，没有 payload。 也就是说，信号量是把队列中传输数据的那部分结构删除后剩下的：

  1. 去掉 r、w 和环形buffer，因为没有数据
  2. 去掉 sender list，因为 Give 失败就返回
  3. 留下 cnt 和 receiver list
  4. 增加计数上限 max，相当于队列的长度

cnt 为 0 时候的阻塞和读空队列一样：把自己挂进等待链表；Give 时 cnt++，从链表头取出一个任务放回就绪链表，它再 cnt--，然后继续跑。

计数信号量和二值信号量

信号量就是一个整数加一条等待链表，创建信号量时要定两个数：上限 max 和计数的初值。

两种信号量机制完全一样，只是 max 不同：

  • max > 1：计数信号量，计数可以在 0 到 max 之间。
  • max = 1：二值信号量，计数只能是 0 或 1。

不存在第二套机制。分成两个名字，是因为它们对应两种不同的用法。

用法A：限制同时使用某个资源的任务数

假设有 N 份资源，初值设为 N，表示全部可用。每个任务用资源之前 Take，用完之后 Give。这样，任何时刻处在 Take 和 Give 之间的任务最多只有 N 个。N > 1 时用计数信号量；N = 1 时就是"同一时刻只允许一个任务"，也就是锁。

用法B：通知某件事发生了

初值设为 0，表示还没发生。事件发生的一方 Give，等待的一方 Take：没有事件的时候睡觉，有事件就被叫醒。 如果等待方还没来得及处理，事件又发生了好几次：二值信号量只能记录“至少发生过一次”。

Give到上限为什么不阻塞

只有同时满足两个条件，阻塞等待才是有意义的：

  1. 当前状态是协议允许的状态，只是暂时的
  2. 放弃这次操作会丢失信息

队列满了的时候，两条都满足：

  1. 生产者暂时比消费者快是正常现象
  2. 数据一旦丢弃就找不回来

所以写队列可以选择阻塞等待。

但如果信号量的计数到了 max，计数会代表什么呢？

  • 如果用计数代表资源：每次 Give 都对应一次 Take。计数到了 max，说明所有资源都已经归还了。这时候再 Give，找不到对应的 Take，是调用方的逻辑错误。错误应该报告，而阻塞等待会把错误藏起来。

  • 如果用计数代表事件：计数到了 max，说明待处理的事件已存满，等待方一定会被叫醒。对二值信号量，多出来的这次 Give 没有带来新信息。对计数信号量，会丢掉一次计数，这时候应该把 max 设大一点，而不是让发信号的一方停下来等。

两种情况下，Give 阻塞等待都换不回任何东西！

此外，Give 很常在中断里调用，而中断不能阻塞。所以 FreeRTOS 把 Give 的等待时间固定为 0，信号量里也就不需要发送者等待链表。

Take唤醒的优先级

多个任务都在等待信号量的时候，等待链表会按照任务的优先级排序（通常是 xTaskCreate 时设定的优先级），优先级相同就先到先得。

这个排序是在插入时候实现的。当任务因为 Take 不到而休眠时，内核会按照它当时的优先级将它插入等待链表中合适的位置：

  • 优先级高的排在前面；
  • 优先级相同的，排在已有同级任务的后面，所以同级之间先来先得。

Give 时只叫醒链表最前面的那一个，所以高优先级任务先拿到。

这条规则不是信号量独有的，队列也是这样，因为信号量的底层就是队列。

Give在中断里调用，会发生什么？

Give 可以在中断里调用，这时候在 Take 中阻塞的任务会不会马上被唤醒取决于两件事：

  1. 被叫醒的任务优先级是否高于被中断打断的那个任务
  2. 中断结尾有没有请求切换

这个机制和前面队列说到的 xQueueSendFromISR 完全一样，因为信号量的底层就是队列。

中断里要用 xSemaphoreGiveFromISR(sem, &woken)，不能用普通的 xSemaphoreGive。 它只做两件事：计数 +1，把等待链表最前面的任务移动到就绪链表（没任务等待就跳过）。 被叫醒的任务这时候只是可以跑了，并没有在跑。中断还没执行完，内核不会在中断里换任务。

退出中断的时候：

  • 如果被叫醒的任务优先级更高，并且结尾调用了 portYIELD_FROM_ISR(woken)，被叫醒的任务就会马上开跑；
  • 如果被叫醒的任务优先级更高，但是没有调用 yield，被打断的任务会继续跑、被叫醒的任务会在下一次调度的时候跑；
  • 如果被叫醒的任务优先级相同或者更低，被打断的任务会继续跑。

woken 由 GiveFromISR 填写，只有被叫醒的任务优先级高于被打断任务时才会设为 pdTRUE，所以标准写法是：

1
2
3

BaseType_t woken = pdFALSE;
xSemaphoreGiveFromISR(sem, &woken);
portYIELD_FROM_ISR(woken);

这样写，需要立刻切换时就会切，不需要时就回到原来的任务。