linux的netlink机制

netlink作为一种用户空间和内核空间通信的机制已经有一定年头了，它不光为了内核和用户通信，还可以作为IPC机制进行进程间通信。 其实netlink定义了一个框架，人们可以基于这个框架用它来做可以做的任何事情，linux中不乏这些类似的好的框架。它们的共同点就是内核并不管它 们能做什么，然而它们真的很强大，往往可以做到的事情很多，这就是内核不问策略只管实现机制，所有策略让用户实现，netlink框架就是用来传递数据 的，内核只知道它可以传递数据而不知道为何要传递这些数据也不管这些数据是什么。你甚至可以将它用于真正的网络而不仅仅限于本机，这些都是可以的，它也用 到了sk_buff结构体，和网络套接字一样，更好的事情是它并没有触及sk_buff里面的标准字段，而仅仅用了一个扩展的cb字段，cb在 sk_buff里面的定义是char cb[40];在netlink模块里面NETLINK_CB宏就是取cb字段的，也就是netlink所用的私有字段，这样的话你就可以用 netlink向任何执行实体传输任何数据了，不限于本机。

关于用户空间的netlink套接字很简单，就和传统的网络套接字一样一样的，只不过修改了一些参数罢了。如下：

sd = socket(AF_NETLINK, SOCK_RAW,NETLINK_GENERIC);

就是建立一个用户netlink套接字。之后的bind也是很简单，注意数据结构的意义就是了。这里就不说了，下面详细说一下内核的netlink实现。内核里面建立一个netlink套接字需要如下调用：
struct sock * netlink_kernel_create(int unit, void (*input)(struct sock *sk, int len))
{
struct socket *sock;
struct sock *sk;
if (unit=MAX_LINKS)
return NULL;
if (sock_create_lite(PF_NETLINK, SOCK_DGRAM, unit, &sock))
return NULL;
if (netlink_create(sock, unit) sock_release(sock);
return NULL;
}
sk = sock->sk;
sk->sk_data_ready = netlink_data_ready; //之所以将sk_data_ready设为新的函数而不用默认的是因为为了实现一些用户策略，比如可以传入自己的input函数，待到有数据的时候自行处理。
if (input)
nlk_sk(sk)->data_ready = input;
netlink_insert(sk, 0);
return sk;
}

注 意该函数的参数input是个回调函数，在有数据的时候内核会调用它。另外sk_data_ready回调函数是套接字标准中定义的，不管什么套接字都有 sk_data_ready回调机制。在input中，你可以直接处理收到的数据，也可以不处理，在大量数据传输的情况下，在input中处理是不明智 的，正确的方式应该是建立一个内核线程专门接收数据，没有数据的时候该内核线程睡眠，一旦有了数据，input回调函数唤醒这个内核线程就是了。

