IOCP模型C++入门级服务端搭建-接收客户端数据并处理

关于客户端连接服务端可参考链接：https://ufgnix0802.blog.csdn.net/article/details/125549167

WSARecv函数

  WSARecv函数是从面向连接socket或面向无连接的socket接收数据。

int WSAAPI WSARecv(
    _In_ SOCKET s, 客户端句柄
    _In_reads_(dwBufferCount) __out_data_source(NETWORK) LPWSABUF lpBuffers,接收缓冲区
    _In_ DWORD dwBufferCount, 接收缓冲区的数量
    _Out_opt_ LPDWORD lpNumberOfBytesRecvd,（如果接收操作立即完成）指向接收字节数
    _Inout_ LPDWORD lpFlags,指向标记位
    _Inout_opt_ LPWSAOVERLAPPED lpOverlapped,重叠I/O结构
    _In_opt_ LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine 指向例程的指针
);

参数：

• s，指向已连接的套接字（即客户端句柄）。
• lpBuffers，一个指向WSABUF结构数组的指针（也可以是WASBUF结构指针）。每一个WSABUF结构包含一个缓冲区的指针和缓冲区的长度。
• dwBufferCount，WSABUF结构数组中WSABUF结构的数量（如果lpBuffers为WSABUF结构指针，则该参数可以填1）。
• lpNumberOfBytesRecvd，如果接受数据的操作立即结束，那么接收的数据字节长度会返回给lpNumberOfBytesRecvd指针所指向的对象。（如果lpOverlapped结构不是NULL，那么应该使用NULL来填充此参数，以避免潜在的错误结果。即只要lpOverlapped不为NULL，那么这个参数必须为NULL或0）。
• lpFlags，用于更改WSARecv函数的行为的标记指针。（即用来控制套接字的行为。例如，指出当前套接字是面向流（连接）的还是面向消息（无连接）的。通常设置为0。）
• lpOverlapped，一个指向重叠I/O结构的指针。
• lpCompletionRoutine，一个指向完成例程的指针。在接收操作完成后，会调用指针所指向的例程函数。在重叠I/O结构中该参数可设置为NULL。

返回值：

  在完成I/O端口下，如果未发生错误并且接收操作已经完成，则WSARecv函数返回0。请注意，在这种情况下，完成指示（即启动指定的完成例程或者设置一个时间对象）将早已发生。否则的话，将返回SOCKET_ERROR，应用程序可以通过调用**WSAGetLastError()**函数来获取具体的错误代码。只有当返回的错误代码为WSA_IO_PENDING时表示重叠操作启动成功，过后依然将会有完成端口指示。而对于任何其它的错误代码都是表示未能成功启动重叠操作，以后也不会再有完成端口指示。

#ifndef UNETCORE_H_
#define UNETCORE_H_

//表示当前IO内核使用的是select模型
//#define U_SELECT_NETCORE

//表示当前IO内核使用的是IOCP模型
#define U_IOCP_NETCORE

#define _WINSOCK_DEPRECATED_NO_WARNINGS
#include 

namespace U
{
	//EventLoop抽象类，事件循环机制
	class IEventLoop
	{
	public:
		virtual bool Init() = 0;

		virtual void LoopOnce() = 0;

		virtual void UnInit() = 0;

		virtual ~IEventLoop() {}
	};
    

	class ITcpSocketCB
	{
	public:
		virtual bool OnRecv(int len) = 0;

		virtual bool OnSend(int len) = 0;

		~ITcpSocketCB() {}
	};


	class ITcpSocket
	{
	public:
		virtual void Init(IEventLoop* loop, ITcpSocketCB* cb) = 0;

		virtual bool DoSend(const char* buf, int len) = 0;

		virtual bool DoRecv(const char* buf, int len) = 0;

		~ITcpSocket() {}
	};


	//响应客户端连接回调函数
	typedef std::function FTcpServerCB;

	//ITcpServer抽象类，服务端
	class ITcpServer
	{
	public:
		/// 

		/// 初始化TCP服务，与循环事件进行关联
		/// 
		/// 事件循环机制
		/// 当用户连接成功后的回调函数
		/// 
		virtual bool Init(IEventLoop* loop, FTcpServerCB cb) = 0;

