UDP的epoll併發處理問題-解決Open***的多線程併發問題

時間 2020-01-15

標籤 udp epoll 併發處理問題解決 open 多線程問題欄目 Java 简体版

原文原文鏈接

UDP具備是一種很好的封裝協議，好比Open×××使用UDP封裝會比TCP好不少，如今愈來愈多的業務採用UDP傳輸，而後本身定義按序到達以及流控邏輯，然而就我我的的使用經驗來看，UDP太難作併發，大多數狀況下，使用UDP會讓epoll等高性能event機制優點全無。本文以Open×××爲例，說明一下我是怎麼解決UDP併發問題的。
nginx

異步併發模型與epoll

和apache相比，nginx採用異步的處理方式，也就是說，一個線程能夠處理多個鏈接，基於event模型，來了個數據包就讀，可能依次到達的數據不屬於同一個鏈接，可是不要緊，只要能將可讀的 socket描述符和具體的鏈接對應上便可。這樣會使得在大併發場景下，讓CPU逼近其極限運轉，由於它幾乎沒有時間閒着，它會一直處理到達的數據包。 apache的模型就不是這樣，它會讓一個鏈接單獨佔有一個線程，若是有大量的鏈接就會有大量的線程，然而對於每個線程而言，其數據讀寫的壓力並非很大，這就會致使大量線程之間頻繁切換，而切換會致使cache的刷新等反作用...所以在一樣的硬件配置情形下，nginx的異步模型要比apache好不少。
算法

咱們已經知道，異步處理是搞定大併發的根本，接下來的問題是，如何讓一個就緒的socket和一個業務邏輯鏈接對應起來，這個問題在同步模型下並不存在，由於一個線程只處理一個鏈接。曾經的event機制好比select，poll，它們只能告訴你socket n就緒了，你不得不本身去經過數據結構來組織socket n和該鏈接信息之間的關係，典型的以下：apache

struct conn {
    int sd;
    void *others;
};

list conns;

一個鏈表conns囊括了該線程負責的全部鏈接，若是select/poll告訴你socket n就緒了，你不得不遍歷這個conns鏈表，比較誰的sd是n，而後取出conn來處理，雖然能夠用更加高效的數據結構，可是查找是必不可少的。然而 epoll解決了這個問題。
編程

在調用epoll_ctrl將一個socket加入到epoll中時，API會爲你提供一個指針，讓你直接綁定一個socket描述符和一個指針，一旦socket就緒，取出的是一個結構體，其中包含了與該socket對應的指針，所以你即可以這麼作：數據結構

conn.sd = sd;
conn.others = all;
ev.events = EPOLLIN;
ev.data.ptr = &conn;
epoll_ctl(kdpfd, EPOLL_CTL_ADD, sd, &ev);
while (1) {      
    nfds = epoll_wait(kdpfd, events, 10000, -1);
    for (n = 0; n < nfds; ++n) {
        conn = events[n].data.ptr;
        recv(conn.sd, ....);
        ....
    }
}

conn會一會兒取出來。這是合理的方式。畢竟，內核中已經通過socket查找了，一個5元組惟一表明瞭一個鏈接，爲什麼要在用戶態程序再找一次呢？所以除了epoll不須要遍歷全部的被監視socket以外，能夠保存用戶的指針也是其相對於select/poll的一大優點。 nginx正是用的這種方式。咱們回到Open×××。
多線程

使用TCP的Open×××

使用TCP的Open×××跟nginx幾乎是如出一轍，其核心處理邏輯以下：

併發

/* 加入偵聽socket */
context.sd = listener;
context.others = dont_care;
listen_ev.events = EPOLLIN;
listen_ev.data.ptr = context;
epoll_ctl(kdpfd, EPOLL_CTL_ADD, listener, &listen_ev);

/* 加入TUN網卡 */
tun.sd = tun;
tun.others = dont_care;
entry.ptr = tun;
entry.type = TUN;
tun_ev.events = EPOLLIN;
tun_ev.data.ptr = entry;
epoll_ctl(kdpfd, EPOLL_CTL_ADD, tun, &tun_ev);

