歡迎來到Linux教程網
Linux教程網
Linux教程網
Linux教程網
您现在的位置: Linux教程網 >> UnixLinux >  >> Linux基礎 >> 關於Linux

UDP的epoll並發框架-UDP Listener解決OpenVPN的並發問題

UDP具有是一種很好的封裝協議,比如OpenVPN使用UDP封裝會比TCP好很多,現在越來越多的業務采用UDP傳輸,然後自己定義按序到達以及流控邏輯,然而就我個人的使用經驗來看,UDP太難做並發,大多數情況下,使用UDP會讓epoll等高性能event機制優勢全無。本文以OpenVPN為例,說明一下我是怎麼解決UDP並發問題的。

異步並發模型與epoll

和apache相比,nginx采用異步的處理方式,也就是說,一個線程可以處理多個連接,基於event模型,來了個數據包就讀,可能依次到達的數據不屬於同一個連接,但是沒關系,只要能將可讀的socket描述符和具體的連接對應上即可。這樣會使得在大並發場景下,讓CPU逼近其極限運轉,因為它幾乎沒有時間閒著,它會一直處理到達的數據包。apache的模型就不是這樣,它會讓一個連接單獨占有一個線程,如果有大量的連接就會有大量的線程,然而對於每一個線程而言,其數據讀寫的壓力並不是很大,這就會導致大量線程之間頻繁切換,而切換會導致cache的刷新等副作用...因此在同樣的硬件配置情形下,nginx的異步模型要比apache好很多。

我們已經知道,異步處理是搞定大並發的根本,接下來的問題是,如何讓一個就緒的socket和一個業務邏輯連接對應起來,這個問題在同步模型下並不存在,因為一個線程只處理一個連接。曾經的event機制比如select,poll,它們只能告訴你socket n就緒了,你不得不自己去通過數據結構來組織socket n和該連接信息之間的關系,典型的如下:

 

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正是用的這種方式。我們回到OpenVPN。

使用TCP的OpenVPN

使用TCP的OpenVPN跟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模式下的OpenVPN全部邏輯,可以看到,如果socket可讀,那麼就可以直接取到multi_instance,然後順序處理就是了。我記得去年我就把OpenVPN改成多線程了,但是現在看來那是個失敗的做法。如果使用TCP,從上述邏輯可以看到,就算使用多線程,在socket-to-tun這個路徑上也不用加鎖,因此multi_instance直接通過epoll_wait就可以取的到。

使用UDP的OpenVPN

然而對於UDP而言,OpenVPN的處理邏輯根上面TCP的邏輯就截然不同了。因為全程只有一個UDP socket,接受所有客戶端的連接,此時根本不存在什麼多路復用的問題,充其量也就是那唯一的UDP socket和tun網卡字符設備二者之間的兩路復用,使用epoll完全沒有必要。為了定位了具體的multi_instance,你不得不先去read唯一的那個UDP socket,然後根據recvfrom返回參數中的sockaddr結構體來構造4元組,然後根據這4元組在全局的multi_instance hash表中去查找具體multi_instance實例。其邏輯如下所示:

/* 加入唯一的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的OpenVPN和UDP的OpenVPN處理方式完全不同,UDP的問題在於,完全沒有充分利用epoll的多路復用機制,不得不根據數據包的recvfrom返回地址來查找multi_instance...

讓UDP socket也Listen起來

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

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

#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來改造OpenVPN

有必要重構一下OpenVPN了,現如今它的邏輯變成了:

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的OpenVPN完全一致了。
如此一來,2014年改造的OpenVPN多線程版本就完美了,用戶態根本不需要再使用recvfrom返回的address信息來定位multi_instance了,一個multi_instance唯一和一個socket綁定,而每一個socket都由epoll來管理,大大降低了用戶態查找multi_instance的開銷,同時也避免了鎖定。
Copyright © Linux教程網 All Rights Reserved