		/// 

		/// 初始化socket
		/// 
		/// 服务端IP
		/// 监听端口号
		/// 错误返回false
		virtual bool Listen(const char* ip, int port) = 0;

		virtual ~ITcpServer() {}
	};


	//加载Windows网络库
	void InitNetCore();

	//卸载Windows网络库
	void UnNetCore();

	//实例化事件循环机制
	IEventLoop* CreateEventLoop();

	//实例化服务端类
	ITcpServer* CreateTcpServer();
}
#endif // !UNETCORE_H_

#include "UNetCore.h"

#ifdef U_IOCP_NETCORE
#include 
#include 
#include 
#pragma comment(lib,"ws2_32")
#pragma comment(lib, "Mswsock")
#pragma comment(lib, "shlwapi")
#pragma comment(lib, "psapi")

#ifndef EVENTLOOP_H_
#define EVENTLOOP_H_


namespace U
{
	//保存服务端单元的信息。即与服务端句柄关联，保存至事件循环机制中
	class sEvent
	{
	public:
		//该服务端是属于什么类型的（TCP、UDP、PIPE）
		enum class Type
		{
			E_TCPSERVER,
			//UDP，PIPE
			E_TCPCLIENT,
		};
		Type type;
		SOCKET sock;//保存服务端句柄
		//由于sEvent类必须要附带服务端类，那么如果不使用联合体，会造成过度的内存资源浪费，即这里使用联合体保存各种类型服务端类的指针。
		union
		{
			class TcpServer* tcpServer;
			//class UcpServer* ucpServer;
			//...
			class TcpSocket* tcpSocket;
		};
	};

	//用于保存I/O完成端口中的重叠结构以及投递类型和投递对象本身
	struct sReqHandle
	{
		OVERLAPPED	overlapped;
		enum class sReqHandleE
		{
			HANDLE_ACCEPT = 0,
			HANDLE_RECV,
			HANDLE_SEND,
			HANDLE_CONNECT,
			HANDLE_RECVFROM,
			HANDLE_SENDTO,
		};
		sReqHandleE eType;//投递的类型
		void* ptrThis;//记录投递对象本身
	};


	class EventLoop :public IEventLoop
	{
	private:
		HANDLE _iocp;
	public:
		bool Init() override;

		void LoopOnce() override;

		void UnInit() override;

		//添加监听的循环事件，压入sEvent单元
		void AddEvent(sEvent* event);

		//将文件描述符与完成端口进行关联
		bool AssioIOCP(SOCKET sock, void* ptr);
	};
}


#endif // !EVENTLOOP_H_

#endif // U_SELECT_NETCORE

#include "IOCPEventLoop.h"
#include "IOCPTcpServer.h"
#include "IOCPTcpSocket.h";

#ifdef U_IOCP_NETCORE

namespace U
{
	void InitNetCore()
	{
		WSADATA wsa;
		WSAStartup(MAKEWORd(2, 2), &wsa);

		//TODO 可以添加异常导出库，在程序崩溃生成崩溃信息
	}

	void UnNetCore()
	{
		WSACleanup();
	}

	IEventLoop* CreateEventLoop()
	{
		return new EventLoop;
	}
}

bool U::EventLoop::Init()
{
	//CreateIoCompletionPort函数的作用：
	//创建输入/输出（I/O）完成端口并将其与指定的文件句柄（文件描述符）相关联，
	//或创建尚未与文件句柄（文件描述符）关联的I/O完成端口，以便稍后关联。
	//将打开的文件句柄（文件描述符）的实例与I/O完成端口相关联，使进程能够接收涉及该文件句柄（文件描述符）的异步I/O操作完成通知。
	

	//第一步，创建尚未与文件句柄（文件描述符）关联的I/O完成端口，以便稍后关联
	//由于目前没有文件描述符，即创建I/O完成端口，同时开辟一个线程（主线程）
	_iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 1);
	if (INVALID_HANDLE_VALUE == _iocp)
	{
		std::cout << "创建完成端口失败" << std::endl;
		return false;
	}
	std::cout << "创建完成端口成功" << std::endl;
	return true;
}