while(1) {
    nfds = epoll_wait(kdpfd, events, 10000, -1);
    for (n = 0; n < nfds; ++n) {
        if (events[n].data.ptr == context) {
            child_sd = accept(context.sd, remote_addr....);
            multi_instance *mi = create_mi(child_sd, remote_addr, ...);
            entry.ptr = mi;
            entry.type = SOCKET;
            new_ev.events = EPOLLIN;
            new_ev.data.ptr = entry;
            epoll_ctl(kdpfd, EPOLL_CTL_ADD, child_sd, &new_ev);
            ....
        } else if (events[n].data.ptr.type == SOCKET){
            multi_instance *mi = events[n].data.ptr;
            data = read_from_socket(mi);
            // 這裏簡化了處理，由於並非每個數據包都是須要加密解密的，還有控制通道的包
            decrypt(mi, data);
            write_to_tun(data);
        } else {
            tun *tun = events[n].data.ptr.ptr;
            packet = read_from_tun(tun);
            lock(mi_hashtable);
            multi_instance *mi = lookup_multi_instance_from(packet);
            unlock(mi_hashtable);
            encrypt(packet);
            write_to_socket(packet, mi);
        }
    }
    ...
}

以上就是TCP模式下的Open×××所有邏輯，能夠看到，若是socket可讀，那麼就能夠直接取到 multi_instance，而後順序處理就是了。我記得去年我就把Open×××改爲多線程了，可是如今看來那是個失敗的作法。若是使用TCP，從上述邏輯能夠看到，就算使用多線程，在socket-to-tun這個路徑上也不用加鎖，所以multi_instance直接經過epoll_wait就能夠取的到。
異步

使用UDP的Open×××

然而對於UDP而言，Open×××的處理邏輯根上面TCP的邏輯就大相徑庭了。由於全程只有一個UDP socket，接受全部客戶端的鏈接，此時根本不存在什麼多路複用的問題，充其量也就是那惟一的UDP socket和tun網卡字符設備兩者之間的兩路複用，使用epoll徹底沒有必要。爲了定位了具體的multi_instance，你不得不先去 read惟一的那個UDP socket，而後根據recvfrom返回參數中的sockaddr結構體來構造4元組，而後根據這4元組在全局的multi_instance hash表中去查找具體multi_instance實例。其邏輯以下所示：

socket

/* 加入惟一的UDP socket */
context.sd = udp_sd;
context.others = dont_care;
listen_ev.events = EPOLLIN;
listen_ev.data.ptr = context;
epoll_ctl(kdpfd, EPOLL_CTL_ADD, listener, &listen_ev);

/* 加入TUN網卡 */
tun.sd = tun;
tun.others = dont_care;
entry.ptr = tun;
entry.type = TUN;
tun_ev.events = EPOLLIN;
tun_ev.data.ptr = entry;
epoll_ctl(kdpfd, EPOLL_CTL_ADD, tun, &tun_ev);

while(1) {
    nfds = epoll_wait(kdpfd, events, 10000, -1);
    for (n = 0; n < nfds; ++n) {  //實際上nfds最多也就是2
        if (events[n].data.ptr == context) {
            data = recvfrom(context.sd, remote_addr....);
            lock(mi_hashtable);   //若是多線程，這個鎖將會成爲瓶頸，即使是RW鎖也同樣
            multi_instance *mi = lookup_mi(child_sd, remote_addr, ...);  //再好的hash算法，也不是0成本的！
            unlock(mi_hashtable);
            // 這裏簡化了處理，由於並非每個數據包都是須要加密解密的，還有控制通道的包
            decrypt(mi, data);
            write_to_tun(data);  
            ....
        } else {
            tun *tun = events[n].data.ptr.ptr;
            packet = read_from_tun(tun);
            lock(mi_hashtable);
            multi_instance *mi = lookup_multi_instance_from(packet);
            unlock(mi_hashtable);
            encrypt(packet);
            write_to_socket(packet, mi);
        }
    }
    ...
}

