rculist_nulls使用详解：Netfilter会话表的锁保护

独孤九贱

马上就要2012.12.21日了，在这个特殊的日子来临之际，发个灌水贴，以示纪念。

作者：独孤九贱
转载请注明出处。

1、概述
RCU锁相对于RWLOCK来讲，对于很少的写者，而多个读取并发的情况下，可以带来更好的并发性，关于原理及对比数据分析，这里就不再赘述了。内核也为链表使用RCU锁做了实现,即RCUList，事实上，这也是RCU锁一个最重要的应用之一。但是做为对RCULIst的扩展的RCUList_null，虽然引入有较长一段时间了，但是由于内核子系统使用较少，所以还不太知名，但是不巧的是网络栈的Netfilter的连接跟踪使用了它。内核对这个接口是这样描述的：

1. Using hlist_nulls to protect read-mostly linked lists and objects using SLAB_DESTROY_BY_RCU allocations.
   
2. Please read the basics in Documentation/RCU/listRCU.txt

复制代码

本文将结合Netfilter的conntrack表以及内核文档中对rculist_nulls的使用的介绍，从rculist_null、自旋锁、引用计数器、定时器的同步几个方面来详细分析其用法。所以阅读本文需要参考的nf_conntrack_core.c和Documentation/RCU/rculist_nulls.txt 两个文件。

2、数据结构

1. struct nf_conn {
   
2.         /* Usage count in here is 1 for hash table/destruct timer, 1 per skb,
   
3.            plus 1 for any connection(s) we are `master' for */
   
4.         struct nf_conntrack ct_general;
   
5. 
   
6.         spinlock_t lock;
   
7.        
   
8.         /* XXX should I move this to the tail ? - Y.K */
   
9.         /* These are my tuples; original and reply */
   
10.         struct nf_conntrack_tuple_hash tuplehash[IP_CT_DIR_MAX];
    
11.        

复制代码

ct_general中包含了引用计数器：

1. #if defined(CONFIG_NF_CONNTRACK) || defined(CONFIG_NF_CONNTRACK_MODULE)
   
2. struct nf_conntrack {
   
3.         atomic_t use;
   
4. };
   
5. #endif

复制代码

lock是用于整个结构的互斥锁

ct的两个方向的tuple都包含了hlist_nulls节点成员：

1. /* Connections have two entries in the hash table: one for each way */
   
2. struct nf_conntrack_tuple_hash {
   
3.         struct hlist_nulls_node hnnode;
   
4.         ……
   
5. };

复制代码

另外，整个conntrack表，有一个全局锁

1. spinlock_t nf_conntrack_lock ;

复制代码

它主要用于对表的添加、修改、遍历（不是查找）等操作。

conntrack使用了以上结构体成员或变量来实现相应的并发互斥。

3、节点分配与分初始化

整个高速缓存的创建，注意SLAB_DESTROY_BY_RCU的使用：

1. static int nf_conntrack_init_net(struct net *net)
   
2. {
   
3.         //初始化节点总的计数器
   
4.         atomic_set(&net->ct.count, 0);
   
5.        
   
6.         //分配高速缓存
   
7.         net->ct.nf_conntrack_cachep = kmem_cache_create(net->ct.slabname,
   
8.                                                         sizeof(struct nf_conn), 0,
   
9.                                                         SLAB_DESTROY_BY_RCU, NULL);       
   
10. }

复制代码

init_conntrack函数完成节点的分配与初始化：

1.         //分配高速缓存
   
2.         ct = kmem_cache_alloc(net->ct.nf_conntrack_cachep, gfp);
   
3. 
   
4.         //初始化锁、引用计数器等成员       
   
5.         spin_lock_init(&ct->lock);
   
6.        
   
7.         ct->tuplehash[IP_CT_DIR_ORIGINAL].hnnode.pprev = NULL;
   
8.        
   
9.         /*
   
10.          * changes to lookup keys must be done before setting refcnt to 1
    
11.          */
    
12.         smp_wmb();
    
13.         atomic_set(&ct->ct_general.use, 1);

复制代码

这与RCULIst_null.txt中介绍是相同的，当然，文档中同时讲了节点的插入操作，关于conntrack的插入，后文将会分析到：

1. 2) Insert function :
   
2. --------------------
   
