diff --git a/src/client/cluster_client.cpp b/src/client/cluster_client.cpp index e7c54edb..45102594 100644 --- a/src/client/cluster_client.cpp +++ b/src/client/cluster_client.cpp @@ -27,7 +27,7 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd msgr.ringloop = ringloop; msgr.repeer_pgs = [this](osd_num_t peer_osd) { - if (msgr.osd_peer_fds.find(peer_osd) != msgr.osd_peer_fds.end()) + if (msgr.osd_peers.find(peer_osd) != msgr.osd_peers.end()) { // peer_osd just connected continue_ops(); @@ -47,8 +47,8 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd msgr.exec_op = [this](osd_op_t *op) { // Garbage in - fprintf(stderr, "Incoming garbage from peer %d\n", op->peer_fd); - msgr.stop_client(op->peer_fd); + fprintf(stderr, "Can't handle incoming operation from client %lu\n", op->client_id); + msgr.stop_client(op->client_id); delete op; }; msgr.parse_config(config); @@ -156,7 +156,7 @@ void cluster_client_t::continue_raw_ops(osd_num_t peer_osd) { auto op = it->second; op->op_type = OSD_OP_OUT; - op->peer_fd = msgr.osd_peer_fds.at(peer_osd); + op->client_id = msgr.osd_peers.at(peer_osd)->client_id; msgr.outbox_push(op); raw_ops.erase(it++); } @@ -871,13 +871,13 @@ void cluster_client_t::execute_cas(cluster_op_t *op) if (op->retval != expected && op->retval >= 0) op->retval = -EIO; op->retval = op->retval == -EPIPE ? -EINTR : op->retval; - auto peer_it = msgr.osd_peer_fds.find(op->parts[0].osd_num); + auto peer_it = msgr.osd_peers.find(op->parts[0].osd_num); if (op->retval != 0 || (op->flags & OP_IMMEDIATE_COMMIT)) { auto cb = std::move(op->callback); cb(op); } - else if (peer_it == msgr.osd_peer_fds.end()) + else if (peer_it == msgr.osd_peers.end()) { // Care must be taken to make sure that the client doesn't reconnect to the OSD // before executing the previously completed operation callback (!) @@ -888,10 +888,10 @@ void cluster_client_t::execute_cas(cluster_op_t *op) else { // CAS writes have a built-in sync - auto peer_fd = peer_it->second; + osd_client_t *cl = peer_it->second; *part = (osd_op_t){ .op_type = OSD_OP_OUT, - .peer_fd = peer_fd, + .client_id = cl->client_id, .req = { .hdr = { .magic = SECONDARY_OSD_OP_MAGIC, @@ -1004,11 +1004,11 @@ bool cluster_client_t::check_rw(cluster_op_t *op) void cluster_client_t::execute_raw(osd_num_t osd_num, osd_op_t *op) { - auto fd_it = msgr.osd_peer_fds.find(osd_num); - if (fd_it != msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(osd_num); + if (peer_it != msgr.osd_peers.end()) { op->op_type = OSD_OP_OUT; - op->peer_fd = fd_it->second; + op->client_id = peer_it->second->client_id; msgr.outbox_push(op); } else @@ -1401,10 +1401,10 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::functionosd_num = primary_osd; - auto peer_it = msgr.osd_peer_fds.find(primary_osd); - if (peer_it != msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(primary_osd); + if (peer_it != msgr.osd_peers.end()) { - int peer_fd = peer_it->second; + osd_client_t *cl = peer_it->second; part->flags |= PART_SENT|PART_VALID; op->inflight_count++; uint64_t pg_bitmap_size = (pool_cfg.data_block_size / pool_cfg.bitmap_granularity / 8) * ( @@ -1419,7 +1419,7 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::functionop = (osd_op_t){ .op_type = OSD_OP_OUT, - .peer_fd = peer_fd, + .client_id = cl->client_id, .req = { .rw = { .header = { .magic = SECONDARY_OSD_OP_MAGIC, @@ -1468,8 +1468,8 @@ int cluster_client_t::continue_sync(cluster_op_t *op) for (auto do_it = dirty_osds.begin(); do_it != dirty_osds.end(); ) { osd_num_t sync_osd = *do_it; - auto peer_it = msgr.osd_peer_fds.find(sync_osd); - if (peer_it == msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(sync_osd); + if (peer_it == msgr.osd_peers.end()) dirty_osds.erase(do_it++); else do_it++; @@ -1522,12 +1522,12 @@ resume_1: void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part) { - auto peer_fd = msgr.osd_peer_fds.at(part->osd_num); + osd_client_t *cl = msgr.osd_peers.at(part->osd_num); part->flags |= PART_SENT; op->inflight_count++; part->op = (osd_op_t){ .op_type = OSD_OP_OUT, - .peer_fd = peer_fd, + .client_id = cl->client_id, .req = { .hdr = { .magic = SECONDARY_OSD_OP_MAGIC, @@ -1567,10 +1567,10 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part) // Error priority: EIO > ENOSPC > ETIMEDOUT > EPIPE op->retval = part->op.reply.hdr.retval; } - int stop_fd = -1; + uint64_t stop_client_id = 0; if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC) { - stop_fd = part->op.peer_fd; + stop_client_id = part->op.client_id; if (op->retval != -EPIPE || log_level > 0) { fprintf( @@ -1597,9 +1597,9 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part) op->retry_after = op->retval != -EPIPE ? client_eio_retry_interval : client_retry_interval; } reset_retry_timer(op->retry_after); - if (stop_fd >= 0) + if (stop_client_id) { - msgr.stop_client(stop_fd); + msgr.stop_client(stop_client_id); } op->inflight_count--; if (op->inflight_count == 0 && !op->retry_after) diff --git a/src/client/cluster_client_list.cpp b/src/client/cluster_client_list.cpp index a6a73da4..d205b250 100644 --- a/src/client/cluster_client_list.cpp +++ b/src/client/cluster_client_list.cpp @@ -295,7 +295,7 @@ int cluster_client_t::start_pg_listing(inode_list_pg_t *pg) bool conn = true; for (osd_num_t peer_osd: all_peers) { - if (msgr.osd_peer_fds.find(peer_osd) == msgr.osd_peer_fds.end()) + if (msgr.osd_peers.find(peer_osd) == msgr.osd_peers.end()) { // Initiate connection if (st_cli.peer_states[peer_osd].is_null()) @@ -340,7 +340,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list) osd_op_t *op = new osd_op_t(); op->op_type = OSD_OP_OUT; // Already checked that it exists above, but anyway - op->peer_fd = msgr.osd_peer_fds.at(cur_list->osd_num); + op->client_id = msgr.osd_peers.at(cur_list->osd_num)->client_id; op->req = (osd_any_op_t){ .sec_list = { .header = { diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp index 20f04df8..d9ccc76f 100644 --- a/src/client/messenger.cpp +++ b/src/client/messenger.cpp @@ -187,7 +187,7 @@ void osd_messenger_t::init() { auto cl = cl_it->second; cl_it++; - auto peer_fd = cl->peer_fd; + auto client_id = cl->client_id; if (!cl->osd_num && !cl->in_osd_num || cl->peer_state != PEER_CONNECTED && cl->peer_state != PEER_RDMA) { // Do not run keepalive on regular clients @@ -199,10 +199,11 @@ void osd_messenger_t::init() if (!cl->ping_time_remaining) { // Ping timed out, stop the client - fprintf(stderr, "Ping timed out for OSD %ju (client %d), disconnecting peer\n", cl->in_osd_num ? cl->in_osd_num : cl->osd_num, cl->peer_fd); - stop_client(peer_fd, true); + fprintf(stderr, "Ping timed out for OSD %ju (client %ju), disconnecting peer\n", + cl->in_osd_num ? cl->in_osd_num : cl->osd_num, cl->client_id); + stop_client(cl->client_id); // Restart iterator because it may be invalidated - cl_it = clients.upper_bound(peer_fd); + cl_it = clients.upper_bound(client_id); } } else if (cl->idle_time_remaining > 0) @@ -213,7 +214,7 @@ void osd_messenger_t::init() // Connection is idle for , send ping osd_op_t *op = new osd_op_t(); op->op_type = OSD_OP_OUT; - op->peer_fd = cl->peer_fd; + op->client_id = cl->client_id; op->req = (osd_any_op_t){ .hdr = { .magic = SECONDARY_OSD_OP_MAGIC, @@ -222,28 +223,28 @@ void osd_messenger_t::init() }; op->callback = [this, cl](osd_op_t *op) { - auto cl_it = clients.find(op->peer_fd); - if (cl_it == clients.end() || cl_it->second != cl) + auto cl_it = clients.find(op->client_id); + if (cl_it == clients.end()) { // client is already dropped delete op; return; } - int fail_fd = (op->reply.hdr.retval != 0 ? op->peer_fd : -1); + uint64_t fail_client_id = (op->reply.hdr.retval != 0 ? op->client_id : 0); auto fail_osd_num = cl->in_osd_num ? cl->in_osd_num : cl->osd_num; cl->ping_time_remaining = 0; delete op; - if (fail_fd >= 0) + if (fail_client_id) { - fprintf(stderr, "Ping failed for OSD %ju (client %d), disconnecting peer\n", fail_osd_num, fail_fd); - stop_client(fail_fd, true); + fprintf(stderr, "Ping failed for OSD %ju (client %ju), disconnecting peer\n", fail_osd_num, fail_client_id); + stop_client(fail_client_id); } }; cl->ping_time_remaining = osd_ping_timeout; cl->idle_time_remaining = osd_idle_timeout; outbox_push(op); // Restart iterator because it may be invalidated - cl_it = clients.upper_bound(peer_fd); + cl_it = clients.upper_bound(client_id); } } else @@ -263,7 +264,7 @@ osd_messenger_t::~osd_messenger_t() } while (clients.size() > 0) { - stop_client(clients.begin()->first, true, true); + stop_client(clients.begin()->first, true); } if (iothreads.size()) { @@ -440,7 +441,7 @@ void osd_messenger_t::try_connect_peer(uint64_t peer_osd) { return; } - if (osd_peer_fds.find(peer_osd) != osd_peer_fds.end()) + if (osd_peers.find(peer_osd) != osd_peers.end()) { wanted_peers.erase(peer_osd); return; @@ -467,20 +468,20 @@ void osd_messenger_t::try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_ #ifdef WITH_RDMACM if (disable_tcp) { - on_connect_peer(peer_osd, -EINVAL); + on_connect_peer(peer_osd, -EINVAL, 0); return; } #endif struct sockaddr_storage addr; if (!string_to_addr(peer_host, 0, peer_port, &addr)) { - on_connect_peer(peer_osd, -EINVAL); + on_connect_peer(peer_osd, -EINVAL, 0); return; } int peer_fd = socket(addr.ss_family, SOCK_STREAM, 0); if (peer_fd < 0) { - on_connect_peer(peer_osd, -errno); + on_connect_peer(peer_osd, -errno, 0); return; } fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK); @@ -488,21 +489,25 @@ void osd_messenger_t::try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_ if (r < 0 && errno != EINPROGRESS) { close(peer_fd); - on_connect_peer(peer_osd, -errno); + on_connect_peer(peer_osd, -errno, 0); return; } - clients[peer_fd] = new osd_client_t(); + const uint64_t client_id = next_client_id++; + osd_client_t *cl = new osd_client_t(); if (log_level > 0) { - fprintf(stderr, "Connecting to OSD %ju at %s:%d (client %d)\n", peer_osd, peer_host, peer_port, peer_fd); + fprintf(stderr, "Connecting to OSD %ju at %s:%d (client %ju, FD %d)\n", peer_osd, peer_host, peer_port, client_id, peer_fd); } - clients[peer_fd]->peer_addr = addr; - clients[peer_fd]->peer_port = peer_port; - clients[peer_fd]->peer_fd = peer_fd; - clients[peer_fd]->peer_state = PEER_CONNECTING; - clients[peer_fd]->connect_timeout_id = -1; - clients[peer_fd]->osd_num = peer_osd; - clients[peer_fd]->in_buf = malloc_or_die(receive_buffer_size); + cl->client_id = client_id; + cl->peer_addr = addr; + cl->peer_port = peer_port; + cl->peer_fd = peer_fd; + cl->peer_state = PEER_CONNECTING; + cl->connect_timeout_id = -1; + cl->osd_num = peer_osd; + cl->in_buf = 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) { // Either OUT (connected) or HUP @@ -510,11 +515,11 @@ void osd_messenger_t::try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_ }); if (peer_connect_timeout > 0) { - clients[peer_fd]->connect_timeout_id = tfd->set_timer(1000*peer_connect_timeout, false, [this, peer_fd](int timer_id) + cl->connect_timeout_id = tfd->set_timer(1000*peer_connect_timeout, false, [this, client_id](int timer_id) { - osd_num_t peer_osd = clients.at(peer_fd)->osd_num; - stop_client(peer_fd, true); - on_connect_peer(peer_osd, -EPIPE); + osd_num_t peer_osd = clients.at(client_id)->osd_num; + stop_client(client_id); + on_connect_peer(peer_osd, -EPIPE, 0); return; }); } @@ -522,7 +527,7 @@ void osd_messenger_t::try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_ void osd_messenger_t::handle_connect_epoll(int peer_fd) { - auto cl = clients[peer_fd]; + auto cl = clients_by_fd.at(peer_fd); if (cl->connect_timeout_id >= 0) { tfd->clear_timer(cl->connect_timeout_id); @@ -537,8 +542,8 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd) } if (result != 0) { - stop_client(peer_fd, true); - on_connect_peer(peer_osd, -result); + stop_client(cl->client_id); + on_connect_peer(peer_osd, -result, 0); return; } int one = 1; @@ -555,23 +560,23 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd) void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events) { // Mark client as ready (i.e. some data is available) + auto cl = clients_by_fd.at(peer_fd); if (epoll_events & EPOLLRDHUP) { // Stop client if (log_level > 0) { - fprintf(stderr, "[OSD %ju] client %d disconnected\n", this->osd_num, peer_fd); + fprintf(stderr, "[OSD %ju] client %ju disconnected\n", this->osd_num, cl->client_id); } - stop_client(peer_fd, true); + stop_client(cl->client_id); } else if (epoll_events & EPOLLIN) { // Mark client as ready (i.e. some data is available) - auto cl = clients[peer_fd]; cl->read_ready++; if (cl->read_ready == 1) { - read_ready_clients.push_back(cl->peer_fd); + read_ready_clients.push_back(cl->client_id); if (ringloop) ringloop->wakeup(); else @@ -580,13 +585,13 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events) } } -void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int peer_fd) +void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int errcode, uint64_t client_id) { auto & wp = wanted_peers.at(peer_osd); wp.connecting = false; - if (peer_fd < 0) + if (errcode < 0) { - fprintf(stderr, "Failed to connect to peer OSD %ju address %s port %d: %s\n", peer_osd, wp.cur_addr.c_str(), wp.cur_port, strerror(-peer_fd)); + fprintf(stderr, "Failed to connect to peer OSD %ju address %s port %d: %s\n", peer_osd, wp.cur_addr.c_str(), wp.cur_port, strerror(-errcode)); if (wp.address_changed) { wp.address_changed = false; @@ -613,7 +618,7 @@ void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int peer_fd) } if (log_level > 0) { - fprintf(stderr, "[OSD %ju] Connected with peer OSD %ju (client %d)\n", osd_num, peer_osd, peer_fd); + fprintf(stderr, "[OSD %ju] Connected with peer OSD %ju (client %ju)\n", osd_num, peer_osd, client_id); } wanted_peers.erase(peer_osd); repeer_pgs(peer_osd); @@ -623,7 +628,7 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) { osd_op_t *op = new osd_op_t(); op->op_type = OSD_OP_OUT; - op->peer_fd = cl->peer_fd; + op->client_id = cl->client_id; op->req = (osd_any_op_t){ .show_conf = { .header = { @@ -647,7 +652,7 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) if (!selected_ctx) { if (log_level > 0) - fprintf(stderr, "No RDMA context for OSD %ju connection (peer %d), using only TCP\n", cl->osd_num, cl->peer_fd); + fprintf(stderr, "No RDMA context for OSD %ju connection (client %ju), using only TCP\n", cl->osd_num, cl->client_id); } else { @@ -708,8 +713,8 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) if (err) { osd_num_t peer_osd = cl->osd_num; - stop_client(op->peer_fd); - on_connect_peer(peer_osd, -EINVAL); + stop_client(op->client_id); + on_connect_peer(peer_osd, -EINVAL, 0); delete op; return; } @@ -744,8 +749,8 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) } } #endif - osd_peer_fds[cl->osd_num] = cl->peer_fd; - on_connect_peer(cl->osd_num, cl->peer_fd); + osd_peers[cl->osd_num] = cl; + on_connect_peer(cl->osd_num, 0, cl->client_id); delete op; }; outbox_push(op); @@ -760,13 +765,16 @@ void osd_messenger_t::accept_connections(int listen_fd) while ((peer_fd = accept(listen_fd, (sockaddr*)&addr, &peer_addr_size)) >= 0) { assert(peer_fd != 0); - fprintf(stderr, "[OSD %ju] new client %d: connection from %s\n", this->osd_num, peer_fd, + const uint64_t client_id = next_client_id++; + fprintf(stderr, "[OSD %ju] new client %ju (FD %d): connection from %s\n", this->osd_num, client_id, peer_fd, addr_to_string(addr).c_str()); fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK); int one = 1; setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one)); auto cl = new osd_client_t(); - clients[peer_fd] = cl; + cl->client_id = client_id; + clients[cl->client_id] = cl; + clients_by_fd[peer_fd] = cl; cl->is_incoming = true; cl->peer_addr = addr; cl->peer_addr = addr; diff --git a/src/client/messenger.h b/src/client/messenger.h index aa7b0d52..83e07ce6 100644 --- a/src/client/messenger.h +++ b/src/client/messenger.h @@ -50,6 +50,7 @@ struct msgr_rdma_context_t; struct osd_client_t { + uint64_t client_id = 0; int refs = 0; sockaddr_storage peer_addr = {}; @@ -206,8 +207,8 @@ protected: #endif std::vector iothreads; - std::vector read_ready_clients; - std::vector write_ready_clients; + std::vector read_ready_clients; + std::vector write_ready_clients; // We don't use ringloop->set_immediate here because we may have no ringloop in client :) std::deque set_immediate_ops; @@ -216,10 +217,12 @@ public: ring_loop_t *ringloop = NULL; bool has_sendmsg_zc = false; // osd_num_t is only for logging and asserts + uint64_t next_client_id = 1; osd_num_t osd_num; - std::map clients; + std::map clients; + std::map osd_peers; + std::map clients_by_fd; std::map wanted_peers; - std::map osd_peer_fds; std::vector osd_networks; std::vector osd_network_masks; std::vector osd_cluster_networks; @@ -232,7 +235,7 @@ public: void init(); void parse_config(const json11::Json & config); void connect_peer(uint64_t osd_num, json11::Json peer_state); - void stop_client(int peer_fd, bool force = false, bool force_delete = false); + void stop_client(uint64_t client_id, bool force_delete = false); void destroy_client(osd_client_t *cl); void outbox_push(osd_op_t *cur_op); std::function exec_op; @@ -252,7 +255,7 @@ public: #ifdef WITH_RDMA bool is_rdma_enabled(); - bool connect_rdma(int peer_fd, std::string rdma_address, uint64_t client_max_msg); + bool connect_rdma(uint64_t client_id, std::string rdma_address, uint64_t client_max_msg); #endif #ifdef WITH_RDMACM bool is_use_rdmacm(); @@ -268,7 +271,7 @@ protected: void try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_host, int peer_port); void handle_peer_epoll(int peer_fd, int epoll_events); void handle_connect_epoll(int peer_fd); - void on_connect_peer(osd_num_t peer_osd, int peer_fd); + void on_connect_peer(osd_num_t peer_osd, int errcode, uint64_t client_id); void check_peer_config(osd_client_t *cl); void cancel_osd_ops(osd_client_t *cl); void cancel_op(osd_op_t *op); @@ -283,7 +286,6 @@ protected: bool handle_reply_hdr(osd_client_t *cl); void handle_reply_ready(osd_op_t *op); void handle_immediate_ops(); - void clear_immediate_ops(int peer_fd); #ifdef WITH_RDMA void try_send_rdma(osd_client_t *cl); diff --git a/src/client/msgr_op.h b/src/client/msgr_op.h index 61fd2429..020b9530 100644 --- a/src/client/msgr_op.h +++ b/src/client/msgr_op.h @@ -156,7 +156,7 @@ struct __attribute__((visibility("default"))) osd_op_t { timespec tv_begin = { 0 }, tv_end = { 0 }; uint64_t op_type = OSD_OP_IN; - int peer_fd; + uint64_t client_id = 0; osd_any_op_t req; osd_any_reply_t reply; blockstore_op_t *bs_op = NULL; diff --git a/src/client/msgr_rdma.cpp b/src/client/msgr_rdma.cpp index 141b3a2d..bff210fe 100644 --- a/src/client/msgr_rdma.cpp +++ b/src/client/msgr_rdma.cpp @@ -493,7 +493,7 @@ int msgr_rdma_connection_t::connect(msgr_rdma_address_t *dest) return 0; } -bool osd_messenger_t::connect_rdma(int peer_fd, std::string rdma_address, uint64_t client_max_msg) +bool osd_messenger_t::connect_rdma(uint64_t client_id, std::string rdma_address, uint64_t client_max_msg) { // Try to connect to the peer using RDMA msgr_rdma_address_t addr; @@ -503,12 +503,12 @@ bool osd_messenger_t::connect_rdma(int peer_fd, std::string rdma_address, uint64 { client_max_msg = rdma_max_msg; } - auto cl = clients.at(peer_fd); + auto cl = clients.at(client_id); msgr_rdma_context_t *selected_ctx = choose_rdma_context(cl); if (!selected_ctx) { if (log_level > 0) - fprintf(stderr, "No RDMA context for peer %d, using only TCP\n", cl->peer_fd); + fprintf(stderr, "No RDMA context for peer %ju, using only TCP\n", client_id); return false; } msgr_rdma_connection_t *rdma_conn = msgr_rdma_connection_t::create(selected_ctx, rdma_max_send, rdma_max_recv, rdma_max_sge, client_max_msg); @@ -519,14 +519,13 @@ bool osd_messenger_t::connect_rdma(int peer_fd, std::string rdma_address, uint64 { delete rdma_conn; fprintf( - stderr, "Failed to connect RDMA queue pair to %s (client %d)\n", - addr.to_string().c_str(), peer_fd + stderr, "Failed to connect RDMA queue pair to %s (client %ju)\n", + addr.to_string().c_str(), client_id ); } else { // Remember connection, but switch to RDMA only after sending the configuration response - auto cl = clients.at(peer_fd); cl->rdma_conn = rdma_conn; cl->peer_state = PEER_RDMA_CONNECTING; return true; @@ -540,7 +539,7 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge) { ibv_send_wr *bad_wr = NULL; ibv_send_wr wr = { - .wr_id = (uint64_t)(cl->peer_fd*2+1), + .wr_id = cl->client_id, .sg_list = sge, .num_sge = op_sge, .opcode = IBV_WR_SEND, @@ -631,7 +630,7 @@ static void try_recv_rdma_wr(osd_client_t *cl, void *buf) }; ibv_recv_wr *bad_wr = NULL; ibv_recv_wr wr = { - .wr_id = (uint64_t)(cl->peer_fd*2), + .wr_id = cl->client_id, .sg_list = &sge, .num_sge = 1, }; @@ -688,8 +687,8 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) event_count = ibv_poll_cq(rdma_context->cq, RDMA_EVENTS_AT_ONCE, wc); for (int i = 0; i < event_count; i++) { - int client_id = wc[i].wr_id >> 1; - bool is_send = wc[i].wr_id & 1; + uint64_t client_id = wc[i].wr_id; + bool is_send = wc[i].opcode == IBV_WC_SEND; auto cl_it = clients.find(client_id); if (cl_it == clients.end()) { @@ -703,7 +702,7 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) auto rc = cl->rdma_conn; if (wc[i].status != IBV_WC_SUCCESS) { - fprintf(stderr, "RDMA work request failed for client %d", client_id); + fprintf(stderr, "RDMA work request failed for client %ju", client_id); if (cl->osd_num) { fprintf(stderr, " (OSD %ju)", cl->osd_num); diff --git a/src/client/msgr_rdmacm.cpp b/src/client/msgr_rdmacm.cpp index b539603c..4affe91d 100644 --- a/src/client/msgr_rdmacm.cpp +++ b/src/client/msgr_rdmacm.cpp @@ -11,7 +11,7 @@ struct rdmacm_connecting_t { rdma_cm_id *cmid = NULL; - int peer_fd = -1; + uint64_t client_id = 0; osd_num_t peer_osd = 0; std::string addr; sockaddr_storage parsed_addr = {}; @@ -117,9 +117,9 @@ void osd_messenger_t::handle_rdmacm_events() auto cli_it = rdmacm_connections.find(ev->id); if (cli_it != rdmacm_connections.end()) { - fprintf(stderr, "Received %s event for peer %d, closing connection\n", - event_type_name, cli_it->second->peer_fd); - stop_client(cli_it->second->peer_fd); + fprintf(stderr, "Received %s event for client %ju, closing connection\n", + event_type_name, cli_it->second->client_id); + stop_client(cli_it->second->client_id); } else if (rdmacm_connecting.find(ev->id) != rdmacm_connecting.end()) { @@ -265,14 +265,6 @@ msgr_rdma_context_t* osd_messenger_t::rdmacm_create_qp(rdma_cm_id *cmid) void osd_messenger_t::rdmacm_accept(rdma_cm_event *ev) { - // Make a fake FD (FIXME: do not use FDs for identifying clients!) - int fake_fd = socket(AF_INET, SOCK_STREAM, 0); - if (fake_fd < 0) - { - fprintf(stderr, "Failed to allocate a fake socket for RDMA-CM client: %s (code %d)\n", strerror(errno), errno); - rdma_destroy_id(ev->id); - return; - } auto rdma_context = rdmacm_create_qp(ev->id); if (!rdma_context) { @@ -297,12 +289,12 @@ void osd_messenger_t::rdmacm_accept(rdma_cm_event *ev) // Wait for RDMA_CM_ESTABLISHED, and enable the connection only after it auto conn = new rdmacm_connecting_t; conn->cmid = ev->id; - conn->peer_fd = fake_fd; + conn->client_id = next_client_id++; conn->parsed_addr = *(sockaddr_storage*)rdma_get_peer_addr(ev->id); conn->rdma_context = rdma_context; rdmacm_set_conn_timeout(conn); rdmacm_connecting[ev->id] = conn; - fprintf(stderr, "[OSD %ju] new client %d: connection from %s via RDMA-CM\n", this->osd_num, conn->peer_fd, + fprintf(stderr, "[OSD %ju] new client %ju: connection from %s via RDMA-CM\n", this->osd_num, conn->client_id, addr_to_string(conn->parsed_addr).c_str()); } @@ -332,8 +324,6 @@ void osd_messenger_t::rdmacm_on_connect_peer_error(rdma_cm_id *cmid, int res) auto peer_osd = conn->peer_osd; if (conn->timeout_id >= 0) tfd->clear_timer(conn->timeout_id); - if (conn->peer_fd >= 0) - close(conn->peer_fd); if (conn->rdma_context) conn->rdma_context->reserve_cqe(-rdma_max_send-rdma_max_recv); if (conn->cmid) @@ -354,7 +344,7 @@ void osd_messenger_t::rdmacm_on_connect_peer_error(rdma_cm_id *cmid, int res) else { // TCP is disabled - on_connect_peer(peer_osd, res == 0 ? -EINVAL : (res > 0 ? -res : res)); + on_connect_peer(peer_osd, res == 0 ? -EINVAL : (res > 0 ? -res : res), 0); } } } @@ -365,7 +355,7 @@ void osd_messenger_t::rdmacm_try_connect_peer(uint64_t peer_osd, const std::stri if (!string_to_addr(addr, false, rdmacm_port, &sa)) { fprintf(stderr, "Address %s is invalid\n", addr.c_str()); - on_connect_peer(peer_osd, -EINVAL); + on_connect_peer(peer_osd, -EINVAL, 0); return; } rdma_cm_id *cmid = NULL; @@ -376,17 +366,7 @@ void osd_messenger_t::rdmacm_try_connect_peer(uint64_t peer_osd, const std::stri if (!disable_tcp) try_connect_peer_tcp(peer_osd, addr.c_str(), fallback_tcp_port); else - on_connect_peer(peer_osd, res); - return; - } - // Make a fake FD (FIXME: do not use FDs for identifying clients!) - int fake_fd = socket(AF_INET, SOCK_STREAM, 0); - if (fake_fd < 0) - { - int res = -errno; - rdma_destroy_id(cmid); - // Can't create socket, pointless to try TCP - on_connect_peer(peer_osd, res); + on_connect_peer(peer_osd, res, 0); return; } if (log_level > 0) @@ -394,7 +374,7 @@ void osd_messenger_t::rdmacm_try_connect_peer(uint64_t peer_osd, const std::stri auto conn = new rdmacm_connecting_t; rdmacm_connecting[cmid] = conn; conn->cmid = cmid; - conn->peer_fd = fake_fd; + conn->client_id = next_client_id++; conn->peer_osd = peer_osd; conn->addr = addr; conn->parsed_addr = sa; @@ -511,13 +491,13 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev) auto cl = new osd_client_t(); cl->peer_addr = conn->parsed_addr; cl->peer_port = conn->rdmacm_port; - cl->peer_fd = conn->peer_fd; + cl->client_id = conn->client_id; 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->rdma_conn = rc; - clients[conn->peer_fd] = cl; + clients[conn->client_id] = cl; if (conn->timeout_id >= 0) { tfd->clear_timer(conn->timeout_id); diff --git a/src/client/msgr_receive.cpp b/src/client/msgr_receive.cpp index bd45c4de..2015c044 100644 --- a/src/client/msgr_receive.cpp +++ b/src/client/msgr_receive.cpp @@ -7,8 +7,8 @@ void osd_messenger_t::read_requests() { for (int i = 0; i < read_ready_clients.size(); i++) { - int peer_fd = read_ready_clients[i]; - auto cl_it = clients.find(peer_fd); + uint64_t client_id = read_ready_clients[i]; + auto cl_it = clients.find(client_id); if (cl_it == clients.end() || !cl_it->second || cl_it->second->read_msg.msg_iovlen || cl_it->second->peer_state != PEER_CONNECTED) { @@ -32,7 +32,7 @@ void osd_messenger_t::read_requests() cl->refs++; if (ringloop && !use_sync_send_recv) { - auto iothread = iothreads.size() ? iothreads[peer_fd % iothreads.size()] : NULL; + auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL; io_uring_sqe sqe_local; ring_data_t data_local; io_uring_sqe* sqe = (iothread ? &sqe_local : ringloop->get_sqe()); @@ -50,7 +50,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, peer_fd, &cl->read_msg, 0); + io_uring_prep_recvmsg(sqe, cl->peer_fd, &cl->read_msg, 0); if (iothread) { iothread->add_sqe(sqe_local); @@ -58,7 +58,7 @@ void osd_messenger_t::read_requests() } else { - int result = recvmsg(peer_fd, &cl->read_msg, 0); + int result = recvmsg(cl->peer_fd, &cl->read_msg, 0); if (result < 0) { result = -errno; @@ -92,20 +92,20 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl) // this is a client socket, so don't panic on error. just disconnect it if (result != 0) { - fprintf(stderr, "Client %d socket read error: %d (%s). Disconnecting client\n", cl->peer_fd, -result, strerror(-result)); + fprintf(stderr, "Client %ju socket read error: %d (%s). Disconnecting client\n", cl->client_id, -result, strerror(-result)); } - stop_client(cl->peer_fd); + stop_client(cl->client_id); return false; } if (result == -EAGAIN || result == -EINTR || result < cl->read_iov.iov_len) { cl->read_ready--; if (cl->read_ready > 0) - read_ready_clients.push_back(cl->peer_fd); + read_ready_clients.push_back(cl->client_id); } else { - read_ready_clients.push_back(cl->peer_fd); + read_ready_clients.push_back(cl->client_id); } if (result > 0) { @@ -140,26 +140,6 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl) return ret; } -void osd_messenger_t::clear_immediate_ops(int peer_fd) -{ - size_t i = 0, j = 0; - while (i < set_immediate_ops.size()) - { - if (set_immediate_ops[i]->peer_fd == peer_fd && set_immediate_ops[i]->op_type == OSD_OP_IN) - { - delete set_immediate_ops[i]; - } - else - { - if (i != j) - set_immediate_ops[j] = set_immediate_ops[i]; - j++; - } - i++; - } - set_immediate_ops.resize(j); -} - void osd_messenger_t::handle_immediate_ops() { while (set_immediate_ops.size()) @@ -168,7 +148,11 @@ void osd_messenger_t::handle_immediate_ops() set_immediate_ops.pop_front(); if (op->op_type == OSD_OP_IN) { - exec_op(op); + auto cl_it = clients.find(op->client_id); + if (cl_it != clients.end() && cl_it->second->peer_state != PEER_STOPPED) + exec_op(op); + else + delete op; } else { @@ -186,7 +170,7 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, void *curbuf, int rem if (!cl->read_op) { cl->read_op = new osd_op_t; - cl->read_op->peer_fd = cl->peer_fd; + 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; @@ -240,8 +224,8 @@ bool osd_messenger_t::handle_finished_read(osd_client_t *cl) { 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); - stop_client(cl->peer_fd); + 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++; @@ -250,8 +234,8 @@ bool osd_messenger_t::handle_finished_read(osd_client_t *cl) } else { - fprintf(stderr, "Received garbage: magic=%jx id=%ju opcode=%jx from %d\n", cl->read_op->req.hdr.magic, cl->read_op->req.hdr.id, cl->read_op->req.hdr.opcode, cl->peer_fd); - stop_client(cl->peer_fd); + 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; } } @@ -367,8 +351,8 @@ bool osd_messenger_t::handle_reply_hdr(osd_client_t *cl) if (req_it == cl->sent_ops.end()) { // Command out of sync. Drop connection - fprintf(stderr, "Client %d command out of sync: id %ju\n", cl->peer_fd, cl->read_op->req.hdr.id); - stop_client(cl->peer_fd); + 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; @@ -382,10 +366,10 @@ bool osd_messenger_t::handle_reply_hdr(osd_client_t *cl) if (op->reply.hdr.retval >= 0 && (op->reply.hdr.retval != expected_size || bmp_len > op->bitmap_len)) { // Check reply length to not overflow the buffer - fprintf(stderr, "Client %d read reply of different length: expected %u+%u, got %jd+%u\n", - cl->peer_fd, expected_size, op->bitmap_len, op->reply.hdr.retval, bmp_len); + 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->peer_fd); + stop_client(cl->client_id); return false; } if (bmp_len > 0) diff --git a/src/client/msgr_send.cpp b/src/client/msgr_send.cpp index 07ed11e2..ae281003 100644 --- a/src/client/msgr_send.cpp +++ b/src/client/msgr_send.cpp @@ -9,8 +9,14 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) { - assert(cur_op->peer_fd); - osd_client_t *cl = clients.at(cur_op->peer_fd); + assert(cur_op->client_id); + auto cl_it = clients.find(cur_op->client_id); + if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED) + { + delete cur_op; + return; + } + osd_client_t *cl = cl_it->second; if (cur_op->op_type == OSD_OP_OUT) { clock_gettime(CLOCK_REALTIME, &cur_op->tv_begin); @@ -18,8 +24,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) } else { - // Check that operation actually belongs to this client - // FIXME: Review if this is still needed + // Remove the operation from received op list bool found = false; for (auto it = cl->received_ops.begin(); it != cl->received_ops.end(); it++) { @@ -30,11 +35,8 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) break; } } - if (!found) - { - delete cur_op; - return; - } + // 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; @@ -126,7 +128,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) if ((cl->write_msg.msg_iovlen > 0 || !try_send(cl)) && (cl->write_state == 0)) { cl->write_state = CL_WRITE_READY; - write_ready_clients.push_back(cur_op->peer_fd); + write_ready_clients.push_back(cur_op->client_id); } ringloop->wakeup(); } @@ -183,15 +185,14 @@ void osd_messenger_t::measure_exec(osd_op_t *cur_op) bool osd_messenger_t::try_send(osd_client_t *cl) { - int peer_fd = cl->peer_fd; - if (!cl->send_list.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED) + if (!cl->send_list.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED || cl->peer_fd < 0) { return true; } assert(cl->peer_state != PEER_RDMA); if (ringloop && !use_sync_send_recv) { - auto iothread = iothreads.size() ? iothreads[peer_fd % iothreads.size()] : NULL; + auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL; io_uring_sqe sqe_local; ring_data_t data_local; io_uring_sqe* sqe = (iothread ? &sqe_local : ringloop->get_sqe()); @@ -218,11 +219,11 @@ bool osd_messenger_t::try_send(osd_client_t *cl) } if (use_zc) { - io_uring_prep_sendmsg_zc(sqe, peer_fd, &cl->write_msg, MSG_WAITALL); + io_uring_prep_sendmsg_zc(sqe, cl->peer_fd, &cl->write_msg, MSG_WAITALL); } else { - io_uring_prep_sendmsg(sqe, peer_fd, &cl->write_msg, MSG_WAITALL); + io_uring_prep_sendmsg(sqe, cl->peer_fd, &cl->write_msg, MSG_WAITALL); } if (iothread) { @@ -234,7 +235,7 @@ bool osd_messenger_t::try_send(osd_client_t *cl) 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++; - int result = sendmsg(peer_fd, &cl->write_msg, MSG_NOSIGNAL); + int result = sendmsg(cl->peer_fd, &cl->write_msg, MSG_NOSIGNAL); if (result < 0) { result = -errno; @@ -249,8 +250,8 @@ void osd_messenger_t::send_replies() { for (int i = 0; i < write_ready_clients.size(); i++) { - int peer_fd = write_ready_clients[i]; - auto cl_it = clients.find(peer_fd); + uint64_t client_id = write_ready_clients[i]; + auto cl_it = clients.find(client_id); if (cl_it != clients.end() && cl_it->second->peer_state != PEER_RDMA && !try_send(cl_it->second)) { write_ready_clients.erase(write_ready_clients.begin(), write_ready_clients.begin() + i); @@ -281,8 +282,8 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t if (result < 0 && result != -EAGAIN && result != -EINTR) { // this is a client socket, so don't panic. just disconnect it - fprintf(stderr, "Client %d socket write error: %d (%s). Disconnecting client\n", cl->peer_fd, -result, strerror(-result)); - stop_client(cl->peer_fd); + fprintf(stderr, "Client %ju socket write error: %d (%s). Disconnecting client\n", cl->client_id, -result, strerror(-result)); + stop_client(cl->client_id); return; } if (result >= 0) @@ -326,9 +327,9 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t int expected = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX; if (done != expected) { - fprintf(stderr, "Client %d socket write error: expected to send " - "%d iovecs with MSG_WAITALL but sent %d. Disconnecting client\n", cl->peer_fd, expected, done); - stop_client(cl->peer_fd); + 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 @@ -352,7 +353,7 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t // FIXME: Ignore pings during RDMA state transition if (log_level > 0) { - fprintf(stderr, "Successfully connected with client %d using RDMA\n", cl->peer_fd); + fprintf(stderr, "Successfully connected with client %ju using RDMA\n", cl->client_id); } cl->peer_state = PEER_RDMA; // Add the initial receive request @@ -362,6 +363,6 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t } if (cl->write_state != 0) { - write_ready_clients.push_back(cl->peer_fd); + write_ready_clients.push_back(cl->client_id); } } diff --git a/src/client/msgr_stop.cpp b/src/client/msgr_stop.cpp index e847c34a..97f1a101 100644 --- a/src/client/msgr_stop.cpp +++ b/src/client/msgr_stop.cpp @@ -47,20 +47,15 @@ void osd_op_t::cancel() // the flag should be used in the destructor. // why? - because yes, we could close the FD first and let it fail all requests in the event loop, // but in that case it can be quickly reopened and we can get old failed responses for the new FD. -void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete) +void osd_messenger_t::stop_client(uint64_t client_id, bool force_delete) { - assert(peer_fd != 0); - auto it = clients.find(peer_fd); - if (it == clients.end()) + auto it = clients.find(client_id); + if (!client_id || it == clients.end()) { return; } osd_client_t *cl = it->second; - // FIXME "force" flag is required because otherwise a first failed operation - // may stop the client, make it start reconnecting, and then another failed - // operation may stop it again. The right fix would be to introduce unique peer ID - // and not use FDs for that. - if (cl->peer_state == PEER_CONNECTING && !force || cl->peer_state == PEER_STOPPED) + if (cl->peer_state == PEER_STOPPED) { if (force_delete) { @@ -68,21 +63,20 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete) } return; } - clear_immediate_ops(peer_fd); cl->received_ops.clear(); if (log_level > 0) { if (cl->osd_num) { - fprintf(stderr, "[OSD %ju] Stopping client %d (OSD peer %ju)\n", osd_num, peer_fd, cl->osd_num); + fprintf(stderr, "[OSD %ju] Stopping client %ju (OSD peer %ju)\n", osd_num, client_id, cl->osd_num); } else if (cl->in_osd_num) { - fprintf(stderr, "[OSD %ju] Stopping client %d (incoming OSD peer %ju)\n", osd_num, peer_fd, cl->in_osd_num); + fprintf(stderr, "[OSD %ju] Stopping client %ju (incoming OSD peer %ju)\n", osd_num, client_id, cl->in_osd_num); } else { - fprintf(stderr, "[OSD %ju] Stopping client %d (regular client)\n", osd_num, peer_fd); + fprintf(stderr, "[OSD %ju] Stopping client %ju (regular client)\n", osd_num, client_id); } } // First set state to STOPPED so another stop_client() call doesn't try to free it again @@ -91,11 +85,11 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete) cl->peer_state = PEER_STOPPED; if (cl->osd_num) { - auto osd_it = osd_peer_fds.find(cl->osd_num); - if (osd_it != osd_peer_fds.end() && osd_it->second == cl->peer_fd) + auto osd_it = osd_peers.find(cl->osd_num); + if (osd_it != osd_peers.end() && osd_it->second == cl) { // ...and forget OSD peer - osd_peer_fds.erase(osd_it); + osd_peers.erase(osd_it); } } #ifdef WITH_RDMA @@ -128,6 +122,14 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete) // so do not repeer on it. repeer_pgs(cl->osd_num); } + if (cl->peer_fd >= 0) + { + int r = shutdown(cl->peer_fd, SHUT_RDWR); + if (r != 0 && errno != ENOTCONN) + { + fprintf(stderr, "[OSD %ju] failed to shutdown a socket: %s (code %d)\n", osd_num, strerror(errno), errno); + } + } cl->refs--; if (cl->refs <= 0 || force_delete) { @@ -138,30 +140,30 @@ void osd_messenger_t::stop_client(int peer_fd, bool force, bool force_delete) void osd_messenger_t::destroy_client(osd_client_t *cl) { // Find the item again because it can be invalidated at this point - auto it = clients.find(cl->peer_fd); - if (it != clients.end()) + clients.erase(cl->client_id); + if (cl->peer_fd >= 0) { - clients.erase(it); - } #ifndef __MOCK__ - tfd->set_fd_handler(cl->peer_fd, false, NULL); - for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++) - { - if (*rit == cl->peer_fd) - { - read_ready_clients.erase(rit); - break; - } - } - for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++) - { - if (*wit == cl->peer_fd) - { - write_ready_clients.erase(wit); - break; - } - } + tfd->set_fd_handler(cl->peer_fd, false, NULL); #endif + for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++) + { + if (*rit == cl->client_id) + { + read_ready_clients.erase(rit); + break; + } + } + for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++) + { + if (*wit == cl->client_id) + { + write_ready_clients.erase(wit); + break; + } + } + clients_by_fd.erase(cl->peer_fd); + } delete cl; } diff --git a/src/osd/osd.cpp b/src/osd/osd.cpp index 7c0e9727..2ba3e3b4 100644 --- a/src/osd/osd.cpp +++ b/src/osd/osd.cpp @@ -616,11 +616,11 @@ void osd_t::print_slow() bufprintf("[OSD %ju] Slow op %jx", osd_num, (uint64_t)op); if (kv.second->osd_num) { - bufprintf(" from peer OSD %ju (client %d)", kv.second->osd_num, kv.second->peer_fd); + bufprintf(" from peer OSD %ju (client %ju)", kv.second->osd_num, kv.second->client_id); } else { - bufprintf(" from client %d", kv.second->peer_fd); + bufprintf(" from client %ju", kv.second->client_id); } bufprintf(": %s id=%ju", osd_op_names[op->req.hdr.opcode], op->req.hdr.id); if (op->req.hdr.opcode == OSD_OP_SEC_READ || op->req.hdr.opcode == OSD_OP_SEC_WRITE || diff --git a/src/osd/osd_cluster.cpp b/src/osd/osd_cluster.cpp index 1097ae2b..cf21f8b8 100644 --- a/src/osd/osd_cluster.cpp +++ b/src/osd/osd_cluster.cpp @@ -991,7 +991,7 @@ void osd_t::apply_pg_config() // Add peers for (auto pg_osd: all_peers) { - if (pg_osd != this->osd_num && msgr.osd_peer_fds.find(pg_osd) == msgr.osd_peer_fds.end()) + if (pg_osd != this->osd_num && msgr.osd_peers.find(pg_osd) == msgr.osd_peers.end()) { msgr.connect_peer(pg_osd, st_cli.peer_states[pg_osd]); } diff --git a/src/osd/osd_flush.cpp b/src/osd/osd_flush.cpp index f1ae4353..319667d5 100644 --- a/src/osd/osd_flush.cpp +++ b/src/osd/osd_flush.cpp @@ -4,6 +4,7 @@ #include "osd.h" #define FLUSH_BATCH 512 +#define SELF_CLIENT 0 void osd_t::submit_pg_flush_ops(pg_t & pg) { @@ -91,11 +92,11 @@ void osd_t::handle_flush_op(bool rollback, pool_id_t pool_id, pg_num_t pg_num, p else { printf("Error while doing flush on OSD %ju: %d (%s)\n", osd_num, retval, strerror(-retval)); - auto fd_it = msgr.osd_peer_fds.find(peer_osd); - if (fd_it != msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(peer_osd); + if (peer_it != msgr.osd_peers.end()) { // Will repeer/stop this PG - msgr.stop_client(fd_it->second); + msgr.stop_client(peer_it->second->client_id); } } } @@ -212,10 +213,10 @@ bool osd_t::submit_flush_op(pool_id_t pool_id, pg_num_t pg_num, pg_flush_batch_t handle_flush_op(op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK, pool_id, pg_num, fb, peer_osd, op->reply.hdr.retval); delete op; }; - auto peer_fd_it = msgr.osd_peer_fds.find(peer_osd); - if (peer_fd_it != msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(peer_osd); + if (peer_it != msgr.osd_peers.end()) { - op->peer_fd = peer_fd_it->second; + op->client_id = peer_it->second->client_id; msgr.outbox_push(op); } else @@ -307,7 +308,7 @@ void osd_t::submit_recovery_op(osd_recovery_op_t *op) { printf("Submitting recovery operation for %jx:%jx (%s)\n", op->oid.inode, op->oid.stripe, op->degraded ? "degraded" : "misplaced"); } - op->osd_op->peer_fd = -1; + op->osd_op->client_id = SELF_CLIENT; op->osd_op->callback = [this, op](osd_op_t *osd_op) { ringloop->set_immediate([this, op]() diff --git a/src/osd/osd_peering.cpp b/src/osd/osd_peering.cpp index fe31a104..83243b12 100644 --- a/src/osd/osd_peering.cpp +++ b/src/osd/osd_peering.cpp @@ -9,7 +9,7 @@ #include "str_util.h" #include "osd.h" -#define SELF_FD -1 +#define SELF_CLIENT 0 // Peering loop void osd_t::handle_peers() @@ -175,17 +175,17 @@ void osd_t::drop_dirty_pg_connections(pool_pg_num_t pg) { if (immediate_commit != IMMEDIATE_ALL) { - std::vector to_stop; + std::vector to_stop; for (auto & cp: msgr.clients) { if (cp.second->dirty_pgs.find(pg) != cp.second->dirty_pgs.end()) { - to_stop.push_back(cp.first); + to_stop.push_back(cp.second->client_id); } } - for (auto peer_fd: to_stop) + for (auto client_id: to_stop) { - msgr.stop_client(peer_fd); + msgr.stop_client(client_id); } } } @@ -203,7 +203,7 @@ void osd_t::start_pg_peering(pg_t & pg) for (auto pg_osd: pg.all_peers) { if (pg_osd != this->osd_num && - msgr.osd_peer_fds.find(pg_osd) == msgr.osd_peer_fds.end()) + msgr.osd_peers.find(pg_osd) == msgr.osd_peers.end()) { if (msgr.wanted_peers.find(pg_osd) == msgr.wanted_peers.end()) msgr.connect_peer(pg_osd, st_cli.peer_states[pg_osd]); @@ -224,7 +224,7 @@ void osd_t::start_pg_peering(pg_t & pg) for (int role = 0; role < pg.target_set.size(); role++) { pg.cur_set[role] = pg.target_set[role] == this->osd_num || - msgr.osd_peer_fds.find(pg.target_set[role]) != msgr.osd_peer_fds.end() ? pg.target_set[role] : 0; + msgr.osd_peers.find(pg.target_set[role]) != msgr.osd_peers.end() ? pg.target_set[role] : 0; if (pg.cur_set[role] != 0) { pg.pg_cursize++; @@ -246,7 +246,7 @@ void osd_t::start_pg_peering(pg_t & pg) std::set dead_peers; for (auto pg_osd: pg.all_peers) { - if (pg_osd == this->osd_num || msgr.osd_peer_fds.find(pg_osd) != msgr.osd_peer_fds.end()) + if (pg_osd == this->osd_num || msgr.osd_peers.find(pg_osd) != msgr.osd_peers.end()) cur_peers.insert(pg_osd); else dead_peers.insert(pg_osd); @@ -266,7 +266,7 @@ void osd_t::start_pg_peering(pg_t & pg) { nonzero++; if (history_osd == this->osd_num || - msgr.osd_peer_fds.find(history_osd) != msgr.osd_peer_fds.end()) + msgr.osd_peers.find(history_osd) != msgr.osd_peers.end()) { found++; } @@ -435,8 +435,8 @@ void osd_t::relock_pg(pg_t & pg) bool unlock_peer = (i >= relock_osd_count); uint64_t new_state = unlock_peer ? 0 : pg.state; auto peer_osd = diff_osds[i]; - auto peer_fd_it = msgr.osd_peer_fds.find(peer_osd); - if (peer_fd_it == msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(peer_osd); + if (peer_it == msgr.osd_peers.end()) { if (unlock_peer) { @@ -446,8 +446,7 @@ void osd_t::relock_pg(pg_t & pg) } continue; } - int peer_fd = peer_fd_it->second; - auto cl = msgr.clients.at(peer_fd); + auto cl = peer_it->second; if (!cl->enable_pg_locks) { // Peer does not support locking - just instantly remember the lock as successful @@ -458,7 +457,7 @@ void osd_t::relock_pg(pg_t & pg) pg.inflight_locks++; osd_op_t *op = new osd_op_t(); op->op_type = OSD_OP_OUT; - op->peer_fd = peer_fd; + op->client_id = cl->client_id; op->req = (osd_any_op_t){ .sec_lock = { .header = { @@ -529,7 +528,7 @@ void osd_t::submit_list_subop(osd_num_t role_osd, pg_peering_state_t *ps) // Self osd_op_t *op = new osd_op_t(); op->op_type = 0; - op->peer_fd = SELF_FD; + op->client_id = SELF_CLIENT; clock_gettime(CLOCK_REALTIME, &op->tv_begin); op->bs_op = new blockstore_op_t(); op->bs_op->opcode = BS_OP_LIST; @@ -567,8 +566,8 @@ void osd_t::submit_list_subop(osd_num_t role_osd, pg_peering_state_t *ps) } else { - auto role_fd_it = msgr.osd_peer_fds.find(role_osd); - if (role_fd_it == msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(role_osd); + if (peer_it == msgr.osd_peers.end()) { printf("Failed to get object list from OSD %ju because it is disconnected\n", role_osd); return; @@ -576,7 +575,7 @@ void osd_t::submit_list_subop(osd_num_t role_osd, pg_peering_state_t *ps) // Peer osd_op_t *op = new osd_op_t(); op->op_type = OSD_OP_OUT; - op->peer_fd = role_fd_it->second; + op->client_id = peer_it->second->client_id; op->req = (osd_any_op_t){ .sec_list = { .header = { @@ -595,10 +594,10 @@ void osd_t::submit_list_subop(osd_num_t role_osd, pg_peering_state_t *ps) if (op->reply.hdr.retval < 0) { printf("Failed to get object list from OSD %ju (retval=%jd), disconnecting peer\n", role_osd, op->reply.hdr.retval); - int fail_fd = op->peer_fd; + uint64_t fail_client_id = op->client_id; ps->list_ops.erase(role_osd); delete op; - msgr.stop_client(fail_fd); + msgr.stop_client(fail_client_id); return; } printf( @@ -622,7 +621,7 @@ void osd_t::submit_list_subop(osd_num_t role_osd, pg_peering_state_t *ps) void osd_t::discard_list_subop(osd_op_t *list_op) { - if (list_op->peer_fd == SELF_FD) + if (list_op->client_id == SELF_CLIENT) { // Self list_op->bs_op->callback = [list_op](blockstore_op_t *bs_op) diff --git a/src/osd/osd_primary.cpp b/src/osd/osd_primary.cpp index 2fccfe53..d33b2958 100644 --- a/src/osd/osd_primary.cpp +++ b/src/osd/osd_primary.cpp @@ -795,7 +795,7 @@ resume_5: { this->dirty_osds.insert(chunk.osd_num); } - for (auto cl_it = msgr.clients.find(cur_op->peer_fd); cl_it != msgr.clients.end(); ) + for (auto cl_it = msgr.clients.find(cur_op->client_id); cl_it != msgr.clients.end(); ) { cl_it->second->dirty_pgs.insert({ .pool_id = pg.pool_id, .pg_num = pg.pg_num }); break; diff --git a/src/osd/osd_primary_chain.cpp b/src/osd/osd_primary_chain.cpp index 600c00d0..adb03200 100644 --- a/src/osd/osd_primary_chain.cpp +++ b/src/osd/osd_primary_chain.cpp @@ -312,16 +312,16 @@ int osd_t::submit_bitmap_subops(osd_op_t *cur_op, pg_t & pg) } handle_primary_subop(subop, cur_op); }; - auto peer_fd_it = msgr.osd_peer_fds.find(subop_osd_num); - if (peer_fd_it != msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(subop_osd_num); + if (peer_it != msgr.osd_peers.end()) { - subop->peer_fd = peer_fd_it->second; + subop->client_id = peer_it->second->client_id; msgr.outbox_push(subop); } else { // Fail it immediately - subop->peer_fd = -1; + subop->client_id = 0; subop->reply.hdr.retval = -EPIPE; ringloop->set_immediate([subop]() { std::function(subop->callback)(subop); }); } @@ -607,7 +607,7 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op) { if (cur > prev) { - // Send buffer in parts to avoid copying + // Send buffer in parts to avoid copying if (!prev_set) { while ((cur-prev) > zero_buffer_size/bs_bitmap_granularity) diff --git a/src/osd/osd_primary_subops.cpp b/src/osd/osd_primary_subops.cpp index 91c4eb63..4f2ff8ce 100644 --- a/src/osd/osd_primary_subops.cpp +++ b/src/osd/osd_primary_subops.cpp @@ -3,7 +3,7 @@ #include "osd_primary.h" -#define SELF_FD -1 +#define SELF_CLIENT 0 void osd_t::autosync() { @@ -15,7 +15,7 @@ void osd_t::autosync() } autosync_op = new osd_op_t(); autosync_op->op_type = OSD_OP_IN; - autosync_op->peer_fd = SELF_FD; + autosync_op->client_id = SELF_CLIENT; autosync_op->req = (osd_any_op_t){ .sync = { .header = { @@ -90,7 +90,7 @@ void osd_t::finish_op(osd_op_t *cur_op, int retval) cur_op->reply.hdr.id = cur_op->req.hdr.id; cur_op->reply.hdr.opcode = cur_op->req.hdr.opcode; cur_op->reply.hdr.retval = retval; - if (cur_op->peer_fd == SELF_FD) + if (cur_op->client_id == SELF_CLIENT) { // Do not include internal primary writes (recovery/rebalance) into client op statistics if (cur_op->req.hdr.opcode != OSD_OP_WRITE) @@ -103,7 +103,7 @@ void osd_t::finish_op(osd_op_t *cur_op, int retval) else { // FIXME add separate magic number for primary ops - auto cl_it = msgr.clients.find(cur_op->peer_fd); + auto cl_it = msgr.clients.find(cur_op->client_id); if (cl_it != msgr.clients.end()) { msgr.outbox_push(cur_op); @@ -243,7 +243,7 @@ void osd_t::submit_primary_subop(osd_op_t *cur_op, osd_op_t *subop, .offset = wr ? si->write_start : si->read_start, .len = subop_len, .attr_len = wr ? clean_entry_bitmap_size : 0, - .flags = cur_op->peer_fd == SELF_FD && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0, + .flags = cur_op->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0, }; #ifdef OSD_DEBUG printf( @@ -270,16 +270,16 @@ void osd_t::submit_primary_subop(osd_op_t *cur_op, osd_op_t *subop, { handle_primary_subop(subop, cur_op); }; - auto peer_fd_it = msgr.osd_peer_fds.find(si->osd_num); - if (peer_fd_it != msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(si->osd_num); + if (peer_it != msgr.osd_peers.end()) { - subop->peer_fd = peer_fd_it->second; + subop->client_id = peer_it->second->client_id; msgr.outbox_push(subop); } else { // Fail it immediately - subop->peer_fd = -1; + subop->client_id = 0; subop->reply.hdr.retval = -EPIPE; ringloop->set_immediate([subop]() { std::function(subop->callback)(subop); }); } @@ -326,7 +326,7 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop) } throw std::runtime_error("local blockstore modification failed"); } - bool recovery_related = cur_op->peer_fd == SELF_FD && cur_op->req.hdr.opcode != OSD_OP_SCRUB; + bool recovery_related = cur_op->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB; add_bs_subop_stats(subop, recovery_related); subop->req.hdr.opcode = bs_op_to_osd_op[bs_op->opcode]; subop->reply.hdr.retval = bs_op->retval; @@ -339,7 +339,7 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop) } delete bs_op; subop->bs_op = NULL; - subop->peer_fd = SELF_FD; + subop->client_id = SELF_CLIENT; if (recovery_related && recovery_target_sleep_us) { tfd->set_timer_us(recovery_target_sleep_us, false, [=](int timer_id) @@ -399,11 +399,15 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op) { uint64_t version = subop->reply.sec_rw.version; #ifdef OSD_DEBUG - int64_t peer_osd = subop->peer_fd == SELF_FD ? osd_num : - (msgr.clients.find(subop->peer_fd) != msgr.clients.end() - ? msgr.clients[subop->peer_fd]->osd_num : -subop->peer_fd); - printf("subop %s %jx:%jx from osd %jd: version = %ju\n", osd_op_names[opcode], - subop->req.sec_rw.oid.inode, subop->req.sec_rw.oid.stripe, peer_osd, version); + if (subop->client_id == SELF_CLIENT) + printf("subop %s %jx:%jx from local: version = %ju\n", osd_op_names[opcode], + subop->req.sec_rw.oid.inode, subop->req.sec_rw.oid.stripe, version); + else if (msgr.clients.find(subop->client_id) != msgr.clients.end()) + printf("subop %s %jx:%jx from osd %ju: version = %ju\n", osd_op_names[opcode], + subop->req.sec_rw.oid.inode, subop->req.sec_rw.oid.stripe, msgr.clients.at(subop->client_id)->osd_num, version); + else + printf("subop %s %jx:%jx from client %ju: version = %ju\n", osd_op_names[opcode], + subop->req.sec_rw.oid.inode, subop->req.sec_rw.oid.stripe, subop->client_id, version); #endif if (version != 0 && op_data->fact_ver != UINT64_MAX) { @@ -422,16 +426,16 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op) } if (retval != expected) { - int64_t peer_osd = (msgr.clients.find(subop->peer_fd) != msgr.clients.end() - ? msgr.clients[subop->peer_fd]->osd_num : 0); + int64_t peer_osd = (msgr.clients.find(subop->client_id) != msgr.clients.end() + ? msgr.clients.at(subop->client_id)->osd_num : 0); if (opcode == OSD_OP_SEC_READ || opcode == OSD_OP_SEC_WRITE || opcode == OSD_OP_SEC_WRITE_STABLE) { printf("%s subop to %jx:%jx v%ju failed ", osd_op_names[opcode], subop->req.sec_rw.oid.inode, subop->req.sec_rw.oid.stripe, subop->req.sec_rw.version); - if (subop->peer_fd >= 0 && peer_osd > 0) + if (subop->client_id && peer_osd > 0) printf("on osd %ju: retval = %d (expected %d)\n", peer_osd, retval, expected); - else if (peer_osd > 0) - printf("on peer %d: retval = %d (expected %d)\n", subop->peer_fd, retval, expected); + else if (subop->client_id) + printf("on client %ju: retval = %d (expected %d)\n", subop->client_id, retval, expected); else printf("locally: retval = %d (expected %d)\n", retval, expected); } @@ -446,7 +450,7 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op) else { printf( - "%s subop failed on osd %jd: retval = %d (expected %d)\n", + "%s subop failed on osd %ju: retval = %d (expected %d)\n", osd_op_names[opcode], peer_osd, retval, expected ); } @@ -460,12 +464,12 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op) { op_data->errcode = retval; } - if (subop->peer_fd >= 0 && retval != -EDOM && retval != -ERANGE && + if (subop->client_id && retval != -EDOM && retval != -ERANGE && (retval != -ENOSPC || opcode != OSD_OP_SEC_WRITE && opcode != OSD_OP_SEC_WRITE_STABLE) && (retval != -EIO || opcode != OSD_OP_SEC_READ)) { // Drop connection on unexpected errors - msgr.stop_client(subop->peer_fd); + msgr.stop_client(subop->client_id); op_data->drops++; } // Increase op_data->errors after stop_client to prevent >= n_subops running twice @@ -606,22 +610,22 @@ void osd_t::submit_primary_del_batch(osd_op_t *cur_op, obj_ver_osd_t *chunks_to_ }, .oid = chunk.oid, .version = chunk.version, - .flags = cur_op->peer_fd == SELF_FD && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0, + .flags = cur_op->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0, } }; subops[i].callback = [cur_op, this](osd_op_t *subop) { handle_primary_subop(subop, cur_op); }; - auto peer_fd_it = msgr.osd_peer_fds.find(chunk.osd_num); - if (peer_fd_it != msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(chunk.osd_num); + if (peer_it != msgr.osd_peers.end()) { - subops[i].peer_fd = peer_fd_it->second; + subops[i].client_id = peer_it->second->client_id; msgr.outbox_push(&subops[i]); } else { // Fail it immediately - subops[i].peer_fd = -1; + subops[i].client_id = 0; subops[i].reply.hdr.retval = -EPIPE; ringloop->set_immediate([subop = &subops[i]]() { std::function(subop->callback)(subop); }); } @@ -637,7 +641,7 @@ int osd_t::submit_primary_sync_subops(osd_op_t *cur_op) op_data->done = op_data->errors = op_data->errcode = 0; op_data->n_subops = n_osds; op_data->subops = subops; - std::map::iterator peer_it; + std::map::iterator peer_it; for (int i = 0; i < n_osds; i++) { osd_num_t sync_osd = op_data->dirty_osds[i]; @@ -654,16 +658,16 @@ int osd_t::submit_primary_sync_subops(osd_op_t *cur_op) }); bs->enqueue_op(subops[i].bs_op); } - else if ((peer_it = msgr.osd_peer_fds.find(sync_osd)) != msgr.osd_peer_fds.end()) + else if ((peer_it = msgr.osd_peers.find(sync_osd)) != msgr.osd_peers.end()) { subops[i].op_type = OSD_OP_OUT; - subops[i].peer_fd = peer_it->second; + subops[i].client_id = peer_it->second->client_id; subops[i].req = (osd_any_op_t){ .sec_sync = { .header = { .magic = SECONDARY_OSD_OP_MAGIC, .opcode = OSD_OP_SEC_SYNC, }, - .flags = cur_op->peer_fd == SELF_FD && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0, + .flags = cur_op->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0, } }; subops[i].callback = [cur_op, this](osd_op_t *subop) { @@ -722,23 +726,23 @@ void osd_t::submit_primary_stab_subops(osd_op_t *cur_op) .opcode = OSD_OP_SEC_STABILIZE, }, .len = (uint64_t)(stab_osd.len * sizeof(obj_ver_id)), - .flags = cur_op->peer_fd == SELF_FD && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0, + .flags = cur_op->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0, } }; subops[i].iov.push_back(op_data->unstable_writes + stab_osd.start, stab_osd.len * sizeof(obj_ver_id)); subops[i].callback = [cur_op, this](osd_op_t *subop) { handle_primary_subop(subop, cur_op); }; - auto peer_fd_it = msgr.osd_peer_fds.find(stab_osd.osd_num); - if (peer_fd_it != msgr.osd_peer_fds.end()) + auto peer_it = msgr.osd_peers.find(stab_osd.osd_num); + if (peer_it != msgr.osd_peers.end()) { - subops[i].peer_fd = peer_fd_it->second; + subops[i].client_id = peer_it->second->client_id; msgr.outbox_push(&subops[i]); } else { // Fail it immediately - subops[i].peer_fd = -1; + subops[i].client_id = 0; subops[i].reply.hdr.retval = -EPIPE; ringloop->set_immediate([subop = &subops[i]]() { std::function(subop->callback)(subop); }); } @@ -756,7 +760,7 @@ void osd_t::submit_primary_rollback_subops(osd_op_t *cur_op, const uint64_t* osd for (int role = 0; role < op_data->pg->pg_size; role++) { if (osd_set[role] != 0 && !stripes[role].read_error && - (osd_set[role] == this->osd_num || msgr.osd_peer_fds.find(osd_set[role]) != msgr.osd_peer_fds.end())) + (osd_set[role] == this->osd_num || msgr.osd_peers.find(osd_set[role]) != msgr.osd_peers.end())) { n_subops++; } @@ -773,7 +777,7 @@ void osd_t::submit_primary_rollback_subops(osd_op_t *cur_op, const uint64_t* osd for (int role = 0; role < op_data->pg->pg_size; role++) { if (osd_set[role] != 0 && !stripes[role].read_error && - (osd_set[role] == this->osd_num || msgr.osd_peer_fds.find(osd_set[role]) != msgr.osd_peer_fds.end())) + (osd_set[role] == this->osd_num || msgr.osd_peers.find(osd_set[role]) != msgr.osd_peers.end())) { osd_op_t *subop = &op_data->subops[i]; op_data->unstable_writes[i] = (obj_ver_id){ @@ -827,7 +831,7 @@ void osd_t::submit_primary_rollback_subops(osd_op_t *cur_op, const uint64_t* osd op_data->oid.inode, op_data->oid.stripe | role, op_data->target_ver-1 ); #endif - subop->peer_fd = msgr.osd_peer_fds.at(osd_set[role]); + subop->client_id = msgr.osd_peers.at(osd_set[role])->client_id; msgr.outbox_push(subop); } i++; diff --git a/src/osd/osd_primary_sync.cpp b/src/osd/osd_primary_sync.cpp index 6a56274d..3a8905f1 100644 --- a/src/osd/osd_primary_sync.cpp +++ b/src/osd/osd_primary_sync.cpp @@ -247,9 +247,9 @@ resume_8: else { finish: - if (cur_op->peer_fd) + if (cur_op->client_id) { - auto it = msgr.clients.find(cur_op->peer_fd); + auto it = msgr.clients.find(cur_op->client_id); if (it != msgr.clients.end()) it->second->dirty_pgs.clear(); } diff --git a/src/osd/osd_primary_write.cpp b/src/osd/osd_primary_write.cpp index 5618d160..0714e8a1 100644 --- a/src/osd/osd_primary_write.cpp +++ b/src/osd/osd_primary_write.cpp @@ -545,7 +545,7 @@ lazy: } // Remember PG as dirty to drop the connection when PG goes offline // (this is required because of the "lazy sync") - auto cl_it = msgr.clients.find(cur_op->peer_fd); + auto cl_it = msgr.clients.find(cur_op->client_id); if (cl_it != msgr.clients.end()) { cl_it->second->dirty_pgs.insert({ .pool_id = pg.pool_id, .pg_num = pg.pg_num }); diff --git a/src/osd/osd_scrub.cpp b/src/osd/osd_scrub.cpp index caa46557..f28659b4 100644 --- a/src/osd/osd_scrub.cpp +++ b/src/osd/osd_scrub.cpp @@ -3,7 +3,7 @@ #include "osd_primary.h" -#define SELF_FD -1 +#define SELF_CLIENT 0 void osd_t::scrub_list(pool_pg_num_t pg_id, osd_num_t role_osd, object_id min_oid) { @@ -16,7 +16,7 @@ void osd_t::scrub_list(pool_pg_num_t pg_id, osd_num_t role_osd, object_id min_oi // Self osd_op_t *op = new osd_op_t(); op->op_type = 0; - op->peer_fd = SELF_FD; + op->client_id = SELF_CLIENT; clock_gettime(CLOCK_REALTIME, &op->tv_begin); op->bs_op = new blockstore_op_t(); op->bs_op->opcode = BS_OP_LIST; @@ -61,7 +61,7 @@ void osd_t::scrub_list(pool_pg_num_t pg_id, osd_num_t role_osd, object_id min_oi // Peer osd_op_t *op = new osd_op_t(); op->op_type = OSD_OP_OUT; - op->peer_fd = msgr.osd_peer_fds.at(role_osd); + op->client_id = msgr.osd_peers.at(role_osd)->client_id; op->req = (osd_any_op_t){ .sec_list = { .header = { @@ -83,9 +83,9 @@ void osd_t::scrub_list(pool_pg_num_t pg_id, osd_num_t role_osd, object_id min_oi if (op->reply.hdr.retval < 0) { printf("Failed to get object list from OSD %ju (retval=%jd), disconnecting peer\n", role_osd, op->reply.hdr.retval); - int fail_fd = op->peer_fd; + uint64_t fail_client_id = op->client_id; delete op; - msgr.stop_client(fail_fd); + msgr.stop_client(fail_client_id); return; } scrub_cur_list = { @@ -224,7 +224,7 @@ void osd_t::submit_scrub_op(object_id oid) { auto osd_op = new osd_op_t(); osd_op->op_type = OSD_OP_OUT; - osd_op->peer_fd = -1; + osd_op->client_id = SELF_CLIENT; osd_op->req = (osd_any_op_t){ .rw = { .header = { diff --git a/src/osd/osd_secondary.cpp b/src/osd/osd_secondary.cpp index ec5c28a6..d553a128 100644 --- a/src/osd/osd_secondary.cpp +++ b/src/osd/osd_secondary.cpp @@ -128,7 +128,7 @@ void osd_t::exec_secondary_real(osd_op_t *cur_op) exec_sec_lock(cur_op); return; } - auto cl = msgr.clients.at(cur_op->peer_fd); + osd_client_t *cl = msgr.clients.at(cur_op->client_id); cur_op->bs_op = new blockstore_op_t(); cur_op->bs_op->callback = [this, cur_op](blockstore_op_t* bs_op) { secondary_op_callback(cur_op); }; cur_op->bs_op->opcode = (cur_op->req.hdr.opcode == OSD_OP_SEC_READ ? BS_OP_READ @@ -249,7 +249,7 @@ void osd_t::exec_secondary_real(osd_op_t *cur_op) void osd_t::exec_sec_read_bmp(osd_op_t *cur_op) { - auto cl = msgr.clients.at(cur_op->peer_fd); + auto cl = msgr.clients.at(cur_op->client_id); int n = cur_op->req.sec_read_bmp.len / sizeof(obj_ver_id); if (n > 0) { @@ -278,7 +278,7 @@ void osd_t::exec_sec_read_bmp(osd_op_t *cur_op) void osd_t::exec_sec_lock(osd_op_t *cur_op) { cur_op->reply.sec_lock.cur_primary = 0; - auto cl = msgr.clients.at(cur_op->peer_fd); + auto cl = msgr.clients.at(cur_op->client_id); if (!cl->in_osd_num || cur_op->req.sec_lock.flags != OSD_SEC_LOCK_PG && cur_op->req.sec_lock.flags != OSD_SEC_UNLOCK_PG || @@ -340,7 +340,7 @@ void osd_t::exec_show_config(osd_op_t *cur_op) ? json11::Json::parse(std::string((char *)cur_op->buf), json_err) : json11::Json(); auto peer_osd_num = req_json["osd_num"].uint64_value(); - auto cl = msgr.clients.at(cur_op->peer_fd); + auto cl = msgr.clients.at(cur_op->client_id); cl->in_osd_num = peer_osd_num; if (req_json["features"]["check_sequencing"].bool_value()) { @@ -369,7 +369,7 @@ void osd_t::exec_show_config(osd_op_t *cur_op) if (req_json["connect_rdma"].is_string()) { // Peer is trying to connect using RDMA, try to satisfy him - bool ok = msgr.connect_rdma(cur_op->peer_fd, req_json["connect_rdma"].string_value(), req_json["rdma_max_msg"].uint64_value()); + bool ok = msgr.connect_rdma(cur_op->client_id, req_json["connect_rdma"].string_value(), req_json["rdma_max_msg"].uint64_value()); if (ok) { auto rc = cl->rdma_conn; diff --git a/src/test/mock/messenger.cpp b/src/test/mock/messenger.cpp index 911b0afd..bfcb5e90 100644 --- a/src/test/mock/messenger.cpp +++ b/src/test/mock/messenger.cpp @@ -15,13 +15,13 @@ osd_messenger_t::~osd_messenger_t() { while (clients.size() > 0) { - stop_client(clients.begin()->first, true, true); + stop_client(clients.begin()->first, true); } } void osd_messenger_t::outbox_push(osd_op_t *cur_op) { - auto cl = clients.at(cur_op->peer_fd); + auto cl = clients.at(cur_op->client_id); cur_op->req.hdr.id = ++cl->send_op_id; cl->sent_ops[cur_op->req.hdr.id] = cur_op; } @@ -57,7 +57,3 @@ json11::Json::object osd_messenger_t::merge_configs(const json11::Json::object & { return cli_config; } - -void osd_messenger_t::clear_immediate_ops(int peer_fd) -{ -} diff --git a/src/test/stub_uring_osd.cpp b/src/test/stub_uring_osd.cpp index cc5c7188..1a8b614f 100644 --- a/src/test/stub_uring_osd.cpp +++ b/src/test/stub_uring_osd.cpp @@ -89,7 +89,7 @@ void stub_exec_op(osd_messenger_t *msgr, osd_op_t *op) } else { - printf("client %d: unsupported stub opcode: %ju\n", op->peer_fd, op->req.hdr.opcode); + printf("client %ju: unsupported stub opcode: %ju\n", op->client_id, op->req.hdr.opcode); op->reply.hdr.retval = -EINVAL; } msgr->outbox_push(op); diff --git a/src/test/test_cluster_client.cpp b/src/test/test_cluster_client.cpp index e3fd91e3..945971b1 100644 --- a/src/test/test_cluster_client.cpp +++ b/src/test/test_cluster_client.cpp @@ -118,11 +118,14 @@ void pretend_connected(cluster_client_t *cli, osd_num_t osd_num) { printf("OSD %ju connected\n", osd_num); int peer_fd = cli->msgr.clients.size() ? std::prev(cli->msgr.clients.end())->first+1 : 10; - cli->msgr.osd_peer_fds[osd_num] = peer_fd; - cli->msgr.clients[peer_fd] = new osd_client_t(); - cli->msgr.clients[peer_fd]->osd_num = osd_num; - cli->msgr.clients[peer_fd]->peer_fd = peer_fd; - cli->msgr.clients[peer_fd]->peer_state = PEER_CONNECTED; + auto cl = new osd_client_t(); + cl->client_id = cli->msgr.next_client_id++; + cl->osd_num = osd_num; + cl->peer_fd = peer_fd; + cl->peer_state = PEER_CONNECTED; + cli->msgr.osd_peers[osd_num] = cl; + cli->msgr.clients[cl->client_id] = cl; + cli->msgr.clients_by_fd[peer_fd] = cl; cli->msgr.wanted_peers.erase(osd_num); cli->msgr.repeer_pgs(osd_num); } @@ -130,12 +133,12 @@ void pretend_connected(cluster_client_t *cli, osd_num_t osd_num) void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num) { printf("OSD %ju disconnected\n", osd_num); - cli->msgr.stop_client(cli->msgr.osd_peer_fds.at(osd_num)); + cli->msgr.stop_client(cli->msgr.osd_peers.at(osd_num)->client_id); } void check_disconnected(cluster_client_t *cli, osd_num_t osd_num) { - if (cli->msgr.osd_peer_fds.find(osd_num) != cli->msgr.osd_peer_fds.end()) + if (cli->msgr.osd_peers.find(osd_num) != cli->msgr.osd_peers.end()) { printf("OSD %ju not disconnected as it ought to be\n", osd_num); assert(0); @@ -144,8 +147,8 @@ void check_disconnected(cluster_client_t *cli, osd_num_t osd_num) void check_op_count(cluster_client_t *cli, osd_num_t osd_num, int ops) { - int peer_fd = cli->msgr.osd_peer_fds.at(osd_num); - int real_ops = cli->msgr.clients[peer_fd]->sent_ops.size(); + osd_client_t *cl = cli->msgr.osd_peers.at(osd_num); + int real_ops = cl->sent_ops.size(); if (real_ops != ops) { printf("error: %d ops expected, but %d queued\n", ops, real_ops); @@ -155,9 +158,9 @@ void check_op_count(cluster_client_t *cli, osd_num_t osd_num, int ops) osd_op_t *find_op(cluster_client_t *cli, osd_num_t osd_num, uint64_t opcode, uint64_t offset, uint64_t len) { - int peer_fd = cli->msgr.osd_peer_fds.at(osd_num); - auto op_it = cli->msgr.clients[peer_fd]->sent_ops.begin(); - while (op_it != cli->msgr.clients[peer_fd]->sent_ops.end()) + osd_client_t *cl = cli->msgr.osd_peers.at(osd_num); + auto op_it = cl->sent_ops.begin(); + while (op_it != cl->sent_ops.end()) { auto op = op_it->second; if (op->req.hdr.opcode == opcode && (opcode == OSD_OP_SYNC || @@ -167,8 +170,8 @@ osd_op_t *find_op(cluster_client_t *cli, osd_num_t osd_num, uint64_t opcode, uin } op_it++; } - op_it = cli->msgr.clients[peer_fd]->sent_ops.begin(); - while (op_it != cli->msgr.clients[peer_fd]->sent_ops.end()) + op_it = cl->sent_ops.begin(); + while (op_it != cl->sent_ops.end()) { printf("Found opcode %ju offset %jx size %x\n", op_it->second->req.hdr.opcode, op_it->second->req.rw.offset, op_it->second->req.rw.len); op_it++; @@ -183,8 +186,8 @@ void pretend_op_completed(cluster_client_t *cli, osd_op_t *op, int64_t retval) printf("Pretend completed %s %jx+%x\n", op->req.hdr.opcode == OSD_OP_SYNC ? "sync" : (op->req.hdr.opcode == OSD_OP_WRITE ? "write" : "read"), op->req.rw.offset, op->req.rw.len); uint64_t op_id = op->req.hdr.id; - int peer_fd = op->peer_fd; - cli->msgr.clients[peer_fd]->sent_ops.erase(op_id); + uint64_t client_id = op->client_id; + cli->msgr.clients[client_id]->sent_ops.erase(op_id); op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC; op->reply.hdr.id = op->req.hdr.id; op->reply.hdr.opcode = op->req.hdr.opcode; @@ -264,7 +267,7 @@ void test1() uint64_t replay_start = UINT64_MAX; uint64_t replay_end = 0; std::vector replay_ops; - auto osd_cl = cli->msgr.clients.at(cli->msgr.osd_peer_fds.at(1)); + auto osd_cl = cli->msgr.osd_peers.at(1); for (auto & op_p: osd_cl->sent_ops) { auto op = op_p.second;