PG,ReplicatedPG: allow multiple flushes to be in progress

Shortly, we won't know precisely when to flush, so Strays
may need to flush twice.

Signed-off-by: Samuel Just <sam.just@inktank.com>

diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 81b9e2e6a755cb5ebe8eb5b492615e4e8544d22b..1e1adbbfc5e2c8e8209f707aca9570895cf2be7c 100644 (file)
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -173,7 +173,7 @@ PG::PG(OSDService *o, OSDMapRef curmap,
   backfill_target(-1),
   backfill_reserved(0),
   backfill_reserving(0),
-  flushed(false),
+  flushes_in_progress(0),
   pg_stats_publish_lock("PG::pg_stats_publish_lock"),
   pg_stats_publish_valid(false),
   osr(osd->osr_registry.lookup_or_create(p, (stringify(p)))),
@@ -4448,7 +4448,7 @@ void PG::start_flush(ObjectStore::Transaction *t,
   FlushStateRef flush_trigger(
     new FlushState(this, get_osdmap()->get_epoch()));
   t->nop();
-  assert(!flushed);
+  flushes_in_progress++;
   on_applied->push_back(new ContainerContext<FlushStateRef>(flush_trigger));
   on_safe->push_back(new ContainerContext<FlushStateRef>(flush_trigger));
 }
@@ -5149,7 +5149,7 @@ PG::RecoveryState::Reset::Reset(my_context ctx)
 {
   context< RecoveryMachine >().log_enter(state_name);
   PG *pg = context< RecoveryMachine >().pg;
-  pg->flushed = false;
+  pg->flushes_in_progress = 0;
   pg->set_last_peering_reset();
 }
 
@@ -6826,7 +6826,7 @@ PG::RecoveryState::WaitFlushedPeering::WaitFlushedPeering(my_context ctx)
 {
   PG *pg = context< RecoveryMachine >().pg;
   context< RecoveryMachine >().log_enter(state_name);
-  if (context< RecoveryMachine >().pg->flushed)
+  if (context< RecoveryMachine >().pg->flushes_in_progress == 0)
     post_event(Activate(pg->get_osdmap()->get_epoch()));
 }
 

diff --git a/src/osd/PG.h b/src/osd/PG.h
index a1f1d2c1a69ca9a184c57db4f89ce1d355d6b840..06fa042dc8148c36aaddcce63848ba5bcbf8e5d0 100644 (file)
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -524,7 +524,7 @@ protected:
 
 
   // pg waiters
-  bool flushed;
+  unsigned flushes_in_progress;
 
   // Ops waiting on backfill_pos to change
   list<OpRequestRef>            waiting_for_active;

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index e957a415f0c734f9deb6d8c1a3b338b6c278abce..e6a85bf62101ae1c78a20bfe948396eb1a75d5e4 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -836,8 +836,10 @@ void ReplicatedPG::do_request(
   if (can_discard_request(op)) {
     return;
   }
-  if (!flushed) {
-    dout(20) << " !flushed, waiting for active on " << op << dendl;
+  if (flushes_in_progress > 0) {
+    dout(20) << flushes_in_progress
+            << " flushes_in_progress pending "
+            << "waiting for active on " << op << dendl;
     waiting_for_active.push_back(op);
     return;
   }
@@ -7296,14 +7298,19 @@ void ReplicatedPG::apply_and_flush_repops(bool requeue)
 
 void ReplicatedPG::on_flushed()
 {
-  pair<hobject_t, ObjectContextRef> i;
-  while (object_contexts.get_next(i.first, &i)) {
-    derr << "on_flushed: object " << i.first << " obc still alive" << dendl;
+  assert(flushes_in_progress > 0);
+  flushes_in_progress--;
+  if (flushes_in_progress == 0) {
+    requeue_ops(waiting_for_active);
+  }
+  if (!is_active() || !is_primary()) {
+    pair<hobject_t, ObjectContextRef> i;
+    while (object_contexts.get_next(i.first, &i)) {
+      derr << "on_flushed: object " << i.first << " obc still alive" << dendl;
+    }
+    assert(object_contexts.empty());
   }
-  assert(object_contexts.empty());
   pgbackend->on_flushed();
-  flushed = true;
-  requeue_ops(pg->waiting_for_active);
 }
 
 void ReplicatedPG::on_removal(ObjectStore::Transaction *t)