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

可以睡眠的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的冲突链，可谓奇妙！