rcu锁所使用的一个机制--dynticks

痛定思痛。 2022-01-11 00:57 303阅读 0赞

可以睡眠的rcu锁很大程度上有益于系统的实时性，因为不用禁用抢占了，该rcu锁巧妙的使用两个“阶段”来跟踪rcu的状态，内部维持着一个状态机，该状态机的其中两个状态需要所有cpu的确认，也就是说只有所有cpu都确认了之后，状态机才能向前推进，这就暴露出一个缺点，就是在nohz启用的情况下，如果一个cpu不再接收时钟中断，那么它就没有机会执行确认，结果就是所有的和rcu相关的cpu需要等待很长一段时间才能推进rcu状态机，一直等到这个停掉时钟心跳的cpu重新开始心跳，怎么解决这个问题呢？内核中引入了一个叫做dynticks的机制，该机制很简单，就是当一个cpu处于心跳停止状态的时候可以直接跳过它，不需要它的确认，毕竟它上面是不可能操作read-rcu锁的，代码实现也非常清晰：

static inline void rcu\_enter\_nohz(void)

smp\_mb(); /\* CPUs seeing ++ must see prior RCU read-side crit sects \*/

\_\_get\_cpu\_var(dynticks\_progress\_counter)++;

if (unlikely(\_\_get\_cpu\_var(dynticks\_progress\_counter) & 0x1)) \{

...

\_\_get\_cpu\_var(dynticks\_progress\_counter)++;

static inline void rcu\_exit\_nohz(void)

\_\_get\_cpu\_var(dynticks\_progress\_counter)++;

smp\_mb(); /\* CPUs seeing ++ must see later RCU read-side crit sects \*/

if (unlikely(!(\_\_get\_cpu\_var(dynticks\_progress\_counter) & 0x1))) \{

...

\_\_get\_cpu\_var(dynticks\_progress\_counter)++;

每一个cpu被引入了一个变量--dynticks\_progress\_counter，该变量被初始化为1，在进入nohz状态时和退出时该变量都会递增，然后在需要等待cpu确认的地方之需要判断该变量的奇偶即可，如果是偶数，那么就说明进入了nohz心跳停止状态，直接跳过即可，就当它已经确认了，反之如果是奇数就需要等待确认，但是问题出来了，如果一个中断打断了nohz的cpu，而该中断中有操作read-rcu的逻辑，那么就可能导致系统挂起，因为可能dynticks机制跳过了改cpu，但是该cpu即刻被中断所打断，致使它不能被跳过，最终可能导致两个数组混乱掉，于是在每个中断进入和退出的时候必然也需要有相应的逻辑来优化dynticks机制，这就是cu\_irq\_enter和rcu\_irq\_exit要做的：

void rcu\_irq\_enter(void)

int cpu = smp\_processor\_id();

if (per\_cpu(rcu\_update\_flag, cpu))//如果已经在中断中了，那么嵌套中断中同样将rcu\_update\_flag递增，然后在内层中断的rcu\_irq\_exit中会导致递减rcu\_update\_flag不为0，直接退出，再更新dynticks\_progress\_counter的值，该机制只考虑是否被中断，并不考虑具体中断的层数

per\_cpu(rcu\_update\_flag, cpu)++;

if (!in\_interrupt() &&

(per\_cpu(dynticks\_progress\_counter, cpu) & 0x1) == 0) \{//注意，这个判断并不是所有进入中断时都为真的，只有在从nohz状态进入时才为真

per\_cpu(dynticks\_progress\_counter, cpu)++;

smp\_mb(); /\* see above block comment. \*/

per\_cpu(rcu\_update\_flag, cpu)++;

void rcu\_irq\_exit(void)

int cpu = smp\_processor\_id();

if (per\_cpu(rcu\_update\_flag, cpu)) \{

if (--per\_cpu(rcu\_update\_flag, cpu)) //处理嵌套中断的逻辑，如果非嵌套，那么该rcu\_update\_flag在此时应该被减为0的。

return;

smp\_mb(); /\* see above block comment. \*/

per\_cpu(dynticks\_progress\_counter, cpu)++;

WARN\_ON(per\_cpu(dynticks\_progress\_counter, cpu) & 0x1);

有了以上两个函数把关，基本上就没有什么问题了，最后在rcu\_try\_flip\_waitack\_needed中进行判断：

static inline int rcu\_try\_flip\_waitack\_needed(int cpu)

long curr;

long snap;

curr = per\_cpu(dynticks\_progress\_counter, cpu);

snap = per\_cpu(rcu\_dyntick\_snapshot, cpu);

smp\_mb(); /\* force ordering with cpu entering/leaving dynticks. \*/

if ((curr == snap) && ((curr & 0x1) == 0)) //仍然在idle也就是nohz中

return 0;

\* if ((curr - snap) > 2 || (snap & 0x1) == 0) //这里有些复杂哦！

return 0;

return 1;

上面函数的\*行的判断逻辑别看短，非常复杂，注意看一下cpu\_idle函数中的内层while循环，也就是该idlecpu仅仅被中断了一次，但是还没有被彻底从idle状态唤醒的逻辑，里面调用了rcu\_check\_callbacks函数，该函数本质上就是处理本cpu上的工作，“尽可能”的推进状态机，然后更新本cpu的rcu数据，为何是尽可能推进呢？想想看如果大家都等它了，那么只要它一确认，那么状态机就推进了，就是这个意思。我们说rcu\_try\_flip\_waitack\_needed这个函数就是尽可能的跳过处于nohz状态的cpu，不需要它们确认状态机的状态，但是这是有条件的，可抢占rcu的两个阶段的等待加和为0的数组索引是通过rcu\_ctrlblk.completed的奇偶来确定的，这就是说如果这个索引错了，那么一切数据就会乱掉。现在看看这个索引怎么会错，在rcu状态机推进到下一个状态的时候，rcu\_ctrlblk.completed会递增，然后等待所有的cpu“看到”这个“递增事件”，如果看不到，那么数据就乱掉了。如果说if ((curr - snap) > 2 || (snap & 0x1) == 0)很晦涩的话，我们不妨采用反证法，这个if逻辑的反面就是(curr - snap) <= 2 && (snap & 0x1) == 1，(curr - snap) <= 2无非就是0，1，2，而(snap & 0x1) == 1无非就是在rcu推进到该阶段的时候cpu的dynticks值为奇数，为奇数的情况无非有两种，一种是未处于nohz状态，另一种是处于nohz状态，但是被中断了，并且中断还没有退出（因为中断退出时又会将dynticks加为偶数），先看未处于nohz状态的情形，如果未处于nohz状态，那么curr和当时的dynticks差值是1或者2的情形就是进入了nohz状态没有被中断或者进入nohz后被中断，不管哪一类情形，都是在cpu\_idle的内层while循环中，而内层循环已经调用了rcu\_check\_callbacks“尽可能”的推进了rcu状态机，中断处理函数中可能已经释放或者得到了rcu锁，直接导致数组变化，因此必须要确认；另外一种就是cpu被中断了，但是还没有退出中断，这个和上面的分析结果一致。这下这个if逻辑就清晰多了，正过来考虑该if逻辑，如果(curr - snap) > 2成立，那么可以肯定要么发生了进入nohz又出来的情形，要么发生了本来就在nohz中被中断了至少两次的情形，无论哪一种情形都不会导致数组索引错误。

从另一个角度来看上面的if判断，只要是(snap & 0x1) == 0成立，那么dyntick\_save\_progress\_counter肯定是在cpu\_idle中的内层循环中被调用的，那么既然是内层循环，那就说明是nohz状态，时钟中断已然停止，我们知道推进rcu状态机的函数是在时钟中断函数中被调用的，因此如果等待确认的话会带来很长的延时，不等待确认不会有任何问题，进入nohz之后，该cpu肯定进入了一个崭新的状态，不会携带有歧义的数据的；再看(curr - snap) > 2这个条件，分为snap进入nohz和未进入nohz两种情形，第一种情形下，要想使差值大于2，那么必然要么被中断(差值1)再出来(差值2)再被中断（差值3）…如果差值为1或者2时，必然有一次机会调用rcu的callback函数，这样即使等待也不会造成长久的延时；如果是snap时没有进入nohz，那么差值为1或者2就是进入nohz或者进入后再被中断一次，既然是中断那么必然会离开中断，离开的话就有可能调用callback函数，因此可以等待确认。反过来如果(snap & 0x1) == 1，那么加上两个计数器差值为1或者2的条件下就必须等待确认，如果差值为1或者2，那么可能根本就没有执行callback函数，从而导致歧义。

赏析一个函数--udp\_v4\_get\_port

首先呈上这个函数：

static int udp\_v4\_get\_port(struct sock \*sk, unsigned short snum)

struct hlist\_node \*node;

struct sock \*sk2;

struct inet\_sock \*inet = inet\_sk(sk);

write\_lock\_bh(&udp\_hash\_lock);

if (snum == 0) \{

int best\_size\_so\_far, best, result, i;

if (udp\_port\_rover > sysctl\_local\_port\_range\[1\] ||

udp\_port\_rover < sysctl\_local\_port\_range\[0\])

udp\_port\_rover = sysctl\_local\_port\_range\[0\];

best\_size\_so\_far = 32767;

best = result = udp\_port\_rover;

for (i = 0; i < UDP\_HTABLE\_SIZE; i++, result++) \{

struct hlist\_head \*list;

int size;

list = &udp\_hash\[result & (UDP\_HTABLE\_SIZE - 1)\];

if (hlist\_empty(list)) \{ //如果该冲突链是空的，那么直接就可以得到了

...

goto gotit;

size = 0;

sk\_for\_each(sk2, node, list) //否则。如果可以的话记录最小size的冲突链

if (++size >= best\_size\_so\_far)//这里实现了去的最小size冲突链的算法，很巧妙

goto next;

best\_size\_so\_far = size;//到此为止，best\_size\_so\_far是最小size的冲突链

best = result;

next:;

result = best;

for(i = 0; i < (1 << 16) / UDP\_HTABLE\_SIZE; i++, result += UDP\_HTABLE\_SIZE) \{//注意，result一次递增UDP\_HTABLE\_SIZE实际上就到了下一条哈希冲突链上了

...

if (!udp\_lport\_inuse(result))//遍历该链上的所有的socket，只要有一个端口是result，那么就说明失败

break;

if (i >= (1 << 16) / UDP\_HTABLE\_SIZE) //如果没有找到

goto fail;

gotit:

udp\_port\_rover = snum = result;

\} else \{

...

write\_unlock\_bh(&udp\_hash\_lock);

return 0;

...

以上列举的这一段代码的目的是取得一个没有使用的端口，众所周知，udp端口的大小是16位，是65536个，取得一个没有使用的端口的最直观的方法就是设置一个65536个元素的数组，数组元素为一个结构体，包含两个元素，一个是从0到65535的数字，另一个是是否已经被使用，程序逻辑就是遍历这个数组，然后得到一个是否被使用字段是0的即可，但是这个算法非常拙劣，浪费空间不说，时间复杂度也不见得很低，于是想到了哈希算法，将如此之大数目的数字散列到一些有限的桶内，这就必须将这些数字进行分类了，linux内核取了128这个不大不小的数字，也就是result & (UDP\_HTABLE\_SIZE - 1)这句代码体现的，这样的话一共就会有128个哈希桶，显然的会有很多的端口会冲突，从而连接到一条冲突链上。

为何linux内核采用这么一种算法呢，比如为何要取得最小的冲突链呢，这是因为为了在后面的第二个for循环中从最小size的冲突连开始寻找冲突链中的可用端口，size最小的结果就是udp\_lport\_inuse内部遍历的时候开销最小。事实上该函数的两个for分别实现了两种寻找空闲端口的方式，第一个for循环的优先级比第二个for循环的优先级要高，也就是说首先寻找冲突链为空的，如果找到，那么后面就不用费事遍历冲突链了，如果没有找到，那么就从最小size的冲突链开始遍历，一条一条冲突链循环遍历，这也是第二个for的作用，当然从最小size冲突链开始遍历比较节省开销了。

该函数的一个要点就是在于将问题分解成了两个部分，第一个部分是哈希优化，第二个部分是不得已的遍历，同时在问题第一个部分求解当中为第二部分埋下了伏笔，这就是得到了最小size的冲突链，可谓奇妙！

转载于:https://blog.51cto.com/dog250/1273476