diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp index 4f0df5c2..00e46ccc 100644 --- a/src/client/messenger.cpp +++ b/src/client/messenger.cpp @@ -399,7 +399,7 @@ void osd_messenger_t::try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_ cl->peer_state = PEER_CONNECTING; cl->connect_timeout_id = -1; cl->osd_num = peer_osd; - cl->in_buf = malloc_or_die(receive_buffer_size); + cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size); clients[client_id] = cl; clients_by_fd[peer_fd] = cl; tfd->set_fd_handler(peer_fd, true, [this](int peer_fd, int epoll_events) @@ -675,7 +675,7 @@ void osd_messenger_t::accept_connections(int listen_fd) cl->peer_port = ntohs(((sockaddr_in*)&addr)->sin_port); cl->peer_fd = peer_fd; cl->peer_state = PEER_CONNECTED; - cl->in_buf = malloc_or_die(receive_buffer_size); + cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size); // Add FD to epoll tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events) { diff --git a/src/client/messenger.h b/src/client/messenger.h index 17d82d76..e8e53299 100644 --- a/src/client/messenger.h +++ b/src/client/messenger.h @@ -35,9 +35,6 @@ #define DEFAULT_MIN_ZEROCOPY_SEND_SIZE 32*1024 -#define MSGR_SENDP_HDR 1 -#define MSGR_SENDP_FREE 2 - #define MAX_SIMPLE_PAYLOAD_SIZE 1048576 struct msgr_sendp_t @@ -67,7 +64,7 @@ struct osd_client_t osd_num_t in_osd_num = 0; bool is_incoming = false; - void *in_buf = NULL; + uint8_t *in_buf = NULL; #ifdef WITH_RDMA msgr_rdma_connection_t *rdma_conn = NULL; @@ -76,11 +73,12 @@ struct osd_client_t // Read state int read_ready = 0; osd_op_t *read_op = NULL; + size_t read_op_size = 0; + size_t read_op_pos = 0; iovec read_iov = { 0 }; msghdr read_msg = { 0 }; - int read_remaining = 0; - int read_state = 0; - osd_op_buf_list_t recv_list; + std::vector recv_list; + size_t recv_list_size = 0; uint64_t read_op_id = 1; bool check_sequencing = false; bool enable_pg_locks = false; @@ -96,10 +94,14 @@ struct osd_client_t std::set dirty_pgs; // Write state + std::deque write_ops; + osd_op_t *write_op = NULL; + size_t write_op_pos = 0; msghdr write_msg = { 0 }; int write_state = 0; - std::vector send_list, next_send_list; - std::vector outbox, next_outbox; + std::vector send_list; + size_t send_list_size = 0; + std::deque send_free_ops; std::vector zc_free_list; ~osd_client_t(); @@ -250,17 +252,22 @@ protected: bool try_send(osd_client_t *cl); void handle_send(int result, bool prev, bool more, osd_client_t *cl); + size_t op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_len); + void op_get_write_buffers(osd_client_t *cl, std::vector & lst); - bool handle_read(int result, osd_client_t *cl); - bool handle_read_buffer(osd_client_t *cl, void *curbuf, int remain); - bool handle_finished_read(osd_client_t *cl); - bool handle_op_hdr(osd_client_t *cl); - bool handle_reply_hdr(osd_client_t *cl); - void handle_reply_ready(osd_op_t *op); + void handle_read(int result, osd_client_t *cl); + bool handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size_t bufsize); + bool handle_hdr(osd_client_t *cl); + bool allocate_op_buffers(osd_client_t *cl); + bool allocate_reply_buffers(osd_client_t *cl, osd_op_t *op); + size_t op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done); + size_t op_get_read_buffers(osd_client_t *cl, std::vector & lst); + void handle_finished_op(osd_client_t *cl); void handle_immediate_ops(); #ifdef WITH_RDMA void try_send_rdma(osd_client_t *cl); + int try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len); bool init_recv_rdma(osd_client_t *cl); void handle_rdma_events(msgr_rdma_context_t *rdma_context); msgr_rdma_context_t* choose_rdma_context(osd_client_t *cl); diff --git a/src/client/msgr_rdma.cpp b/src/client/msgr_rdma.cpp index a2018993..6e6abf66 100644 --- a/src/client/msgr_rdma.cpp +++ b/src/client/msgr_rdma.cpp @@ -571,23 +571,28 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge) cl->rdma_conn->cur_send++; } -static int try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len) +int osd_messenger_t::try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len) { - auto rc = cl->rdma_conn; int total_dst_len = dst_len; - while (dst_len > 0 && rc->send_pos < cl->send_list.size()) + while (dst_len > 0 && cl->write_ops.size()) { - iovec & iov = cl->send_list[rc->send_pos]; - uint32_t len = (uint32_t)(iov.iov_len-rc->send_buf_pos < dst_len - ? iov.iov_len-rc->send_buf_pos : dst_len); - memcpy(dst, (uint8_t*)iov.iov_base+rc->send_buf_pos, len); - dst += len; - dst_len -= len; - rc->send_buf_pos += len; - if (rc->send_buf_pos >= iov.iov_len) + if (!cl->write_op) { - rc->send_pos++; - rc->send_buf_pos = 0; + cl->write_op = cl->write_ops.front(); + cl->write_ops.pop_front(); + } + osd_op_t *op = cl->write_op; + size_t copied = op_copy_to(cl, dst, dst_len); + if (!copied) + { + break; + } + dst += copied; + dst_len -= copied; + if (!cl->write_op && op->op_type == OSD_OP_IN) + { + // this is a reply, free the op after sending it + cl->send_free_ops.push_back(op); } } return total_dst_len-dst_len; @@ -636,6 +641,7 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl) }; try_send_rdma_wr(cl, &sge, 1); rc->send_sizes.push_back(copied); + cl->send_free_ops.push_back(NULL); // end marker } } } @@ -746,51 +752,20 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) else { rc->cur_send--; - uint64_t sent_size = rc->send_sizes.at(0); - rc->send_sizes.erase(rc->send_sizes.begin(), rc->send_sizes.begin()+1); + // byte_len is not filled for send operations + uint64_t sent_size = rc->send_sizes.front(); + rc->send_sizes.pop_front(); rc->send_done_pos += sent_size; rc->send_out_full = false; if (rc->send_done_pos == rc->send_out_size) rc->send_done_pos = 0; assert(rc->send_done_pos < rc->send_out_size); - int send_pos = 0, send_buf_pos = 0; - while (sent_size > 0) + while (cl->send_free_ops.front()) { - if (sent_size >= cl->send_list.at(send_pos).iov_len) - { - sent_size -= cl->send_list[send_pos].iov_len; - send_pos++; - } - else - { - send_buf_pos = sent_size; - sent_size = 0; - } - } - assert(rc->send_pos >= send_pos); - if (rc->send_pos == send_pos) - { - rc->send_buf_pos -= send_buf_pos; - } - rc->send_pos -= send_pos; - for (int i = 0; i < send_pos; i++) - { - if (cl->outbox[i].flags & MSGR_SENDP_FREE) - { - // Reply fully sent - delete cl->outbox[i].op; - } - } - if (send_pos > 0) - { - cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+send_pos); - cl->outbox.erase(cl->outbox.begin(), cl->outbox.begin()+send_pos); - } - if (send_buf_pos > 0) - { - cl->send_list[0].iov_base = (uint8_t*)cl->send_list[0].iov_base + send_buf_pos; - cl->send_list[0].iov_len -= send_buf_pos; + delete cl->send_free_ops.front(); + cl->send_free_ops.pop_front(); } + cl->send_free_ops.pop_front(); try_send_rdma(cl); } } diff --git a/src/client/msgr_rdma.h b/src/client/msgr_rdma.h index f18b8d10..9bf06668 100644 --- a/src/client/msgr_rdma.h +++ b/src/client/msgr_rdma.h @@ -8,8 +8,11 @@ #include #include #include +#include #include "addr_util.h" +struct osd_op_t; + struct msgr_rdma_address_t { ibv_gid gid; @@ -72,9 +75,9 @@ struct msgr_rdma_connection_t int cur_send = 0, cur_recv = 0; int send_pos = 0, send_buf_pos = 0; int next_recv_buf = 0; - std::vector recv_buffers; + std::vector recv_buffers; msgr_rdma_buf_t recv_buf; - std::vector send_sizes; + std::deque send_sizes; msgr_rdma_buf_t send_out; int send_out_pos = 0, send_done_pos = 0, send_out_size = 0; bool send_out_full = false; diff --git a/src/client/msgr_rdmacm.cpp b/src/client/msgr_rdmacm.cpp index 4affe91d..46ffd4cd 100644 --- a/src/client/msgr_rdmacm.cpp +++ b/src/client/msgr_rdmacm.cpp @@ -495,7 +495,7 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev) cl->peer_state = PEER_RDMA; cl->connect_timeout_id = -1; cl->osd_num = peer_osd; - cl->in_buf = malloc_or_die(receive_buffer_size); + cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size); cl->rdma_conn = rc; clients[conn->client_id] = cl; if (conn->timeout_id >= 0) diff --git a/src/client/msgr_receive.cpp b/src/client/msgr_receive.cpp index d0af75e3..a51b9eb5 100644 --- a/src/client/msgr_receive.cpp +++ b/src/client/msgr_receive.cpp @@ -1,6 +1,8 @@ // Copyright (c) Vitaliy Filippov, 2019+ // License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details) +#define _XOPEN_SOURCE +#include #include "messenger.h" #include "msgr_iothread.h" @@ -16,7 +18,11 @@ void osd_messenger_t::read_requests() continue; } auto cl = cl_it->second; - if (cl->read_remaining < receive_buffer_size) + if (cl->read_op && cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size) + { + op_get_read_buffers(cl, cl->recv_list); + } + if (!cl->recv_list.size()) { cl->read_iov.iov_base = cl->in_buf; cl->read_iov.iov_len = receive_buffer_size; @@ -26,10 +32,11 @@ void osd_messenger_t::read_requests() else { cl->read_iov.iov_base = 0; - cl->read_iov.iov_len = cl->read_remaining; - cl->read_msg.msg_iov = cl->recv_list.get_iovec(); - cl->read_msg.msg_iovlen = cl->recv_list.get_size(); + cl->read_iov.iov_len = 0; + cl->read_msg.msg_iov = cl->recv_list.data(); + cl->read_msg.msg_iovlen = cl->recv_list.size(); } + assert(!cl->read_op || cl->read_op_pos < OSD_PACKET_SIZE || cl->read_op_size >= (cl->read_op_pos-OSD_PACKET_SIZE)); cl->refs++; if (ringloop && !use_sync_send_recv) { @@ -51,7 +58,7 @@ void osd_messenger_t::read_requests() } ring_data_t* data = ((ring_data_t*)sqe->user_data); data->callback = [this, cl](ring_data_t *data) { handle_read(data->res, cl); }; - io_uring_prep_recvmsg(sqe, cl->peer_fd, &cl->read_msg, 0); + io_uring_prep_recvmsg(sqe, cl->peer_fd, &cl->read_msg, cl->recv_list.size() ? MSG_WAITALL : 0); if (iothread) { iothread->add_sqe(sqe_local); @@ -71,14 +78,12 @@ void osd_messenger_t::read_requests() read_ready_clients.clear(); } -bool osd_messenger_t::handle_read(int result, osd_client_t *cl) +void osd_messenger_t::handle_read(int result, osd_client_t *cl) { - bool ret = false; - cl->read_msg.msg_iovlen = 0; cl->refs--; if (cl->peer_state == PEER_RDMA) { - return true; + return; } if (cl->peer_state == PEER_STOPPED) { @@ -86,7 +91,7 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl) { destroy_client(cl); } - return false; + return; } if (result <= 0 && result != -EAGAIN && result != -EINTR) { @@ -96,9 +101,50 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl) fprintf(stderr, "Client %ju socket read error: %d (%s). Disconnecting client\n", cl->client_id, -result, strerror(-result)); } stop_client(cl->client_id); - return false; + return; } - if (result == -EAGAIN || result == -EINTR || result < cl->read_iov.iov_len) + bool full_read = false; + if (result > 0) + { + if (cl->read_iov.iov_base == cl->in_buf) + { + full_read = result >= cl->read_iov.iov_len; + if (!handle_read_buffer(cl, cl->in_buf, result)) + { + handle_immediate_ops(); + return; + } + } + else + { + // Reset OSD ping state + cl->ping_time_remaining = 0; + cl->idle_time_remaining = osd_idle_timeout; + // Long data + size_t i = 0; + while (i < cl->recv_list.size() && result >= cl->recv_list[i].iov_len) + { + result -= cl->recv_list[i].iov_len; + i++; + } + if (i < cl->recv_list.size()) + { + cl->recv_list[i].iov_base += result; + cl->recv_list[i].iov_len -= result; + } + else + { + full_read = true; + } + cl->recv_list.erase(cl->recv_list.begin(), cl->recv_list.begin()+i); + if (!cl->recv_list.size()) + { + handle_finished_op(cl); + } + } + } + cl->read_msg.msg_iovlen = 0; + if (result == -EAGAIN || result == -EINTR || !full_read) { cl->read_ready--; if (cl->read_ready > 0) @@ -108,37 +154,7 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl) { read_ready_clients.push_back(cl->client_id); } - if (result > 0) - { - if (cl->read_iov.iov_base == cl->in_buf) - { - if (!handle_read_buffer(cl, cl->in_buf, result)) - { - handle_immediate_ops(); - return false; - } - } - else - { - // Long data - cl->read_remaining -= result; - cl->recv_list.eat(result); - if (cl->recv_list.done >= cl->recv_list.count) - { - if (!handle_finished_read(cl)) - { - handle_immediate_ops(); - return false; - } - } - } - if (result >= cl->read_iov.iov_len) - { - ret = true; - } - } handle_immediate_ops(); - return ret; } void osd_messenger_t::handle_immediate_ops() @@ -163,117 +179,94 @@ void osd_messenger_t::handle_immediate_ops() } } -bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, void *curbuf, int remain) +bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size_t bufsize) { + // Reset OSD ping state + cl->ping_time_remaining = 0; + cl->idle_time_remaining = osd_idle_timeout; // Compose operation(s) from the buffer - while (remain > 0) + size_t done = 0; + while (done < bufsize) { if (!cl->read_op) { cl->read_op = new osd_op_t; cl->read_op->client_id = cl->client_id; cl->read_op->op_type = OSD_OP_IN; - cl->recv_list.push_back(cl->read_op->req.buf, OSD_PACKET_SIZE); - cl->read_remaining = OSD_PACKET_SIZE; - cl->read_state = CL_READ_HDR; + cl->read_op_pos = 0; + cl->read_op_size = 0; } - while (cl->recv_list.done < cl->recv_list.count && remain > 0) + if (cl->read_op_pos < OSD_PACKET_SIZE) { - iovec* cur = cl->recv_list.get_iovec(); - if (cur->iov_len > remain) - { - memcpy(cur->iov_base, curbuf, remain); - cl->read_remaining -= remain; - cur->iov_len -= remain; - cur->iov_base = (uint8_t*)cur->iov_base + remain; - remain = 0; - } - else - { - memcpy(cur->iov_base, curbuf, cur->iov_len); - curbuf = (uint8_t*)curbuf + cur->iov_len; - cl->read_remaining -= cur->iov_len; - remain -= cur->iov_len; - cur->iov_len = 0; - cl->recv_list.done++; - } - } - if (cl->recv_list.done >= cl->recv_list.count) - { - if (!handle_finished_read(cl)) - { - return false; - } - } - } - return true; -} - -bool osd_messenger_t::handle_finished_read(osd_client_t *cl) -{ - // Reset OSD ping state - cl->ping_time_remaining = 0; - cl->idle_time_remaining = osd_idle_timeout; - cl->recv_list.reset(); - if (cl->read_state == CL_READ_HDR) - { - if (cl->read_op->req.hdr.magic == SECONDARY_OSD_REPLY_MAGIC) - return handle_reply_hdr(cl); - else if (cl->read_op->req.hdr.magic == SECONDARY_OSD_OP_MAGIC) - { - if (cl->check_sequencing) - { - if (cl->read_op->req.hdr.id != cl->read_op_id) - { - fprintf(stderr, "Warning: operation sequencing is broken on client %ju: expected num %ju, got %ju, stopping client\n", cl->client_id, cl->read_op_id, cl->read_op->req.hdr.id); - stop_client(cl->client_id); - return false; - } - cl->read_op_id++; - } - if (!handle_op_hdr(cl)) + int len = OSD_PACKET_SIZE - cl->read_op_pos; + if (len > bufsize-done) + len = bufsize-done; + memcpy(cl->read_op->req.buf + cl->read_op_pos, curbuf+done, len); + done += len; + cl->read_op_pos += len; + if (cl->read_op_pos < OSD_PACKET_SIZE) + return true; + if (!handle_hdr(cl)) { stop_client(cl->client_id); return false; } } - else - { - fprintf(stderr, "Received garbage: magic=%jx id=%ju opcode=%jx from client %ju\n", cl->read_op->req.hdr.magic, cl->read_op->req.hdr.id, cl->read_op->req.hdr.opcode, cl->client_id); - stop_client(cl->client_id); - return false; - } - } - else if (cl->read_state == CL_READ_DATA) - { - // Operation is ready - cl->received_ops.push_back(cl->read_op); - set_immediate_ops.push_back(cl->read_op); - cl->read_op = NULL; - cl->read_state = 0; - } - else if (cl->read_state == CL_READ_REPLY_DATA) - { - // Reply is ready - handle_reply_ready(cl->read_op); - cl->read_op = NULL; - cl->read_state = 0; - } - else - { - assert(0); + op_copy_from(cl, curbuf, bufsize, done); } return true; } -bool osd_messenger_t::handle_op_hdr(osd_client_t *cl) +bool osd_messenger_t::handle_hdr(osd_client_t *cl) +{ + if (cl->read_op->req.hdr.magic == SECONDARY_OSD_REPLY_MAGIC) + { + auto req_it = cl->sent_ops.find(cl->read_op->req.hdr.id); + if (req_it == cl->sent_ops.end()) + { + // Command out of sync. Drop connection + fprintf(stderr, "Client %ju command out of sync: id %ju\n", cl->client_id, cl->read_op->req.hdr.id); + return false; + } + osd_op_t *op = req_it->second; + memcpy(op->reply.buf, cl->read_op->req.buf, OSD_PACKET_SIZE); + if (!allocate_reply_buffers(cl, op)) + { + return false; + } + cl->sent_ops.erase(req_it); + delete cl->read_op; + cl->read_op = op; + } + else if (cl->read_op->req.hdr.magic == SECONDARY_OSD_OP_MAGIC) + { + if (cl->check_sequencing) + { + if (cl->read_op->req.hdr.id != cl->read_op_id) + { + fprintf(stderr, "Warning: operation sequencing is broken on client %d: expected num %ju, got %ju, stopping client\n", cl->peer_fd, cl->read_op_id, cl->read_op->req.hdr.id); + return false; + } + cl->read_op_id++; + } + if (!allocate_op_buffers(cl)) + { + return false; + } + } + else + { + fprintf(stderr, "Received garbage: magic=%jx id=%ju opcode=%jx from client %ju\n", cl->read_op->req.hdr.magic, cl->read_op->req.hdr.id, cl->read_op->req.hdr.opcode, cl->client_id); + return false; + } + return true; +} + +bool osd_messenger_t::allocate_op_buffers(osd_client_t *cl) { osd_op_t *cur_op = cl->read_op; - if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ) - { - cl->read_remaining = 0; - } - else if (cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE || + cl->read_op_size = 0; + if (cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE || cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) { if (cur_op->req.sec_rw.attr_len > 0) @@ -291,7 +284,6 @@ bool osd_messenger_t::handle_op_hdr(osd_client_t *cl) cur_op->bitmap = cur_op->rmw_buf = malloc_or_die(cur_op->req.sec_rw.attr_len); else cur_op->bitmap = &cur_op->bmp_data; - cl->recv_list.push_back(cur_op->bitmap, cur_op->req.sec_rw.attr_len); } if (cur_op->req.sec_rw.len > 0) { @@ -305,9 +297,8 @@ bool osd_messenger_t::handle_op_hdr(osd_client_t *cl) return false; } cur_op->buf = memalign_or_die(MEM_ALIGNMENT, cur_op->req.sec_rw.len); - cl->recv_list.push_back(cur_op->buf, cur_op->req.sec_rw.len); } - cl->read_remaining = cur_op->req.sec_rw.len + cur_op->req.sec_rw.attr_len; + cl->read_op_size = cur_op->req.sec_rw.len + cur_op->req.sec_rw.attr_len; } else if (cur_op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || cur_op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK) @@ -324,9 +315,8 @@ bool osd_messenger_t::handle_op_hdr(osd_client_t *cl) return false; } cur_op->buf = memalign_or_die(MEM_ALIGNMENT, cur_op->req.sec_stab.len); - cl->recv_list.push_back(cur_op->buf, cur_op->req.sec_stab.len); } - cl->read_remaining = cur_op->req.sec_stab.len; + cl->read_op_size = cur_op->req.sec_stab.len; } else if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ_BMP) { @@ -342,9 +332,8 @@ bool osd_messenger_t::handle_op_hdr(osd_client_t *cl) return false; } cur_op->buf = memalign_or_die(MEM_ALIGNMENT, cur_op->req.sec_read_bmp.len); - cl->recv_list.push_back(cur_op->buf, cur_op->req.sec_read_bmp.len); } - cl->read_remaining = cur_op->req.sec_read_bmp.len; + cl->read_op_size = cur_op->req.sec_read_bmp.len; } else if (cur_op->req.hdr.opcode == OSD_OP_WRITE) { @@ -360,9 +349,8 @@ bool osd_messenger_t::handle_op_hdr(osd_client_t *cl) return false; } cur_op->buf = memalign_or_die(MEM_ALIGNMENT, cur_op->req.rw.len); - cl->recv_list.push_back(cur_op->buf, cur_op->req.rw.len); } - cl->read_remaining = cur_op->req.rw.len; + cl->read_op_size = cur_op->req.rw.len; } else if (cur_op->req.hdr.opcode == OSD_OP_SHOW_CONFIG) { @@ -379,45 +367,15 @@ bool osd_messenger_t::handle_op_hdr(osd_client_t *cl) } cur_op->buf = malloc_or_die(cur_op->req.show_conf.json_len+1); ((uint8_t*)cur_op->buf)[cur_op->req.show_conf.json_len] = 0; - cl->recv_list.push_back(cur_op->buf, cur_op->req.show_conf.json_len); } - cl->read_remaining = cur_op->req.show_conf.json_len; - } - /*else if (cur_op->req.hdr.opcode == OSD_OP_READ || - cur_op->req.hdr.opcode == OSD_OP_SCRUB || - cur_op->req.hdr.opcode == OSD_OP_DESCRIBE) - { - cl->read_remaining = 0; - }*/ - if (cl->read_remaining > 0) - { - // Read data - cl->read_state = CL_READ_DATA; - } - else - { - // Operation is ready - cl->received_ops.push_back(cur_op); - set_immediate_ops.push_back(cur_op); - cl->read_op = NULL; - cl->read_state = 0; + cl->read_op_size = cur_op->req.show_conf.json_len; } return true; } -bool osd_messenger_t::handle_reply_hdr(osd_client_t *cl) +bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op) { - auto req_it = cl->sent_ops.find(cl->read_op->req.hdr.id); - if (req_it == cl->sent_ops.end() || req_it->second->req.hdr.opcode != cl->read_op->req.hdr.opcode) - { - // Command out of sync. Drop connection - fprintf(stderr, "Client %ju command out of sync: id %ju\n", cl->client_id, cl->read_op->req.hdr.id); - stop_client(cl->client_id); - return false; - } - osd_op_t *op = req_it->second; - memcpy(op->reply.buf, cl->read_op->req.buf, OSD_PACKET_SIZE); - cl->sent_ops.erase(req_it); + cl->read_op_size = 0; if (op->reply.hdr.opcode == OSD_OP_SEC_READ || op->reply.hdr.opcode == OSD_OP_READ) { // Read data. In this case we assume that the buffer is preallocated by the caller (!) @@ -428,97 +386,279 @@ bool osd_messenger_t::handle_reply_hdr(osd_client_t *cl) // Check reply length to not overflow the buffer fprintf(stderr, "Client %ju read reply of different length: expected %u+%u, got %jd+%u\n", cl->client_id, expected_size, op->bitmap_len, op->reply.hdr.retval, bmp_len); - cl->sent_ops[op->req.hdr.id] = op; - stop_client(cl->client_id); return false; } if (op->reply.hdr.retval >= 0 && bmp_len > 0) { assert(op->bitmap); - cl->recv_list.push_back(op->bitmap, bmp_len); - cl->read_remaining += bmp_len; + cl->read_op_size += bmp_len; } if (op->reply.hdr.retval > 0) { assert(op->iov.count > 0); - cl->recv_list.append(op->iov); - cl->read_remaining += op->reply.hdr.retval; + cl->read_op_size += op->reply.hdr.retval; } - if (cl->read_remaining == 0) - { - goto reuse; - } - delete cl->read_op; - cl->read_op = op; - cl->read_state = CL_READ_REPLY_DATA; } else if (op->reply.hdr.opcode == OSD_OP_SEC_LIST && op->reply.hdr.retval > 0) { assert(!op->iov.count); - delete cl->read_op; - cl->read_op = op; - cl->read_state = CL_READ_REPLY_DATA; - cl->read_remaining = sizeof(obj_ver_id) * op->reply.hdr.retval; - op->buf = memalign_or_die(MEM_ALIGNMENT, cl->read_remaining); - cl->recv_list.push_back(op->buf, cl->read_remaining); + cl->read_op_size = sizeof(obj_ver_id) * op->reply.hdr.retval; + op->buf = memalign_or_die(MEM_ALIGNMENT, cl->read_op_size); } else if (op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP && op->reply.hdr.retval > 0) { assert(!op->iov.count); - delete cl->read_op; - cl->read_op = op; - cl->read_state = CL_READ_REPLY_DATA; - cl->read_remaining = op->reply.hdr.retval; + cl->read_op_size = op->reply.hdr.retval; free(op->buf); - op->buf = memalign_or_die(MEM_ALIGNMENT, cl->read_remaining); - cl->recv_list.push_back(op->buf, cl->read_remaining); + op->buf = memalign_or_die(MEM_ALIGNMENT, cl->read_op_size); } else if (op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG && op->reply.hdr.retval > 0) { - delete cl->read_op; - cl->read_op = op; - cl->read_state = CL_READ_REPLY_DATA; - cl->read_remaining = op->reply.hdr.retval; + cl->read_op_size = op->reply.hdr.retval; free(op->buf); op->buf = malloc_or_die(op->reply.hdr.retval); - cl->recv_list.push_back(op->buf, op->reply.hdr.retval); } else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0) { - delete cl->read_op; - cl->read_op = op; - cl->read_state = CL_READ_REPLY_DATA; - cl->read_remaining = op->reply.describe.result_bytes; + cl->read_op_size = op->reply.describe.result_bytes; free(op->buf); op->buf = malloc_or_die(op->reply.describe.result_bytes); - cl->recv_list.push_back(op->buf, op->reply.describe.result_bytes); - } - else - { -reuse: - // It's fine to reuse cl->read_op for the next reply - handle_reply_ready(op); - cl->recv_list.push_back(cl->read_op->req.buf, OSD_PACKET_SIZE); - cl->read_remaining = OSD_PACKET_SIZE; - cl->read_state = CL_READ_HDR; } return true; } -void osd_messenger_t::handle_reply_ready(osd_op_t *op) +size_t osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done) { - // Measure subop latency - timespec tv_end; - clock_gettime(CLOCK_REALTIME, &tv_end); - stats.subop_stat_count[op->req.hdr.opcode]++; - if (!stats.subop_stat_count[op->req.hdr.opcode]) + osd_op_t *op = cl->read_op; + size_t from = cl->read_op_pos-OSD_PACKET_SIZE; + auto op_read_buf = [&](uint8_t *dst, size_t dst_len) { - stats.subop_stat_count[op->req.hdr.opcode]++; - stats.subop_stat_sum[op->req.hdr.opcode] = 0; + if (from < dst_len) + { + size_t n = dst_len-from; + if (n > src_len-done) + n = src_len-done; + memcpy(dst+from, src+done, n); + done += n; + cl->read_op_pos += n; + from += n; + if (from < dst_len) + return false; + from = 0; + } + else + from -= dst_len; + return true; + }; + if (op->op_type == OSD_OP_IN) + { + if (op->req.hdr.opcode == OSD_OP_SEC_WRITE || + op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) + { + if (!op_read_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len)) + return done; + if (!op_read_buf((uint8_t*)op->buf, op->req.sec_rw.len)) + return done; + } + else if (op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || + op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK) + { + if (!op_read_buf((uint8_t*)op->buf, op->req.sec_stab.len)) + return done; + } + else if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP) + { + if (!op_read_buf((uint8_t*)op->buf, op->req.sec_read_bmp.len)) + return done; + } + else if (op->req.hdr.opcode == OSD_OP_WRITE) + { + if (!op_read_buf((uint8_t*)op->buf, op->req.rw.len)) + return done; + } + else if (op->req.hdr.opcode == OSD_OP_SHOW_CONFIG) + { + if (!op_read_buf((uint8_t*)op->buf, op->req.show_conf.json_len)) + return done; + } } - stats.subop_stat_sum[op->req.hdr.opcode] += ( - (tv_end.tv_sec - op->tv_begin.tv_sec)*1000000 + - (tv_end.tv_nsec - op->tv_begin.tv_nsec)/1000 - ); - set_immediate_ops.push_back(op); + else + { + if (op->reply.hdr.opcode == OSD_OP_SEC_READ) + { + if (op->reply.sec_rw.attr_len > 0) + { + if (!op_read_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len)) + return done; + } + if (op->reply.hdr.retval > 0) + { + for (int i = 0; i < op->iov.count; i++) + if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) + return done; + } + } + else if (op->reply.hdr.opcode == OSD_OP_READ) + { + if (op->reply.rw.bitmap_len > 0) + { + if (!op_read_buf((uint8_t*)op->bitmap, op->reply.rw.bitmap_len)) + return done; + } + if (op->reply.hdr.retval > 0) + { + for (int i = 0; i < op->iov.count; i++) + if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) + return done; + } + } + else if (op->reply.hdr.opcode == OSD_OP_SEC_LIST && op->reply.hdr.retval > 0) + { + if (!op_read_buf((uint8_t*)op->buf, sizeof(obj_ver_id) * op->reply.hdr.retval)) + return done; + } + else if ((op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP || + op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG) && op->reply.hdr.retval > 0) + { + if (!op_read_buf((uint8_t*)op->buf, op->reply.hdr.retval)) + return done; + } + else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0) + { + if (!op_read_buf((uint8_t*)op->buf, op->reply.describe.result_bytes)) + return done; + } + } + handle_finished_op(cl); + return done; +} + +size_t osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector & lst) +{ + osd_op_t *op = cl->read_op; + size_t from = cl->read_op_pos-OSD_PACKET_SIZE; + size_t done = 0; + auto op_read_buf = [&](uint8_t *dst, size_t dst_len) + { + if (lst.size() >= IOV_MAX) + return false; + if (from < dst_len) + { + lst.push_back((iovec){ .iov_base = dst+from, .iov_len = dst_len-from }); + cl->read_op_pos += dst_len-from; + done += dst_len-from; + from = 0; + } + else + from -= dst_len; + return true; + }; + if (op->op_type == OSD_OP_IN) + { + if (op->req.hdr.opcode == OSD_OP_SEC_WRITE || + op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) + { + if (!op_read_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len)) + return done; + if (!op_read_buf((uint8_t*)op->buf, op->req.sec_rw.len)) + return done; + } + else if (op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || + op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK) + { + if (!op_read_buf((uint8_t*)op->buf, op->req.sec_stab.len)) + return done; + } + else if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP) + { + if (!op_read_buf((uint8_t*)op->buf, op->req.sec_read_bmp.len)) + return done; + } + else if (op->req.hdr.opcode == OSD_OP_WRITE) + { + if (!op_read_buf((uint8_t*)op->buf, op->req.rw.len)) + return done; + } + else if (op->req.hdr.opcode == OSD_OP_SHOW_CONFIG) + { + if (!op_read_buf((uint8_t*)op->buf, op->req.show_conf.json_len)) + return done; + } + } + else + { + if (op->reply.hdr.opcode == OSD_OP_SEC_READ) + { + if (op->reply.sec_rw.attr_len > 0) + { + if (!op_read_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len)) + return done; + } + if (op->reply.hdr.retval > 0) + { + for (int i = 0; i < op->iov.count; i++) + if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) + return done; + } + } + else if (op->reply.hdr.opcode == OSD_OP_READ) + { + if (op->reply.rw.bitmap_len > 0) + { + if (!op_read_buf((uint8_t*)op->bitmap, op->reply.rw.bitmap_len)) + return done; + } + if (op->reply.hdr.retval > 0) + { + for (int i = 0; i < op->iov.count; i++) + if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) + return done; + } + } + else if (op->reply.hdr.opcode == OSD_OP_SEC_LIST && op->reply.hdr.retval > 0) + { + if (!op_read_buf((uint8_t*)op->buf, sizeof(obj_ver_id) * op->reply.hdr.retval)) + return done; + } + else if ((op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP || + op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG) && op->reply.hdr.retval > 0) + { + if (!op_read_buf((uint8_t*)op->buf, op->reply.hdr.retval)) + return done; + } + else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0) + { + if (!op_read_buf((uint8_t*)op->buf, op->reply.describe.result_bytes)) + return done; + } + } + return done; +} + +void osd_messenger_t::handle_finished_op(osd_client_t *cl) +{ + osd_op_t *op = cl->read_op; + if (op->op_type == OSD_OP_IN) + { + // Operation is ready + cl->received_ops.push_back(op); + } + else + { + // Measure subop (outbound op) latency + timespec tv_end; + clock_gettime(CLOCK_REALTIME, &tv_end); + stats.subop_stat_count[op->req.hdr.opcode]++; + if (!stats.subop_stat_count[op->req.hdr.opcode]) + { + stats.subop_stat_count[op->req.hdr.opcode]++; + stats.subop_stat_sum[op->req.hdr.opcode] = 0; + } + stats.subop_stat_sum[op->req.hdr.opcode] += ( + (tv_end.tv_sec - op->tv_begin.tv_sec)*1000000 + + (tv_end.tv_nsec - op->tv_begin.tv_nsec)/1000 + ); + } + set_immediate_ops.push_back(op); + cl->read_op = NULL; } diff --git a/src/client/msgr_send.cpp b/src/client/msgr_send.cpp index 32250112..6fbf0b65 100644 --- a/src/client/msgr_send.cpp +++ b/src/client/msgr_send.cpp @@ -5,6 +5,12 @@ #include #include +#ifdef WITH_OPENSSL +#include +#include +#include +#endif + #include "messenger.h" #include "msgr_iothread.h" @@ -22,6 +28,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) { clock_gettime(CLOCK_REALTIME, &cur_op->tv_begin); cur_op->req.hdr.id = ++cl->send_op_id; + cl->sent_ops[cur_op->req.hdr.id] = cur_op; } else { @@ -38,82 +45,9 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) } // Can't be not found because client IDs are unique assert(found); - } - auto & to_send_list = cl->write_msg.msg_iovlen ? cl->next_send_list : cl->send_list; - auto & to_outbox = cl->write_msg.msg_iovlen ? cl->next_outbox : cl->outbox; - if (cur_op->op_type == OSD_OP_IN) - { measure_exec(cur_op); - to_send_list.push_back((iovec){ .iov_base = cur_op->reply.buf, .iov_len = OSD_PACKET_SIZE }); - } - else - { - to_send_list.push_back((iovec){ .iov_base = cur_op->req.buf, .iov_len = OSD_PACKET_SIZE }); - cl->sent_ops[cur_op->req.hdr.id] = cur_op; - } - to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = MSGR_SENDP_HDR }); - // Bitmap - if (cur_op->op_type == OSD_OP_IN && - cur_op->req.hdr.opcode == OSD_OP_SEC_READ && - cur_op->reply.sec_rw.attr_len > 0) - { - to_send_list.push_back((iovec){ - .iov_base = cur_op->bitmap, - .iov_len = cur_op->reply.sec_rw.attr_len, - }); - to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 }); - } - else if (cur_op->op_type == OSD_OP_OUT && - (cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE || cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) && - cur_op->req.sec_rw.attr_len > 0) - { - to_send_list.push_back((iovec){ - .iov_base = cur_op->bitmap, - .iov_len = cur_op->req.sec_rw.attr_len, - }); - to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 }); - } - // Operation data - if ((cur_op->op_type == OSD_OP_IN - ? (cur_op->req.hdr.opcode == OSD_OP_READ || - cur_op->req.hdr.opcode == OSD_OP_SEC_READ || - cur_op->req.hdr.opcode == OSD_OP_SEC_LIST || - cur_op->req.hdr.opcode == OSD_OP_SHOW_CONFIG || - cur_op->req.hdr.opcode == OSD_OP_DESCRIBE) - : (cur_op->req.hdr.opcode == OSD_OP_WRITE || - cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE || - cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE || - cur_op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || - cur_op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK || - cur_op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)) && cur_op->iov.count > 0) - { - for (int i = 0; i < cur_op->iov.count; i++) - { - if (cur_op->iov.buf[i].iov_len > 0) - { - assert(cur_op->iov.buf[i].iov_base); - to_send_list.push_back(cur_op->iov.buf[i]); - to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 }); - } - } - } - if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ_BMP) - { - if (cur_op->op_type == OSD_OP_IN && cur_op->reply.hdr.retval > 0) - { - to_send_list.push_back((iovec){ .iov_base = cur_op->buf, .iov_len = (size_t)cur_op->reply.hdr.retval }); - to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 }); - } - else if (cur_op->op_type == OSD_OP_OUT && cur_op->req.sec_read_bmp.len > 0) - { - to_send_list.push_back((iovec){ .iov_base = cur_op->buf, .iov_len = (size_t)cur_op->req.sec_read_bmp.len }); - to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 }); - } - } - if (cur_op->op_type == OSD_OP_IN) - { - to_outbox[to_outbox.size()-1].flags |= MSGR_SENDP_FREE; } + cl->write_ops.push_back(cur_op); #ifdef WITH_RDMA if (cl->peer_state == PEER_RDMA) { @@ -124,7 +58,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) if (!ringloop) { // FIXME: It's worse because it doesn't allow batching - while (cl->outbox.size()) + while (cl->write_ops.size()) { try_send(cl); } @@ -142,11 +76,25 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) bool osd_messenger_t::try_send(osd_client_t *cl) { - if (!cl->send_list.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED || cl->peer_fd < 0) + if (!cl->write_op && !cl->write_ops.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED || cl->peer_fd < 0) { return true; } assert(cl->peer_state != PEER_RDMA); + while ((cl->write_op || cl->write_ops.size()) && cl->send_list.size() < IOV_MAX) + { + if (!cl->write_op) + { + cl->write_op = cl->write_ops.front(); + cl->write_ops.pop_front(); + } + osd_op_t *op = cl->write_op; + op_get_write_buffers(cl, cl->send_list); + if (!cl->write_op && op->op_type == OSD_OP_IN) + { + cl->send_free_ops.push_back(op); + } + } if (ringloop && !use_sync_send_recv) { auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL; @@ -159,20 +107,24 @@ bool osd_messenger_t::try_send(osd_client_t *cl) data_local = {}; } if (!sqe) + { return false; + } + cl->send_list_size = 0; + for (auto & iov: cl->send_list) + { + cl->send_list_size += iov.iov_len; + } cl->write_msg.msg_iov = cl->send_list.data(); cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX; cl->refs++; ring_data_t* data = ((ring_data_t*)sqe->user_data); data->callback = [this, cl](ring_data_t *data) { handle_send(data->res, data->prev, data->more, cl); }; bool use_zc = has_sendmsg_zc && min_zerocopy_send_size >= 0; - if (use_zc && min_zerocopy_send_size > 0) + if (use_zc && min_zerocopy_send_size > 0 && + cl->send_list_size/cl->write_msg.msg_iovlen < min_zerocopy_send_size) { - size_t avg_size = 0; - for (size_t i = 0; i < cl->write_msg.msg_iovlen; i++) - avg_size += cl->write_msg.msg_iov[i].iov_len; - if (avg_size/cl->write_msg.msg_iovlen < min_zerocopy_send_size) - use_zc = false; + use_zc = false; } if (use_zc) { @@ -223,6 +175,7 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t if (!prev) { cl->write_msg.msg_iovlen = 0; + cl->send_list.clear(); } if (!more) { @@ -255,57 +208,26 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t cl->zc_free_list.erase(cl->zc_free_list.begin(), cl->zc_free_list.begin()+i+1); return; } - int done = 0; - while (result > 0 && done < cl->send_list.size()) + if (cl->send_list_size > result) { - iovec & iov = cl->send_list[done]; - if (iov.iov_len <= result) - { - if (cl->outbox[done].flags & MSGR_SENDP_FREE) - { - // Reply fully sent - if (more) - cl->zc_free_list.push_back(cl->outbox[done].op); - else - delete cl->outbox[done].op; - } - result -= iov.iov_len; - done++; - } + fprintf(stderr, "Client %ju socket write error: expected to send " + "%zu bytes with MSG_WAITALL but sent %u. Disconnecting client\n", cl->client_id, cl->send_list_size, result); + stop_client(cl->peer_fd); + return; + } + for (auto op: cl->send_free_ops) + { + if (more) + cl->zc_free_list.push_back(op); else - { - iov.iov_len -= result; - iov.iov_base = (uint8_t*)iov.iov_base + result; - break; - } + delete op; } if (more) - { - int expected = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX; - if (done != expected) - { - fprintf(stderr, "Client %ju socket write error: expected to send " - "%d iovecs with MSG_WAITALL but sent %d. Disconnecting client\n", cl->client_id, expected, done); - stop_client(cl->client_id); - return; - } cl->zc_free_list.push_back(NULL); // end marker - } - if (done > 0) - { - cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+done); - cl->outbox.erase(cl->outbox.begin(), cl->outbox.begin()+done); - } - if (cl->next_send_list.size()) - { - cl->send_list.insert(cl->send_list.end(), cl->next_send_list.begin(), cl->next_send_list.end()); - cl->outbox.insert(cl->outbox.end(), cl->next_outbox.begin(), cl->next_outbox.end()); - cl->next_send_list.clear(); - cl->next_outbox.clear(); - } - cl->write_state = cl->outbox.size() > 0 ? CL_WRITE_READY : 0; + cl->send_free_ops.clear(); + cl->write_state = cl->write_op || cl->write_ops.size() ? CL_WRITE_READY : 0; #ifdef WITH_RDMA - if (cl->rdma_conn && !cl->outbox.size() && cl->peer_state == PEER_RDMA_CONNECTING) + if (cl->rdma_conn && !cl->write_op && !cl->write_ops.size() && cl->peer_state == PEER_RDMA_CONNECTING) { // FIXME: Ignore pings during RDMA state transition if (log_level > 0) @@ -323,3 +245,130 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t write_ready_clients.push_back(cl->client_id); } } + +static inline bool op_write_headers(osd_op_t *op, std::function op_write_buf) +{ + // Header + if (!op_write_buf((op->op_type == OSD_OP_IN ? op->reply.buf : op->req.buf), OSD_PACKET_SIZE)) + return false; + // Bitmap + if (op->op_type == OSD_OP_IN && + op->req.hdr.opcode == OSD_OP_SEC_READ && + op->reply.sec_rw.attr_len > 0) + { + if (!op_write_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len)) + return false; + } + else if (op->op_type == OSD_OP_OUT && + (op->req.hdr.opcode == OSD_OP_SEC_WRITE || op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) && + op->req.sec_rw.attr_len > 0) + { + if (!op_write_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len)) + return false; + } + if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP) + { + if (op->op_type == OSD_OP_IN && op->reply.hdr.retval > 0) + { + if (!op_write_buf((uint8_t*)op->buf, (size_t)op->reply.hdr.retval)) + return false; + } + else if (op->op_type == OSD_OP_OUT && op->req.sec_read_bmp.len > 0) + { + if (!op_write_buf((uint8_t*)op->buf, (size_t)op->req.sec_read_bmp.len)) + return false; + } + } + return true; +} + +static inline bool op_has_data(osd_op_t *op) +{ + return (op->op_type == OSD_OP_IN + ? (op->req.hdr.opcode == OSD_OP_READ || + op->req.hdr.opcode == OSD_OP_SEC_READ || + op->req.hdr.opcode == OSD_OP_SEC_LIST || + op->req.hdr.opcode == OSD_OP_SHOW_CONFIG || + op->req.hdr.opcode == OSD_OP_DESCRIBE) + : (op->req.hdr.opcode == OSD_OP_WRITE || + op->req.hdr.opcode == OSD_OP_SEC_WRITE || + op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE || + op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || + op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK || + op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)) && op->iov.count > 0; +} + +size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_len) +{ + size_t done = 0; + size_t from = cl->write_op_pos; + auto op_write_buf = [&](uint8_t *src, size_t src_len) + { + if (from < src_len) + { + size_t n = src_len-from; + if (n > dst_len-done) + n = dst_len-done; + memcpy(dst+done, src+from, n); + done += n; + cl->write_op_pos += n; + from += n; + if (from < src_len) + return false; + from = 0; + } + else + from -= src_len; + return true; + }; + if (!op_write_headers(cl->write_op, op_write_buf)) + { + return done; + } + // Operation data + if (op_has_data(cl->write_op)) + { + for (int i = 0; i < cl->write_op->iov.count; i++) + { + if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len)) + return done; + } + } + cl->write_op = NULL; + cl->write_op_pos = 0; + return done; +} + +void osd_messenger_t::op_get_write_buffers(osd_client_t *cl, std::vector & lst) +{ + size_t from = cl->write_op_pos; + auto op_write_buf = [&](uint8_t *src, size_t src_len) + { + if (lst.size() >= IOV_MAX) + return false; + if (from < src_len) + { + lst.push_back((iovec){ .iov_base = src+from, .iov_len = src_len-from }); + cl->write_op_pos += src_len-from; + from = 0; + } + else + from -= src_len; + return true; + }; + if (!op_write_headers(cl->write_op, op_write_buf)) + { + return; + } + // Operation data + if (op_has_data(cl->write_op)) + { + for (int i = 0; i < cl->write_op->iov.count; i++) + { + if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len)) + return; + } + } + cl->write_op = NULL; + cl->write_op_pos = 0; +} diff --git a/src/client/msgr_stop.cpp b/src/client/msgr_stop.cpp index 83afaf30..7cc77271 100644 --- a/src/client/msgr_stop.cpp +++ b/src/client/msgr_stop.cpp @@ -179,6 +179,13 @@ osd_client_t::~osd_client_t() } // Cancel outbound ops cancel_ops(); + for (osd_op_t *op: send_free_ops) + { + if (op) + { + delete op; + } + } for (osd_op_t *op: zc_free_list) { if (op)