3. /*
   
4. * Please note that new inserts are done at the head of list,
   
5. * not in the middle or end.
   
6. */ obj = kmem_cache_alloc(cachep); lock_chain(); // typically a spin_lock() obj->key = key; /*
   
7. * changes to obj->key must be visible before refcnt one
   
8. */ smp_wmb(); atomic_set(&obj->refcnt, 1); /*
   
9. * insert obj in RCU way (readers might be traversing chain)
   
10. */ hlist_nulls_add_head_rcu(&obj->obj_node, list); unlock_chain(); // typically a spin_unlock()

复制代码

4、节点的插入
__nf_conntrack_confirm函数完成相应的插入操作：

1. 
   
2. /* Confirm a connection given skb; places it in hash table */
   
3. int
   
4. __nf_conntrack_confirm(struct sk_buff *skb)
   
5. {
   
6.         ……
   
7.         //加全局锁
   
8.         spin_lock_bh(&nf_conntrack_lock);
   
9. 
   
10.         //加引用计数器
    
11.         atomic_inc(&ct->ct_general.use);
    
12. 
    
13.         /* Since the lookup is lockless, hash insertion must be done after
    
14.          * starting the timer and setting the CONFIRMED bit. The RCU barriers
    
15.          * guarantee that no other CPU can find the conntrack before the above
    
16.          * stores are visible.
    
17.          */
    
18.         __nf_conntrack_hash_insert(ct, hash, repl_hash);
    
19.        
    
20.         spin_unlock_bh(&nf_conntrack_lock);
    
21.         ……
    
22. }

复制代码

分配节点和插入的时候，引用计数器都加了1.这与netfilter的语义有关，分配的时候，将其设置为1，
与查找的时候，增加引用计数器原因相同：它们都是为skb使用conntrack准备的。而这里加1，则是为hash
表准备的：表明hash表已经在使用它了。

1. //具体的插入函数       
   
2. static void __nf_conntrack_hash_insert(struct nf_conn *ct,
   
3.                                        unsigned int hash,
   
4.                                        unsigned int repl_hash)
   
5. {
   
6.         struct net *net = nf_ct_net(ct);
   
7. 
   
8.         hlist_nulls_add_head_rcu(&ct->tuplehash[IP_CT_DIR_ORIGINAL].hnnode,
   
9.                            &net->ct.hash[hash]);
   
10.         hlist_nulls_add_head_rcu(&ct->tuplehash[IP_CT_DIR_REPLY].hnnode,
    
11.                            &net->ct.hash[repl_hash]);
    
12. }

复制代码

可以看到，这里的分配、插入操作，都是rculist_nulls.txt中代码的翻版。
未完，待续……

独孤九贱

5、查找
查找是rculist_null最为重要的应用，可以说所有的准备都是为它而来，
____nf_conntrack_find完成hash表的查找工作，它返回要查找的节点，
该函数的结构为标准的hlist_nulls的查找，但是这个函数没有RCU锁的保护，
作者将其一分为二，锁放在上层了调用函数了：

1. 
   
2. /*
   
3. * Warning :
   
4. * - Caller must take a reference on returned object
   
5. *   and recheck nf_ct_tuple_equal(tuple, &h->tuple)
   
6. * OR
   
7. * - Caller must lock nf_conntrack_lock before calling this function
   
8. */
   
9. static struct nf_conntrack_tuple_hash *
   
10. ____nf_conntrack_find(struct net *net, u16 zone,
    
11.                       const struct nf_conntrack_tuple *tuple, u32 hash)
    
12. {
    
13.         struct nf_conntrack_tuple_hash *h;
    
14.         struct hlist_nulls_node *n;
    
15.         unsigned int bucket = hash_bucket(hash, net);
    
16. 
    
17.         /* Disable BHs the entire time since we normally need to disable them
    
18.          * at least once for the stats anyway.
    
19.          */
    
20.         local_bh_disable();
    
21. begin:
    
22.         hlist_nulls_for_each_entry_rcu(h, n, &net->ct.hash[bucket], hnnode) {
    
23.                 if (nf_ct_tuple_equal(tuple, &h->tuple) &&
    
24.                     nf_ct_zone(nf_ct_tuplehash_to_ctrack(h)) == zone) {
    
25.                         NF_CT_STAT_INC(net, found);
    
26.                         local_bh_enable();
    
27.                         return h;
    
28.                 }
    
29.                 NF_CT_STAT_INC(net, searched);
    
30.         }
    
31.         /*
    
32.          * if the nulls value we got at the end of this lookup is
    
33.          * not the expected one, we must restart lookup.
    
34.          * We probably met an item that was moved to another chain.
    
35.          */
    
36.         if (get_nulls_value(n) != bucket) {
    
37.                 NF_CT_STAT_INC(net, search_restart);
    
38.                 goto begin;
    
39.         }
    
40.         local_bh_enable();
    
41. 
    
42.         return NULL;
    
43. }

