Posted at 2021-01-25 网络 kcp

简介

地址,一句话描述,牺牲带宽占用来降低延迟。

KCP 源码注释

头文件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;
	
    // 会话ID,发送方和接收方的会话必须要一致才能收发
	IUINT32 conv;
	
	// 报文类型,
	// ICKP_CMD_PUSH(数据报文)、
	// IKCP_CMD_ACK(确认报文)、
	// IKCP_CMD_WASK(窗口探测报文),询问对端剩余接收窗口大小
	// IKCP_CMD_WINS(窗口通知报文),
	IUINT32 cmd;
	
	// 分片数量,表示后面还有多少报文属于同一个包,0表示最后一个包,应用层的数据会被分包
	IUINT32 frg;
	
	// 发送方剩余接收窗口大小, 接收窗口大小-接收窗口队列大小。发送方发送窗口不能大于接收方接收窗口大小
	IUINT32 wnd;
	
	// 发送的时间戳
	IUINT32 ts;
	
	// 报文的编号，按1累加，和frg有啥区别。该编号用于ack
	IUINT32 sn;
	
	// 待接收的消息的编号，ack相关,表示我想要收数据报的sn=una的没有收到。
	// 未发生丢包的话，una为下一个可接收数据包序号。当前收到的sn = 10, una = 11。
	IUINT32 una;
	
	// 数据data的长度
	IUINT32 len;
	
	// 下一次超时重发的时间戳
	IUINT32 resendts;
	
	// 该分片超时重传的等待时间
	IUINT32 rto;
	
	// ack被跳过大小，会影响重传。如果发送1,2,3,4,接收到1,3,4。那么2的数据包已经被跳过2次了（4 - 2 = 2）
	IUINT32 fastack;
	
	// 发送该分片的次数
	IUINT32 xmit;
	
	// data指针
	char data[1];
};


//---------------------------------------------------------------------
// IKCPCB
//---------------------------------------------------------------------
struct IKCPCB
{
    // conv  : 会话标识
    // mtu   : 最大传输单元（Maximum Transmission Unit）
    // mss   : 最大报文段大小, mss = mtu - 包头长度(24)
    // state : 连接状态, 0 连接建立, -1(0xffffffff) 断开连接
	IUINT32 conv, mtu, mss, state;
	
	// snd_una : 发送缓冲区中最小还未确认送达的报文段标号，比它小的已经确认送达了
	// snd_nxt : 下一个等待发送的报文段的编号
	// rcv_nxt : 下一个等待接收的报文段的编号
	IUINT32 snd_una, snd_nxt, rcv_nxt;
	
	// ts_recent  : 未使用
	// ts_lastack : 未使用
	// ssthresh   : 慢启动阈值Slow Start Threshold
	IUINT32 ts_recent, ts_lastack, ssthresh;
	
	
	// rx_rttval、rx_srtt、rx_minrto 计算出 => rx_rto
	IINT32 rx_rttval, rx_srtt, rx_rto, rx_minrto;
	
	// snd_wnd : 发送窗口大小
	// rcv_wnd : 接收窗口大小
	// rmt_wnd : 对端剩余接收窗口大小
	// cwnd    : congestion window 拥塞窗口,用于拥塞控制
	// probe   : 是否要发送控制报文的标志
	IUINT32 snd_wnd, rcv_wnd, rmt_wnd, cwnd, probe;
	
	// current  : 当前时间
	// interval : flush的时间粒度,多长进行flush
	// ts_flush : 下一次flush时间
	// xmit     : 该链接超时重传的总次数
	IUINT32 current, interval, ts_flush, xmit;
	
	// nrcv_buf、nsnd_buf 接收发送缓冲区长度
	IUINT32 nrcv_buf, nsnd_buf;
	
	// 接收/发送队列长度
	IUINT32 nrcv_que, nsnd_que;
	
	// nodelay : 使用启用快速模式
	// updated : 使用调用过ikcp_update
	IUINT32 nodelay, updated;
	
	// ts_probe、probe_wait :用于确定何时需要发送窗口询问报文
	IUINT32 ts_probe, probe_wait;
	
	// dead_link : 当一个报文发送超时次数达到deal_link次,则可以认为连接断开 (state = 0xffffffff)
	// incr      : 用于计算cwnd
	IUINT32 dead_link, incr;
	
	// 发送队列
	struct IQUEUEHEAD snd_queue;
	
	// 接收队列
	struct IQUEUEHEAD rcv_queue;
	
	// 发送缓冲区,能在发送缓冲区的数据
	// 1. 新加入snd_buf中的,从未发送过的报文
	// 2. 发送过，在RTO内未收到ACK报文
	// 3. 发送过，ACK失序报文，需要执行快速重传
	struct IQUEUEHEAD snd_buf;
	
	// 接收缓冲区
	struct IQUEUEHEAD rcv_buf;
	
	// ACK列表,待发送的ACK的信息会存在列表中，flush时一并发送
	IUINT32 *acklist;
	
	// ACK列表长度
	IUINT32 ackcount;
	
	// ACK列表容量
	IUINT32 ackblock;
	
	// 自定义的数据
	void *user;
	
	// flush 临时缓冲区
	char *buffer;
	
	// ACK失序次数大于等于fastresend，触发快速重传
	int fastresend;
	
	// 传输次数小于fastlimit,执行快速重传
	int fastlimit;
	
	// nocwnd : 是否不考虑拥塞窗口
	// stream : 是否开启流模式,开启后可能会粘包
	int nocwnd, stream;
	
	// 用于控制日志
	int logmask;
	