可見，TCP的Open×××和UDP的Open×××處理方式徹底不一樣，UDP的問題在於，徹底沒有充分利用epoll的多路複用機制，不得不根據數據包的recvfrom返回地址來查找multi_instance...
ide

讓UDP socket也Listen起來

如果UDP也能像TCP同樣，每個用戶接進來就爲之建立一個單獨的socket爲其專門服務該多好，這樣在大併發的時候，就能夠充分複用內核UDP層的 socket查找結論加上epoll的通知機制了。理論上這是可行的，由於UDP的4元組能夠惟一識別一個與之通訊的客戶端，雖然UDP生成無鏈接，不可靠，可是爲每個鏈接的客戶端建立一個socket並無破壞UDP的語義，只是改變了UDP的編程模型而已，內核協議棧依然不會去刻意維護一個UDP連接，也不會進行任何的數據確認。
須要說明的是，這種方案僅僅對「長鏈接」的UDP有意義，好比Open×××這類。由於UDP是沒有鏈接的，那麼你也就不知道一個客戶端何時會永遠中止發送數據，所以必然要經過定時器來定時關閉那些在必定時間段內沒有數據的socket。
爲了驗證可行性，我先在用戶態作實驗，也就是說，接受一個客戶端的「鏈接請求」(其實就是一個數據包)時，我手工爲其建立一個socket，而後bind 本地地址，而且connect從recvfrom返回的對端地址，這樣理論上對於後續的數據包，epoll都應該觸發這個新的socket，畢竟它更精確。事實是否是這樣呢？如下的程序能夠證實：

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <pthread.h>
#include <assert.h>

#define SO_REUSEPORT    15

#define MAXBUF 10240
#define MAXEPOLLSIZE 100

int flag = 0;

int read_data(int sd)
{
    char recvbuf[MAXBUF + 1];
    int  ret;
    struct sockaddr_in client_addr;
    socklen_t cli_len=sizeof(client_addr);

    bzero(recvbuf, MAXBUF + 1);
  
    ret = recvfrom(sd, recvbuf, MAXBUF, 0, (struct sockaddr *)&client_addr, &cli_len);
    if (ret > 0) {
        printf("read[%d]: %s  from  %d\n", ret, recvbuf, sd);
    } else {
        printf("read err:%s  %d\n", strerror(errno), ret);
      
    }
    fflush(stdout);
}

int udp_accept(int sd, struct sockaddr_in my_addr)
{
    int new_sd = -1;
    int ret = 0;
    int reuse = 1;
    char buf[16];
    struct sockaddr_in peer_addr;
    socklen_t cli_len = sizeof(peer_addr);

    ret = recvfrom(sd, buf, 16, 0, (struct sockaddr *)&peer_addr, &cli_len);
    if (ret > 0) {
    }

    if ((new_sd = socket(PF_INET, SOCK_DGRAM, 0)) == -1) {
        perror("child socket");
        exit(1);
    } else {
        printf("parent:%d  new:%d\n", sd, new_sd);
    }

    ret = setsockopt(new_sd, SOL_SOCKET, SO_REUSEADDR, &reuse,sizeof(reuse));
    if (ret) {
        exit(1);
    }

    ret = setsockopt(new_sd, SOL_SOCKET, SO_REUSEPORT, &reuse, sizeof(reuse));
    if (ret) {
        exit(1);
    }

    ret = bind(new_sd, (struct sockaddr *) &my_addr, sizeof(struct sockaddr));
    if (ret){
        perror("chid bind");
        exit(1);
    } else {
    }

    peer_addr.sin_family = PF_INET;
    printf("aaa:%s\n", inet_ntoa(peer_addr.sin_addr));
    if (connect(new_sd, (struct sockaddr *) &peer_addr, sizeof(struct sockaddr)) == -1) {
        perror("chid connect");
        exit(1);
    } else {
    }

out:
    return new_sd;
}