static void netlink_data_ready(struct sock *sk, int len)
{
struct netlink_opt *nlk = nlk_sk(sk);
if (nlk->data_ready)
nlk->data_ready(sk, len); //这里调用的回调函数就是内核netlink套接字建立的时候传入的那个函数。
netlink_rcv_wake(sk); //告知别的进程该sock上刚完成了一次接收，可能会腾出地方以便接收新的skb
}
static inline void netlink_rcv_wake(struct sock *sk)
{
struct netlink_opt *nlk = nlk_sk(sk);
if (!skb_queue_len(&sk->sk_receive_queue))
clear_bit(0, &nlk->state);
if (!test_bit(0, &nlk->state))
wake_up_interruptible(&nlk->wait); //唤醒可能等待发送的进程
}
int netlink_unicast(struct sock *ssk, struct sk_buff *skb, u32 pid, int nonblock)
{
struct sock *sk;
int err;
long timeo;
netlink_trim(skb, gfp_any());
timeo = sock_sndtimeo(ssk, nonblock);
retry:
sk = netlink_getsockbypid(ssk, pid);
...
err = netlink_attachskb(sk, skb, nonblock, timeo); //将sock和sk_buff绑定在一起，在netlink中套接字和skb的绑定与解绑定是很频繁的。
if (err == 1)
goto retry;
if (err)
return err;
return netlink_sendskb(sk, skb, ssk->sk_protocol); //在套接字sk上传输这个skb，其实就是将这个skb排入了该sk的接收队列的后头。
}
int netlink_attachskb(struct sock *sk, struct sk_buff *skb, int nonblock, long timeo)
{//这个函数将一个sk_buff给了一个套接字sock，也就是skb与sock的绑定，在绑定之前有很多工作要做。
struct netlink_opt *nlk;
nlk = nlk_sk(sk);
if (atomic_read(&sk->sk_rmem_alloc) > sk->sk_rcvbuf || test_bit(0, &nlk->state)) {
DECLARE_WAITQUEUE(wait, current);
...
__set_current_state(TASK_INTERRUPTIBLE);
add_wait_queue(&nlk->wait, &wait);
if ((atomic_read(&sk->sk_rmem_alloc) > sk->sk_rcvbuf || test_bit(0, &nlk->state)) &&
!sock_flag(sk, SOCK_DEAD)) //如果此时这个sock不能接受这个sk，那么就要等待了，正好等在nlk->wait上，待到和该sock相关的进程在netlink_rcv_wake中唤醒之，说明可以过继skb了。
timeo = schedule_timeout(timeo);
__set_current_state(TASK_RUNNING);
remove_wait_queue(&nlk->wait, &wait);
sock_put(sk);
if (signal_pending(current)) {
kfree_skb(skb);
return sock_intr_errno(timeo);
}
return 1;
}
skb_orphan(skb);
skb_set_owner_r(skb, sk); //该sock正式接受这个sk_buff
return 0;
}

那 么谁会调用netlink_attachskb呢？这是显而易见的，在发送的时候，要把一个要发送的消息初始化成一个sk_buff结构体，但是这个 skb归谁所有呢？确定绑定的主客双方的过程就是绑定，也就是上面的函数做的事。在netlink的消息发送过程中的第一步就是sock和sk_buff 的绑定，于是调用上述绑定函数的就是netlink_sendmsg中调用的netlink_unicast，也就是单播，单播的意思就是只发给一个 sock而不是多个，于是要做的就是找到接收此skb的sock，netlink_unicast的参数dst_pid为确定目标sock提供了方向，在 netlink_unicast中（见上）正是通过这个dst_pid找到了目标sock。
static int netlink_sendmsg(struct kiocb *kiocb, struct socket *sock, struct msghdr *msg, size_t len)
{
...
err = netlink_unicast(sk, skb, dst_pid, msg->msg_flags&MSG_DONTWAIT);
out:
return err;
}
发 送是一个主动的过程，因此需要主动寻找目标sock，然后把根据要发送的消息初始化好的sk_buff找机会排入目标sock的接收队列就完事了，如此看 来的话，接收更是一个简单的过程了，它只要等着接收就可以了，纯粹就是一个被动的过程了，在对应的netlink_recvmsg中循环接收，没有数据时 就睡在一个sock->sk_sleep队列上就是了，一旦有数据过来，该接收进程就被唤醒，具体过程就是，当发送方调用 netlink_sendmsg时，后者调用netlink_unicast，然后它进一步调用netlink_sendskb，这个 netlink_sendskb最后将sk_buff排入目标接收sock的接收队列后调用目标sock的sk_data_ready，而这个 sk_data_ready对于用户空间的netlink就是sock_def_readable，它会唤醒睡眠了sk->sk_sleep上的接 收sock进程，而对于内核netlink套接字，其sk_data_ready将会是netlink_data_ready，就是上面所说的那个函数。 这个netlink_data_ready里面调用了一个程序设计者传入内核的data_ready回调函数从而可以实现用户策略，再次引入了机制和策略 的分离。在netlink_rcv_wake中会判断当前接收队列sk->sk_receive_queue是否已经空了，空了的话证明接收已经完 成，这种情况下就要唤醒等待排入队列新skb的发送进程，也就是调用：