	// 下层传输协议输出函数
	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'
ikcpcb* ikcp_create(IUINT32 conv, void *user);

// release kcp control object
void ikcp_release(ikcpcb *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));

// user/upper level recv: returns size, returns below zero for EAGAIN
// 用户上层调用的接收数据的函数,这个函数得到的数据是通过kcp算法处理的
// 可以认为该数据一定是对端发送成功的，
int ikcp_recv(ikcpcb *kcp, char *buffer, int len);

// user/upper level send, returns below zero for error
// 用户上层调用的发送数据的函数,该函数内创建报文实例,将报文放入sdn_queue中
// 错误小于0
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. 
// 更新函数,需要不断重复调用该函数,用来驱动数据的收发。会调用ikcp_flush函数进行数据发送
// 会根据传进来的时间和上一次进行flush的时间进行比较,需要满足
// 当前时间戳(current) - 上一次flush的时间戳(kcp->ts_flush) >= 0,
// 则说明需要发送了
// 将kcp->ts_flush += kcp->interval之后 和 kcp->current进行比较。
// 如果current还是大过kcp->ts_flush,将kcp->ts_flush设置为kcp->current + kcp->interval
// 为什么要这样？需要控制ikcp_flush发送间隔,
// 如果调用ikcp_update的时间间隔太短(距离上一次调用间隔小于interval时间),就不会触发ikcp_flush了
// 我要怎么知道我调用update的时机?
// 通过调用ickp_check函数可以知道下一次调用update生效的时间距离现在多久
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的话，调用recvfrom之后得到的数据需要调用这个函数
int ikcp_input(ikcpcb *kcp, const char *data, long size);

// flush pending data
// 该函数会完成如下动作
// + 发送ACK列表中的ACK
// 
// + 检查是否需要发送窗口探测和通知报文
// 
// + 计算拥塞窗口大小
//   从自身的发送窗口大小和远端接收窗口中选择最小的作为拥塞窗口
// 
// + 根据上一步计算得到的拥塞窗口，从发送队列中取出符合拥塞发送窗口大小的数据放入发送缓冲区kcp->snd_buf中,
//   上一步计算是为了这一步取出的数据大小能够发送到对端
// 
// + 从kcp->snd_buf中取出需要先发送的数据
//   新报文,需要重传的报文，ACK失序的报文
// 
// + 将kcp->snd_buf中剩余的数据发送出去
// 
// + 根据丢包计算ssthresh和cwnd
// 
// 数据发送的优先级:
// 1. ACK报文
// 2. 响应对端窗口询问报文
// 3. 询问对端窗口大小报文
// 4. 需要先发送的数据报文
// 5. 剩余的报文
// 
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
// 更改KCP的MTU，注意MTU不能小于50或小于24,
// 其中24为IKCP_OVERHEAD,KCP协议的头部大小,小过这个值,这个协议没法用的。
// 为什么不能小于50呢？？
// 设置MTU会影响切片的分包,最大分片大小kcp->mss = kcp->mtu - IKCP_OVERHEAD
int ikcp_setmtu(ikcpcb *kcp, int mtu);

// set maximum window size: sndwnd=32, rcvwnd=32 by default
// 设置发送和接收窗口大小
// 发送窗口>0就行，但是接收窗口大小有要求，必须>=128（IKCP_WND_RCV）,传小了被设成IKCP_WND_RCV
int ikcp_wndsize(ikcpcb *kcp, int sndwnd, int rcvwnd);

// get how many packet is waiting to be sent
// 获得当前等待发送数据量大小:发送队列大小 + 发送缓冲区大小
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
// 设置ikcp的模式
int ikcp_nodelay(ikcpcb *kcp, int nodelay, int interval, int resend, int nc);


// ikcp_log日志模块,用来查看日志kcp的日志,
// 使用前需要先设置kcp->writelog函数
void ikcp_log(ikcpcb *kcp, int mask, const char *fmt, ...);

// setup allocator
// 设置内存分配和释放器,默认使用malloc
void ikcp_allocator(void* (*new_malloc)(size_t), void (*new_free)(void*));

// read conv
// 可以传入UDP接收到数据,会返回数据包的conv值，
// 可以在调用ikcp_input的时候做不同的过滤,只把符合kcp协议的包放入
// 可以用来融合其他协议
IUINT32 ikcp_getconv(const void *ptr);


#ifdef __cplusplus
}
#endif

#endif

源文件ickp.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;		// no delay min rto
const IUINT32 IKCP_RTO_MIN = 100;		// normal min rto
const IUINT32 IKCP_RTO_DEF = 200;
const IUINT32 IKCP_RTO_MAX = 60000;
const IUINT32 IKCP_CMD_PUSH = 81;		// cmd: push data
const IUINT32 IKCP_CMD_ACK  = 82;		// cmd: ack
const IUINT32 IKCP_CMD_WASK = 83;		// cmd: window probe (ask)
const IUINT32 IKCP_CMD_WINS = 84;		// cmd: window size (tell)
const IUINT32 IKCP_ASK_SEND = 1;		// need to send IKCP_CMD_WASK
const IUINT32 IKCP_ASK_TELL = 2;		// need to send 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;
const IUINT32 IKCP_ACK_FAST	= 3;
const IUINT32 IKCP_INTERVAL	= 100;
const IUINT32 IKCP_OVERHEAD = 24;
const IUINT32 IKCP_DEADLINK = 20;
const IUINT32 IKCP_THRESH_INIT = 2;
const IUINT32 IKCP_THRESH_MIN = 2;
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, 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
//---------------------------------------------------------------------
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;
	kcp->mss = kcp->mtu - IKCP_OVERHEAD;
	kcp->stream = 0;

	kcp->buffer = (char*)ikcp_malloc((kcp->mtu + IKCP_OVERHEAD) * 3);
	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;
	kcp->logmask = 0;
	kcp->ssthresh = IKCP_THRESH_INIT;
	kcp->fastresend = 0;
	kcp->fastlimit = IKCP_FASTACK_LIMIT;
	kcp->nocwnd = 0;
	kcp->xmit = 0;
	kcp->dead_link = IKCP_DEADLINK;
	kcp->output = NULL;
	kcp->writelog = NULL;

	return kcp;
}


