읽기/쓰기 일단 완료

Client/src/asteroid/main.cpp

@@ -4,6 +4,7 @@
 #include "glfw/glfw_window.h"
 #include "socket/iocp.h"
 #include "socket/tcp_socket.h"
+#include "socket/udp_socket.h"
 #include "socket/wsa_manager.h"
 #include "utils/log.h"
 #include "vulkan_engine/vulkan/engine.h"
@@ -40,17 +41,46 @@
    data->event = Network::IOCPEVENT::WRITE;
    data->wsabuf.buf[0] = 'a';
    data->wsabuf.buf[1] = 'b';
-   data->wsabuf.buf[2] = '\0';
+   data->wsabuf.buf[2] = ' ';
    data->wsabuf.len = 3;
    send_data.push_back(data);
+   Network::IOCPPASSINDATA* data2 = new Network::IOCPPASSINDATA(16 * 1024);
+   data2->wsabuf.buf[0] = 'b';
+   data2->wsabuf.buf[1] = 'a';
+   data2->wsabuf.buf[2] = '\0';
+   data2->wsabuf.len = 3;
+   send_data.push_back(data2);
    iocp.send(sock.sock, &send_data);
 
-   Sleep(10000);
-
    Network::IOCPPASSINDATA* recv_data = new Network::IOCPPASSINDATA(16 * 1024);
    recv_data->socket = std::make_shared<Network::TCPSocket>(sock);
    recv_data->IOCPInstance = &iocp;
-   iocp.recv(recv_data);
+   while (!iocp.recv(recv_data)); // 어떤 데이터를 읽는걸 보장받고 싶다면 그냥 스린락 걸어버리기.
+
+   spdlog::info("recv_data: {}", recv_data->wsabuf.buf);
+
+   std::vector<Network::IOCPPASSINDATA*> send_data2;
+   data->event = Network::IOCPEVENT::WRITE;
+   data->wsabuf.buf[0] = 'c';
+   data->wsabuf.buf[1] = 'd';
+   data->wsabuf.buf[2] = ' ';
+   data->wsabuf.len = 3;
+   send_data2.push_back(data);
+   data2 = new Network::IOCPPASSINDATA(16 * 1024);
+   data2->wsabuf.buf[0] = 'd';
+   data2->wsabuf.buf[1] = 'c';
+   data2->wsabuf.buf[2] = '\0';
+   data2->wsabuf.len = 3;
+   send_data2.push_back(data2);
+   iocp.send(sock.sock, &send_data2);
+
+   recv_data = new Network::IOCPPASSINDATA(16 * 1024);
+   recv_data->socket = std::make_shared<Network::TCPSocket>(sock);
+   recv_data->IOCPInstance = &iocp;
+   while (!iocp.recv(recv_data));  // 어떤 데이터를 읽는걸 보장받고 싶다면 그냥
+                                   // 스린락 걸어버리기.
+
+   spdlog::info("recv_data: {}", recv_data->wsabuf.buf);
 
    const veng::GlfwInitialization _glfw;
 

impl/socket/iocp.cpp

