Fix writeback incorrectly calculating queue size which was leading to client hangs

src/client/cluster_client_impl.h

@@ -45,7 +45,6 @@ public:
     dirty_buf_it_t find_dirty(uint64_t inode, uint64_t offset);
     bool is_left_merged(dirty_buf_it_t dirty_it);
     bool is_right_merged(dirty_buf_it_t dirty_it);
-    bool is_merged(const dirty_buf_it_t & dirty_it);
     void copy_write(cluster_op_t *op, int state, uint64_t new_flush_id = 0);
     int repeat_ops_for(cluster_client_t *cli, osd_num_t peer_osd, pool_id_t pool_id, pg_num_t pg_num);
     void start_writebacks(cluster_client_t *cli, int count);

src/client/cluster_client_wb.cpp

@@ -68,11 +68,6 @@ bool writeback_cache_t::is_right_merged(dirty_buf_it_t dirty_it)
     return false;
 }
 
-bool writeback_cache_t::is_merged(const dirty_buf_it_t & dirty_it)
-{
-    return is_left_merged(dirty_it) || is_right_merged(dirty_it);
-}
-
 void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flush_id)
 {
     // Save operation for replay when one of PGs goes out of sync
@@ -121,7 +116,7 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
                 if (dirty_it->second.state == CACHE_DIRTY)
                 {
                     writeback_bytes -= old_end - op->offset;
-                    if (is_left_merged(dirty_it) && !is_right_merged(dirty_it))
+                    if (is_right_merged(dirty_it))
                     {
                         writeback_queue_size++;
                     }
@@ -136,7 +131,7 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
             if (dirty_it->second.state == CACHE_DIRTY)
             {
                 writeback_bytes -= new_end - dirty_it->first.stripe;
-                if (!is_left_merged(dirty_it) && is_right_merged(dirty_it))
+                if (is_left_merged(dirty_it))
                 {
                     writeback_queue_size++;
                 }
@@ -158,11 +153,20 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
         else
         {
             // Remove the whole buffer
-            if (dirty_it->second.state == CACHE_DIRTY && !is_merged(dirty_it))
+            if (dirty_it->second.state == CACHE_DIRTY)
             {
                 writeback_bytes -= dirty_it->second.len;
-                assert(writeback_queue_size > 0);
+                bool lm = is_left_merged(dirty_it);
-                writeback_queue_size--;
+                bool rm = is_right_merged(dirty_it);
+                if (!lm && !rm)
+                {
+                    assert(writeback_queue_size > 0);
+                    writeback_queue_size--;
+                }
+                else if (lm && rm)
+                {
+                    writeback_queue_size++;
+                }
             }
             if (!--(*dirty_it->second.refcnt))
             {
@@ -190,7 +194,9 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
     {
         writeback_bytes += is_del ? 0 : op->len;
         // Track consecutive write-back operations
-        if (!is_merged(dirty_it))
+        bool lm = is_left_merged(dirty_it);
+        bool rm = is_right_merged(dirty_it);
+        if (!lm && !rm)
         {
             // <writeback_queue> is OK to contain more than actual number of consecutive
             // requests as long as it doesn't miss anything. But <writeback_queue_size>
@@ -201,6 +207,11 @@ void writeback_cache_t::copy_write(cluster_op_t *op, int state, uint64_t new_flu
                 .stripe = op->offset,
             });
         }
+        else if (lm && rm)
+        {
+            assert(writeback_queue_size > 0);
+            writeback_queue_size--;
+        }
     }
     if (!is_del)
     {

src/test/test_cluster_client.cpp

@@ -480,10 +480,76 @@ void test_writeback()
     printf("[ok] writeback test\n");
 }
 
+static void copy_write_for_test(writeback_cache_t *wb, uint64_t offset, uint64_t len, int state, uint64_t new_flush_id)
+{
+    void *buf = malloc_or_die(len);
+    cluster_op_t *op = new cluster_op_t();
+    op->opcode = OSD_OP_WRITE;
+    op->inode = 0x1000000000001;
+    op->offset = offset;
+    op->len = len;
+    op->iov.push_back(buf, len);
+    wb->copy_write(op, state, new_flush_id);
+    delete op;
+    free(buf);
+}
+
+void test_writeback_merge()
+{
+    writeback_cache_t *wb = new writeback_cache_t;
+    // [1000..3000]
+    copy_write_for_test(wb, 1000, 2000, CACHE_DIRTY, 0);
+    assert(wb->writeback_bytes == 2000);
+    assert(wb->writeback_queue_size == 1);
+    // [1000..3000][3000..4000]
+    copy_write_for_test(wb, 3000, 1000, CACHE_DIRTY, 0);
+    assert(wb->writeback_bytes == 3000);
+    assert(wb->writeback_queue_size == 1);
+    // [1000..3000][3000..4000], [5000..6000]
+    copy_write_for_test(wb, 5000, 1000, CACHE_DIRTY, 0);
+    assert(wb->writeback_bytes == 4000);
+    assert(wb->writeback_queue_size == 2);
+    // [1000..2500], [3500..4000], [5000..6000]
+    copy_write_for_test(wb, 2500, 1000, CACHE_WRITTEN, 0);
+    assert(wb->writeback_bytes == 3000);
+    assert(wb->writeback_queue_size == 3);
+    // [1000..2500], [3500..4000][4000...5000][5000..6000]
+    copy_write_for_test(wb, 4000, 1000, CACHE_DIRTY, 0);
+    assert(wb->writeback_bytes == 4000);
+    assert(wb->writeback_queue_size == 2);
+    // [1000..2500], [3500..4500][4500...5000][5000..6000]
+    copy_write_for_test(wb, 3500, 1000, CACHE_DIRTY, 0);
+    assert(wb->writeback_bytes == 4000);
+    assert(wb->writeback_queue_size == 2);
+    // [1000..2500], [3500..4500], [5000..6000]
+    copy_write_for_test(wb, 4500, 500, CACHE_WRITTEN, 0);
+    assert(wb->writeback_bytes == 3500);
+    assert(wb->writeback_queue_size == 3);
+    // [1000..2500][2500..3500][3500..4500], [5000..6000]
+    copy_write_for_test(wb, 2500, 1000, CACHE_DIRTY, 0);
+    assert(wb->writeback_bytes == 4500);
+    assert(wb->writeback_queue_size == 2);
+    // [1000..2500][2500..3500][3500..4500], [5500..6000]
+    copy_write_for_test(wb, 5000, 500, CACHE_WRITTEN, 0);
+    assert(wb->writeback_bytes == 4000);
+    assert(wb->writeback_queue_size == 2);
+    // [1000..2500][2500..3500][3500..4000], [5500..6000]
+    copy_write_for_test(wb, 4000, 1000, CACHE_WRITTEN, 0);
+    assert(wb->writeback_bytes == 3500);
+    assert(wb->writeback_queue_size == 2);
+    // [1000..2500][2500..3500][3500..4000][4000..5500][5500..6000]
+    copy_write_for_test(wb, 4000, 1500, CACHE_DIRTY, 0);
+    assert(wb->writeback_bytes == 5000);
+    assert(wb->writeback_queue_size == 1);
+    delete wb;
+    printf("[ok] writeback merge test\n");
+}
+
 int main(int narg, char *args[])
 {
     test1();
     test2();
     test_writeback();
+    test_writeback_merge();
     return 0;
 }