Test ZCTR

src/fio_sec_osd.cpp

@@ -42,6 +42,7 @@ struct op_buf_t
 struct sec_data
 {
     int connect_fd;
+    int data_fd;
     /* block_size = 1 << block_order (128KB by default) */
     uint64_t block_order = 17, block_size = 1 << 17;
     std::unordered_map<uint64_t, op_buf_t*> queue;
@@ -157,6 +158,7 @@ static void sec_cleanup(struct thread_data *td)
     sec_data *bsd = (sec_data*)td->io_ops_data;
     if (bsd)
     {
+        close(bsd->data_fd);
         close(bsd->connect_fd);
         delete bsd;
     }
@@ -170,20 +172,20 @@ static int sec_init(struct thread_data *td)
     bsd->block_order = o->block_order == 0 ? 17 : o->block_order;
     bsd->block_size = 1 << o->block_order;
 
-    sockaddr addr;
-    if (!string_to_addr(std::string(o->host ? o->host : "127.0.0.1"), false, o->port > 0 ? o->port : 11203, &addr))
+    struct sockaddr_storage addr = { 0 };
+    if (!string_to_addr(o->host ? o->host : "127.0.0.1", false, o->port > 0 ? o->port : 11203, (struct sockaddr*)&addr))
     {
         fprintf(stderr, "server address: %s is not valid\n", o->host ? o->host : "127.0.0.1");
         return 1;
     }
 
-    bsd->connect_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    bsd->connect_fd = socket(addr.ss_family, SOCK_STREAM, 0);
     if (bsd->connect_fd < 0)
     {
         perror("socket");
         return 1;
     }
-    if (connect(bsd->connect_fd, (sockaddr*)&addr, sizeof(addr)) < 0)
+    if (connect(bsd->connect_fd, (sockaddr*)&addr, addr.ss_family == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6)) < 0)
     {
         perror("connect");
         return 1;
@@ -199,6 +201,39 @@ static int sec_init(struct thread_data *td)
         }
     }
 
