用户态网络协议栈的原理与代码实现（C++）

网络通信方式

从网卡取到一个完整的数据的方式：

raw socket
netmap
dpdk

负载均衡产品：

nginx：工作在应用层
haproxy：工作在传输层
lvs：工作在网络层
f5：工作在数据链路层

网络协议栈

MAC地址是以太网产物
IP地址是网络层产物
port端口是传输层产物
NAT转换ip+端口网络地址映射工作是在传输层的

内核协议栈与用户协议栈区别

内核协议栈

用户态协议栈

UDP、ICMP、ARP用户态协议栈设计

协议头

以太网协议头

IP协议头

UDP协议头

ARP协议

ICMP协议

数据帧

UDP数据帧

UDP数据帧协议栈设计代码示例

基于netmap实现UPD数据帧处理的用户态协议栈。

#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>#include <sys/poll.h>
#include <arpa/inet.h>#define NETMAP_WITH_LIBS#include <net/netmap_user.h>  //netmap库
#pragma pack(1) //结构体以1个字节进行字节对齐#define ETH_ALEN	6
#define PROTO_IP	0x0800
#define PROTO_ARP	0x0806#define PROTO_UDP	17
#define PROTO_ICMP	1
#define PROTO_IGMP	2//以太网头
struct ethhdr {unsigned char h_dest[ETH_ALEN];	//目的地址unsigned char h_source[ETH_ALEN];	//源地址unsigned short h_proto;	//类型
};//ip头
struct iphdr {unsigned char version;	//版本(4位版本+4位首部长度)unsigned char tos;	//服务类型unsigned short tot_len;	//总长度(字节数)unsigned short id;	//标识unsigned short flag_off;	//偏移(3位标志+13位偏移)unsigned char ttl;	//生存时间unsigned char protocol;	//协议unsigned short check;	//首部检验和unsigned int saddr;	//源IPunsigned int daddr;	//目标IP
};//udp头
struct udphdr {unsigned short source; //源端口unsigned short dest; //目标端口unsigned short len; //udp长度unsigned short check; //udp校验
};//udp数据包
struct udppkt {struct ethhdr eh;	//以太网头struct iphdr ip;	//ip头struct udphdr udp;	//udp头unsigned char body[128];	//数据
};struct arphdr {unsigned short h_type;unsigned short h_proto;unsigned char h_addrlen;unsigned char protolen;unsigned short oper;unsigned char smac[ETH_ALEN];unsigned int sip;unsigned char dmac[ETH_ALEN];unsigned int dip;
};struct arppkt {struct ethhdr eh;struct arphdr arp;
};struct icmphdr {unsigned char type;unsigned char code;unsigned short check;unsigned short identifier;unsigned short seq;unsigned char data[32];
};struct icmppkt {struct ethhdr eh;struct iphdr ip;struct icmphdr icmp;
};void print_mac(unsigned char *mac) {int i = 0;for (i = 0;i < ETH_ALEN-1;i ++) {printf("%02x:", mac[i]);}printf("%02x", mac[i]);
}void print_ip(unsigned char *ip) {int i = 0;for (i = 0;i < 3;i ++) {printf("%d.", ip[i]);}printf("%d", ip[i]);
}void print_arp(struct arppkt *arp) {print_mac(arp->eh.h_dest);printf(" ");print_mac(arp->eh.h_source);printf(" ");printf("0x%04x ", ntohs(arp->eh.h_proto));printf("  ");
}int str2mac(char *mac, char *str) {char *p = str;unsigned char value = 0x0;int i = 0;while (p != '\0') {if (*p == ':') {mac[i++] = value;value = 0x0;} else {unsigned char temp = *p;if (temp <= '9' && temp >= '0') {temp -= '0';} else if (temp <= 'f' && temp >= 'a') {temp -= 'a';temp += 10;} else if (temp <= 'F' && temp >= 'A') {temp -= 'A';temp += 10;} else {	break;}value <<= 4;value |= temp;}p ++;}mac[i] = value;return 0;
}void echo_arp_pkt(struct arppkt *arp, struct arppkt *arp_rt, char *hmac) {memcpy(arp_rt, arp, sizeof(struct arppkt));memcpy(arp_rt->eh.h_dest, arp->eh.h_source, ETH_ALEN);str2mac(arp_rt->eh.h_source, hmac);arp_rt->eh.h_proto = arp->eh.h_proto;arp_rt->arp.h_addrlen = 6;arp_rt->arp.protolen = 4;arp_rt->arp.oper = htons(2);str2mac(arp_rt->arp.smac, hmac);arp_rt->arp.sip = arp->arp.dip;memcpy(arp_rt->arp.dmac, arp->arp.smac, ETH_ALEN);arp_rt->arp.dip = arp->arp.sip;
}void echo_udp_pkt(struct udppkt *udp, struct udppkt *udp_rt) {memcpy(udp_rt, udp, sizeof(struct udppkt));memcpy(udp_rt->eh.h_dest, udp->eh.h_source, ETH_ALEN);memcpy(udp_rt->eh.h_source, udp->eh.h_dest, ETH_ALEN);udp_rt->ip.saddr = udp->ip.daddr;udp_rt->ip.daddr = udp->ip.saddr;udp_rt->udp.source = udp->udp.dest;udp_rt->udp.dest = udp->udp.source;}unsigned short in_cksum(unsigned short *addr, int len)
{register int nleft = len;register unsigned short *w = addr;register int sum = 0;unsigned short answer = 0;while (nleft > 1)  {sum += *w++;nleft -= 2;}if (nleft == 1) {*(u_char *)(&answer) = *(u_char *)w ;sum += answer;}sum = (sum >> 16) + (sum & 0xffff);	sum += (sum >> 16);			answer = ~sum;return (answer);}void echo_icmp_pkt(struct icmppkt *icmp, struct icmppkt *icmp_rt) {memcpy(icmp_rt, icmp, sizeof(struct icmppkt));icmp_rt->icmp.type = 0x0; //icmp_rt->icmp.code = 0x0; //icmp_rt->icmp.check = 0x0;icmp_rt->ip.saddr = icmp->ip.daddr;icmp_rt->ip.daddr = icmp->ip.saddr;memcpy(icmp_rt->eh.h_dest, icmp->eh.h_source, ETH_ALEN);memcpy(icmp_rt->eh.h_source, icmp->eh.h_dest, ETH_ALEN);icmp_rt->icmp.check = in_cksum((unsigned short*)&icmp_rt->icmp, sizeof(struct icmphdr));}int main() {struct ethhdr *eh;struct pollfd pfd = {0};struct nm_pkthdr h;unsigned char *stream = NULL;//把eth0网卡的所有数据映射到*nmr内存中, 网卡数据就不会走内核协议栈了, 而是直接发到*nmr中。struct nm_desc *nmr = nm_open("netmap:eth0", NULL, 0, NULL);if (nmr == NULL) {return -1;}pfd.fd = nmr->fd;pfd.events = POLLIN;while (1) {int ret = poll(&pfd, 1, -1);if (ret < 0) continue;if (pfd.revents & POLLIN) {stream = nm_nextpkt(nmr, &h);eh = (struct ethhdr*)stream;if (ntohs(eh->h_proto) == PROTO_IP) {struct udppkt *udp = (struct udppkt*)stream;if (udp->ip.protocol == PROTO_UDP) {struct in_addr addr;addr.s_addr = udp->ip.saddr;int udp_length = ntohs(udp->udp.len);printf("%s:%d:length:%d, ip_len:%d --> ", inet_ntoa(addr), udp->udp.source, udp_length, ntohs(udp->ip.tot_len));udp->body[udp_length-8] = '\0';printf("udp --> %s\n", udp->body);struct udppkt udp_rt;echo_udp_pkt(udp, &udp_rt);nm_inject(nmr, &udp_rt, sizeof(struct udppkt));} }  } }
}

使用nm_open()函数时，需要指定的是物理网卡名。eth0是物理显卡名，ens33是虚拟网卡名。

修改网卡名字：

sudo vim /etc/default/grub//修改GRUB_CMDLINE_LINUX为如下，主要是增加 net.ifnames=0 biosdevname=0 这句
GRUB_CMDLINE_LINUX="find_presend=/presend.cfg noprompt net.ifnames=0 biosdevname=0 default_hugepagesz=1G hugepagesz=2M hugepages=1024 isolcpus=0-2"

启动程序后，过段时间会出现两个问题：

1.刚开始可以接收udp包，过一段时间后就接受不到了。

原因：程序把网卡的数据发送到了共享内存，不经过协议栈。而局域网内所有机器每隔一段时间会发送arp协议告知局域网内其他机器自己的IP和MAC地址，如果一段时间内没有收到对方的arp协议，那么本机就会把arp表对应的arp协议信息（IP和MAC地址）删掉。

因此，因为一开始发送udp包对方的时候，还知道对方的IP和MAC地址。对方因为没有走协议栈，对方就会不发arp协议给我，那么过段时间后，我的arp表就会把对方的IP和MAC地址信息删掉，我就没办法知道对方的IP和MAC地址，因此后面就无法发送upd包给到对方了。

2.没开启进程前，可以ping通进程所在的机器，过段时间后无法ping通。

原因：程序把网卡的数据发送到了共享内存，不经过协议栈。而ping协议的反馈是走ICMP协议的