复制代码

上层调用函数使用RCU锁保护，来调用该查找函数，并且如果查找命中，增加引用计数器：

1. 
   
2. /* Find a connection corresponding to a tuple. */
   
3. static struct nf_conntrack_tuple_hash *
   
4. __nf_conntrack_find_get(struct net *net, u16 zone,
   
5.                         const struct nf_conntrack_tuple *tuple, u32 hash)
   
6. {
   
7.         struct nf_conntrack_tuple_hash *h;
   
8.         struct nf_conn *ct;
   
9. 
   
10.         rcu_read_lock();
    
11. begin:
    
12.         h = ____nf_conntrack_find(net, zone, tuple, hash);
    
13.         if (h) {
    
14.                 ct = nf_ct_tuplehash_to_ctrack(h);
    
15.                 if (unlikely(nf_ct_is_dying(ct) ||
    
16.                              !atomic_inc_not_zero(&ct->ct_general.use)))
    
17.                         h = NULL;
    
18.                 else {
    
19.                         if (unlikely(!nf_ct_tuple_equal(tuple, &h->tuple) ||
    
20.                                      nf_ct_zone(ct) != zone)) {
    
21.                                 nf_ct_put(ct);
    
22.                                 goto begin;
    
23.                         }
    
24.                 }
    
25.         }
    
26.         rcu_read_unlock();
    
27. 
    
28.         return h;
    
29. }

复制代码

这同样与rculist_nulls.txt中描述是完全相同的，只是实现的时候切割开了：

1. 1) lookup algo
   
2. head = &table[slot];
   
3. rcu_read_lock(); begin: hlist_nulls_for_each_entry_rcu(obj, node, head, member) { if (obj->key == key) { if (!try_get_ref(obj)) // might fail for free objects goto begin; if (obj->key != key) { // not the object we expected put_ref(obj);
   
4.          goto begin; } goto out; } /*
   
5. * if the nulls value we got at the end of this lookup is
   
6. * not the expected one, we must restart lookup.
   
7. * We probably met an item that was moved to another chain.
   
8. */
   
9. if (get_nulls_value(node) != slot) goto begin; obj = NULL;
   
10. out: rcu_read_unlock();

复制代码

6、更新
当查找到的节点需要对其成员进行操作时，需要加节点锁，如下所示：

1. static int tcp_packet(struct nf_conn *ct,
   
2.                       const struct sk_buff *skb,
   
3.                       unsigned int dataoff,
   
4.                       enum ip_conntrack_info ctinfo,
   
5.                       u_int8_t pf,
   
6.                       unsigned int hooknum)
   
7. {
   
8.         spin_lock_bh(&ct->lock);
   
9.        
   
10.         //对ct成员的若干操作
    
11.         ……
    
12.        
    
13.         spin_unlock_bh(&ct->lock);
    
14. }

复制代码

这里，最重要的理解查找操作不需要更多的自旋锁保护，只需rcu_read_lock即可。这涉及到RCU锁的原理与实现，本文侧重于rculist_null的使用接口介绍与分析，不再RCU相关东东了。
未完，待续……


   

独孤九贱

7、删除
7.1 超时释放

节点的定时器超时，会调用death_by_timeout函数：

1. 
   
2. static void death_by_timeout(unsigned long ul_conntrack)
   
3. {
   
4.         ……
   
5.         nf_ct_delete_from_lists(ct);
   
6.         nf_ct_put(ct);
   
7. }

复制代码

nf_ct_delete_from_lists完成从链表中释放

1. void nf_ct_delete_from_lists(struct nf_conn *ct)
   
2. {
   
3.         ……
   
4.         //全局锁
   
5.         spin_lock_bh(&nf_conntrack_lock);
   
6.         /* Inside lock so preempt is disabled on module removal path.
   
7.          * Otherwise we can get spurious warnings. */
   
8.         ……
   
9.         clean_from_lists(ct);
   
10.         spin_unlock_bh(&nf_conntrack_lock);
    
11. }