@@ -11,6 +11,29 @@ IOCP::IOCP() {
 
 IOCP::~IOCP() { destruct(); }
 
+void IOCP::init(utils::ThreadPool* __IOCPThread, SessionProtocol proto) {
+  IOCPThread_ = __IOCPThread;
+  proto_ = proto;
+
+#ifdef _WIN32
+  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(
+        [this](utils::ThreadPool* th, std::uint8_t __) { iocpWatcher_(th); },
+        0);
+#endif
+}
+
 void IOCP::destruct() {
 #ifdef __linux__
 
@@ -23,15 +46,24 @@ void IOCP::registerSocket(IOCPPASSINDATA* data) {
       (HANDLE)data->socket->sock, completionPort_, data->socket->sock, 0);
   if (returnData == 0) completionPort_ = returnData;
 
-  data->event = IOCPEVENT::READ;
+  IOCPPASSINDATA* recv_data = new IOCPPASSINDATA(data->bufsize);
+  recv_data->event = IOCPEVENT::READ;
+  recv_data->socket = data->socket;
   DWORD recvbytes = 0, flags = 0;
-  ::WSARecv(data->socket->sock, &data->wsabuf, 1, &recvbytes, &flags,
-            &data->overlapped, NULL);
+  int result = ::WSARecv(recv_data->socket->sock, &recv_data->wsabuf, 1,
+                         &recvbytes, &flags, &recv_data->overlapped, NULL);
+  if (result == SOCKET_ERROR) {
+    int err = ::WSAGetLastError();
+    if (err != WSA_IO_PENDING) {
+      spdlog::error("WSARecv failed: {}", err);
+      // 반드시 여기서 리턴하거나 처리해야 합니다.
+    }
+  }
 
 #endif
 }
 
-int IOCP::recv(IOCPPASSINDATA* data) {
+int IOCP::recv(IOCPPASSINDATA* data) { //읽은 바이트수가 무조건 100임? 왜..?
   SOCKET sock = data->socket->sock;
   std::lock_guard lock(*GetRecvQueueMutex_(sock));
   auto queue = GetRecvQueue_(sock);
@@ -79,6 +111,73 @@ int IOCP::send(SOCKET sock, std::vector<IOCPPASSINDATA*>* data) {
   return 0;
 }
 
+int IOCP::GetRecvedPacketCount(SOCKET sock) {
+  std::lock_guard lock(socket_mod_mutex_);
+  auto queue = GetRecvQueue_(sock);
+  return queue->size();
+}
+
+void IOCP::iocpWatcher_(utils::ThreadPool* IOCPThread) {
+  IOCPPASSINDATA* data;
+  SOCKET sock;
+  DWORD cbTransfrred;
+  int jitter = jitterDist_(gen_);
+  int retVal = GetQueuedCompletionStatus(completionPort_, &cbTransfrred,
+                                         (PULONG_PTR)&sock,
+                                         (LPOVERLAPPED*)&data, 1000 + jitter);
+
+  if (retVal == 0 || cbTransfrred == 0) {
+    DWORD lasterror = GetLastError();
+    if (lasterror == WAIT_TIMEOUT) {
+      IOCPThread->enqueueJob(
+          [this](utils::ThreadPool* th, std::uint8_t __) { iocpWatcher_(th); },
+          0);
+      return;
+    }
+    data->event = IOCPEVENT::QUIT;
+    spdlog::debug("Disconnected. [{}]",
+                  (std::string)(data->socket->remoteAddr));
+    delete data;
+  } else {
+    data->transferredbytes = cbTransfrred;
+  }
+
+  std::vector<char> buf(16384);  // SSL_read최대 반환 크기
+  int red_data = 0;
+  std::lock_guard lock(*GetRecvQueueMutex_(sock));
+  auto queue_list = GetRecvQueue_(data->socket->sock);
+  if (data->event == IOCPEVENT::READ) {
+    if (proto_ == SessionProtocol::TLS || proto_ == SessionProtocol::QUIC) {
+      ::BIO_write(::SSL_get_rbio(data->ssl.get()), data->wsabuf.buf,
+                  cbTransfrred);
+
+      while ((red_data = ::SSL_read(data->ssl.get(), buf.data(), buf.size())) >
+             0) {
+        queue_list->emplace_back(std::make_pair(
+            std::vector<char>(buf.begin(), buf.begin() + red_data), 0));
+      }
+    } else {
+      ::memcpy(buf.data(), data->wsabuf.buf, data->transferredbytes);
+      queue_list->emplace_back(std::make_pair(
+          std::vector<char>(buf.begin(), buf.begin() + data->transferredbytes),
+          0));
+    }
+    DWORD recvbytes = 0, flags = 0;
+
+    IOCPPASSINDATA* recv_data = new IOCPPASSINDATA(data->bufsize);
+    recv_data->event = IOCPEVENT::READ;
+    recv_data->socket = data->socket;
+
+    delete data;
+    ::WSARecv(recv_data->socket->sock, &recv_data->wsabuf, 1, &recvbytes,
+              &flags, &recv_data->overlapped, NULL);
+  } else {  // WRITE 시, 무시한다.
+    delete data;
+  }
+  IOCPThread->enqueueJob(
+      [this](utils::ThreadPool* th, std::uint8_t __) { iocpWatcher_(th); }, 0);
+}
+
 std::shared_ptr<std::list<IOCPPASSINDATA*>> IOCP::GetSendQueue_(SOCKET sock) {
   std::lock_guard lock(socket_mod_mutex_);
   if (send_queue_.find(sock) == send_queue_.end()) {
@@ -131,9 +230,9 @@ void IOCP::packet_sender_(SOCKET sock) {
     int data_len = 0;
 
     if (proto_ == SessionProtocol::TLS || proto_ == SessionProtocol::QUIC) {
-      int ret = ::SSL_write(front->ssl, front->wsabuf.buf, front->wsabuf.len);
+      int ret = ::SSL_write(front->ssl.get(), front->wsabuf.buf, front->wsabuf.len);
       if (ret <= 0) {
-        int err = ::SSL_get_error(front->ssl, ret);
+        int err = ::SSL_get_error(front->ssl.get(), ret);
         if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) {
           queue->push_front(front);
           break;
@@ -143,7 +242,7 @@ void IOCP::packet_sender_(SOCKET sock) {
         break;
       }
 
-      while ((data_len = ::BIO_read(front->wbio, buf.data(), buf.size())) > 0) {
+      while ((data_len = ::BIO_read(::SSL_get_wbio(front->ssl.get()), buf.data(), buf.size())) > 0) {
         wsabuf.buf = buf.data();
         wsabuf.len = data_len;
 

include/socket/iocp.h

@@ -5,8 +5,8 @@
 
 #include <functional>
 #include <queue>
-#include <vector>
 #include <random>
+#include <vector>
 
 #include "socket.h"
 #include "utils/thread_pool.h"
@@ -31,13 +31,11 @@ class IOCP;
 
 enum class IOCPEVENT { QUIT, READ, WRITE };
 
-struct IOCPPASSINDATA { // 얘 double free 문제 있음..
+struct IOCPPASSINDATA {
   OVERLAPPED overlapped;
   IOCPEVENT event;
   std::shared_ptr<Socket> socket;
-  SSL* ssl;
-  BIO* rbio; // bio는 ssl별로 달라야 하므로 분리해야 함..
-  BIO* wbio;
+  std::shared_ptr<SSL> ssl;
   std::uint32_t transferredbytes;
   WSABUF wsabuf;
   std::uint32_t bufsize;
@@ -50,8 +48,7 @@ struct IOCPPASSINDATA { // 얘 double free 문제 있음..
     std::memset(&overlapped, 0, sizeof(overlapped));
     event = IOCPEVENT::QUIT;
     socket = nullptr;
-    rbio = nullptr;
-    wbio = nullptr;
+    ssl = nullptr;
     transferredbytes = 0;
     this->bufsize = bufsize;
     IOCPInstance = nullptr;
@@ -63,10 +60,9 @@ struct IOCPPASSINDATA { // 얘 double free 문제 있음..
     std::memset(&overlapped, 0, sizeof(overlapped));
     event = IOCPEVENT::QUIT;
     socket = nullptr;
-    ssl = ::SSL_new(ctx);
-    rbio = ::BIO_new(::BIO_s_mem());
-    wbio = ::BIO_new(::BIO_s_mem());
-    ::SSL_set_bio(ssl, rbio, wbio);
+    ssl = std::shared_ptr<SSL>(::SSL_new(ctx), ::SSL_free);
+    ::SSL_set_bio(ssl.get(), ::BIO_new(::BIO_s_mem()),
+                  ::BIO_new(::BIO_s_mem()));
     transferredbytes = 0;
     this->bufsize = bufsize;
     IOCPInstance = nullptr;
@@ -80,8 +76,6 @@ struct IOCPPASSINDATA { // 얘 double free 문제 있음..
       std::memset(&overlapped, 0, sizeof(overlapped));
       event = other.event;
       socket = other.socket;
-      rbio = other.rbio;
-      wbio = other.wbio;
       transferredbytes = other.transferredbytes;
       bufsize = other.bufsize;
       IOCPInstance = other.IOCPInstance;
@@ -104,8 +98,6 @@ struct IOCPPASSINDATA { // 얘 double free 문제 있음..
       std::memset(&overlapped, 0, sizeof(overlapped));
       event = other.event;
       socket = other.socket;
-      rbio = other.rbio;
-      wbio = other.wbio;
       transferredbytes = other.transferredbytes;
       bufsize = other.bufsize;
       IOCPInstance = other.IOCPInstance;
@@ -126,93 +118,21 @@ class IOCP {
   IOCP();
   ~IOCP();
 
-  void init(utils::ThreadPool* __IOCPThread, SessionProtocol proto) {
-    IOCPThread_ = __IOCPThread;
-    proto_ = proto;
-
-#ifdef _WIN32
-    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(
-          [this](utils::ThreadPool* th, std::uint8_t __) { iocpWatcher_(th); },
-          0);
-#endif
-  }
+  void init(utils::ThreadPool* __IOCPThread, SessionProtocol proto);
 
   void destruct();
 
   void registerSocket(IOCPPASSINDATA* data);
 
+  int recv(IOCPPASSINDATA* data);
   // data는 한 가지 소켓에 보내는 패킷만 담아야 합니다
   int send(SOCKET sock, std::vector<IOCPPASSINDATA*>* data);
-  int recv(IOCPPASSINDATA* data);
+
+  int GetRecvedPacketCount(SOCKET sock);
 
  private:
 #ifdef _WIN32
-  void iocpWatcher_(utils::ThreadPool* IOCPThread) {
-    IOCPPASSINDATA* data;
-    SOCKET sock;
-    DWORD cbTransfrred;
-    int jitter = jitterDist_(gen_);
-    int retVal = GetQueuedCompletionStatus(completionPort_, &cbTransfrred,
-                                           (PULONG_PTR)&sock,
-                                           (LPOVERLAPPED*)&data, 1000 + jitter);
-
-    if (retVal == 0 || cbTransfrred == 0) {
-      DWORD lasterror = GetLastError();
-      if (lasterror == WAIT_TIMEOUT) {
-        IOCPThread->enqueueJob([this](utils::ThreadPool* th,
-                                      std::uint8_t __) { iocpWatcher_(th); },
-                               0);
-        return;
-      }
-      data->event = IOCPEVENT::QUIT;
-      spdlog::debug("Disconnected. [{}]",
-                    (std::string)(data->socket->remoteAddr));
-      delete data;
-    } else {
-      data->transferredbytes = cbTransfrred;
-    }
-
-    std::vector<char> buf(16384);  // SSL_read최대 반환 크기
-    int red_data = 0;
-    auto queue_list = GetRecvQueue_(data->socket->sock);
-    if (data->event == IOCPEVENT::READ) {
-      if (proto_ == SessionProtocol::TLS || proto_ == SessionProtocol::QUIC) {
-        ::BIO_write(data->rbio, data->wsabuf.buf, cbTransfrred);
-
-        while ((red_data = ::SSL_read(data->ssl, buf.data(), buf.size())) > 0) {
-          queue_list->emplace_back(std::make_pair(
-              std::vector<char>(buf.begin(), buf.begin() + red_data), 0));
-        }
-      } else {
-        ::memcpy(buf.data(), data->wsabuf.buf, data->transferredbytes);
-        queue_list->emplace_back(std::make_pair(
-            std::vector<char>(buf.begin(),
-                              buf.begin() + data->transferredbytes),
-            0));
-      }
-      DWORD recvbytes = 0, flags = 0;
-      ::WSARecv(data->socket->sock, &data->wsabuf, 1, &recvbytes, &flags,
-                &data->overlapped, NULL);
-    } else {  // WRITE 시, 무시한다.
-      delete data;
-    }
-    IOCPThread->enqueueJob(
-        [this](utils::ThreadPool* th, std::uint8_t __) { iocpWatcher_(th); },
-        0);
-  }
+  void iocpWatcher_(utils::ThreadPool* IOCPThread);
 #elif __linux__
 
 #endif