+    if (!string_to_addr(o->host ? o->host : "127.0.0.1", false, 1 + (o->port > 0 ? o->port : 11203), (sockaddr*)&addr))
+    {
+        fprintf(stderr, "server address: %s is not valid\n", o->host ? o->host : "127.0.0.1");
+        return 1;
+    }
+
+    bsd->data_fd = socket(addr.ss_family, SOCK_STREAM, 0);
+    if (bsd->data_fd < 0)
+    {
+        perror("socket");
+        return 1;
+    }
+/*    int mss = 4096;
+    if (setsockopt(bsd->data_fd, IPPROTO_TCP, TCP_MAXSEG, &mss, sizeof(mss)) < 0)
+    {
+        perror("setsockopt TCP_MAXSEG");
+        return 1;
+    }*/
+    if (connect(bsd->data_fd, (sockaddr*)&addr, addr.ss_family == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6)) < 0)
+    {
+        perror("connect");
+        return 1;
+    }
+    setsockopt(bsd->data_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
+    if (o->zerocopy_send)
+    {
+        if (setsockopt(bsd->data_fd, SOL_SOCKET, SO_ZEROCOPY, &one, sizeof(one)) < 0)
+        {
+            perror("setsockopt zerocopy");
+            return 1;
+        }
+    }
+
     // FIXME: read config (block size) from OSD
 
     return 0;
@@ -298,16 +333,18 @@ static enum fio_q_status sec_queue(struct thread_data *td, struct io_u *io)
     bsd->op_n++;
     bsd->queue[n] = op_buf;
 
-    iovec iov[2] = { { .iov_base = op.buf, .iov_len = OSD_PACKET_SIZE } };
-    int iovcnt = 1, wtotal = OSD_PACKET_SIZE;
     if (io->ddir == DDIR_WRITE)
     {
-        iov[iovcnt++] = { .iov_base = io->xfer_buf, .iov_len = io->xfer_buflen };
-        wtotal += io->xfer_buflen;
+        // It may make you laugh but ZCTR is only stable if we write data before header :-) O_o
+        if (write_blocking(bsd->data_fd, io->xfer_buf, io->xfer_buflen) != io->xfer_buflen)
+        {
+            perror("write");
+            exit(1);
+        }
     }
-    if (sendv_blocking(bsd->connect_fd, iov, iovcnt, opt->zerocopy_send ? MSG_ZEROCOPY : 0) != wtotal)
+    if (write_blocking(bsd->connect_fd, op.buf, OSD_PACKET_SIZE) != OSD_PACKET_SIZE)
     {
-        perror("sendmsg");
+        perror("write");
         exit(1);
     }
 
@@ -346,23 +383,21 @@ static int sec_getevents(struct thread_data *td, unsigned int min, unsigned int
                 fprintf(stderr, "Short read: retval = %ld instead of %llu\n", reply.hdr.retval, io->xfer_buflen);
                 exit(1);
             }
-            // Support bitmap
-            uint64_t bitmap = 0;
-            int iovcnt = 0;
-            iovec iov[2];
             if (reply.sec_rw.attr_len > 0)
             {
                 if (reply.sec_rw.attr_len <= 8)
-                    iov[iovcnt++] = { .iov_base = &bitmap, .iov_len = reply.sec_rw.attr_len };
+                {
+                    uint64_t bitmap = 0;
+                    read_blocking(bsd->connect_fd, &bitmap, reply.sec_rw.attr_len);
+                }
                 else
-                    iov[iovcnt++] = { .iov_base = (void*)(bitmap = (uint64_t)malloc(reply.sec_rw.attr_len)), .iov_len = reply.sec_rw.attr_len };
-            }
-            iov[iovcnt++] = { .iov_base = io->xfer_buf, .iov_len = io->xfer_buflen };
-            readv_blocking(bsd->connect_fd, iov, iovcnt);
-            if (reply.sec_rw.attr_len > 8)
-            {
-                free((void*)bitmap);
+                {
+                    void *bitmap = malloc(reply.sec_rw.attr_len);
+                    read_blocking(bsd->connect_fd, bitmap, reply.sec_rw.attr_len);
+                    free(bitmap);
+                }
             }
+            read_blocking(bsd->data_fd, io->xfer_buf, io->xfer_buflen);
         }
         else if (io->ddir == DDIR_WRITE)
         {

src/stub_osd.cpp

@@ -24,7 +24,9 @@
  */
 
 #include <sys/types.h>
+#include <sys/mman.h>
 #include <sys/socket.h>
+#include <poll.h>
 #include <netinet/in.h>
 #include <netinet/tcp.h>
 #include <arpa/inet.h>
@@ -43,19 +45,32 @@
 
 int bind_stub(std::string bind_address, int bind_port);
 
-void run_stub(int peer_fd);
+void run_stub(int peer_fd, int peer_data_fd);
 
 int main(int narg, char *args[])
 {
     int listen_fd = bind_stub("0.0.0.0", 11203);
+    int listen_data_fd = bind_stub("0.0.0.0", 11204);
+/*    int mss = 8192;
+    if (setsockopt(listen_data_fd, IPPROTO_TCP, TCP_MAXSEG, &mss, sizeof(mss)) < 0)
+    {
+        throw std::runtime_error(std::string("setsockopt TCP_MAXSEG: ") + strerror(errno));
+    }
+    int rcvlowat = 4096;
+    if (setsockopt(listen_data_fd, SOL_SOCKET, SO_RCVLOWAT, &rcvlowat, sizeof(rcvlowat)) < 0)
+    {
+        throw std::runtime_error(std::string("setsockopt SO_RCVLOWAT: ") + strerror(errno));
+    }*/
     // Accept new connections
     sockaddr addr;
-    socklen_t peer_addr_size = sizeof(addr);
-    int peer_fd;
+    socklen_t peer_addr_size;
+    int peer_fd, peer_data_fd;
+    const int one = 1;
     while (1)
     {
         printf("stub_osd: waiting for 1 client\n");
-        peer_fd = accept(listen_fd, &addr, &peer_addr_size);
+        peer_addr_size = sizeof(addr);
+        peer_fd = accept(listen_fd, (sockaddr*)&addr, &peer_addr_size);
         if (peer_fd == -1)
         {
             if (errno == EAGAIN)
@@ -63,15 +78,27 @@ int main(int narg, char *args[])
             else
                 throw std::runtime_error(std::string("accept: ") + strerror(errno));
         }
-        printf("stub_osd: new client %d: connection from %s\n", peer_fd,
-            addr_to_string(addr).c_str());
-        int one = 1;
         setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
-        run_stub(peer_fd);
-        close(peer_fd);
-        printf("stub_osd: client %d disconnected\n", peer_fd);
-        // Try to accept next connection
+        printf("stub_osd: new client %d: connection from %s\n", peer_fd,
+            addr_to_string(*((sockaddr*)&addr)).c_str());
+        printf("stub_osd: waiting for 1 data connection\n");
         peer_addr_size = sizeof(addr);
+        peer_data_fd = accept(listen_data_fd, (sockaddr*)&addr, &peer_addr_size);
+        if (peer_data_fd == -1)
+        {
+            if (errno == EAGAIN)
+                continue;
+            else
+                throw std::runtime_error(std::string("accept: ") + strerror(errno));
+        }
+        setsockopt(peer_data_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
+        printf("stub_osd: new client %d: data connection from %s\n", peer_data_fd,
+            addr_to_string(*((sockaddr*)&addr)).c_str());
+        run_stub(peer_fd, peer_data_fd);
+        close(peer_data_fd);
+        close(peer_fd);
+        printf("stub_osd: client %d / data %d disconnected\n", peer_fd, peer_data_fd);
+        // Try to accept next connection
     }
     return 0;
 }
@@ -80,13 +107,13 @@ int bind_stub(std::string bind_address, int bind_port)
 {
     int listen_backlog = 128;
 
-    sockaddr addr;
-    if (!string_to_addr(bind_address, 0, bind_port, &addr))
+    sockaddr_storage addr = { 0 };
+    if (!string_to_addr(bind_address, 0, bind_port, (sockaddr*)&addr))
     {
         throw std::runtime_error("bind address "+bind_address+" is not valid");
     }
 
-    int listen_fd = socket(addr.sa_family, SOCK_STREAM, 0);
+    int listen_fd = socket(addr.ss_family, SOCK_STREAM, 0);
     if (listen_fd < 0)
     {
         throw std::runtime_error(std::string("socket: ") + strerror(errno));
@@ -94,7 +121,7 @@ int bind_stub(std::string bind_address, int bind_port)
     int enable = 1;
     setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
 
-    if (bind(listen_fd, &addr, sizeof(addr)) < 0)
+    if (bind(listen_fd, (sockaddr*)&addr, addr.ss_family == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6)) < 0)
     {
         close(listen_fd);
         throw std::runtime_error(std::string("bind: ") + strerror(errno));
@@ -109,11 +136,16 @@ int bind_stub(std::string bind_address, int bind_port)
     return listen_fd;
 }
 
-void run_stub(int peer_fd)
+void run_stub(int peer_fd, int peer_data_fd)
 {
     osd_any_op_t op;
     osd_any_reply_t reply = { 0 };
-    void *buf = NULL;
+    unsigned bufsize = 4*1024*1024;
+    void *buf = mmap(NULL, bufsize, PROT_READ, MAP_SHARED, peer_data_fd, 0);
+    if (buf == MAP_FAILED)
+    {
+        throw std::runtime_error(std::string("mmap: ") + strerror(errno));
+    }
     while (1)
     {
         int r = read_blocking(peer_fd, op.buf, OSD_PACKET_SIZE);
@@ -132,19 +164,36 @@ void run_stub(int peer_fd)
         if (op.hdr.opcode == OSD_OP_SEC_READ)
         {
             reply.hdr.retval = op.sec_rw.len;
-            buf = malloc(op.sec_rw.len);
+            void *buf = malloc(op.sec_rw.len);
             r = write_blocking(peer_fd, reply.buf, OSD_PACKET_SIZE);
             if (r == OSD_PACKET_SIZE)
-                r = write_blocking(peer_fd, buf, op.sec_rw.len);
+                r = write_blocking(peer_data_fd, buf, op.sec_rw.len);
             free(buf);
             if (r < op.sec_rw.len)
                 break;
         }
         else if (op.hdr.opcode == OSD_OP_SEC_WRITE || op.hdr.opcode == OSD_OP_SEC_WRITE_STABLE)
         {
-            buf = malloc(op.sec_rw.len);
-            r = read_blocking(peer_fd, buf, op.sec_rw.len);
-            free(buf);
+            struct pollfd pfd = { .fd = peer_data_fd, .events = POLLIN };
+            poll(&pfd, 1, 10000);
+            struct tcp_zerocopy_receive zc = { .address = (uint64_t)buf, .length = op.sec_rw.len };
+            socklen_t zc_len = sizeof(zc);
+            r = getsockopt(peer_data_fd, IPPROTO_TCP, TCP_ZEROCOPY_RECEIVE, &zc, &zc_len);
+            r = r == -1 ? 0 : zc.length;
+            if (r > 0)
+            {
+                uint64_t hash = 0;
+                for (int k = 0; k < r/8; k++)
+                    hash ^= ((uint64_t*)buf)[k];
+                printf("ZCTR: op=%lx r=%d len=%d skip=%d hash=%lx\n", op.hdr.id, r, zc.length, zc.recv_skip_hint, hash);
+            }
+            if (r < op.sec_rw.len)
+            {
+                int rest = op.sec_rw.len - r;
+                void *buf = malloc(rest);
+                r += read_blocking(peer_data_fd, buf, rest);
+                free(buf);
+            }
             reply.hdr.retval = op.sec_rw.len;
             if (r == op.sec_rw.len)
                 r = write_blocking(peer_fd, reply.buf, OSD_PACKET_SIZE);
@@ -166,5 +215,4 @@ void run_stub(int peer_fd)
             break;
         }
     }
-    free(buf);
 }