Limit the number of unstable versions per object

src/osd/osd.h

@@ -98,6 +98,12 @@ struct osd_pg_lock_t
     uint64_t state = 0;
 };
 
+struct osd_unstable_wr_t
+{
+    uint64_t latest_ver = 0;
+    uint64_t ver_count = 0;
+};
+
 class osd_t
 {
     // config
@@ -123,6 +129,7 @@ class osd_t
     int slow_log_interval = 10;
     int immediate_commit = IMMEDIATE_NONE;
     int autosync_interval = DEFAULT_AUTOSYNC_INTERVAL; // "emergency" sync every 5 seconds
+    int autosync_dirty_per_object = 16;
     int autosync_writes = DEFAULT_AUTOSYNC_WRITES;
     uint64_t recovery_queue_depth = 1;
     uint64_t recovery_sleep_us = 0;
@@ -195,7 +202,8 @@ class osd_t
 
     // Unstable writes
     uint64_t unstable_write_count = 0;
-    std::map<osd_object_id_t, uint64_t> unstable_writes;
+    uint64_t unstable_per_object = 0;
+    std::map<osd_object_id_t, osd_unstable_wr_t> unstable_writes;
     std::deque<osd_op_t*> syncs_in_progress;
 
     // client & peer I/O

src/osd/osd_primary.cpp

@@ -790,9 +790,9 @@ resume_5:
         if (immediate_commit == IMMEDIATE_NONE)
         {
             unstable_write_count++;
-            if (unstable_write_count >= autosync_writes)
+            if (unstable_write_count >= autosync_writes ||
+                unstable_per_object >= autosync_dirty_per_object)
             {
-                unstable_write_count = 0;
                 autosync();
             }
         }

src/osd/osd_primary.h

@@ -43,6 +43,7 @@ struct osd_primary_op_data_t
             osd_num_t *dirty_osds;
             int dirty_osd_count;
             obj_ver_id *unstable_writes;
+            uint64_t *unstable_ver_counts;
             obj_ver_osd_t *copies_to_delete;
             int copies_to_delete_count;
         };

src/osd/osd_primary_sync.cpp

@@ -45,7 +45,10 @@ resume_2:
     if (unstable_writes.size() > 0)
     {
         op_data->unstable_write_osds = new std::vector<unstable_osd_num_t>();
-        op_data->unstable_writes = new obj_ver_id[this->unstable_writes.size()];
+        op_data->unstable_writes = (obj_ver_id*)malloc_or_die(
+            (sizeof(obj_ver_id) + sizeof(uint64_t)) * this->unstable_writes.size());
+        op_data->unstable_ver_counts = (uint64_t*)((uint8_t*)op_data->unstable_writes +
+            sizeof(obj_ver_id) * this->unstable_writes.size());
         osd_num_t last_osd = 0;
         int last_start = 0, last_end = 0;
         for (auto it = this->unstable_writes.begin(); it != this->unstable_writes.end(); it++)
@@ -65,8 +68,9 @@ resume_2:
             }
             op_data->unstable_writes[last_end] = (obj_ver_id){
                 .oid = it->first.oid,
-                .version = it->second,
+                .version = it->second.latest_ver,
             };
+            op_data->unstable_ver_counts[last_end] = it->second.ver_count;
             last_end++;
         }
         if (last_osd != 0)
@@ -78,6 +82,8 @@ resume_2:
             });
         }
         this->unstable_writes.clear();
+        this->unstable_write_count = 0;
+        this->unstable_per_object = 0;
     }
     {
         op_data->dirty_pg_count = dirty_pgs.size();
@@ -175,11 +181,12 @@ resume_6:
                     };
                     if (pgs.at(wpg).state & PG_ACTIVE)
                     {
-                        uint64_t & dest = this->unstable_writes[(osd_object_id_t){
+                        auto & dest = this->unstable_writes[(osd_object_id_t){
                             .osd_num = unstable_osd.osd_num,
                             .oid = w.oid,
                         }];
-                        dest = dest < w.version ? w.version : dest;
+                        dest.latest_ver = dest.latest_ver < w.version ? w.version : dest.latest_ver;
+                        dest.ver_count += op_data->unstable_ver_counts[unstable_osd.start + i];
                         dirty_pgs.insert(wpg);
                     }
                 }
@@ -236,7 +243,7 @@ resume_8:
     if (op_data->unstable_writes)
     {
         delete op_data->unstable_write_osds;
-        delete[] op_data->unstable_writes;
+        free(op_data->unstable_writes);
         op_data->unstable_writes = NULL;
         op_data->unstable_write_osds = NULL;
     }

src/osd/osd_primary_write.cpp

@@ -408,9 +408,9 @@ continue_others:
     }
     // finish_op would invalidate next_it if it cleared pg.write_queue, but it doesn't do that :)
     finish_op(cur_op, cur_op->reply.hdr.retval);
-    if (unstable_write_count >= autosync_writes)
+    if (unstable_write_count >= autosync_writes ||
+        unstable_per_object >= autosync_dirty_per_object)
     {
-        unstable_write_count = 0;
         autosync();
     }
     if (next_op)
@@ -543,13 +543,17 @@ lazy:
             for (auto & chunk: loc_set)
             {
                 this->dirty_osds.insert(chunk.osd_num);
-                this->unstable_writes[(osd_object_id_t){
+                auto & unst = this->unstable_writes[(osd_object_id_t){
                     .osd_num = chunk.osd_num,
                     .oid = {
                         .inode = op_data->oid.inode,
                         .stripe = op_data->oid.stripe | chunk.role,
                     },
-                }] = op_data->fact_ver;
+                }];
+                unst.latest_ver = op_data->fact_ver;
+                unst.ver_count++;
+                if (unstable_per_object < unst.ver_count)
+                    unstable_per_object = unst.ver_count;
             }
         }
         else