From 83939f5a22032371ef549a05d734c275394c40d1 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Fri, 31 Dec 2021 00:10:51 +0300 Subject: [PATCH] Test ZCTR --- src/fio_sec_osd.cpp | 79 ++++++++++++++++++++++++++----------- src/stub_osd.cpp | 94 ++++++++++++++++++++++++++++++++++----------- 2 files changed, 128 insertions(+), 45 deletions(-) diff --git a/src/fio_sec_osd.cpp b/src/fio_sec_osd.cpp index 0d0f6c33..f5f92938 100644 --- a/src/fio_sec_osd.cpp +++ b/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 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) { diff --git a/src/stub_osd.cpp b/src/stub_osd.cpp index 485f4bbe..61819ea7 100644 --- a/src/stub_osd.cpp +++ b/src/stub_osd.cpp @@ -24,7 +24,9 @@ */ #include +#include #include +#include #include #include #include @@ -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); }