int main(int argc, char **argv)
{
    int listener, kdpfd, nfds, n, curfds;
    socklen_t len;
    struct sockaddr_in my_addr, their_addr;
    unsigned int port;
    struct epoll_event ev;
    struct epoll_event events[MAXEPOLLSIZE];
    int opt = 1;;
    int ret = 0;

    port = 1234;
  
    if ((listener = socket(PF_INET, SOCK_DGRAM, 0)) == -1) {
        perror("socket");
        exit(1);
    } else {
        printf("socket OK\n");
    }

    ret = setsockopt(listener,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));
    if (ret) {
        exit(1);
    }

    ret = setsockopt(listener, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
    if (ret) {
        exit(1);
    }
  
    bzero(&my_addr, sizeof(my_addr));
    my_addr.sin_family = PF_INET;
    my_addr.sin_port = htons(port);
    my_addr.sin_addr.s_addr = INADDR_ANY;
    if (bind(listener, (struct sockaddr *) &my_addr, sizeof(struct sockaddr)) == -1) {
        perror("bind");
        exit(1);
    } else {
        printf("IP bind OK\n");
    }
  
    kdpfd = epoll_create(MAXEPOLLSIZE);

    ev.events = EPOLLIN|EPOLLET;
    ev.data.fd = listener;

    if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, listener, &ev) < 0) {
        fprintf(stderr, "epoll set insertion error: fd=%dn", listener);
        return -1;
    } else {
        printf("ep add OK\n");
    }
 
    while (1) {
      
        nfds = epoll_wait(kdpfd, events, 10000, -1);
        if (nfds == -1) {
            perror("epoll_wait");
            break;
        }
      
        for (n = 0; n < nfds; ++n) {
            if (events[n].data.fd == listener) {
                printf("listener:%d\n", n);
                int new_sd;               
                struct epoll_event child_ev;

                new_sd = udp_accept(listener, my_addr);
                child_ev.events = EPOLLIN;
                child_ev.data.fd = new_sd;
                if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, new_sd, &child_ev) < 0) {
                    fprintf(stderr, "epoll set insertion error: fd=%dn", new_sd);
                    return -1;
                }
            } else {
                read_data(events[n].data.fd);
            }
        }
    }
    close(listener);
    return 0;
}

須要說明的是，REUSEPORT是必要的，由於在connect以前，你必須爲新建的socket bind跟listener同樣的IP地址和端口，所以就須要這個socket選項。
此時，若是你用多個udp客戶端去給這個服務端發數據，會發現徹底實現了想要的效果。

內核中的UDP Listener

雖然在用戶態能夠實現效果，可是編程模型並不太好用，爲了建立一個socket，你不得不先去recvfrom一下數據，好獲得對端的地址，雖然使用 PEEK標誌可讓建立好child socket後再讀一次，可是仔細想一想，最完全的方案仍是直接擴展內核，我基於3.9.6內核，對__udp4_lib_rcv這個UDP協議棧接收函數做了如下的修改：

