深入理解Netlink内核实现（第二部分）：通信机制解析
最编程
2024-07-24 12:32:08
...
int type：WRITE； size_t iov_offset：初始化为0； size_t count：所有iovec结构数据的总长度（即iov->iov_len的总和）； const struct iovec *iov：首个iov结构指针； unsigned long nr_segs：iovec结构的个数。
6.详解4 --- used_address && msg_sys->msg_name && used_address->name_len == msg_sys->msg_namelen && !memcmp(&used_address->name, msg_sys->msg_name, used_address->name_len)
根据传入的used_address指针判断当前发送消息的目的地址是否同它记录的一致，如果一致则调用sock_sendmsg_nosec()函数发送数据，否则调用sock_sendmsg()函数发送数据，sock_sendmsg()其实最终也是通过调用sock_sendmsg_nosec()来发送数据的，它们的区别就在于是否调用安全检查函数，如下：
int sock_sendmsg(struct socket *sock, struct msghdr *msg) { int err = security_socket_sendmsg(sock, msg, msg_data_left(msg)); return err ?: sock_sendmsg_nosec(sock, msg); ------ 详解8 }
7.详解5 --- used_address && err >= 0
在sendmmsg系统调用每一次发送多个消息时，由于发送的目的地一般都是一致的，所以只需要在发送第一个消息爆时执行检查就可以了，通过这种策略就可以加速数据的发送。最后，在发送完数据后，如果传入的used_address指针非空，就会将本次成功发送数据的目的地址记录下来，供下次发送数据比较。
8.详解8 --- sock_sendmsg_nosec
static inline int sock_sendmsg_nosec(struct socket *sock, struct msghdr *msg) { int ret = sock->ops->sendmsg(sock, msg, msg_data_left(msg)); BUG_ON(ret == -EIOCBQUEUED); return ret; }
这里调用了socket所绑定协议特有的数据发送钩子函数，其中最后一个参数为msg->msg_iter->count，即消息实际载荷的总长度。在前一篇文章中已经看到了对于netlink类型的套接字来说该函数被注册为netlink_sendmsg()。
static int netlink_sendmsg(struct socket *sock, struct msghdr *msg, size_t len) { struct sock *sk = sock->sk; struct netlink_sock *nlk = nlk_sk(sk); DECLARE_SOCKADDR(struct sockaddr_nl *, addr, msg->msg_name);
//定义了一个struct sockaddr_nl *addr指针，它指向了msg->msg_name表示消息的目的地址（会做地址长度检查） u32 dst_portid; u32 dst_group; struct sk_buff *skb; int err; struct scm_cookie scm; u32 netlink_skb_flags = 0; if (msg->msg_flags&MSG_OOB) return -EOPNOTSUPP; err = scm_send(sock, msg, &scm, true);//发送消息辅助数据 if (err < 0) return err; if (msg->msg_namelen) {--------------- 详解9 err = -EINVAL; if (addr->nl_family != AF_NETLINK) goto out; dst_portid = addr->nl_pid; dst_group = ffs(addr->nl_groups); err = -EPERM; if ((dst_group || dst_portid) && !netlink_allowed(sock, NL_CFG_F_NONROOT_SEND)) goto out; netlink_skb_flags |= NETLINK_SKB_DST; } else { dst_portid = nlk->dst_portid; dst_group = nlk->dst_group; } if (!nlk->bound) { --------------------- 详解10 err = netlink_autobind(sock); if (err) goto out; } else { /* Ensure nlk is hashed and visible. */ smp_rmb(); } err = -EMSGSIZE; if (len > sk->sk_sndbuf - 32)------------------- 详解11 goto out; err = -ENOBUFS; skb = netlink_alloc_large_skb(len, dst_group); if (skb == NULL) goto out; NETLINK_CB(skb).portid = nlk->portid; ------------------ 详解12 NETLINK_CB(skb).dst_group = dst_group; NETLINK_CB(skb).creds = scm.creds; NETLINK_CB(skb).flags = netlink_skb_flags; err = -EFAULT; if (memcpy_from_msg(skb_put(skb, len), msg, len)) { kfree_skb(skb); goto out; } err = security_netlink_send(sk, skb);----------------- 详解13 if (err) { kfree_skb(skb); goto out; } if (dst_group) { atomic_inc(&skb->users); netlink_broadcast(sk, skb, dst_portid, dst_group, GFP_KERNEL);//组播方式 --- 详解15 } err = netlink_unicast(sk, skb, dst_portid, msg->msg_flags&MSG_DONTWAIT);//单播方式 --- 详解14 out: scm_destroy(&scm); return err; }
9.详解9 --- msg->msg_namelen
这里如果用户指定了netlink消息的目的地址，则对其进行校验，然后判断当前netlink协议的NL_CFG_F_NONROOT_SEND标识是否设置，如果设置了该标识则允许非root用户发送组播，对于NETLINK_ROUTE类型的netlink套接字，并没有设置该标识，表明非root用户不能发送组播消息；然后设置NETLINK_SKB_DST标识。如果用户没有指定netlink消息的目的地址，则使用netlink套接字默认的（该值默认为0，会在调用connect系统调用时在netlink_connect()中被赋值为用户设置的值）。注意这里dst_group经过ffs的处理后转化为组播地址位数（找到最低有效位）。
10.详解10 --- !nlk->bound
接下来判断当前的netlink套接字是否被绑定过，如果没有绑定过这里调用netlink_autobind()进行动态绑定，该函数在前一篇文章中已经分析.继续往下分析
/* It's a really convoluted way for userland to ask for mmaped * sendmsg(), but that's what we've got... */ if (netlink_tx_is_mmaped(sk) && msg->msg_iter.type == ITER_IOVEC && msg->msg_iter.nr_segs == 1 && msg->msg_iter.iov->iov_base == NULL) { err = netlink_mmap_sendmsg(sk, msg, dst_portid, dst_group, &scm); goto out; }
如果内核配置了CONFIG_NETLINK_MMAP内核选项，则表示内核空间和应用层的消息发送队列支持内存映射，然后通过调用netlink_mmap_sendmsg来发送netlink消息，该种方式将减少数据的内存数据的拷贝动作，减少发送时间和资源占用。现我的环境中并不支持，继续往下分析：
11.详解11 --- if (len > sk->sk_sndbuf - 32)
接下来判断需要发送的数据是否过长（长于发送缓存大小），然后通过netlink_alloc_large_skb分配skb结构（传入的参数为消息载荷的长度以及组播地址）。
12.详解12 --- NETLINK_CB(skb).portid = nlk->portid;
在成功创建skb结构之后，这里就开始初始化它，这里使用到了skb中的扩展cb字段（char cb[48] __aligned(8)，一共48个字节用于存放netlink的地址和标识相关的附加信息足够了），同时使用宏NETLINK_CB来操作这些字段。netlink将skb的cb字段强制定义为struct netlink_skb_parms结构：
struct netlink_skb_parms { struct scm_creds creds; /* Skb credentials */ __u32 portid; __u32 dst_group; __u32 flags; struct sock *sk; };
其中portid表示原端套接字所绑定的id，dst_group表示消息目的组播地址，flag为标识，sk指向原端套接字的sock结构。
这里首先将套接字绑定的portid赋值到skb的cb字段中、同时设置组播地址的数量以及netlink_skb标识（这里是已经置位NETLINK_SKB_DST）。接下来调用最关键的调用memcpy_from_msg拷贝数据，它首先调用skb_put调整skb->tail指针，然后执行copy_from_iter(data, len, &msg->msg_iter)将数据从msg->msg_iter中传输到skb->data中（这是第一次内存拷贝动作！将用户空间数据直接拷贝到内核skb中）。
13.详解13 --- security_netlink_send
调用security_netlink_send()执行security检查.
最后如果是组播发送则调用netlink_broadcast()发送消息，否则调用netlink_unicast()发送单播消息。
14.详解14 --- netlink_unicast 发送单播消息
int netlink_unicast(struct sock *ssk, struct sk_buff *skb, u32 portid, int nonblock) { struct sock *sk; int err; long timeo; skb = netlink_trim(skb, gfp_any());---------- 详解16 timeo = sock_sndtimeo(ssk, nonblock); ---------------- 详解17 retry: sk = netlink_getsockbyportid(ssk, portid);--------------- 详解18 if (IS_ERR(sk)) { kfree_skb(skb); return PTR_ERR(sk); } if (netlink_is_kernel(sk))-------------- 详解19 return netlink_unicast_kernel(sk, skb, ssk); ------------ 详解20 if (sk_filter(sk, skb)) { err = skb->len; kfree_skb(skb); sock_put(sk); return err; } err = netlink_attachskb(sk, skb, &timeo, ssk); if (err == 1) goto retry; if (err) return err; return netlink_sendskb(sk, skb); }
15.详解16 --- netlink_trim
调用netlink_trim()重新裁剪skb的数据区的大小，这可能会clone出一个新的skb结构同时重新分配skb->data的内存空间（这就出现了第三次的内存拷贝动作！），当然如果原本skb中多余的内存数据区非常小或者该内存空间是在vmalloc空间中的就不会执行上述操作，我们现在跟随的情景上下文中就是后一种情况，并不会重新分配空间。
16.详解17 --- sock_sndtimeo
记下发送超时等待时间，如果已经设置了MSG_DONTWAIT标识，则等待时间为0，否则返回sk->sk_sndtimeo（该值在sock初始化时由sock_init_data()函数赋值为MAX_SCHEDULE_TIMEOUT）。
17.详解18 --- netlink_getsockbyportid
接下来调用netlink_getsockbyportid根据目的portid号和原端sock结构查找目的端的sock结构。其定义如下：
static struct sock *netlink_getsockbyportid(struct sock *ssk, u32 portid) { struct sock *sock; struct netlink_sock *nlk; /*调用netlink_lookup执行查找工作，查找的命名空间和协议号同原端sock，它会从nl_table[protocol]的哈希表中找到已经注册的目的端sock套接字。找到以后执行校验，如若找到的socket已经connect了，则它的目的portid必须是原端的portid*/ sock = netlink_lookup(sock_net(ssk), ssk->sk_protocol, portid); if (!sock) return ERR_PTR(-ECONNREFUSED); /* Don't bother queuing skb if kernel socket has no input function */ nlk = nlk_sk(sock); if (sock->sk_state == NETLINK_CONNECTED && nlk->dst_portid != nlk_sk(ssk)->portid) { sock_put(sock); return ERR_PTR(-ECONNREFUSED); } return sock; }
18.详解19 --- netlink_is_kernel(sk)
判断目的的netlink socket是否是内核的netlink socket，如果目的地址是内核空间，则调用netlink_unicast_kernel向内核进行单播，入参是目的sock、原端sock和数据skb。目前目的地址是内核。
函数netlink_is_kernel定义如下：
static inline int netlink_is_kernel(struct sock *sk) { return nlk_sk(sk)->flags & NETLINK_F_KERNEL_SOCKET; }
19.详解20 --- netlink_unicast_kernel
函数netlink_unicast_kernel定义如下：
static int netlink_unicast_kernel(struct sock *sk, struct sk_buff *skb, struct sock *ssk) { int ret; struct netlink_sock *nlk = nlk_sk(sk); ret = -ECONNREFUSED; if (nlk->netlink_rcv != NULL) { ret = skb->len; netlink_skb_set_owner_r(skb, sk); NETLINK_CB(skb).sk = ssk; netlink_deliver_tap_kernel(sk, ssk, skb); nlk->netlink_rcv(skb); consume_skb(skb); } else { kfree_skb(skb); } sock_put(sk); return ret; }
检查目标netlink套接字是否注册了netlink_rcv()接收函数，如果没有则直接丢弃该数据包，否则继续发送流程，这里首先设置一些标识：
skb->sk = sk; /* 将目的sock赋值给skb->sk指针 */ skb->destructor = netlink_skb_destructor; /* 注册destructor钩子函数 */ NETLINK_CB(skb).sk = ssk; /* 将原端的sock保存早skb的cb扩展字段中 */
struct sock * __netlink_kernel_create(struct net *net, int unit, struct module *module, struct netlink_kernel_cfg *cfg) { ...... if (cfg && cfg->input) nlk_sk(sk)->netlink_rcv = cfg->input; ... }
对于NETLINK_ROUTE类型的套接字来说就是rtnetlink_rcv了，netlink_rcv()钩子函数会接收并解析用户传下来的数据，不同类型的netlink协议各不相同，这里就不进行分析了。至此应用层下发单播的netlink数据就下发完成了。
20.详解15 --- netlink_broadcast() 发送组播消息
int netlink_broadcast(struct sock *ssk, struct sk_buff *skb, u32 portid, u32 group, gfp_t allocation) { return netlink_broadcast_filtered(ssk, skb, portid, group, allocation, NULL, NULL); }
函数netlink_broadcast_filtered定义如下:
int netlink_broadcast_filtered(struct sock *ssk, struct sk_buff *skb, u32 portid, u32 group, gfp_t allocation, int (*filter)(struct sock *dsk, struct sk_buff *skb, void *data), void *filter_data) { struct net *net = sock_net(ssk); struct netlink_broadcast_data info; struct sock *sk; skb = netlink_trim(skb, allocation); //初始化netlink组播数据结构netlink_broadcast_data info.exclude_sk = ssk; info.net = net; info.portid = portid; info.group = group;//保存了目的组播地址 info.failure = 0; info.delivery_failure = 0; info.congested = 0; info.delivered = 0; info.allocation = allocation; info.skb = skb; info.skb2 = NULL; info.tx_filter = filter; info.tx_data = filter_data; /* While we sleep in clone, do not allow to change socket list */ netlink_lock_table(); sk_for_each_bound(sk, &nl_table[ssk->sk_protocol].mc_list)//从nl_table[ssk->sk_protocol].mc_list里边查找加入组播组的socket do_one_broadcast(sk, &info);//依次发送组播数据 consume_skb(skb); netlink_unlock_table(); if (info.delivery_failure) { kfree_skb(info.skb2); return -ENOBUFS; } consume_skb(info.skb2); if (info.delivered) { if (info.congested && gfpflags_allow_blocking(allocation)) yield(); return 0; } return -ESRCH; }
这里首先初始化netlink组播数据结构netlink_broadcast_data，其中info.group中保存了目的组播地址，然后从nl_table[ssk->sk_protocol].mc_list里边查找加入组播组的socket，并调用do_one_broadcast()函数依次发送组播数据：
static void do_one_broadcast(struct sock *sk, struct netlink_broadcast_data *p) { struct netlink_sock *nlk = nlk_sk(sk); int val; //做必要的检查 if (p->exclude_sk == sk) return; ... val = netlink_broadcast_deliver(sk, p->skb2);//对目的sock发送数据skb ... }
当然，在发送之前会做一些必要的检查，例如这里会确保原端sock和目的端sock不是同一个，它们属于同一个网络命名空间，目的的组播地址为发送的目的组播地址等等，然后会对skb和组播数据结构netlink_broadcast_data进行一些处理，最后调用
netlink_broadcast_deliver()函数对目的sock发送数据skb：
static int netlink_broadcast_deliver(struct sock *sk, struct sk_buff *skb) { struct netlink_sock *nlk = nlk_sk(sk); if (atomic_read(&sk->sk_rmem_alloc) <= sk->sk_rcvbuf && !test_bit(NETLINK_S_CONGESTED, &nlk->state)) { netlink_skb_set_owner_r(skb, sk); __netlink_sendskb(sk, skb); return atomic_read(&sk->sk_rmem_alloc) > (sk->sk_rcvbuf >> 1); } return -1; } static int __netlink_sendskb(struct sock *sk, struct sk_buff *skb) { int len = skb->len; netlink_deliver_tap(skb); skb_queue_tail(&sk->sk_receive_queue, skb);//将要发送的skb添加到目的sock的接收队列末尾 sk->sk_data_ready(sk);//通知钩子函数，告知目的sock有数据到达，执行处理流程 ----- 补充1 return len; }
补充1：可以看到，这里将要发送的skb添加到目的sock的接收队列末尾，然后调用sk_data_ready()通知钩子函数，告知目的sock有数据到达，执行处理流程。对于内核的netlink来说内核netlink的创建函数中已经将其注册为：
struct sock * __netlink_kernel_create(struct net *net, int unit, struct module *module, struct netlink_kernel_cfg *cfg) { ...... sk->sk_data_ready = netlink_data_ready; ...... } static void netlink_data_ready(struct sock *sk) { BUG(); }
非常明显了，内核netlink套接字是无论如何也不应该接收到组播消息的。但是对于应用层netlink套接字，该sk_data_ready()钩子函数在初始化netlink函数sock_init_data()中被注册为sock_def_readable()，这个函数待分析。
三:内核接收应用层消息
当进程有数据发送过来时，内核部分会接收数据，上送的包是struct sk_buff *skb,我们可以通过netlink提供的一系列操作函数来获取消息头以及数据。
消息头 = nlmsg_hdr(skb)；
消息数据 = NLMSG_DATA(nlh)；
四:内核向应用层发送消息
(一)内核发送netlink单播消息
内核可以通过nlmsg_unicast()函数向应用层发送单播消息，由各个netlink协议负责调用，也有的协议是直接调用netlink_unicast()函数，其实nlmsg_unicast()也仅是netlink_unicast()的一个封装而已：
/** * nlmsg_unicast - unicast a netlink message * @sk: netlink socket to spread message to * @skb: netlink message as socket buffer * @portid: netlink portid of the destination socket */ static inline int nlmsg_unicast(struct sock *sk, struct sk_buff *skb, u32 portid) { int err; err = netlink_unicast(sk, skb, portid, MSG_DONTWAIT); if (err > 0) err = 0; return err; }
这里以非阻塞（MSG_DONTWAIT）的形式向应用层发送消息，这时的portid为应用层套接字所绑定的id号。我们再次进入到netlink_unicast()内部，这次由于目的sock不再是内核，所以要走不同的的分支了
int netlink_unicast(struct sock *ssk, struct sk_buff *skb, u32 portid, int nonblock) { struct sock *sk; int err; long timeo; skb = netlink_trim(skb, gfp_any()); timeo = sock_sndtimeo(ssk, nonblock); retry: sk = netlink_getsockbyportid(ssk, portid); if (IS_ERR(sk)) { kfree_skb(skb); return PTR_ERR(sk); } if (netlink_is_kernel(sk)) return netlink_unicast_kernel(sk, skb, ssk);//应用层向内核发送消息 /*以下为内核向应用层发送消息的flow*/ if (sk_filter(sk, skb)) {//首先sk_filter执行防火墙的过滤，确保可以发送以后调用netlink_attachskb将要发送的skb绑定到netlink sock上 err = skb->len; kfree_skb(skb); sock_put(sk); return err; } err = netlink_attachskb(sk, skb, &timeo, ssk);---------------- 详解1 if (err == 1)//若执行netlink_attachskb()的返回值为1，就会再次尝试发送操作 goto retry; if (err) return err; return netlink_sendskb(sk, skb); ------------------- 详解2 }
1.详解1 --- netlink_attachskb
这里首先sk_filter执行防火墙的过滤，确保可以发送以后,调用netlink_attachskb将要发送的skb绑定到netlink sock上。
如果目的sock的接收缓冲区剩余的的缓存大小小于已经提交的数据量，或者标志位已经置位了阻塞标识NETLINK_CONGESTED，这表明数据不可以立即的送到目的端的接收缓存中。
因此，在原端不是内核socket且没有设置非阻塞标识的情况下会定义一个等待队列并等待指定的时间并返回1，否则直接丢弃该skb数据包并返回失败。
int netlink_attachskb(struct sock *sk, struct sk_buff *skb, long *timeo, struct sock *ssk) { struct netlink_sock *nlk; nlk = nlk_sk(sk); if ((atomic_read(&sk->sk_rmem_alloc) > sk->sk_rcvbuf || test_bit(NETLINK_S_CONGESTED, &nlk->state))) { DECLARE_WAITQUEUE(wait, current); if (!*timeo) { if (!ssk || netlink_is_kernel(ssk)) netlink_overrun(sk); sock_put(sk); kfree_skb(skb); return -EAGAIN; } __set_current_state(TASK_INTERRUPTIBLE); add_wait_queue(&nlk->wait, &wait); if ((atomic_read(&sk->sk_rmem_alloc) > sk->sk_rcvbuf || test_bit(NETLINK_S_CONGESTED, &nlk->state)) && !sock_flag(sk, SOCK_DEAD)) *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; } netlink_skb_set_owner_r(skb, sk);//目的端的接收缓存区空间足够，就会调用netlink_skb_set_owner_r进行绑定 return 0; }
2.详解2 --- netlink_sendskb
调用netlink_sendskb()执行发送操作
int netlink_sendskb(struct sock *sk, struct sk_buff *skb) { int len = __netlink_sendskb(sk, skb); sock_put(sk); return len; } /*这里又一次回到了__netlink_sendskb函数执行发送流程*/ static int __netlink_sendskb(struct sock *sk, struct sk_buff *skb) { int len = skb->len; netlink_deliver_tap(skb); skb_queue_tail(&sk->sk_receive_queue, skb); sk->sk_data_ready(sk); return len; }
这里的sk_data_ready()钩子函数在初始化netlink函数sock_init_data()中被注册为sock_def_readable()：
static void sock_def_readable(struct sock *sk) { struct socket_wq *wq; rcu_read_lock(); wq = rcu_dereference(sk->sk_wq); if (wq_has_sleeper(wq)) wake_up_interruptible_sync_poll(&wq->wait, POLLIN | POLLPRI | POLLRDNORM | POLLRDBAND); sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);//唤醒目的接收端socket的等待队列，这样应用层套接字就可以接收并处理消息了 rcu_read_unlock(); }
(二)内核发送netlink组播消息
内核发送多播消息是通过函数nlmsg_multicast()，详细分析见上文，不再重复。
static inline int nlmsg_multicast(struct sock *sk, struct sk_buff *skb, u32 portid, unsigned int group, gfp_t flags) { int err; NETLINK_CB(skb).dst_group = group; err = netlink_broadcast(sk, skb, portid, group, flags); if (err > 0) err = 0; return err; }
五:应用层接收内核的消息
使用如下示例程序可以以阻塞的方式接收内核发送的netlink消息：
#define TEST_DATA_LEN 16 struct sockaddr_nl nladdr; struct msghdr msg; struct nlmsghdr *nlhdr; struct iovec iov; /* 清空源地址结构 */ memset(&nladdr, 0, sizeof(nladdr)); /* 清空netlink消息头 */ nlhdr = (struct nlmsghdr *)malloc(NLMSG_SPACE(TEST_DATA_LEN)); memset(nlhdr, 0, NLMSG_SPACE(TEST_DATA_LEN)); /* 封装netlink消息 */ iov.iov_base = (void *)nlhdr; /* 接收缓存地址 */ iov.iov_len = NLMSG_LENGTH(TEST_DATA_LEN);; /* 接收缓存大小 */ /* 填充数据消息结构 */ memset(&msg, 0, sizeof(msg)); msg.msg_name = (void *)&(nladdr); msg.msg_namelen = sizeof(nladdr); /* 地址长度由内核赋值 */ msg.msg_iov = &iov; msg.msg_iovlen = 1; /* 接收netlink消息 */ recvmsg(sock_fd, &msg, 0);
本示例程序同前文中的发送程序类似，需要有接收端组装接收msg消息。同发送流程的不同之处在于：
（1）msg.msg_name地址结构中存放的是消息源的地址信息，由内核负责填充。
（2）iov.iov_base为接收缓存的地址空间，其需要在接收前清空。
（3）iov.iov_len为单个iov接收缓存的长度，需要指明。
（4）msg.msg_namelen：为地址占用长度，有内核负责填充。
（5）msg.msg_iovlen:为接收iov空间的个数，需要指明。
这里用到了recvmsg系统调用，现进入该系统调用分析消息的整个接收的过程（需要注意的是，在不使用NETLINK_MMAP技术的情况下，整个接收的过程中存在1次数据的内存拷贝动作！）：
应用层通过API recvmsg接收内核的消息，其对应的系统调用如下：
SYSCALL_DEFINE3(recvmsg, int, fd, struct user_msghdr __user *, msg, unsigned int, flags) { if (flags & MSG_CMSG_COMPAT) return -EINVAL; return __sys_recvmsg(fd, msg, flags); } long __sys_recvmsg(int fd, struct user_msghdr __user *msg, unsigned flags) { int fput_needed, err; struct msghdr msg_sys; struct socket *sock; sock = sockfd_lookup_light(fd, &err, &fput_needed);//也是通过fd描述符查找对应的套接字socket结构 if (!sock) goto out; err = ___sys_recvmsg(sock, msg, &msg_sys, flags, 0); fput_light(sock->file, fput_needed); out: return err; }
同sendmsg系统调用类似，这里也同样首先通过fd描述符查找对应的套接字socket结构，然后调用___sys_recvmsg()执行实际的工作
static int ___sys_recvmsg(struct socket *sock, struct user_msghdr __user *msg, struct msghdr *msg_sys, unsigned int flags, int nosec) { struct compat_msghdr __user *msg_compat = (struct compat_msghdr __user *)msg; struct iovec iovstack[UIO_FASTIOV]; struct iovec *iov = iovstack;//定义了一个大小为8的iovstack数组缓存，用来加速消息处理 unsigned long cmsg_ptr; int len; ssize_t err; /* kernel mode address */ struct sockaddr_storage addr; /* user mode address pointers */ struct sockaddr __user *uaddr; int __user *uaddr_len = COMPAT_NAMELEN(msg);//获取用户空间的地址长度字段的地址 msg_sys->msg_name = &addr; if (MSG_CMSG_COMPAT & flags) err = get_compat_msghdr(msg_sys, msg_compat, &uaddr, &iov); else err = copy_msghdr_from_user(msg_sys, msg, &uaddr, &iov);//拷贝用户态msg中的数据到内核态msg_sys中 ------------- 详解1 if (err < 0) return err; cmsg_ptr = (unsigned long)msg_sys->msg_control; msg_sys->msg_flags = flags & (MSG_CMSG_CLOEXEC|MSG_CMSG_COMPAT); /* We assume all kernel code knows the size of sockaddr_storage */ msg_sys->msg_namelen = 0;//将地址的长度字段清零 /*根据nosec的值是否为0而调用sock_recvmsg_nosec()或sock_recvmsg()函数接收数据，nosec在recvmsg系统调用传入的为0，在recvmmsg系统能够调用接收多个消息时传入已经接受的消息个数*/ if (sock->file->f_flags & O_NONBLOCK) flags |= MSG_DO
推荐阅读
-
深入理解Linux内核调试技术——kprobe的使用和实现（第二部分）
-
深入理解Linux Netlink通信机制（第二部分）
-
【2022新手指南】Java编程进阶之路 - 六、技术架构篇 ### MySQL索引底层解析与优化实战 - 你会讲解MySQL索引的数据结构吗？性能调优技巧知多少？ - Redis深度揭秘：你知道多少？从基础到哨兵、主从复制全梳理 - Redis持久化及哨兵模式详解，还有集群搭建和Leader选举黑箱打开 - Zookeeper是个啥？特性和应用场景大公开 - ZooKeeper集群搭建攻略及 Leader选举、读写一致性、共享锁实现细节 - 探究ZooKeeper中的Leader选举机制及其在分布式环境中的作用 - Zab协议深入剖析：原理、功能与在Zookeeper中的核心地位 - RabbitMQ全方位解读：工作模式、消费限流、可靠投递与配置策略 - 设计者视角：RabbitMQ过期时间、死信队列与延时队列实践指南 - RocketMQ特性和应用场景揭示：理解其精髓与差异化优势 - Kafka详细介绍：特性及广泛应用于实时数据处理的场景解析 - ElasticSearch实力揭秘：特性概述与作为搜索引擎的广泛应用 - MongoDB认知升级：非关系型数据库的优势阐述，安装与使用实战教学 - BIO/NIO/AIO网络模型对比：掌握它们的区别与在网络编程中的实际应用 - Netty带你飞：理解其超快速度背后的秘密，包括线程模型分析 - 网络通信黑科技：Netty编解码原理与常用编解码器的应用，Protostuff实战演示 - 解密Netty粘包与拆包现象，怎样有效应对这一常见问题 - 自定义Netty心跳检测机制，轻松调整检测间隔时间的艺术 - Dubbo轻骑兵介绍：核心特性概览，服务降级实战与其实现益处 - Dubbo三大神器解读：本地存根与本地伪装的实战运用与优势呈现 ----------------------- 七、结语与回顾
-
深入理解Netlink内核实现（第二部分）：通信机制解析
-
鸿蒙v78.01内核源码详解：消息映射深度解析 - LiteIpc进程通信机制第二部分 | 一系列百篇博文中揭秘OpenHarmony源代码探索
-
深入解析ODL技术的秘密：第二部分——详细解读项目管理设计与实现机制
-
深入解析Generic Netlink内核实现（二）：通信的详细分析
-
【Netty】「萌新入门」（七）ByteBuf 的性能优化-堆内存的分配和释放都是由 Java 虚拟机自动管理的，这意味着它们可以快速地被分配和释放，但是也会产生一些开销。 直接内存需要手动分配和释放，因为它由操作系统管理，这使得分配和释放的速度更快，但是也需要更多的系统资源。 另外，直接内存可以映射到本地文件中，这对于需要频繁读写文件的应用程序非常有用。 此外，直接内存还可以避免在使用 NIO 进行网络传输时发生数据拷贝的情况。在使用传统的 I/O 时，数据必须先从文件或网络中读取到堆内存中，然后再从堆内存中复制到直接缓冲区中，最后再通过 SocketChannel 发送到网络中。而使用直接缓冲区时，数据可以直接从文件或网络中读取到直接缓冲区中，并且可以直接从直接缓冲区中发送到网络中，避免了不必要的数据拷贝和内存分配。 通过 ByteBufAllocator.DEFAULT.directBuffer 方法来创建基于直接内存的 ByteBuf： ByteBuf directBuf = ByteBufAllocator.DEFAULT.directBuffer(16); 通过 ByteBufAllocator.DEFAULT.heapBuffer 方法来创建基于堆内存的 ByteBuf： ByteBuf heapBuf = ByteBufAllocator.DEFAULT.heapBuffer(16); 注意： 直接内存是一种特殊的内存分配方式，可以通过在堆外申请内存来避免 JVM 堆内存的限制，从而提高读写性能和降低 GC 压力。但是，直接内存的创建和销毁代价昂贵，因此需要慎重使用。 此外，由于直接内存不受 JVM 垃圾回收的管理，我们需要主动释放这部分内存，否则会造成内存泄漏。通常情况下，可以使用 ByteBuffer.clear 方法来释放直接内存中的数据，或者使用 ByteBuffer.cleaner 方法来手动释放直接内存空间。 测试代码： public static void testCreateByteBuf { ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(16); System.out.println(buf.getClass); ByteBuf heapBuf = ByteBufAllocator.DEFAULT.heapBuffer(16); System.out.println(heapBuf.getClass); ByteBuf directBuf = ByteBufAllocator.DEFAULT.directBuffer(16); System.out.println(directBuf.getClass); } 运行结果： class io.netty.buffer.PooledUnsafeDirectByteBuf class io.netty.buffer.PooledUnsafeHeapByteBuf class io.netty.buffer.PooledUnsafeDirectByteBuf 池化技术 在 Netty 中，池化技术指的是通过对象池来重用已经创建的对象，从而避免了频繁地创建和销毁对象，这种技术可以提高系统的性能和可伸缩性。 通过设置 VM options，来决定池化功能是否开启： -Dio.netty.allocator.type={unpooled|pooled} 在 Netty 4.1 版本以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现； 这里我们使用非池化功能进行测试，依旧使用的是上面的测试代码 testCreateByteBuf，运行结果如下所示： class io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeDirectByteBuf class io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf class io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeDirectByteBuf 可以看到，ByteBuf 类由 PooledUnsafeDirectByteBuf 变成了 UnpooledUnsafeDirectByteBuf； 在没有池化的情况下，每次使用都需要创建新的 ByteBuf 实例，这个操作会涉及到内存的分配和初始化，如果是直接内存则代价更为昂贵，而且频繁的内存分配也可能导致内存碎片问题，增加 GC 压力。 使用池化技术可以避免频繁内存分配带来的开销，并且重用池中的 ByteBuf 实例，减少了内存占用和内存碎片问题。另外，池化技术还可以采用类似 jemalloc 的内存分配算法，进一步提升分配效率。 在高并发环境下，池化技术的优点更加明显，因为内存的分配和释放都是比较耗时的操作，频繁的内存分配和释放会导致系统性能下降，甚至可能出现内存溢出的风险。使用池化技术可以将内存分配和释放的操作集中到预先分配的池中，从而有效地降低系统的内存开销和风险。 内存释放 当在 Netty 中使用 ByteBuf 来处理数据时，需要特别注意内存回收问题。 Netty 提供了不同类型的 ByteBuf 实现，包括堆内存（JVM 内存）实现 UnpooledHeapByteBuf 和堆外内存（直接内存）实现 UnpooledDirectByteBuf，以及池化技术实现的 PooledByteBuf 及其子类。 UnpooledHeapByteBuf：通过 Java 的垃圾回收机制来自动回收内存； UnpooledDirectByteBuf：由于 JVM 的垃圾回收机制无法管理这些内存，因此需要手动调用 release 方法来释放内存； PooledByteBuf：使用了池化机制，需要更复杂的规则来回收内存; 由于池化技术的特殊性质，释放 PooledByteBuf 对象所使用的内存并不是立即被回收的，而是被放入一个内存池中，待下次分配内存时再次使用。因此，释放 PooledByteBuf 对象的内存可能会延迟到后续的某个时间点。为了避免内存泄漏和占用过多内存，我们需要根据实际情况来设置池化技术的相关参数，以便及时回收内存； Netty 采用了引用计数法来控制 ByteBuf 对象的内存回收，在博文 「源码解析」ByteBuf 的引用计数机制 中将会通过解读源码的形式对 ByteBuf 的引用计数法进行深入理解； 每个 ByteBuf 对象被创建时，都会初始化为1，表示该对象的初始计数为1。 在使用 ByteBuf 对象过程中，如果当前 handler 已经使用完该对象，需要通过调用 release 方法将计数减1，当计数为0时，底层内存会被回收，该对象也就被销毁了。此时即使 ByteBuf 对象还在，其各个方法均无法正常使用。 但是，如果当前 handler 还需要继续使用该对象，可以通过调用 retain 方法将计数加1，这样即使其他 handler 已经调用了 release 方法，该对象的内存仍然不会被回收。这种机制可以有效地避免了内存泄漏和意外访问已经释放的内存的情况。 一般来说，应该尽可能地保证 retain 和 release 方法成对出现，以确保计数正确。