//---------------------------------------------------------------------
// release a new kcpcb
//---------------------------------------------------------------------
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
//---------------------------------------------------------------------
int ikcp_recv(ikcpcb *kcp, char *buffer, int len)
{
	struct IQUEUEHEAD *p;
	int ispeek = (len < 0)? 1 : 0;
	int peeksize;
	int recover = 0;
	IKCPSEG *seg;
	assert(kcp);

	if (iqueue_is_empty(&kcp->rcv_queue))
		return -1;

	if (len < 0) len = -len;

	peeksize = ikcp_peeksize(kcp);

	if (peeksize < 0) 
		return -2;

	if (peeksize > len) 
		return -3;

	if (kcp->nrcv_que >= kcp->rcv_wnd)
		recover = 1;

	// merge fragment
	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);
			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--;
		}

		if (fragment == 0) 
			break;
	}

	assert(len == peeksize);

	// move available data from rcv_buf -> rcv_queue
	while (! iqueue_is_empty(&kcp->rcv_buf)) {
		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++;
		}	else {
			break;
		}
	}

	// fast recover
	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
//---------------------------------------------------------------------
int ikcp_send(ikcpcb *kcp, const char *buffer, int len)
{
	IKCPSEG *seg;
	int count, i;

	assert(kcp->mss > 0);
	if (len < 0) return -1;

	// append to previous segment in streaming mode (if possible)
	if (kcp->stream != 0) {
		if (!iqueue_is_empty(&kcp->snd_queue)) {
			IKCPSEG *old = iqueue_entry(kcp->snd_queue.prev, IKCPSEG, node);
			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);
				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;
		}
	}

	if (len <= (int)kcp->mss) count = 1;
	else count = (len + kcp->mss - 1) / kcp->mss;

	if (count >= (int)IKCP_WND_RCV) return -2;

	if (count == 0) count = 1;

	// fragment
	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->frg = (kcp->stream == 0)? (count - i - 1) : 0;
		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
//---------------------------------------------------------------------
static void ikcp_update_ack(ikcpcb *kcp, IINT32 rtt)
{
	IINT32 rto = 0;
	if (kcp->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;
		kcp->rx_srtt = (7 * kcp->rx_srtt + rtt) / 8;
		if (kcp->rx_srtt < 1) kcp->rx_srtt = 1;
	}
	rto = kcp->rx_srtt + _imax_(kcp->interval, 4 * kcp->rx_rttval);
	kcp->rx_rto = _ibound_(kcp->rx_minrto, rto, IKCP_RTO_MAX);
}

static void ikcp_shrink_buf(ikcpcb *kcp)
{
	struct IQUEUEHEAD *p = kcp->snd_buf.next;
	if (p != &kcp->snd_buf) {
		IKCPSEG *seg = iqueue_entry(p, IKCPSEG, node);
		kcp->snd_una = seg->sn;
	}	else {
		kcp->snd_una = kcp->snd_nxt;
	}
}

static void ikcp_parse_ack(ikcpcb *kcp, IUINT32 sn)
{
	struct IQUEUEHEAD *p, *next;

	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;
		}
	}
}

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;
		if (_itimediff(una, seg->sn) > 0) {
			iqueue_del(p);
			ikcp_segment_delete(kcp, seg);
			kcp->nsnd_buf--;
		}	else {
			break;
		}
	}
}

static void ikcp_parse_fastack(ikcpcb *kcp, IUINT32 sn, IUINT32 ts)
{
	struct IQUEUEHEAD *p, *next;

	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 (_itimediff(sn, seg->sn) < 0) {
			break;
		}
		else if (sn != seg->sn) {
		#ifndef IKCP_FASTACK_CONSERVE
			seg->fastack++;
		#else
			if (_itimediff(ts, seg->ts) >= 0)
				seg->fastack++;
		#endif
		}
	}
}


//---------------------------------------------------------------------
// ack append
//---------------------------------------------------------------------
static void ikcp_ack_push(ikcpcb *kcp, IUINT32 sn, IUINT32 ts)
{
	IUINT32 newsize = kcp->ackcount + 1;
	IUINT32 *ptr;

	if (newsize > kcp->ackblock) {
		IUINT32 *acklist;
		IUINT32 newblock;

		for (newblock = 8; newblock < newsize; newblock <<= 1);
		acklist = (IUINT32*)ikcp_malloc(newblock * sizeof(IUINT32) * 2);

		if (acklist == NULL) {
			assert(acklist != NULL);
			abort();
		}

		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++;
}

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];
}


//---------------------------------------------------------------------
// parse data
//---------------------------------------------------------------------
void ikcp_parse_data(ikcpcb *kcp, IKCPSEG *newseg)
{
	struct IQUEUEHEAD *p, *prev;
	IUINT32 sn = newseg->sn;
	int repeat = 0;
	
	if (_itimediff(sn, kcp->rcv_nxt + kcp->rcv_wnd) >= 0 ||
		_itimediff(sn, kcp->rcv_nxt) < 0) {
		ikcp_segment_delete(kcp, newseg);
		return;
	}

	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++;
		}	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
}


