diff --git a/src/client/cluster_client.cpp b/src/client/cluster_client.cpp index b631d4d2..480d06f7 100644 --- a/src/client/cluster_client.cpp +++ b/src/client/cluster_client.cpp @@ -103,6 +103,46 @@ cluster_op_t::~cluster_op_t() } } +bool cluster_op_t::support_left_on_dead() +{ + if (!parts.size()) + { + return false; + } + for (auto & part: parts) + { + if (!(part.flags & PART_DONE) || + part.op.reply.hdr.opcode != OSD_OP_DELETE || + part.op.reply.hdr.retval != 0 || + !(part.op.reply.del.flags & OSD_DEL_SUPPORT_LEFT_ON_DEAD)) + { + return false; + } + } + return true; +} + +std::vector cluster_op_t::get_left_on_dead() +{ + std::set osds; + for (auto & part: parts) + { + if ((part.flags & PART_DONE) || + part.op.reply.hdr.opcode == OSD_OP_DELETE && + part.op.reply.hdr.retval == 0 && + (part.op.reply.del.flags & OSD_DEL_LEFT_ON_DEAD) != 0) + { + int del_count = (OSD_PACKET_SIZE-sizeof(part.op.reply.del)) / sizeof(uint32_t); + if (del_count > part.op.reply.del.left_on_dead_count) + del_count = part.op.reply.del.left_on_dead_count; + uint32_t *left_on_dead = (uint32_t*)((&part.op.reply.del) + 1); + for (int i = 0; i < del_count; i++) + osds.insert(left_on_dead[i]); + } + } + return std::vector(osds.begin(), osds.end()); +} + void cluster_client_t::continue_raw_ops(osd_num_t peer_osd) { auto it = raw_ops.find(peer_osd); diff --git a/src/client/cluster_client.h b/src/client/cluster_client.h index ac7f93c5..632d8bf0 100644 --- a/src/client/cluster_client.h +++ b/src/client/cluster_client.h @@ -53,6 +53,10 @@ struct cluster_op_t void *bitmap_buf = NULL; std::function callback; ~cluster_op_t(); + + // for deletions, remove after 'atomic delete': + bool support_left_on_dead(); + std::vector get_left_on_dead(); protected: int state = 0; uint64_t cur_inode; // for snapshot reads @@ -142,7 +146,7 @@ public: void continue_ops(int time_passed = 0); void list_inode(inode_t inode, uint64_t min_offset, uint64_t max_offset, int max_parallel_pgs, std::function&& objects, std::vector && inactive_osds)> pg_callback); + int status, int pgs_left, pg_num_t pg_num, std::set&& objects)> pg_callback); //inline uint32_t get_bs_bitmap_granularity() { return st_cli.global_bitmap_granularity; } //inline uint64_t get_bs_block_size() { return st_cli.global_block_size; } diff --git a/src/client/cluster_client_list.cpp b/src/client/cluster_client_list.cpp index 5d7f4fb6..c7898db6 100644 --- a/src/client/cluster_client_list.cpp +++ b/src/client/cluster_client_list.cpp @@ -55,11 +55,11 @@ struct inode_list_t int onstack = 0; std::vector pgs; pg_num_t real_pg_count = 0; - std::function&& objects, std::vector && inactive_osds)> callback; + std::function&& objects)> callback; }; void cluster_client_t::list_inode(inode_t inode, uint64_t min_offset, uint64_t max_offset, int max_parallel_pgs, std::function&& objects, std::vector && inactive_osds)> pg_callback) + int status, int pgs_left, pg_num_t pg_num, std::set&& objects)> pg_callback) { init_msgr(); pool_id_t pool_id = INODE_POOL(inode); @@ -67,7 +67,7 @@ void cluster_client_t::list_inode(inode_t inode, uint64_t min_offset, uint64_t m { if (log_level > 0) fprintf(stderr, "Pool %u does not exist\n", pool_id); - pg_callback(-EINVAL, 0, 0, std::set(), std::vector()); + pg_callback(-EINVAL, 0, 0, std::set()); return; } auto pg_stripe_size = st_cli.pool_config.at(pool_id).pg_stripe_size; @@ -139,13 +139,13 @@ bool cluster_client_t::restart_listing(inode_list_t* lst) if (pool_it == st_cli.pool_config.end()) { // Unknown pool - lst->callback(-EINVAL, 0, 0, std::set(), std::vector()); + lst->callback(-EINVAL, 0, 0, std::set()); return false; } else if (lst->done_pgs) { // PG count changed during listing, it should fail - lst->callback(-EAGAIN, 0, 0, std::set(), std::vector()); + lst->callback(-EAGAIN, 0, 0, std::set()); return false; } else @@ -245,7 +245,7 @@ void cluster_client_t::set_list_retry_timeout(int ms, timespec new_time) int cluster_client_t::start_pg_listing(inode_list_pg_t *pg) { - auto & pool_cfg = st_cli.pool_config[pg->lst->pool_id]; + auto & pool_cfg = st_cli.pool_config.at(pg->lst->pool_id); auto pg_it = pool_cfg.pg_config.find(pg->pg_num); assert(pg->lst->real_pg_count == pool_cfg.real_pg_count); if (pg_it == pool_cfg.pg_config.end() || @@ -258,7 +258,7 @@ int cluster_client_t::start_pg_listing(inode_list_pg_t *pg) } pg->inactive_osds.clear(); std::set all_peers; - if (pg_it->second.cur_state != PG_ACTIVE) + if (pg_it->second.cur_state != PG_ACTIVE && pg->lst->fallback) { // Not clean and OSDs don't support listing from primary for (osd_num_t pg_osd: pg_it->second.target_set) @@ -283,10 +283,9 @@ int cluster_client_t::start_pg_listing(inode_list_pg_t *pg) peer_it++; } } - if (pg_it->second.cur_state == PG_ACTIVE || !pg->lst->fallback) + else { // Clean - all_peers.clear(); all_peers.insert(pg_it->second.cur_primary); } // Check that we're connected to all PG OSDs @@ -440,7 +439,7 @@ void cluster_client_t::finish_list_pg(inode_list_pg_t *pg, bool retry_epipe) } lst->done_pgs++; pg->state = LIST_PG_DONE; - lst->callback(pg->errcode, lst->pgs.size()-lst->done_pgs, pg->pg_num, std::move(pg->objects), std::move(pg->inactive_osds)); + lst->callback(pg->errcode, lst->pgs.size()-lst->done_pgs, pg->pg_num, std::move(pg->objects)); pg->objects.clear(); pg->inactive_osds.clear(); } diff --git a/src/client/osd_ops.h b/src/client/osd_ops.h index 941ed34f..109995be 100644 --- a/src/client/osd_ops.h +++ b/src/client/osd_ops.h @@ -53,6 +53,9 @@ #define OSD_LIST_PRIMARY 1 +#define OSD_DEL_SUPPORT_LEFT_ON_DEAD 1 +#define OSD_DEL_LEFT_ON_DEAD 2 + // common request and reply headers struct __attribute__((__packed__)) osd_op_header_t { @@ -242,6 +245,20 @@ struct __attribute__((__packed__)) osd_reply_rw_t uint64_t version; }; +struct __attribute__((__packed__)) osd_reply_del_t +{ + osd_reply_header_t header; + // OSD_DEL_SUPPORT_LEFT_ON_DEAD and/or OSD_DEL_LEFT_ON_DEAD or 0 + uint32_t flags; + // for deletes, if flags & OSD_DEL_LEFT_ON_DEAD: + // count of OSDs from which the object could be not deleted + // these come directly after this del_left_on_dead_list_size as uint32_t[] + // FIXME it's kind of a hack and will be removed in the future, when Vitastor will + // have 'atomic deletions', i.e. when it will be able to remember deleted objects + // and complete deletions automatically after extra OSDs are started + uint32_t left_on_dead_count; +}; + // sync to the primary OSD struct __attribute__((__packed__)) osd_op_sync_t { @@ -314,6 +331,7 @@ union osd_any_reply_t osd_reply_sec_list_t sec_list; osd_reply_show_config_t show_conf; osd_reply_rw_t rw; + osd_reply_del_t del; osd_reply_sync_t sync; osd_reply_describe_t describe; uint8_t buf[OSD_PACKET_SIZE]; diff --git a/src/cmd/cli_merge.cpp b/src/cmd/cli_merge.cpp index a0f3e1d9..8081ce3b 100644 --- a/src/cmd/cli_merge.cpp +++ b/src/cmd/cli_merge.cpp @@ -390,7 +390,7 @@ struct snap_merger_t { lists_todo++; parent->cli->list_inode(src, 0, 0, parent->parallel_osds, [this, src]( - int errcode, int pgs_left, pg_num_t pg_num, std::set&& objects, std::vector && inactive_osds) + int errcode, int pgs_left, pg_num_t pg_num, std::set&& objects) { if (errcode) { diff --git a/src/cmd/cli_rm_data.cpp b/src/cmd/cli_rm_data.cpp index fe6a405f..877839af 100644 --- a/src/cmd/cli_rm_data.cpp +++ b/src/cmd/cli_rm_data.cpp @@ -32,6 +32,7 @@ struct rm_inode_t std::vector lists; std::set inactive_osds; std::set inactive_pgs; + std::set fallback_pgs; uint64_t total_count = 0, total_done = 0, total_prev_pct = 0; bool lists_done = false; int pgs_to_list = 0; @@ -52,7 +53,7 @@ struct rm_inode_t } pgs_to_list = pool_it->second.real_pg_count; parent->cli->list_inode(inode, min_offset, max_offset, parent->parallel_osds, [this]( - int errcode, int pgs_left, pg_num_t pg_num, std::set&& objects, std::vector && inactive_osds) + int errcode, int pgs_left, pg_num_t pg_num, std::set&& objects) { if (errcode) { @@ -60,10 +61,6 @@ struct rm_inode_t } else { - for (auto osd_num: inactive_osds) - { - this->inactive_osds.insert(osd_num); - } rm_pg_t *rm = new rm_pg_t((rm_pg_t){ .pg_num = pg_num, .objects = std::move(objects), @@ -114,6 +111,13 @@ struct rm_inode_t op->inode, op->offset, cur_list->pg_num, op->retval); error_count++; } + else + { + if (!op->support_left_on_dead()) + fallback_pgs.insert(cur_list->pg_num); + for (auto inactive_osd: op->get_left_on_dead()) + inactive_osds.insert(inactive_osd); + } delete op; cur_list->obj_done++; total_done++; @@ -188,7 +192,36 @@ struct rm_inode_t { fprintf(stderr, "\n"); } - // FIXME: for 100% correctness inactive_osds should be taken from OSD_OP_DELETE reply, not from the listing + if (fallback_pgs.size() && !parent->json_output) + { + fprintf(stderr, "Warning: some OSDs don't indicate left_on_dead PG OSDs" + " in delete replies, falling back to simpler checks\n"); + auto pool_it = parent->cli->st_cli.pool_config.find(pool_id); + if (pool_it != parent->cli->st_cli.pool_config.end()) + { + std::set all_peers; + for (auto pg_num: fallback_pgs) + { + auto pg_it = pool_it->second.pg_config.find(pg_num); + if (pg_it != pool_it->second.pg_config.end()) + { + for (osd_num_t pg_osd: pg_it->second.target_set) + all_peers.insert(pg_osd); + for (osd_num_t pg_osd: pg_it->second.all_peers) + all_peers.insert(pg_osd); + for (auto & hist_item: pg_it->second.target_history) + for (auto pg_osd: hist_item) + all_peers.insert(pg_osd); + } + } + all_peers.erase(0); + for (auto peer_osd: all_peers) + { + if (parent->cli->st_cli.peer_states[peer_osd].is_null()) + inactive_osds.insert(peer_osd); + } + } + } if (inactive_osds.size() && !parent->json_output) { fprintf(stderr, "Some data may remain after delete on OSDs which are currently down: "); @@ -209,7 +242,7 @@ struct rm_inode_t } fprintf(stderr, "\n"); } - if (error_count > 0) + if (error_count > 0 && !parent->json_output) { fprintf(stderr, "Failed to delete %u objects from active OSD(s).\n", error_count); } diff --git a/src/osd/osd_peering.cpp b/src/osd/osd_peering.cpp index f739ff6d..4db2b2e8 100644 --- a/src/osd/osd_peering.cpp +++ b/src/osd/osd_peering.cpp @@ -133,6 +133,7 @@ void osd_t::repeer_pgs(osd_num_t peer_osd) void osd_t::reset_pg(pg_t & pg) { pg.cur_peers.clear(); + pg.dead_peers.clear(); pg.state_dict.clear(); copies_to_delete_after_sync_count -= pg.copies_to_delete_after_sync.size(); pg.copies_to_delete_after_sync.clear(); @@ -235,13 +236,16 @@ void osd_t::start_pg_peering(pg_t & pg) return; } std::set cur_peers; + 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()) - { cur_peers.insert(pg_osd); - } + else + dead_peers.insert(pg_osd); } + pg.cur_peers.insert(pg.cur_peers.begin(), cur_peers.begin(), cur_peers.end()); + pg.dead_peers.insert(pg.dead_peers.begin(), dead_peers.begin(), dead_peers.end()); if (pg.target_history.size()) { // Refuse to start PG if no peers are available from any of the historical OSD sets @@ -269,7 +273,6 @@ void osd_t::start_pg_peering(pg_t & pg) } } } - pg.cur_peers.insert(pg.cur_peers.begin(), cur_peers.begin(), cur_peers.end()); if (pg.peering_state) { // Adjust the peering operation that's still in progress - discard unneeded results diff --git a/src/osd/osd_peering_pg.h b/src/osd/osd_peering_pg.h index c7397794..593f43cb 100644 --- a/src/osd/osd_peering_pg.h +++ b/src/osd/osd_peering_pg.h @@ -97,6 +97,8 @@ struct pg_t bool history_changed = false; // peer list from the last peering event std::vector cur_peers; + // dead_peers = all_peers - cur_peers + std::vector dead_peers; // target_set is the "correct" peer OSD set for this PG std::vector target_set; // cur_set is the current set of connected peer OSDs for this PG diff --git a/src/osd/osd_primary.cpp b/src/osd/osd_primary.cpp index 629ed9a1..33c9f477 100644 --- a/src/osd/osd_primary.cpp +++ b/src/osd/osd_primary.cpp @@ -783,6 +783,20 @@ resume_5: } pg.total_count--; cur_op->reply.hdr.retval = 0; + // indicate possibly unfinished (left_on_dead) deletions + cur_op->reply.del.flags = OSD_DEL_SUPPORT_LEFT_ON_DEAD; + if (pg.dead_peers.size() > 0) + { + int max_del = (OSD_PACKET_SIZE-sizeof(cur_op->reply.del)) / sizeof(uint32_t); + cur_op->reply.del.flags |= OSD_DEL_LEFT_ON_DEAD; + cur_op->reply.del.left_on_dead_count = pg.dead_peers.size() < max_del + ? pg.dead_peers.size() : max_del; + uint32_t *left_on_dead = (uint32_t*)((&cur_op->reply.del) + 1); + for (int i = 0; i < cur_op->reply.del.left_on_dead_count; i++) + { + left_on_dead[i] = pg.dead_peers[i]; + } + } continue_others: osd_op_t *next_op = NULL; auto next_it = pg.write_queue.find(op_data->oid);