diff --git a/src/blockstore/blockstore_flush.cpp b/src/blockstore/blockstore_flush.cpp index 47e55c67..ab1a2512 100644 --- a/src/blockstore/blockstore_flush.cpp +++ b/src/blockstore/blockstore_flush.cpp @@ -361,6 +361,10 @@ bool journal_flusher_co::loop() else if (wait_state == 28) goto resume_28; else if (wait_state == 29) goto resume_29; else if (wait_state == 30) goto resume_30; + else if (wait_state == 31) goto resume_31; + else if (wait_state == 32) goto resume_32; + else if (wait_state == 33) goto resume_33; + resume_0: if (flusher->flush_queue.size() < flusher->min_flusher_count && !flusher->trim_wanted || !flusher->flush_queue.size() || !flusher->dequeuing) @@ -486,13 +490,14 @@ resume_2: resume_10: resume_11: resume_12: + resume_13: if (fill_incomplete && !clear_incomplete_csum_block_bits(5)) return false; // Wait for journal data reads if the journal is not inmemory - resume_13: + resume_14: if (wait_journal_count > 0) { - wait_state = wait_base+13; + wait_state = wait_base+14; return false; } if (bs->dsk.csum_block_size) @@ -509,7 +514,7 @@ resume_2: { if (it->copy_flags == COPY_BUF_JOURNAL || it->copy_flags == (COPY_BUF_JOURNAL|COPY_BUF_COALESCED)) { - await_sqe(14); + await_sqe(15); data->iov = (struct iovec){ it->buf, (size_t)it->len }; data->callback = simple_callback_w; my_uring_prep_writev( @@ -519,23 +524,25 @@ resume_2: } } // Wait for data writes and metadata reads - resume_15: resume_16: - if (!wait_meta_reads(15)) + resume_17: + if (!wait_meta_reads(16)) return false; // Sync data before writing metadata - resume_17: resume_18: resume_19: - if (copy_count && !fsync_batch(false, 17)) + resume_20: + if (copy_count && !fsync_batch(false, 18)) return false; - // Modify the new metadata entry - update_metadata_entry(); - // Update clean_db - it must be equal to the metadata entry - update_clean_db(); - // And write metadata entries if (old_clean_loc != UINT64_MAX && old_clean_loc != clean_loc) { + resume_21: + inflight_meta_sector = flusher->inflight_meta_sectors.find(meta_old.sector); + if (inflight_meta_sector != flusher->inflight_meta_sectors.end()) + { + wait_state = wait_base+20; + return false; + } // zero out old metadata entry { clean_disk_entry *old_entry = (clean_disk_entry*)((uint8_t*)meta_old.buf + meta_old.pos*bs->dsk.clean_entry_size); @@ -548,26 +555,50 @@ resume_2: } } memset((uint8_t*)meta_old.buf + meta_old.pos*bs->dsk.clean_entry_size, 0, bs->dsk.clean_entry_size); - resume_20: - if (meta_old.sector != meta_new.sector && !write_meta_block(meta_old, 20)) - return false; - } - resume_21: - if (!write_meta_block(meta_new, 21)) - return false; + if (meta_old.sector != meta_new.sector) + { + flusher->inflight_meta_sectors.insert(meta_old.sector); resume_22: + if (!write_meta_block(meta_old, 22)) + return false; + resume_23: + if (wait_count > 0) + { + wait_state = wait_base+23; + return false; + } + flusher->inflight_meta_sectors.erase(meta_old.sector); + } + } + resume_24: + inflight_meta_sector = flusher->inflight_meta_sectors.find(meta_new.sector); + if (inflight_meta_sector != flusher->inflight_meta_sectors.end()) { + wait_state = wait_base+24; + return false; + } + flusher->inflight_meta_sectors.insert(meta_new.sector); + // Modify the new metadata entry + update_metadata_entry(); + // Update clean_db - it must be equal to the metadata entry + update_clean_db(); + // And write metadata entries + resume_25: + if (!write_meta_block(meta_new, 25)) + return false; + resume_26: if (wait_count > 0) { - wait_state = wait_base+22; + wait_state = wait_base+26; return false; } + flusher->inflight_meta_sectors.erase(meta_new.sector); // Done, free all buffers free_buffers(); // And sync metadata (in batches - not per each operation!) - resume_23: - resume_24: - resume_25: - if (!fsync_batch(true, 23)) + resume_27: + resume_28: + resume_29: + if (!fsync_batch(true, 27)) return false; // Free the data block only when metadata is synced free_data_blocks(); @@ -590,12 +621,12 @@ resume_2: if (bs->journal_trim_interval && !((++flusher->journal_trim_counter) % bs->journal_trim_interval) || flusher->trim_wanted > 0) { - resume_26: - resume_27: - resume_28: - resume_29: resume_30: - if (!trim_journal(26)) + resume_31: + resume_32: + resume_33: + resume_34: + if (!trim_journal(30)) return false; } // All done @@ -734,6 +765,7 @@ bool journal_flusher_co::clear_incomplete_csum_block_bits(int wait_base) else if (wait_state == wait_base+5) goto resume_5; else if (wait_state == wait_base+6) goto resume_6; else if (wait_state == wait_base+7) goto resume_7; + else if (wait_state == wait_base+8) goto resume_8; cleared_incomplete = false; for (auto it = v.begin(); it != v.end(); it++) { @@ -754,11 +786,19 @@ bool journal_flusher_co::clear_incomplete_csum_block_bits(int wait_base) if (!wait_meta_reads(wait_base+0)) return false; resume_2: - if (wait_journal_count > 0) + inflight_meta_sector = flusher->inflight_meta_sectors.find(meta_new.sector); + if (inflight_meta_sector != flusher->inflight_meta_sectors.end()) { wait_state = wait_base+2; return false; } + flusher->inflight_meta_sectors.insert(meta_new.sector); + resume_3: + if (wait_journal_count > 0) + { + wait_state = wait_base+3; + return false; + } // Verify data checksums for (i = v.size()-1; i >= 0 && (v[i].copy_flags & COPY_BUF_CSUM_FILL); i--) { @@ -837,19 +877,20 @@ bool journal_flusher_co::clear_incomplete_csum_block_bits(int wait_base) } } // Write and fsync the modified metadata entry - resume_3: - if (!write_meta_block(meta_new, wait_base+3)) - return false; resume_4: + if (!write_meta_block(meta_new, wait_base+4)) + return false; + resume_5: if (wait_count > 0) { - wait_state = wait_base+4; + wait_state = wait_base+5; return false; } - resume_5: + flusher->inflight_meta_sectors.erase(meta_new.sector); resume_6: resume_7: - if (!fsync_batch(true, wait_base+5)) + resume_8: + if (!fsync_batch(true, wait_base+6)) return false; } return true; diff --git a/src/blockstore/blockstore_flush.h b/src/blockstore/blockstore_flush.h index 1f4392a4..3741a986 100644 --- a/src/blockstore/blockstore_flush.h +++ b/src/blockstore/blockstore_flush.h @@ -76,6 +76,7 @@ class journal_flusher_co uint8_t *new_clean_bitmap; uint64_t new_trim_pos; + std::unordered_set::iterator inflight_meta_sector; friend class journal_flusher_t; void scan_dirty(); @@ -120,6 +121,7 @@ class journal_flusher_t std::map meta_sectors; std::deque flush_queue; std::map flush_versions; // FIXME: consider unordered_map? + std::unordered_set inflight_meta_sectors; bool try_find_older(std::map::iterator & dirty_end, obj_ver_id & cur); bool try_find_other(std::map::iterator & dirty_end, obj_ver_id & cur); diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index e3743542..0bea0cbc 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -19,6 +19,7 @@ #include #include #include +#include #include "cpp-btree/btree_map.h"