从内核看SO_REUSEPORT的实现（基于5.9.9）
从内核看SO_REUSEPORT的实现（基于5.9.9）
前言：SO_REUSEPORT是提高服务器性能的一个特性，从Linux3.9后支持，本文从内核5.9.9的源码分析SO_REUSEPORT的实现，因为内核源码非常复杂，尽量把自己的思路说一下。大家有兴趣的可以自己研究。
首先我们来看看SO_REUSEPORT是什么
SO_REUSEPORT (since Linux 3.9)
Permits multiple AF_INET or AF_INET6 sockets to be bound
to an identical socket address. This option must be set
on each socket (including the first socket) prior to
calling bind(2) on the socket. To prevent port hijacking,
all of the processes binding to the same address must have
the same effective UID. This option can be employed with
both TCP and UDP sockets.
For TCP sockets, this option allows accept(2) load
distribution in a multi-threaded server to be improved by
using a distinct listener socket for each thread. This
provides improved load distribution as compared to
traditional techniques such using a single accept(2)ing
thread that distributes connections, or having multiple
threads that compete to accept(2) from the same socket.
For UDP sockets, the use of this option can provide better
distribution of incoming datagrams to multiple processes
(or threads) as compared to the traditional technique of
having multiple processes compete to receive datagrams on
the same socket.SO_REUSEPORT主要是支持同用户下的多个进程同时绑定同一个ip和端口，他的作用主要分为两部分。
1 UDP
单播
在UDP中，单播的情况下，如果多个进程绑定同一个ip和端口，则只会有一个进程收到请求，具体哪个进程不同的操作系统实现不一样。我们写个测试例子。新建两个js用作服务器，代码如下。
const dgram = require('dgram');const udp = dgram.createSocket({type: 'udp4', reuseAddr: true});const socket = udp.bind(5678);socket.on('message', (msg) => {
console.log(msg)});同时执行这两个js，则有两个进程同时绑定到同一个ip和端口中。然后新建一个js用作客户端。
const dgram = require('dgram');const udp = dgram.createSocket({type: 'udp4'});const socket = udp.bind(1234);udp.send('hi', 5678)；执行以上代码，首先执行客户端，再执行服务器，我们会发现只有一个进程会收到数据。下面我们分析具体的原因。我们看一下UDP中执行bind时的逻辑。
int inet_bind(struct socket *sock, struct sockaddr *uaddr, int addr_len)
{
if (sk->sk_prot->get_port(sk, snum)) {
inet->saddr = inet->rcv_saddr = 0;
err = -EADDRINUSE;
goto out_release_sock;
}
}每个协议都可以实现自己的get_port钩子函数。用来判断当前的端口是否允许被绑定。如果不允许则返回EADDRINUSE，我们看看UDP协议的实现。
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);
// 通过端口找到对应的链表，然后遍历链表
sk_for_each(sk2, node, &udp_hash[snum & (UDP_HTABLE_SIZE - 1)]) {
struct inet_sock *inet2 = inet_sk(sk2);
// 端口已使用，则判断是否可以复用
if (inet2->num == snum &&
sk2 != sk &&
(!inet2->rcv_saddr ||
!inet->rcv_saddr ||
inet2->rcv_saddr == inet->rcv_saddr) &&
// 每个socket都需要设置端口复用标记
(!sk2->sk_reuse || !sk->sk_reuse))
// 不可以复用，报错
goto fail;
}
// 可以复用
inet->num = snum;
if (sk_unhashed(sk)) {
// 找到端口对应的位置
struct hlist_head *h = &udp_hash[snum & (UDP_HTABLE_SIZE - 1)];
// 插入链表
sk_add_node(sk, h);
sock_prot_inc_use(sk->sk_prot);
}
return 0;
fail:
write_unlock_bh(&udp_hash_lock);
return 1;
}UDP协议的实现中，会使用udp_hash记录每一个UDP socket，udp_hash是一个用数组实现的哈希表，每次bind socket的时候，首先会根据socket的源端口和哈希算法计算得到一个数组索引，然后把socket插入索引锁对应的链表中，即哈希冲突的解决方法是链地址法。回到代码的逻辑，当用户想绑定一个端口的时候，操作系统会根据端口拿到对应的socket链表，然后逐个判断是否有相等的端口，如果有则判断是否可以复用。例如两个socket都设置了复用标记则可以复用。最后把socket插入到链表中。
static inline void hlist_add_head(struct hlist_node *n, struct hlist_head *h)
{
// 头结点
struct hlist_node *first = h->first;
n->next = first;
if (first)
first->pprev = &n->next;
h->first = n;
n->pprev = &h->first;
}我们看到操作系统是以头插法的方式插入新节点的。接着我们看一下操作系统是如何使用这些数据结构的。下面是操作系统收到一个UDP数据包时的逻辑。
int udp_rcv(struct sk_buff *skb)
{
struct sock *sk;
struct udphdr *uh;
unsigned short ulen;
struct rtable *rt = (struct rtable*)skb->dst;
// ip头中记录的源ip和目的ip
u32 saddr = skb->nh.iph->saddr;
u32 daddr = skb->nh.iph->daddr;
int len = skb->len;
// udp协议头结构体
uh = skb->h.uh;
ulen = ntohs(uh->len);
// 广播或多播包
if(rt->rt_flags & (RTCF_BROADCAST|RTCF_MULTICAST))
return udp_v4_mcast_deliver(skb, uh, saddr, daddr);
// 单播
sk = udp_v4_lookup(saddr, uh->source, daddr, uh->dest, skb->dev->ifindex);
// 找到对应的socket
if (sk != NULL) {
// 把数据插到socket的消息队列
int ret = udp_queue_rcv_skb(sk, skb);
sock_put(sk);
if (ret > 0)
return -ret;
return 0;
}
return(0);
}单播时，收到UDP数据包会调用udp_v4_lookup函数找到接收该UDP数据包的socket，然后把数据包挂载到socket的接收队列中。我们看看udp_v4_lookup。
static __inline__ struct sock *udp_v4_lookup(u32 saddr, u16 sport,
u32 daddr, u16 dport, int dif)
{
struct sock *sk;
sk = udp_v4_lookup_longway(saddr, sport, daddr, dport, dif);
return sk;
}
static struct sock *udp_v4_lookup_longway(u32 saddr, u16 sport,
u32 daddr, u16 dport, int dif)
{
struct sock *sk, *result = NULL;
struct hlist_node *node;
unsigned short hnum = ntohs(dport);
int badness = -1;
// 遍历端口对应的链表
sk_for_each(sk, node, &udp_hash[hnum & (UDP_HTABLE_SIZE - 1)]) {
struct inet_sock *inet = inet_sk(sk);
if (inet->num == hnum && !ipv6_only_sock(sk)) {
int score = (sk->sk_family == PF_INET ? 1 : 0);
if (inet->rcv_saddr) {
if (inet->rcv_saddr != daddr)
continue;
score+=2;
}
if (inet->daddr) {
if (inet->daddr != saddr)
continue;
score+=2;
}
if (inet->dport) {
if (inet->dport != sport)
continue;
score+=2;
}
if (sk->sk_bound_dev_if) {
if (sk->sk_bound_dev_if != dif)
continue;
score+=2;
}
// 全匹配，直接返回，否则记录当前最好的匹配结果
if(score == 9) {
result = sk;
break;
} else if(score > badness) {
result = sk;
badness = score;
}
}
}
return result;
}我们看到代码很多，但是逻辑并不复杂，操作系统收到根据端口从哈希表中拿到对应的链表，然后遍历该链表找出最匹配的socket。然后把数据挂载到socket上。从Linux源码我们看到，插入socket的时候是使用头插法，查找的时候是从头开始找最匹配的socket。即后面插入的socket会先被搜索到。但是Windows下结构却相反，先监听了该IP端口的进程会收到数据。
多播
多播的情况下，多个绑定同一个ip和端口的进程，同一个请求，每个进程都会收到。我们写一个测试例子。我们在同主机上新建两个JS文件当作服务器，代码如下
const dgram = require('dgram');
const udp = dgram.createSocket({type: 'udp4', reuseAddr: true});
udp.bind(1234, ‘192.168.8.164‘, () => {
udp.addMembership('224.0.0.114', '192.168.8.164');
});
udp.on('message', (msg) => {
console.log(msg)
});上面代码使得两个进程都监听了同样的IP和端口。接下来我们写一个UDP客户端。
const dgram = require('dgram');
const udp = dgram.createSocket({type: 'udp4'});
const socket = udp.bind(5678);
socket.send('hi', 1234, '224.0.0.114', (err) => {
console.log(err)
});上面的代码给一个多播组发送了一个数据，执行上面的代码，我们可以看到两个服务器进程都收到了数据。我们看一下收到数据时，操作系统是如何把数据分发给每个监听了同样IP和端口的进程的。我们看一下udp_v4_mcast_deliver的实现。
static int udp_v4_mcast_deliver(struct sk_buff *skb, struct udphdr *uh,
u32 saddr, u32 daddr)
{
struct sock *sk;
int dif;
read_lock(&udp_hash_lock);
// 通过端口找到对应的链表
sk = sk_head(&udp_hash[ntohs(uh->dest) & (UDP_HTABLE_SIZE - 1)]);
dif = skb->dev->ifindex;
sk = udp_v4_mcast_next(sk, uh->dest, daddr, uh->source, saddr, dif);
if (sk) {
struct sock *sknext = NULL;
// 遍历每一个需要处理该数据包的socket
do {
struct sk_buff *skb1 = skb;
sknext = udp_v4_mcast_next(sk_next(sk),
uh->dest, daddr,
uh->source,
saddr,
dif);
if(sknext)
// 复制一份
skb1 = skb_clone(skb, GFP_ATOMIC);
// 插入每一个socket的数据包队列
if(skb1) {
int ret = udp_queue_rcv_skb(sk, skb1);
if (ret > 0)
kfree_skb(skb1);
}
sk = sknext;
} while(sknext);
} else
kfree_skb(skb);
read_unlock(&udp_hash_lock);
return 0;
}在多播的情况下，操作系统会遍历链表找到每一个可以接收该数据包的socket，然后把数据包复制一份，挂载到socket的接收队列。这就解释了测试的例子，即两个服务器进程都会收到UDP数据包。
2 TCP
而在TCP中意义就不太一样了。首先TCP没有多播的概念，所以我们只需要看单播，单播的情况下，多个进程绑定同一个ip和端口，在操作系统中会对应多个socket结构体。操作系统会负责负载均衡地分发请求。
而在没有SO_REUSEPORT之前，可以通过传递文件描述符或者fork的方式实现多个进程绑定到同一个ip和端口，但是有些不完美的地方，旧版的Linux内核会引起惊群现象，即一个请求会唤醒多个阻塞的进程，但是只有一个进程accept该请求进行处理，而其他进程被无效唤醒。新版内核则解决了这个问题，但是在依然存在其他问题，因为虽然多个进程绑定到了同一个ip和端口，但是底层对应的是一个socket结构体，即只有一个连接队列，当一个进程被唤醒的时候，他是accept全部请求呢，还是只accept一个，这会导致编程的复杂性，同时唤醒哪个进程也取决于操作系统的实现。另外一种处理方式是主进程accept，然后分发连接给子进程处理，这种方式下，只有一个进程可以accept，从进程调度角度来看，会导致进程被选中的概率变小，从而导致连接处理变慢。而SO_REUSEPORT中，多个进程可以同时accept，这意味着可处理连接的进程数变多，处理连接的速度变快。下面是不支持SO_REUSEPORT时的结构图。
最后我们从内核bind函数开始分析一下TCP中SO_REUSEPORT的实现。
if (sk->sk_prot->get_port(sk, snum)) {
inet->inet_saddr = inet->inet_rcv_saddr = 0;
err = -EADDRINUSE;
goto out_release_sock;}在get_port函数中内核会判断端口的合法性，get_port的实现取决于具体协议的实现，我们这里是TCP，函数是inet_csk_get_port。我们看当第一次bind这个socket的时候的逻辑
// 根据port从哈希表找到对应的列表（哈希表的每一项指向一个链表），hinfo是协议级别的数据结构
head = &hinfo->bhash[inet_bhashfn(net, port, hinfo->bhash_size)];
spin_lock_bh(&head->lock);
// 遍历这个链表，判断端口是否相等
inet_bind_bucket_for_each(tb, &head->chain)
if (net_eq(ib_net(tb), net) && tb->l3mdev == l3mdev &&
tb->port == port)
goto tb_found;
srtuct inet_bind_bucket *tb = inet_bind_bucket_create(hinfo->bind_bucket_cachep,
net, head, port, l3mdev);我们看到第一次bind的时候，链表是空的，然后调用inet_bind_bucket_create创建一个inet_bind_bucket结构体。
struct inet_bind_bucket *inet_bind_bucket_create(struct kmem_cache *cachep,
struct net *net,
struct inet_bind_hashbucket *head,
const unsigned short snum,
int l3mdev){
struct inet_bind_bucket *tb = kmem_cache_alloc(cachep, GFP_ATOMIC);
if (tb) {
write_pnet(&tb->ib_net, net);
tb->l3mdev = l3mdev;
// 记录端口
tb->port = snum;
tb->fastreuse = 0;
tb->fastreuseport = 0;
INIT_HLIST_HEAD(&tb->owners);
hlist_add_head(&tb->node, &head->chain);
}
return tb;}我们继续看创建完后的逻辑。
if (!inet_csk(sk)->icsk_bind_hash)
inet_bind_hash(sk, tb, port);会调用inet_bind_hash
void inet_bind_hash(struct sock *sk, struct inet_bind_bucket *tb,
const unsigned short snum){
inet_sk(sk)->inet_num = snum;
sk_add_bind_node(sk, &tb->owners);
inet_csk(sk)->icsk_bind_hash = tb;}把socket和inet_bind_bucket互相关联起来。下面我们来看第二次bind时的逻辑。
head = &hinfo->bhash[inet_bhashfn(net, port,
hinfo->bhash_size)];
spin_lock_bh(&head->lock);
inet_bind_bucket_for_each(tb, &head->chain)
if (net_eq(ib_net(tb), net) && tb->l3mdev == l3mdev &&
tb->port == port)
goto tb_found;
tb_found:
if (!hlist_empty(&tb->owners)) {
if (inet_csk_bind_conflict(sk, tb, true, true))
goto fail_unlock;
}if (!inet_csk(sk)->icsk_bind_hash)
inet_bind_hash(sk, tb, port);我们看到这时候从链表中可以找到对应的inet_bind_bucket tb了，并且tb->owners非空，因为第一次bind的时候往链表里插入了一个节点，接着调用inet_csk_bind_conflict判断端口是否冲突，判断逻辑非常复杂就不具体展开，比如判断每个socekt是否都设置SO_REUSEPORT，uid是否一样等等，代码大致如下。
// 遍历链表里的每个socketsk_for_each_bound(sk2, &tb->owners) { // 判断 }判断没问题之后，内核同样会把该socket插入到inet_bind_bucket维护的链表中。至此，bind的过程就分析完成了。接着我们分析调用listen函数时内核的处理。我们从inet_listen函数开始分析。
int inet_listen(struct socket *sock, int backlog){
struct sock *sk = sock->sk;
int err;
err = -EINVAL;
// 更新backlog字段
WRITE_ONCE(sk->sk_max_ack_backlog, backlog);
err = inet_csk_listen_start(sk, backlog);
if (err)
goto out;}接着看inet_csk_listen_start。
int inet_csk_listen_start(struct sock *sk, int backlog){
struct inet_connection_sock *icsk = inet_csk(sk);
struct inet_sock *inet = inet_sk(sk);
int err = -EADDRINUSE;
reqsk_queue_alloc(&icsk->icsk_accept_queue);
// 修改socket状态为TCP_LISTEN
inet_sk_state_store(sk, TCP_LISTEN);
// 再次校验端口，返回0说明ok。
if (!sk->sk_prot->get_port(sk, inet->inet_num)) {
inet->inet_sport = htons(inet->inet_num);
sk_dst_reset(sk);
// 调用钩子函数
err = sk->sk_prot->hash(sk);
if (likely(!err))
return 0;
}
inet_sk_set_state(sk, TCP_CLOSE);
return err;}inet_csk_listen_start的逻辑比较简单，主要是修改socekt的状态并且再次校验端口，然后调用钩子函数hash。对应TCP协议函数是inet_hash。
int inet_hash(struct sock *sk){
int err = 0;
if (sk->sk_state != TCP_CLOSE) {
local_bh_disable();
err = __inet_hash(sk, NULL);
local_bh_enable();
}
return err;}接着看__inet_hash。
int __inet_hash(struct sock *sk, struct sock *osk){
struct inet_hashinfo *hashinfo = sk->sk_prot->h.hashinfo;
struct inet_listen_hashbucket *ilb;
int err = 0;
ilb = &hashinfo->listening_hash[inet_sk_listen_hashfn(sk)];
// 设置了SO_REUSEPORT
if (sk->sk_reuseport) {
err = inet_reuseport_add_sock(sk, ilb);
if (err)
goto unlock;
}
// 插入nulls_head链表
__sk_nulls_add_node_rcu(sk, &ilb->nulls_head);
inet_hash2(hashinfo, sk);}__inet_hash的逻辑主要有两个，分别是inet_reuseport_add_sock和inet_hash2。我们先看inet_reuseport_add_sock。
static int inet_reuseport_add_sock(struct sock *sk,
struct inet_listen_hashbucket *ilb){
struct inet_bind_bucket *tb = inet_csk(sk)->icsk_bind_hash;
const struct hlist_nulls_node *node;
struct sock *sk2;
kuid_t uid = sock_i_uid(sk);
// 遍历nulls_head链表，第一个socket执行listen时为空
sk_nulls_for_each_rcu(sk2, node, &ilb->nulls_head) {
if (sk2 != sk &&
sk2->sk_family == sk->sk_family &&
ipv6_only_sock(sk2) == ipv6_only_sock(sk) &&
sk2->sk_bound_dev_if == sk->sk_bound_dev_if &&
inet_csk(sk2)->icsk_bind_hash == tb &&
sk2->sk_reuseport && uid_eq(uid, sock_i_uid(sk2)) &&
inet_rcv_saddr_equal(sk, sk2, false))
return reuseport_add_sock(sk, sk2,
inet_rcv_saddr_any(sk));
}
return reuseport_alloc(sk, inet_rcv_saddr_any(sk));}reuseport_add_sock和reuseport_alloc的逻辑类似。第一个socket执行listen时会执行reuseport_alloc，第二个socket执行listen时会执行reuseport_add_sock。
int reuseport_alloc(struct sock *sk, bool bind_inany){
struct sock_reuseport *reuse;
int id, ret = 0;
// 分配一个sock_reuseport结构体
reuse = __reuseport_alloc(INIT_SOCKS);
id = ida_alloc(&reuseport_ida, GFP_ATOMIC);
reuse->reuseport_id = id;
// 把socket记录到reuse结构体中
reuse->socks[0] = sk;
reuse->num_socks = 1;
reuse->bind_inany = bind_inany;
// 把sock_reuseport结构体保存到socket的sk_reuseport_cb字段
rcu_assign_pointer(sk->sk_reuseport_cb, reuse);
return ret;}reuseport_alloc是分配了一个sock_reuseport结构体，并且和socket互相关联起来。接着我们看reuseport_add_sock。
int reuseport_add_sock(struct sock *sk, struct sock *sk2, bool bind_inany){
struct sock_reuseport *old_reuse, *reuse;
// 拿到现存（执行过listen的socket）
reuse = rcu_dereference_protected(sk2->sk_reuseport_cb,
lockdep_is_held(&reuseport_lock));
// num_socks是数组当前可用索引，把执行listen的socket插入数组中
reuse->socks[reuse->num_socks] = sk;
reuse->num_socks++;
// 保存sock_reuseport到socket中
rcu_assign_pointer(sk->sk_reuseport_cb, reuse);
return 0;}以上两个函数最终会形成以下结构。
我们继续分析inet_hash2。
static void inet_hash2(struct inet_hashinfo *h, struct sock *sk){
struct inet_listen_hashbucket *ilb2;
// 从h中拿到lhash2字段
ilb2 = inet_lhash2_bucket_sk(h, sk);
spin_lock(&ilb2->lock);
// 把socket插入lhash2链表
hlist_add_head_rcu(&inet_csk(sk)->icsk_listen_portaddr_node,
&ilb2->head);
ilb2->count++;
spin_unlock(&ilb2->lock);}至此listen函数就分析完毕。接着我们分析收到连接之后，内核是如何分发的，入口函数是tcp_v4_rcv。
sk = __inet_lookup_skb(&tcp_hashinfo,
skb, __tcp_hdrlen(th),
th->source,
th->dest, sdif, &refcounted);if (sk->sk_state == TCP_LISTEN) {
ret = tcp_v4_do_rcv(sk, skb);
goto put_and_return;}主要分为两步，第一个是根据TCP报文找到对应的监听socket，第二步是把报文交给该socket处理。我们先看inet_lookup_skb，调用链很长（inet_lookup_skb->inet_lookup->inet_lookup_listener->inet_lhash2_lookup），我们只看关键代码。
struct sock *__inet_lookup_listener(struct net *net,
struct inet_hashinfo *hashinfo,
struct sk_buff *skb, int doff,
const __be32 saddr, __be16 sport,
const __be32 daddr, const unsigned short hnum,
const int dif, const int sdif){
struct inet_listen_hashbucket *ilb2;
struct sock *result = NULL;
unsigned int hash2;
hash2 = ipv4_portaddr_hash(net, daddr, hnum);
// 拿到hashinfo的lhash2字段某个元素的值
ilb2 = inet_lhash2_bucket(hashinfo, hash2);
result = inet_lhash2_lookup(net, ilb2, skb, doff,
saddr, sport, daddr, hnum,
dif, sdif);
return result;}继续看inet_lhash2_lookup
static struct sock *inet_lhash2_lookup(struct net *net,
struct inet_listen_hashbucket *ilb2,
struct sk_buff *skb, int doff,
const __be32 saddr, __be16 sport,
const __be32 daddr, const unsigned short hnum,
const int dif, const int sdif){
bool exact_dif = inet_exact_dif_match(net, skb);
struct inet_connection_sock *icsk;
struct sock *sk, *result = NULL;
int score, hiscore = 0;
// 遍历找到对应的socket
inet_lhash2_for_each_icsk_rcu(icsk, &ilb2->head) {
sk = (struct sock *)icsk;
// 计算分数
score = compute_score(sk, net, hnum, daddr,
dif, sdif, exact_dif);
// 保存得分最高的
if (score > hiscore) {
result = lookup_reuseport(net, sk, skb, doff,
saddr, sport, daddr, hnum);
if (result)
return result;
result = sk;
hiscore = score;
}
}
return result;}inet_lhash2_lookup负责从链表中选择处理该连接的socket。接着唤醒阻塞到该socket的进程。
后记：从内核实现的角度我们可以看到，SO_REUSEPORT的实现大概原理是内核会把每个进程的每个socket（设置了SO_REUSEPORT）维护起来。然后连接到来的时候通过选择算法从多个socket中选择一个socket，并且唤醒阻塞在该socket上的进程处理连接。