◼ ACK机制
◼ 重传机制
◼ 序号机制 3 2 1 -》2 3 1
◼ 重排机制 2 3 1 ->3 2 1
◼ 窗口机制
Tcp不用我们管
可靠性udp 5种机制都需要用户层处理
以10%-20%带宽浪费的代价换取了比 TCP快30%-40%的传输速度。
RTO翻倍vs不翻倍:
TCP超时计算是RTOx2,这样连续丢三次包就变成RTOx8了,十分恐怖,而 KCP启动快速模式后不x2,只是x1.5(实验证明1.5这个值相对比较好), 提高了传输速度。 200 300 450 675 – 200 400 800 1600
选择性重传 vs 全部重传:
TCP丢包时会全部重传从丢的那个包开始以后的数据,KCP是选择性重传, 只重传真正丢失的数据包。
快速重传(跳过多少个包马上重传)(如果使用了快速重传,可以不考虑 RTO)): 发送端发送了1,2,3,4,5几个包,然后收到远端的ACK: 1, 3, 4, 5,当收到ACK3时,KCP知道2被跳过1次,收到ACK4时,知道2被跳过了2次,此时可以认为2号丢失,不用等超时,直接重传2号包,大大改善了丢包时的传输速度。 fastresend =2
延迟ACK vs 非延迟ACK:
TCP为了充分利用带宽,延迟发送ACK(NODELAY都没用),这样超时计算会算出较大 RTT时间,延长了丢包时的判断过程。KCP的ACK是否延迟发送可以调节。
UNA vs ACK+UNA:
ARQ模型响应有两种,UNA(此编号前所有包已收到,如TCP)和ACK(该编号包已收到),光用UNA将导致全部重传,光用ACK则丢失成本太高,以往协议都是二选其一,而 KCP协议中,除去单独的 ACK包外,所有包都有UNA信息。
非退让流控:
KCP正常模式同TCP一样使用公平退让法则,即发送窗口大小由:发送缓存大小、接收端剩余接收缓存大小、丢包退让及慢启动这四要素决定。但传送及时性要求很高的小数据时,可选择通过配置跳过后两步,仅用前两项来控制发送频率。
以牺牲部分公平性及带宽利用率之代价,换取了开着BT都能流畅传输的效果。
◼ kcp官方:https://github.com/skywind3000/kcp
◼ 名词说明
真正发送数据需要调用sendto
ikcp_input(kcp,received_udp_packet,received_udp_size);
我们要使用recvfrom接收,然后扔到kcp里面做解析
conv:连接号。UDP是无连接的,conv用于表示来自于哪个客户端。对连接的一种替代
cmd:命令字。如,
IKCP_CMD_ACK确认命令,
IKCP_CMD_WASK接收窗口大小询问命令,
IKCP_CMD_WINS接收窗口大小告知命令,
frg:分片,用户数据可能会被分成多个KCP包,发送出去
wnd:接收窗口大小,发送方的发送窗口不能超过接收方给出的数值
ts:时间序列
sn:序列号
una:下一个可接收的序列号。其实就是确认号,收到 sn=10的包,una为11
len:数据长度
data:用户数据
RTO计算(与TCP完全一样)
RTT:一个报文段发送出去,到收到对应确认包的时间差。
SRTT(kcp->rx_srtt):RTT的一个加权RTT平均值,平滑值。
RTTVAR(kcp->rx_rttval):RTT的平均偏差,用来衡量
RTT的抖动。
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include "ikcp.h"
#define RECV_BUF 1500
static int number = 0;
typedef struct
{
unsigned char *ipstr;
int port;
ikcpcb *pkcp;
int sockfd;
struct sockaddr_in addr; //存放服务器信息的结构体
struct sockaddr_in CientAddr; //存放客户机信息的结构体
char buff[RECV_BUF]; //存放收发的消息
} kcpObj;
// 编译: gcc -o server server.c ikcp.c
// 特别需要注意,这里的服务器端也只能一次使用,即是等客户端退出后,服务端也要停止掉再启动
// 之所以是这样,主要是因为sn的问题,比如客户端第一次启动 sn 0~5, 第二次启动发送的sn还是0 ~5 如果服务器端不停止则自己以为0~5已经收到过了就不会回复。
// 在真正使用的时候,还需要另外的通道让客户端和服务器端之前重新创建ikcpcb,以匹配ikcpcb的conv
/* get system time */
void itimeofday(long *sec, long *usec)
{
#if defined(__unix)
struct timeval time;
gettimeofday(&time, NULL);
if (sec)
*sec = time.tv_sec;
if (usec)
*usec = time.tv_usec;
#else
static long mode = 0, addsec = 0;
BOOL retval;
static IINT64 freq = 1;
IINT64 qpc;
if (mode == 0)
{
retval = QueryPerformanceFrequency((LARGE_INTEGER *)&freq);
freq = (freq == 0) ? 1 : freq;
retval = QueryPerformanceCounter((LARGE_INTEGER *)&qpc);
addsec = (long)time(NULL);
addsec = addsec - (long)((qpc / freq) & 0x7fffffff);
mode = 1;
}
retval = QueryPerformanceCounter((LARGE_INTEGER *)&qpc);
retval = retval * 2;
if (sec)
*sec = (long)(qpc / freq) + addsec;
if (usec)
*usec = (long)((qpc % freq) * 1000000 / freq);
#endif
}
/* get clock in millisecond 64 */
IINT64 iclock64(void)
{
long s, u;
IINT64 value;
itimeofday(&s, &u);
value = ((IINT64)s) * 1000 + (u / 1000);
return value;
}
IUINT32 iclock()
{
return (IUINT32)(iclock64() & 0xfffffffful);
}
int64_t first_recv_time = 0;
/* sleep in millisecond */
void isleep(unsigned long millisecond)
{
#ifdef __unix /* usleep( time * 1000 ); */
struct timespec ts;
ts.tv_sec = (time_t)(millisecond / 1000);
ts.tv_nsec = (long)((millisecond % 1000) * 1000000);
/*nanosleep(&ts, NULL);*/
usleep((millisecond << 10) - (millisecond << 4) - (millisecond << 3));
#elif defined(_WIN32)
Sleep(millisecond);
#endif
}
int udp_output(const char *buf, int len, ikcpcb *kcp, void *user)
{
kcpObj *send = (kcpObj *)user;
//发送信息
int n = sendto(send->sockfd, buf, len, 0, (struct sockaddr *)&send->CientAddr, sizeof(struct sockaddr_in));
if (n >= 0)
{
//会重复发送,因此牺牲带宽
printf("send: %d bytes, t:%lld\n", n, iclock64() - first_recv_time); //24字节的KCP头部
return n;
}
else
{
printf("error: %d bytes send, error\n", n);
return -1;
}
}
int init(kcpObj *send)
{
send->sockfd = socket(AF_INET, SOCK_DGRAM, 0);
if (send->sockfd < 0)
{
perror("socket error!");
exit(1);
}
bzero(&send->addr, sizeof(send->addr));
send->addr.sin_family = AF_INET;
send->addr.sin_addr.s_addr = htonl(INADDR_ANY); //INADDR_ANY
send->addr.sin_port = htons(send->port);
printf("服务器socket: %d port:%d\n", send->sockfd, send->port);
if (send->sockfd < 0)
{
perror("socket error!");
exit(1);
}
if (bind(send->sockfd, (struct sockaddr *)&(send->addr), sizeof(struct sockaddr_in)) < 0)
{
perror("bind");
exit(1);
}
}
void loop(kcpObj *send)
{
unsigned int len = sizeof(struct sockaddr_in);
int n, ret;
//接收到第一个包就开始循环处理
int recv_count = 0;
isleep(1);
ikcp_update(send->pkcp, iclock());
char buf[RECV_BUF] = {0};
while (1)
{
isleep(1);
ikcp_update(send->pkcp, iclock());
//处理收消息
n = recvfrom(send->sockfd, buf, RECV_BUF, MSG_DONTWAIT, (struct sockaddr *)&send->CientAddr, &len);
if (n > 0)
{
printf("UDP recv[%d] size= %d \n", recv_count++, n);
if (first_recv_time == 0)
{
first_recv_time = iclock64();
}
//预接收数据:调用ikcp_input将裸数据交给KCP,这些数据有可能是KCP控制报文,并不是我们要的数据。
//kcp接收到下层协议UDP传进来的数据底层数据buffer转换成kcp的数据包格式
ret = ikcp_input(send->pkcp, buf, n);
if (ret < 0)
{
continue;
}
//kcp将接收到的kcp数据包还原成之前kcp发送的buffer数据
ret = ikcp_recv(send->pkcp, buf, n); //从 buf中 提取真正数据,返回提取到的数据大小
if (ret < 0)
{ // 没有检测ikcp_recv提取到的数据
isleep(1);
continue;
}
int send_size = ret;
//ikcp_send只是把数据存入发送队列,没有对数据加封kcp头部数据
//应该是在kcp_update里面加封kcp头部数据
//ikcp_send把要发送的buffer分片成KCP的数据包格式,插入待发送队列中。
ret = ikcp_send(send->pkcp, buf, send_size);
printf("Server reply -> bytes[%d], ret = %d\n", send_size, ret);
ikcp_flush(send->pkcp); // 快速flush一次 以更快让客户端收到数据
number++;
}
else if (n == 0)
{
printf("finish loop\n");
break;
}
else
{
// printf("n:%d\n", n);
}
}
}
int main(int argc, char *argv[])
{
printf("this is kcpServer\n");
if (argc < 2)
{
printf("请输入服务器端口号\n");
return -1;
}
kcpObj send;
send.port = atoi(argv[1]);
send.pkcp = NULL;
bzero(send.buff, sizeof(send.buff));
char Msg[] = "Server:Hello!"; //与客户机后续交互
memcpy(send.buff, Msg, sizeof(Msg));
ikcpcb *kcp = ikcp_create(0x1, (void *)&send); //创建kcp对象把send传给kcp的user变量
ikcp_setmtu(kcp, 1400);
kcp->output = udp_output; //设置kcp对象的回调函数
ikcp_nodelay(kcp, 0, 10, 0, 0); //1, 10, 2, 1
ikcp_wndsize(kcp, 128, 128);
send.pkcp = kcp;
init(&send); //服务器初始化套接字
loop(&send); //循环处理
return 0;
}
#include
#include
#include
#include
#include
#include
#include
#include
#include "ikcp.h"
#include
#include
#include
#include "delay.h"
#define DELAY_TEST2_N 5
#define UDP_RECV_BUF_SIZE 1500
// 编译: gcc -o client client.c ikcp.c delay.c -lpthread
typedef struct {
unsigned char *ipstr;
int port;
ikcpcb *pkcp;
int sockfd;
struct sockaddr_in addr;//存放服务器的结构体
char buff[UDP_RECV_BUF_SIZE];//存放收发的消息
}kcpObj;
/* sleep in millisecond */
void isleep(unsigned long millisecond)
{
#ifdef __unix /* usleep( time * 1000 ); */
struct timespec ts;
ts.tv_sec = (time_t)(millisecond / 1000);
ts.tv_nsec = (long)((millisecond % 1000) * 1000000);
/*nanosleep(&ts, NULL);*/
usleep((millisecond << 10) - (millisecond << 4) - (millisecond << 3));
#elif defined(_WIN32)
Sleep(millisecond);
#endif
}
int udp_output(const char *buf, int len, ikcpcb *kcp, void *user){
// printf("使用udp_output发送数据\n");
kcpObj *send = (kcpObj *)user;
//发送信息
int n = sendto(send->sockfd, buf, len, 0,(struct sockaddr *) &send->addr,sizeof(struct sockaddr_in));//【】
if (n >= 0)
{
//会重复发送,因此牺牲带宽
printf("send:%d bytes\n", n);//24字节的KCP头部
return n;
}
else
{
printf("udp_output: %d bytes send, error\n", n);
return -1;
}
}
int init(kcpObj *send)
{
send->sockfd = socket(AF_INET,SOCK_DGRAM,0);
if(send->sockfd < 0)
{
perror("socket error!");
exit(1);
}
bzero(&send->addr, sizeof(send->addr));
//设置服务器ip、port
send->addr.sin_family=AF_INET;
send->addr.sin_addr.s_addr = inet_addr((char*)send->ipstr);
send->addr.sin_port = htons(send->port);
printf("sockfd = %d ip = %s port = %d\n",send->sockfd,send->ipstr,send->port);
}
// 特别说明,当我们使用kcp测试rtt的时候,如果发现rtt过大,很大一种可能是分片数据没有及时发送出去,需要调用ikcp_flush更快速将分片发送出去。
void delay_test2(kcpObj *send) {
// 初始化 100个 delay obj
char buf[UDP_RECV_BUF_SIZE];
unsigned int len = sizeof(struct sockaddr_in);
size_t obj_size = sizeof(t_delay_obj);
t_delay_obj *objs = malloc(DELAY_TEST2_N * sizeof(t_delay_obj));
int ret = 0;
int recv_objs = 0;
//ikcp_update包含ikcp_flush,ikcp_flush将发送队列中的数据通过下层协议UDP进行发送
ikcp_update(send->pkcp,iclock());//不是调用一次两次就起作用,要loop调用
for(int i = 0; i < DELAY_TEST2_N; i++) {
// isleep(1);
delay_set_seqno_send_time(&objs[i], i);
ret = ikcp_send(send->pkcp, (char *) &objs[i], obj_size);
if(ret < 0) {
printf("send %d seqno:%u failed, ret:%d, obj_size:%ld\n", i, objs[i].seqno, ret, obj_size);
return;
}
// ikcp_flush(send->pkcp); // 调用flush能更快速把分片发送出去
//ikcp_update包含ikcp_flush,ikcp_flush将发送队列中的数据通过下层协议UDP进行发送
ikcp_update(send->pkcp,iclock());//不是调用一次两次就起作用,要loop调用
int n = recvfrom(send->sockfd, buf, UDP_RECV_BUF_SIZE, MSG_DONTWAIT,(struct sockaddr *) &send->addr,&len);
// printf("print recv1:%d\n", n);
if(n < 0) {//检测是否有UDP数据包
// isleep(1);
continue;
}
ret = ikcp_input(send->pkcp, buf, n); // 从 linux api recvfrom先扔到kcp引擎
if(ret < 0)//检测ikcp_input是否提取到真正的数据
{
//printf("ikcp_input ret = %d\n",ret);
continue; // 没有读取到数据
}
ret = ikcp_recv(send->pkcp, (char *)&objs[i], obj_size);
if(ret < 0)//检测ikcp_recv提取到的数据
{
printf("ikcp_recv1 ret = %d\n",ret);
continue;
}
delay_set_recv_time(&objs[recv_objs]);
recv_objs++;
printf("recv1 %d seqno:%d, ret:%d\n", recv_objs, objs[i].seqno, ret);
if(ret != obj_size) {
printf("recv1 %d seqno:%d failed, size:%d\n", i, objs[i].seqno, ret);
delay_print_rtt_time(objs, i);
return;
}
}
// 还有没有发送完毕的数据
for(int i = recv_objs; i < DELAY_TEST2_N; ) {
// isleep(1);
//ikcp_update包含ikcp_flush,ikcp_flush将发送队列中的数据通过下层协议UDP进行发送
ikcp_update(send->pkcp,iclock());//不是调用一次两次就起作用,要loop调用
//ikcp_flush(send->pkcp); // 调用flush能更快速把分片发送出去
int n = recvfrom(send->sockfd, buf, UDP_RECV_BUF_SIZE, MSG_DONTWAIT,(struct sockaddr *) &send->addr,&len);
// printf("recv2:%d\n", n);
if(n < 0) {//检测是否有UDP数据包
printf("recv2:%d\n", n);
isleep(1);
continue;
}
ret = ikcp_input(send->pkcp, buf, n);
if(ret < 0)//检测ikcp_input是否提取到真正的数据
{
printf("ikcp_input2 ret = %d\n",ret);
continue; // 没有读取到数据
}
ret = ikcp_recv(send->pkcp, (char *)&objs[i], obj_size);
if(ret < 0)//检测ikcp_recv提取到的数据
{
printf("ikcp_recv2 ret = %d\n",ret);
continue;
}
printf("recv2 %d seqno:%d, ret:%d\n", recv_objs, objs[i].seqno, ret);
delay_set_recv_time(&objs[recv_objs]);
recv_objs++;
i++;
if(ret != obj_size) {
printf("recv2 %d seqno:%d failed, size:%d\n", i, objs[i].seqno, ret);
delay_print_rtt_time(objs, i);
return;
}
}
ikcp_flush(send->pkcp);
delay_print_rtt_time(objs, DELAY_TEST2_N);
}
void loop(kcpObj *send)
{
unsigned int len = sizeof(struct sockaddr_in);
int n,ret;
// while(1)
{
isleep(1);
delay_test2(send);
}
printf("loop finish\n");
close(send->sockfd);
}
int main(int argc,char *argv[])
{
//printf("this is kcpClient,请输入服务器 ip地址和端口号:\n");
if(argc != 3)
{
printf("请输入服务器ip地址和端口号\n");
return -1;
}
printf("this is kcpClient\n");
unsigned char *ipstr = (unsigned char *)argv[1];
unsigned char *port = (unsigned char *)argv[2];
kcpObj send;
send.ipstr = ipstr;
send.port = atoi(argv[2]);
init(&send);//初始化send,主要是设置与服务器通信的套接字对象
bzero(send.buff,sizeof(send.buff));
// 每个连接都是需要对应一个ikcpcb
ikcpcb *kcp = ikcp_create(0x1, (void *)&send);//创建kcp对象把send传给kcp的user变量
kcp->output = udp_output;//设置kcp对象的回调函数
ikcp_nodelay(kcp,0, 10, 0, 0);//(kcp1, 0, 10, 0, 0); 1, 10, 2, 1
ikcp_wndsize(kcp, 128, 128);
ikcp_setmtu(kcp, 1400);
send.pkcp = kcp;
loop(&send);//循环处理
ikcp_release(send.pkcp);
printf("main finish\n");
return 0;
}
https://www.yuque.com/docs/share/01d83f75-0fd5-4c9f-976a-0dfcf417e0cc?#
《阿里XQUIC:标准QUIC实现自研之路》