因此，因为ping对方的时候，对方因为没有走协议栈，对如果对方处理网卡信息的时候，没有实现对ICMP协议的解析和回复，那么我ping对方就没办法收到对方的反馈。

解决办法：在程序中增加处理ARP和ICMP协议。

需要C/C++ Linux服务器架构师学习资料加qun812855908获取（资料包括C/C++，Linux，golang技术，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，ffmpeg等），免费分享

UDP-ICMP-ARP数据帧协议栈设计代码示例

基于netmap实现UPD、ICMP、ARP数据帧处理的用户态协议栈。

#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>#include <sys/poll.h>
#include <arpa/inet.h>#define NETMAP_WITH_LIBS#include <net/netmap_user.h>  //netmap库
#pragma pack(1) //结构体以1个字节进行字节对齐#define ETH_ALEN	6
#define PROTO_IP	0x0800
#define PROTO_ARP	0x0806#define PROTO_UDP	17
#define PROTO_ICMP	1
#define PROTO_IGMP	2//以太网头
struct ethhdr {unsigned char h_dest[ETH_ALEN];	//目的地址unsigned char h_source[ETH_ALEN];	//源地址unsigned short h_proto;	//类型
};//ip头
struct iphdr {unsigned char version;	//版本(4位版本+4位首部长度)unsigned char tos;	//服务类型unsigned short tot_len;	//总长度(字节数)unsigned short id;	//标识unsigned short flag_off;	//偏移(3位标志+13位偏移)unsigned char ttl;	//生存时间unsigned char protocol;	//协议unsigned short check;	//首部检验和unsigned int saddr;	//源IPunsigned int daddr;	//目标IP
};//udp头
struct udphdr {unsigned short source; //源端口unsigned short dest; //目标端口unsigned short len; //udp长度unsigned short check; //udp校验
};//udp数据包
struct udppkt {struct ethhdr eh;	//以太网头struct iphdr ip;	//ip头struct udphdr udp;	//udp头unsigned char body[128];	//数据
};struct arphdr {unsigned short h_type;unsigned short h_proto;unsigned char h_addrlen;unsigned char protolen;unsigned short oper;unsigned char smac[ETH_ALEN];unsigned int sip;unsigned char dmac[ETH_ALEN];unsigned int dip;
};struct arppkt {struct ethhdr eh;struct arphdr arp;
};struct icmphdr {unsigned char type;unsigned char code;unsigned short check;unsigned short identifier;unsigned short seq;unsigned char data[32];
};struct icmppkt {struct ethhdr eh;struct iphdr ip;struct icmphdr icmp;
};void print_mac(unsigned char *mac) {int i = 0;for (i = 0;i < ETH_ALEN-1;i ++) {printf("%02x:", mac[i]);}printf("%02x", mac[i]);
}void print_ip(unsigned char *ip) {int i = 0;for (i = 0;i < 3;i ++) {printf("%d.", ip[i]);}printf("%d", ip[i]);
}void print_arp(struct arppkt *arp) {print_mac(arp->eh.h_dest);printf(" ");print_mac(arp->eh.h_source);printf(" ");printf("0x%04x ", ntohs(arp->eh.h_proto));printf("  ");
}int str2mac(char *mac, char *str) {char *p = str;unsigned char value = 0x0;int i = 0;while (p != '\0') {if (*p == ':') {mac[i++] = value;value = 0x0;} else {unsigned char temp = *p;if (temp <= '9' && temp >= '0') {temp -= '0';} else if (temp <= 'f' && temp >= 'a') {temp -= 'a';temp += 10;} else if (temp <= 'F' && temp >= 'A') {temp -= 'A';temp += 10;} else {	break;}value <<= 4;value |= temp;}p ++;}mac[i] = value;return 0;
}void echo_arp_pkt(struct arppkt *arp, struct arppkt *arp_rt, char *hmac) {memcpy(arp_rt, arp, sizeof(struct arppkt));memcpy(arp_rt->eh.h_dest, arp->eh.h_source, ETH_ALEN);str2mac(arp_rt->eh.h_source, hmac);arp_rt->eh.h_proto = arp->eh.h_proto;arp_rt->arp.h_addrlen = 6;arp_rt->arp.protolen = 4;arp_rt->arp.oper = htons(2);str2mac(arp_rt->arp.smac, hmac);arp_rt->arp.sip = arp->arp.dip;memcpy(arp_rt->arp.dmac, arp->arp.smac, ETH_ALEN);arp_rt->arp.dip = arp->arp.sip;
}void echo_udp_pkt(struct udppkt *udp, struct udppkt *udp_rt) {memcpy(udp_rt, udp, sizeof(struct udppkt));memcpy(udp_rt->eh.h_dest, udp->eh.h_source, ETH_ALEN);memcpy(udp_rt->eh.h_source, udp->eh.h_dest, ETH_ALEN);udp_rt->ip.saddr = udp->ip.daddr;udp_rt->ip.daddr = udp->ip.saddr;udp_rt->udp.source = udp->udp.dest;udp_rt->udp.dest = udp->udp.source;
}unsigned short in_cksum(unsigned short *addr, int len)
{register int nleft = len;register unsigned short *w = addr;register int sum = 0;unsigned short answer = 0;while (nleft > 1)  {sum += *w++;nleft -= 2;}if (nleft == 1) {*(u_char *)(&answer) = *(u_char *)w ;sum += answer;}sum = (sum >> 16) + (sum & 0xffff);	sum += (sum >> 16);			answer = ~sum;return (answer);
}void echo_icmp_pkt(struct icmppkt *icmp, struct icmppkt *icmp_rt) {memcpy(icmp_rt, icmp, sizeof(struct icmppkt));icmp_rt->icmp.type = 0x0; //icmp_rt->icmp.code = 0x0; //icmp_rt->icmp.check = 0x0;icmp_rt->ip.saddr = icmp->ip.daddr;icmp_rt->ip.daddr = icmp->ip.saddr;memcpy(icmp_rt->eh.h_dest, icmp->eh.h_source, ETH_ALEN);memcpy(icmp_rt->eh.h_source, icmp->eh.h_dest, ETH_ALEN);icmp_rt->icmp.check = in_cksum((unsigned short*)&icmp_rt->icmp, sizeof(struct icmphdr));
}int main() {struct ethhdr *eh;struct pollfd pfd = {0};struct nm_pkthdr h;unsigned char *stream = NULL;//把eth0网卡的所有数据映射到*nmr内存中, 网卡数据就不会走内核协议栈了, 而是直接发到*nmr中。struct nm_desc *nmr = nm_open("netmap:eth0", NULL, 0, NULL);if (nmr == NULL) {return -1;}pfd.fd = nmr->fd;pfd.events = POLLIN;while (1) {int ret = poll(&pfd, 1, -1);if (ret < 0) continue;if (pfd.revents & POLLIN) {stream = nm_nextpkt(nmr, &h);eh = (struct ethhdr*)stream;if (ntohs(eh->h_proto) == PROTO_IP) {struct udppkt *udp = (struct udppkt*)stream;if (udp->ip.protocol == PROTO_UDP) {struct in_addr addr;addr.s_addr = udp->ip.saddr;int udp_length = ntohs(udp->udp.len);printf("%s:%d:length:%d, ip_len:%d --> ", inet_ntoa(addr), udp->udp.source, udp_length, ntohs(udp->ip.tot_len));udp->body[udp_length-8] = '\0';printf("udp --> %s\n", udp->body);
#if 1struct udppkt udp_rt;echo_udp_pkt(udp, &udp_rt);nm_inject(nmr, &udp_rt, sizeof(struct udppkt));
#endif
#if 0} else if (udp->ip.protocol == PROTO_ICMP) {struct icmppkt *icmp = (struct icmppkt*)stream;printf("icmp ---------- --> %d, %x\n", icmp->icmp.type, icmp->icmp.check);if (icmp->icmp.type == 0x08) {struct icmppkt icmp_rt = {0};echo_icmp_pkt(icmp, &icmp_rt);//printf("icmp check %x\n", icmp_rt.icmp.check);nm_inject(nmr, &icmp_rt, sizeof(struct icmppkt));}
#endif	} else if (udp->ip.protocol == PROTO_IGMP) {} else {printf("other ip packet");}
#if 0}  else if (ntohs(eh->h_proto) == PROTO_ARP) {struct arppkt *arp = (struct arppkt *)stream;struct arppkt arp_rt;if (arp->arp.dip == inet_addr("192.168.2.217")) {echo_arp_pkt(arp, &arp_rt, "00:50:56:33:1c:ca");nm_inject(nmr, &arp_rt, sizeof(struct arppkt));}
#endif}} }
}

TCP用户态协议栈设计

设计理论

TCP半连接队列和全连接队列

在TCP进行三次握手时，Liunx会为其维护两个队列：

半连接队列，也叫syn队列
全连接队列，也叫accept队列

应用程序中 listen(fd, backlog) 中的第二个参数backlog就是服务端半连接队列的大小，一般用5/15左右即可。

在tcp两端没建立连接前，服务端如何标识每一个客户端？

答：五元组（sip, dip, sport, dport, proto）。这个五元组作为一个客户端标识。五元组从何而来？从IP头和TCP头拿到的信息。**在客户端发起第一次连接时，服务端会将标识客户端的数据结构（五元组）加入到syn队列中。