复制代码

对于添加和删除链表的操作，是需要一个链表的自旋锁保护，这里是nf的全局锁nf_conntrack_lock。

1. static void
   
2. clean_from_lists(struct nf_conn *ct)
   
3. {
   
4.         pr_debug("clean_from_lists(%p)\n", ct);
   
5.         hlist_nulls_del_rcu(&ct->tuplehash[IP_CT_DIR_ORIGINAL].hnnode);
   
6.         hlist_nulls_del_rcu(&ct->tuplehash[IP_CT_DIR_REPLY].hnnode);
   
7. 
   
8.         /* Destroy all pending expectations */
   
9.         nf_ct_remove_expectations(ct);
   
10. }

复制代码

nf_ct_put释放节点高速缓存：

1. 
   
2. /* decrement reference count on a conntrack */
   
3. static inline void nf_ct_put(struct nf_conn *ct)
   
4. {
   
5.         NF_CT_ASSERT(ct);
   
6.         nf_conntrack_put(&ct->ct_general);
   
7. }
   
8. 
   
9. static inline void nf_conntrack_put(struct nf_conntrack *nfct)
   
10. {
    
11.         if (nfct && atomic_dec_and_test(&nfct->use))
    
12.                 nf_conntrack_destroy(nfct);
    
13. }

复制代码

nf_conntrack_destroy事实上会调用destroy_conntrack->nf_conntrack_free函数，
最终释放高速缓存：

1. 
   
2. void nf_conntrack_free(struct nf_conn *ct)
   
3. {
   
4.         ……
   
5.         //递减计数器
   
6.         atomic_dec(&net->ct.count);
   
7.         //释放高速缓存
   
8.         kmem_cache_free(net->ct.nf_conntrack_cachep, ct);
   
9. }

复制代码

这段代码示例很好地展示了全局锁与单个节点的引用计数器的配合使用，完整全部的释放工作。

7.2 使用者释放
conntrack查找命中后，会将找到的ct节点指针赋给skb

1. skb->nfct = &ct->ct_general;                        //resolve_normal_ct函数

复制代码

当skb被释放时，会调用上述nf_conntrack_put函数尝试释放引用：

1. static void skb_release_head_state(struct sk_buff *skb)
   
2. {
   
3. #if defined(CONFIG_NF_CONNTRACK) || defined(CONFIG_NF_CONNTRACK_MODULE)
   
4.         nf_conntrack_put(skb->nfct);
   
5. #endif
   
6. }

复制代码

这里有意思是的skb释放是直接释放高速缓存，并不涉及链表操作，有两种场景：
a、超时函数并没有触发，那么只里仅是增加、减少引用计数器而已，不涉及其它；
b、超时函数被调用，节点被从hash表中删除并减少引用计数器，而释放则放到这里；

同样地，这也与文档中描述相同：

1. 3) Remove algo
   
2. -------------- Nothing special here, we can use a standard RCU hlist deletion. But thanks to SLAB_DESTROY_BY_RCU, beware a deleted object can be reused very very fast (before the end of RCU grace period)
   
