diff --git a/src/client/cluster_client_impl.h b/src/client/cluster_client_impl.h index a3b53343..7fdd0509 100644 --- a/src/client/cluster_client_impl.h +++ b/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); diff --git a/src/client/cluster_client_wb.cpp b/src/client/cluster_client_wb.cpp index 34b4a548..a75964fe 100644 --- a/src/client/cluster_client_wb.cpp +++ b/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); - writeback_queue_size--; + bool lm = is_left_merged(dirty_it); + 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) { // is OK to contain more than actual number of consecutive // requests as long as it doesn't miss anything. But @@ -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) { diff --git a/src/test/test_cluster_client.cpp b/src/test/test_cluster_client.cpp index 9dc00c6f..6f754b18 100644 --- a/src/test/test_cluster_client.cpp +++ b/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; }