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
 424   // The only dentries elegible for purging are those
 425   // in the stray directories
 426   assert(dn->get_dir()->get_inode()->is_stray());
 427
 428   // Inode may not pass through this function if it
 429   // was already identified for purging (i.e. cannot
 430   // call eval_stray() after purge()
 431   assert(!dn->state_test(CDentry::STATE_PURGING));
 432
 433   if (!dn->is_auth()) {
 434     return false;
 435   }
 436
 437   if (dn->item_stray.is_on_list()) {
 438     if (delay)
 439       return false;
 440
 441     dn->item_stray.remove_myself();
 442     num_strays_delayed--;
 443     logger->set(l_mdc_num_strays_delayed, num_strays_delayed);
 444   }
 445
 446   // purge?
 447   if (in->inode.nlink == 0) {
 448     // past snaprealm parents imply snapped dentry remote links.
 449     // only important for directories.  normal file data snaps are handled
 450     // by the object store.
 451     if (in->snaprealm) {
 452       if (!in->snaprealm->have_past_parents_open() &&
 453           !in->snaprealm->open_parents(new C_MDC_EvalStray(this, dn))) {
 454         return false;
 455       }
 456       in->snaprealm->prune_past_parents();
 457       in->purge_stale_snap_data(in->snaprealm->get_snaps());
 458     }
 459     if (in->is_dir()) {
 460       if (in->snaprealm && in->snaprealm->has_past_parents()) {
 461         dout(20) << "  directory has past parents "
 462                  << in->snaprealm->srnode.past_parents << dendl;
 463         if (in->state_test(CInode::STATE_MISSINGOBJS)) {
 464           mds->clog->error() << "previous attempt at committing dirfrag of ino "
 465                              << in->ino() << " has failed, missing object";
 466           mds->handle_write_error(-ENOENT);
 467         }
 468         return false;  // not until some snaps are deleted.
 469       }
 470
 471       in->mdcache->clear_dirty_bits_for_stray(in);
 472
 473       if (!in->remote_parents.empty()) {
 474         // unlink any stale remote snap dentry.
 475         for (compact_set<CDentry*>::iterator p = in->remote_parents.begin();
 476              p != in->remote_parents.end(); ) {
 477           CDentry *remote_dn = *p;
 478           ++p;
 479           assert(remote_dn->last != CEPH_NOSNAP);
 480           remote_dn->unlink_remote(remote_dn->get_linkage());
 481         }
 482       }
 483     }
 484     if (dn->is_replicated()) {
 485       dout(20) << " replicated" << dendl;
 486       return false;
 487     }
 488     if (dn->is_any_leases() || in->is_any_caps()) {
 489       dout(20) << " caps | leases" << dendl;
 490       return false;  // wait
 491     }
 492     if (in->state_test(CInode::STATE_NEEDSRECOVER) ||
 493         in->state_test(CInode::STATE_RECOVERING)) {
 494       dout(20) << " pending recovery" << dendl;
 495       return false;  // don't mess with file size probing
 496     }
 497     if (in->get_num_ref() > (int)in->is_dirty() + (int)in->is_dirty_parent()) {
 498       dout(20) << " too many inode refs" << dendl;
 499       return false;
 500     }
 501     if (dn->get_num_ref() > (int)dn->is_dirty() + !!in->get_num_ref()) {
 502       dout(20) << " too many dn refs" << dendl;
 503       return false;
 504     }
 505     if (delay) {
 506       if (!dn->item_stray.is_on_list()) {
 507         delayed_eval_stray.push_back(&dn->item_stray);
 508         num_strays_delayed++;
 509         logger->set(l_mdc_num_strays_delayed, num_strays_delayed);
 510       }
 511     // don't purge multiversion inode with snap data
 512     } else if (in->snaprealm && in->snaprealm->has_past_parents() &&
 513               !in->old_inodes.empty()) {
 514       // A file with snapshots: we will truncate the HEAD revision
 515       // but leave the metadata intact.
 516       assert(!in->is_dir());
 517       dout(20) << " file has past parents "
 518         << in->snaprealm->srnode.past_parents << dendl;
 519       if (in->is_file() && in->get_projected_inode()->size > 0) {
 520         enqueue(dn, true); // truncate head objects
 521       }
 522     } else {
 523       // A straightforward file, ready to be purged.  Enqueue it.
 524       if (in->is_dir()) {
 525         in->close_dirfrags();
 526       }
 527
 528       enqueue(dn, false);
 529     }
 530
 531     return true;
 532   } else {
 533     /*
 534      * Where a stray has some links, they should be remotes, check
 535      * if we can do anything with them if we happen to have them in
 536      * cache.
 537      */
 538     _eval_stray_remote(dn, NULL);
 539     return false;
 540   }
 541 }
 542
 543 void StrayManager::activate()
 544 {
 545   dout(10) << __func__ << dendl;
 546   started = true;
 547 }
 548
 549 bool StrayManager::eval_stray(CDentry *dn, bool delay)
 550 {
 551   // avoid nested eval_stray
 552   if (dn->state_test(CDentry::STATE_EVALUATINGSTRAY))
 553       return false;
 554
 555   dn->state_set(CDentry::STATE_EVALUATINGSTRAY);
 556   bool ret = _eval_stray(dn, delay);
 557   dn->state_clear(CDentry::STATE_EVALUATINGSTRAY);
 558   return ret;
 559 }
 560
 561 void StrayManager::eval_remote(CDentry *remote_dn)
 562 {
 563   dout(10) << __func__ << " " << *remote_dn << dendl;
 564
 565   CDentry::linkage_t *dnl = remote_dn->get_projected_linkage();
 566   assert(dnl->is_remote());
 567   CInode *in = dnl->get_inode();
 568
 569   if (!in) {
 570     dout(20) << __func__ << ": no inode, cannot evaluate" << dendl;
 571     return;
 572   }
 573
 574   if (remote_dn->last != CEPH_NOSNAP) {
 575     dout(20) << __func__ << ": snap dentry, cannot evaluate" << dendl;
 576     return;
 577   }
 578
 579   // refers to stray?
 580   CDentry *primary_dn = in->get_projected_parent_dn();
 581   assert(primary_dn != NULL);
 582   if (primary_dn->get_dir()->get_inode()->is_stray()) {
 583     _eval_stray_remote(primary_dn, remote_dn);
 584   } else {
 585     dout(20) << __func__ << ": inode's primary dn not stray" << dendl;
 586   }
 587 }
 588
 589 class C_RetryEvalRemote : public StrayManagerContext {
 590   CDentry *dn;
 591   public:
 592     C_RetryEvalRemote(StrayManager *sm_, CDentry *dn_) :
 593       StrayManagerContext(sm_), dn(dn_) {
 594       dn->get(CDentry::PIN_PTRWAITER);
 595     }
 596     void finish(int r) override {
 597       if (dn->get_projected_linkage()->is_remote())
 598         sm->eval_remote(dn);
 599       dn->put(CDentry::PIN_PTRWAITER);
 600     }
 601 };
 602
 603 void StrayManager::_eval_stray_remote(CDentry *stray_dn, CDentry *remote_dn)
 604 {
 605   dout(20) << __func__ << " " << *stray_dn << dendl;
 606   assert(stray_dn != NULL);
 607   assert(stray_dn->get_dir()->get_inode()->is_stray());
 608   CDentry::linkage_t *stray_dnl = stray_dn->get_projected_linkage();
 609   assert(stray_dnl->is_primary());
 610   CInode *stray_in = stray_dnl->get_inode();
 611   assert(stray_in->inode.nlink >= 1);
 612   assert(stray_in->last == CEPH_NOSNAP);
 613
 614   /* If no remote_dn hinted, pick one arbitrarily */
 615   if (remote_dn == NULL) {
 616     if (!stray_in->remote_parents.empty()) {
 617       for (compact_set<CDentry*>::iterator p = stray_in->remote_parents.begin();
 618            p != stray_in->remote_parents.end();
 619            ++p)
 620         if ((*p)->last == CEPH_NOSNAP && !(*p)->is_projected()) {
 621           if ((*p)->is_auth()) {
 622             remote_dn = *p;
 623             if (remote_dn->dir->can_auth_pin())
 624               break;
 625           } else if (!remote_dn) {
 626             remote_dn = *p;
 627           }
 628         }
 629     }
 630     if (!remote_dn) {
 631       dout(20) << __func__ << ": not reintegrating (no remote parents in cache)" << dendl;
 632       return;
 633     }
 634   }
 635   assert(remote_dn->last == CEPH_NOSNAP);
 636   // NOTE: we repeat this check in _rename(), since our submission path is racey.
 637   if (!remote_dn->is_projected()) {
 638     if (remote_dn->is_auth()) {
 639       if (remote_dn->dir->can_auth_pin()) {
 640         reintegrate_stray(stray_dn, remote_dn);
 641       } else {
 642         remote_dn->dir->add_waiter(CDir::WAIT_UNFREEZE, new C_RetryEvalRemote(this, remote_dn));
 643         dout(20) << __func__ << ": not reintegrating (can't authpin remote parent)" << dendl;
 644       }
 645
 646     } else if (!remote_dn->is_auth() && stray_dn->is_auth()) {
 647       migrate_stray(stray_dn, remote_dn->authority().first);
 648     } else {
 649       dout(20) << __func__ << ": not reintegrating" << dendl;
 650     }
 651   } else {
 652     // don't do anything if the remote parent is projected, or we may
 653     // break user-visible semantics!
 654     dout(20) << __func__ << ": not reintegrating (projected)" << dendl;
 655   }
 656 }
 657
 658 void StrayManager::reintegrate_stray(CDentry *straydn, CDentry *rdn)
 659 {
 660   dout(10) << __func__ << " " << *straydn << " into " << *rdn << dendl;
 661
 662   logger->inc(l_mdc_strays_reintegrated);
 663
 664   // rename it to another mds.
 665   filepath src;
 666   straydn->make_path(src);
 667   filepath dst;
 668   rdn->make_path(dst);
 669
 670   MClientRequest *req = new MClientRequest(CEPH_MDS_OP_RENAME);
 671   req->set_filepath(dst);
 672   req->set_filepath2(src);
 673   req->set_tid(mds->issue_tid());
 674
 675   mds->send_message_mds(req, rdn->authority().first);
 676 }
 677
 678 void StrayManager::migrate_stray(CDentry *dn, mds_rank_t to)
 679 {
 680   CInode *in = dn->get_projected_linkage()->get_inode();
 681   assert(in);
 682   CInode *diri = dn->dir->get_inode();
 683   assert(diri->is_stray());
 684   dout(10) << "migrate_stray from mds." << MDS_INO_STRAY_OWNER(diri->inode.ino)
 685            << " to mds." << to
 686            << " " << *dn << " " << *in << dendl;
 687
 688   logger->inc(l_mdc_strays_migrated);
 689
 690   // rename it to another mds.
 691   filepath src;
 692   dn->make_path(src);
 693   assert(src.depth() == 2);
 694
 695   filepath dst(MDS_INO_MDSDIR(to));
 696   dst.push_dentry(src[0]);
 697   dst.push_dentry(src[1]);
 698
 699   MClientRequest *req = new MClientRequest(CEPH_MDS_OP_RENAME);
 700   req->set_filepath(dst);
 701   req->set_filepath2(src);
 702   req->set_tid(mds->issue_tid());
 703
 704   mds->send_message_mds(req, to);
 705 }
 706
 707 StrayManager::StrayManager(MDSRank *mds, PurgeQueue &purge_queue_)
 708   : delayed_eval_stray(member_offset(CDentry, item_stray)),
 709     mds(mds), logger(NULL), started(false), num_strays(0),
 710     num_strays_delayed(0), num_strays_enqueuing(0),
 711     purge_queue(purge_queue_)
 712 {
 713   assert(mds != NULL);
 714 }
 715
 716 void StrayManager::truncate(CDentry *dn)
 717 {
 718   const CDentry::linkage_t *dnl = dn->get_projected_linkage();
 719   const CInode *in = dnl->get_inode();
 720   assert(in);
 721   dout(10) << __func__ << ": " << *dn << " " << *in << dendl;
 722   assert(!dn->is_replicated());
 723
 724   const SnapRealm *realm = in->find_snaprealm();
 725   assert(realm);
 726   dout(10) << " realm " << *realm << dendl;
 727   const SnapContext *snapc = &realm->get_snap_context();
 728
 729   uint64_t to = in->inode.get_max_size();
 730   to = MAX(in->inode.size, to);
 731   // when truncating a file, the filer does not delete stripe objects that are
 732   // truncated to zero. so we need to purge stripe objects up to the max size
 733   // the file has ever been.
 734   to = MAX(in->inode.max_size_ever, to);
 735
 736   assert(to > 0);
 737
 738   PurgeItem item;
 739   item.ino = in->inode.ino;
 740   item.layout = in->inode.layout;
 741   item.snapc = *snapc;
 742   item.size = to;
 743
 744   purge_queue.push(item, new C_IO_PurgeStrayPurged(
 745         this, dn, true));
 746 }
 747
 748 void StrayManager::_truncate_stray_logged(CDentry *dn, LogSegment *ls)
 749 {
 750   CInode *in = dn->get_projected_linkage()->get_inode();
 751
 752   dout(10) << __func__ << ": " << *dn << " " << *in << dendl;
 753
 754   dn->state_clear(CDentry::STATE_PURGING | CDentry::STATE_PURGINGPINNED);
 755   dn->put(CDentry::PIN_PURGING);
 756
 757   in->pop_and_dirty_projected_inode(ls);
 758
 759   eval_stray(dn);
 760 }
 761