HappyTanuki/Np_Term
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

iocp 구현하기

Browse Source
This commit is contained in:
HappyTanuki
2025-05-31 03:19:58 +09:00
parent 466a80f02b
commit a65483a9c3
55 changed files with 608 additions and 683 deletions
Download Patch File Download Diff File Expand all files Collapse all files

360
include/socket/iocp.h
View File

@@ -1,11 +1,15 @@
#pragma once
#include <openssl/bio.h>
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <functional>
#include <queue>
#include <vector>
#include "socket/tcp_socket.h"
#include "socket/wsa_manager.h"
#include "socket.h"
#include "utils/thread_pool.h"
#include "wsa_manager.h"
#ifdef __linux__
@@ -20,7 +24,7 @@ typedef struct __WSABUF {
#endif
namespace Socket {
namespace Network {
class IOCP;
@@ -29,9 +33,8 @@ enum class IOCPEVENT { QUIT, READ, WRITE };
struct IOCPPASSINDATA {
OVERLAPPED overlapped;
IOCPEVENT event;
std::shared_ptr<UDPSocket> socket;
std::uint32_t recvbytes;
std::uint32_t sendbytes;
std::shared_ptr<Socket> socket;
SSL* ssl;
std::uint32_t transferredbytes;
WSABUF wsabuf;
std::uint32_t bufsize;
@@ -44,8 +47,6 @@ struct IOCPPASSINDATA {
std::memset(&overlapped, 0, sizeof(overlapped));
event = IOCPEVENT::QUIT;
socket = nullptr;
recvbytes = 0;
sendbytes = 0;
transferredbytes = 0;
this->bufsize = bufsize;
IOCPInstance = nullptr;
@@ -64,8 +65,6 @@ struct IOCPPASSINDATA {
#endif
{
std::memset(&overlapped, 0, sizeof(overlapped));
recvbytes = 0;
sendbytes = 0;
wsabuf.buf = new char[bufsize];
std::memcpy(wsabuf.buf, other.wsabuf.buf, other.wsabuf.len);
}
@@ -79,8 +78,6 @@ struct IOCPPASSINDATA {
std::memset(&overlapped, 0, sizeof(overlapped));
event = other.event;
socket = other.socket;
recvbytes = 0;
sendbytes = 0;
transferredbytes = other.transferredbytes;
bufsize = other.bufsize;
IOCPInstance = other.IOCPInstance;
@@ -96,10 +93,53 @@ struct IOCPPASSINDATA {
class IOCP {
public:
~IOCP();
template <typename... _args>
void init(utils::ThreadPool* __IOCPThread, SessionProtocol proto,
std::function<void(utils::ThreadPool*, IOCPPASSINDATA*, _args&&...)>
callback,
_args&&... args) {
IOCPThread_ = __IOCPThread;
proto_ = proto;
if (proto == SessionProtocol::TLS || proto == SessionProtocol::QUIC) {
rbio_ = ::BIO_new(::BIO_s_mem());
wbio_ = ::BIO_new(::BIO_s_mem());
}
#ifdef _WIN32
static void iocpWather(
utils::ThreadPool* threadPool, HANDLE completionPort_,
std::function<void(utils::ThreadPool*, IOCPPASSINDATA*)> callback) {
completionPort_ =
::CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
if (completionPort_ == NULL) {
spdlog::critical("CreateIoCompletionPort()");
std::exit(EXIT_FAILURE);
}
int tCount = __IOCPThread->threadCount;
spdlog::info("Resizing threadpool size to: {}", tCount * 2);
__IOCPThread->respawnWorker(tCount * 2);
for (int i = 0; i < tCount; i++)
__IOCPThread->enqueueJob(iocpWatcher_, IOCPThread_, callback, args...);
#endif
}
void destruct();
void registerSocket(IOCPPASSINDATA* data);
int send(std::vector<IOCPPASSINDATA> data);
int recv(IOCPPASSINDATA& data);
private:
#ifdef _WIN32
template <typename... _args>
void iocpWatcher_(
utils::ThreadPool* IOCPThread,
std::function<void(utils::ThreadPool*, IOCPPASSINDATA*, _args&&...)>
callback,
_args&&... args) {
IOCPPASSINDATA* data;
SOCKET sock;
DWORD cbTransfrred;
@@ -111,266 +151,74 @@ class IOCP {
spdlog::debug("Disconnected. [{}]",
(std::string)(data->socket->remoteAddr));
} else {
if (data->event == IOCPEVENT::READ &&
(proto_ == SessionProtocol::TLS || proto_ == SessionProtocol::QUIC)) {
::BIO_write(rbio_, data->wsabuf.buf, cbTransfrred);
::SSL_read_ex(data->ssl, data->wsabuf.buf, data->bufsize);
}
data->transferredbytes = cbTransfrred;
}
threadPool->enqueueJob(callback, data);
threadPool->enqueueJob(iocpWather, completionPort_, callback);
IOCPThread->enqueueJob(callback, IOCPThread, data, args...);
IOCPThread->enqueueJob(iocpWatcher_, callback, args...);
};
#elif __linux__
static void socketReader(
ThreadPool* threadPool, epoll_event event, int epollfd,
std::function<void(ThreadPool*, IOCPPASSINDATA*)> callback) {
pthread_t tid = pthread_self();
if (event.data.ptr == nullptr) {
spdlog::error("invalid call on {}", tid);
return;
}
IOCPPASSINDATA* rootIocpData = (IOCPPASSINDATA*)event.data.ptr;
std::lock_guard<std::mutex> lock(rootIocpData->socket->readMutex);
while (true) {
char peekBuffer[1];
int rc = rootIocpData->socket->recv(peekBuffer, 1, MSG_PEEK);
if (rc > 0)
;
else if (rc == 0) {
rootIocpData->event = IOCPEVENT::QUIT;
spdlog::debug("Disconnected. [{}]",
(std::string)(rootIocpData->socket->remoteAddr));
::epoll_ctl(epollfd, EPOLL_CTL_DEL, rootIocpData->socket->sock, NULL);
threadPool->enqueueJob(callback, rootIocpData);
// delete rootIocpData;
return;
} else {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
spdlog::trace("No data to read on {}", tid);
return;
} else {
rootIocpData->event = IOCPEVENT::QUIT;
spdlog::debug("Disconnected. [{}]",
(std::string)(rootIocpData->socket->remoteAddr));
::epoll_ctl(epollfd, EPOLL_CTL_DEL, rootIocpData->socket->sock, NULL);
threadPool->enqueueJob(callback, rootIocpData);
// delete rootIocpData;
return;
}
}
Chattr::IOCPPASSINDATA* ptr = new Chattr::IOCPPASSINDATA(*rootIocpData);
ptr->wsabuf.len = 1500;
int redSize = 0;
int headerSize = 8;
int totalRedSize = 0;
while (totalRedSize < headerSize) {
redSize = ptr->socket->recv(ptr->buf + totalRedSize,
headerSize - totalRedSize, 0);
if (redSize == SOCKET_ERROR) {
ptr->event = IOCPEVENT::QUIT;
spdlog::debug("Disconnected. [{}]",
(std::string)(ptr->socket->remoteAddr));
::epoll_ctl(epollfd, EPOLL_CTL_DEL, ptr->socket->sock, NULL);
threadPool->enqueueJob(callback, ptr);
// delete ptr;
return;
} else if (redSize == 0) {
ptr->event = IOCPEVENT::QUIT;
spdlog::debug("Disconnected. [{}]",
(std::string)(ptr->socket->remoteAddr));
::epoll_ctl(epollfd, EPOLL_CTL_DEL, ptr->socket->sock, NULL);
threadPool->enqueueJob(callback, ptr);
// delete ptr;
return;
}
totalRedSize += redSize;
}
Packet packet;
::memcpy(packet.serialized, ptr->buf, headerSize);
redSize = 0;
int dataLength = ntohs(packet.__data.packetLength);
while (totalRedSize < dataLength + headerSize) {
redSize = ptr->socket->recv(ptr->buf + totalRedSize,
dataLength + headerSize - totalRedSize, 0);
if (redSize == SOCKET_ERROR) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
spdlog::trace("No data to read on {}", tid);
return;
}
ptr->event = IOCPEVENT::QUIT;
spdlog::debug("Disconnected. [{}]",
(std::string)(ptr->socket->remoteAddr));
::epoll_ctl(epollfd, EPOLL_CTL_DEL, ptr->socket->sock, NULL);
threadPool->enqueueJob(callback, ptr);
// delete ptr;
return;
} else if (redSize == 0) {
ptr->event = IOCPEVENT::QUIT;
spdlog::debug("Disconnected. [{}]",
(std::string)(ptr->socket->remoteAddr));
::epoll_ctl(epollfd, EPOLL_CTL_DEL, ptr->socket->sock, NULL);
threadPool->enqueueJob(callback, ptr);
// delete ptr;
return;
}
totalRedSize += redSize;
}
ptr->transferredbytes = totalRedSize;
threadPool->enqueueJob(callback, ptr);
}
};
static void socketWriter(
ThreadPool* threadPool, epoll_event event, int epollfd,
std::function<void(ThreadPool*, IOCPPASSINDATA*)> callback) {
pthread_t tid = pthread_self();
if (event.data.ptr == nullptr) {
spdlog::error("invalid call on {}", tid);
return;
}
IOCPPASSINDATA* rootIocpData = (IOCPPASSINDATA*)event.data.ptr;
std::lock_guard<std::mutex> lock(rootIocpData->socket->writeMutex);
while (!rootIocpData->sendQueue->empty()) {
IOCPPASSINDATA* data = rootIocpData->sendQueue->front();
rootIocpData->sendQueue->pop();
if (data == nullptr) {
spdlog::error("invalid call on {}", tid);
break;
}
int packetSize = data->wsabuf.len;
int totalSentSize = 0;
int sentSize = 0;
spdlog::trace("Sending to: [{}]", (std::string)data->socket->remoteAddr);
while (totalSentSize < packetSize) {
sentSize = data->socket->send(data->buf + totalSentSize,
packetSize - totalSentSize, 0);
if (sentSize == SOCKET_ERROR) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
spdlog::warn("buffer full");
continue;
}
data->event = IOCPEVENT::QUIT;
spdlog::debug("Disconnected. [{}]",
(std::string)(data->socket->remoteAddr));
::epoll_ctl(epollfd, EPOLL_CTL_DEL, data->socket->sock, NULL);
threadPool->enqueueJob(callback, data);
// delete data;
return;
}
totalSentSize += sentSize;
}
data->transferredbytes = totalSentSize;
threadPool->enqueueJob(callback, data);
}
};
static void iocpWatcher(
ThreadPool* threadPool, int epollfd, int epollDetroyerFd,
std::function<void(ThreadPool*, IOCPPASSINDATA*)> callback) {
struct epoll_event events[FD_SETSIZE];
pthread_t tid = pthread_self();
int nready = ::epoll_wait(epollfd, events, FD_SETSIZE, -1);
for (int i = 0; i < nready; i++) {
struct epoll_event current_event = events[i];
if (current_event.events & EPOLLIN) {
if (current_event.data.fd == epollDetroyerFd) {
return;
}
std::function<void(ThreadPool*, IOCPPASSINDATA*)> task(callback);
threadPool->enqueueJob(socketReader, current_event, epollfd, task);
} else if (current_event.events & EPOLLOUT) {
std::function<void(ThreadPool*, IOCPPASSINDATA*)> task(callback);
threadPool->enqueueJob(socketWriter, current_event, epollfd, task);
}
if (--nready <= 0) break;
}
threadPool->enqueueJob(iocpWatcher, epollfd, epollDetroyerFd, callback);
};
#endif
IOCP();
~IOCP();
template <typename _Callable>
void init(utils::ThreadPool* __IOCPThread, _Callable&& callback) {
IOCPThread_ = __IOCPThread;
void packet_sender_(utils::ThreadPool* IOCPThread) {
#ifdef _WIN32
completionPort_ =
::CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
if (completionPort_ == NULL) {
spdlog::critical("CreateIoCompletionPort()");
std::exit(EXIT_FAILURE);
}
std::lock_guard<std::mutex> lock(send_queue_mutex_);
#elif __linux__
epollfd_ = ::epoll_create(1);
epollDetroyerFd = ::eventfd(0, EFD_NONBLOCK);
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = epollDetroyerFd;
::epoll_ctl(epollfd_, EPOLL_CTL_ADD, epollDetroyerFd, &ev);
#endif
auto boundFunc = [callback = std::move(callback)](
utils::ThreadPool* __IOCPThread,
IOCPPASSINDATA* data) mutable {
callback(__IOCPThread, data);
};
#ifdef _WIN32
int tCount = __IOCPThread->threadCount;
spdlog::info("Resizing threadpool size to: {}", tCount * 2);
__IOCPThread->respawnWorker(tCount * 2);
for (int i = 0; i < tCount; i++) {
std::function<void(utils::ThreadPool*, IOCPPASSINDATA*)> task(boundFunc);
__IOCPThread->enqueueJob(iocpWather, completionPort_, task);
}
#elif __linux__
__IOCPThread->respawnWorker(__IOCPThread->threadCount + 1);
spdlog::info("Spawning 1 Epoll Waiter...");
__IOCPThread->enqueueJob(iocpWatcher, epollfd_, epollDetroyerFd, boundFunc);
#endif
}
void destruct();
void registerSocket(IOCPPASSINDATA* data);
int recv(IOCPPASSINDATA* data, int bufferCount);
int send(IOCPPASSINDATA* data, int bufferCount, int __flags);
#ifdef __linux__
int epollDetroyerFd = -1;
#endif
private:
struct WSAManager* wsaManager = WSAManager::GetInstance();
utils::ThreadPool* IOCPThread_;
BIO* rbio_ = nullptr;
BIO* wbio_ = nullptr;
SessionProtocol proto_;
// 밑의 unordered_map들에 키를 추가/제거 하려는 스레드는 이 뮤텍스를 잡아야 함.
std::mutex socket_mod_mutex_;
// 각 소켓별 뮤텍스. 다른 스레드가 읽는 중이라면 수신 순서 보장을 위해 다른
// 스레드는 수신을 포기하고 이전 스레드에 전송을 위임해야 함. (ONESHOT으로
// 인해 발생하지 않을 것으로 기대되는 행동이기는 하나 보장되지 않을 수
// 있으므로) 수신 스레드는 epoll이나 iocp를 통해 적당히 수신받아 이 큐를
// 채워야 함. (항시 socket에 대한 큐에 대해 읽기 시도가 행해질 수 있어야
// 한다는 뜻임) EPOLLIN에 대해서는 ONESHOT으로 등록해 놓고 읽는 도중에 버퍼에
// 새 값이 채워질 수 있으므로 읽기가 끝나고 나서 재등록한다
std::unordered_map<SOCKET, std::mutex> recv_queue_mutex_;
// 각 소켓별 패킷, int는 그 vector의 시작 인덱스(vector의 끝까지 다 읽었으면
// 그 vector는 list에서 삭제되어야 하며, 데이터는 평문으로 변환하여 저장한다)
std::unordered_map<SOCKET,
std::list<std::pair<std::vector<char>, std::uint32_t>>>
recv_queue_;
// 각 소켓별 뮤텍스. 다른 스레드가 쓰는 중이라면 송신 순서 보장을 위해 다른
// 스레드는 대기한다. condition variable을 쓰지 않는 이유는 소켓 갯수만큼의
// 스레드가 대기할 수 없기 때문이며, 송신 등록 시에 적당히 송신 스레드를
// 스폰해야 함. 이 변수는 항상 송신 스레드에 의해 관리되어야만 하며,
// 윈도우에서는 송신을 iocp가 대행하기 때문에 이 큐가 필요 없는 것처럼 느껴질
// 수 있으나, 송신 순서를 보장하기 위해 WSASend를 한 스레드가 연속해서
// 호출해야만 하는데 이는 한번 쓰기 호출 시에 송신 중 데이터를 추가하려는
// 스레드가 데이터를 추가하고 미전송된 경우를 대비하여 최대 1개까지의 스레드가
// 대기하도록 한다. 리눅스에서도 send_queue에 데이터를 쌓고 최대 1개까지의
// 스레드가 대기하도록 한다.
std::unordered_map<SOCKET, std::atomic_flag> pending_try_empty_;
std::unordered_map<SOCKET, std::atomic_flag> sending_;
std::unordered_map<SOCKET, std::mutex> send_queue_mutex_;
std::unordered_map<SOCKET, std::list<IOCPPASSINDATA>> send_queue_;
#ifdef _WIN32
HANDLE completionPort_ = INVALID_HANDLE_VALUE;
#elif __linux__
int epollfd_ = -1;
std::unordered_map<std::shared_ptr<TCPSocket>, std::mutex> writeMutex;
std::unordered_map<std::shared_ptr<TCPSocket>, std::queue<IOCPPASSINDATA*>>
writeBuffer;
#endif
};
} // namespace Socket
} // namespace Network