ceph/src/mds/StrayManager.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2015 Red Hat
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15
  16 #include "common/perf_counters.h"
  17
  18 #include "mds/MDSRank.h"
  19 #include "mds/MDCache.h"
  20 #include "mds/MDLog.h"
  21 #include "mds/CDir.h"
  22 #include "mds/CDentry.h"
  23 #include "events/EUpdate.h"
  24 #include "messages/MClientRequest.h"
  25
  26 #include "StrayManager.h"
  27
  28 #define dout_context g_ceph_context
  29 #define dout_subsys ceph_subsys_mds
  30 #undef dout_prefix
  31 #define dout_prefix _prefix(_dout, mds)
  32 static ostream& _prefix(std::ostream *_dout, MDSRank *mds) {
  33   return *_dout << "mds." << mds->get_nodeid() << ".cache.strays ";
  34 }
  35
  36 class StrayManagerIOContext : public virtual MDSIOContextBase {
  37 protected:
  38   StrayManager *sm;
  39   MDSRank *get_mds() override
  40   {
  41     return sm->mds;
  42   }
  43 public:
  44   explicit StrayManagerIOContext(StrayManager *sm_) : sm(sm_) {}
  45 };
  46
  47 class StrayManagerLogContext : public virtual MDSLogContextBase {
  48 protected:
  49   StrayManager *sm;
  50   MDSRank *get_mds() override
  51   {
  52     return sm->mds;
  53   }
  54 public:
  55   explicit StrayManagerLogContext(StrayManager *sm_) : sm(sm_) {}
  56 };
  57
  58 class StrayManagerContext : public virtual MDSInternalContextBase {
  59 protected:
  60   StrayManager *sm;
  61   MDSRank *get_mds() override
  62   {
  63     return sm->mds;
  64   }
  65 public:
  66   explicit StrayManagerContext(StrayManager *sm_) : sm(sm_) {}
  67 };
  68
  69
  70 /**
  71  * Context wrapper for _purge_stray_purged completion
  72  */
  73 class C_IO_PurgeStrayPurged : public StrayManagerIOContext {
  74   CDentry *dn;
  75   bool only_head;
  76   // How many ops_in_flight were allocated to this purge?
  77   uint32_t ops_allowance;
  78 public:
  79   C_IO_PurgeStrayPurged(StrayManager *sm_, CDentry *d, bool oh) :
  80     StrayManagerIOContext(sm_), dn(d), only_head(oh) { }
  81   void finish(int r) override {
  82     assert(r == 0 || r == -ENOENT);
  83     sm->_purge_stray_purged(dn, ops_allowance, only_head);
  84   }
  85 };
  86
  87
  88 void StrayManager::purge(CDentry *dn)
  89 {
  90   CDentry::linkage_t *dnl = dn->get_projected_linkage();
  91   CInode *in = dnl->get_inode();
  92   dout(10) << __func__ << " " << *dn << " " << *in << dendl;
  93   assert(!dn->is_replicated());
  94
  95   // CHEAT.  there's no real need to journal our intent to purge, since
  96   // that is implicit in the dentry's presence and non-use in the stray
  97   // dir.  on recovery, we'll need to re-eval all strays anyway.
  98
  99   SnapContext nullsnapc;
 100
 101   PurgeItem item;
 102   item.ino = in->inode.ino;
 103   if (in->is_dir()) {
 104     item.action = PurgeItem::PURGE_DIR;
 105     item.fragtree = in->dirfragtree;
 106   } else {
 107     item.action = PurgeItem::PURGE_FILE;
 108
 109     const SnapContext *snapc;
 110     SnapRealm *realm = in->find_snaprealm();
 111     if (realm) {
 112       dout(10) << " realm " << *realm << dendl;
 113       snapc = &realm->get_snap_context();
 114     } else {
 115       dout(10) << " NO realm, using null context" << dendl;
 116       snapc = &nullsnapc;
 117       assert(in->last == CEPH_NOSNAP);
 118     }
 119
 120     uint64_t to = 0;
 121     if (in->is_file()) {
 122       to = in->inode.get_max_size();
 123       to = MAX(in->inode.size, to);
 124       // when truncating a file, the filer does not delete stripe objects that are
 125       // truncated to zero. so we need to purge stripe objects up to the max size
 126       // the file has ever been.
 127       to = MAX(in->inode.max_size_ever, to);
 128     }
 129
 130     inode_t *pi = in->get_projected_inode();
 131
 132     item.size = to;
 133     item.layout = pi->layout;
 134     item.old_pools = pi->old_pools;
 135     item.snapc = *snapc;
 136   }
 137
 138   purge_queue.push(item, new C_IO_PurgeStrayPurged(
 139         this, dn, false));
 140 }
 141
 142 class C_PurgeStrayLogged : public StrayManagerLogContext {
 143   CDentry *dn;
 144   version_t pdv;
 145   LogSegment *ls;
 146 public:
 147   C_PurgeStrayLogged(StrayManager *sm_, CDentry *d, version_t v, LogSegment *s) :
 148     StrayManagerLogContext(sm_), dn(d), pdv(v), ls(s) { }
 149   void finish(int r) override {
 150     sm->_purge_stray_logged(dn, pdv, ls);
 151   }
 152 };
 153
 154 class C_TruncateStrayLogged : public StrayManagerLogContext {
 155   CDentry *dn;
 156   LogSegment *ls;
 157 public:
 158   C_TruncateStrayLogged(StrayManager *sm, CDentry *d, LogSegment *s) :
 159     StrayManagerLogContext(sm), dn(d), ls(s) { }
 160   void finish(int r) override {
 161     sm->_truncate_stray_logged(dn, ls);
 162   }
 163 };
 164
 165 void StrayManager::_purge_stray_purged(
 166     CDentry *dn, uint32_t ops_allowance, bool only_head)
 167 {
 168   CInode *in = dn->get_projected_linkage()->get_inode();
 169   dout(10) << "_purge_stray_purged " << *dn << " " << *in << dendl;
 170
 171   logger->inc(l_mdc_strays_enqueued);
 172   num_strays_enqueuing--;
 173   logger->set(l_mdc_num_strays_enqueuing, num_strays_enqueuing);
 174
 175   if (only_head) {
 176     /* This was a ::truncate */
 177     EUpdate *le = new EUpdate(mds->mdlog, "purge_stray truncate");
 178     mds->mdlog->start_entry(le);
 179
 180     inode_t *pi = in->project_inode();
 181     pi->size = 0;
 182     pi->max_size_ever = 0;
 183     pi->client_ranges.clear();
 184     pi->truncate_size = 0;
 185     pi->truncate_from = 0;
 186     pi->version = in->pre_dirty();
 187
 188     le->metablob.add_dir_context(dn->dir);
 189     le->metablob.add_primary_dentry(dn, in, true);
 190
 191     mds->mdlog->submit_entry(le,
 192         new C_TruncateStrayLogged(
 193           this, dn, mds->mdlog->get_current_segment()));
 194   } else {
 195     if (in->get_num_ref() != (int)in->is_dirty() ||
 196         dn->get_num_ref() != (int)dn->is_dirty() + !!in->get_num_ref() + 1/*PIN_PURGING*/) {
 197       // Nobody should be taking new references to an inode when it
 198       // is being purged (aside from it were
 199
 200       derr << "Rogue reference after purge to " << *dn << dendl;
 201       assert(0 == "rogue reference to purging inode");
 202     }
 203
 204     // kill dentry.
 205     version_t pdv = dn->pre_dirty();
 206     dn->push_projected_linkage(); // NULL
 207
 208     EUpdate *le = new EUpdate(mds->mdlog, "purge_stray");
 209     mds->mdlog->start_entry(le);
 210
 211     // update dirfrag fragstat, rstat
 212     CDir *dir = dn->get_dir();
 213     fnode_t *pf = dir->project_fnode();
 214     pf->version = dir->pre_dirty();
 215     if (in->is_dir())
 216       pf->fragstat.nsubdirs--;
 217     else
 218       pf->fragstat.nfiles--;
 219     pf->rstat.sub(in->inode.accounted_rstat);
 220
 221     le->metablob.add_dir_context(dn->dir);
 222     EMetaBlob::dirlump& dl = le->metablob.add_dir(dn->dir, true);
 223     le->metablob.add_null_dentry(dl, dn, true);
 224     le->metablob.add_destroyed_inode(in->ino());
 225
 226     mds->mdlog->submit_entry(le, new C_PurgeStrayLogged(this, dn, pdv,
 227           mds->mdlog->get_current_segment()));
 228
 229     logger->set(l_mdc_num_strays, num_strays);
 230   }
 231 }
 232
 233 void StrayManager::_purge_stray_logged(CDentry *dn, version_t pdv, LogSegment *ls)
 234 {
 235   CInode *in = dn->get_linkage()->get_inode();
 236   dout(10) << "_purge_stray_logged " << *dn << " " << *in << dendl;
 237
 238   assert(!in->state_test(CInode::STATE_RECOVERING));
 239
 240   bool new_dn = dn->is_new();
 241
 242   // unlink
 243   assert(dn->get_projected_linkage()->is_null());
 244   dn->dir->unlink_inode(dn, !new_dn);
 245   dn->pop_projected_linkage();
 246   dn->mark_dirty(pdv, ls);
 247
 248   dn->dir->pop_and_dirty_projected_fnode(ls);
 249
 250   in->state_clear(CInode::STATE_ORPHAN);
 251   dn->state_clear(CDentry::STATE_PURGING | CDentry::STATE_PURGINGPINNED);
 252   dn->put(CDentry::PIN_PURGING);
 253
 254   // drop dentry?
 255   if (new_dn) {
 256     dout(20) << " dn is new, removing" << dendl;
 257     dn->mark_clean();
 258     dn->dir->remove_dentry(dn);
 259   }
 260
 261   // drop inode
 262   if (in->is_dirty())
 263     in->mark_clean();
 264   in->mdcache->remove_inode(in);
 265 }
 266
 267 void StrayManager::enqueue(CDentry *dn, bool trunc)
 268 {
 269   CDentry::linkage_t *dnl = dn->get_projected_linkage();
 270   assert(dnl);
 271   CInode *in = dnl->get_inode();
 272   assert(in);
 273
 274   /* We consider a stray to be purging as soon as it is enqueued, to avoid
 275    * enqueing it twice */
 276   dn->state_set(CDentry::STATE_PURGING);
 277   in->state_set(CInode::STATE_PURGING);
 278
 279   /* We must clear this as soon as enqueuing it, to prevent the journal
 280    * expiry code from seeing a dirty parent and trying to write a backtrace */
 281   if (!trunc) {
 282     if (in->is_dirty_parent()) {
 283       in->clear_dirty_parent();
 284     }
 285   }
 286
 287   dout(20) << __func__ << ": purging dn: " << *dn << dendl;
 288
 289   if (!dn->state_test(CDentry::STATE_PURGINGPINNED)) {
 290     dn->get(CDentry::PIN_PURGING);
 291     dn->state_set(CDentry::STATE_PURGINGPINNED);
 292   }
 293
 294   ++num_strays_enqueuing;
 295   logger->set(l_mdc_num_strays_enqueuing, num_strays_enqueuing);
 296
 297   // Resources are available, acquire them and execute the purge
 298   _enqueue(dn, trunc);
 299
 300   dout(10) << __func__ << ": purging this dentry immediately: "
 301     << *dn << dendl;
 302 }
 303
 304 class C_OpenSnapParents : public StrayManagerContext {
 305   CDentry *dn;
 306   bool trunc;
 307   public:
 308     C_OpenSnapParents(StrayManager *sm_, CDentry *dn_, bool t) :
 309       StrayManagerContext(sm_), dn(dn_), trunc(t) { }
 310     void finish(int r) override {
 311       sm->_enqueue(dn, trunc);
 312     }
 313 };
 314
 315 void StrayManager::_enqueue(CDentry *dn, bool trunc)
 316 {
 317   CInode *in = dn->get_linkage()->get_inode();
 318   if (in->snaprealm &&
 319       !in->snaprealm->have_past_parents_open() &&
 320       !in->snaprealm->open_parents(new C_OpenSnapParents(this, dn, trunc))) {
 321     // this can happen if the dentry had been trimmed from cache.
 322     return;
 323   }
 324
 325   if (!started) {
 326     // If the MDS is not yet active, defer executing this purge
 327     // in order to avoid the mdlog writes we do on purge completion.
 328     mds->wait_for_active(
 329         new MDSInternalContextWrapper(mds,
 330           new FunctionContext([this, dn, trunc](int r){
 331             // It is safe to hold on to this CDentry* pointer
 332             // because the dentry is pinned with PIN_PURGING
 333            _enqueue(dn, trunc);
 334             })
 335         )
 336       );
 337
 338     return;
 339   }
 340
 341   if (trunc) {
 342     truncate(dn);
 343   } else {
 344     purge(dn);
 345   }
 346 }
 347
 348
 349 void StrayManager::advance_delayed()
 350 {
 351   for (elist<CDentry*>::iterator p = delayed_eval_stray.begin(); !p.end(); ) {
 352     CDentry *dn = *p;
 353     ++p;
 354     dn->item_stray.remove_myself();
 355     num_strays_delayed--;
 356
 357     if (dn->get_projected_linkage()->is_null()) {
 358       /* A special case: a stray dentry can go null if its inode is being
 359        * re-linked into another MDS's stray dir during a shutdown migration. */
 360       dout(4) << __func__ << ": delayed dentry is now null: " << *dn << dendl;
 361       continue;
 362     }
 363
 364     const bool purging = eval_stray(dn);
 365     if (!purging) {
 366       derr << "Dentry " << *dn << " was purgeable but no longer is!" << dendl;
 367       /*
 368        * This can happen if a stray is purgeable, but has gained an extra
 369        * reference by virtue of having its backtrace updated.
 370        * FIXME perhaps we could simplify this further by
 371        * avoiding writing the backtrace of purge-ready strays, so
 372        * that this code could be more rigid?
 373        */
 374     }
 375   }
 376   logger->set(l_mdc_num_strays_delayed, num_strays_delayed);
 377 }
 378
 379 void StrayManager::set_num_strays(uint64_t num)
 380 {
 381   assert(!started);
 382   num_strays = num;
 383   logger->set(l_mdc_num_strays, num_strays);
 384 }
 385
 386 void StrayManager::notify_stray_created()
 387 {
 388   num_strays++;
 389   logger->set(l_mdc_num_strays, num_strays);
 390   logger->inc(l_mdc_strays_created);
 391 }
 392
 393 void StrayManager::notify_stray_removed()
 394 {
 395   num_strays--;
 396   logger->set(l_mdc_num_strays, num_strays);
 397 }
 398
 399 struct C_EvalStray : public StrayManagerContext {
 400   CDentry *dn;
 401   C_EvalStray(StrayManager *sm_, CDentry *d) : StrayManagerContext(sm_), dn(d) {}
 402   void finish(int r) override {
 403     sm->eval_stray(dn);
 404   }
 405 };
 406
 407 struct C_MDC_EvalStray : public StrayManagerContext {
 408   CDentry *dn;
 409   C_MDC_EvalStray(StrayManager *sm_, CDentry *d) : StrayManagerContext(sm_), dn(d) {}
 410   void finish(int r) override {
 411     sm->eval_stray(dn);
 412   }
 413 };
 414
 415 bool StrayManager::_eval_stray(CDentry *dn, bool delay)
 416 {
 417   dout(10) << "eval_stray " << *dn << dendl;
 418   CDentry::linkage_t *dnl = dn->get_projected_linkage();
 419   assert(dnl->is_primary());
 420   dout(10) << " inode is " << *dnl->get_inode() << dendl;
 421   CInode *in = dnl->get_inode();
 422   assert(in);
 423   assert(!in->state_test(CInode::STATE_REJOINUNDEF));
 424
 425   // The only dentries elegible for purging are those
 426   // in the stray directories
 427   assert(dn->get_dir()->get_inode()->is_stray());
 428
 429   // Inode may not pass through this function if it
 430   // was already identified for purging (i.e. cannot
 431   // call eval_stray() after purge()
 432   assert(!dn->state_test(CDentry::STATE_PURGING));
 433
 434   if (!dn->is_auth()) {
 435     return false;
 436   }
 437
 438   if (dn->item_stray.is_on_list()) {
 439     if (delay)
 440       return false;
 441
 442     dn->item_stray.remove_myself();
 443     num_strays_delayed--;
 444     logger->set(l_mdc_num_strays_delayed, num_strays_delayed);
 445   }
 446
 447   // purge?
 448   if (in->inode.nlink == 0) {
 449     // past snaprealm parents imply snapped dentry remote links.
 450     // only important for directories.  normal file data snaps are handled
 451     // by the object store.
 452     if (in->snaprealm) {
 453       if (!in->snaprealm->have_past_parents_open() &&
 454           !in->snaprealm->open_parents(new C_MDC_EvalStray(this, dn))) {
 455         return false;
 456       }
 457       in->snaprealm->prune_past_parents();
 458       in->purge_stale_snap_data(in->snaprealm->get_snaps());
 459     }
 460     if (in->is_dir()) {
 461       if (in->snaprealm && in->snaprealm->has_past_parents()) {
 462         dout(20) << "  directory has past parents "
 463                  << in->snaprealm->srnode.past_parents << dendl;
 464         if (in->state_test(CInode::STATE_MISSINGOBJS)) {
 465           mds->clog->error() << "previous attempt at committing dirfrag of ino "
 466                              << in->ino() << " has failed, missing object";
 467           mds->handle_write_error(-ENOENT);
 468         }
 469         return false;  // not until some snaps are deleted.
 470       }
 471
 472       in->mdcache->clear_dirty_bits_for_stray(in);
 473
 474       if (!in->remote_parents.empty()) {
 475         // unlink any stale remote snap dentry.
 476         for (compact_set<CDentry*>::iterator p = in->remote_parents.begin();
 477              p != in->remote_parents.end(); ) {
 478           CDentry *remote_dn = *p;
 479           ++p;
 480           assert(remote_dn->last != CEPH_NOSNAP);
 481           remote_dn->unlink_remote(remote_dn->get_linkage());
 482         }
 483       }
 484     }
 485     if (dn->is_replicated()) {
 486       dout(20) << " replicated" << dendl;
 487       return false;
 488     }
 489     if (dn->is_any_leases() || in->is_any_caps()) {
 490       dout(20) << " caps | leases" << dendl;
 491       return false;  // wait
 492     }
 493     if (in->state_test(CInode::STATE_NEEDSRECOVER) ||
 494         in->state_test(CInode::STATE_RECOVERING)) {
 495       dout(20) << " pending recovery" << dendl;
 496       return false;  // don't mess with file size probing
 497     }
 498     if (in->get_num_ref() > (int)in->is_dirty() + (int)in->is_dirty_parent()) {
 499       dout(20) << " too many inode refs" << dendl;
 500       return false;
 501     }
 502     if (dn->get_num_ref() > (int)dn->is_dirty() + !!in->get_num_ref()) {
 503       dout(20) << " too many dn refs" << dendl;
 504       return false;
 505     }
 506     if (delay) {
 507       if (!dn->item_stray.is_on_list()) {
 508         delayed_eval_stray.push_back(&dn->item_stray);
 509         num_strays_delayed++;
 510         logger->set(l_mdc_num_strays_delayed, num_strays_delayed);
 511       }
 512     // don't purge multiversion inode with snap data
 513     } else if (in->snaprealm && in->snaprealm->has_past_parents() &&
 514               !in->old_inodes.empty()) {
 515       // A file with snapshots: we will truncate the HEAD revision
 516       // but leave the metadata intact.
 517       assert(!in->is_dir());
 518       dout(20) << " file has past parents "
 519         << in->snaprealm->srnode.past_parents << dendl;
 520       if (in->is_file() && in->get_projected_inode()->size > 0) {
 521         enqueue(dn, true); // truncate head objects
 522       }
 523     } else {
 524       // A straightforward file, ready to be purged.  Enqueue it.
 525       if (in->is_dir()) {
 526         in->close_dirfrags();
 527       }
 528
 529       enqueue(dn, false);
 530     }
 531
 532     return true;
 533   } else {
 534     /*
 535      * Where a stray has some links, they should be remotes, check
 536      * if we can do anything with them if we happen to have them in
 537      * cache.
 538      */
 539     _eval_stray_remote(dn, NULL);
 540     return false;
 541   }
 542 }
 543
 544 void StrayManager::activate()
 545 {
 546   dout(10) << __func__ << dendl;
 547   started = true;
 548 }
 549
 550 bool StrayManager::eval_stray(CDentry *dn, bool delay)
 551 {
 552   // avoid nested eval_stray
 553   if (dn->state_test(CDentry::STATE_EVALUATINGSTRAY))
 554       return false;
 555
 556   dn->state_set(CDentry::STATE_EVALUATINGSTRAY);
 557   bool ret = _eval_stray(dn, delay);
 558   dn->state_clear(CDentry::STATE_EVALUATINGSTRAY);
 559   return ret;
 560 }
 561
 562 void StrayManager::eval_remote(CDentry *remote_dn)
 563 {
 564   dout(10) << __func__ << " " << *remote_dn << dendl;
 565
 566   CDentry::linkage_t *dnl = remote_dn->get_projected_linkage();
 567   assert(dnl->is_remote());
 568   CInode *in = dnl->get_inode();
 569
 570   if (!in) {
 571     dout(20) << __func__ << ": no inode, cannot evaluate" << dendl;
 572     return;
 573   }
 574
 575   if (remote_dn->last != CEPH_NOSNAP) {
 576     dout(20) << __func__ << ": snap dentry, cannot evaluate" << dendl;
 577     return;
 578   }
 579
 580   // refers to stray?
 581   CDentry *primary_dn = in->get_projected_parent_dn();
 582   assert(primary_dn != NULL);
 583   if (primary_dn->get_dir()->get_inode()->is_stray()) {
 584     _eval_stray_remote(primary_dn, remote_dn);
 585   } else {
 586     dout(20) << __func__ << ": inode's primary dn not stray" << dendl;
 587   }
 588 }
 589
 590 class C_RetryEvalRemote : public StrayManagerContext {
 591   CDentry *dn;
 592   public:
 593     C_RetryEvalRemote(StrayManager *sm_, CDentry *dn_) :
 594       StrayManagerContext(sm_), dn(dn_) {
 595       dn->get(CDentry::PIN_PTRWAITER);
 596     }
 597     void finish(int r) override {
 598       if (dn->get_projected_linkage()->is_remote())
 599         sm->eval_remote(dn);
 600       dn->put(CDentry::PIN_PTRWAITER);
 601     }
 602 };
 603
 604 void StrayManager::_eval_stray_remote(CDentry *stray_dn, CDentry *remote_dn)
 605 {
 606   dout(20) << __func__ << " " << *stray_dn << dendl;
 607   assert(stray_dn != NULL);
 608   assert(stray_dn->get_dir()->get_inode()->is_stray());
 609   CDentry::linkage_t *stray_dnl = stray_dn->get_projected_linkage();
 610   assert(stray_dnl->is_primary());
 611   CInode *stray_in = stray_dnl->get_inode();
 612   assert(stray_in->inode.nlink >= 1);
 613   assert(stray_in->last == CEPH_NOSNAP);
 614
 615   /* If no remote_dn hinted, pick one arbitrarily */
 616   if (remote_dn == NULL) {
 617     if (!stray_in->remote_parents.empty()) {
 618       for (compact_set<CDentry*>::iterator p = stray_in->remote_parents.begin();
 619            p != stray_in->remote_parents.end();
 620            ++p)
 621         if ((*p)->last == CEPH_NOSNAP && !(*p)->is_projected()) {
 622           if ((*p)->is_auth()) {
 623             remote_dn = *p;
 624             if (remote_dn->dir->can_auth_pin())
 625               break;
 626           } else if (!remote_dn) {
 627             remote_dn = *p;
 628           }
 629         }
 630     }
 631     if (!remote_dn) {
 632       dout(20) << __func__ << ": not reintegrating (no remote parents in cache)" << dendl;
 633       return;
 634     }
 635   }
 636   assert(remote_dn->last == CEPH_NOSNAP);
 637   // NOTE: we repeat this check in _rename(), since our submission path is racey.
 638   if (!remote_dn->is_projected()) {
 639     if (remote_dn->is_auth()) {
 640       if (remote_dn->dir->can_auth_pin()) {
 641         reintegrate_stray(stray_dn, remote_dn);
 642       } else {
 643         remote_dn->dir->add_waiter(CDir::WAIT_UNFREEZE, new C_RetryEvalRemote(this, remote_dn));
 644         dout(20) << __func__ << ": not reintegrating (can't authpin remote parent)" << dendl;
 645       }
 646
 647     } else if (!remote_dn->is_auth() && stray_dn->is_auth()) {
 648       migrate_stray(stray_dn, remote_dn->authority().first);
 649     } else {
 650       dout(20) << __func__ << ": not reintegrating" << dendl;
 651     }
 652   } else {
 653     // don't do anything if the remote parent is projected, or we may
 654     // break user-visible semantics!
 655     dout(20) << __func__ << ": not reintegrating (projected)" << dendl;
 656   }
 657 }
 658
 659 void StrayManager::reintegrate_stray(CDentry *straydn, CDentry *rdn)
 660 {
 661   dout(10) << __func__ << " " << *straydn << " into " << *rdn << dendl;
 662
 663   logger->inc(l_mdc_strays_reintegrated);
 664
 665   // rename it to another mds.
 666   filepath src;
 667   straydn->make_path(src);
 668   filepath dst;
 669   rdn->make_path(dst);
 670
 671   MClientRequest *req = new MClientRequest(CEPH_MDS_OP_RENAME);
 672   req->set_filepath(dst);
 673   req->set_filepath2(src);
 674   req->set_tid(mds->issue_tid());
 675
 676   mds->send_message_mds(req, rdn->authority().first);
 677 }
 678
 679 void StrayManager::migrate_stray(CDentry *dn, mds_rank_t to)
 680 {
 681   CInode *in = dn->get_projected_linkage()->get_inode();
 682   assert(in);
 683   CInode *diri = dn->dir->get_inode();
 684   assert(diri->is_stray());
 685   dout(10) << "migrate_stray from mds." << MDS_INO_STRAY_OWNER(diri->inode.ino)
 686            << " to mds." << to
 687            << " " << *dn << " " << *in << dendl;
 688
 689   logger->inc(l_mdc_strays_migrated);
 690
 691   // rename it to another mds.
 692   filepath src;
 693   dn->make_path(src);
 694   assert(src.depth() == 2);
 695
 696   filepath dst(MDS_INO_MDSDIR(to));
 697   dst.push_dentry(src[0]);
 698   dst.push_dentry(src[1]);
 699
 700   MClientRequest *req = new MClientRequest(CEPH_MDS_OP_RENAME);
 701   req->set_filepath(dst);
 702   req->set_filepath2(src);
 703   req->set_tid(mds->issue_tid());
 704
 705   mds->send_message_mds(req, to);
 706 }
 707
 708 StrayManager::StrayManager(MDSRank *mds, PurgeQueue &purge_queue_)
 709   : delayed_eval_stray(member_offset(CDentry, item_stray)),
 710     mds(mds), logger(NULL), started(false), num_strays(0),
 711     num_strays_delayed(0), num_strays_enqueuing(0),
 712     purge_queue(purge_queue_)
 713 {
 714   assert(mds != NULL);
 715 }
 716
 717 void StrayManager::truncate(CDentry *dn)
 718 {
 719   const CDentry::linkage_t *dnl = dn->get_projected_linkage();
 720   const CInode *in = dnl->get_inode();
 721   assert(in);
 722   dout(10) << __func__ << ": " << *dn << " " << *in << dendl;
 723   assert(!dn->is_replicated());
 724
 725   const SnapRealm *realm = in->find_snaprealm();
 726   assert(realm);
 727   dout(10) << " realm " << *realm << dendl;
 728   const SnapContext *snapc = &realm->get_snap_context();
 729
 730   uint64_t to = in->inode.get_max_size();
 731   to = MAX(in->inode.size, to);
 732   // when truncating a file, the filer does not delete stripe objects that are
 733   // truncated to zero. so we need to purge stripe objects up to the max size
 734   // the file has ever been.
 735   to = MAX(in->inode.max_size_ever, to);
 736
 737   assert(to > 0);
 738
 739   PurgeItem item;
 740   item.ino = in->inode.ino;
 741   item.layout = in->inode.layout;
 742   item.snapc = *snapc;
 743   item.size = to;
 744
 745   purge_queue.push(item, new C_IO_PurgeStrayPurged(
 746         this, dn, true));
 747 }
 748
 749 void StrayManager::_truncate_stray_logged(CDentry *dn, LogSegment *ls)
 750 {
 751   CInode *in = dn->get_projected_linkage()->get_inode();
 752
 753   dout(10) << __func__ << ": " << *dn << " " << *in << dendl;
 754
 755   dn->state_clear(CDentry::STATE_PURGING | CDentry::STATE_PURGINGPINNED);
 756   dn->put(CDentry::PIN_PURGING);
 757
 758   in->pop_and_dirty_projected_inode(ls);
 759
 760   eval_stray(dn);
 761 }
 762