在客户端发起第一次连接时，服务端会将其加入到syn队列中，并且响应客户端syn+ack报文，等到客户端发送ack应答报文时，服务端将该连接从半连接队列中取出，并新建一个新的连接，加入到accept队列当中。等待进程调用accept请求时，将该连接取出来

不管是半连接队列还是全连接队列，都有最大长度限制，超过限制时，内核会直接丢弃，或返回 RST 包。

半连接队列(syn队列)

如何查看半连接队列的长度呢？？

我们可以抓住半连接队列的特点，处于syn_recv状态的tcp连接，就是我们的半连接队列

于是，我们使用如下命令查看处于syn_recv状态的tcp连接

netstat -natp | grep SYN_RECV | wc -l

如何模拟tcp半连接队列溢出场景

我们只需要一直对服务端发送syn包，但是不回ack回应包，这样就会使得服务端有大量请求处于syn_recv状态，这就是所谓的syn洪泛，syn攻击，DDos攻击

如何抵御syn攻击

1.增大半连接队列

不能只增大tcp_max_syn_backlog,还需要一同增大somaconn和backlog，也就是增大全连接队列

2.开启tcp_syncookies功能

开启tcp_syncookies就可以在不使用syn半连接队列的情况下建立连接

syncookies在接收到客户端的syn报文时，计算出一个值，放到syn+ack报文中发出。当客户端返回ack报文时，取出该值验证，成功则建立连接，如下图：

3.减少ack+syn报文的重传次数

因为我们在收到syn攻击时，服务端会重传syn+ack报文到最大次数，才会断开连接。针对syn攻击的场景，我们可以减少ack+syn报文的重传次数，使处于syn_recv状态的它们更快断开连接

修改重传次数:/proc/sys/net/ipv4/tcp_synack_retries

全连接队列(accept队列)

如何知道TCP全连接队列的大小

可以使用ss命令，来查看TCP全连接队列的情况

全连接队列溢出

当服务端的全连接队列过小时，容易发生全连接队列溢出。发生全连接队列溢出，后续的请求就会别丢弃。

Linux有个参数可以指定TCP全连接队列满了，会使用什么策略来回应客户端

丢弃连接只是liunx的默认行为，我们还可以向客户端发送RST报文终止连接，告诉客户端连接失败

cp_abort_on_overflow共有两个值分别是0和1

0：如果全连接队列满了，那么服务端丢弃ack报文
1：如果全连接队列满了，那么服务端会想客户端发送RST报文，终止这个握手连接

通常情况下设置为0更好，可以提高效率

如果设置为0的话，此时服务端全连接队列满了，客户端发送过来的ack报文，服务端丢弃。而此时客户端还会继续重传，如果此时服务端的全连接队列有空闲，那么就会接受重传的ack包，这样就能直接建立连接了。而设置为1的话，还需要重新连接

如何增大全连接队列呢？

当全连接队列溢出后，我们需要增大全连接队列的长度，以提高请求容量

TCP 全连接队列的最大值取决于 somaxconn 和 backlog 之间的最小值，也就是 min(somaxconn, backlog)，所以我们需要提高这两个参数的大小才能拿增大全连接队列

TCP慢启动、拥塞控制、超时重传、快速重传

慢启动

TCP 在刚建立连接完成后，首先是有个慢启动的过程，这个慢启动的意思就是一点一点的提高发送数据包的数量，如果一上来就发大量的数据，这不是给网络添堵吗？

慢启动的算法记住一个规则就行：当发送方每收到一个 ACK，拥塞窗口 cwnd 的大小就会加 1。

可以看出慢启动算法，发包的个数是指数性的增长。

那慢启动涨到什么时候是个头呢？

有一个叫慢启动门限 ssthresh （slow start threshold）状态变量。

当 cwnd < ssthresh 时，使用慢启动算法。
当 cwnd >= ssthresh 时，就会使用「拥塞避免算法」。

拥塞控制

前面说道，当拥塞窗口 cwnd 「超过」慢启动门限 ssthresh 就会进入拥塞避免算法。

一般来说 ssthresh 的大小是 65535 字节。

那么进入拥塞避免算法后，它的规则是：每当收到一个 ACK 时，cwnd 增加 1/cwnd。

接上前面的慢启动的栗子，现假定 ssthresh 为 8：

当 8 个 ACK 应答确认到来时，每个确认增加 1/8，8 个 ACK 确认 cwnd 一共增加 1，于是这一次能够发送 9 个 MSS 大小的数据，变成了线性增长。

所以，我们可以发现，拥塞避免算法就是将原本慢启动算法的指数增长变成了线性增长，还是增长阶段，但是增长速度缓慢了一些。

就这么一直增长着后，网络就会慢慢进入了拥塞的状况了，于是就会出现丢包现象，这时就需要对丢失的数据包进行重传。

当触发了重传机制，也就进入了「拥塞发生算法」。

拥塞发生

当网络出现拥塞，也就是会发生数据包重传，重传机制主要有两种：

超时重传
快速重传

这两种使用的拥塞发送算法是不同的，接下来分别来说说。

当发生了「超时重传」，则就会使用拥塞发生算法。

这个时候，ssthresh 和 cwnd 的值会发生变化：

ssthresh 设为 cwnd/2，
cwnd 重置为 1

接着，就重新开始慢启动，慢启动是会突然减少数据流的。这真是一旦「超时重传」，马上回到解放前。但是这种方式太激进了，反应也很强烈，会造成网络卡顿。

就好像本来在秋名山高速漂移着，突然来个紧急刹车，轮胎受得了吗。。。

发生快速重传的拥塞发生算法

还有更好的方式，前面我们讲过「快速重传算法」。当接收方发现丢了一个中间包的时候，发送三次前一个包的 ACK，于是发送端就会快速地重传，不必等待超时再重传。

TCP 认为这种情况不严重，因为大部分没丢，只丢了一小部分，则 ssthresh 和 cwnd 变化如下：

cwnd = cwnd/2 ，也就是设置为原来的一半;
ssthresh = cwnd;
进入快速恢复算法

快速恢复

快速重传和快速恢复算法一般同时使用，快速恢复算法是认为，你还能收到 3 个重复 ACK 说明网络也不那么糟糕，所以没有必要像 RTO 超时那么强烈。

正如前面所说，进入快速恢复之前，cwnd 和 ssthresh 已被更新了：

cwnd = cwnd/2 ，也就是设置为原来的一半;
ssthresh = cwnd;

然后，进入快速恢复算法如下：

拥塞窗口 cwnd = ssthresh + 3 （ 3 的意思是确认有 3 个数据包被收到了）；
重传丢失的数据包；
如果再收到重复的 ACK，那么 cwnd 增加 1；
如果收到新数据的 ACK 后，把 cwnd 设置为第一步中的 ssthresh 的值，原因是该 ACK 确认了新的数据，说明从 duplicated ACK 时的数据都已收到，该恢复过程已经结束，可以回到恢复之前的状态了，也即再次进入拥塞避免状态；

也就是没有像「超时重传」一夜回到解放前，而是还在比较高的值，后续呈线性增长。

超时重传

重传机制的其中一个方式，就是在发送数据时，设定一个定时器，当超过指定的时间后，没有收到对方的 ACK 确认应答报文，就会重发该数据，也就是我们常说的超时重传。

TCP 会在以下两种情况发生超时重传：

数据包丢失
确认应答丢失

超时时间设置

**RTT：**Round-Trip Time 往返时延，数据从网络一端传送到另一端所需的时间，也就是包的往返时间。

**RTO：**Retransmission Timeout 超时重传时间。

假设在重传的情况下，超时时间 RTO 「较长或较短」时，会发生什么事情呢？

上图中有两种超时时间不同的情况：

当超时时间 RTO 较大时，重发就慢，丢了老半天才重发，没有效率，性能差；
当超时时间 RTO 较小时，会导致可能并没有丢就重发，于是重发的就快，会增加网络拥塞，导致更多的超时，更多的超时导致更多的重发。

根据上述的两种情况，我们可以得知，超时重传时间 RTO 的值应该略大于报文往返 RTT 的值。

可能大家觉得超时重传时间 RTO 的值计算，也不是很复杂嘛。

好像就是在发送端发包时记下 t0 ，然后接收端再把这个 ack 回来时再记一个 t1，于是 RTT = t1 – t0。没那么简单，这只是一个采样，不能代表普遍情况。

实际上「报文往返 RTT 的值」是经常变化的，因为我们的网络也是时常变化的。也就因为「报文往返 RTT 的值」是经常波动变化的，所以「超时重传时间 RTO 的值」应该是一个动态变化的值。

Linux计算RTO

估计往返时间，通常需要采样以下两个：

需要 TCP 通过采样 RTT 的时间，然后进行加权平均，算出一个平滑 RTT 的值，而且这个值还是要不断变化的，因为网络状况不断地变化。
除了采样 RTT，还要采样 RTT 的波动范围，这样就避免如果 RTT 有一个大的波动的话，很难被发现的情况。

RFC6289 建议使用以下的公式计算 RTO：

RFC6289 建议的 RTO 计算

其中 SRTT 是计算平滑的RTT ，DevRTR 是计算平滑的RTT 与最新 RTT 的差距。

在 Linux 下，α = 0.125，β = 0.25， μ = 1，∂ = 4。别问怎么来的，问就是大量实验中调出来的。

如果超时重发的数据，再次超时的时候，又需要重传的时候，TCP 的策略是超时间隔加倍。