//---------------------------------------------------------------------
// input data
//---------------------------------------------------------------------
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);
	}

	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);
		if (conv != kcp->conv) return -1;

		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;

		if (cmd != IKCP_CMD_PUSH && cmd != IKCP_CMD_ACK &&
			cmd != IKCP_CMD_WASK && cmd != IKCP_CMD_WINS) 
			return -3;

		kcp->rmt_wnd = wnd;
		ikcp_parse_una(kcp, una);
		ikcp_shrink_buf(kcp);

		if (cmd == IKCP_CMD_ACK) {
			if (_itimediff(kcp->current, ts) >= 0) {
				ikcp_update_ack(kcp, _itimediff(kcp->current, ts));
			}
			ikcp_parse_ack(kcp, sn);
			ikcp_shrink_buf(kcp);
			if (flag == 0) {
				flag = 1;
				maxack = sn;
				latest_ts = ts;
			}	else {
				if (_itimediff(sn, maxack) > 0) {
				#ifndef IKCP_FASTACK_CONSERVE
					maxack = sn;
					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);
				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);
					}

					ikcp_parse_data(kcp, seg);
				}
			}
		}
		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
			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);
	}

	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
//---------------------------------------------------------------------
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;
	seg.frg = 0;
	seg.wnd = ikcp_wnd_unused(kcp);
	seg.una = kcp->rcv_nxt;
	seg.len = 0;
	seg.sn = 0;
	seg.ts = 0;

	// 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);
		ptr = ikcp_encode_seg(ptr, &seg);
	}

	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;
			kcp->ts_probe = kcp->current + kcp->probe_wait;
		}	
		else {
			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 {
		kcp->ts_probe = 0;
		kcp->probe_wait = 0;
	}

	// flush window probing commands
	if (kcp->probe & IKCP_ASK_SEND) {
		seg.cmd = IKCP_CMD_WASK;
		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 = (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);
	if (kcp->nocwnd == 0) cwnd = _imin_(kcp->cwnd, cwnd);

	// move data from snd_queue to snd_buf
	while (_itimediff(kcp->snd_nxt, kcp->snd_una + cwnd) < 0) {
		IKCPSEG *newseg;
		if (iqueue_is_empty(&kcp->snd_queue)) break;

		newseg = iqueue_entry(kcp->snd_queue.next, IKCPSEG, node);

		iqueue_del(&newseg->node);
		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
	resent = (kcp->fastresend > 0)? (IUINT32)kcp->fastresend : 0xffffffff;
	rtomin = (kcp->nodelay == 0)? (kcp->rx_rto >> 3) : 0;

	// flush data segments
	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) {
		    // 新的报文，需要发送
			needsend = 1;
			segment->xmit++;
			segment->rto = kcp->rx_rto;
			// 失败的话下一次重传时间 = 当前时间 + rto + rtomin
			segment->resendts = current + segment->rto + rtomin;
		}
		else if (_itimediff(current, segment->resendts) >= 0) {
			// current - segment->resendts >= 0
			// 当前的segment已经到了重传的时间，需要发送
			needsend = 1;
			// 发送次数 + 1
			segment->xmit++;
			// 总的重传次数 + 1
			kcp->xmit++;
			// 计算该segment的RTO
			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;
			}
			// 设置下一次重传的时间
			segment->resendts = current + segment->rto;
			lost = 1;
		}
		else if (segment->fastack >= resent) {
		    // 这个segment的ack已经失序了,出现了发送1,2,3,4 收到ack 1,3,4情况
		    // 编号2的segment的ack已经失序(4 - 2 = 2)次了
			if ((int)segment->xmit <= kcp->fastlimit || 
				kcp->fastlimit <= 0) {
			    // 该segment的发送次数小于快速重传次数或者需要快速重传
				needsend = 1;
				// 发送次数+1,
				// 如果发现 segment->xmit > kcp->fastlimit了该segment会怎么处理
				segment->xmit++;
				segment->fastack = 0;
				segment->resendts = current + segment->rto;
				change++;
			}
			// 发现 segment->xmit > kcp->fastlimit了该segment会怎么处理，
			// 这个segment不用快速重传了,留到后面发
		}

		if (needsend) {
		    // 该segment需要先发送
			int need;
			segment->ts = current;
			segment->wnd = seg.wnd;
			segment->una = kcp->rcv_nxt;

			size = (int)(ptr - buffer);
			need = IKCP_OVERHEAD + segment->len;

			if (size + need > (int)kcp->mtu) {
				ikcp_output(kcp, buffer, size);
				ptr = buffer;
			}

			ptr = ikcp_encode_seg(ptr, segment);

			if (segment->len > 0) {
				memcpy(ptr, segment->data, segment->len);
				ptr += segment->len;
			}

			// 该segment的发送次超过限制了（默认20次）,
			// 出现这种情况可以认为网络出现了问题，一个包发送20几次都发送不出去是真的有问题
			if (segment->xmit >= kcp->dead_link) {
				// kcp->state = 0xffffffff
				kcp->state = (IUINT32)-1;
			}
		}
	}

	// flash remain segments
	size = (int)(ptr - buffer);
	if (size > 0) {
		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;
		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) {
		kcp->ts_flush = kcp->current;
		slap = 0;
	}

	if (slap >= 0) {
		kcp->ts_flush += kcp->interval;
		if (_itimediff(kcp->current, kcp->ts_flush) >= 0) {
			kcp->ts_flush = kcp->current + kcp->interval;
		}
		ikcp_flush(kcp);
	}
}