if (!skb_queue_len(&sk->sk_receive_queue)) //是否接收已经完成
clear_bit(0, &nlk->state); //完成的话state清位
if (!test_bit(0, &nlk->state)) //如果清位说明接收完成，那么就唤醒等待发送的进程，这个接收进程可以继续接收了。 其实只有在溢出的情况下才会置位state
wake_up_interruptible(&nlk->wait);
以 上就是简要的netlink发送和接收过程，netlink之所以如此简洁和高效，靠的就是它的一个巧妙的数据组织形式，内核要在接收到用户 netlink套接字数据时快速确定策略和快速定位发送的目标都是问题，linux的netlink实现了一个表结构，其实是一个链结构，只要有 netlink形成，不管是内核的还是用户空间的，都要将netlink套接字本身和它的信息一并插入到这个链表结构中，然后在发送时定位目标的时候，只 要遍历这个表就可以了，插入代码如下：
static int netlink_insert(struct sock *sk, u32 pid)
{
int err = -EADDRINUSE;
struct sock *osk;
struct hlist_node *node;
netlink_table_grab();
sk_for_each(osk, node, &nl_table[sk->sk_protocol]) {
if (nlk_sk(osk)->pid == pid)
break;
}
if (!node) {
err = -EBUSY;
if (nlk_sk(sk)->pid == 0) {
nlk_sk(sk)->pid = pid;
sk_add_node(sk, &nl_table[sk->sk_protocol]);
err = 0;
}
}
netlink_table_ungrab();
return err;
}
相应的，查找代码如下：
static __inline__ struct sock *netlink_lookup(int protocol, u32 pid)
{
struct sock *sk;
struct hlist_node *node;
read_lock(&nl_table_lock);
sk_for_each(sk, node, &nl_table[protocol]) {
if (nlk_sk(sk)->pid == pid) {
sock_hold(sk);
goto found;
}
}
sk = NULL;
found:
read_unlock(&nl_table_lock);
return sk;
}

最 后看看netlink的广播是如何实现的，其实这里的广播并没有用什么网络协议的知识，和那也没有关系，所谓的广播其实就是将数据发给所有的可以接收的 netlink套接字而已。其实就是遍历这个nl_table表，然后抽出其中每一个netlink套接字，经检验可以发送时，就将数据发给它们，实现起 来很简单。

int netlink_broadcast(struct sock *ssk, struct sk_buff *skb, u32 pid, u32 group, int allocation)
{
struct sock *sk;
struct hlist_node *node;
struct sk_buff *skb2 = NULL;
int protocol = ssk->sk_protocol;
int failure = 0, delivered = 0;
netlink_trim(skb, allocation);
netlink_lock_table();
sk_for_each(sk, node, &nl_table[protocol]) { //遍历整个netlink套接字链表
struct netlink_opt *nlk = nlk_sk(sk);
if (ssk == sk) //不再向自己发送
continue;
...//简单判断并且递增该sock结构的引用计数，因为下面要用它了。
} else if (netlink_broadcast_deliver(sk, skb2)) {
...
}
static __inline__ int netlink_broadcast_deliver(struct sock *sk, struct sk_buff *skb)
{
struct netlink_opt *nlk = nlk_sk(sk);
if (atomic_read(&sk->sk_rmem_alloc) sk_rcvbuf && !test_bit(0, &nlk->state)) {
skb_orphan(skb);
skb_set_owner_r(skb, sk); //将skb直接过继给需要接收的sock结构
skb_queue_tail(&sk->sk_receive_queue, skb);
sk->sk_data_ready(sk, skb->len); //告知需要接收的sock结构有数据到来
return 0;
}
return -1;
}
其实，linux的很多机制都是netlink套接字实现的，比如udev守护进程，netlink机制很好很强大！