//采用IOCP IO模型
void U::EventLoop::LoopOnce()
{
	//第五步，获取排列队列的完成状态
	
	DWORD NumberOfBytesTransferred;
	void* lpCompletionKey = NULL;
	OVERLAPPED* lpOverlapped;
	BOOL bRet = GetQueuedCompletionStatus(_iocp, &NumberOfBytesTransferred, (PULONG_PTR)&lpCompletionKey, &lpOverlapped, 0);
	if (!bRet && NULL == lpOverlapped)
	{
		//std::cout << "错误消息：" << GetLastError() << std::endl;
		return;
	}
	sEvent* event = (sEvent*)lpCompletionKey;
	switch (event->type)
	{
	case U::sEvent::Type::E_TCPSERVER:
		event->tcpServer->OnAccept();
		break;
	case U::sEvent::Type::E_TCPCLIENT:
		event->tcpSocket->OnNetMsg((sReqHandle*)lpOverlapped, bRet, NumberOfBytesTransferred);
		break;
	default:
		break;
	}
}

void U::EventLoop::UnInit()
{

}


void U::EventLoop::AddEvent(sEvent* event)
{

}

bool U::EventLoop::AssioIOCP(SOCKET sock, void* ptr)
{
	//第二步，将创建的服务端句柄跟I/O完成端口进行关联，同时绑定I/O完成密钥（使用文件描述符对应的sEvent对象进行绑定），
	//由于I/O完成端口不为NULL，此时并发线程数忽略。
	//文件描述符为有效，且I/O完成端口句柄有效，则返回值为I/O完成端口句柄本身，否则则返回NULL。
	return _iocp == CreateIoCompletionPort((HANDLE)sock, _iocp, (ULONG_PTR)ptr, 0);
}

#endif // U_SELECT_NETCORE

#include "UNetCore.h"

#ifdef U_IOCP_NETCORE
#include "IOCPEventLoop.h"
#ifndef TCPSERVER_H_
#define TCPSERVER_H_

namespace U
{
	class TcpServer :public ITcpServer
	{
	private:
		FTcpServerCB _cb;//客户端连接回调事件
		SOCKET _sock;//服务端socket
		sEvent _event;
		EventLoop* _loop;

		//客户端句柄（文件描述符）及输出缓冲区
		SOCKET _clientSock;
		char _buffer[1024];
		DWORD _recvLen;


		//重叠结构体，供IOCP内部机制使用
		sReqHandle _handleOVERLAPPED;

		std::unordered_map _tcs;
	public:
		TcpServer();

		~TcpServer();

		bool Init(IEventLoop* loop, FTcpServerCB cb) override;

		bool Listen(const char* ip, int port) override;

	public:
		//处理客户端连接
		void OnAccept();

		void DeleteTcpSocket(SOCKET sock);

	private:
		bool PostAccept();
	};
}

#endif // !TCPSERVER_H_
#endif // U_SELECT_NETCORE

#include "IOCPTcpServer.h"
#ifdef U_IOCP_NETCORE
//IOCP框架头文件
#include 
#include 
#include "IOCPTcpSocket.h"

namespace U
{
	ITcpServer* CreateTcpServer()
	{
		return new TcpServer;
	}
}

U::TcpServer::TcpServer()
{
	std::cout << "初始化 TCPServer" << std::endl;
	_cb = nullptr;
	_sock = INVALID_SOCKET;

	_event.tcpServer = this;
	_event.type = sEvent::Type::E_TCPSERVER;
	_event.sock = INVALID_SOCKET;

	_handleOVERLAPPED.eType = sReqHandle::sReqHandleE::HANDLE_ACCEPT;
	_handleOVERLAPPED.ptrThis = this;
	memset(&_handleOVERLAPPED.overlapped, 0, sizeof(_handleOVERLAPPED.overlapped));
}

U::TcpServer::~TcpServer()
{
	_cb = nullptr;
	if (INVALID_SOCKET != _sock)
		closesocket(_sock);
	_sock = INVALID_SOCKET;
}

bool U::TcpServer::Init(IEventLoop* loop, FTcpServerCB cb)
{
	_cb = cb;
	_loop = dynamic_cast(loop);
	return true;
}

