import/export failure recovery basically working!

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@1153 29311d96-e01e-0410-9327-a35deaab8ce9

diff --git a/branches/sage/cephmds2/TODO b/branches/sage/cephmds2/TODO
index 229da68491a7bc6c5853e9f27ad8490dd4c56a0b..4b3c87e4c016000601c0a7c2ff3a0cb79c1c614e 100644 (file)
--- a/branches/sage/cephmds2/TODO
+++ b/branches/sage/cephmds2/TODO
@@ -16,6 +16,8 @@ doc
  
 
 mds
+- bystanders should avoid contacting auth when it is ambiguous.
+  - CDIR_WAIT_UNAMBIGUOUS?
 - openingdir pins should be handled by open_remote_dir, not explicitly by handle_export_dir_prep
 - bystander recovery from exporter failure
 - does inode need it's own replica list?  no?

diff --git a/branches/sage/cephmds2/mds/MDCache.cc b/branches/sage/cephmds2/mds/MDCache.cc
index c767c294f7d34c87d975a04f78ff323cdc4431ce..cc35db8ca8cb815ce9642685af6adee04892f03c 100644 (file)
--- a/branches/sage/cephmds2/mds/MDCache.cc
+++ b/branches/sage/cephmds2/mds/MDCache.cc
@@ -279,7 +279,8 @@ void MDCache::adjust_subtree_auth(CDir *dir, pair<int,int> auth)
   CDir *root;
   if (dir->ino() == 1) {
     root = dir;  // bootstrap hack.
-    subtrees[root].clear();
+    if (subtrees.count(root) == 0)
+      subtrees[root].clear();
   } else {
     root = get_subtree_root(dir);  // subtree root
   }
@@ -509,7 +510,6 @@ void MDCache::adjust_bounded_subtree_auth(CDir *dir, set<CDir*>& bounds, pair<in
     if (bounds.count(*p) == 0) {
       CDir *stray = *p;
       dout(10) << "  swallowing extra subtree at " << *stray << endl;
-      assert(stray->auth_is_ambiguous());
       adjust_subtree_auth(stray, auth);
       try_subtree_merge_at(stray);
     }
@@ -847,6 +847,39 @@ void MDCache::send_import_map_now(int who)
 }
 
 
