Test: replace EPOLLIN with blocking sockets + io_uring reads

osd.cpp

@@ -196,7 +196,6 @@ restart:
             {
                 char peer_str[256];
                 printf("osd: new client %d: connection from %s port %d\n", peer_fd, inet_ntop(AF_INET, &addr.sin_addr, peer_str, 256), ntohs(addr.sin_port));
-                fcntl(peer_fd, F_SETFL, fcntl(listen_fd, F_GETFL, 0) | O_NONBLOCK);
                 int one = 1;
                 setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
                 clients[peer_fd] = {
@@ -208,7 +207,7 @@ restart:
                 // Add FD to epoll
                 epoll_event ev;
                 ev.data.fd = peer_fd;
-                ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
+                ev.events = EPOLLET | EPOLLRDHUP;
                 if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, peer_fd, &ev) < 0)
                 {
                     throw std::runtime_error(std::string("epoll_ctl: ") + strerror(errno));

osd.h

@@ -131,8 +131,8 @@ struct osd_client_t
     int read_ready = 0;
     osd_op_t *read_op = NULL;
     int read_reply_id = 0;
-    iovec read_iov;
-    msghdr read_msg;
+    iovec read_iov = { 0 };
+    msghdr read_msg = { 0 };
     void *read_buf = NULL;
     int read_remaining = 0;
     int read_state = 0;
@@ -215,6 +215,7 @@ class osd_t
     // event loop, socket read/write
     void loop();
     void handle_epoll_events();
+    bool try_receive(osd_client_t & cl);
     void read_requests();
     void handle_read(ring_data_t *data, int peer_fd);
     void handle_op_hdr(osd_client_t *cl);

osd_peering.cpp

@@ -89,7 +89,7 @@ void osd_t::connect_peer(osd_num_t osd_num, const char *peer_host, int peer_port
     // Add FD to epoll (EPOLLOUT for tracking connect() result)
     epoll_event ev;
     ev.data.fd = peer_fd;
-    ev.events = EPOLLOUT | EPOLLIN | EPOLLRDHUP | EPOLLET;
+    ev.events = EPOLLOUT | EPOLLRDHUP | EPOLLET;
     if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, peer_fd, &ev) < 0)
     {
         throw std::runtime_error(std::string("epoll_ctl: ") + strerror(errno));
@@ -115,12 +115,13 @@ void osd_t::handle_connect_result(int peer_fd)
     }
     int one = 1;
     setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
+    fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) & ~O_NONBLOCK);
     // Disable EPOLLOUT on this fd
     cl.connect_callback = NULL;
     cl.peer_state = PEER_CONNECTED;
     epoll_event ev;
     ev.data.fd = peer_fd;
-    ev.events = EPOLLIN | EPOLLRDHUP | EPOLLET;
+    ev.events = EPOLLRDHUP | EPOLLET;
     if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, peer_fd, &ev) < 0)
     {
         throw std::runtime_error(std::string("epoll_ctl: ") + strerror(errno));

osd_receive.cpp

@@ -1,16 +1,12 @@
 #include "osd.h"
 
-void osd_t::read_requests()
+bool osd_t::try_receive(osd_client_t & cl)
 {
-    for (int i = 0; i < read_ready_clients.size(); i++)
-    {
-        int peer_fd = read_ready_clients[i];
-        auto & cl = clients[peer_fd];
+    int peer_fd = cl.peer_fd;
     io_uring_sqe* sqe = ringloop->get_sqe();
     if (!sqe)
     {
-            read_ready_clients.erase(read_ready_clients.begin(), read_ready_clients.begin() + i);
-            return;
+        return false;
     }
     ring_data_t* data = ((ring_data_t*)sqe->user_data);
     if (!cl.read_buf)
@@ -33,8 +29,18 @@ void osd_t::read_requests()
     cl.read_msg.msg_iovlen = 1;
     data->callback = [this, peer_fd](ring_data_t *data) { handle_read(data, peer_fd); };
     my_uring_prep_recvmsg(sqe, peer_fd, &cl.read_msg, 0);
+    return true;
+}
+
+void osd_t::read_requests()
+{
+    for (auto & p: clients)
+    {
+        if (p.second.peer_state == PEER_CONNECTED && p.second.read_iov.iov_len == 0)
+        {
+            try_receive(p.second);
+        }
     }
-    read_ready_clients.clear();
 }
 
 void osd_t::handle_read(ring_data_t *data, int peer_fd)
@@ -43,11 +49,9 @@ void osd_t::handle_read(ring_data_t *data, int peer_fd)
     if (cl_it != clients.end())
     {
         auto & cl = cl_it->second;
+        cl.read_iov.iov_len = 0;
         if (data->res == -EAGAIN)
         {
-            cl.read_ready--;
-            if (cl.read_ready > 0)
-                read_ready_clients.push_back(peer_fd);
             return;
         }
         else if (data->res < 0)
@@ -57,7 +61,6 @@ void osd_t::handle_read(ring_data_t *data, int peer_fd)
             stop_client(peer_fd);
             return;
         }
-        read_ready_clients.push_back(peer_fd);
         if (data->res > 0)
         {
             cl.read_remaining -= data->res;