也就是每当遇到一次超时重传的时候，都会将下一次超时时间间隔设为先前值的两倍。两次超时，就说明网络环境差，不宜频繁反复发送。

超时触发重传存在的问题是，超时周期可能相对较长。那是不是可以有更快的方式呢？

于是就可以用「快速重传」机制来解决超时重发的时间等待。

快速重传

快速重传（Fast Retransmit）机制，它不以时间为驱动，而是以数据驱动重传。

在上图，发送方发出了 1，2，3，4，5 份数据：

第一份 Seq1 先送到了，于是就回ack=2；
结果 Seq2 因为某些原因没收到，Seq3 到达了，于是还是回ack=2；
后面的 Seq4 和 Seq5 都到了，但还是回ack=2，因为 Seq2 还是没有收到；

发送端收到了三个 ack= 2 的确认，知道了 Seq2 还没有收到，就会在定时器过期之前，重传丢失的 Seq2。

最后，收到了 Seq2，此时因为 Seq3，Seq4，Seq5 都收到了，于是 Ack 回 6 。

所以，快速重传的工作方式是当收到三个相同的 ACK 报文时，会在定时器过期之前，重传丢失的报文段。

快速重传机制只解决了一个问题，就是超时时间的问题，但是它依然面临着另外一个问题。就是重传的时候，是重传之前的一个，还是重传所有的问题。

比如对于上面的例子，是重传 Seq2 呢？还是重传 Seq2、Seq3、Seq4、Seq5 呢？因为发送端并不清楚这连续的三个 Ack 2 是谁传回来的。

根据 TCP 不同的实现，以上两种情况都是有可能的。可见，这是一把双刃剑。

为了解决不知道该重传哪些 TCP 报文，于是就有 SACK 方法。

SACK方法

SACK（ Selective Acknowledgment 选择性确认）。

这种方式需要在 TCP 头部「选项」字段里加一个 SACK 的东西，它可以将缓存的地图发送给发送方，这样发送方就可以知道哪些数据收到了，哪些数据没收到，知道了这些信息，就可以只重传丢失的数据。

如下图，发送方收到了三次同样的 ACK 确认报文，于是就会触发快速重发机制，通过 SACK 信息发现只有 200~299 这段数据丢失，则重发时，就只选择了这个 TCP 段进行重复。

如果要支持 SACK，必须双方都要支持。在 Linux 下，可以通过 net.ipv4.tcp_sack 参数打开这个功能（Linux 2.4 后默认打开）。

Duplicate SACK

Duplicate SACK 又称 D-SACK，其主要使用了 SACK 来告诉「发送方」有哪些数据被重复接收了。

ACK丢包

「接收方」发给「发送方」的两个 ACK 确认应答都丢失了，所以发送方超时后，重传第一个数据包（3000 ~ 3499）
于是「接收方」发现数据是重复收到的，于是回了一个 SACK = 3000~3500，告诉「发送方」 3000~3500 的数据早已被接收了，因为 ACK 都到了 4000 了，已经意味着 4000 之前的所有数据都已收到，所以这个 SACK 就代表着 D-SACK。
这样「发送方」就知道了，数据没有丢，是「接收方」的 ACK 确认报文丢了。

网络延时

数据包（1000~1499）被网络延迟了，导致「发送方」没有收到 Ack 1500 的确认报文。
而后面报文到达的三个相同的 ACK 确认报文，就触发了快速重传机制，但是在重传后，被延迟的数据包（1000~1499）又到了「接收方」；
所以「接收方」回了一个 SACK=1000~1500，因为 ACK 已经到了 3000，所以这个 SACK 是 D-SACK，表示收到了重复的包。
这样发送方就知道快速重传触发的原因不是发出去的包丢了，也不是因为回应的 ACK 包丢了，而是因为网络延迟了。

D-SACK 有这么几个好处：

可以让「发送方」知道，是发出去的包丢了，还是接收方回应的 ACK 包丢了;
可以知道是不是「发送方」的数据包被网络延迟了;
可以知道网络中是不是把「发送方」的数据包给复制了;

在 Linux 下可以通过 net.ipv4.tcp_dsack 参数开启/关闭这个功能（Linux 2.4 后默认打开）。

TCP的4个定时器

超时重传定时器：tcp客户端每发送一个包，如果在规定时间内没有收到对端的ack确认，那么重传该数据包。
探测定时器：当对端反馈接收窗口为空时，每隔一段时间发送一个探测包询问对面接收窗口是否可以开始接收数据。
keepalive定时器：两端长期不进行消息通信时，会每隔一段时间发送一个keepalive包，确认和对方还进行正常连接。（不建议用，传输层断开连接，应用层使用过程无法感知。我们的应用程序没法把控与对端的断开时机，很可能tcp层自动断开连接并且回收tcb，那么我们应用程序的socketfd失效却无法感知，导致socketfd指向一个未知）
time_wait定时器：四次挥手期间，最后自己断开连接前，保持一段时间监听是否对方收到自己的ack，如果在此期间对方没收到我方ack并且发数据包询问，那么可以重新发一个ack包给对面。

代码实现

安装netmap
下载编译源码NtyTcp（基于netmap开发）

NtyTcp

用户态协议栈之 TCP/IP 的设计

1. Netmap 简介

Netmap 是一个高性能收发原始数据包的框架，由 Luigi Rizzo 等人开发完成，其包含了内核模块以及用户态库函数。其目标是，不修改现有操作系统软件以及不需要特殊硬件支持，实现用户态和网卡之间数据包的高性能传递。其原理图如下，数据包不经过操作系统内核进行处理，用户空间程序收发数据包时，直接与网卡进行通信。

代码位置：https://github.com/luigirizzo/netmap

1. 数据结构

在 Netmap 框架下，内核拥有数据包池，发送环接收环上的数据包不需要动态申请，有数据到达网卡时，当有数据到达后，直接从数据包池中取出一个数据包，然后将数据放入此数据包中，再将数据包的描述符放入接收环中。内核中的数据包池，通过 mmap 技术映射到用户空间。用户态程序最终通过 netmap_if 获取接收发送环 netmap_ring，进行数据包的获取发送。

2. 特点总结

（1）性能高：数据包不走传统协议栈，不需要层层解析，用户态直接与网卡的接受环和发送环交互。性能高的具体原因有一下三个：

（a）系统调用以及处理数据包的时间花费少

（b）不需要进行数据包的内存分配：采用数据包池，当有数据到达后，直接从数据包池中取出一个数据包，然后将数据放入此数据包中，再将数据包的描述符放入接收环中。

（c）数据拷贝次数少：内核中的数据包采用 mmap 技术映射到用户态。所以数据包在到达用户态时，不需要进行数据包的拷贝。

（2）稳定性高：有关网卡寄存器数据的维护都是在内核模块进行，用户不会直接操作寄存器。所以在用户态操作时，不会导致操作系统崩溃

（3）亲和性：可采用了 CPU 亲和性，实现 CPU 和网卡绑定，提高性能。

（4）易用性好：API 操作简单，用户态只需要调用 ioctl 函数即可完成数据包收发工作

（5）与硬件解耦：不依赖硬件，只需要对网卡驱动程序稍微做点修改就可以使用此框架（几十行行），传统网卡驱动将数据包传递给操作系统内核中协议栈，而修改后的数据包直接放入 Netmap_ring 供用户使用。

2. Netmap API 介绍

1. 简要说明

netmap API 主要为两个头文件 netmap.h 和 netmap_user.h ，当解压下载好的 netmap 程序后，在./netmap/sys/net/目录下，本文主要对这两个头文件进行分析。
我们从 netmap_user.h 头文件开始看起。

2. likely()和 unlikely()

这两个宏定义是对编译器做优化的，并不会对变量做什么改变。后面看到这两个宏的调用自动忽略就好了。

#ifndef likely
#define likely(x) __builtin_expect(!!(x), 1)
#define unlikely(x) __builtin_expect(!!(x), 0)
#endif /* likely and unlikely */

3. netmap.h 头文件

netmap.h 被 netmap_user.h 调用，里面定义了一些宏和几个主要的结构体，如 nmreq{}, netmap_if{}, netmap_ring{}, netmap_slot{}。
一个网卡(或者网络接口)只有一个 netmap_if{}结构，在使用 mmap()申请的共享内存中，通过 netmap_if{}结构可以访问到任何一个发送/接收环(也就是 netmap_ring{}结构，一个netmap_if{}可以对应多发送/接收环,这应该和物理硬件有关，我在虚拟机下只有一对环，在真实主机上有两队环)。
找到 netmap_ring{}的地址后，我们就可以找到环中每一个 buffer 的地址(buffer 里面存储的是将要发送/接收的数据包)。后面会讲解这是如何实现的。
通过一个 nifp 是如何访问到多个收/发环的，通过一个 ring 如何找到多个不同的 buffer 地址的，其实都是通过存储这些结构体相邻的后面一部分空间实现。(申请共享内存的时候，这些均已被设计好)

4. 几个重要的宏定义

_NETMAP_OFFSET

#define _NETMAP_OFFSET(type, ptr, offset) \((type)(void *)((char *)(ptr) + (offset)))

解释：该宏定义的作用是将 ptr 指针(强转成 char *类型)向右偏移 offset 个字节，再将其转化为指定的类型 type。

NETMAP_IF

#define NETMAP_IF(_base, _ofs) _NETMAP_OFFSET(struct netmap_if *, _base, _ofs)

