目录
- 系列文章
- unix网络编程1.1——TCP协议详解（一）
- unix网络编程2.1——高并发服务器（一）基础——io与文件描述符、socket编程与单进程服务端客户端实现
- unix网络编程2.2——高并发服务器（二）多进程与多线程实现
- unix网络编程2.3——高并发服务器（三）多路IO复用之select
- unix网络编程2.4——高并发服务器（四）epoll基础篇
- unix网络编程2.5——高并发服务器（五）epoll进阶篇——基于epoll实现reactor
- unix网络编程2.6——高并发服务器（六）基于epoll&&reactor实现http服务器
- unix网络编程2.7——高并发服务器（七）基于epoll&&reactor实现web_socket服务器
- unix网络编程2.8——高并发服务器（八）unix网络编程系统调用与网络协议栈
- unix网络编程3.1——UDP（一）UDP入门
- unix网络编程3.2——UDP（二）UDP可靠性传输1——KCP协议（上）
- KCP概述
- 相关名词说明
- kcp发数据和收数据流程
- kcp使用方式
- kcp协议头
- kcp发送数据过程
- kcp接收数据过程
- ikcp.c
- ikcp.h
系列文章
阅读本文需要先阅读下面的文章：
unix网络编程1.1——TCP协议详解（一）
unix网络编程2.1——高并发服务器（一）基础——io与文件描述符、socket编程与单进程服务端客户端实现
unix网络编程2.2——高并发服务器（二）多进程与多线程实现
unix网络编程2.3——高并发服务器（三）多路IO复用之select
unix网络编程2.4——高并发服务器（四）epoll基础篇
unix网络编程2.5——高并发服务器（五）epoll进阶篇——基于epoll实现reactor
unix网络编程2.6——高并发服务器（六）基于epoll&&reactor实现http服务器
unix网络编程2.7——高并发服务器（七）基于epoll&&reactor实现web_socket服务器
unix网络编程2.8——高并发服务器（八）unix网络编程系统调用与网络协议栈
unix网络编程3.1——UDP（一）UDP入门
unix网络编程3.2——UDP（二）UDP可靠性传输1——KCP协议（上）
KCP概述
- 官网
- TCP保证数据准确交付，UDP保证数据快速到达，KCP则是两种协议的一个折中。
- KCP的设计目标是为了解决在网络拥堵的情况下TCP传输速度慢的问题。
KCP是一个快速可靠协议，能以比 TCP浪费10%-20%的带宽的代价，换取平均延迟降低 30%-40%，且最大延迟降低三倍的传输效果。纯算法实现，并不负责底层协议（如UDP）的收发，需要使用者自己定义下层数据的发送方式，以 callback的方式提供给 KCP。连时钟都需要外部传递进来，内部不会有任何一次系统调用。
TCP是为流量设计的（每秒内可以传输多少KB的数据），讲究的是充分利用带宽。而 KCP是为流速设计的（单个数据从一端发送到一端需要多少时间），以10%-20%带宽浪费的代价换取了比 TCP快30%-40%的传输速度。
- KCP力求在保证可靠性的情况下提高传输速度。
相关名词说明
- 用户数据：应用层发送的数据，如一张图片2Kb 的数据
- MTU：最大传输单元。即每次发送的最大数据 1500 实际使用 1400
- RTO：Retransmission TimeOut ，重传超时时间。
- cwnd: congestion window ，拥塞窗口，表示发送方可发送多少个 KCP 数据包。与接收方窗口有关，与网络状况（拥塞控制）有关，与发送窗口大小有关。
- rwnd: receiver 接收方窗口大小，表示接收方还可接收多少个 KCP 数据包
- snd_queue: 待发送 KCP 数据包队列
- snd_buf:
- snd_nxt：下一个即将发送的 kcp 数据包序列号
- snd_una：下一个待确认的序列号，即是之前的包接收端都已经收到。
kcp发数据和收数据流程
发送数据
- 应用层发送数据，调用ikcp_send接口，根据MSS进行分片，分片后的包插入待发送队列(snd_queue)中
- 再通过调度单元(一定时间间隔)：ikcp_update -> 会调用ikcp_flush，数据从snd_queue放入到snd_buf(发送窗口)中
- ikcp_flush： 检查 kcp->update 是否更新，未更新直接返回。kcp->update 由 ikcp_update 更新，上层应用需要每隔一段时间（10-100ms）调用 ikcp_update 来驱动 KCP 发送数据；
- 通过snd_buf，调用ikcp_output，在ikcp_output中回调真正的发送数据接口: sendto，进行数据发送
接收数据
- 应用层recvfrom接收到数据，将收到的udp数据写入kcp模型，调用的是ikcp_input接口，准备解析数据
- 再通过调度单元: ikcp_update -> 会调用ikcp_flush，数据从rcv_buf(接收窗口)中，进行包排序等处理
- 数据从recv_buf放入到recv_queue中，排好序的包拷贝到rcv_queue中；
- 通过ikcp_recv中读取recv_queue中的数据，应用层真正接收到数据
kcp使用方式
- 创建KCP 对象：
ikcpcb *kcp = ikcp_create(conv, user);- 通过会话id(conv id)建立连接，客户端与服务端协商好
- 设置发送回调函数（如 UDP 的 send 函数）： kcp->output = udp_output;
- 真正发送数据需要调用sendto
- 循环调用update ikcp_update(kcp , millisec);
- 在一个线程或者定时器5ms/10ms做调度
- 输入一个应用层数据包（如UDP 收到的数据包）:
ikcp_input(kcp, received_udp_packet, received_udp_size);- 我们要使用recvfrom 接收，然后扔到 kcp 里面做解析
- 发送数据：ikcp_send(kcp1, buffer, 8); 用户层接口
- 接收数据：hr = ikcp_recv(kcp2, buffer, 10); 用户层接口
kcp协议头
- [0,3] conv: 连接号。 UDP 是无连接的， conv 用于表示来自于哪个客户端。对连接的一种替代
- [4] cmd 命令字。如， IKCP_CMD_ACK 确认命令，IKCP_CMD_WASK 接收窗口大小询问命令， IKCP_CMD_WINS 接收窗口大小告知命令，
- [5] frg 分片，用户数据可能会被分成多个 KCP 包，发送出去
- [6,7] wnd 接收窗口大小，发送方的发送窗口不能超过接收方给出的数值
- [8,11] ts 时间序列
- [12,15] sn 序列号
- [16,19] una:下一个可接收的序列号。其实就是确认号，收到sn = 10 的包， una 为 11
- [20,23] len ：数据长度
- data: 用户 数据，这一次发送的数据长度
conv id如何分配
- 参考asio_kcp
cmd 命令字
- IKCP_CMD_PUSH 和 IKCP_CMD_ACK 关联
- IKCP_CMD_WASK 和 IKCP_CMD_WINS 关联
- IKCP_CMD_ACK是确认单独的包
- 什么时候确认时不用IKCP_CMD_ACK?
frg 分片序号
ts 时间序列
- 两个作用：
- 计算rtt，包1发送的时间ts1, 与收到包1的ack应答包时间ts2，
rtt = ts2 - ts1，注意ts1与ts2都是“自己的本地时间”; - 与重传时间阈值作比较，判断是否需要重传;
kcp发送数据过程
- 发送ack应答包
- 检测窗口是否等于0，判断是否发探测包
- 询问接收窗口大小
- 告知接收窗口大小
- 窗口机制snd_buf: 如，最大32个，未确认19个，则只能从snd_queue中读取13个包
- 发送snd_buf数据分片 —— 核心逻辑
- 将snd_queue数据拷贝到snd_buf中
- 判断包是否要发送
- 正常发送
- 超时重传
- 快速重传，避免超时重传
- 开始发送 —— needsend
- 更新拥塞窗口
C 库宏 - offsetof()
- C 库宏 offsetof(type, member-designator) 会生成一个类型为 size_t 的整型常量，它是一个结构成员相对于结构开头的字节偏移量。成员是由 member-designator 给定的，结构的名称是在 type 中给定的。
- 参考
kcp接收数据过程
ikcp.c
//=====================================================================
//
// KCP - A Better ARQ Protocol Implementation
// skywind3000 (at) gmail.com, 2010-2011
//
// Features:
// + Average RTT reduce 30% - 40% vs traditional ARQ like tcp.
// + Maximum RTT reduce three times vs tcp.
// + Lightweight, distributed as a single source file.
//
//=====================================================================
#include "ikcp.h"
#include <stddef.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include <stdio.h>
//=====================================================================
// KCP BASIC
//=====================================================================
const IUINT32 IKCP_RTO_NDL = 30; // rto: no delay min rto 无延迟的最小重传超时时间
const IUINT32 IKCP_RTO_MIN = 100; // rto: normal min rto 正常模式最小超时重传
const IUINT32 IKCP_RTO_DEF = 200; // rto: 默认超时重传
const IUINT32 IKCP_RTO_MAX = 60000; // rto: 最大超时超时
const IUINT32 IKCP_CMD_PUSH = 81; // cmd: push data 协议类型 [正常接收数据]
const IUINT32 IKCP_CMD_ACK = 82; // cmd: ack 协议类型 [收到ack回复]
const IUINT32 IKCP_CMD_WASK = 83; // cmd: window probe (ask) 协议类型 [询问对方窗口size]
const IUINT32 IKCP_CMD_WINS = 84; // cmd: window size (tell) 协议类型 [告知对方我的窗口size]
const IUINT32 IKCP_ASK_SEND = 1; // need to send IKCP_CMD_WASK 是否需要发送 IKCP_CMD_WASK
const IUINT32 IKCP_ASK_TELL = 2; // need to send IKCP_CMD_WINS 是否需要发送 IKCP_CMD_WINS
const IUINT32 IKCP_WND_SND = 32; // 发送队列滑动窗口最大值
const IUINT32 IKCP_WND_RCV = 128; // 接收队列滑动窗口最大值 must >= max fragment size
const IUINT32 IKCP_MTU_DEF = 1400; // segment: 报文默认大小 [mtu 网络最小传输单元]
const IUINT32 IKCP_ACK_FAST = 3; // null: 没有被用使用
const IUINT32 IKCP_INTERVAL = 100; // flush: 控制刷新时间间隔
const IUINT32 IKCP_OVERHEAD = 24; // segment: 报文默认大小 [mtu 网络最小传输单元]
const IUINT32 IKCP_DEADLINK = 20;
const IUINT32 IKCP_THRESH_INIT = 2; // ssthresh: 慢热启动 初始窗口大小
const IUINT32 IKCP_THRESH_MIN = 2; // ssthresh: 慢热启动 最小窗口大小
const IUINT32 IKCP_PROBE_INIT = 7000; // 7 secs to probe window size 请求询问远端窗口大小的初始时间
const IUINT32 IKCP_PROBE_LIMIT = 120000; // up to 120 secs to probe window 请求询问远端窗口大小的最大时间
const IUINT32 IKCP_FASTACK_LIMIT = 5; // max times to trigger fastack
//---------------------------------------------------------------------
// encode / decode
//---------------------------------------------------------------------
/* encode 8 bits unsigned int */
static inline char *ikcp_encode8u(char *p, unsigned char c)
{
*(unsigned char*)p++ = c;
return p;
}
/* decode 8 bits unsigned int */
static inline const char *ikcp_decode8u(const char *p, unsigned char *c)
{
*c = *(unsigned char*)p++;
return p;
}
/* encode 16 bits unsigned int (lsb) */
static inline char *ikcp_encode16u(char *p, unsigned short w)
{
#if IWORDS_BIG_ENDIAN || IWORDS_MUST_ALIGN
*(unsigned char*)(p + 0) = (w & 255);
*(unsigned char*)(p + 1) = (w >> 8);
#else
memcpy(p, &w, 2);
#endif
p += 2;
return p;
}
/* decode 16 bits unsigned int (lsb) */
static inline const char *ikcp_decode16u(const char *p, unsigned short *w)
{
#if IWORDS_BIG_ENDIAN || IWORDS_MUST_ALIGN
*w = *(const unsigned char*)(p + 1);
*w = *(const unsigned char*)(p + 0) + (*w << 8);
#else
memcpy(w, p, 2);
#endif
p += 2;
return p;
}
/* encode 32 bits unsigned int (lsb) */
static inline char *ikcp_encode32u(char *p, IUINT32 l)
{
#if IWORDS_BIG_ENDIAN || IWORDS_MUST_ALIGN
*(unsigned char*)(p + 0) = (unsigned char)((l >> 0) & 0xff);
*(unsigned char*)(p + 1) = (unsigned char)((l >> 8) & 0xff);
*(unsigned char*)(p + 2) = (unsigned char)((l >> 16) & 0xff);
*(unsigned char*)(p + 3) = (unsigned char)((l >> 24) & 0xff);
#else
memcpy(p, &l, 4);
#endif
p += 4;
return p;
}
/* decode 32 bits unsigned int (lsb) */
static inline const char *ikcp_decode32u(const char *p, IUINT32 *l)
{
#if IWORDS_BIG_ENDIAN || IWORDS_MUST_ALIGN
*l = *(const unsigned char*)(p + 3);
*l = *(const unsigned char*)(p + 2) + (*l << 8);
*l = *(const unsigned char*)(p + 1) + (*l << 8);
*l = *(const unsigned char*)(p + 0) + (*l << 8);
#else
memcpy(l, p, 4);
#endif
p += 4;
return p;
}
static inline IUINT32 _imin_(IUINT32 a, IUINT32 b) {
return a <= b ? a : b;
}
static inline IUINT32 _imax_(IUINT32 a, IUINT32 b) {
return a >= b ? a : b;
}
static inline IUINT32 _ibound_(IUINT32 lower, IUINT32 middle, IUINT32 upper)
{
return _imin_(_imax_(lower, middle), upper);
}
static inline long _itimediff(IUINT32 later, IUINT32 earlier)
{
return ((IINT32)(later - earlier));
}
//---------------------------------------------------------------------
// manage segment
//---------------------------------------------------------------------
typedef struct IKCPSEG IKCPSEG;
static void* (*ikcp_malloc_hook)(size_t) = NULL;
static void (*ikcp_free_hook)(void *) = NULL;
// internal malloc
static void* ikcp_malloc(size_t size) {
if (ikcp_malloc_hook)
return ikcp_malloc_hook(size);
return malloc(size);
}
// internal free
static void ikcp_free(void *ptr) {
if (ikcp_free_hook) {
ikcp_free_hook(ptr);
} else {
free(ptr);
}
}
// redefine allocator
void ikcp_allocator(void* (*new_malloc)(size_t), void (*new_free)(void*))
{
ikcp_malloc_hook = new_malloc;
ikcp_free_hook = new_free;
}
// allocate a new kcp segment
static IKCPSEG* ikcp_segment_new(ikcpcb *kcp, int size)
{
return (IKCPSEG*)ikcp_malloc(sizeof(IKCPSEG) + size);
}
// delete a segment
static void ikcp_segment_delete(ikcpcb *kcp, IKCPSEG *seg)
{
ikcp_free(seg);
}
// write log
void ikcp_log(ikcpcb *kcp, int mask, const char *fmt, ...)
{
char buffer[1024];
va_list argptr;
if ((mask & kcp->logmask) == 0 || kcp->writelog == 0) return;
va_start(argptr, fmt);
vsprintf(buffer, fmt, argptr);
va_end(argptr);
kcp->writelog(buffer, (struct IKCPCB *)kcp, kcp->user);
}
// check log mask
static int ikcp_canlog(const ikcpcb *kcp, int mask)
{
if ((mask & kcp->logmask) == 0 || kcp->writelog == NULL) return 0;
return 1;
}
// output segment
static int ikcp_output(ikcpcb *kcp, const void *data, int size)
{
assert(kcp);
assert(kcp->output);
if (ikcp_canlog(kcp, IKCP_LOG_OUTPUT)) {
ikcp_log(kcp, IKCP_LOG_OUTPUT, "[RO] %ld bytes", (long)size);
}
if (size == 0) return 0;
return kcp->output((const char*)data, size, kcp, kcp->user);
}
// output queue
void ikcp_qprint(const char *name, const struct IQUEUEHEAD *head)
{
#if 0
const struct IQUEUEHEAD *p;
printf("<%s>: [", name);
for (p = head->next; p != head; p = p->next) {
const IKCPSEG *seg = iqueue_entry(p, const IKCPSEG, node);
printf("(%lu %d)", (unsigned long)seg->sn, (int)(seg->ts % 10000));
if (p->next != head) printf(",");
}
printf("]\n");
#endif
}
//---------------------------------------------------------------------
// create a new kcpcb 创建kcp对象，然后进行初始化
//---------------------------------------------------------------------
ikcpcb* ikcp_create(IUINT32 conv, void *user)
{
ikcpcb *kcp = (ikcpcb*)ikcp_malloc(sizeof(struct IKCPCB));
if (kcp == NULL) return NULL;
kcp->conv = conv;
kcp->user = user;
kcp->snd_una = 0;
kcp->snd_nxt = 0;
kcp->rcv_nxt = 0;
kcp->ts_recent = 0;
kcp->ts_lastack = 0;
kcp->ts_probe = 0;
kcp->probe_wait = 0;
kcp->snd_wnd = IKCP_WND_SND; // 默认发送的窗口
kcp->rcv_wnd = IKCP_WND_RCV; // 默认接收的窗口
kcp->rmt_wnd = IKCP_WND_RCV; // 默认远端的接收窗口
kcp->cwnd = 0; // 拥塞动态窗口
kcp->incr = 0;
kcp->probe = 0; // 当出现拥塞的时候，我们就可能要探测窗口
kcp->mtu = IKCP_MTU_DEF; // 缺省的mtu大小
kcp->mss = kcp->mtu - IKCP_OVERHEAD; // mtu - 分片 header头部占用字节
kcp->stream = 0;
kcp->buffer = (char*)ikcp_malloc((kcp->mtu + IKCP_OVERHEAD) * 3); // 消息字节流, 类似tcp方式使用
if (kcp->buffer == NULL) {
ikcp_free(kcp);
return NULL;
}
iqueue_init(&kcp->snd_queue);
iqueue_init(&kcp->rcv_queue);
iqueue_init(&kcp->snd_buf);
iqueue_init(&kcp->rcv_buf);
kcp->nrcv_buf = 0; // 队列 缓存的数量
kcp->nsnd_buf = 0;
kcp->nrcv_que = 0;
kcp->nsnd_que = 0;
kcp->state = 0;
kcp->acklist = NULL; // 需要应答的序号
kcp->ackblock = 0;
kcp->ackcount = 0;
kcp->rx_srtt = 0;
kcp->rx_rttval = 0;
kcp->rx_rto = IKCP_RTO_DEF; // 缺省的超时重传时间
kcp->rx_minrto = IKCP_RTO_MIN; // 最小重传时间
kcp->current = 0; // 当前时间
kcp->interval = IKCP_INTERVAL; // 缺省刷新循环的时间间隔
kcp->ts_flush = IKCP_INTERVAL;
kcp->nodelay = 0; // 无延迟是否开启
kcp->updated = 0; // 是否调用过ikcp_update
kcp->logmask = 0;
kcp->ssthresh = IKCP_THRESH_INIT; // 窗口启动初始化
kcp->fastresend = 0; // 触发ack的数量, 本质是跳过了多少个ack就重传
kcp->fastlimit = IKCP_FASTACK_LIMIT;
kcp->nocwnd = 0; // 是否开启拥塞控制
kcp->xmit = 0; // 只是计数器
kcp->dead_link = IKCP_DEADLINK; // 重传的最大次数, 如果达到则认为链路断开
kcp->output = NULL; // udp输出函数
kcp->writelog = NULL; // 可以设置外部的log
return kcp;
}
//---------------------------------------------------------------------
// release a new kcpcb 释放一个kcp对象
//---------------------------------------------------------------------
void ikcp_release(ikcpcb *kcp)
{
assert(kcp);
if (kcp) {
IKCPSEG *seg;
while (!iqueue_is_empty(&kcp->snd_buf)) {
seg = iqueue_entry(kcp->snd_buf.next, IKCPSEG, node);
iqueue_del(&seg->node);
ikcp_segment_delete(kcp, seg);
}
while (!iqueue_is_empty(&kcp->rcv_buf)) {
seg = iqueue_entry(kcp->rcv_buf.next, IKCPSEG, node);
iqueue_del(&seg->node);
ikcp_segment_delete(kcp, seg);
}
while (!iqueue_is_empty(&kcp->snd_queue)) {
seg = iqueue_entry(kcp->snd_queue.next, IKCPSEG, node);
iqueue_del(&seg->node);
ikcp_segment_delete(kcp, seg);
}
while (!iqueue_is_empty(&kcp->rcv_queue)) {
seg = iqueue_entry(kcp->rcv_queue.next, IKCPSEG, node);
iqueue_del(&seg->node);
ikcp_segment_delete(kcp, seg);
}
if (kcp->buffer) {
ikcp_free(kcp->buffer);
}
if (kcp->acklist) {
ikcp_free(kcp->acklist);
}
kcp->nrcv_buf = 0;
kcp->nsnd_buf = 0;
kcp->nrcv_que = 0;
kcp->nsnd_que = 0;
kcp->ackcount = 0;
kcp->buffer = NULL;
kcp->acklist = NULL;
ikcp_free(kcp);
}
}
//---------------------------------------------------------------------
// set output callback, which will be invoked by kcp 设置输出函数回调
//---------------------------------------------------------------------
void ikcp_setoutput(ikcpcb *kcp, int (*output)(const char *buf, int len,
ikcpcb *kcp, void *user))
{
kcp->output = output;
}
//---------------------------------------------------------------------
// user/upper level recv: returns size, returns below zero for EAGAIN
// 读取组好的数据
//---------------------------------------------------------------------
/*
主要做三件事情:
1. 读取组好包的数据
2. 将接收缓存rcv_buf的分片转移到接收队列rcv_queue
3. 如果有接收空间则将kcp->probe |= IKCP_ASK_TELL; 以在update的时候告知对方可以发送数据了。
*/
int ikcp_recv(ikcpcb *kcp, char *buffer, int len)
{
struct IQUEUEHEAD *p;
int ispeek = (len < 0)? 1 : 0; // peek:窥视; 偷看. 如果是 ispeek 说明只是为了拿数据看看，则不用将数据从queue删除
int peeksize;
int recover = 0;
IKCPSEG *seg;
assert(kcp);
// 排序好的数据存放在rcv_queue, 反过来待发送的数据则存放在snd_queue
if (iqueue_is_empty(&kcp->rcv_queue)) // 如果为空则没有数据可读
return -1;
if (len < 0) len = -len;
//计算当前接收队列中的属于同一个消息的数据总长度(不是所有消息的总长度)，注意这里的同一个消息是seg->frg进行标记
peeksize = ikcp_peeksize(kcp); // 当我们没有采用流式传输的时候, 我们接收的则是类似 udp一样的报文传输方式
if (peeksize < 0) // 没有数据可读
return -2;
if (peeksize > len) // 可读数据大于 用户传入的长度，每次读取需要一次性读取完毕，类似udp报文的读取
return -3;
// KCP 协议在远端窗口为0的时候将会停止发送数据
if (kcp->nrcv_que >= kcp->rcv_wnd) // 接收队列segment数量大于等于接收窗口，标记窗口可以恢复
recover = 1; // 标记可以开始窗口恢复
// merge fragment 将属于同一个消息的各分片重组完整数据，并删除rcv_queue中segment，nrcv_que减少
// 经过 ikcp_send 发送的数据会进行分片，分片编号为倒序序号，因此 frg 为 0 的数据包标记着完整接收到了一次 send 发送过来的数据
for (len = 0, p = kcp->rcv_queue.next; p != &kcp->rcv_queue; ) {
int fragment;
seg = iqueue_entry(p, IKCPSEG, node);
p = p->next;
if (buffer) {
memcpy(buffer, seg->data, seg->len); // 把queue的数据就放入用户buffer
buffer += seg->len;
}
len += seg->len;
fragment = seg->frg;
if (ikcp_canlog(kcp, IKCP_LOG_RECV)) {
ikcp_log(kcp, IKCP_LOG_RECV, "recv sn=%lu", (unsigned long)seg->sn);
}
if (ispeek == 0) {
iqueue_del(&seg->node);
ikcp_segment_delete(kcp, seg); // 删除节点
kcp->nrcv_que--; // nrcv_que接收队列-1
}
if (fragment == 0) // 完整的数据接收到, send的时候如果数据被分片，比如分成 n个片，则0 .. n-2的fragment为1，n-1的为0
break;
}
assert(len == peeksize);
// move available data from rcv_buf -> rcv_queue
// 将合适的数据从接收缓存rcv_buf 转移到接收队列rev_queue, 已经确认的以及接收窗口未满
while (! iqueue_is_empty(&kcp->rcv_buf)) {
seg = iqueue_entry(kcp->rcv_buf.next, IKCPSEG, node);
// 条件1 序号是该接收的数据
// 条件2 接收队列nrcv_que < 接收窗口rcv_wnd;
if (seg->sn == kcp->rcv_nxt && kcp->nrcv_que < kcp->rcv_wnd) {
iqueue_del(&seg->node);
kcp->nrcv_buf--;
iqueue_add_tail(&seg->node, &kcp->rcv_queue);
kcp->nrcv_que++; // 接收队列 有多少个分片 + 1
kcp->rcv_nxt++; // 接收序号 + 1
} else {
break;
}
}
// fast recover ，nrcv_que小于rcv_wnd, 说明接收端有空间继续接收数据了
if (kcp->nrcv_que < kcp->rcv_wnd && recover) {
// ready to send back IKCP_CMD_WINS in ikcp_flush
// tell remote my window size
kcp->probe |= IKCP_ASK_TELL;
}
return len;
}
//---------------------------------------------------------------------
// peek data size 计算当前一帧数据的总大小(一个或多个分片组成的数据帧)
//---------------------------------------------------------------------
int ikcp_peeksize(const ikcpcb *kcp)
{
struct IQUEUEHEAD *p;
IKCPSEG *seg;
int length = 0;
assert(kcp);
if (iqueue_is_empty(&kcp->rcv_queue)) return -1;
seg = iqueue_entry(kcp->rcv_queue.next, IKCPSEG, node);
if (seg->frg == 0) return seg->len;
if (kcp->nrcv_que < seg->frg + 1) return -1;
for (p = kcp->rcv_queue.next; p != &kcp->rcv_queue; p = p->next) {
seg = iqueue_entry(p, IKCPSEG, node);
length += seg->len; //
if (seg->frg == 0) break;
}
return length;
}
//---------------------------------------------------------------------
// user/upper level send, returns below zero for error
//---------------------------------------------------------------------
/*
把用户发送的数据根据MSS(max segment size)分片成KCP的数据分片格式，插入待发送队列中。当用户的数据超过一个MSS(最大分片大小)
的时候，会对发送的数据进行分片处理。通过frg进行排序区分，frg即message中的segment分片ID，在message中的索引，由大到小，
0表示最后一个分片。分成4片时，frg为3,2,1,0。
如用户发送2900字节的数据，MSS为1400byte。因此，该函数会把1900byte的用户数据分成两个分片，一个数据大小为1400，头frg设置为2，
len设置为1400；第二个分片，头frg设置为1，len设置为1400; 第三个分片, 头frg设置为0，len设置为100。
切好KCP分片之后，放入到名为snd_queue的待发送队列中。
*/
/*分片方式共有两种。
(1) 流模式情况下，检测每个发送队列里的分片是否达到最大MSS，如果没有达到就会用新的数据填充分片。
接收端会把多片发送的数据重组为一个完整的KCP帧。
(2)消息模式下，将用户数据分片，为每个分片设置sn和frag，将分片后的数据一个一个地存入发送队列，
接收方通过sn和frag解析原来的包，消息方式一个分片的数据量可能不能达到MSS，也会作为一个包发送出去。*/
/*
不能一下send太长的数据, 当数据长度/mss大于对方接收窗口的时候则返回错误
*/
int ikcp_send(ikcpcb *kcp, const char *buffer, int len)
{
IKCPSEG *seg;
int count, i;
assert(kcp->mss > 0); // 从mtu
if (len < 0) return -1;
// append to previous segment in streaming mode (if possible)
// 1 如果当前的 KCP 开启流模式，取出 `snd_queue` 中的最后一个报文 将其填充到 mss 的长度，并设置其 frg 为 0.
if (kcp->stream != 0) { // 先不考虑流式的
if (!iqueue_is_empty(&kcp->snd_queue)) {
IKCPSEG *old = iqueue_entry(kcp->snd_queue.prev, IKCPSEG, node);
/*节点内数据长度小于mss，计算还可容纳的数据大小，以及本次占用的空间大小，
以此新建segment，将新建segment附加到发送队列尾，将old节点内数据拷贝过去，
然后将buffer中也拷贝其中，如果buffer中的数据没有拷贝完，extend为拷贝数据，
开始frg计数。更新len为剩余数据，删除old */
if (old->len < kcp->mss) {
int capacity = kcp->mss - old->len;
int extend = (len < capacity)? len : capacity;
seg = ikcp_segment_new(kcp, old->len + extend);
assert(seg);
if (seg == NULL) {
return -2;
}
iqueue_add_tail(&seg->node, &kcp->snd_queue); // 重新一个新segment加入
memcpy(seg->data, old->data, old->len);
if (buffer) {
memcpy(seg->data + old->len, buffer, extend);
buffer += extend;
}
seg->len = old->len + extend;
seg->frg = 0;
len -= extend;
iqueue_del_init(&old->node);
ikcp_segment_delete(kcp, old);
}
}
if (len <= 0) {
return 0;
}
}
// 2 计算数据可以被最多分成多少个frag
if (len <= (int)kcp->mss) count = 1; // 分片(头部24) + user data(mss 1376) = mtu （1400）
else count = (len + kcp->mss - 1) / kcp->mss; // mss = mtu(1400) - 24 = 1376 ->
// 1376: cout =1; 1377:cout =2, 1376, 1
if (count >= (int)IKCP_WND_RCV) return -2; // 超过对方的初始接收窗口
// 这里的设计有个疑问，如果一致send则snd_queue一直增长, 如果去也去判断主机的发送窗口大小可能更优些
if (count == 0) count = 1; // ?
// fragment
// 3 将数据全部新建segment插入发送队列尾部，队列计数递增, frag递减
for (i = 0; i < count; i++) {
int size = len > (int)kcp->mss ? (int)kcp->mss : len;
seg = ikcp_segment_new(kcp, size);
assert(seg);
if (seg == NULL) {
return -2;
}
if (buffer && len > 0) {
memcpy(seg->data, buffer, size); // 拷贝数据
}
seg->len = size; // 每seg 数据大小
// count =4; frg 3 2 1 0
seg->frg = (kcp->stream == 0)? (count - i - 1) : 0; // frg编号 , 流模式情况下分片编号不用填写
iqueue_init(&seg->node);
iqueue_add_tail(&seg->node, &kcp->snd_queue); // 发送队列
kcp->nsnd_que++; // 发送队列++
if (buffer) {
buffer += size;
}
len -= size;
}
return 0;
}
//---------------------------------------------------------------------
// parse ack
//---------------------------------------------------------------------
/*
RTT: Round Trip Time，也就是一个数据包从发出去到回来的时间.
这样发送端就大约知道需要多少的时间，从而可以方便地设置Timeout——RTO（Retransmission TimeOut）
RTO: (Retransmission TimeOut)即重传超时时间
rx_srtt: smoothed round trip time，平滑后的RTT
rx_rttval：RTT的变化量，代表连接的抖动情况
interval：内部flush刷新间隔，对系统循环效率有非常重要影响
作用：更新RTT和RTO等参数，该算法与TCP保持一致:
1. 第一次测量，rtt 是我们测量的结果，rx_srtt = rtt，rx_rttval = rtt / 2
2. 以后每次测量：
rx_srtt =(1-a) * rx_srtt + a * rtt，a取值1/8
rx_rttval= (1-b) * rx_rttval + b * |rtt - rx_srtt|，b取值1/4
rto = rx_srtt + 4 * rx_rttval
rx_rto = MIN(MAX(rx_minrto, rto), IKCP_RTO_MAX)
*/
static void ikcp_update_ack(ikcpcb *kcp, IINT32 rtt)
{
IINT32 rto = 0;
if (kcp->rx_srtt == 0) { //rx_srtt初始为0时
kcp->rx_srtt = rtt;
kcp->rx_rttval = rtt / 2;
} else {
long delta = rtt - kcp->rx_srtt; //计算这次和之前的差值
if (delta < 0) delta = -delta;
kcp->rx_rttval = (3 * kcp->rx_rttval + delta) / 4; //权重计算 rtt的变化量
kcp->rx_srtt = (7 * kcp->rx_srtt + rtt) / 8; // 计算平滑后的rtt
if (kcp->rx_srtt < 1) kcp->rx_srtt = 1;
}
/*! 通过抖动情况与内部调度间隔计算出RTO时间 */
rto = kcp->rx_srtt + _imax_(kcp->interval, 4 * kcp->rx_rttval);
/*! 使得最后结果在minrto <= x <= IKCP_RTO_MAX 之间 */
kcp->rx_rto = _ibound_(kcp->rx_minrto, rto, IKCP_RTO_MAX);
}
// 更新下一个需要对方应答的序号 snd_una
static void ikcp_shrink_buf(ikcpcb *kcp)
{
// 更新未确认的segment的序号
struct IQUEUEHEAD *p = kcp->snd_buf.next;
if (p != &kcp->snd_buf) {/*! 判断发送队列不为空，与iqueue_is_empty一个意思 */
IKCPSEG *seg = iqueue_entry(p, IKCPSEG, node);
kcp->snd_una = seg->sn;
} else {
kcp->snd_una = kcp->snd_nxt;
}
}
// 该函数主要工作从发送buf中删除相应编号的分片
static void ikcp_parse_ack(ikcpcb *kcp, IUINT32 sn)
{
struct IQUEUEHEAD *p, *next;
// 当前确认数据包ack的编号小于已经接收到的编号(una)或数据包的ack编号大于待分配的编号则不合法
if (_itimediff(sn, kcp->snd_una) < 0 || _itimediff(sn, kcp->snd_nxt) >= 0)
return;
// 遍历发送队列释放该编号分片，已经确认就可以删除了
for (p = kcp->snd_buf.next; p != &kcp->snd_buf; p = next) {
IKCPSEG *seg = iqueue_entry(p, IKCPSEG, node);
next = p->next;
if (sn == seg->sn) {
iqueue_del(p);
ikcp_segment_delete(kcp, seg);
kcp->nsnd_buf--;
break;
}
// 如果该编号小于则表明不需要继续遍历下去，因为队列里面的序号是递增的
if (_itimediff(sn, seg->sn) < 0) {
break;
}
}
}
// una前的序号都已经收到，所以将对方已经确认收到的数据从发送缓存删除
// 确定已经发送的数据包有哪些被对方接收到
static void ikcp_parse_una(ikcpcb *kcp, IUINT32 una)
{
struct IQUEUEHEAD *p, *next;
for (p = kcp->snd_buf.next; p != &kcp->snd_buf; p = next) {
IKCPSEG *seg = iqueue_entry(p, IKCPSEG, node);
next = p->next;
/*! 发送队列满足sn由大到小的顺序规则 */
if (_itimediff(una, seg->sn) > 0) {
iqueue_del(p);
ikcp_segment_delete(kcp, seg);
kcp->nsnd_buf--;
} else {
break;
}
}
}
// 解析fastack sn=3
// 4 3 2
// 解析fastack sn=4
// 4 3 2
static void ikcp_parse_fastack(ikcpcb *kcp, IUINT32 sn, IUINT32 ts)
{
struct IQUEUEHEAD *p, *next;
// 当前确认数据包ack的编号小于已经接收到的编号(una)或数据包的ack编号大于待分配的编号则不合法
// 说明已经发过了
if (_itimediff(sn, kcp->snd_una) < 0 || _itimediff(sn, kcp->snd_nxt) >= 0)
return;
// 遍历发送buf，进行快速确认(ack)
//snd_buf =
for (p = kcp->snd_buf.next; p != &kcp->snd_buf; p = next) {
IKCPSEG *seg = iqueue_entry(p, IKCPSEG, node);
next = p->next;
// seg->sn = 2；
// sn =4
if (_itimediff(sn, seg->sn) < 0) {
break;
}
else if (sn != seg->sn) {
#ifndef IKCP_FASTACK_CONSERVE
seg->fastack++; // 增加跳过的计数
// printf("sn:%u, c_sn:%u,fastack:%d\n", sn, seg->sn, seg->fastack);
#else
if (_itimediff(ts, seg->ts) >= 0)
seg->fastack++;
#endif
}
}
}
/**
* ts: message发送时刻的时间戳
* sn: 分片编号
* ackcount: acklist中ack的数量，每个ack在acklist中存储ts，sn两个量
* ackblock: 2的倍数，标识acklist最大可容纳的ack数量
* 该函数主要用与添加ack确认数据包信息
*/
//---------------------------------------------------------------------
// ack append 注意这里的append，意思是先把要sn的序号收集起来
//---------------------------------------------------------------------
static void ikcp_ack_push(ikcpcb *kcp, IUINT32 sn, IUINT32 ts)
{
IUINT32 newsize = kcp->ackcount + 1;
IUINT32 *ptr;
if (newsize > kcp->ackblock) {//判断是否超出acklist可容纳的数量
IUINT32 *acklist;
IUINT32 newblock;
//进行扩容，以8的N次方扩充
for (newblock = 8; newblock < newsize; newblock <<= 1);
//分配数组
acklist = (IUINT32*)ikcp_malloc(newblock * sizeof(IUINT32) * 2);
if (acklist == NULL) {
assert(acklist != NULL);
abort();
}
/*! 不为空则需要copy */
if (kcp->acklist != NULL) {
IUINT32 x;
for (x = 0; x < kcp->ackcount; x++) {
acklist[x * 2 + 0] = kcp->acklist[x * 2 + 0];
acklist[x * 2 + 1] = kcp->acklist[x * 2 + 1];
}
ikcp_free(kcp->acklist); //释放旧数据
}
kcp->acklist = acklist; //数组赋值
kcp->ackblock = newblock; // --> 容量赋值
}
ptr = &kcp->acklist[kcp->ackcount * 2]; //进行数组下标偏移
ptr[0] = sn;
ptr[1] = ts;
kcp->ackcount++; //增加数量
}
// sn:message分片segment的序号,ts:message发送时刻的时间戳
// 通过下标偏移获取相应位置p上的ack确认包数据信息
static void ikcp_ack_get(const ikcpcb *kcp, int p, IUINT32 *sn, IUINT32 *ts)
{
if (sn) sn[0] = kcp->acklist[p * 2 + 0];
if (ts) ts[0] = kcp->acklist[p * 2 + 1];
}
//---------------------------------------------------------------------
// 该函数主要将数据分片追加到buf中，并将buf中数据有序的移置接收队列中
// 1. 先检测该分片是否需要进行接收处理
// 2.将分片放入rcv_buf
// 3.将已经确认的rcv_buf里面的分片放入rcv_queue
//---------------------------------------------------------------------
void ikcp_parse_data(ikcpcb *kcp, IKCPSEG *newseg)
{
struct IQUEUEHEAD *p, *prev;
IUINT32 sn = newseg->sn;
int repeat = 0; // 接收也有窗口大小 rcv_nxt =4, rcv_wnd =30, sn > 34
// 判断该数据分片的编号是否超出接收窗口可接收的范围，或者该编号小于需要的则直接丢弃
if (_itimediff(sn, kcp->rcv_nxt + kcp->rcv_wnd) >= 0 || // 检测序号是否超过 缓存可以容纳的
_itimediff(sn, kcp->rcv_nxt) < 0) { // 检测是否是过期的数据 rcv_wnd = 4
ikcp_segment_delete(kcp, newseg); // 不在窗口内的都删除
return;
}
// 在接收buf中寻找编号为sn的分片用来判断是否重复 先插入 接收窗口， 是递增的排序
// 1 2 3 4 6
for (p = kcp->rcv_buf.prev; p != &kcp->rcv_buf; p = prev) {
IKCPSEG *seg = iqueue_entry(p, IKCPSEG, node);
prev = p->prev;
if (seg->sn == sn) { // 重复的包 检测序列化
repeat = 1;
break;
}
if (_itimediff(sn, seg->sn) > 0) {//由于分片编号是递增的
break;
}
}
if (repeat == 0) {
iqueue_init(&newseg->node);
iqueue_add(&newseg->node, p); // 加入到队列 按着顺序插入的
kcp->nrcv_buf++;
} else {
ikcp_segment_delete(kcp, newseg);
}
#if 0
ikcp_qprint("rcvbuf", &kcp->rcv_buf);
printf("rcv_nxt=%lu\n", kcp->rcv_nxt);
#endif
// move available data from rcv_buf -> rcv_queue
while (! iqueue_is_empty(&kcp->rcv_buf)) {
IKCPSEG *seg = iqueue_entry(kcp->rcv_buf.next, IKCPSEG, node);
if (seg->sn == kcp->rcv_nxt && kcp->nrcv_que < kcp->rcv_wnd) { // 将顺序正确的加入到队列
iqueue_del(&seg->node);
kcp->nrcv_buf--;
iqueue_add_tail(&seg->node, &kcp->rcv_queue);
kcp->nrcv_que++;
kcp->rcv_nxt++; // rcv_nxt = 4, seg->sn = 4; rcv_nxt = 5, seg->sn = 5
} else {
break;
}
}
#if 0
ikcp_qprint("queue", &kcp->rcv_queue);
printf("rcv_nxt=%lu\n", kcp->rcv_nxt);
#endif
#if 1
// printf("snd(buf=%d, queue=%d)\n", kcp->nsnd_buf, kcp->nsnd_que);
// printf("rcv(buf=%d, queue=%d)\n", kcp->nrcv_buf, kcp->nrcv_que);
#endif
}
/**
* 接收对方的数据输入
* 该函数主要是处理接收到的数据
* 校验数据=》解析数据=》处理数据（将合法的数据分片添加到接收buf中）=》拥塞窗口处理
*
* 1.检测una，将una之前的分片从snd_buf清除(批量)
* 2. 检测ack，对应ack sn分片从snd_buf清除（单个）
*/
//---------------------------------------------------------------------
// input data 我们在应用主动调用recvfrom读取udp数据，然后写入 ikcp_input
//---------------------------------------------------------------------
int ikcp_input(ikcpcb *kcp, const char *data, long size)
{
IUINT32 prev_una = kcp->snd_una; // 最新应答的序号
IUINT32 maxack = 0, latest_ts = 0;
int flag = 0;
if (ikcp_canlog(kcp, IKCP_LOG_INPUT)) {
ikcp_log(kcp, IKCP_LOG_INPUT, "[RI] %d bytes", (int)size);
}
// 据和长度的初步校验 数据太小时异常，因为kcp头部都占用了24字节了，即时sendto最小的大小为 IKCP_OVERHEAD
if (data == NULL || (int)size < (int)IKCP_OVERHEAD) return -1;
while (1) {
IUINT32 ts, sn, len, una, conv;
IUINT16 wnd;
IUINT8 cmd, frg;
IKCPSEG *seg;
if (size < (int)IKCP_OVERHEAD) break;
// 校验数据分片
data = ikcp_decode32u(data, &conv); // 获取segment头部信息
if (conv != kcp->conv) return -1; // 特别需要注意会话id的匹配
data = ikcp_decode8u(data, &cmd);
data = ikcp_decode8u(data, &frg);
data = ikcp_decode16u(data, &wnd);
data = ikcp_decode32u(data, &ts);
data = ikcp_decode32u(data, &sn);
data = ikcp_decode32u(data, &una);
data = ikcp_decode32u(data, &len);
size -= IKCP_OVERHEAD; //剔除固定的包头信息长度
if ((long)size < (long)len || (int)len < 0) return -2; // 数据不足或者, 没有真正的数据存在
//只支持{IKCP_CMD_PUSH, IKCP_CMD_ACK, IKCP_CMD_WASK, IKCP_CMD_WINS}指令
//其他不合法
if (cmd != IKCP_CMD_PUSH && cmd != IKCP_CMD_ACK &&
cmd != IKCP_CMD_WASK && cmd != IKCP_CMD_WINS)
return -3;
kcp->rmt_wnd = wnd; // 携带了远端的接收窗口 IKCP_CMD_WINS
ikcp_parse_una(kcp, una); // 删除小于snd_buf中小于una的segment, 意思是una之前的都已经收到了
ikcp_shrink_buf(kcp); // 更新snd_una为snd_buf中seg->sn或kcp->snd_nxt ，更新下一个待应答的序号
if (cmd == IKCP_CMD_ACK) {
if (_itimediff(kcp->current, ts) >= 0) { // 根据应答判断rtt
//更新rx_srtt，rx_rttval，计算kcp->rx_rto
ikcp_update_ack(kcp, _itimediff(kcp->current, ts));
}
//遍历snd_buf中（snd_una, snd_nxt），将sn相等的删除，直到大于sn
ikcp_parse_ack(kcp, sn); // 将已经ack的分片删除
ikcp_shrink_buf(kcp); // 更新控制块的 snd_una
if (flag == 0) {
flag = 1; //快速重传标记
maxack = sn; // 记录最大的 ACK 编号
latest_ts = ts;
} else {
if (_itimediff(sn, maxack) > 0) {
#ifndef IKCP_FASTACK_CONSERVE
maxack = sn; // 记录最大的 ACK 编号
latest_ts = ts;
#else
if (_itimediff(ts, latest_ts) > 0) {
maxack = sn;
latest_ts = ts;
}
#endif
}
}
if (ikcp_canlog(kcp, IKCP_LOG_IN_ACK)) {
ikcp_log(kcp, IKCP_LOG_IN_ACK,
"input ack: sn=%lu rtt=%ld rto=%ld", (unsigned long)sn,
(long)_itimediff(kcp->current, ts),
(long)kcp->rx_rto);
}
}
else if (cmd == IKCP_CMD_PUSH) { //接收到具体的数据包
if (ikcp_canlog(kcp, IKCP_LOG_IN_DATA)) {
ikcp_log(kcp, IKCP_LOG_IN_DATA,
"input psh: sn=%lu ts=%lu", (unsigned long)sn, (unsigned long)ts);
}
if (_itimediff(sn, kcp->rcv_nxt + kcp->rcv_wnd) < 0) {
ikcp_ack_push(kcp, sn, ts); // 对该报文的确认 ACK 报文放入 ACK 列表中
// 判断接收的数据分片编号是否符合要求，即：在接收窗口（滑动窗口）范围之内
if (_itimediff(sn, kcp->rcv_nxt) >= 0) { // 是要接受起始的序号
seg = ikcp_segment_new(kcp, len);
seg->conv = conv;
seg->cmd = cmd;
seg->frg = frg;
seg->wnd = wnd;
seg->ts = ts;
seg->sn = sn;
seg->una = una;
seg->len = len;
if (len > 0) {
memcpy(seg->data, data, len);
}
//1. 丢弃sn > kcp->rcv_nxt + kcp->rcv_wnd的segment;
//2. 逐一比较rcv_buf中的segment，若重复丢弃，非重复，新建segment加入;
//3. 检查rcv_buf的包序号sn，如果是待接收的序号rcv_nxt，且可以接收（接收队列小 于接收窗口），
// 转移segment到rcv_buf，nrcv_buf减少，nrcv_que增加，rcv_nxt增加;
ikcp_parse_data(kcp, seg); // 将该报文插入到 rcv_buf 链表中
}
}
}
else if (cmd == IKCP_CMD_WASK) {
// ready to send back IKCP_CMD_WINS in ikcp_flush // 如果是探测包
// tell remote my window size 添加相应的标识位
kcp->probe |= IKCP_ASK_TELL; // 收到对方请求后标记自己要告诉对方自己的窗口
if (ikcp_canlog(kcp, IKCP_LOG_IN_PROBE)) {
ikcp_log(kcp, IKCP_LOG_IN_PROBE, "input probe");
}
}
else if (cmd == IKCP_CMD_WINS) {
// do nothing 如果是tell me 远端窗口大小，什么都不做
if (ikcp_canlog(kcp, IKCP_LOG_IN_WINS)) {
ikcp_log(kcp, IKCP_LOG_IN_WINS,
"input wins: %lu", (unsigned long)(wnd));
}
}
else {
return -3;
}
data += len;
size -= len;
}
if (flag != 0) {
ikcp_parse_fastack(kcp, maxack, latest_ts);
}
// 如果snd_una增加了那么就说明对端正常收到且回应了发送方发送缓冲区第一个待确认的包，
// 此时需要更新cwnd（拥塞窗口）
// snd_una = 1; prev_una = 1;
// snd_una = 10,
// snd_una -prev_una = 10 - 1= 9
if (_itimediff(kcp->snd_una, prev_una) > 0) {
//如何拥塞窗口小于远端窗口
if (kcp->cwnd < kcp->rmt_wnd) {
IUINT32 mss = kcp->mss; //最大分片大小
if (kcp->cwnd < kcp->ssthresh) { //拥塞窗口小于阈值
kcp->cwnd++; // 扩大窗口?
kcp->incr += mss;
} else {
if (kcp->incr < mss) kcp->incr = mss;
kcp->incr += (mss * mss) / kcp->incr + (mss / 16);
if ((kcp->cwnd + 1) * mss <= kcp->incr) {
#if 1
kcp->cwnd = (kcp->incr + mss - 1) / ((mss > 0)? mss : 1);
#else
kcp->cwnd++;
#endif
}
}
//如果拥塞窗口大于远端窗口
if (kcp->cwnd > kcp->rmt_wnd) {
kcp->cwnd = kcp->rmt_wnd; //则使用远端窗口
kcp->incr = kcp->rmt_wnd * mss; //并设置相应数据量，该数据量以字节数
}
}
}
return 0;
}
//---------------------------------------------------------------------
// ikcp_encode_seg 打包分片
//---------------------------------------------------------------------
static char *ikcp_encode_seg(char *ptr, const IKCPSEG *seg)
{
ptr = ikcp_encode32u(ptr, seg->conv);
ptr = ikcp_encode8u(ptr, (IUINT8)seg->cmd);
ptr = ikcp_encode8u(ptr, (IUINT8)seg->frg);
ptr = ikcp_encode16u(ptr, (IUINT16)seg->wnd);
ptr = ikcp_encode32u(ptr, seg->ts);
ptr = ikcp_encode32u(ptr, seg->sn);
ptr = ikcp_encode32u(ptr, seg->una);
ptr = ikcp_encode32u(ptr, seg->len);
return ptr;
}
// 计算可接收长度，以分片为单位
static int ikcp_wnd_unused(const ikcpcb *kcp)
{
if (kcp->nrcv_que < kcp->rcv_wnd) {
return kcp->rcv_wnd - kcp->nrcv_que;
}
return 0;
}
//---------------------------------------------------------------------
// ikcp_flush 检查 kcp->update 是否更新，未更新直接返回。kcp->update 由 ikcp_update 更新，
// 上层应用需要每隔一段时间（10-100ms）调用 ikcp_update 来驱动 KCP 发送数据；
//---------------------------------------------------------------------
/*
准备将 acklist 中记录的 ACK 报文发送出去，即从 acklist 中填充 ACK 报文的 sn 和 ts 字段；
检查当前是否需要对远端窗口进行探测。由于 KCP 流量控制依赖于远端通知其可接受窗口的大小，
一旦远端接受窗口 kcp->rmt_wnd 为0，那么本地将不会再向远端发送数据，因此就没有机会从远端接受 ACK 报文，
从而没有机会更新远端窗口大小。在这种情况下，KCP 需要发送窗口探测报文到远端，待远端回复窗口大小后，后续传输可以继续。
在发送数据之前，先设置快重传的次数和重传间隔；KCP 允许设置快重传的次数，即 fastresend 参数。
例如设置 fastresend 为2，并且发送端发送了1,2,3,4,5几个分片，收到远端的ACK: 1, 3, 4, 5，
当收到ACK3时，KCP知道2被跳过1次，收到ACK4时，知道2被“跳过”了2次，此时可以认为2号丢失，
不用等超时，直接重传2号包；每个报文的 fastack 记录了该报文被跳过了几次，由函数 ikcp_parse_fastack 更新。
于此同时，KCP 也允许设置 nodelay 参数，当激活该参数时，每个报文的超时重传时间将由 x2 变为 x1.5，即加快报文重传：
* 1. ack确认包
* 2. 探测远端窗口
* 3. 发送snd_buf数据分片
* 4. 更新拥塞窗口
*/
void ikcp_flush(ikcpcb *kcp) // 主要发送数据
{
IUINT32 current = kcp->current;
char *buffer = kcp->buffer;
char *ptr = buffer;
int count, size, i;
IUINT32 resent, cwnd;
IUINT32 rtomin;
struct IQUEUEHEAD *p;
int change = 0;
int lost = 0;
IKCPSEG seg;
// 'ikcp_update' haven't been called.
if (kcp->updated == 0) return;
seg.conv = kcp->conv;
seg.cmd = IKCP_CMD_ACK; // 这里的ack后
seg.frg = 0;
seg.wnd = ikcp_wnd_unused(kcp); // 应答的时候携带了剩余的接收窗口大小
seg.una = kcp->rcv_nxt; // 已经处理到具体的分片 期望接收的下一个包
seg.len = 0;
seg.sn = 0;
seg.ts = 0;
// 逐一获取acklist中的sn和ts，编码成segment
// flush acknowledges
count = kcp->ackcount; // 需要应答的分片数量
for (i = 0; i < count; i++) {
size = (int)(ptr - buffer);
if (size + (int)IKCP_OVERHEAD > (int)kcp->mtu) {
ikcp_output(kcp, buffer, size);
ptr = buffer;
}
ikcp_ack_get(kcp, i, &seg.sn, &seg.ts); // 应答包 把时间戳发回去是为了能够计算RTT
// printf("get %p ack-> una:%u, sn:%d, ts:%u, sz:%d\n",kcp, seg.una, seg.sn, seg.ts, size); //
ptr = ikcp_encode_seg(ptr, &seg); // 编码segment协议头
}
kcp->ackcount = 0;
// probe window size (if remote window size equals zero)
if (kcp->rmt_wnd == 0) {
if (kcp->probe_wait == 0) { // 初始化探测间隔和下一次探测时间
kcp->probe_wait = IKCP_PROBE_INIT; // 默认7秒探测
kcp->ts_probe = kcp->current + kcp->probe_wait; // 下一次探测时间
}
else {
//远端窗口为0，发送过探测请求，但是已经超过下次探测的时间
//更新probe_wait，增加为IKCP_PROBE_INIT+ probe_wait /2,但满足KCP_PROBE_LIMIT
//更新下次探测时间 ts_probe与 探测变量 为 IKCP_ASK_SEND，立即发送探测消息
if (_itimediff(kcp->current, kcp->ts_probe) >= 0) { // 检测是否到了探测时间
if (kcp->probe_wait < IKCP_PROBE_INIT)
kcp->probe_wait = IKCP_PROBE_INIT;
kcp->probe_wait += kcp->probe_wait / 2;
if (kcp->probe_wait > IKCP_PROBE_LIMIT)
kcp->probe_wait = IKCP_PROBE_LIMIT;
kcp->ts_probe = kcp->current + kcp->probe_wait;
kcp->probe |= IKCP_ASK_SEND;
}
}
} else {
// 远端窗口正常，则不需要探测 远端窗口不等于0，更新下次探测时间与探测窗口等待时间为0，不发送窗口探测
kcp->ts_probe = 0;
kcp->probe_wait = 0;
}
// flush window probing commands
if (kcp->probe & IKCP_ASK_SEND) {
seg.cmd = IKCP_CMD_WASK; // 窗口探测 [询问对方窗口size]
size = (int)(ptr - buffer);
if (size + (int)IKCP_OVERHEAD > (int)kcp->mtu) {
ikcp_output(kcp, buffer, size);
ptr = buffer;
}
ptr = ikcp_encode_seg(ptr, &seg);
}
// flush window probing commands
if (kcp->probe & IKCP_ASK_TELL) {
seg.cmd = IKCP_CMD_WINS; // [告诉对方我方窗口size], 如果不为0，可以往我方发送数据
size = (int)(ptr - buffer);
if (size + (int)IKCP_OVERHEAD > (int)kcp->mtu) {
ikcp_output(kcp, buffer, size);
ptr = buffer;
}
ptr = ikcp_encode_seg(ptr, &seg);
}
kcp->probe = 0; //清空标识
// calculate window size 取发送窗口和远端窗口最小值得到拥塞窗口小
cwnd = _imin_(kcp->snd_wnd, kcp->rmt_wnd); // 当rmt_wnd为0的时候,
// 如果做了流控制则取配置拥塞窗口、发送窗口和远端窗口三者最小值
if (kcp->nocwnd == 0) cwnd = _imin_(kcp->cwnd, cwnd); // 进一步控制cwnd大小， kcp->cwnd拥塞窗口
// kcp->nocwnd = 1 少了一次_imin_的判断，可以尽量多发送数据
// move data from snd_queue to snd_buf
// 从snd_queue移动到snd_buf的数量不能超出对方的接收能力 此时如果
// 发送那些符合拥塞范围的数据分片
// kcp->snd_una = 10
// cwnd = 3
//
while (_itimediff(kcp->snd_nxt, kcp->snd_una + cwnd) < 0) {
IKCPSEG *newseg;
if (iqueue_is_empty(&kcp->snd_queue)) break;
// 从snd_queue读取segment
newseg = iqueue_entry(kcp->snd_queue.next, IKCPSEG, node);
iqueue_del(&newseg->node);
// 插入到snd buf 发送窗口
iqueue_add_tail(&newseg->node, &kcp->snd_buf); // 从发送队列添加到发送缓存
kcp->nsnd_que--;
kcp->nsnd_buf++;
//设置数据分片的属性
newseg->conv = kcp->conv;
newseg->cmd = IKCP_CMD_PUSH;
newseg->wnd = seg.wnd; // 告知对方当前的接收窗口
newseg->ts = current; // 当前时间
newseg->sn = kcp->snd_nxt++; // 序号
newseg->una = kcp->rcv_nxt; // 告诉对方可以发送的下一个包序号
newseg->resendts = current; // 当前发送的时间
newseg->rto = kcp->rx_rto; // 超时重传的时间， 重传间隔的时间
newseg->fastack = 0; // 是否快速重传
newseg->xmit = 0; // 重传次数
}
// calculate resent (1)使用快速重传时 resent = fastresend; (2)不使用时resent = 0xffffffff
resent = (kcp->fastresend > 0)? (IUINT32)kcp->fastresend : 0xffffffff; // 使用快速重传时
// fastresend = 2 , resent = 2
// fastresend=0, resent=0xffffffff
rtomin = (kcp->nodelay == 0)? (kcp->rx_rto >> 3) : 0; // 最小超时时间
// nodelay = 0 -> rtomin = rx_rto/8
// nodelay = 1 -> rtomin = 0
// flush data segments
// 只要还在snd_buf 说明对方还没有应答
// 发送snd buf的分片
// 1. 超时重传、需要重传的就重传
// 2. 新的segmen，正常发送
for (p = kcp->snd_buf.next; p != &kcp->snd_buf; p = p->next) {
IKCPSEG *segment = iqueue_entry(p, IKCPSEG, node);
int needsend = 0;
if (segment->xmit == 0) { //1 如果该报文是第一次传输，那么直接发送
needsend = 1;
segment->xmit++; // 发送次数技术
segment->rto = kcp->rx_rto; // 超时时间 间隔
segment->resendts = current + segment->rto + rtomin; // 下一次要发送的时间
}
else if (_itimediff(current, segment->resendts) >= 0) { //2 当前时间达到了重发时间，但并没有新的ack到达，出现丢包, 重传
needsend = 1;
segment->xmit++;
kcp->xmit++;
if (kcp->nodelay == 0) {
segment->rto += _imax_(segment->rto, (IUINT32)kcp->rx_rto);
} else {
IINT32 step = (kcp->nodelay < 2)?
((IINT32)(segment->rto)) : kcp->rx_rto;
segment->rto += step / 2; //1.5 1.5+(1.5/2)
}
segment->resendts = current + segment->rto;
lost = 1; // 丢包，反应到拥塞控制策略去了
}
else if (segment->fastack >= resent) { //3 segment的累计被跳过次数大于快速重传设定，需要重传
if ((int)segment->xmit <= kcp->fastlimit || // fastlimit默认为5, 超过这个次数后走正常重传方式
kcp->fastlimit <= 0) {
needsend = 1;
segment->xmit++; // 重传次数
// printf("req resend: sn:%u, fastack:%d, %d\n", segment->sn, segment->fastack, resent);
segment->fastack = 0; // fastack 这个值是发送方收到应答的其他序号时进行统计的，跳过的序号
segment->resendts = current + segment->rto; // 下一次重传的时间
change++;
}
}
if (needsend) {
int need;
segment->ts = current;
segment->wnd = seg.wnd; // 剩余接收窗口大小(接收窗口大小-接收队列大小), 告诉对方目前自己的接收能力
segment->una = kcp->rcv_nxt; // 待接收的下一个包序号, 即是告诉对方una之前的包都收到了, 你不用再发送发送缓存了
size = (int)(ptr - buffer);
need = IKCP_OVERHEAD + segment->len;
// 希望把多个小的seg封装到一个较大的udp报文发送 1000
if (size + need > (int)kcp->mtu) { // 小包封装成大包取发送 500 500 ， 按1000发
ikcp_output(kcp, buffer, size);
ptr = buffer;
}
ptr = ikcp_encode_seg(ptr, segment); // 把segment封装成线性buffer发送 头部+数据
if (segment->len > 0) {
memcpy(ptr, segment->data, segment->len);
ptr += segment->len;
}
if (segment->xmit >= kcp->dead_link) {
kcp->state = (IUINT32)-1;
}
}
}
// flash remain segments
size = (int)(ptr - buffer); // 剩余的数据
if (size > 0) { // 内部调用sendto
ikcp_output(kcp, buffer, size); // 最终只要有数据要发送，一定发出去
}
// update ssthresh 看完 用户态协议栈再来看这里的拥塞控制
if (change) { //如果发生了快速重传，拥塞窗口阈值降低为当前未确认包数量的一半或最小值
IUINT32 inflight = kcp->snd_nxt - kcp->snd_una;
kcp->ssthresh = inflight / 2;
if (kcp->ssthresh < IKCP_THRESH_MIN)
kcp->ssthresh = IKCP_THRESH_MIN;
kcp->cwnd = kcp->ssthresh + resent; // 动态调整拥塞控制窗口
kcp->incr = kcp->cwnd * kcp->mss;
}
if (lost) {
kcp->ssthresh = cwnd / 2;
if (kcp->ssthresh < IKCP_THRESH_MIN)
kcp->ssthresh = IKCP_THRESH_MIN;
kcp->cwnd = 1; //丢失则阈值减半, cwd 窗口保留为 1 动态调整拥塞控制窗口
kcp->incr = kcp->mss;
}
if (kcp->cwnd < 1) {
kcp->cwnd = 1;
kcp->incr = kcp->mss;
}
}
//---------------------------------------------------------------------
// update state (call it repeatedly, every 10ms-100ms), or you can ask
// ikcp_check when to call it again (without ikcp_input/_send calling).
// 'current' - current timestamp in millisec. 时钟来源
//---------------------------------------------------------------------
void ikcp_update(ikcpcb *kcp, IUINT32 current)
{
IINT32 slap;
kcp->current = current; // 超时重传要不要？
if (kcp->updated == 0) {
kcp->updated = 1;
kcp->ts_flush = kcp->current;
}
slap = _itimediff(kcp->current, kcp->ts_flush);
if (slap >= 10000 || slap < -10000) { // 至少10ms的间隔触发一次
kcp->ts_flush = kcp->current;
slap = 0;
}
if (slap >= 0) {
kcp->ts_flush += kcp->interval; // 先按interval叠加下一次要刷新的时间
if (_itimediff(kcp->current, kcp->ts_flush) >= 0) { // 如果下一次要刷新的时间已经落后则需要做校正
kcp->ts_flush = kcp->current + kcp->interval; // 使用当前时间 + interval进行校正
}
ikcp_flush(kcp); //interval = 5ms, ts_flush = 1000ms, current = 1005ms,
}
}
//---------------------------------------------------------------------
// Determine when should you invoke ikcp_update:
// returns when you should invoke ikcp_update in millisec, if there
// is no ikcp_input/_send calling. you can call ikcp_update in that
// time, instead of call update repeatly.
// Important to reduce unnacessary ikcp_update invoking. use it to
// schedule ikcp_update (eg. implementing an epoll-like mechanism,
// or optimize ikcp_update when handling massive kcp connections)
//---------------------------------------------------------------------
IUINT32 ikcp_check(const ikcpcb *kcp, IUINT32 current)
{
IUINT32 ts_flush = kcp->ts_flush;
IINT32 tm_flush = 0x7fffffff;
IINT32 tm_packet = 0x7fffffff;
IUINT32 minimal = 0;
struct IQUEUEHEAD *p;
if (kcp->updated == 0) {
return current;
}
if (_itimediff(current, ts_flush) >= 10000 ||
_itimediff(current, ts_flush) < -10000) {
ts_flush = current;
}
if (_itimediff(current, ts_flush) >= 0) {
return current;
}
tm_flush = _itimediff(ts_flush, current);
for (p = kcp->snd_buf.next; p != &kcp->snd_buf; p = p->next) {
const IKCPSEG *seg = iqueue_entry(p, const IKCPSEG, node);
IINT32 diff = _itimediff(seg->resendts, current);
if (diff <= 0) {
return current;
}
if (diff < tm_packet) tm_packet = diff;
}
minimal = (IUINT32)(tm_packet < tm_flush ? tm_packet : tm_flush);
if (minimal >= kcp->interval) minimal = kcp->interval;
return current + minimal;
}
int ikcp_setmtu(ikcpcb *kcp, int mtu)
{
char *buffer;
if (mtu < 50 || mtu < (int)IKCP_OVERHEAD)
return -1;
buffer = (char*)ikcp_malloc((mtu + IKCP_OVERHEAD) * 3);
if (buffer == NULL)
return -2;
kcp->mtu = mtu; // 本质是sendto的最大size值
kcp->mss = kcp->mtu - IKCP_OVERHEAD; // 默认mtu 1400, - IKCP_OVERHEAD = 1400-24= 1376
ikcp_free(kcp->buffer);
kcp->buffer = buffer;
return 0;
}
// 设置调度间隔
int ikcp_interval(ikcpcb *kcp, int interval)
{
if (interval > 5000) interval = 5000;
else if (interval < 10) interval = 10;
kcp->interval = interval;
return 0;
}
// 设置无延迟机制
int ikcp_nodelay(ikcpcb *kcp, int nodelay, int interval, int resend, int nc)
{
if (nodelay >= 0) {
kcp->nodelay = nodelay;
if (nodelay) {
kcp->rx_minrto = IKCP_RTO_NDL; //设置无延迟，设置相应的最小重传时间
}
else {
kcp->rx_minrto = IKCP_RTO_MIN; //设置正常的最小重传时间
}
}
if (interval >= 0) { //设置调度间隔
if (interval > 5000) interval = 5000;
else if (interval < 10) interval = 10;
kcp->interval = interval;
}
if (resend >= 0) { //设置快速重传数
kcp->fastresend = resend;
// printf("init fastresend:%d\n", kcp->fastresend);
}
if (nc >= 0) { // 0 不关闭流控， 1关闭流控
kcp->nocwnd = nc;
}
return 0;
}
// 设置发送和接收窗口
int ikcp_wndsize(ikcpcb *kcp, int sndwnd, int rcvwnd)
{
if (kcp) {
if (sndwnd > 0) {
kcp->snd_wnd = sndwnd;
}
if (rcvwnd > 0) { // must >= max fragment size
kcp->rcv_wnd = _imax_(rcvwnd, IKCP_WND_RCV);
}
}
return 0;
}
// 我们可以获取等待发送的, 如果有就flush?
int ikcp_waitsnd(const ikcpcb *kcp)
{
return kcp->nsnd_buf + kcp->nsnd_que;
}
// read conv 获取会话id
IUINT32 ikcp_getconv(const void *ptr)
{
IUINT32 conv;
ikcp_decode32u((const char*)ptr, &conv);
return conv;
}
ikcp.h
//=====================================================================
//
// KCP - A Better ARQ Protocol Implementation
// skywind3000 (at) gmail.com, 2010-2011
//
// Features:
// + Average RTT reduce 30% - 40% vs traditional ARQ like tcp.
// + Maximum RTT reduce three times vs tcp.
// + Lightweight, distributed as a single source file.
//
//=====================================================================
#ifndef __IKCP_H__
#define __IKCP_H__
#include <stddef.h>
#include <stdlib.h>
#include <assert.h>
//=====================================================================
// 32BIT INTEGER DEFINITION
//=====================================================================
#ifndef __INTEGER_32_BITS__
#define __INTEGER_32_BITS__
#if defined(_WIN64) || defined(WIN64) || defined(__amd64__) || \
defined(__x86_64) || defined(__x86_64__) || defined(_M_IA64) || \
defined(_M_AMD64)
typedef unsigned int ISTDUINT32;
typedef int ISTDINT32;
#elif defined(_WIN32) || defined(WIN32) || defined(__i386__) || \
defined(__i386) || defined(_M_X86)
typedef unsigned long ISTDUINT32;
typedef long ISTDINT32;
#elif defined(__MACOS__)
typedef UInt32 ISTDUINT32;
typedef SInt32 ISTDINT32;
#elif defined(__APPLE__) && defined(__MACH__)
#include <sys/types.h>
typedef u_int32_t ISTDUINT32;
typedef int32_t ISTDINT32;
#elif defined(__BEOS__)
#include <sys/inttypes.h>
typedef u_int32_t ISTDUINT32;
typedef int32_t ISTDINT32;
#elif (defined(_MSC_VER) || defined(__BORLANDC__)) && (!defined(__MSDOS__))
typedef unsigned __int32 ISTDUINT32;
typedef __int32 ISTDINT32;
#elif defined(__GNUC__)
#include <stdint.h>
typedef uint32_t ISTDUINT32;
typedef int32_t ISTDINT32;
#else
typedef unsigned long ISTDUINT32;
typedef long ISTDINT32;
#endif
#endif
//=====================================================================
// Integer Definition
//=====================================================================
#ifndef __IINT8_DEFINED
#define __IINT8_DEFINED
typedef char IINT8;
#endif
#ifndef __IUINT8_DEFINED
#define __IUINT8_DEFINED
typedef unsigned char IUINT8;
#endif
#ifndef __IUINT16_DEFINED
#define __IUINT16_DEFINED
typedef unsigned short IUINT16;
#endif
#ifndef __IINT16_DEFINED
#define __IINT16_DEFINED
typedef short IINT16;
#endif
#ifndef __IINT32_DEFINED
#define __IINT32_DEFINED
typedef ISTDINT32 IINT32;
#endif
#ifndef __IUINT32_DEFINED
#define __IUINT32_DEFINED
typedef ISTDUINT32 IUINT32;
#endif
#ifndef __IINT64_DEFINED
#define __IINT64_DEFINED
#if defined(_MSC_VER) || defined(__BORLANDC__)
typedef __int64 IINT64;
#else
typedef long long IINT64;
#endif
#endif
#ifndef __IUINT64_DEFINED
#define __IUINT64_DEFINED
#if defined(_MSC_VER) || defined(__BORLANDC__)
typedef unsigned __int64 IUINT64;
#else
typedef unsigned long long IUINT64;
#endif
#endif
#ifndef INLINE
#if defined(__GNUC__)
#if (__GNUC__ > 3) || ((__GNUC__ == 3) && (__GNUC_MINOR__ >= 1))
#define INLINE __inline__ __attribute__((always_inline))
#else
#define INLINE __inline__
#endif
#elif (defined(_MSC_VER) || defined(__BORLANDC__) || defined(__WATCOMC__))
#define INLINE __inline
#else
#define INLINE
#endif
#endif
#if (!defined(__cplusplus)) && (!defined(inline))
#define inline INLINE
#endif
//=====================================================================
// QUEUE DEFINITION
//=====================================================================
#ifndef __IQUEUE_DEF__
#define __IQUEUE_DEF__
struct IQUEUEHEAD {
struct IQUEUEHEAD *next, *prev;
};
typedef struct IQUEUEHEAD iqueue_head;
//---------------------------------------------------------------------
// queue init
//---------------------------------------------------------------------
#define IQUEUE_HEAD_INIT(name) { &(name), &(name) }
#define IQUEUE_HEAD(name) \
struct IQUEUEHEAD name = IQUEUE_HEAD_INIT(name)
#define IQUEUE_INIT(ptr) ( \
(ptr)->next = (ptr), (ptr)->prev = (ptr))
#define IOFFSETOF(TYPE, MEMBER) ((size_t) &((TYPE *)0)->MEMBER)
#define ICONTAINEROF(ptr, type, member) ( \
(type*)( ((char*)((type*)ptr)) - IOFFSETOF(type, member)) )
#define IQUEUE_ENTRY(ptr, type, member) ICONTAINEROF(ptr, type, member)
//---------------------------------------------------------------------
// queue operation
//---------------------------------------------------------------------
#define IQUEUE_ADD(node, head) ( \
(node)->prev = (head), (node)->next = (head)->next, \
(head)->next->prev = (node), (head)->next = (node))
#define IQUEUE_ADD_TAIL(node, head) ( \
(node)->prev = (head)->prev, (node)->next = (head), \
(head)->prev->next = (node), (head)->prev = (node))
#define IQUEUE_DEL_BETWEEN(p, n) ((n)->prev = (p), (p)->next = (n))
#define IQUEUE_DEL(entry) (\
(entry)->next->prev = (entry)->prev, \
(entry)->prev->next = (entry)->next, \
(entry)->next = 0, (entry)->prev = 0)
#define IQUEUE_DEL_INIT(entry) do { \
IQUEUE_DEL(entry); IQUEUE_INIT(entry); } while (0)
#define IQUEUE_IS_EMPTY(entry) ((entry) == (entry)->next)
#define iqueue_init IQUEUE_INIT
#define iqueue_entry IQUEUE_ENTRY
#define iqueue_add IQUEUE_ADD
#define iqueue_add_tail IQUEUE_ADD_TAIL
#define iqueue_del IQUEUE_DEL
#define iqueue_del_init IQUEUE_DEL_INIT
#define iqueue_is_empty IQUEUE_IS_EMPTY
#define IQUEUE_FOREACH(iterator, head, TYPE, MEMBER) \
for ((iterator) = iqueue_entry((head)->next, TYPE, MEMBER); \
&((iterator)->MEMBER) != (head); \
(iterator) = iqueue_entry((iterator)->MEMBER.next, TYPE, MEMBER))
#define iqueue_foreach(iterator, head, TYPE, MEMBER) \
IQUEUE_FOREACH(iterator, head, TYPE, MEMBER)
#define iqueue_foreach_entry(pos, head) \
for( (pos) = (head)->next; (pos) != (head) ; (pos) = (pos)->next )
#define __iqueue_splice(list, head) do { \
iqueue_head *first = (list)->next, *last = (list)->prev; \
iqueue_head *at = (head)->next; \
(first)->prev = (head), (head)->next = (first); \
(last)->next = (at), (at)->prev = (last); } while (0)
#define iqueue_splice(list, head) do { \
if (!iqueue_is_empty(list)) __iqueue_splice(list, head); } while (0)
#define iqueue_splice_init(list, head) do { \
iqueue_splice(list, head); iqueue_init(list); } while (0)
#ifdef _MSC_VER
#pragma warning(disable:4311)
#pragma warning(disable:4312)
#pragma warning(disable:4996)
#endif
#endif
//---------------------------------------------------------------------
// BYTE ORDER & ALIGNMENT
//---------------------------------------------------------------------
#ifndef IWORDS_BIG_ENDIAN
#ifdef _BIG_ENDIAN_
#if _BIG_ENDIAN_
#define IWORDS_BIG_ENDIAN 1
#endif
#endif
#ifndef IWORDS_BIG_ENDIAN
#if defined(__hppa__) || \
defined(__m68k__) || defined(mc68000) || defined(_M_M68K) || \
(defined(__MIPS__) && defined(__MIPSEB__)) || \
defined(__ppc__) || defined(__POWERPC__) || defined(_M_PPC) || \
defined(__sparc__) || defined(__powerpc__) || \
defined(__mc68000__) || defined(__s390x__) || defined(__s390__)
#define IWORDS_BIG_ENDIAN 1
#endif
#endif
#ifndef IWORDS_BIG_ENDIAN
#define IWORDS_BIG_ENDIAN 0
#endif
#endif
#ifndef IWORDS_MUST_ALIGN
#if defined(__i386__) || defined(__i386) || defined(_i386_)
#define IWORDS_MUST_ALIGN 0
#elif defined(_M_IX86) || defined(_X86_) || defined(__x86_64__)
#define IWORDS_MUST_ALIGN 0
#elif defined(__amd64) || defined(__amd64__)
#define IWORDS_MUST_ALIGN 0
#else
#define IWORDS_MUST_ALIGN 1
#endif
#endif
//=====================================================================
// SEGMENT 每个分片
//=====================================================================
struct IKCPSEG
{
struct IQUEUEHEAD node;
IUINT32 conv; // 会话编号,和TCP的con一样,确保双方需保证conv相同,相互的数据包才能被接收.conv唯一标识一个会话
IUINT32 cmd; // 区分不同的分片.IKCP_CMD_PUSH数据分片;IKCP_CMD_ACK:ack分片;IKCP_CMD_WASK:请求告知窗口大小;IKCP_CMD_WINS:告知窗口大小
IUINT32 frg; // 标识segment分片ID,用户数据可能被分成多个kcp包发送
IUINT32 wnd; // 剩余接收窗口大小(接收窗口大小-接收队列大小),发送方的发送窗口不能超过接收方给出的数值
IUINT32 ts; // 发送时刻的时间戳
IUINT32 sn; // 分片segment的序号,按1累加递增
IUINT32 una; // 待接收消息序号(接收滑动窗口左侧).对于未丢包的网络来说,una是下一个可接收的序号,如收到sn=10的包,una为11
IUINT32 len; // 数据长度
IUINT32 resendts; // 下次超时重传时间戳
IUINT32 rto; //该分片的超时等待时间,其计算方法同TCP
IUINT32 fastack; // 收到ack时计算该分片被跳过的累计次数,此字段用于快速重传,自定义需要几次确认开始快速重传
IUINT32 xmit; // 发送分片的次数,每发一次加1.发送的次数对RTO的计算有影响,但是比TCP来说,影响会小一些.
char data[1];
};
// ts_xxx 时间戳相关
// wnd 窗口相关
// snd发送相关
// rcv接收相关
struct IKCPCB
{
IUINT32 conv; // 标识会话
IUINT32 mtu; // 最大传输单元,默认数据为1400,最小为50
IUINT32 mss; // 最大分片大小,不大于mtu
IUINT32 state; // 连接状态(0xffffffff表示断开连接)
IUINT32 snd_una; // 第一个未确认的包
IUINT32 snd_nxt; // 下一个待分配包的序号，这个值实际是用来分配序号的
IUINT32 rcv_nxt; // 待接收消息序号.为了保证包的顺序,接收方会维护一个接收窗口,接收窗口有一个起始序号rcv_nxt
// 以及尾序号rcv_nxt + rcv_wnd(接收窗口大小)
IUINT32 ts_recent;
IUINT32 ts_lastack;
IUINT32 ssthresh; // 拥塞窗口的阈值
IINT32 rx_rttval; // RTT的变化量,代表连接的抖动情况
IINT32 rx_srtt; // smoothed round trip time，平滑后的RTT；
IINT32 rx_rto; // 收ACK接收延迟计算出来的重传超时时间
IINT32 rx_minrto; // 最小重传超时时间
IUINT32 snd_wnd; // 发送窗口大小
IUINT32 rcv_wnd; // 接收窗口大小，本质上而言如果接收端一直不去读取数据则rcv_queue就会满(达到rcv_wnd)
IUINT32 rmt_wnd; // 远端接收窗口大小
IUINT32 cwnd; // 拥塞窗口大小, 动态变化
IUINT32 probe; // 探查变量, IKCP_ASK_TELL表示告知远端窗口大小。IKCP_ASK_SEND表示请求远端告知窗口大小；
IUINT32 current;
IUINT32 interval; // 内部flush刷新间隔，对系统循环效率有非常重要影响, 间隔小了cpu占用率高, 间隔大了响应慢
IUINT32 ts_flush; // 下次flush刷新的时间戳
IUINT32 xmit; // 发送segment的次数, 当segment的xmit增加时，xmit增加(重传除外)
IUINT32 nrcv_buf; // 接收缓存中的消息数量
IUINT32 nsnd_buf; // 发送缓存中的消息数量
IUINT32 nrcv_que; // 接收队列中消息数量
IUINT32 nsnd_que; // 发送队列中消息数量
IUINT32 nodelay; // 是否启动无延迟模式。无延迟模式rtomin将设置为0，拥塞控制不启动；
IUINT32 updated; //是 否调用过update函数的标识；
IUINT32 ts_probe; // 下次探查窗口的时间戳；
IUINT32 probe_wait; // 探查窗口需要等待的时间；
IUINT32 dead_link; // 最大重传次数，被认为连接中断；
IUINT32 incr; // 可发送的最大数据量；
struct IQUEUEHEAD snd_queue; //发送消息的队列
struct IQUEUEHEAD rcv_queue; //接收消息的队列, 是已经确认可以供用户读取的数据
struct IQUEUEHEAD snd_buf; //发送消息的缓存 和snd_queue有什么区别
struct IQUEUEHEAD rcv_buf; //接收消息的缓存, 还不能直接供用户读取的数据
IUINT32 *acklist; //待发送的ack的列表 当收到一个数据报文时，将其对应的 ACK 报文的 sn 号以及时间戳 ts
//同时加入到acklist 中，即形成如 [sn1, ts1, sn2, ts2 …] 的列表
IUINT32 ackcount; //是当前计数, 记录 acklist 中存放的 ACK 报文的数量
IUINT32 ackblock; //是容量, acklist 数组的可用长度，当 acklist 的容量不足时，需要进行扩容
void *user; // 指针，可以任意放置代表用户的数据，也可以设置程序中需要传递的变量；
char *buffer; //
int fastresend; // 触发快速重传的重复ACK个数；
int fastlimit;
int nocwnd; // 取消拥塞控制
int stream; // 是否采用流传输模式
int logmask; // 日志的类型，如IKCP_LOG_IN_DATA，方便调试
int (*output)(const char *buf, int len, struct IKCPCB *kcp, void *user);//发送消息的回调函数
void (*writelog)(const char *log, struct IKCPCB *kcp, void *user); // 写日志的回调函数
};
typedef struct IKCPCB ikcpcb;
// 日志相关的开关
#define IKCP_LOG_OUTPUT 1
#define IKCP_LOG_INPUT 2
#define IKCP_LOG_SEND 4
#define IKCP_LOG_RECV 8
#define IKCP_LOG_IN_DATA 16
#define IKCP_LOG_IN_ACK 32
#define IKCP_LOG_IN_PROBE 64
#define IKCP_LOG_IN_WINS 128
#define IKCP_LOG_OUT_DATA 256
#define IKCP_LOG_OUT_ACK 512
#define IKCP_LOG_OUT_PROBE 1024
#define IKCP_LOG_OUT_WINS 2048
#ifdef __cplusplus
extern "C" {
#endif
//---------------------------------------------------------------------
// interface
//---------------------------------------------------------------------
// create a new kcp control object, 'conv' must equal in two endpoint
// from the same connection. 'user' will be passed to the output callback
// output callback can be setup like this: 'kcp->output = my_udp_output'
// 创建一个kcp对象，每个不同的会话产生不同的对象.
// 因为kcp协议本身没有提供网络部分的代码,所以需要将udp发送函数的回调设置到kcp中,在有需要的时候,调用回调函数即可.
ikcpcb* ikcp_create(IUINT32 conv, void *user);
// release kcp control object
// 释放kcp对象
void ikcp_release(ikcpcb *kcp);
// set output callback, which will be invoked by kcp
// 设置输出函数，需要的时候将被kcp内部调用
void ikcp_setoutput(ikcpcb *kcp, int (*output)(const char *buf, int len,
ikcpcb *kcp, void *user));
// user/upper level recv: returns size, returns below zero for EAGAIN
// 接收数据
int ikcp_recv(ikcpcb *kcp, char *buffer, int len);
// user/upper level send, returns below zero for error
// 发送数据
int ikcp_send(ikcpcb *kcp, const char *buffer, int len);
// update state (call it repeatedly, every 10ms-100ms), or you can ask
// ikcp_check when to call it again (without ikcp_input/_send calling).
// 'current' - current timestamp in millisec.
// 更新状态
void ikcp_update(ikcpcb *kcp, IUINT32 current);
// Determine when should you invoke ikcp_update:
// returns when you should invoke ikcp_update in millisec, if there
// is no ikcp_input/_send calling. you can call ikcp_update in that
// time, instead of call update repeatly.
// Important to reduce unnacessary ikcp_update invoking. use it to
// schedule ikcp_update (eg. implementing an epoll-like mechanism,
// or optimize ikcp_update when handling massive kcp connections)
IUINT32 ikcp_check(const ikcpcb *kcp, IUINT32 current);
// when you received a low level packet (eg. UDP packet), call it
// 输入数据，主要是将udp协议收到的数据传给kcp进行处理
int ikcp_input(ikcpcb *kcp, const char *data, long size);
// flush pending data
// 调用 ikcp_flush 时将数据从 snd_queue 中 移入到 snd_buf 中，然后调用 kcp->output() 发送。
void ikcp_flush(ikcpcb *kcp);
// check the size of next message in the recv queue
int ikcp_peeksize(const ikcpcb *kcp);
// change MTU size, default is 1400
// 设置MTU大小
int ikcp_setmtu(ikcpcb *kcp, int mtu);
// set maximum window size: sndwnd=32, rcvwnd=32 by default
// 设置最大的发送窗口和接收窗口
int ikcp_wndsize(ikcpcb *kcp, int sndwnd, int rcvwnd);
// get how many packet is waiting to be sent
// 有多少数据还没有被发送, 包括发送缓存的(即使发送出去但是没有ack也算)
int ikcp_waitsnd(const ikcpcb *kcp);
// fastest: ikcp_nodelay(kcp, 1, 20, 2, 1)
// nodelay: 0:disable(default), 1:enable
// interval: internal update timer interval in millisec, default is 100ms
// resend: 0:disable fast resend(default), 1:enable fast resend
// nc: 0:normal congestion control(default), 1:disable congestion control
/*
工作模式
nodelay ：是否启用 nodelay模式，0不启用；1启用。
interval ：协议内部工作的 interval，单位毫秒，比如 10ms或者 20ms
resend ：快速重传模式，默认0关闭，可以设置2（2次ACK跨越将会直接重传）
nc ：是否关闭流控，默认是0代表不关闭，1代表关闭。
普通模式： ikcp_nodelay(kcp, 0, 40, 0, 0);
极速模式： ikcp_nodelay(kcp, 1, 10, 2, 1)
*/
int ikcp_nodelay(ikcpcb *kcp, int nodelay, int interval, int resend, int nc);
void ikcp_log(ikcpcb *kcp, int mask, const char *fmt, ...);
// setup allocator
void ikcp_allocator(void* (*new_malloc)(size_t), void (*new_free)(void*));
// read conv
IUINT32 ikcp_getconv(const void *ptr);
#ifdef __cplusplus
}
#endif
#endif
文章参考与<零声教育>的C/C++linux服务期高级架构系统教程学习:https://ke.qq.com/course/417774?flowToken=1010783
标签：__,UDP,IUINT32,IKCP,ikcp,seg,unix,kcp,3.3 From： https://www.cnblogs.com/kongweisi/p/17016446.html