+void MDCache::handle_mds_failure(int who)
+{
+  dout(7) << "handle_mds_failure mds" << who << endl;
+  
+  // adjust subtree auth
+  for (map<CDir*,set<CDir*> >::iterator p = subtrees.begin();
+       p != subtrees.end();
+       ++p) {
+    CDir *dir = p->first;
+    // only if we are a _bystander_.
+    if (dir->dir_auth.first == who &&
+       dir->dir_auth.second >= 0 &&
+       dir->dir_auth.second != mds->get_nodeid()) {
+      dout(7) << "disambiguating auth for " << *dir << endl;
+      adjust_subtree_auth(dir, dir->dir_auth.second);
+      try_subtree_merge(dir);
+    }
+    else if (dir->dir_auth.second == who &&
+            dir->dir_auth.first != mds->get_nodeid()) {
+      dout(7) << "disambiguating auth for " << *dir << endl;
+      adjust_subtree_auth(dir, dir->dir_auth.first);
+      try_subtree_merge(dir);
+    }      
+  }
+
+  // tell the migrator too.
+  migrator->handle_mds_failure(who);
+
+  show_subtrees();  
+}
+
+
+
 /*
  * during resolve state, we share import_maps to determine who
  * is authoritative for which trees.  we expect to get an import_map
@@ -895,12 +928,14 @@ void MDCache::handle_import_map(MMDSImportMap *m)
     }
   }
 
-  // note ambiguous imports too
-  for (map<inodeno_t, list<inodeno_t> >::iterator pi = m->ambiguous_imap.begin();
-       pi != m->ambiguous_imap.end();
-       ++pi) {
-    dout(10) << "noting ambiguous import on " << pi->first << " bounds " << pi->second << endl;
-    other_ambiguous_imports[from][pi->first].swap( pi->second );
+  // note ambiguous imports too.. unless i'm already active
+  if (!mds->is_active() && !mds->is_stopping()) {
+    for (map<inodeno_t, list<inodeno_t> >::iterator pi = m->ambiguous_imap.begin();
+        pi != m->ambiguous_imap.end();
+        ++pi) {
+      dout(10) << "noting ambiguous import on " << pi->first << " bounds " << pi->second << endl;
+      other_ambiguous_imports[from][pi->first].swap( pi->second );
+    }
   }
 
   show_subtrees();
@@ -2229,6 +2264,12 @@ int MDCache::path_traverse(filepath& origpath,
                                              want,
                                              true),  // need this dir too
                                cur->authority().first, MDS_PORT_CACHE);
+         if (cur->authority().second >= 0) 
+           mds->send_message_mds(new MDiscover(mds->get_nodeid(),
+                                               cur->ino(),
+                                               want,
+                                               true),  // need this dir too
+                                 cur->authority().second, MDS_PORT_CACHE);
         }
         cur->add_waiter(CINODE_WAIT_DIR, ondelay);
         if (onfinish) delete onfinish;
@@ -2380,11 +2421,11 @@ int MDCache::path_traverse(filepath& origpath,
     
     // MISS.  don't have it.
 
-    int dauth = cur->dir->dentry_authority( path[depth] ).first;
+    pair<int,int> dauth = cur->dir->dentry_authority( path[depth] );
     dout(12) << "traverse: miss on dentry " << path[depth] << " dauth " << dauth << " in " << *cur->dir << endl;
     
 
-    if (dauth == whoami) {
+    if (dauth.first == whoami) {
       // dentry is mine.
       if (cur->dir->is_complete()) {
         // file not found
@@ -2438,7 +2479,14 @@ int MDCache::path_traverse(filepath& origpath,
                                              cur->ino(),
                                              want,
                                              false),
-                               dauth, MDS_PORT_CACHE);
+                               dauth.first, MDS_PORT_CACHE);
+         if (dauth.second >= 0) 
+           mds->send_message_mds(new MDiscover(mds->get_nodeid(),
+                                               cur->ino(),
+                                               want,
+                                               false),
+                                 dauth.second, MDS_PORT_CACHE);
+
           if (mds->logger) mds->logger->inc("dis");
         }
         
@@ -2462,7 +2510,7 @@ int MDCache::path_traverse(filepath& origpath,
           req->clear_payload();  // reencode!
         }
 
-        mds->send_message_mds(req, dauth, req->get_dest_port());
+        mds->send_message_mds(req, dauth.first, req->get_dest_port());
         //show_imports();
         
         if (mds->logger) mds->logger->inc("cfw");

diff --git a/branches/sage/cephmds2/mds/MDCache.h b/branches/sage/cephmds2/mds/MDCache.h
index 410cc6585f3351843ab6c97fde9e0281ee5fa037..45ed437b1360d9dd966a3c420b854735058ff081 100644 (file)
--- a/branches/sage/cephmds2/mds/MDCache.h
+++ b/branches/sage/cephmds2/mds/MDCache.h
@@ -182,6 +182,7 @@ protected:
   set<int> got_import_map;     // nodes i need to send my import map to (when exports finish)
   set<int> rejoin_ack_gather;  // nodes i need a rejoin ack from
   
+  void handle_mds_failure(int who);
   void handle_import_map(MMDSImportMap *m);
   void handle_cache_rejoin(MMDSCacheRejoin *m);
   void handle_cache_rejoin_ack(MMDSCacheRejoinAck *m);

diff --git a/branches/sage/cephmds2/mds/MDS.cc b/branches/sage/cephmds2/mds/MDS.cc
index 9d06bea2f674568a031b19567241bbb6675987e3..1acb1781a9e5ca8acae4cee22366b351306f277c 100644 (file)
--- a/branches/sage/cephmds2/mds/MDS.cc
+++ b/branches/sage/cephmds2/mds/MDS.cc
@@ -573,7 +573,7 @@ void MDS::handle_mds_map(MMDSMap *m)
       // newly so?
       if (oldfailed.count(*p)) continue;      
 
-      mdcache->migrator->handle_mds_failure(*p);
+      mdcache->handle_mds_failure(*p);
     }
   }
 

diff --git a/branches/sage/cephmds2/mds/Migrator.cc b/branches/sage/cephmds2/mds/Migrator.cc
index 1b8adcc35db4e3ef35cde1e250e41600e6a345e7..b3875efa6ae8fb20373dffcfe789d7e4422241c2 100644 (file)
--- a/branches/sage/cephmds2/mds/Migrator.cc
+++ b/branches/sage/cephmds2/mds/Migrator.cc
@@ -252,13 +252,14 @@ void Migrator::handle_mds_failure(int who)
        export_state.erase(dir); // clean up
        break;
 
+       // NOTE: state order reversal, warning comes after loggingstart+prepping
       case EXPORT_WARNING:
        dout(10) << "export state=warning : unpinning bounds, unfreezing, notifying" << endl;
-       export_notify_abort(dir);   // tell peers about abort
+       //export_notify_abort(dir);   // tell peers about abort
 
        // fall-thru
 
-      case EXPORT_LOGGINGSTART:
+       //case EXPORT_LOGGINGSTART:
       case EXPORT_PREPPING:
        if (p->second != EXPORT_WARNING) 
          dout(10) << "export state=loggingstart|prepping : unpinning bounds, unfreezing" << endl;
@@ -356,7 +357,7 @@ void Migrator::handle_mds_failure(int who)
     if (import_peer[dirino] == who) {
       switch (import_state[dirino]) {
       case IMPORT_DISCOVERED:
-       dout(10) << "import state=discovered : unpinning " << *diri << endl;
+       dout(10) << "import state=discovered : unpinning inode " << *diri << endl;
        assert(diri);
        // unpin base
        diri->put(CInode::PIN_IMPORTING);
@@ -364,54 +365,35 @@ void Migrator::handle_mds_failure(int who)
        import_peer.erase(dirino);
        break;
 
-       // NOTE: state order reversal + fall-thru, pay attention.
+      case IMPORT_PREPPING:
+       if (import_state[dirino] == IMPORT_PREPPING) {
+         dout(10) << "import state=prepping : unpinning base+bounds " << *dir << endl;
+       }
+       assert(dir);
+       import_reverse_unpin(dir);    // unpin
+       break;
 
       case IMPORT_PREPPED:
-       dout(10) << "import state=prepping : unpinning base+bounds, unfreezing,  " << *dir << endl;
+       dout(10) << "import state=prepping : unpinning base+bounds, unfreezing " << *dir << endl;
        assert(dir);
        
-       // unfreeze
-       dir->unfreeze_tree();
-
        // adjust auth back to me
        cache->adjust_subtree_auth(dir, import_peer[dirino]);
        cache->try_subtree_merge(dir);
        
-       // FIXME what about bystanders
-
-       // fall-thru to unpin base+bounds
-
-      case IMPORT_PREPPING:
-       if (import_state[dirino] == IMPORT_PREPPING) {
-         dout(10) << "import state=prepping : unpinning base+bounds " << *dir << endl;
+       // bystanders?
+       if (import_bystanders[dir].empty()) {
+         import_reverse_unfreeze(dir);
+       } else {
+         // notify them; wait in aborting state
+         import_notify_abort(dir);
+         import_state[dirino] = IMPORT_ABORTING;
        }
-       assert(dir);
-
-       // unpin base
-       dir->put(CDir::PIN_IMPORTING);
-
-       // unpin bounds
-       for (set<CDir*>::iterator it = import_bounds[dir].begin();
-            it != import_bounds[dir].end();
-            it++) {
-         CDir *bd = *it;
-         assert(bd->state_test(CDir::STATE_IMPORTBOUND));
-         bd->state_clear(CDir::STATE_IMPORTBOUND);
-         bd->put(CDir::PIN_IMPORTBOUND);
-       }
-
-       import_state.erase(dirino);
-       import_peer.erase(dirino);
-       import_bound_inos.erase(dirino);
-       import_bounds.erase(dir);
        break;
 
-
       case IMPORT_LOGGINGSTART:
        dout(10) << "import state=loggingstart : reversing import on " << *dir << endl;
        import_reverse(dir);
-
-       // FIXME what about bystanders
        break;
 
       case IMPORT_ACKING:
@@ -419,6 +401,10 @@ void Migrator::handle_mds_failure(int who)
        dout(10) << "import state=acking : noting ambiguous import " << *dir << endl;
        cache->add_ambiguous_import(dir, import_bounds[dir]);
        break;
+
+      case IMPORT_ABORTING:
+       dout(10) << "import state=aborting : ignoring repeat failure " << *dir << endl;
+       break;
       }
     }
 
@@ -548,7 +534,7 @@ void Migrator::export_frozen(CDir *dir,
   assert(dir->is_frozen());
 
   // ok!
-  export_state[dir] = EXPORT_LOGGINGSTART;
+  //export_state[dir] = EXPORT_LOGGINGSTART;
 
   cache->show_imports();
 
@@ -561,6 +547,16 @@ void Migrator::export_frozen(CDir *dir,
   // generate prep message, log entry.
   MExportDirPrep *prep = new MExportDirPrep(dir->inode);
 
+  // include list of bystanders
+  for (map<int,int>::iterator p = dir->replicas_begin();
+       p != dir->replicas_end();
+       p++) {
+    if (p->first != dest) {
+      dout(10) << "bystander mds" << p->first << endl;
+      prep->add_bystander(p->first);
+    }
+  }
+
   // include spanning tree for all nested exports.
   // these need to be on the destination _before_ the final export so that
   // dir_auth updates on any nested exports are properly absorbed.
@@ -660,8 +656,9 @@ void Migrator::handle_export_prep_ack(MExportDirPrepAck *m)
 
     //mds->send_message_mds(new MExportDirWarning(dir->ino(), export_peer[dir]),
     //p->first, MDS_PORT_MIGRATOR);
-    MExportDirNotify *notify = new MExportDirNotify(dir->ino(), 
-                                                   mds->get_nodeid(), export_peer[dir]);
+    MExportDirNotify *notify = new MExportDirNotify(dir->ino(), true,
+                                                   pair<int,int>(mds->get_nodeid(),CDIR_AUTH_UNKNOWN),
+                                                   pair<int,int>(mds->get_nodeid(),export_peer[dir]));
     notify->copy_exports(export_bounds[dir]);
     mds->send_message_mds(notify, p->first, MDS_PORT_MIGRATOR);
     
@@ -1055,7 +1052,7 @@ void Migrator::export_reverse(CDir *dir)
   cache->process_delayed_expire(dir);
   
   // tell peers
-  export_notify_abort(dir);
+  //export_notify_abort(dir);
   
   // unfreeze
   dir->unfreeze_tree();
@@ -1069,6 +1066,7 @@ void Migrator::export_reverse(CDir *dir)
   cache->show_cache();
 }
 
+/*
 void Migrator::export_notify_abort(CDir* dir)
 {
   dout(10) << "export_notify_abort " << *dir << endl;
@@ -1077,12 +1075,14 @@ void Migrator::export_notify_abort(CDir* dir)
   for (set<int>::iterator p = export_notify_ack_waiting[dir].begin();
        p != export_notify_ack_waiting[dir].end();
        ++p) {
-    MExportDirNotify *notify = new MExportDirNotify(dir->ino(), 
-                                                   mds->get_nodeid(), CDIR_AUTH_UNKNOWN);
+    MExportDirNotify *notify = new MExportDirNotify(dir->ino(), false,
+                                                   pair<int,int>(mds->get_nodeid(), export_peer[dir]),
+                                                   pair<int,int>(mds->get_nodeid(), CDIR_AUTH_UNKNOWN));
     notify->copy_exports(export_bounds[dir]);
     mds->send_message_mds(notify, *p, MDS_PORT_MIGRATOR);
   }
 }
+*/
 
 /*
  * once i get the ack, and logged the EExportFinish(true),
@@ -1094,16 +1094,6 @@ void Migrator::export_logged_finish(CDir *dir)
   dout(7) << "export_logged_finish " << *dir << endl;
   dir->put(CDir::PIN_LOGGINGEXPORTFINISH);
 
-  if (mds->get_nodeid() == 0 && g_clock.now() > 20.0) assert(0); // hack fake death
-
-
-  if (export_state.count(dir) == 0||
-      export_state[dir] != EXPORT_LOGGINGFINISH) {
-    assert(0);  // this won't happen.
-    dout(7) << "target must have failed, not sending final commit message.  export succeeded anyway." << endl;
-    return;
-  }
-
   cache->verify_subtree_bounds(dir, export_bounds[dir]);
 
   // send notifies
@@ -1112,8 +1102,18 @@ void Migrator::export_logged_finish(CDir *dir)
   for (set<int>::iterator p = export_notify_ack_waiting[dir].begin();
        p != export_notify_ack_waiting[dir].end();
        ++p) {
-    MExportDirNotify *notify = new MExportDirNotify(dir->ino(), 
-                                                   dest, CDIR_AUTH_UNKNOWN);
+    MExportDirNotify *notify;
+    if (mds->mdsmap->is_active_or_stopping(export_peer[dir])) 
+      // dest is still alive.
+      notify = new MExportDirNotify(dir->ino(), true,
+                                   pair<int,int>(mds->get_nodeid(), dest),
+                                   pair<int,int>(dest, CDIR_AUTH_UNKNOWN));
+    else 
+      // dest is dead.  bystanders will think i am only auth, as per mdcache->handle_mds_failure()
+      notify = new MExportDirNotify(dir->ino(), true,
+                                   pair<int,int>(mds->get_nodeid(), CDIR_AUTH_UNKNOWN),
+                                   pair<int,int>(dest, CDIR_AUTH_UNKNOWN));
+
     notify->copy_exports(export_bounds[dir]);
     
     mds->send_message_mds(notify, *p, MDS_PORT_MIGRATOR);
@@ -1140,42 +1140,43 @@ void Migrator::handle_export_notify_ack(MExportDirNotifyAck *m)
   CInode *in = cache->get_inode(m->get_ino());
   CDir *dir = in ? in->dir : 0;
 
-  if (dir) {
-    dout(7) << "handle_export_notify_ack from " << m->get_source()
-           << " on " << *dir << endl;
-  } else {
-    dout(7) << "handle_export_notify_ack from " << m->get_source()
-           << " on dir " << m->get_ino() << endl;
-  }
-  
-  // aborted?
-  if (!dir ||
-      export_state.count(dir) == 0 ||
-      (export_state[dir] != EXPORT_NOTIFYING &&
-       export_state[dir] != EXPORT_WARNING)) {
-    assert(0);  // this won't happen.
-    dout(7) << "target must have failed, ignoring." << endl;
-    delete m;
-    return;
-  }
-
+  assert(dir);
   int from = m->get_source().num();
     
-  if (export_state[dir] == EXPORT_WARNING) {
-    // process warning.
+  if (export_state.count(dir) && export_state[dir] == EXPORT_WARNING) {
+    // exporting. process warning.
+    dout(7) << "handle_export_notify_ack from " << m->get_source()
+           << ": exporting, processing warning on "
+           << *dir << endl;
     assert(export_warning_ack_waiting.count(dir));
     export_warning_ack_waiting[dir].erase(from);
     
     if (export_warning_ack_waiting[dir].empty()) 
       export_go(dir);     // start export.
-  } else {
-    // process notify.
+  } 
+  else if (export_state.count(dir) && export_state[dir] == EXPORT_NOTIFYING) {
+    // exporting. process notify.
+    dout(7) << "handle_export_notify_ack from " << m->get_source()
+           << ": exporting, processing notify on "
+           << *dir << endl;
     assert(export_notify_ack_waiting.count(dir));
     export_notify_ack_waiting[dir].erase(from);
     
     if (export_notify_ack_waiting[dir].empty())
       export_finish(dir);
   }
+  else if (import_state.count(dir->ino()) && import_state[dir->ino()] == IMPORT_ABORTING) {
+    // reversing import
+    dout(7) << "handle_export_notify_ack from " << m->get_source()
+           << ": aborting import on "
+           << *dir << endl;
+    assert(import_bystanders[dir].count(from));
+    import_bystanders[dir].erase(from);
+    if (import_bystanders[dir].empty()) {
+      import_bystanders.erase(dir);
+      import_reverse_unfreeze(dir);
+    }
+  }
 
   delete m;
 }
@@ -1376,6 +1377,10 @@ void Migrator::handle_export_prep(MExportDirPrep *m)
     // change import state
     import_state[diri->ino()] = IMPORT_PREPPING;
     
+    // bystander list
+    import_bystanders[dir] = m->get_bystanders();
+    dout(7) << "bystanders are " << import_bystanders[dir] << endl;
+
     // assimilate traces to exports
     for (list<CInodeDiscover*>::iterator it = m->get_inodes().begin();
          it != m->get_inodes().end();
@@ -1521,7 +1526,7 @@ void Migrator::handle_export_dir(MExportDir *m)
   assert(dir->is_auth() == false);
 
   cache->show_imports();
-  
+
   // start the journal entry
   EImportStart *le = new EImportStart(dir->ino(), m->get_exports());
   le->metablob.add_dir_context(dir);
@@ -1592,18 +1597,6 @@ void Migrator::import_reverse(CDir *dir, bool fix_dir_auth)
 {
   dout(7) << "import_reverse " << *dir << endl;
 
-  // remove importing pin
-  dir->put(CDir::PIN_IMPORTING);
-
-  // remove bound pins
-  for (set<CDir*>::iterator it = import_bounds[dir].begin();
-       it != import_bounds[dir].end();
-       it++) {
-    CDir *bd = *it;
-    bd->put(CDir::PIN_IMPORTBOUND);
-    bd->state_clear(CDir::STATE_IMPORTBOUND);
-  }
-
   // update auth, with possible subtree merge.
   if (fix_dir_auth) {
     assert(dir->is_subtree_root());
@@ -1653,20 +1646,74 @@ void Migrator::import_reverse(CDir *dir, bool fix_dir_auth)
     }
   }
 
+  // log our failure
+  mds->mdlog->submit_entry(new EImportFinish(dir,false));      // log failure
+
+  // bystanders?
+  if (import_bystanders[dir].empty()) {
+    dout(7) << "no bystanders, finishing reverse now" << endl;
+    import_reverse_unfreeze(dir);
+  } else {
+    // notify them; wait in aborting state
+    dout(7) << "notifying bystanders of abort" << endl;
+    import_notify_abort(dir);
+    import_state[dir->ino()] = IMPORT_ABORTING;
+  }
+}
+
+void Migrator::import_notify_abort(CDir *dir)
+{
+  dout(7) << "import_notify_abort " << *dir << endl;
+  
+  for (set<int>::iterator p = import_bystanders[dir].begin();
+       p != import_bystanders[dir].end();
+       ++p) {
+    // NOTE: the bystander will think i am _only_ auth, because they will have seen
+    // the exporter's failure and updated the subtree auth.  see mdcache->handle_mds_failure().
+    MExportDirNotify *notify = 
+      new MExportDirNotify(dir->ino(), true,
+                          pair<int,int>(mds->get_nodeid(), CDIR_AUTH_UNKNOWN),
+                          pair<int,int>(import_peer[dir->ino()], CDIR_AUTH_UNKNOWN));
+    notify->copy_exports(import_bounds[dir]);
+    mds->send_message_mds(notify, *p, MDS_PORT_MIGRATOR);
+  }
+}
+
+void Migrator::import_reverse_unfreeze(CDir *dir)
+{
+  dout(7) << "import_reverse_unfreeze " << *dir << endl;
+
   // unfreeze
   dir->unfreeze_tree();
 
   // discard expire crap
   cache->discard_delayed_expire(dir);
+  
+  import_reverse_unpin(dir);
+}
 
-  // log our failure
-  mds->mdlog->submit_entry(new EImportFinish(dir,false));      // log failure
+void Migrator::import_reverse_unpin(CDir *dir) 
+{
+  dout(7) << "import_reverse_unpin " << *dir << endl;
+
+  // remove importing pin
+  dir->put(CDir::PIN_IMPORTING);
+
+  // remove bound pins
+  for (set<CDir*>::iterator it = import_bounds[dir].begin();
+       it != import_bounds[dir].end();
+       it++) {
+    CDir *bd = *it;
+    bd->put(CDir::PIN_IMPORTBOUND);
+    bd->state_clear(CDir::STATE_IMPORTBOUND);
+  }
 
   // clean up
   import_state.erase(dir->ino());
   import_peer.erase(dir->ino());
   import_bound_inos.erase(dir->ino());
   import_bounds.erase(dir);
+  import_bystanders.erase(dir);
 
   cache->show_subtrees();
   cache->show_cache();
@@ -1733,6 +1780,7 @@ void Migrator::import_finish(CDir *dir, bool now)
   import_peer.erase(dir->ino());
   import_bound_inos.erase(dir->ino());
   import_bounds.erase(dir);
+  import_bystanders.erase(dir);
 
   // process delayed expires
   cache->process_delayed_expire(dir);
@@ -1980,31 +2028,33 @@ void Migrator::handle_export_notify(MExportDirNotify *m)
   CDir *dir = cache->get_dir(m->get_ino());
 
   int from = m->get_source().num();
-  pair<int,int> auth = m->get_auth();
+  pair<int,int> old_auth = m->get_old_auth();
+  pair<int,int> new_auth = m->get_new_auth();
   
   if (!dir) {
-    dout(7) << "handle_export_notify " << auth
+    dout(7) << "handle_export_notify " << old_auth << " -> " << new_auth
            << " on missing dir " << m->get_ino() << endl;
+  } else if (dir->authority() != old_auth) {
+    dout(7) << "handle_export_notify old_auth was " << dir->authority() 
+           << " != " << old_auth << " -> " << new_auth
+           << " on " << *dir << endl;
   } else {
-    dout(7) << "handle_export_notify " << auth
+    dout(7) << "handle_export_notify " << old_auth << " -> " << new_auth
            << " on " << *dir << endl;
     // adjust auth
-    cache->adjust_bounded_subtree_auth(dir, m->get_exports(), auth);
+    cache->adjust_bounded_subtree_auth(dir, m->get_exports(), new_auth);
     
     // induce a merge?
     cache->try_subtree_merge(dir);
   }
   
   // send ack
-  if (auth.first == from && 
-      auth.second == CDIR_AUTH_UNKNOWN) {
-    // aborted.  no ack.
-    dout(7) << "handle_export_notify mds" << auth.first
-           << " aborted export, not sending ack for "
-           << *dir << endl;
-  } else {
+  if (m->wants_ack()) {
     mds->send_message_mds(new MExportDirNotifyAck(m->get_ino()),
                          from, MDS_PORT_MIGRATOR);
+  } else {
+    // aborted.  no ack.
+    dout(7) << "handle_export_notify no ack requested" << endl;
   }
   
   delete m;

diff --git a/branches/sage/cephmds2/mds/Migrator.h b/branches/sage/cephmds2/mds/Migrator.h
index 17b7e17eeb0e336c2ec3538b6fecee25adf9ad27..2fca1503efeacb83e7bd2ce295fb798fc8b4fe7a 100644 (file)
--- a/branches/sage/cephmds2/mds/Migrator.h
+++ b/branches/sage/cephmds2/mds/Migrator.h
@@ -68,12 +68,13 @@ public:
   // export stages.  used to clean up intelligently if there's a failure.
   const static int EXPORT_DISCOVERING   = 1;  // dest is disovering export dir
   const static int EXPORT_FREEZING      = 2;  // we're freezing the dir tree
-  const static int EXPORT_LOGGINGSTART  = 3;  // we're logging EExportStart
+  //const static int EXPORT_LOGGINGSTART  = 3;  // we're logging EExportStart
   const static int EXPORT_PREPPING      = 4;  // sending dest spanning tree to export bounds
   const static int EXPORT_WARNING       = 5;  // warning bystanders of dir_auth_pending
   const static int EXPORT_EXPORTING     = 6;  // sent actual export, waiting for ack
   const static int EXPORT_LOGGINGFINISH = 7;  // logging EExportFinish
   const static int EXPORT_NOTIFYING     = 8;  // waiting for notifyacks
+  const static int EXPORT_ABORTING      = 9;  // notifying bystanders of abort
 
 protected:
   // export fun
@@ -95,12 +96,14 @@ public:
   const static int IMPORT_LOGGINGSTART  = 4; // got import, logging EImportStart
   const static int IMPORT_ACKING        = 5; // logged EImportStart, sent ack, waiting for finish
   //const static int IMPORT_LOGGINGFINISH = 6; // logging EImportFinish
+  const static int IMPORT_ABORTING      = 7; // notifying bystanders of an abort before unfreezing
 
 protected:
   map<inodeno_t,int>              import_state;
   map<inodeno_t,int>              import_peer;
   map<inodeno_t,list<inodeno_t> > import_bound_inos;
   map<CDir*,set<CDir*> >          import_bounds;
+  map<CDir*,set<int> >            import_bystanders;
 
 
   // -- hashing madness --
@@ -205,6 +208,9 @@ public:
 public:
   void import_reverse(CDir *dir, bool fix_dir_auth=true);
 protected:
+  void import_reverse_unfreeze(CDir *dir);
+  void import_reverse_unpin(CDir *dir);
+  void import_notify_abort(CDir *dir);
   void import_logged_start(CDir *dir, int from,
                               list<inodeno_t> &imported_subdirs,
                               list<inodeno_t> &exports);

diff --git a/branches/sage/cephmds2/messages/MExportDirNotify.h b/branches/sage/cephmds2/messages/MExportDirNotify.h
index 68de14e7fb57fbbb1da436a4b813fd225f632c37..36d8d7c01aef8130ef5e4901e71f8779fd6d6bc4 100644 (file)
--- a/branches/sage/cephmds2/messages/MExportDirNotify.h
+++ b/branches/sage/cephmds2/messages/MExportDirNotify.h
@@ -20,25 +20,36 @@ using namespace std;
 
 class MExportDirNotify : public Message {
   inodeno_t ino;
-  pair<int,int> auth;
+  bool ack;
+  pair<int,int> old_auth, new_auth;
   list<inodeno_t> exports;  // bounds; these dirs are _not_ included (tho the inodes are)
 
   //list<inodeno_t> subdirs;
 
  public:
   inodeno_t get_ino() { return ino; }
-  pair<int,int> get_auth() { return auth; }
+  pair<int,int> get_old_auth() { return old_auth; }
+  pair<int,int> get_new_auth() { return new_auth; }
+  bool wants_ack() { return ack; }
   list<inodeno_t>& get_exports() { return exports; }
   //list<inodeno_t>::iterator subdirs_begin() { return subdirs.begin(); }
   //list<inodeno_t>::iterator subdirs_end() { return subdirs.end(); }
   //int num_subdirs() { return subdirs.size(); }
 
   MExportDirNotify() {}
-  MExportDirNotify(inodeno_t i, int a, int b) :
+  MExportDirNotify(inodeno_t i, bool a, pair<int,int> oa, pair<int,int> na) :
     Message(MSG_MDS_EXPORTDIRNOTIFY),
-    ino(i), auth(a,b) { }
-
+    ino(i), ack(a), old_auth(oa), new_auth(na) { }
+  
   virtual char *get_type_name() { return "ExNot"; }
+  void print(ostream& o) {
+    o << "export_notify(" << ino;
+    o << " " << old_auth << " -> " << new_auth;
+    if (ack) 
+      o << " ack)";
+    else
+      o << " no ack)";
+  }
   
   /*
   void copy_subdirs(list<inodeno_t>& s) {
@@ -61,16 +72,22 @@ class MExportDirNotify : public Message {
 
   virtual void decode_payload() {
     int off = 0;
-    payload.copy(off, sizeof(auth), (char*)&auth);
-    off += sizeof(auth);
     payload.copy(off, sizeof(ino), (char*)&ino);
     off += sizeof(ino);
+    payload.copy(off, sizeof(ack), (char*)&ack);
+    off += sizeof(ack);
+    payload.copy(off, sizeof(old_auth), (char*)&old_auth);
+    off += sizeof(old_auth);
+    payload.copy(off, sizeof(new_auth), (char*)&new_auth);
+    off += sizeof(new_auth);
     ::_decode(exports, payload, off);
     //::_decode(subdirs, payload, off);
   }
   virtual void encode_payload() {
-    payload.append((char*)&auth, sizeof(auth));
     payload.append((char*)&ino, sizeof(ino));
+    payload.append((char*)&ack, sizeof(ack));
+    payload.append((char*)&old_auth, sizeof(old_auth));
+    payload.append((char*)&new_auth, sizeof(new_auth));
     ::_encode(exports, payload);
     //::_encode(subdirs, payload);
   }

diff --git a/branches/sage/cephmds2/messages/MExportDirNotifyAck.h b/branches/sage/cephmds2/messages/MExportDirNotifyAck.h
index caa824af177f88d5b685d92586aabc94e47c7ac0..e1b996717d37ce245324c55a22831f372a50e40f 100644 (file)
--- a/branches/sage/cephmds2/messages/MExportDirNotifyAck.h
+++ b/branches/sage/cephmds2/messages/MExportDirNotifyAck.h
@@ -30,6 +30,9 @@ class MExportDirNotifyAck : public Message {
     this->ino = ino;
   }
   virtual char *get_type_name() { return "ExNotA"; }
+  void print(ostream& o) {
+    o << "export_notify_ack(" << ino << ")";
+  }
 
   virtual void decode_payload() {
     int off = 0;

diff --git a/branches/sage/cephmds2/messages/MExportDirPrep.h b/branches/sage/cephmds2/messages/MExportDirPrep.h
index 8b5141706d61d46621c62cae7b04f156142d73f7..798f3af8d84fa0c2a885c6a17685517ceacbce81 100644 (file)
--- a/branches/sage/cephmds2/messages/MExportDirPrep.h
+++ b/branches/sage/cephmds2/messages/MExportDirPrep.h
@@ -35,6 +35,8 @@ class MExportDirPrep : public Message {
 
   map<inodeno_t,CDirDiscover*>   dirs;
 
+  set<int>                       bystanders;
+
   bool b_did_assim;
 
  public:
@@ -53,6 +55,7 @@ class MExportDirPrep : public Message {
   CDirDiscover* get_dir(inodeno_t ino) {
     return dirs[ino];
   }
+  set<int> &get_bystanders() { return bystanders; }
 
   bool did_assim() { return b_did_assim; }
   void mark_assim() { b_did_assim = true; }
@@ -80,8 +83,6 @@ class MExportDirPrep : public Message {
   virtual char *get_type_name() { return "ExP"; }
 
 
-
-
   void add_export(inodeno_t dirino) {
     exports.push_back( dirino );
   }
@@ -93,7 +94,9 @@ class MExportDirPrep : public Message {
   void add_dir(CDirDiscover *dir) {
     dirs.insert(pair<inodeno_t, CDirDiscover*>(dir->get_ino(), dir));
   }
-
+  void add_bystander(int who) {
+    bystanders.insert(who);
+  }
 
   virtual void decode_payload() {
     int off = 0;
@@ -133,6 +136,8 @@ class MExportDirPrep : public Message {
       dir->_decode(payload, off);
       dirs[dir->get_ino()] = dir;
     }
+    
+    ::_decode(bystanders, payload, off);
   }
 
   virtual void encode_payload() {
@@ -163,6 +168,8 @@ class MExportDirPrep : public Message {
          dit != dirs.end();
          dit++)
       dit->second->_encode(payload);
+
+    ::_encode(bystanders, payload);
   }
 };