diff --git a/src/blockstore/v1/flush.cpp b/src/blockstore/v1/flush.cpp index 8832952d..a555770a 100644 --- a/src/blockstore/v1/flush.cpp +++ b/src/blockstore/v1/flush.cpp @@ -176,6 +176,22 @@ void journal_flusher_t::unshift_flush(obj_ver_id ov, bool force) } } +// If the object is being flushed, let that flush requeue when it finishes. Queueing it now +// would leave it in the queue after the flush erases it, and it would hide newer versions +bool journal_flusher_t::requeue_after_flush(obj_ver_id ov) +{ + auto repeat_it = sync_to_repeat.find(ov.oid); + if (repeat_it == sync_to_repeat.end()) + { + return false; + } + if (repeat_it->second < ov.version) + { + repeat_it->second = ov.version; + } + return true; +} + void journal_flusher_t::remove_flush(object_id oid) { #ifdef BLOCKSTORE_DEBUG @@ -231,6 +247,50 @@ void journal_flusher_t::release_trim() trim_wanted--; } +// The flusher puts back entries whose journal sector isn't written yet and stops if it finds +// nothing else. Writing the sector makes them flushable, and nothing else may wake it then +void journal_flusher_t::wakeup_on_journal_written() +{ + if (!dequeuing && flush_queue.size() && (flush_queue.size() >= flusher_start_threshold || trim_wanted > 0)) + { + dequeuing = true; + bs->ringloop->wakeup(); + } +} + +// An object's newest stable version is reachable when the object is being flushed, or when +// it's queued under a version that is still in dirty_db and not older than that one: the +// flusher skips a queued version it can't find, and flushes only up to the queued version +std::vector journal_flusher_t::find_unflushable_stable() +{ + std::vector lost; + auto & dirty_db = bs->dirty_db; + auto it = dirty_db.begin(); + while (it != dirty_db.end()) + { + object_id oid = it->first.oid; + uint64_t stable_ver = 0; + for (; it != dirty_db.end() && it->first.oid == oid; it++) + { + if (IS_STABLE(it->second.state)) + stable_ver = it->first.version; + } + if (!stable_ver || sync_to_repeat.find(oid) != sync_to_repeat.end()) + continue; + auto fv = flush_versions.find(oid); + if (fv != flush_versions.end() && fv->second >= stable_ver && + dirty_db.find((obj_ver_id){ .oid = oid, .version = fv->second }) != dirty_db.end()) + continue; + lost.push_back((obj_ver_id){ .oid = oid, .version = stable_ver }); + } + return lost; +} + +std::vector blockstore_impl_t::find_unflushable_stable() +{ + return flusher->find_unflushable_stable(); +} + void journal_flusher_t::dump_diagnostics() { const char *unflushable_type = ""; diff --git a/src/blockstore/v1/flush.h b/src/blockstore/v1/flush.h index 06a6f6bc..201ff267 100644 --- a/src/blockstore/v1/flush.h +++ b/src/blockstore/v1/flush.h @@ -132,9 +132,12 @@ public: void mark_trim_possible(); void request_trim(); void release_trim(); + void wakeup_on_journal_written(); void enqueue_flush(obj_ver_id oid); void unshift_flush(obj_ver_id oid, bool force); + bool requeue_after_flush(obj_ver_id ov); void remove_flush(object_id oid); void dump_diagnostics(); bool is_mutated(uint64_t clean_loc); + std::vector find_unflushable_stable(); }; diff --git a/src/blockstore/v1/impl.h b/src/blockstore/v1/impl.h index 0e67640a..df3d3825 100644 --- a/src/blockstore/v1/impl.h +++ b/src/blockstore/v1/impl.h @@ -346,6 +346,9 @@ public: // Print diagnostics to stdout void dump_diagnostics(); + // Stable dirty versions that no flush will ever reach. Scans all of dirty_db: for tests only + std::vector find_unflushable_stable(); + // Get diagnostic string for an operation std::string get_op_diag(blockstore_op_t *op); diff --git a/src/blockstore/v1/write.cpp b/src/blockstore/v1/write.cpp index c00c75ec..b5c14eec 100644 --- a/src/blockstore/v1/write.cpp +++ b/src/blockstore/v1/write.cpp @@ -353,7 +353,7 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op) void blockstore_impl_t::unshift_stable_flush(obj_ver_id ov) { auto dirty_it = dirty_db.find(ov); - if (dirty_it != dirty_db.end() && IS_STABLE(dirty_it->second.state)) + if (dirty_it != dirty_db.end() && IS_STABLE(dirty_it->second.state) && !flusher->requeue_after_flush(ov)) { flusher->unshift_flush(ov, true); } @@ -925,6 +925,7 @@ void blockstore_impl_t::mark_not_dirty(int journal_sector) (new_ds + (new_ds >= journal.used_start ? 0 : journal.len))) { journal.dirty_start = new_ds; + flusher->wakeup_on_journal_written(); } } diff --git a/src/test/test_blockstore_chaos.cpp b/src/test/test_blockstore_chaos.cpp index b0b66990..72676e45 100644 --- a/src/test/test_blockstore_chaos.cpp +++ b/src/test/test_blockstore_chaos.cpp @@ -66,6 +66,8 @@ struct chaos_probs_t }; static chaos_probs_t probs; +// Check the store's internal state after every drain, which takes full scans of it +static bool hardened = false; // What a single step decided to do enum chaos_action_t @@ -114,6 +116,28 @@ static void dump_extra(v1::blockstore_impl_t *bs) { } +// A stable version no flush reaches pins the journal for good, and the store only shows that +// as writes refused with EAGAIN - expected under pressure - so --hardened checks it explicitly +static void check_flush_liveness(blockstore_impl_t *bs, uint64_t step_no) +{ +} + +static void check_flush_liveness(v1::blockstore_impl_t *bs, uint64_t step_no) +{ + if (!hardened) + return; + auto lost = bs->find_unflushable_stable(); + if (!lost.size()) + return; + fprintf(stderr, "step %ju: %zu stable version(s) will never be flushed:", step_no, lost.size()); + for (auto & ov: lost) + fprintf(stderr, " %jx:%jx v%ju", ov.oid.inode, ov.oid.stripe, ov.version); + fprintf(stderr, "\n"); + bs->dump_diagnostics(); + fflush(stdout); + abort(); +} + // What we believe about one object. // // `submitted` is the highest version we ever enqueued a write for - not necessarily @@ -1052,6 +1076,7 @@ struct chaos_t void do_restart() { drain(); + check_flush_liveness(bs, step_no); restarts++; std::vector> before(OBJ_COUNT); std::vector before_ver(OBJ_COUNT, 0); @@ -1201,6 +1226,7 @@ struct chaos_t step(); // Everything still in flight must complete, and everything durable must still be there drain(); + check_flush_liveness(bs, step_no); for (int obj = 0; obj < OBJ_COUNT; obj++) check_object(obj); } @@ -1283,6 +1309,9 @@ static const char *help_text = " --steps Steps per seed (default %ju)\n" " --config Force configuration instead of deriving it from the seed\n" " --max-inflight Operations in flight at once (default %u)\n" + " --hardened After every drain, also check the store's internal state, which\n" + " takes full scans of it: in store v1, that every stable version\n" + " gets flushed\n" " -v, --trace Trace every operation and every simulated disk request\n" " -h, --help Show this help\n" "\n" @@ -1383,6 +1412,8 @@ int main(int narg, char *args[]) } else if (!strcmp(opt, "-v") || !strcmp(opt, "--trace")) trace = true; + else if (!strcmp(opt, "--hardened")) + hardened = true; else if (!strcmp(opt, "--seed")) { take(v);