解释：该宏定义将_base 指针向右偏移_ofs 个字节后，强转为 netmap_if *类型返回。在 nemap 中通过此宏得到 d->nifp 的地址。

NETMAP_TXRING

#define NETMAP_TXRING(nifp, index) _NETMAP_OFFSET(struct netmap_ring *, \nifp, (nifp)->ring_ofs[index] )

解释：

1.通过该宏定义，可以找到 nifp 的第 index 个发送环的地址(index 是从 0 开始的)， ring_ofs[index]为偏移量，由内核生成。

2.其中，我们注意到 struct netmap_if{}最后面只定义了 const ssize_t ring_ofs[0]，实际上其它的 netmap 环的偏移量都写在了该结构体后面的内存地址里面，直接访问就可以了。

NETMAP_RXRING

#define NETMAP_RXRING(nifp, index) _NETMAP_OFFSET(struct netmap_ring *, \nifp, (nifp)->ring_ofs[index + (nifp)->ni_tx_rings + 1] )

解释：通过该宏定义，可以找到 nifp 的第 index 个接收环的地址，其中(nifp)->ring_ofs[]里面的下标为 index+(nifp)->ni_tx_rings+1，正好与发送环的偏移量区间隔开 1 个。(我想这应该是作者特意设计的)

NETMAP_BUF

#define NETMAP_BUF(ring, index) \((char *)(ring) + (ring)->buf_ofs + ((index)*(ring)->nr_buf_size))

解释：

1.通过该宏定义，可以找到 ring 这个环的第 index 个 buffer 的地址(buffer 里面存的就是我们接收/将发送的完整数据包)，每个 buffer 占的长度是 2048 字节(在(ring)->nr_buf_size也给出了)。

2.其中(ring) ->buf_ofs 是固定的偏移量，不同的环这个值不相同，但所有的(char*)(ring)+(ring)->buf_ofs 会指向同一个地址，也就是存放 buffer 的连续内存的开始地址 (d->buf_start 会指向该地址)。

NETMAP_BUF_IDX

#define NETMAP_BUF_IDX(ring, buf) \( ((char *)(buf) - ((char *)(ring) + (ring)->buf_ofs) ) / \(ring)->nr_buf_size )

解释：在讲 NETMAP_BUF 的时候我们说(char *)(ring) + (ring)->buf_ofs)总会指向存放 buffer 的起始位置(无论是哪一个环)，在这段内存中将第一个 buffer 下标标记为 0 的话， NETMAP_BUF_IDX 计算的恰好是指针 buf 所指 buffer 的下标。

上面几个宏一时没弄懂也没关系，下面调用的时候还会提的。

5. nm_open 函数

1.调用 nm_open 函数时，如：nmr = nm_open(“netmap:eth0”, NULL, 0, NULL); nm_open()会对传递的 ifname 指针里面的字符串进行分析，提取出网络接口名。

2.nm_open() 会对 struct nm_desc *d 申请内存空间，并通过 d->fd =open(NETMAP_DEVICE_NAME, O_RDWR);打开一个特殊的设备/dev/netmap 来创建文件描述符 d->fd。

3.通过 ioctl(d->fd, NIOCREGIF, &d->req)语句，将 d->fd 绑定到一个特殊的接口，并对 d->req 结构体里面的成员做初始化，包括 **a.**在共享内存区域中 nifp 的偏移，**b.**共享区域的大小 nr_memsize，**c.**tx/rx 环的大小 nr_tx_slots/nr_rx_slots(大小为 256)，**d.**tx/rx 环的数量 nr_tx_rings、 nr_rx_rings(视硬件性能而定)等。

4.接着在 if ((!(new_flags & NM_OPEN_NO_MMAP) || parent) && nm_mmap(d, parent))语句中调用 nm_mmap 函数，继续给 d 指针指向的内存赋值。

6. nm_mmap 函数

nm_mmap()源码：

static int nm_mmap(struct nm_desc *d, const struct nm_desc *parent)
{//XXX TODO: check if mmap is already doneif (IS_NETMAP_DESC(parent) && parent->mem && parent->req.nr_arg2 == d->req.nr_arg2){/* do not mmap, inherit from parent */D("do not mmap, inherit from parent");d->memsize = parent->memsize;d->mem = parent->mem;} else{/* XXX TODO: 检查如果想申请的内存太大 (or there is overflow) */d->memsize = d->req.nr_memsize; /* 将需要申请的内存大小赋值给 d->memsize */d->mem = mmap(0, d->memsize, PROT_WRITE | PROT_READ, MAP_SHARED, d->fd, 0); /* 申请共享内存 */if (d->mem == MAP_FAILED){goto fail;}d->done_mmap = 1;}{struct netmap_if *nifp = NETMAP_IF(d->mem, d->req.nr_offset); /*通过 d->req.nr_offset 这个偏移量的到 nifp 的地址，NETMAP_IF 前面说过*/int i;/**for(i=0; i<=2; i++)* printf("ring_ofs[%d]:0x%x\n",i,nifp->ring_ofs[i]); // 这里是我自己加的，为了手动计算收/发环的偏移量*/struct netmap_ring *r = NETMAP_RXRING(nifp,); //对 nifp，找接收包的环 r，因为 index 为 0，所以省略了*(struct netmap_if **) (uintptr_t) &(d->nifp) = nifp; //对d->nifp 赋值，虽然 d->nifp 使用 const 定义的，但对其取地址再强值类型转换后，依然可以对其指向的空间进行操作*(struct netmap_ring **) (uintptr_t) &d->some_ring = r; //同理，对 d->some_ring 进行赋值，此处指向了第一个接受(rx)环。//printf("buf_ofs:0x%x\n", (u_int)r->buf_ofs);*(void **) (uintptr_t) &d->buf_start = NETMAP_BUF(r, 0);//计算第一个 buffer 的地址，并存入 d->buf_start 指针中*(void **) (uintptr_t) &d->buf_end = (char *) d->mem + d->memsize; //计算共享区间的最后一个地址，赋值给 d->buf_end}return 0;fail: return EINVAL;
}

其中：

1.nifp 为申请的共享内存首地址 d->mem 向右偏移 d->req.nr_offset(该值在调用前面的ioctl()时得到)得到。并且一个网络接口(网卡)只对应一个 nifp。(使用宏 NETMAP_IF 计算)

2.得到的nifp 的地址，nifp 结构体里最后定义的 ring_ofs[0]以及接下来内存中的 ring_ofs[1]，ring_ofs[2]…，这些内存中存储的是访问每一个环(tx or rx ring)的偏移量，通过这个偏移量我们可以得到每一个环的地址(使用宏 NETMAP_RXRING/NETMAP_TXRING 进行计算)。

3.得到每个收/发环的地址了，netmap_ring 结构体最后面有一个 struct netmap_slot slot[0];，通过 slot[0],后面内存的 slot[1],slot[2],slot[3]…,取出里面的偏移量就可以得到每一个 buffer(也叫数据包槽)的地址了(使用宏 NETMAP_BUF 计算得到)。到这里，netmap 如何访问到内存槽中的每一个 buffer 的，我们都知道了。实际上 netmap 运行的数据结构就和下图描述的一样：

4.在 struct nm_desc 中，nifp，some_ring，buf_start，buf_end 等指针都定义为 const 的，但我们通过对其取地址再强转指针的方式去往这些指针指向的内存中赋值。

注:在 nm_mmap()中使用 mmap()申请共享的时候，这些数据结构里数据的设计是内核模块就已写好了的，我们在这里其实是在做验证。

7. nm_nextpkt 函数

1.nm_nextpkt()是用来接收网卡上到来的数据包的函数。

2.nm_nextpkt()会将所有 rx 环都检查一遍，当发现有一个 rx 环有需要接收的数据包时，得到这个数据包的地址，并返回。所以 nm_nextpkt()每次只能取一个数据包。

nm_nextpkt()源代码：

static u_char *nm_nextpkt(struct nm_desc *d, struct nm_pkthdr *hdr)
{int ri = d->cur_rx_ring; //当前的接收环的编号do{/* compute current ring to use */struct netmap_ring *ring = NETMAP_RXRING(d->nifp, ri); //得到当前 rx 环的地址if (!nm_ring_empty(ring)) //判断环里是否有新到的包{u_int i = ring->cur; //当前该访问哪个槽(buffer)了u_int idx = ring->slot[i].buf_idx; //得到第 i 个 buffer 的下标//printf("%d\n", idx);u_char *buf = (u_char *) NETMAP_BUF(ring, idx); //得到存有到来数据包的地址// __builtin_prefetch(buf);hdr->ts = ring->ts;hdr->len = hdr->caplen = ring->slot[i].len;ring->cur = nm_ring_next(ring, i); //ring->cur 向后移动一位/* we could postpone advancing head if we want* to hold the buffer. This can be supported in* the future.*/ring->head = ring->cur;d->cur_rx_ring = ri; //将当前环(d->cur_rx_ring)指向第 ri个(因为可能有多个环)。return buf; //将数据包地址返回}ri++;if (ri > d->last_rx_ring) //如果 ri 超过了 rx 环的数量，则再从第一个 rx 环开始检测是否有包到来。ri = d->first_rx_ring;} while (ri != d->cur_rx_ring);return NULL; /* 什么也没发现 */
}

8. nm_inject 函数