3. if (put_last_reference_on(obj) { lock_chain(); // typically a spin_lock()
   
4.    hlist_del_init_rcu(&obj->obj_node);
   
5.    unlock_chain(); // typically a spin_unlock()
   
6.    kmem_cache_free(cachep, obj); }

复制代码

8、全清       

下面来看对整个hash表的节点的清空操作：

1. 
   
2. static void nf_conntrack_cleanup_net(struct net *net)
   
3. {
   
4. i_see_dead_people:
   
5.         nf_ct_iterate_cleanup(net, kill_all, NULL);
   
6. 
   
7.         //ct.count为整个hash表的节点计数器，这里循环直至所有节点被释放，
   
8.         //这是因为此时模块移除时，Netfilter Hook函数或网络软中断函数可能拥有conntrack节点。
   
9.         if (atomic_read(&net->ct.count) != 0) {
   
10.                 schedule();
    
11.                 goto i_see_dead_people;
    
12.         }
    
13. 
    
14.         ……
    
15.         //释放高速缓存       
    
16.         kmem_cache_destroy(net->ct.nf_conntrack_cachep);
    
17.         ……
    
18. }

复制代码

nf_ct_iterate_cleanup遍历整个hash表，释放节点：

1. 
   
2. void nf_ct_iterate_cleanup(struct net *net,
   
3.                            int (*iter)(struct nf_conn *i, void *data),
   
4.                            void *data)
   
5. {
   
6.         struct nf_conn *ct;
   
7.         unsigned int bucket = 0;
   
8. 
   
9.         while ((ct = get_next_corpse(net, iter, data, &bucket)) != NULL) {
   
10.                 /* Time to push up daises... */
    
11.                 if (del_timer(&ct->timeout))
    
12.                         death_by_timeout((unsigned long)ct);
    
13.                 /* ... else the timer will get him soon. */
    
14. 
    
15.                 nf_ct_put(ct);
    
16.         }
    
17. }

复制代码

这里需要注意的是定时器的删除操作，即del_timer函数的调用：
1、当定时器已经被调度执行，其将直接返回0，nf_ct_put将被执行，
这意味着后续的nf_ct_put函数可与在其它CPU上执行的定时器超时函数death_by_timeout并行执行，
它们的互斥显然由引用计数器来实现。而不需要再调用del_timer_sync等待同步；

2、另一种情况，定时器并没有被调度，则直接删除定时器，并且直接调用death_by_timeout来实现删除操作；

get_next_corpse在遍历hash表时，查找节点的时候，会增加引用计数器，所以，循环内部最后调用nf_ct_put(ct)释放之，而不是：

1. 
   
2.                 if (del_timer(&ct->timeout)){
   
3.                         death_by_timeout((unsigned long)ct);
   
4.                         continue;
   
5.                 }

复制代码

get_next_corpsehash表的遍历，逐个返回节点：

1. 
   
2. /* Bring out ya dead! */
   
3. static struct nf_conn *
   
4. get_next_corpse(struct net *net, int (*iter)(struct nf_conn *i, void *data),
   
5.                 void *data, unsigned int *bucket)
   
6. {
   
7.         struct nf_conntrack_tuple_hash *h;
   
8.         struct nf_conn *ct;
   
9.         struct hlist_nulls_node *n;
   
10. 
    
11.         spin_lock_bh(&nf_conntrack_lock);
    
12.         for (; *bucket < net->ct.htable_size; (*bucket)++) {
    
13.                 hlist_nulls_for_each_entry(h, n, &net->ct.hash[*bucket], hnnode) {
    
14.                         ct = nf_ct_tuplehash_to_ctrack(h);
    
15.                         if (iter(ct, data))
    
16.                                 goto found;
    
17.                 }
    
18.         }
    
19.         hlist_nulls_for_each_entry(h, n, &net->ct.unconfirmed, hnnode) {
    
20.                 ct = nf_ct_tuplehash_to_ctrack(h);
    
21.                 if (iter(ct, data))
    
22.                         set_bit(IPS_DYING_BIT, &ct->status);
    
23.         }
    
24.         spin_unlock_bh(&nf_conntrack_lock);
    
25.         return NULL;
    
26. found:
    
27.         atomic_inc(&ct->ct_general.use);
    
28.         spin_unlock_bh(&nf_conntrack_lock);
    
29.         return ct;
    
30. }

复制代码

可以看到，遍历的时候，同样需要全局锁的保护，而单个节点的使用，依靠的是引用计数器。

全局终！！！

Godbach

回复 2# 独孤九贱
九贱兄又有大作问世，好好拜读先。


   

独孤九贱

Godbach 发表于 2012-12-19 11:45
回复 2# 独孤九贱
九贱兄又有大作问世，好好拜读先。

客气，我也是因为工作需要，看一下，然后把笔记放上来了，大家共同学习嘛

daniel_11

支持楼主大作！学习下～

瀚海书香

回复 1# 独孤九贱
一到年底九贱兄就有干货啊

   

liuiang

马克一下，回头要研究。

915086731

你好，独孤老师。
我也在看Linux内核连接跟踪这一块的代码，大部分函数明白了。有个问题， 为什么要增加ct->ct_general.use的引用计数，既然是要清楚这个连接跟踪，增加引用计数有什么意义？

1.         atomic_inc(&ct->ct_general.use);
   

复制代码

915086731

不明白为什么get_next_corpse要增加计数

get_next_corpse在遍历hash表时，查找节点的时候，会增加引用计数器，所以，循环内部最后调用nf_ct_put(ct)释放之