//---------------------------------------------------------------------
// 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;
	kcp->mss = kcp->mtu - IKCP_OVERHEAD;
	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;
	}
	if (nc >= 0) {
		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
		    // 接收大小需要大过128
			kcp->rcv_wnd = _imax_(rcvwnd, IKCP_WND_RCV);
		}
	}
	return 0;
}

int ikcp_waitsnd(const ikcpcb *kcp)
{
	return kcp->nsnd_buf + kcp->nsnd_que;
}


// read conv
IUINT32 ikcp_getconv(const void *ptr)
{
	IUINT32 conv;
	ikcp_decode32u((const char*)ptr, &conv);
	return conv;
}

收发demo

使用lsof -i:8080

服务端

#include <sys/types.h>
#include <sys/socket.h>
#include <pthread.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include "ikcp.c"

#include <string.h>

#include <sys/time.h>
#include <sys/wait.h>
#include <arpa/inet.h>
#include <stdbool.h>
#include <errno.h>

typedef struct {
    unsigned char *ipstr;
    int port;
    ikcpcb *pkcpb;
    int sockfd;
    struct sockaddr_in addr;
    struct sockaddr_in clientAddr;
    socklen_t clientAddrLen;
    
    volatile bool is_avliable;
    volatile bool run_sendread;
    volatile bool run_send;
    volatile bool run_read;
    pthread_mutex_t lock;
    pthread_t thread_send;
    pthread_t thread_read;
    pthread_t thread_update;
}my_kcp_t;