int __udp4_lib_rcv(struct sk_buff *skb, struct udp_table *udptable,
                   int proto)
{
......................
        sk = __udp4_lib_lookup_skb(skb, uh->source, uh->dest, udptable);

        if (sk != NULL) {
                int ret;
#if 1
                // 這個UDP_LISTEN，經過setsockopt來設置
                if (sk->sk_state == UDP_LISTEN) {
                        // 若是是UDP的listener，建立一個數據socket
                        struct sock *newsk = inet_udp_clone_lock(sk, skb, GFP_ATOMIC);
                        if (newsk) {
                                struct inet_sock *newinet;
                                // 爲這個數據傳輸socket根據skb來填充4元組信息
                                newinet               = inet_sk(newsk);
                                newinet->inet_daddr   = ip_hdr(skb)->saddr;
                                newinet->inet_rcv_saddr = ip_hdr(skb)->daddr;
                                newinet->inet_saddr           = ip_hdr(skb)->daddr;
                                rcu_assign_pointer(newinet->inet_opt, NULL);

                                newinet->mc_index     = inet_iif(skb);
                                newinet->mc_ttl       = ip_hdr(skb)->ttl;
                                newinet->rcv_tos      = ip_hdr(skb)->tos;
                                newinet->inet_id = 0xffffffff ^ jiffies;
                                inet_sk_rx_dst_set(newsk, skb);
                                // sock結構體新增csk變量，相似TCP的accept queue，可是爲了簡單，目前每一個Listen socket只能持有一個csk，即child sock。
                                sk->csk = newsk;

                                // 將新的數據傳輸socket排入全局的UDP socket hash表
                                if (newsk->sk_prot->get_port(newsk, newinet->inet_num)) {
                                        printk("[UDP listen] get port error\n");
                                        release_sock(newsk);
                                        err = -2;
                                        goto out_go;
                                }
                                ret = udp_queue_rcv_skb(newsk, skb);
                                // 喚醒epoll，讓epoll返回UDP Listener
                                sk->sk_data_ready(sk, 0);
                                sock_put(newsk);
                        } else {
                                printk("[UDP listen] create new error\n");
                                sock_put(sk);
                                return -1;
                        }
out_go:
                        sock_put(sk);
                        if (ret > 0)
                                return -ret;
                        return 0;
                }
#endif
                ret = udp_queue_rcv_skb(sk, skb);
                sock_put(sk);
......................
}

我只是測試，所以並無擴展UDP的accept方法，只是簡單的用getsocketopt來得到這個新的socket描述符併爲task安裝該文件描述符，setsockopt能夠設置一個UDP socket爲listener。這樣用戶態的編程模型就很簡單了。

使用新的Listen UDP來改造Open×××

有必要重構一下Open×××了，現現在它的邏輯變成了：

listen = 1;
listener = socket(PF_INET, SOCK_DGRAM, 0);
setsockopt(new_sd, SOL_SOCKET, SO_UDPLISTEN, &listen,sizeof(listen));

/* 加入偵聽socket */
context.sd = listener;
context.others = dont_care;
listen_ev.events = EPOLLIN;
listen_ev.data.ptr = context;
epoll_ctl(kdpfd, EPOLL_CTL_ADD, listener, &listen_ev);

/* 加入TUN網卡 */
tun.sd = tun;
tun.others = dont_care;
entry.ptr = tun;
entry.type = TUN;
tun_ev.events = EPOLLIN;
tun_ev.data.ptr = entry;
epoll_ctl(kdpfd, EPOLL_CTL_ADD, tun, &tun_ev);

while(1) {
    nfds = epoll_wait(kdpfd, events, 10000, -1);
    for (n = 0; n < nfds; ++n) {
        if (events[n].data.ptr == context) {
            getsockopt(context.sd, SOL_SOCKET, &newsock_info....);
            child_sd = newsock_info.sd;
            multi_instance *mi = create_mi(child_sd, newsock_info.remote_addr, ...);
            entry.ptr = mi;
            entry.type = SOCKET;
            new_ev.events = EPOLLIN;
            new_ev.data.ptr = entry;
            epoll_ctl(kdpfd, EPOLL_CTL_ADD, child_sd, &new_ev);
            // 這是UDP，內核除了通知Listener以外，還會將數據排入child_sd，所以須要去讀取，能夠參考TCP的Fastopen邏輯
            data = recvfrom(child_sd, ....);
            ....
        } else if (events[n].data.ptr.type == SOCKET){
            multi_instance *mi = events[n].data.ptr;
            data = read_from_socket(mi);
            // 這裏簡化了處理，由於並非每個數據包都是須要加密解密的，還有控制通道的包
            decrypt(mi, data);
            write_to_tun(data);
        } else {
            tun *tun = events[n].data.ptr.ptr;
            packet = read_from_tun(tun);
            lock(mi_hashtable);
            multi_instance *mi = lookup_multi_instance_from(packet);
            unlock(mi_hashtable);
            encrypt(packet);
            write_to_socket(packet, mi);
        }
    }
    ...
}

除了把accept改爲了getsockopt以外，別的幾乎和TCP的Open×××徹底一致了。如此一來，2014年改造的Open×××多線程版本就完美了，用戶態根本不須要再使用recvfrom返回的address信息來定位 multi_instance了，一個multi_instance惟一和一個socket綁定，而每個socket都由epoll來管理，大大下降了用戶態查找multi_instance的開銷，同時也避免了鎖定。