mds: implement enqueue scrub in MDCache

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

diff --git a/src/include/ceph_fs.h b/src/include/ceph_fs.h
index 08ef460bfe055626596e52c91d8603563f5551ac..1dd333e0f8224780ba63720e6e222f4cba6618d9 100644 (file)
--- a/src/include/ceph_fs.h
+++ b/src/include/ceph_fs.h
@@ -351,7 +351,8 @@ enum {
        CEPH_MDS_OP_FRAGMENTDIR= 0x01500,
        CEPH_MDS_OP_EXPORTDIR  = 0x01501,
        CEPH_MDS_OP_VALIDATE   = 0x01502,
-       CEPH_MDS_OP_FLUSH      = 0x01503
+       CEPH_MDS_OP_FLUSH      = 0x01503,
+       CEPH_MDS_OP_ENQUEUE_SCRUB = 0x01504
 };
 
 extern const char *ceph_mds_op_name(int op);

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 3ec852a97d136f909ec3a6a82943887219226fda..bf654735214ae82beff4dcacf15bc5c672bc7eb4 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -26,6 +26,7 @@
 #include "MDLog.h"
 #include "MDBalancer.h"
 #include "Migrator.h"
+#include "ScrubStack.h"
 
 #include "SnapClient.h"
 
@@ -8821,6 +8822,9 @@ void MDCache::dispatch_request(MDRequestRef& mdr)
     case CEPH_MDS_OP_VALIDATE:
       scrub_dentry_work(mdr);
       break;
+    case CEPH_MDS_OP_ENQUEUE_SCRUB:
+      enqueue_scrub_work(mdr);
+      break;
     case CEPH_MDS_OP_FLUSH:
       flush_dentry_work(mdr);
       break;
@@ -11674,6 +11678,21 @@ void MDCache::scrub_dentry(const string& path, Formatter *f, Context *fin)
   scrub_dentry_work(mdr);
 }
 
+/**
+ * The private data for an OP_ENQUEUE_SCRUB MDRequest
+ */
+class EnqueueScrubParams
+{
+  public:
+  const bool recursive;
+  const bool children;
+  const std::string tag;
+  EnqueueScrubParams(bool r, bool c, const std::string &tag_)
+    : recursive(r), children(c), tag(tag_)
+  {}
+};
+
+
 void MDCache::scrub_dentry_work(MDRequestRef& mdr)
 {
   set<SimpleLock*> rdlocks, wrlocks, xlocks;
@@ -11695,6 +11714,87 @@ void MDCache::scrub_dentry_work(MDRequestRef& mdr)
   return;
 }
 
+
+class C_ScrubEnqueued : public Context
+{
+public:
+  MDRequestRef mdr;
+  Context *on_finish;
+  Formatter *formatter;
+  C_ScrubEnqueued(MDRequestRef& mdr,
+                  Context *fin, Formatter *f) :
+    mdr(mdr), on_finish(fin), formatter(f) {}
+
+  void finish(int r) {
+#if 0
+    if (r >= 0) { // we got into the scrubbing dump it
+      results.dump(formatter);
+    } else { // we failed the lookup or something; dump ourselves
+      formatter->open_object_section("results");
+      formatter->dump_int("return_code", r);
+      formatter->close_section(); // results
+    }
+#endif
+    on_finish->complete(r);
+  }
+};
+
+void MDCache::enqueue_scrub(const string& path, Formatter *f, Context *fin)
+{
+  dout(10) << "scrub_dentry " << path << dendl;
+  MDRequestRef mdr = request_start_internal(CEPH_MDS_OP_ENQUEUE_SCRUB);
+  filepath fp(path.c_str());
+  mdr->set_filepath(fp);
+
+  C_ScrubEnqueued *se = new C_ScrubEnqueued(mdr, fin, f);
+  mdr->internal_op_finish = se;
+  // TODO pass through tag/args
+  mdr->internal_op_private = new EnqueueScrubParams(true, true, "foobar");
+  enqueue_scrub_work(mdr);
+}
+
+void MDCache::enqueue_scrub_work(MDRequestRef& mdr)
+{
+  set<SimpleLock*> rdlocks, wrlocks, xlocks;
+  CInode *in = mds->server->rdlock_path_pin_ref(mdr, 0, rdlocks, true);
+  if (NULL == in)
+    return;
+
+  // TODO: Remove this restriction
+  assert(in->is_auth());
+
+  bool locked = mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks);
+  if (!locked)
+    return;
+
+  CDentry *dn = in->get_parent_dn();
+
+  // We got to this inode by path, so it must have a parent
+  assert(dn != NULL);
+
+  // Not setting a completion context here because we don't
+  // want to block asok caller on long running scrub
+  EnqueueScrubParams *args = static_cast<EnqueueScrubParams*>(
+      mdr->internal_op_private);
+  assert(args != NULL);
+
+  // Cannot scrub same dentry twice at same time
+  if (dn->scrub_info()->dentry_scrubbing) {
+    mds->server->respond_to_request(mdr, -EBUSY);
+    return;
+  }
+
+  mds->scrubstack->enqueue_dentry_bottom(dn, true, true, args->tag, NULL);
+  delete args;
+  mdr->internal_op_private = NULL;
+
+  // Successfully enqueued
+  mds->server->respond_to_request(mdr, 0);
+  return;
+}
+
+
+
 void MDCache::flush_dentry(const string& path, Context *fin)
 {
   if (is_readonly()) {

diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index c369acdbf8d8cc8bf5c97f5291d44f2e92f48e52..3158b9cc7a8f78a94afe772cde796281af62da8c 100644 (file)
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -1115,10 +1115,30 @@ public:
     while (n--) ++p;
     return p->second;
   }
+
   void scrub_dentry(const string& path, Formatter *f, Context *fin);
+  /**
+   * Scrub the named dentry only (skip the scrubstack)
+   */
   void scrub_dentry_work(MDRequestRef& mdr);
+
   void flush_dentry(const string& path, Context *fin);
   void flush_dentry_work(MDRequestRef& mdr);
+
+  /**
+   * Create and start an OP_ENQUEUE_SCRUB
+   */
+  void enqueue_scrub(const string& path, Formatter *f, Context *fin);
+
+  /**
+   * Resolve path to a dentry and pass it onto the ScrubStack.
+   *
+   * TODO: return enough information to the original mdr formatter
+   * and completion that they can subsequeuntly check the progress of
+   * this scrub (we won't block them on a whole scrub as it can take a very
+   * long time)
+   */
+  void enqueue_scrub_work(MDRequestRef& mdr);
 };
 
 class C_MDS_RetryRequest : public MDSInternalContext {