OSD: add pg deletion cancelation

DeletingState now allows _create_lock_pg() to attempt to cancel
pg deletion.

PG::init() must mark the PG as backfill iff we stopped a deletion.

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

1 files changed, 48 insertions, 8 deletions

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index bf630ec0ee3..14f88d5e7fc 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -1644,9 +1644,24 @@ PG *OSD::_create_lock_pg(
 
   PG *pg = _open_lock_pg(createmap, pgid, true, hold_map_lock);
 
-  t.create_collection(coll_t(pgid));
+  DeletingStateRef df = service.deleting_pgs.lookup(pgid);
+  bool backfill = false;
 
-  pg->init(role, up, acting, history, pi, &t);
+  if (df && df->try_stop_deletion()) {
+    dout(10) << __func__ << ": halted deletion on pg " << pgid << dendl;
+    backfill = true;
+    service.deleting_pgs.remove(pgid); // PG is no longer being removed!
+  } else {
+    if (df) {
+      // raced, ensure we don't see DeletingStateRef when we try to
+      // delete this pg
+      service.deleting_pgs.remove(pgid);
+    }
+    // either it's not deleting, or we failed to get to it in time
+    t.create_collection(coll_t(pgid));
+  }
+
+  pg->init(role, up, acting, history, pi, backfill, &t);
 
   dout(7) << "_create_lock_pg " << *pg << dendl;
   return pg;
@@ -2780,7 +2795,7 @@ void TestOpsSocketHook::test_ops(OSDService *service, ObjectStore *store,
 }
 
 // =========================================
-void remove_dir(
+bool remove_dir(
   ObjectStore *store, SnapMapper *mapper,
   OSDriver *osdriver,
   ObjectStore::Sequencer *osr,
@@ -2801,12 +2816,17 @@ void remove_dir(
     if (num >= g_conf->osd_target_transaction_size) {
       store->apply_transaction(osr, *t);
       delete t;
+      if (!dstate->check_canceled()) {
+	// canceled!
+	return false;
+      }
       t = new ObjectStore::Transaction;
       num = 0;
     }
   }
   store->apply_transaction(*t);
   delete t;
+  return true;
 }
 
 void OSD::RemoveWQ::_process(pair<PGRef, DeletingStateRef> item)
@@ -2817,10 +2837,22 @@ void OSD::RemoveWQ::_process(pair<PGRef, DeletingStateRef> item)
   coll_t coll = coll_t(pg->info.pgid);
   pg->osr->flush();
 
-  if (pg->have_temp_coll())
-    remove_dir(
+  if (!item.second->start_clearing())
+    return;
+
+  if (pg->have_temp_coll()) {
+    bool cont = remove_dir(
       store, &mapper, &driver, pg->osr.get(), pg->get_temp_coll(), item.second);
-  remove_dir(store, &mapper, &driver, pg->osr.get(), coll, item.second);
+    if (!cont)
+      return;
+  }
+  bool cont = remove_dir(
+    store, &mapper, &driver, pg->osr.get(), coll, item.second);
+  if (!cont)
+    return;
+
+  if (!item.second->start_deleting())
+    return;
 
   ObjectStore::Transaction *t = new ObjectStore::Transaction;
   PG::clear_info_log(
@@ -2831,11 +2863,19 @@ void OSD::RemoveWQ::_process(pair<PGRef, DeletingStateRef> item)
   if (pg->have_temp_coll())
     t->remove_collection(pg->get_temp_coll());
   t->remove_collection(coll);
+
+  // We need the sequencer to stick around until the op is complete
   store->queue_transaction(
     pg->osr.get(),
     t,
-    new ObjectStore::C_DeleteTransactionHolder<pair<PGRef, DeletingStateRef> >(
-      t, item));
+    0, // onapplied
+    0, // oncommit
+    0, // onreadable sync
+    new ObjectStore::C_DeleteTransactionHolder<PGRef>(
+      t, pg), // oncomplete
+    TrackedOpRef());
+
+  item.second->finish_deleting();
 }
 // =========================================