osd: fix race when queuing recovery ops

Previously we would sample how many ops to start under the lock, drop it,
and start that many.  This is racy because multiple threads can jump in
and we start too many ops.  Instead, claim as many slots as we can and
release them back later if we do not end up using them.

Take care to re-wake the work-queue since we are releasing more resources
for wq use.

Signed-off-by: Sage Weil <sage@inktank.com>
Reviewed-by: Samuel Just <sam.just@inktank.com>

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index bab345324d0254e77a39207b7389a125881419dd..c3645ab49e7ff06337d94bd3a3c12e6efd34ef4c 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -6495,29 +6495,35 @@ void OSD::do_recovery(PG *pg)
   // see how many we should try to start.  note that this is a bit racy.
   recovery_wq.lock();
   int max = g_conf->osd_recovery_max_active - recovery_ops_active;
+  if (max > 0) {
+    dout(10) << "do_recovery can start " << max << " (" << recovery_ops_active << "/" << g_conf->osd_recovery_max_active
+            << " rops)" << dendl;
+    recovery_ops_active += max;  // take them now, return them if we don't use them.
+  } else {
+    dout(10) << "do_recovery can start 0 (" << recovery_ops_active << "/" << g_conf->osd_recovery_max_active
+            << " rops)" << dendl;
+  }
   recovery_wq.unlock();
+
   if (max <= 0) {
     dout(10) << "do_recovery raced and failed to start anything; requeuing " << *pg << dendl;
     recovery_wq.queue(pg);
+    return;
   } else {
     pg->lock();
     if (pg->deleting || !(pg->is_active() && pg->is_primary())) {
       pg->unlock();
-      return;
+      goto out;
     }
     
-    dout(10) << "do_recovery starting " << max
-            << " (" << recovery_ops_active << "/" << g_conf->osd_recovery_max_active << " rops) on "
-            << *pg << dendl;
+    dout(10) << "do_recovery starting " << max << " " << *pg << dendl;
 #ifdef DEBUG_RECOVERY_OIDS
     dout(20) << "  active was " << recovery_oids[pg->info.pgid] << dendl;
 #endif
     
     PG::RecoveryCtx rctx = create_context();
     int started = pg->start_recovery_ops(max, &rctx);
-    dout(10) << "do_recovery started " << started
-            << " (" << recovery_ops_active << "/" << g_conf->osd_recovery_max_active << " rops) on "
-            << *pg << dendl;
+    dout(10) << "do_recovery started " << started << "/" << max << " on " << *pg << dendl;
 
     /*
      * if we couldn't start any recovery ops and things are still
@@ -6540,6 +6546,15 @@ void OSD::do_recovery(PG *pg)
     pg->unlock();
     dispatch_context(rctx, pg, curmap);
   }
+
+ out:
+  recovery_wq.lock();
+  if (max > 0) {
+    assert(recovery_ops_active >= max);
+    recovery_ops_active -= max;
+  }
+  recovery_wq._wake();
+  recovery_wq.unlock();
 }
 
 void OSD::start_recovery_op(PG *pg, const hobject_t& soid)