osd: Fix problems in ReplicatedPG::do_op() logic

Fix assert(is_degraded_object(soid)) in ReplicatedPG::wait_for_degraded_object()
  Use last_backfill_started as the backfill line
  Handle uncommon case of multi op source after backfill line and target before
  backfill line and !is_degraded_object().
  Include backfill line itself for before_backfill (<= instead of <)

Signed-off-by: David Zafman <david.zafman@inktank.com>

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 3c1715212ae97655e20cdef0028fad622d2a309a..274c9a7f3d11912c3f4c288d70ed6b09b82055dc 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -1165,12 +1165,14 @@ void ReplicatedPG::do_op(OpRequestRef op)
   // operation won't apply properly on the backfill_target.  (the
   // opposite is not a problem; if the target is after the line, we
   // don't apply on the backfill_target and it doesn't matter.)
+  // The last_backfill_started is used as the backfill line since
+  // that determines the boundary for writes.
   pg_info_t *backfill_target_info = NULL;
   int backfill_target = get_backfill_target();
   bool before_backfill = false;
   if (backfill_target >= 0) {
     backfill_target_info = &peer_info[backfill_target];
-    before_backfill = obc->obs.oi.soid < backfill_target_info->last_backfill;
+    before_backfill = obc->obs.oi.soid <= last_backfill_started;
   }
 
   // src_oids
@@ -1218,8 +1220,13 @@ void ReplicatedPG::do_op(OpRequestRef op)
                  << obc->obs.oi.soid << dendl;
            osd->reply_op_error(op, -EINVAL);
          } else if (is_degraded_object(sobc->obs.oi.soid) ||
-                  (before_backfill && sobc->obs.oi.soid > backfill_target_info->last_backfill)) {
-           wait_for_degraded_object(sobc->obs.oi.soid, op);
+                  (before_backfill && sobc->obs.oi.soid > last_backfill_started)) {
+           if (is_degraded_object(sobc->obs.oi.soid)) {
+             wait_for_degraded_object(sobc->obs.oi.soid, op);
+           } else {
+             waiting_for_degraded_object[sobc->obs.oi.soid].push_back(op);
+             op->mark_delayed("waiting for degraded object");
+           }
            dout(10) << " writes for " << obc->obs.oi.soid << " now blocked by "
                     << sobc->obs.oi.soid << dendl;
            obc->blocked_by = sobc;