1.nm_inject()是用来往共享内存中写入待发送的数据包数据的。数据包经共享内存拷贝到网卡，然后发送出去。所以 nm_inject()是用来发包的。

2.nm_inject()也会查找所有的发送环(tx 环)，找到一个可以发送的槽，就将数据包写入并返回，所以每次函数调用也只能发送一个包。

源代码：

static int nm_inject(struct nm_desc *d, const void *buf, size_t size)
{u_int c, n = d->last_tx_ring - d->first_tx_ring + 1;for (c = 0; c < n; c++){/* 计算当前的环去使用(compute current ring to use) */struct netmap_ring *ring;uint32_t i, idx;uint32_t ri = d->cur_tx_ring + c; //该访问第几个 tx 环了if (ri > d->last_tx_ring) //当超过访问的 tx 环的下标范围时，从头开始访问ri = d->first_tx_ring;ring = NETMAP_TXRING(d->nifp, ri); //得到当前 tx 环的地址if (nm_ring_empty(ring)) //如果当前 tx 环是满的(ring->cur=ring->tail 表示没地方存数据包了)，就跳过{continue;}i = ring->cur; //当前要往哪个槽(槽指向 buffer)中写入数据idx = ring->slot[i].buf_idx; //得到这个槽相对于 buffer 起始地址(d->buf_start)的下标编号ring->slot[i].len = size; //size 为待发送数据包的长度nm_pkt_copy(buf, NETMAP_BUF(ring, idx), size); //将 buf 里存的数据包拷贝给 ring 这个环的第 i 个槽d->cur_tx_ring = ri;ring->head = ring->cur = nm_ring_next(ring, i); //将 head 和 cur 指向下一个槽return size;}return 0; /* 失败 */
}

9. nm_close 函数

1.nm_close 函数就是回收动态内存，回收共享内存，关闭文件描述符什么的了。

源代码：

static int nm_close(struct nm_desc *d)
{/** ugly trick to avoid unused warnings*/static void *__xxzt[] __attribute__ ((unused)) ={ (void *) nm_open, (void *) nm_inject, (void *) nm_dispatch, (void *) nm_nextpkt};if (d == NULL || d->self != d)return EINVAL;if (d->done_mmap && d->mem)munmap(d->mem, d->memsize); //释放申请的共享内存if (d->fd != -1){close(d->fd); //关闭文件描述符}bzero(d, sizeof(*d)); //将 d 指向的空间全部置 0free(d); //释放指针 d 指向的空间return 0;
}

3. NtyTCP 安装

VMWare 编译与调试

1. 添加两个网络适配器

系统启动后，

添加绑定的网卡的 IP 地址，十六进制 IP 地址，网卡相应的 MAC 地址。

代码地址 https://github.com/wangbojing/NtyTcp.git

环境编译，下面以ubuntuserver 版本为例。

先安装 netmap 
Ubuntu 14.04 
https://github.com/wangbojing/netmap.git 
Ubuntu 16.04 
https://github.com/luigirizzo/netmap.git # ./configure 
# make 
# make install进入 ntytcp 的目录
直接 make

4. C10M 的问题

截至目前，40gpbs、32-cores、256G RAM 的 X86 服务器在 Newegg 网站上的报价是几千美元。实际上以这样的硬件配置来看，它完全可以处理 1000 万个以上的并发连接，如果它们不能，那是因为你选择了错误的软件，而不是底层硬件的问题。

可以预见在接下来的 10 年里，因为 IPv6 协议下每个服务器的潜在连接数都是数以百万级的，单机服务器处理数百万的并发连接（甚至千万）并非不可能，但我们需要重新审视目前主流 OS 针对网络编程这一块的具体技术实现。

1、解决 C10M 问题并非不可能

很多人会想当然的认为，要实现 C10M（即单机千万）并发连接和处理能力，是不可能的。不过事实并非如此，现在系统已经在用你可能不熟悉甚至激进的方式支持千万级别的并发连接。

要知道它是如何做到的，我们首先要了解 Errata Security 的 CEO Robert Graham，以及他在 Shmoocon 2013 大会上的“天方夜谈”视频记录： C10M Defending The Internet At Scale（此为 Yutube 视频，你懂的）。

Robert 用一种我以前从未听说的方式来很巧妙地解释了这个问题。他首先介绍了一点有关 Unix 的历史，Unix 的设计初衷并不是一般的服务器操作系统，而是电话网络的控制系统。由于是实际传送数据的电话网络，所以在控制层和数据层之间有明确的界限。问题是我们现在根本不应该使用 Unix 服务器作为数据层的一部分。正如设计只运行一个应用程序的服务器内核，肯定和设计多用户的服务器内核是不同的。

**Robert Graham 的结论是：**OS 的内核不是解决 C10M 问题的办法，恰恰相反 OS 的内核正是导致 C10M 问题的关键所在。

这也就意味着：

不要让 OS 内核执行所有繁重的任务：将数据包处理、内存管理、处理器调度等任务从内核转移到应用程序高效地完成，让诸如 Linux 这样的 OS 只处理控制层，数据层完全交给应用程序来处理。

最终就是要设计这样一个系统，该系统可以处理千万级别的并发连接，它在 200 个时钟周期内处理数据包，在 14 万个时钟周期内处理应用程序逻辑。由于一次主存储器访问就要花费300 个时钟周期，所以这是最大限度的减少代码和缓存丢失的关键。

面向数据层的系统可以每秒处理 1 千万个数据包，面向控制层的系统，每秒只能处理 1 百万个数据包。这似乎很极端，请记住一句老话：可扩展性是专业化的，为了做好一些事情，你不能把性能问题外包给操作系统来解决，你必须自己做。

2、回顾一下 C10K 问题

10 年前，开发人员处理 C10K 可扩展性问题时，尽量避免服务器处理超过 1 万个的并发连接。通过改进操作系统内核以及用事件驱动服务器（典型技术实现如：Nginx 和 Node）代替线程服务器（典型代表：Apache），使得这个问题已经被解决。人们用十年的时间从 Apache 转移到可扩展服务器，在近几年，可扩展服务器的采用率增长得更快了。

以传统网络编程模型作为代表的 Apache 为例，我们来看看它在 C10K 问题上的局限表现在哪些方面，并针对性的讨论对应的解决方法。Apache 的问题在于服务器的性能会随着连接数的增多而变差，实际上性能和可扩展性并不是一回事。当人们谈论规模时，他们往往是在谈论性能，但是规模和性能是不同的，比如 Apache。持续几秒的短期连接：比如快速事务，如果每秒处理 1000 个事务，只能有约 1000 个并发连接到服务器。如果事务延长到 10 秒，要维持每秒 1000 个事务则必须打开 1 万个并发连接。这种情况下：尽管你不顾DoS 攻击，Apache 也会性能陡降，同时大量的下载操作也会使 Apache 崩溃。

如果每秒处理的连接从 5 千增加到 1 万，你会怎么做？比方说，你升级硬件并且提高处理器速度到原来的 2 倍。到底发生了什么？你得到两倍的性能，但你没有得到两倍的处理规模。每秒处理的连接可能只达到了 6000。你继续提高速度，情况也没有改善。甚至 16 倍的性能时，仍然不能处理 1 万个并发连接。所以说性能和可扩展性是不一样的。

问题在于 Apache 会创建一个 CGI 进程，然后关闭，这个步骤并没有扩展。为什么呢？内核使用的 O(N^2)算法使服务器无法处理 1 万个并发连接。

OS 内核中的两个基本问题：

连接数=线程数/进程数：当一个数据包进来，内核会遍历其所有进程以决定由哪个进程来处理这个数据包。
连接数=选择数/轮询次数（单线程）：同样的可扩展性问题，每个包都要走一遭列表上所有的 socket。

通过上述针对 Apache 所表现出的问题，实际上彻底解决并发性能问题的解决方法的根本就是改进 OS 内核使其在常数时间内查找，使线程切换时间与线程数量无关，使用一个新的可扩展 epoll()/IOCompletionPort 常数时间去做 socket 查询。

因为线程调度并没有得到扩展，所以服务器大规模对 socket 使用 epoll 方法，这样就导致需要使用异步编程模式，而这些编程模式正是 Nginx 和 Node 类型服务器具有的。所以当从 Apache 迁移到 Nginx 和 Node 类型服务器时，即使在一个配置较低的服务器上增加连接数，性能也不会突降。所以在处理 C10K 连接时，一台笔记本电脑的速度甚至超过了 16 核的服务器。这也是前一个 10 年解决 C10K 问题的普遍方法。

3、实现 C10M 意味着什么？

实现 10M（即 1 千万）的并发连接挑战意味着什么：

1 千万的并发连接数；
**100 万个连接/秒：**每个连接以这个速率持续约 10 秒；
**10GB/秒的连接：**快速连接到互联网；
**1 千万个数据包/秒：**据估计目前的服务器每秒处理 50K 数据包，以后会更多；
**10 微秒的延迟：**可扩展服务器也许可以处理这个规模（但延迟可能会飙升）；
**10 微秒的抖动：**限制最大延迟；
**并发 10 核技术：**软件应支持更多核的服务器（通常情况下，软件能轻松扩展到四核，服务器可以扩展到更多核，因此需要重写软件，以支持更多核的服务器）。

4、为什么说实现 C10M 的挑战不在硬件而在软件？

1. 理由概述

硬件不是 10M 问题的性能瓶颈所在处，真正的问题出在软件上，尤其是*nux 操作系统。理由如下面这几点：

首先：最初的设计是让 Unix 成为一个电话网络的控制系统，而不是成为一个服务器操作系统。对于控制系统而言，针对的主要目标是用户和任务，而并没有针对作为协助功能的数据处理做特别设计，也就是既没有所谓的快速路径、慢速路径，也没有各种数据服务处理的优先级差别。

其次：传统的 CPU，因为只有一个核，操作系统代码以多线程或多任务的形式来提升整体性能。而现在，4 核、8 核、32 核、64 核和 100 核，都已经是真实存在的 CPU 芯片，如何提高多核的性能可扩展性，是一个必须面对的问题。比如让同一任务分割在多个核心上执行，以避免 CPU 的空闲浪费，当然，这里面要解决的技术点有任务分割、任务同步和异步等。

再次：核心缓存大小与内存速度是一个关键问题。现在，内存已经变得非常的便宜，随便一台普通的笔记本电脑，内存至少也就是 4G 以上，高端服务器的内存上 24G 那是相当的平常。但是，内存的访问速度仍然很慢，CPU 访问一次内存需要约 60~100 纳秒，相比很久以前的内存访问速度，这基本没有增长多少。对于在一个带有 1GHZ 主频 CPU 的电脑硬件里，如果要实现 10M 性能，那么平均每一个包只有 100 纳秒，如果存在两次 CPU 访问内存，那么 10M 性能就达不到了。核心缓存，也就是 CPU L1/L2/LL Cache，虽然访问速度会快些，但大小仍然不够，我之前接触到的高端至强，LLC 容量大小貌似也就是 12M。

2. 解决思路

解决这些问题的关键在于如何将功能逻辑做好恰当的划分，比如专门负责控制逻辑的控制面和专门负责数据逻辑的数据面。数据面专门负责数据的处理，属于资源消耗的主要因素，压力巨大，而相比如此，控制面只负责一些偶尔才有非业务逻辑，比如与外部用户的交互、信息的统计等等。我之前接触过几种网络数据处理框架，比如 Intel 的 DPDK、6wind、windriver，它们都针对 Linux 系统做了特别的补充设计，增加了数据面、快速路径等等特性，其性能的提升自然是相当巨大。

看一下这些高性能框架的共同特点：

数据包直接传递到业务逻辑：

而不是经过 Linux 内核协议栈。这是很明显的事情，因为我们知道，Linux 协议栈是复杂和繁琐的，数据包经过它无非会导致性能的巨大下降，并且会占用大量的内存资源，之前有同事测试过，Linux 内核要吃掉 2.5KB 内存/socket。我研究过很长一段时间的 DPDK 源码，其提供的 82576 和 82599 网卡驱动就直接运行在应用层，将接管网卡收到的数据包直接传递到应用层的业务逻辑里进行处理，而无需经过 Linux 内核协议栈。当然，发往本服务器的非业务逻辑数据包还是要经过 Linux 内核协议栈的，比如用户的 SSH 远程登录操作连接等。

多线程的核间绑定：

一个具有 8 核心的设备，一般会有 1 个控制面线程和 7 个或 8 个数据面线程，每一个线程绑定到一个处理核心（其中可能会存在一个控制面线程和一个数据面线程都绑定到同一个处理核心的情况）。这样做的好处是最大化核心 CACHE 利用、实现无锁设计、避免进程切换消耗等等。

内存是另外一个核心要素：

常见的内存池设计必须在这里得以切实应用。有几个考虑点，首先，可以在 Linux 系统启动时把业务所需内存直接预留出来，脱离 Linux 内核的管理。其次，Linux 一般采用 4K每页，而我们可以采用更大内存分页，比如 2M，这样能在一定程度上减少地址转换等的性能消耗。

3. 关于 Intel 的 DPDK 框架/ Netmap 开源框架

随着网络技术的不断创新和市场的发展，越来越多的网络设备基础架构开始向基于通用处理器平台的架构方向融合，期望用更低的成本和更短的产品开发周期来提供多样的网络单元和丰富的功能，如应用处理、控制处理、包处理、信号处理等。为了适应这一新的产业趋势，Intel 推出了基于 Intel x86 架构 DPDK (Data Plane Development Kit，数据平面开发套件) 实现了高效灵活的包处理解决方案。经过近 6 年的发展，DPDK 已经发展成支持多种高性能网卡和多通用处理器平台的开源软件工具包。

5、解决 C10M 问题的思路总结

综上所述，解决 C10M 问题的关键主要是从下面几个方面入手：

网卡问题

网卡问题：通过内核工作效率不高

解决方案：使用自己的驱动程序并管理它们，使适配器远离操作系统。

CPU 问题

CPU 问题：使用传统的内核方法来协调你的应用程序是行不通的。

解决方案：Linux 管理前两个 CPU，你的应用程序管理其余的 CPU，中断只发生在你允许的CPU 上。

内存问题

内存问题：内存需要特别关注，以求高效。

解决方案：在系统启动时就分配大部分内存给你管理的大内存页。

以 Linux 为例，解决的思路就是将控制层交给 Linux，应用程序管理数据。应用程序与内核之间没有交互、没有线程调度、没有系统调用、没有中断，什么都没有。然而，你有的是在 Linux 上运行的代码，你可以正常调试，这不是某种怪异的硬件系统，需要特定的工程师。你需要定制的硬件在数据层提升性能，但是必须是在你熟悉的编程和开发环境上进行。

Epoll与协议栈

send() 详解

应用程序调用 send() 操作返回>0 ，说明仅仅是把用户程序内存copy到内核协议栈发送缓存区（注意与mov的区别，mov是是转移操作，不复制），无法保证数据已经发送对端。
应用程序调用 send() 操作返回 -1 ，说明内核协议栈缓存满了，暂时无法copy更多数据到内核协议栈发送缓存区，需要等到内核协议栈缓存有空余空间（即内核协议栈缓存区已发送一部分数据出去），才能继续调用 send()。
如果多个线程同时调用 send() ，会导致内核协议栈发送缓存区内的数据混乱，即 send() 操作是非线程安全的。解决方法是通过epoll管理fd，保证某一时刻只能有一个线程调用 send()，类似于加锁思想。

recv() 详解

如果fd是非阻塞（setnoblock），应用程序调用 recv() 返回-1 ，说明目前内核协议栈接收缓存区暂时无数据。

tcp粘包与分包

协议头加入识别长度的域（推荐）
分隔符
发送定长的包

TCP四次挥手详解

TCP四次挥手与业务代码关系

客户端主动调用close()请求关闭连接，发送FIN包给服务器。服务器接收到FIN包后，recv()函数返回0。服务器发送ACK包给客户端，并且调用close()请求关闭连接。客户端收到服务器的FIN包，发送ACK给服务器，并且后进入time_wait状态，开启定时器，在这段时间内如果没有收到服务器的FIN包，意味着服务器已经收到了我们的ACK包，服务器进入了CLOSED状态，那么客户端也可以进入CLOSED状态。

常见TCP四次挥手问题

1.为什么是四次挥手？

a. 第一次客户端发送 FIN 报文后，只是代表客户端没有数据发给服务端，但是还有接收数据能力

b. 服务端收到 FIN 报文后，可能还有数据要发送给客户端，只能先回复 ACK 报文应答

c. 当服务端不再有数据要发送时，才发送 FIN 报文给你客户端，表示可以关闭

d. 客户端收到服务端 FIN 报文要做一次应答，所以一共就四次

2.为什么服务端收到FIN请求，不直接回复FIN+ACK包，而是先发送ACK, 后面再发送FIN?

答：不直接回复FIN+ACK是因为，客户端要断开链接了，不表示服务端没有数据继续发送出去。如果直接发送FIN+ACK, 服务端就不能把剩下的数据包发送出去。等数据包发送完了，服务端才能发送FIN包，表示自己也要关闭连接了。

3.出现大量close_wait状态的原因：

查看：

netstat -anop | grep 6666

原因：首先close_wait状态代表本机是作为tcp服务端的一方。因此最大的可能是close()调用不及时。业务代码中，服务器管理大量的tcp连接fd，可能与其他字段组成结构体一起管理，当断开tcp连接时，可能没调用close()之间就将fd置空，这导致了无法有效释放tcp连接，停留在close_wait状态。

解决：a.及时调用close()关闭tcp连接。b.把业务清理掉，做异步处理。

4.出现大量fin_wait_2状态的原因：

查看：

netstat -anop | grep 6666

原因：首先fin_wait_2状态代表本机是作为tcp客户端的一方。因此最大的可能是对方服务端没有及时调用close()关闭连接并且发FIN包给我。

解决：a.需要服务端处理。b.自己kill掉进程。

5.出现大量time_wait状态的原因：

查看：

netstat -anop | grep 6666

原因：首先time_wait状态代表本机是作为tcp客户端的一方。因此最大的可能是自己有很多主动断开tcp连接的操作。最有可能是本机和数据库服务器是短连接状态，查询完语句就断开操作。

解决：a.与服务器建立长连接。

6.设置SO_REUSEADDR会把time_wait取消掉吗？

答：不会，只是time_wait结束后，这个tcb块不会被操作系统回收，而是等有新连接的时候，重新用这个tcb。

7.双方同时调用close怎么办？

8.服务端的fin比服务端的ack先到客户端怎么办？

EPOLL设计思路

epoll具体实现

epoll具体数据结构
epoll如何线程安全
如何与协议栈合作
ET和LT的具体实现

EPOLL数据结构思考

服务器如何管理那么多与客户端的tcp连接fd？首先明确一下两点： a. 所有的fd只有两个状态，空闲与就绪 b. fd中，就绪状态是很少的，大部分处于空闲状态

针对a，我们应该用两个数据结构分类fd：

一个数据结构装空闲的fd list。
一个数据结构装就绪的fd list。

接下来思考，空闲fd用什么数据结构来存储？

(1) hash结构？不行，因为hash结构一开始就需要固定下来，不适合经常变动数量的情况。

(2) b/b+树？不行，b/b+树的高度，太低，查找效率不算高。

(3) 跳表？不行，跳表有太多冗余数据结构只为了用于提高查找效率，如果节点数量<5000的情况下不推荐用这种空间换时间的做法。

(4) 红黑树？可以，节点数量在500这个量级比较适合。

接下来思考，就绪fd用什么数据结构来存储？

答：可以选择用链式结构的队列。

为什么用链式结构，是因为数量少吗？不全是，主要是因为就绪的fd业务层需要遍历！

为什么用队列？链式结构有队列和栈，为什么不用栈结构

因为epoll_wait(epfd, events, length, timeout) 返回的events 可能不是一次取出所有的就绪fd，下次再次调用这个接口继续上次的位置，因此用队列结构合适。

EPOLL线程安全思考

epoll需要考虑线程安全吗？当然需要！因为很可能有多个线程同时操作同一个epollfd。

那么如何保证epoll的线程安全？加锁！加锁的对象是？

epoll用于存储所有fd的红黑树
epoll用于存储所有就绪fd的就绪队列

对红黑树的加锁有两种方案：

锁整颗树
锁子树

目前推荐做法是锁整棵树，虽然粒度更大，但是操作时间相差不远而且方便操作。用的锁是互斥锁！为什么不用自旋锁呢？因为如果红黑树较大，对红黑树的操作耗时比较长，因此不推荐自旋锁和读写锁。

对就绪队列的加锁有两种方案：

自旋锁
cas

目前用的比较多是自旋锁。

EPOLL与协议栈怎么一起工作思考

三次握手完成的时候，服务器把fd添加到epoll管理。
recvbuffer有数据的时候，epoll把fd添加到就绪队列并通知应用程序读取。
sendbuffer有空间的时候，epoll把fd添加到就绪队列并通知应用程序可以发数据。
接收到FIN包的时候，服务器把fd从epoll中删除，不再管理。

一些关于EPOLL的误区：

epoll比select/poll效率要高：

错误的，当数量级<500的时候，select/poll比epoll效率更高。数量级>500或1024时，epoll效率更高。

ET/LT是怎么实现的：

其实就是fd对应的内核发送缓冲区和接收缓冲区，当检测到缓冲区有数据，就会一直触发ET，如果缓冲区从无数据到有数据，就会触发LT。

内核EPOLL实现的源码位置：

EPOLL 的实现原理

Epoll 是 Linux IO 多路复用的管理机制。作为现在 Linux 平台高性能网络 IO 必要的组件。内核的实现可以参照：fs/eventpoll.c .

为什么需要自己实现 epoll 呢？现在自己打算做一个用户态的协议栈。采用单线程的模式。 https://github.com/wangbojing/NtyTcp，至于为什么要实现用户态协议栈？可以自行百度C10M 的问题。

由于协议栈做到了用户态故需要自己实现高性能网络 IO 的管理。所以 epoll 就自己实现一下。代码：NtyTcp/src/nty_epoll_rb.c at master · wangbojing/NtyTcp · GitHub

在实现 epoll 之前，先得好好理解内核 epoll 的运行原理。内核的 epoll 可以从四方面来理解。

1. Epoll 的数据结构，rbtree 对<fd, event>的存储，ready 队列存储就绪 io。

2. Epoll 的线程安全，SMP 的运行，以及防止死锁。

3. Epoll 内核回调。

4. Epoll 的 LT（水平触发）与 ET（边沿触发）

下面从这四个方面来实现 epoll。

一、 Epoll 数据结构

Epoll 主要由两个结构体：eventpoll 与 epitem。Epitem 是每一个 IO 所对应的的事件。比如 epoll_ctl EPOLL_CTL_ADD 操作的时候，就需要创建一个 epitem。Eventpoll 是每一个 epoll 所对应的的。比如 epoll_create 就是创建一个 eventpoll。

Epitem 的定义

Eventpoll 的定义

数据结构如下图所示。

List 用来存储准备就绪的 IO。对于数据结构主要讨论两方面：insert 与 remove。同样如此，对于 list 我们也讨论 insert 与 remove。何时将数据插入到 list 中呢？当内核 IO 准备就绪的时候，则会执行 epoll_event_callback 的回调函数，将 epitem 添加到 list 中。

那何时删除 list 中的数据呢？当 epoll_wait 激活重新运行的时候，将 list 的 epitem 逐一 copy 到 events 参数中。

Rbtree 用来存储所有 io 的数据，方便快速通 io_fd 查找。也从 insert 与 remove 来讨论。对于 rbtree 何时添加：当 App 执行 epoll_ctl EPOLL_CTL_ADD 操作，将 epitem 添加到 rbtree 中。何时删除呢？当 App 执行 epoll_ctl EPOLL_CTL_DEL 操作，将 epitem 添加到 rbtree 中。

List 与 rbtree 的操作又如何做到线程安全，SMP，防止死锁呢？

二、 Epoll 锁机制

Epoll 从以下几个方面是需要加锁保护的。List 的操作，rbtree 的操作，epoll_wait 的等待。

List 使用最小粒度的锁 spinlock，便于在 SMP 下添加操作的时候，能够快速操作 list。

List 添加

346 行：获取 spinlock。

347 行：epitem 的 rdy 置为 1，代表 epitem 已经在就绪队列中，后续再触发相同事件就只需更改 event。

348 行：添加到 list 中。

349 行：将 eventpoll 的 rdnum 域加 1。

350 行：释放 spinlock

List 删除

301 行：获取 spinlock

304 行：判读 rdnum 与 maxevents 的大小，避免 event 溢出。

307 行：循环遍历 list，判断添加 list 不能为空

309 行：获取 list 首个结点

310 行：移除 list 首个结点。

311 行：将 epitem 的 rdy 域置为 0，标识 epitem 不再就绪队列中。

313 行：copy epitem 的 event 到用户空间的 events。

316 行：copy 数量加 1

317 行：eventpoll 中 rdnum 减一。

避免 SMP 体系下，多核竞争。此处采用自旋锁，不适合采用睡眠锁。

Rbtree 的添加

149 行：获取互斥锁。

153 行：查找 sockid 的 epitem 是否存在。存在则不能添加，不存在则可以添加。

160 行：分配 epitem。

167 行：sockid 赋值

168 行：将设置的 event 添加到 epitem 的 event 域。

170 行：将 epitem 添加到 rbrtree 中。

173 行：释放互斥锁。

Rbtree 删除：

177 行：获取互斥锁。

181 行：删除 sockid 的结点，如果不存在，则 rbtree 返回-1。

188 行：释放 epitem

190 行：释放互斥锁。

Epoll_wait 的挂起。

采用 pthread_cond_wait，具体实现可以参照。

https://github.com/wangbojing/NtyTcp/blob/master/src/nty_epoll_rb.c

三、 Epoll 回调

Epoll 的回调函数何时执行，此部分需要与 Tcp 的协议栈一起来阐述。Tcp 协议栈的时序图如下图所示，epoll 从协议栈回调的部分从下图的编号 1,2,3,4。具体 Tcp 协议栈的实现，后续从另外的文章中表述出来。下面分别对四个步骤详细描述

编号 1：是 tcp 三次握手，对端反馈 ack 后，socket 进入 rcvd 状态。需要将监听 socket 的 event 置为 EPOLLIN，此时标识可以进入到 accept 读取 socket 数据。

编号 2：在 established 状态，收到数据以后，需要将 socket 的 event 置为 EPOLLIN 状态。

编号 3：在 established 状态，收到 fin 时，此时 socket 进入到 close_wait。需要 socket 的 event 置为 EPOLLIN。读取断开信息。

编号 4：检测 socket 的 send 状态，如果对端 cwnd>0 是可以，发送的数据。故需要将 socket 置为 EPOLLOUT。

所以在此四处添加 EPOLL 的回调函数，即可使得 epoll 正常接收到 io 事件。

四、 LT 与 ET

LT（水平触发）与 ET（边沿触发）是电子信号里面的概念。不清楚可以 man epoll 查看的。

如下图所示：

比如：event = EPOLLIN | EPOLLLT，将 event 设置为 EPOLLIN 与水平触发。只要 event 为 EPOLLIN 时就能不断调用 epoll 回调函数。

比如: event = EPOLLIN | EPOLLET，event 如果从 EPOLLOUT 变化为 EPOLLIN 的时候，就会触发。在此情形下，变化只发生一次，故只调用一次 epoll 回调函数。关于水平触发与边沿触发放在 epoll 回调函数执行的时候，如果为 EPOLLET（边沿触发），与之前的 event 对比，如果发生改变则调用 epoll 回调函数，如果为 EPOLLLT（水平触发），则查看 event 是否为 EPOLLIN，即可调用 epoll 回调函数。