bool U::TcpServer::Listen(const char* ip, int port)
{
	std::cout << "Listen IP:" << ip << " port:" << port << std::endl;

	//TODO 待完善
	if (_sock != INVALID_SOCKET)
	{
		closesocket(_sock);
		return false;
	}

	_sock = socket(AF_INET, SOCK_STREAM, 0);
	SOCKADDR_IN addr;
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	addr.sin_addr.s_addr = inet_addr(ip);
	if (SOCKET_ERROR == bind(_sock, (sockaddr*)&addr, sizeof(addr)))
		return false;

	if (SOCKET_ERROR == listen(_sock, 5))
		return false;

	_event.sock = _sock;
	if (_loop->AssioIOCP(_sock, (void*)&_event))
		std::cout << "服务端句柄关联IOCP成功" << __FUNCTION__ << std::endl;
	else
		std::cout << "服务端句柄关联IOCP失败" << __FUNCTION__ << std::endl;

	//投递IOCP accept，预处理
	PostAccept();
	return true;
}


void U::TcpServer::OnAccept()
{
	//处理客户端消息
	//第六步，GetAcceptExSockaddrs解析AcceptEx函数获取的数据，将输出缓冲区和接收字节大小传入至函数，最终将本地和远端地址传递给sockaddr结构中
	
	sockaddr* serverAddr = NULL;
	sockaddr* clientAddr = NULL;
	int serverAddrLen;
	int clientAddrLen;
	GetAcceptExSockaddrs(_buffer, _recvLen, sizeof(SOCKADDR_IN) + 16, sizeof(SOCKADDR_IN) + 16,
		&serverAddr, &serverAddrLen, &clientAddr, &clientAddrLen);
	TcpSocket* client = new TcpSocket;
	client->OnAccept(_clientSock, *clientAddr, clientAddrLen);
	client->_server = this;
	_tcs.insert(std::make_pair(_clientSock, client));

	//连接成功后调用回调函数
	_cb(client);

	//投递IOCP Accept
	PostAccept();
}


void U::TcpServer::DeleteTcpSocket(SOCKET sock)
{
	auto item = _tcs.find(sock);
	if (item != _tcs.end())
	{
		delete item->second;
		_tcs.erase(sock);
	}
}

bool U::TcpServer::PostAccept()
{
	//第三步，创建支持重叠I/O的客户端socket
	//WSASocket函数：创建一个套接字（文件描述符，文件句柄）
	//第一个参数表示采用ipv4族，第二个参数表示采用TCP协议，第三个参数表示可能的协议类型为TCP协议
	//第四个参数如果不为空，则会让创建的套接字与LPWSAPROTOCOL_INFOW指针指向的结构体绑定
	//第五个参数为0表示没有执行组相关的操作
	//第六个参数表示WSA_FLAG_OVERLAPPED表示创建一个支持重叠I/O(I/O完成端口句柄)的socket
	_clientSock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);//等价于调用socket函数，内置默认支持重叠I/O（WSA_FLAG_OVERLAPPED）
	if (_clientSock == INVALID_SOCKET)
	{
		std::cout << "创建客户端socket失败" << std::endl;
		return false;
	}

	//第四步，处理第一次客户端连接的数据块
	_recvLen = 0;//每次都需要重置一下当前用于接收客户端数据大小的长度
	

	//在处理客户端连接的过程中虽然我们不处理第一次连接过程中额外的数据，第四个参数为0，但是客户端第一次连接服务端的时候
	//还是可能会携带一些数据发送至服务端，那么这时我们使用_recvBuf进行存储的数据不仅仅是服务端和客户端的地址信息，在其头部
	//可能存放一些额外的数据，这时我们使用_recvLen进行接收。_recvBuf存储数据的顺序是先存放额外的数据，然后在存放客户端和
	//服务端地址信息，在使用GetAcceptExSockaddrs进行解析客户端和服务端地址信息时，需要将头部额外的数据信息进行偏移掉，
	//从而正确获取客户端和服务端的地址偏移信息。
	if (FALSE == AcceptEx(_sock, _clientSock, _buffer, 0, sizeof(SOCKADDR_IN) + 16, sizeof(SOCKADDR_IN) + 16, &_recvLen, &_handleOVERLAPPED.overlapped))
	{
		if (WSAGetLastError() != ERROR_IO_PENDING)
		{
			std::cout << "AcceptEx ERROR" << std::endl;
			return false;
		}
	}
	std::cout << "AcceptEx OK" << std::endl;
	return true;

}
#endif // U_SELECT_NETCORE

