mds: basic split support

diff --git a/src/TODO b/src/TODO
index ee8dc6bfb43f41f264f48c2a5e846cc953173c75..038558188d2e9a5a8aee2dfd2447834b2f299b3a 100644 (file)
--- a/src/TODO
+++ b/src/TODO
@@ -226,6 +226,7 @@ todo
 
   - SnapRealm open_parents, get_snap_set need to recursively open/examine parents over given ranges...
   - realm split
+  - make realm split notifications safe from races from multiple mds's
 
   - figure out how to fix up rados logging
   - snap collections

diff --git a/src/client/Client.cc b/src/client/Client.cc
index 684fa2aa7fa2f9b1cd8986458c0ad704937dbdaa..5ed633c8de002cbf24c9633003a109552ae00a40 100644 (file)
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -1587,20 +1587,25 @@ void Client::handle_snap(MClientSnap *m)
 
   case CEPH_SNAP_OP_SPLIT:
     {
-      SnapRealm *newrealm = get_snap_realm(m->new_realm);
-      for (list<inodeno_t>::iterator p = m->new_inodes.begin();
-          p != m->new_inodes.end();
+      /*
+       * fixme: this blindly moves the inode from whatever its prior
+       *  realm is.  this needs to somehow be safe from races from
+       *  multiple mds's...
+       */
+      for (list<inodeno_t>::iterator p = m->split_inos.begin();
+          p != m->split_inos.end();
           p++) {
        if (inode_map.count(*p)) {
          Inode *in = inode_map[*p];
-         dout(10) << " moving " << *in << " into new realm " << m->new_realm << dendl;
+         dout(10) << " moving " << *in << " from old realm " << m->split_parent << dendl;
          if (in->snaprealm)
            put_snap_realm(in->snaprealm);
-         in->snaprealm = newrealm;
-         newrealm->nref++;
+         in->snaprealm = realm;
+         realm->nref++;
        }
       }
-      put_snap_realm(newrealm);
+      // oh.. update it too
+      maybe_update_snaprealm(realm, m->snap_highwater, m->snaps);
     }
     break;
 

diff --git a/src/mds/CInode.h b/src/mds/CInode.h
index 3157d17c171e0939365bebb92f1d31ebbb6cd05d..cf88a229026f1b4f02f374cb36ebec2bbfc49ebb 100644 (file)
--- a/src/mds/CInode.h
+++ b/src/mds/CInode.h
@@ -136,6 +136,8 @@ class CInode : public MDSCacheObject {
   fragtree_t       dirfragtree;  // dir frag tree, if any.  always consistent with our dirfrag map.
   SnapRealm        *snaprealm;
 
+  SnapRealm        *containing_realm;
+
   off_t last_journaled;       // log offset for the last time i was journaled
   off_t last_open_journaled;  // log offset for the last journaled EOpen
 
@@ -486,28 +488,45 @@ public:
     return 0;
   }
   Capability *add_client_cap(int client, CInode *in) {
-    if (client_caps.empty())
+    if (client_caps.empty()) {
       get(PIN_CAPS);
+      containing_realm = find_containing_snaprealm();
+      containing_realm->inodes_with_caps.push_back(&xlist_caps);
+    }
+
     assert(client_caps.count(client) == 0);
     Capability *cap = client_caps[client] = new Capability;
     cap->set_inode(in);
-    
-    SnapRealm *realm = find_containing_snaprealm();
-    realm->add_cap(client, cap);
+   
+    containing_realm->add_cap(client, cap);
     
     return cap;
   }
   void remove_client_cap(int client) {
     assert(client_caps.count(client) == 1);
 
-    Capability *cap = client_caps[client];
-    cap->realm->remove_cap(client, cap);
+    containing_realm->remove_cap(client, client_caps[client]);
 
     delete client_caps[client];
     client_caps.erase(client);
-    if (client_caps.empty())
+    if (client_caps.empty()) {
       put(PIN_CAPS);
+      xlist_caps.remove_myself();
+      containing_realm = NULL;
+    }
+  }
+  void move_to_containing_realm(SnapRealm *realm) {
+    for (map<int,Capability*>::iterator q = client_caps.begin();
+        q != client_caps.end();
+        q++) {
+      containing_realm->remove_cap(q->first, q->second);
+      realm->add_cap(q->first, q->second);
+    }
+    xlist_caps.remove_myself();
+    realm->inodes_with_caps.push_back(&xlist_caps);
+    containing_realm = realm;
   }
+
   Capability *reconnect_cap(int client, inode_caps_reconnect_t& icr) {
     Capability *cap = get_client_cap(client);
     if (cap) {

diff --git a/src/mds/Capability.h b/src/mds/Capability.h
index d3c91e892d6898ee686f31d84cdc4f4378242532..53776dfbd20cb5910b1da61e9a6223fd683545a7 100644 (file)
--- a/src/mds/Capability.h
+++ b/src/mds/Capability.h
@@ -30,7 +30,6 @@ using namespace std;
 
 
 class CInode;
-class SnapRealm;
 
 class Capability {
 public:
@@ -69,7 +68,6 @@ private:
 public:
   xlist<Capability*>::item session_caps_item;
 
-  SnapRealm *realm;
   xlist<Capability*>::item snaprealm_caps_item;
 
   Capability(CInode *i=0, int want=0, capseq_t s=0) :
@@ -80,7 +78,7 @@ public:
     last_open(0),
     mseq(0),
     suppress(false), stale(false),
-    session_caps_item(this), realm(0), snaprealm_caps_item(this) { }
+    session_caps_item(this), snaprealm_caps_item(this) { }
   
   capseq_t get_mseq() { return mseq; }
 

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index e026b4e476d3eff238fc8e6471a53145c60a0f07..ed2e6828a1ea27e8b85ccc26d3aaa6520639cf5c 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -4684,6 +4684,7 @@ void Server::handle_client_mksnap(MDRequest *mdr)
   dout(10) << " snapid is " << snapid << dendl;
 
   // create realm?
+  inodeno_t split_parent = 0;
   if (!diri->snaprealm) {
     dout(10) << "creating snaprealm on " << *diri << dendl;
     diri->open_snaprealm();
@@ -4700,8 +4701,9 @@ void Server::handle_client_mksnap(MDRequest *mdr)
     link.dirino = parent->inode->ino();
     diri->snaprealm->parents.insert(pair<snapid_t,snaplink_t>(CEPH_NOSNAP, link));
 
-    // split...
-    // ***
+    // split existing caps
+    parent->split_at(diri->snaprealm);
+    split_parent = parent->inode->ino();
   }
 
   // add the snap
@@ -4717,16 +4719,23 @@ void Server::handle_client_mksnap(MDRequest *mdr)
   vector<snapid_t> snaps;
   diri->snaprealm->get_snap_vector(snaps);
 
-  // notify clients
-  
+  // notify clients of update|split
+  list<inodeno_t> split_inos;
+  if (split_parent)
+    for (xlist<CInode*>::iterator p = diri->snaprealm->inodes_with_caps.begin(); !p.end(); ++p)
+      split_inos.push_back((*p)->ino());
+
   for (map<int, xlist<Capability*> >::iterator p = diri->snaprealm->client_caps.begin();
        p != diri->snaprealm->client_caps.end();
        p++) {
     assert(!p->second.empty());
 
-    MClientSnap *update = new MClientSnap(CEPH_SNAP_OP_UPDATE, diri->ino());
+    MClientSnap *update = new MClientSnap(split_parent ? CEPH_SNAP_OP_SPLIT:CEPH_SNAP_OP_UPDATE,
+                                         diri->ino());
     update->snaps = snaps;
     update->snap_highwater = diri->snaprealm->snap_highwater;
+    update->split_parent = split_parent;
+    update->split_inos = split_inos;
     mds->send_message_client(update, p->first);
   }
 

diff --git a/src/mds/snap.cc b/src/mds/snap.cc
index 864fd86a2b546b9863e44f99ad5a365ded94fe3a..8db7ec9fdf4067385a68cf33c6a2eaa787b36158 100644 (file)
--- a/src/mds/snap.cc
+++ b/src/mds/snap.cc
@@ -16,6 +16,7 @@
 #include "MDCache.h"
 #include "MDS.h"
 
+#include "messages/MClientSnap.h"
 
 /*
  * SnapRealm
@@ -91,3 +92,36 @@ void SnapRealm::get_snap_vector(vector<snapid_t> &v)
   for (set<snapid_t>::reverse_iterator p = s.rbegin(); p != s.rend(); p++)
     v[i++] = *p;
 }
+
+
+void SnapRealm::split_at(SnapRealm *child)
+{
+  dout(10) << "split_at " << *child << dendl;
+
+  xlist<CInode*>::iterator p = inodes_with_caps.begin();
+  while (!p.end()) {
+    CInode *in = *p;
+    ++p;
+
+    // does inode fall within the child realm?
+    CInode *t = in;
+    bool under_child = false;
+    while (t) {
+      t = in->get_parent_dn()->get_dir()->get_inode();
+      if (t == child->inode) {
+       under_child = true;
+       break;
+      }
+      if (t == inode)
+       break;
+    }
+    if (!under_child) {
+      dout(20) << " keeping " << *in << dendl;
+      continue;
+    }
+    
+    dout(20) << " child gets " << *in << dendl;
+    in->move_to_containing_realm(child);
+  }
+
+}

diff --git a/src/mds/snap.h b/src/mds/snap.h
index 66401fa002031321f08b4e88123b65b01ac63e44..cb7c1963cd2d1ec9a6d4b329b4af2f71317112b7 100644 (file)
--- a/src/mds/snap.h
+++ b/src/mds/snap.h
@@ -116,12 +116,12 @@ struct SnapRealm {
   void get_snap_set(set<snapid_t>& s);
   void get_snap_vector(vector<snapid_t>& s);
 
+  void split_at(SnapRealm *child);
+
   void add_cap(int client, Capability *cap) {
     client_caps[client].push_back(&cap->snaprealm_caps_item);
-    cap->realm = this;
   }
   void remove_cap(int client, Capability *cap) {
-    cap->realm = 0;
     cap->snaprealm_caps_item.remove_myself();
     if (client_caps[client].empty())
       client_caps.erase(client);

diff --git a/src/messages/MClientSnap.h b/src/messages/MClientSnap.h
index 38f3e7acb2e79cbde5c0e79661afe8d29df469df..6490b56bbafcfefa91005d797ae917d4308f8094 100644 (file)
--- a/src/messages/MClientSnap.h
+++ b/src/messages/MClientSnap.h
@@ -35,18 +35,21 @@ struct MClientSnap : public Message {
   vector<snapid_t> snaps;
 
   // (for split only)
-  inodeno_t new_realm;
-  list<inodeno_t> new_inodes;
+  inodeno_t split_parent;
+  list<inodeno_t> split_inos;
   
   MClientSnap() : Message(CEPH_MSG_CLIENT_SNAP) {}
   MClientSnap(int o, inodeno_t r) : 
     Message(CEPH_MSG_CLIENT_SNAP),
-    op(o), realm(r) {} 
+    op(o), realm(r),
+    split_parent(0) {} 
   
   const char *get_type_name() { return "Csnap"; }
   void print(ostream& out) {
     out << "client_snap(" << get_opname(op) << " " << realm
        << " " << snaps;
+    if (split_parent)
+      out << " split_parent=" << split_parent;
     out << ")";
   }
 
@@ -55,8 +58,8 @@ struct MClientSnap : public Message {
     ::encode(realm, payload);
     ::encode(snap_highwater, payload);
     ::encode(snaps, payload);
-    ::encode(new_realm, payload);
-    ::encode(new_inodes, payload);
+    ::encode(split_parent, payload);
+    ::encode(split_inos, payload);
   }
   void decode_payload() {
     bufferlist::iterator p = payload.begin();
@@ -64,8 +67,8 @@ struct MClientSnap : public Message {
     ::decode(realm, p);
     ::decode(snap_highwater, p);
     ::decode(snaps, p);
-    ::decode(new_realm, p);
-    ::decode(new_inodes, p);
+    ::decode(split_parent, p);
+    ::decode(split_inos, p);
     assert(p.end());
   }