mds: write_head when reading in PurgeQueue

Previously write_head calls were only generated
on the write side, so if you had a big queue
and were just working through consuming it, you
wouldn't record your progress, and on a daemon
restart would end up repeating a load of work.

Signed-off-by: John Spray <john.spray@redhat.com>

diff --git a/src/mds/PurgeQueue.cc b/src/mds/PurgeQueue.cc
index 40ece7ecb5064a5c7c1b4b3e1351598ec7c2f41f..b8a59798ba07f72f6375d94fd25aca63b1e26f23 100644 (file)
--- a/src/mds/PurgeQueue.cc
+++ b/src/mds/PurgeQueue.cc
@@ -384,22 +384,38 @@ void PurgeQueue::execute_item_complete(
   if (iter == in_flight.begin()) {
     // This was the lowest journal position in flight, so we can now
     // safely expire the journal up to here.
+    dout(10) << "expiring to 0x" << std::hex << expire_to << std::dec << dendl;
     journaler.set_expire_pos(expire_to);
-    journaler.trim();
+  } else {
+    // This is completely fine, we're not supposed to purge files in
+    // order when doing them in parallel.
+    dout(10) << "non-sequential completion, not expiring anything" << dendl;
   }
 
-  dout(10) << "completed item for ino 0x" << std::hex << iter->second.ino
-           << std::dec << dendl;
-
   ops_in_flight -= _calculate_ops(iter->second);
   logger->set(l_pq_executing_ops, ops_in_flight);
 
+  dout(10) << "completed item for ino 0x" << std::hex << iter->second.ino
+           << std::dec << dendl;
+
   in_flight.erase(iter);
   logger->set(l_pq_executing, in_flight.size());
+  dout(10) << "in_flight.size() now " << in_flight.size() << dendl;
 
   logger->inc(l_pq_executed);
 
   _consume();
+
+  // Have we gone idle?  If so, do an extra write_head now instead of
+  // waiting for next flush after journaler_write_head_interval.
+  // Also do this periodically even if not idle, so that the persisted
+  // expire_pos doesn't fall too far behind our progress when consuming
+  // a very long queue.
+  if (in_flight.empty() || journaler.write_head_needed()) {
+    journaler.write_head(new FunctionContext([this](int r){
+          journaler.trim();
+          }));
+  }
 }
 
 void PurgeQueue::update_op_limit(const MDSMap &mds_map)