Store v1: Wake up the flusher when journal sectors are written (test_blockstore_chaos --seed 5 --config 12 --p-delete 30)

The flusher puts back the entries whose journal sector isn't written yet
and stops when it finds nothing else to flush. Writing the sector moves
dirty_start and makes them flushable, but nothing woke the flusher then:
when the writes waiting for journal space or for a deletion were the only
thing left, they hung forever, with the queue full of flushable entries
and a trim requested. Happens with immediate_commit=all too.

src/blockstore/v1/flush.cpp

@@ -127,6 +127,13 @@ void journal_flusher_t::loop()
         co[i].loop();
 }
 
+// Versions restart from 1 after a deletion, so the queued version of an object may be a flushed
+// delete above the one being queued now. The flusher would skip that version and never flush this one
+bool journal_flusher_t::is_gone(object_id oid, uint64_t version)
+{
+    return bs->dirty_db.find((obj_ver_id){ .oid = oid, .version = version }) == bs->dirty_db.end();
+}
+
 void journal_flusher_t::enqueue_flush(obj_ver_id ov)
 {
 #ifdef BLOCKSTORE_DEBUG
@@ -135,7 +142,7 @@ void journal_flusher_t::enqueue_flush(obj_ver_id ov)
     auto it = flush_versions.find(ov.oid);
     if (it != flush_versions.end())
     {
-        if (it->second < ov.version)
+        if (it->second < ov.version || is_gone(ov.oid, it->second))
             it->second = ov.version;
     }
     else
@@ -158,7 +165,7 @@ void journal_flusher_t::unshift_flush(obj_ver_id ov, bool force)
     auto it = flush_versions.find(ov.oid);
     if (it != flush_versions.end())
     {
-        if (it->second < ov.version)
+        if (it->second < ov.version || is_gone(ov.oid, it->second))
             it->second = ov.version;
     }
     else
@@ -231,6 +238,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<obj_ver_id> journal_flusher_t::find_unflushable_stable()
+{
+    std::vector<obj_ver_id> 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<obj_ver_id> blockstore_impl_t::find_unflushable_stable()
+{
+    return flusher->find_unflushable_stable();
+}
+
 void journal_flusher_t::dump_diagnostics()
 {
     const char *unflushable_type = "";

src/blockstore/v1/flush.h

@@ -120,6 +120,7 @@ class journal_flusher_t
     bool try_find_older(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
     bool try_find_other(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
     bool may_advance_except_one();
+    bool is_gone(object_id oid, uint64_t version);
 
 public:
     journal_flusher_t(blockstore_impl_t *bs);
@@ -132,9 +133,11 @@ 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);
     void remove_flush(object_id oid);
     void dump_diagnostics();
     bool is_mutated(uint64_t clean_loc);
+    std::vector<obj_ver_id> find_unflushable_stable();
 };

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<obj_ver_id> find_unflushable_stable();
+
     // Get diagnostic string for an operation
     std::string get_op_diag(blockstore_op_t *op);
 

src/blockstore/v1/write.cpp

@@ -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();
     }
 }
 

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<std::vector<uint8_t>> before(OBJ_COUNT);
         std::vector<uint64_t> 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 <n>         Steps per seed (default %ju)\n"
     "  --config <n>        Force configuration <n> instead of deriving it from the seed\n"
     "  --max-inflight <n>  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);