#include "UNetCore.h"

#ifdef U_IOCP_NETCORE

#include "IOCPEventLoop.h"

#ifndef IOCPTCPSOCKET_H_
#define IOCPTCPSOCKET_H_


namespace U
{
	class TcpSocket :public ITcpSocket
	{
	private:
		EventLoop* _loop;
		SOCKET _sock;
		ITcpSocketCB* _cb;

		sEvent _event;

		//接收
		sReqHandle _recvHandle;
		bool _bRecv;
		WSABUF _recvBuf;

		//发送
		sReqHandle _sendHandle;
		bool _bSend;
		WSABUF _sendBuf;

	public:
		class TcpServer* _server;

		TcpSocket();

		~TcpSocket();

		bool OnAccept(SOCKET sock, SOCKADDR addr, int sockaddrLen);

		bool OnNetMsg(sReqHandle* handle, bool nRet, int len);

	public:
		void Init(IEventLoop* loop, ITcpSocketCB* cb) override;


		bool DoSend(const char* buf, int len) override;


		bool DoRecv(const char* buf, int len) override;

	};
}
#endif // !IOCPTCPSOCKET_H_

#endif // U_IOCP_NETCORE

#include "IOCPTcpSocket.h"
#include "IOCPTcpServer.h"

#ifdef IOCPTCPSOCKET_H_


U::TcpSocket::TcpSocket()
{
	_loop = nullptr;
	_sock = INVALID_SOCKET;
	_cb = nullptr;

	_event.sock = INVALID_SOCKET;
	_event.type = sEvent::Type::E_TCPCLIENT;
	_event.tcpSocket = this;

	_recvHandle.eType = sReqHandle::sReqHandleE::HANDLE_RECV;
	_recvHandle.ptrThis = this;
	memset(&_recvHandle.overlapped, 0, sizeof(_recvHandle.overlapped));
	_bRecv = false;

	_recvBuf.buf = nullptr;
	_recvBuf.len = 0;

	_sendBuf.buf = nullptr;
	_sendBuf.len = 0;
	_bSend = false;
	_sendHandle.eType = sReqHandle::sReqHandleE::HANDLE_SEND;
	_sendHandle.ptrThis = this;
	memset(&_sendHandle.overlapped, 0, sizeof(_sendHandle.overlapped));
}

U::TcpSocket::~TcpSocket()
{
	if (_sock != INVALID_SOCKET)
		closesocket(_sock);
	_sock = INVALID_SOCKET;

	if (_cb != nullptr)
		delete _cb;
	_cb = nullptr;
}

void U::TcpSocket::Init(IEventLoop* loop, ITcpSocketCB* cb)
{
	_loop = (EventLoop*)loop;
	_cb = cb;

	_event.sock = _sock;
	_loop->AssioIOCP(_sock, &_event);
}


bool U::TcpSocket::OnAccept(SOCKET sock, SOCKADDR addr, int sockaddrLen)
{
	if (_sock != INVALID_SOCKET)
		return false;
	_sock = sock;
	return true;
}


bool U::TcpSocket::OnNetMsg(sReqHandle* handle, bool nRet, int len)
{
	if (!nRet)
	{
		//TODO 关闭
		//此时完成端口读取一个失败的I/O完成包或者完成端口中该socket已经被关闭
		return false;
	}

	if (handle->eType == sReqHandle::sReqHandleE::HANDLE_RECV)
	{
		_bRecv = false;
		if (len <= 0)
		{
			//断开连接
			_server->DeleteTcpSocket(_sock);
			std::cout << "客户端断开连接" << std::endl;
			return false;
		}
		//接收到客户端数据并进行处理
		_cb->OnRecv(len);
	}
	else if (handle->eType == sReqHandle::sReqHandleE::HANDLE_SEND)
	{
		_bSend = false;
		
	}

	return true;
}