/* 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);
}

/* 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 udpOutPut(const char *buf, int len, ikcpcb *kcp, void *user){
    my_kcp_t *send = (my_kcp_t *)user;
    //发送信息
    int n = sendto(send->sockfd, buf, len, MSG_DONTWAIT, (struct sockaddr *)&send->clientAddr, sizeof(struct sockaddr_in));
    if (n >= 0)
    {
        printf("server udp send: ip = %s port = %d 字节 =%d bytes   内容=[%s]\n",inet_ntoa(send->clientAddr.sin_addr), ntohs(send->clientAddr.sin_port), n ,buf+24);//24字节的KCP头部
        return n;
    }else{
        printf("error: %d bytes send, error\n", n);
        return -1;
    }
}
void kcp_write_log(const char *log, struct IKCPCB *kcp, void *user){
    printf("kcp_write_log %s", log);
}

static size_t memsize = 0;
static int needsend = 0;
void* my_ikcp_mem_malloc(size_t size) {
    void *ptr = NULL;
    ptr = malloc(size);
    printf("申请内存：%p 大小:%lu\n",ptr, size);
    memsize ++;
    printf("memsize %ld\n", memsize);
    return ptr;
}

void my_ikcp_mem_free(void *ptr){
    printf("释放内存:%p\n", ptr);
    memsize --;
    printf("memsize %ld\n", memsize);
    free(ptr);
}

void* th_read(void *arg) {
    my_kcp_t *kcp = (my_kcp_t *)arg;
    unsigned int len = sizeof(struct sockaddr_in);
    int n, ret;
    bool read_exit = false;
    while (kcp->is_avliable && kcp->run_sendread) {
        isleep(5);
        char buf[1500] = {0};
//        pthread_mutex_lock(&(kcp->lock));
        int timeout = 300;
        while (timeout --) {
            // 收到UDP包
            struct sockaddr_in callAddr;
            socklen_t callAddrLen;
            n = recvfrom(kcp->sockfd, buf, 567, MSG_CONFIRM, (struct sockaddr *)&callAddr, &callAddrLen);
            if (n < 0) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    continue;
                }else{
                    read_exit = true;
                    break;
                }
            }
//            char* callAddrStr = inet_ntoa(callAddr.sin_addr);
//            int callAddrPort = ntohs(callAddr.sin_port);
//            if (0 != strcmp(callAddrStr, inet_ntoa(kcp->clientAddr.sin_addr)) || callAddrPort != ntohs(kcp->clientAddr.sin_port)) {
//                printf("未知连接: ip =%s port=%d data=[%s]\n", callAddrStr, callAddrPort, buf);
//                continue;
//            }
            break;
        }
        if (timeout <= 0) {
//            pthread_mutex_unlock(&(kcp->lock));
            printf("Recfrom %s %d Timeout\n", inet_ntoa(kcp->clientAddr.sin_addr), ntohs(kcp->clientAddr.sin_port));
            kcp->is_avliable = false;
            kcp->run_sendread = false;
            break;
        }
        if (read_exit) {
            kcp->is_avliable = false;
            kcp->run_sendread = false;
            break;
        }
        
        printf("recvfrom ip %s port %d data %s\n", inet_ntoa(kcp->clientAddr.sin_addr),ntohs(kcp->clientAddr.sin_port), buf);
        
        // 将收到的数据包塞给kcp处理
        ret = ikcp_input(kcp->pkcpb, buf, n);
//        pthread_mutex_unlock(&(kcp->lock));
        if (0 != ret) {
            printf("ikcp_input errcode %d\n", ret);
            continue;
        }
        // 再从kcp中获得数据原始数据
//        pthread_mutex_lock(&(kcp->lock));
        ret = ikcp_recv(kcp->pkcpb, buf, 1024);
//        pthread_mutex_unlock(&(kcp->lock));
        if (ret < 0) {
            if (kcp->pkcpb->state == 0xffffffff) {
                kcp->is_avliable = false;
                kcp->run_sendread = false;
                break;
            }
            continue;
        }else{
            printf("收到: ip = %s port = %d 数据:%s\n", inet_ntoa(kcp->clientAddr.sin_addr), ntohs(kcp->clientAddr.sin_port), buf);
            // 返回响应
//             pthread_mutex_lock(&(kcp->lock));
//             needsend ++;
//             pthread_mutex_lock(&(kcp->lock));
        }
    }
    return NULL;
}

void* th_update(void *arg){
//    my_kcp_t  *kcp = (my_kcp_t *)arg;
//    while (kcp->is_avliable && kcp->run_sendread && kcp->pkcpb->state != 0xffffffff) {
//        usleep(10 * 1000);
//        pthread_mutex_lock(&(kcp->lock));
//        if (NULL != kcp->pkcpb) {
//            ikcp_update(kcp->pkcpb, iclock());
//        }else{
//            kcp->is_avliable = false;
//            kcp->run_sendread = false;
//        }
//        pthread_mutex_unlock(&(kcp->lock));
//    }
    return NULL;
}

static int sendNumber = 0;
void* th_send(void *arg) {
    my_kcp_t *kcp = (my_kcp_t *)arg;
    char sendbuf[100];
    while (kcp->is_avliable && kcp->run_sendread) {
        isleep(5);
//        pthread_mutex_lock(&(kcp->lock));
//        if (needsend > 0) {
//            needsend --;
//        }else{
//            pthread_mutex_unlock(&(kcp->lock));
//            continue;
//        }
//        pthread_mutex_unlock(&(kcp->lock));
        memset(sendbuf, 0, sizeof(sendbuf));
        sprintf(sendbuf, "This is Server Send Data n=[%d]", ++sendNumber);
        if (0 != ikcp_send(kcp->pkcpb, sendbuf, sizeof(sendbuf))){
            kcp->run_sendread = false;
            kcp->is_avliable = false;
            break;
        }else{
            // printf("服务器发送数据成功....");
        }
        ikcp_update(kcp->pkcpb,iclock());
    }
    return NULL;
}

// 服务器开启8080端口监听UDP
// 客户端向服务器发送数据,
// 服务器知道了客户端的IP和PORT了,
// 就可以向这个地址发送响应数据了
int main(int argc,char *argv[])
{
    printf("this is kcpServer\n");
    
    my_kcp_t send;
    send.port = 8080;
    send.pkcpb = NULL;
    
    ikcpcb *kcp = ikcp_create(0x1, (void *)&send);//创建kcp对象把send传给kcp的user变量
    
    // 设置日志回调函数
    kcp->writelog = kcp_write_log;
    
    //设置kcp对象的回调函数
    ikcp_setoutput(kcp, udpOutPut);
    
    //1, 10, 2, 1
    ikcp_nodelay(kcp, 1, 10, 2, 1);
    
    // 设置窗口大小
    ikcp_wndsize(kcp, 1024, 1024);
    
    // 设置MTU
    ikcp_setmtu(kcp, 1400);
    
    // 设置内存分配器
    ikcp_allocator(my_ikcp_mem_malloc, my_ikcp_mem_free);
    
    send.pkcpb = kcp;
    
    int sockfd = socket(AF_INET,SOCK_DGRAM,0);
    if(sockfd < 0)
    {
        perror("socket error！");
        exit(1);
    }
    send.sockfd = sockfd;
    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 failed.\n");
        exit(1);
    }
    
    // 先读到Hello::Server之后再去开启读写线程
    char buff[1024];
    int buffer_size = 1024;
    int rec = 0;
    socklen_t socklen;
    while(1){
        usleep(20* 1000);
        memset(buff, 0, sizeof(buff));
        rec = recvfrom(send.sockfd, buff, 567, MSG_DONTWAIT, (struct sockaddr*)&(send.clientAddr), &socklen);
        if (rec > 0) {
            send.clientAddrLen = socklen;
            if (0 == strcmp(buff, "Hello::Server")) {
                printf("recvfrom ip %s port %d data=[%s] timestamp[%lu]\n", inet_ntoa(send.clientAddr.sin_addr), ntohs(send.clientAddr.sin_port), buff,iclock());
                // 响应Hello::Client
                memset(buff, 0, buffer_size);
                strcpy(buff, "Hello::Client");
                int n = sendto(send.sockfd, buff, strlen("Hello::Client") + 1, MSG_DONTWAIT, (struct sockaddr *)&(send.clientAddr), socklen);
                printf("sendto ip %s port %d data=[%s]\n", inet_ntoa(send.clientAddr.sin_addr), ntohs(send.clientAddr.sin_port), buff);
                if (n == strlen("Hello::Client") + 1){
                    printf("协商完成,开始使用kcp交互数据\n");
                    break;
                }else{
                    printf("协商失败 n = %d failed.\n", n);
                }
            }else{
                printf("UNKNOWN Data.... %s\n", buff);
            }
        }
       
    }

    send.is_avliable = true;
    send.run_sendread = true;
    send.run_send = false;
    if (0 != pthread_create(&(send.thread_read), NULL, th_read, &send)) {
        printf("创建读线程失败....");
        ikcp_free(send.pkcpb);
        return -1;
    }
    pthread_detach(send.thread_read);
    
    if (0 != pthread_create(&(send.thread_send), NULL, th_send, &send)) {
        printf("创建写线程失败.....");
        ikcp_free(send.pkcpb);
        return -1;
    }
    pthread_detach(send.thread_send);
    
    if (0 != pthread_create(&(send.thread_update), NULL, th_update, &send)) {
        printf("创建更新线程失败.....");
        ikcp_free(send.pkcpb);
        return -1;
    }
    pthread_detach(send.thread_update);
    
    // 阻塞主线程
    while (send.is_avliable) {
        sleep(2);
        if (!send.is_avliable){
            // 发生错误,将收发线程关闭
            send.run_sendread = false;
            sleep(2); // 等待收发线程释放资源
        }
    }
    
    ikcp_free(send.pkcpb);
    close(send.sockfd);
    return 0;
}

客户端

#include <sys/types.h>
#include <sys/socket.h>
#include <pthread.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include "ikcp.c"

#include <sys/time.h>
#include <sys/wait.h>
#include <arpa/inet.h>
#include <stdbool.h>
#include <errno.h>


typedef struct {
    unsigned char *ipstr;
    int port;
    ikcpcb *pkcpb;
    int sockfd;
    struct sockaddr_in addr;//存放服务器的结构体
    char buff[488];

    volatile bool is_avliable;
    volatile bool run_sendread;
     pthread_mutex_t lock;
    pthread_t thread_send;
    pthread_t thread_read;
    pthread_t thread_update;
}my_kcp_t;


/* 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);
}

/* 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
}
unsigned long malloc_num = 0;
void* my_ikcp_mem_malloc(size_t size) {
    void *ptr = NULL;
    ptr = malloc(size);
    // printf("申请内存：%p 大小:%lu\n",ptr, size);
    malloc_num ++;
    return ptr;
}

void my_ikcp_mem_free(void *ptr){
    // printf("释放内存:%p\n", ptr);
    malloc_num --;
    free(ptr);
}

int udpOutPut(const char *buf, int len, ikcpcb *kcp, void *user){
    my_kcp_t *send = (my_kcp_t *)user;
    int n = sendto(send->sockfd, buf, len, 0,(struct sockaddr *) &send->addr,sizeof(struct sockaddr_in));
    if (n >= 0)
    {
        printf("client udp send: 字节 =%d bytes   内容=[%s]\n", n ,buf+24);//24字节的KCP头部
        return n;
    }else{
        printf("udpOutPut: %d bytes send, error\n", n);
        return -1;
    }
}
void* th_read(void *arg) {
    my_kcp_t *kcp = (my_kcp_t *)arg;
    unsigned int len = sizeof(struct sockaddr_in);
    int n, ret;
    bool read_exit = false;
    while (kcp->is_avliable && kcp->run_sendread) {
        isleep(10);
        char buf[1500] = {0};
        int timeout = 2000;
        while (timeout --) {
            isleep(5);
            // 收到UDP包
            struct sockaddr_in callAddr;
            socklen_t callAddrLen;
            n = recvfrom(kcp->sockfd, buf, 1500, MSG_DONTWAIT, (struct sockaddr *)&callAddr, &callAddrLen);
            if (n < 0) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    continue;
                }else{
                    read_exit = true;
                    break;
                }
            }
//            char* callAddrStr = inet_ntoa(callAddr.sin_addr);
//            int callAddrPort = ntohs(callAddr.sin_port);
//            if (0 != strcmp(callAddrStr, inet_ntoa(kcp->addr.sin_addr)) || callAddrPort != ntohs(kcp->addr.sin_port)){
//                printf("未知连接: ip =%s port=%d data=[%s]\n", callAddrStr, callAddrPort, buf);
//                continue;
//            }
            break;
        }
        
        if (timeout <= 0) {
//            pthread_mutex_unlock(&(kcp->lock));
            printf("Recfrom %s %d Timeout\n", inet_ntoa(kcp->addr.sin_addr), ntohs(kcp->addr.sin_port));
            kcp->is_avliable = false;
            kcp->run_sendread = false;
            break;
        }
        if (read_exit) {
            kcp->is_avliable = false;
            kcp->run_sendread = false;
        }
        
        printf("recvfrom ip %s port %d data %s\n", inet_ntoa(kcp->addr.sin_addr),ntohs(kcp->addr.sin_port), buf);
        // 将收到的数据包塞给kcp处理
//        pthread_mutex_lock(&(kcp->lock));
        ret = ikcp_input(kcp->pkcpb, buf, n);
        if (kcp->pkcpb->state == 0xffffffff) {
            kcp->is_avliable = false;
            kcp->run_sendread = false;
            break;
        }
//        pthread_mutex_unlock(&(kcp->lock));
        if (-1 == ret) {
            continue;
        }
        // 再从kcp中获得数据原始数据
//        pthread_mutex_lock(&(kcp->lock));
        ret = ikcp_recv(kcp->pkcpb, buf,576);
//        pthread_mutex_unlock(&(kcp->lock));
        if (ret < 0) {
            continue;
        }else{
            printf("收到: ip = %s port = %d 数据:%s\n", inet_ntoa(kcp->addr.sin_addr), ntohs(kcp->addr.sin_port), buf);
            // 返回响应
        }
    }
    return NULL;
}

void* th_update(void *arg){
//    my_kcp_t  *kcp = (my_kcp_t *)arg;
//    while (kcp->is_avliable && kcp->run_sendread) {
//        usleep(10 * 1000);
//        pthread_mutex_lock(&(kcp->lock));
//        if (NULL != kcp->pkcpb) {
//            ikcp_update(kcp->pkcpb, iclock());
//        }
//        pthread_mutex_unlock(&(kcp->lock));
//    }
    return NULL;
}
static int sendNumber = 0;
void* th_send(void *arg) {
    my_kcp_t *kcp = (my_kcp_t *)arg;
    char sendbuf[100];
    while (kcp->is_avliable && kcp->run_sendread) {
        memset(sendbuf, 0, sizeof(sendbuf));
        sprintf(sendbuf, "This is Client Send Data n=[%d]", ++sendNumber);
//        pthread_mutex_lock(&(kcp->lock));
        if (kcp->pkcpb->state == 0xffffffff) {
            kcp->is_avliable = false;
            kcp->run_sendread = false;
            break;
        }
        if (0 != ikcp_send(kcp->pkcpb, sendbuf, sizeof(sendbuf))){
            printf("客户端发送数据失败....");
            kcp->run_sendread = false;
            kcp->is_avliable = false;
        }else{
            // printf("服务器发送数据成功....");
        }
        
        ikcp_update(kcp->pkcpb, iclock());
//        pthread_mutex_unlock(&(kcp->lock));
        isleep(20);
    }
    return NULL;
}
void kcp_write_log(const char *log, struct IKCPCB *kcp, void *user){
    printf("kcp_write_log %s", log);
}

// 客户端向服务端发送数据的时候
// 已经知道服务器的IP和端口号了
int main(int argc,char *argv[])
{
    printf("this is kcpClient\n");
    
    my_kcp_t send;
    send.ipstr = (unsigned char *)"127.0.0.1";
    send.port = 8080;
    int sockfd = socket(AF_INET,SOCK_DGRAM,0);
    printf("sockfd %d\n", sockfd);
    send.sockfd = sockfd;
    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);
    
    
    ikcp_allocator(my_ikcp_mem_malloc, my_ikcp_mem_free);
    
    // 初始化kcp对象
    ikcpcb *kcp = ikcp_create(0x1, (void *)&send);//创建kcp对象把send传给kcp的user变量
    ikcp_setoutput(kcp, udpOutPut);//设置kcp对象的回调函数
    ikcp_nodelay(kcp,1, 10, 2, 1);//(kcp1, 0, 10, 0, 0); 1, 10, 2, 1
    ikcp_wndsize(kcp, 1024, 1024);
    ikcp_setmtu(kcp, 1400);
    kcp->writelog = kcp_write_log;
    
    
    send.pkcpb = kcp;
    
    // 先发送Hello::Client,然后等待接收Hello::Server
    int rec = 0;
    char buff[1024];
    int buffer_size = 1024;
    bool connectOK = false;
    while (!connectOK) {
        isleep(50);
        memset(buff, 0, sizeof(buff));
        strcpy(buff, "Hello::Server");
        printf("尝试发送 ip %s port %d ==> Hello::Server数据....timestamp[%lu]\n", inet_ntoa(send.addr.sin_addr), ntohs(send.addr.sin_port), iclock());
        int n = sendto(send.sockfd, buff, strlen("Hello::Server") + 1,MSG_DONTWAIT,(struct sockaddr*)&(send.addr), sizeof(send.addr));
        printf("已经发送 ip %s port %d ==> Hello::Server数据....Return[%d]\n", inet_ntoa(send.addr.sin_addr), ntohs(send.addr.sin_port), n);
        
        if (n == strlen("Hello::Server") + 1) {
            int timeout = 50;
            while (timeout --) {
                isleep(5);
                // 发送成功,接收响应
                struct sockaddr_in callAddr;
                socklen_t callAddrLen;
                rec = recvfrom(send.sockfd, buff,657 ,MSG_DONTWAIT , (struct sockaddr *)&callAddr, &callAddrLen);
                if (rec < 0) {
                    continue;
                }
                printf("收到ip %s port %d 数据包 %s\n", inet_ntoa(callAddr.sin_addr), ntohs(callAddr.sin_port),buff);
                if (0 == strcmp("Hello::Client", buff)) {
                    printf("收到Hello::Client, 开始KCP\n");
                    connectOK = true;
                    break;
                }else{
                    printf("收到其他数据:%s\n", buff);
                    memset(buff, 0, sizeof(buff));
                }
            }
        }else{
            printf("Client sendto ret = %d\n", n);
        }
    }

    printf("协商完成,使用KCP通讯.....\n");
    send.is_avliable = true;
    send.run_sendread = true;
    if (0 != pthread_create(&(send.thread_send), NULL, th_send, &send)) {
        printf("创建写线程失败.....");
        ikcp_free(send.pkcpb);
        return -1;
    }
    pthread_detach(send.thread_send);

    if (0 != pthread_create(&(send.thread_read), NULL, th_read, &send)) {
        printf("创建读线程失败....");
        ikcp_free(send.pkcpb);
        return -1;
    }
    pthread_detach(send.thread_read);
    
    if (0 != pthread_create(&(send.thread_update), NULL, th_update, &send)) {
        printf("创建更新线程失败.....");
        ikcp_free(send.pkcpb);
        return -1;
    }
    pthread_detach(send.thread_update);

    // 客户端只收发10s
    int i = 0;
    while (send.is_avliable) {
        sleep(4);
        if (!send.is_avliable){
            // 发生错误,将收发线程关闭
            send.run_sendread = false;
            sleep(2); // 等待收发线程释放资源
            ikcp_free(send.pkcpb);
            return -1;
        }
        if (++i == 14) {
            send.is_avliable = false;
            send.run_sendread = false;
            sleep(2);
            ikcp_free(send.pkcpb);
            break;
        }
    }
    return 0;
}

现象和总结

收发速度不一致会导致缓存堆积,大量的内存被申请而不被释放。

使用sendto发送数据之后，可以直接使用recvfrom接收响应的数据了。太久不取数据会取不到。

udp响应包不一定通过目的port返回。有可能通过0端口返回。

只有服务端通过一个共享变量来控制收发会出现死锁问题。

Previous post: 碎片化生存 Next post: C-Protoc

long blogs

进一步有进一步惊喜

简介

KCP 源码注释

头文件ikcp.h

源文件ickp.c

收发demo

服务端

客户端

现象和总结