Return left_on_dead OSD list in DELETE replies and use it in rm-data

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<osd_num_t> cluster_op_t::get_left_on_dead()
+{
+    std::set<osd_num_t> 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<osd_num_t>(osds.begin(), osds.end());
+}
+
 void cluster_client_t::continue_raw_ops(osd_num_t peer_osd)
 {
     auto it = raw_ops.find(peer_osd);

src/client/cluster_client.h

@@ -53,6 +53,10 @@ struct cluster_op_t
     void *bitmap_buf = NULL;
     std::function<void(cluster_op_t*)> callback;
     ~cluster_op_t();
+
+    // for deletions, remove after 'atomic delete':
+    bool support_left_on_dead();
+    std::vector<osd_num_t> 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<void(
-        int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects, std::vector<osd_num_t> && inactive_osds)> pg_callback);
+        int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& 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; }

src/client/cluster_client_list.cpp

@@ -55,11 +55,11 @@ struct inode_list_t
     int onstack = 0;
     std::vector<inode_list_pg_t*> pgs;
     pg_num_t real_pg_count = 0;
-    std::function<void(int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects, std::vector<osd_num_t> && inactive_osds)> callback;
+    std::function<void(int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& 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<void(
-    int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects, std::vector<osd_num_t> && inactive_osds)> pg_callback)
+    int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& 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<object_id>(), std::vector<osd_num_t>());
+        pg_callback(-EINVAL, 0, 0, std::set<object_id>());
         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<object_id>(), std::vector<osd_num_t>());
+            lst->callback(-EINVAL, 0, 0, std::set<object_id>());
             return false;
         }
         else if (lst->done_pgs)
         {
             // PG count changed during listing, it should fail
-            lst->callback(-EAGAIN, 0, 0, std::set<object_id>(), std::vector<osd_num_t>());
+            lst->callback(-EAGAIN, 0, 0, std::set<object_id>());
             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<osd_num_t> 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();
     }

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];

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<object_id>&& objects, std::vector<osd_num_t> && inactive_osds)
+                    int errcode, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects)
                 {
                     if (errcode)
                     {

src/cmd/cli_rm_data.cpp

@@ -32,6 +32,7 @@ struct rm_inode_t
     std::vector<rm_pg_t*> lists;
     std::set<osd_num_t> inactive_osds;
     std::set<pg_num_t> inactive_pgs;
+    std::set<pg_num_t> 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<object_id>&& objects, std::vector<osd_num_t> && inactive_osds)
+            int errcode, int pgs_left, pg_num_t pg_num, std::set<object_id>&& 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<osd_num_t> 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);
             }

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<osd_num_t> cur_peers;
+    std::set<osd_num_t> 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

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<osd_num_t> cur_peers;
+    // dead_peers = all_peers - cur_peers
+    std::vector<osd_num_t> dead_peers;
     // target_set is the "correct" peer OSD set for this PG
     std::vector<osd_num_t> target_set;
     // cur_set is the current set of connected peer OSDs for this PG

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);