bool U::TcpSocket::DoSend(const char* buf, int len)
{
	if (_bSend)return false;

	// TODO判断一些失败条件

	std::cout << "WSASend OK" << std::endl;
	_bSend = true;


	_sendBuf.buf = (CHAR*)buf;
	_sendBuf.len = len;

	DWORD dwSend = 0;
	DWORD dwFlag = 0;
	if (WSASend(_sock, &_sendBuf, 1, &dwSend, dwFlag, &_sendHandle.overlapped, NULL) != 0)
	{
		if (WSAGetLastError() != WSA_IO_PENDING)
		{
			return false;
		}
	}

	return true;
}

bool U::TcpSocket::DoRecv(const char* buf, int len)
{
	if (_bRecv)return false;

	// TODO判断一些失败条件
	if (len <= 0)
	{
		std::cout << "接收缓冲区已满" << std::endl;
		return false;
	}

	std::cout << "WSARecv OK" << std::endl;
	_bRecv = true;
	_recvBuf.buf = (CHAR*)buf;
	_recvBuf.len = len;

	DWORD dwRecv = 0;
	DWORD dwFlag = 0;
	if (WSARecv(_sock, &_recvBuf, 1, &dwRecv, &dwFlag, &_recvHandle.overlapped, NULL) != 0)
	{
		if (WSAGetLastError() != WSA_IO_PENDING)
		{
			return false;
		}
	}

	return true;
}


#endif // IOCPTCPSOCKET_H_

#include "UNetCore.h"
#include "HttpServer.h" 
#include 

namespace U
{
	class TcpSocketCB :public ITcpSocketCB
	{
	private:
		//接收缓冲区
		char _recvBuf[1024];
		int _recvLen;
		int _recvMaxLen;

		//发送缓冲区
		char _sendBuf[1024];
		int _sendLen;
		int _sendMaxLen;
	public:
		ITcpSocket* iTcpSocket;

		TcpSocketCB()
		{
			_recvMaxLen = 1024;
			_recvLen = 0;
			iTcpSocket = nullptr;

			_sendLen = 0;
			_sendMaxLen = 1024;
		}


	public:
		virtual bool OnRecv(int len) override
		{
			//更新现有缓冲区长度
			_recvLen += len;

			std::cout << "接收数据内容为：" << std::endl;
			for (int i = 0; i < _recvLen; i++)
				printf("%c", _recvBuf[i]);
			std::cout << std::endl;

			//处理从客户端接收的数据
			//将接收到的所有数据转发给原来的客户端
			

			//处理数据之后更新缓冲区长度
			//_recvLen = 0;
			_recvLen -= len;
			//投递新的缓冲区用于接收数据
			OnRecv();
			return true;
		}

		void OnRecv()
		{
			iTcpSocket->DoRecv(_recvBuf + _recvLen, _recvMaxLen - _recvLen);
		}

		virtual bool OnSend(int len) override
		{
			//将已经发送的数据进行更新
			_sendLen -= len;
			return true;
		}


		void OnSend()
		{
			if (iTcpSocket->DoSend(_sendBuf, _sendLen))
				_sendLen = 0;
		}
	};
}

int main()
{
	//加载Windows网络库
	U::InitNetCore();

	//调用EventLoop循环机制
	U::IEventLoop* loop = U::CreateEventLoop();

	loop->Init();//初始化I/O完成端口

	//初始化服务端
	U::ITcpServer* server = U::CreateTcpServer();

	//初始化HTTP服务端
	//U::ITcpServer* server = U::CreateTcpServer();

	//将服务端与事件循环机制loop关联
	

	server->Init(loop, [loop](U::ITcpSocket* sock) {
		std::cout << "客户端连接" << std::endl;
		U::TcpSocketCB* p = new U::TcpSocketCB;
		p->iTcpSocket = sock;
		sock->Init(loop, p);
		//第七步，投递缓冲区用于接收数据
		p->OnRecv();
		});

	server->Listen("0.0.0.0", 7890);

	//启动循环机制监听消息
	while (true)
		loop->LoopOnce();

	//卸载Windows网络库
	U::UnNetCore();
	return 0;
}