ceph/src/mds/CDir.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) 2004-2006 Sage Weil <sage@newdream.net>
   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 #include <string_view>
  16 #include <algorithm>
  17
  18 #include "include/types.h"
  19
  20 #include "CDir.h"
  21 #include "CDentry.h"
  22 #include "CInode.h"
  23 #include "Mutation.h"
  24
  25 #include "MDSMap.h"
  26 #include "MDSRank.h"
  27 #include "MDCache.h"
  28 #include "Locker.h"
  29 #include "MDLog.h"
  30 #include "LogSegment.h"
  31 #include "MDBalancer.h"
  32
  33 #include "common/bloom_filter.hpp"
  34 #include "include/Context.h"
  35 #include "common/Clock.h"
  36
  37 #include "osdc/Objecter.h"
  38
  39 #include "common/config.h"
  40 #include "include/ceph_assert.h"
  41 #include "include/compat.h"
  42
  43 #define dout_context g_ceph_context
  44 #define dout_subsys ceph_subsys_mds
  45 #undef dout_prefix
  46 #define dout_prefix *_dout << "mds." << mdcache->mds->get_nodeid() << ".cache.dir(" << this->dirfrag() << ") "
  47
  48 using namespace std;
  49
  50 int CDir::num_frozen_trees = 0;
  51 int CDir::num_freezing_trees = 0;
  52
  53 CDir::fnode_const_ptr CDir::empty_fnode = CDir::allocate_fnode();
  54
  55 class CDirContext : public MDSContext
  56 {
  57 protected:
  58   CDir *dir;
  59   MDSRank* get_mds() override {return dir->mdcache->mds;}
  60
  61 public:
  62   explicit CDirContext(CDir *d) : dir(d) {
  63     ceph_assert(dir != NULL);
  64   }
  65 };
  66
  67
  68 class CDirIOContext : public MDSIOContextBase
  69 {
  70 protected:
  71   CDir *dir;
  72   MDSRank* get_mds() override {return dir->mdcache->mds;}
  73
  74 public:
  75   explicit CDirIOContext(CDir *d) : dir(d) {
  76     ceph_assert(dir != NULL);
  77   }
  78 };
  79
  80
  81 // PINS
  82 //int cdir_pins[CDIR_NUM_PINS] = { 0,0,0,0,0,0,0,0,0,0,0,0,0,0 };
  83
  84
  85 ostream& operator<<(ostream& out, const CDir& dir)
  86 {
  87   out << "[dir " << dir.dirfrag() << " " << dir.get_path() << "/"
  88       << " [" << dir.first << ",head]";
  89   if (dir.is_auth()) {
  90     out << " auth";
  91     if (dir.is_replicated())
  92       out << dir.get_replicas();
  93
  94     if (dir.is_projected())
  95       out << " pv=" << dir.get_projected_version();
  96     out << " v=" << dir.get_version();
  97     out << " cv=" << dir.get_committing_version();
  98     out << "/" << dir.get_committed_version();
  99   } else {
 100     mds_authority_t a = dir.authority();
 101     out << " rep@" << a.first;
 102     if (a.second != CDIR_AUTH_UNKNOWN)
 103       out << "," << a.second;
 104     out << "." << dir.get_replica_nonce();
 105   }
 106
 107   if (dir.is_rep()) out << " REP";
 108
 109   if (dir.get_dir_auth() != CDIR_AUTH_DEFAULT) {
 110     if (dir.get_dir_auth().second == CDIR_AUTH_UNKNOWN)
 111       out << " dir_auth=" << dir.get_dir_auth().first;
 112     else
 113       out << " dir_auth=" << dir.get_dir_auth();
 114   }
 115
 116   if (dir.get_auth_pins() || dir.get_dir_auth_pins()) {
 117     out << " ap=" << dir.get_auth_pins()
 118         << "+" << dir.get_dir_auth_pins();
 119 #ifdef MDS_AUTHPIN_SET
 120     dir.print_authpin_set(out);
 121 #endif
 122   }
 123
 124   out << " state=" << dir.get_state();
 125   if (dir.state_test(CDir::STATE_COMPLETE)) out << "|complete";
 126   if (dir.state_test(CDir::STATE_FREEZINGTREE)) out << "|freezingtree";
 127   if (dir.state_test(CDir::STATE_FROZENTREE)) out << "|frozentree";
 128   if (dir.state_test(CDir::STATE_AUXSUBTREE)) out << "|auxsubtree";
 129   if (dir.state_test(CDir::STATE_FROZENDIR)) out << "|frozendir";
 130   if (dir.state_test(CDir::STATE_FREEZINGDIR)) out << "|freezingdir";
 131   if (dir.state_test(CDir::STATE_EXPORTBOUND)) out << "|exportbound";
 132   if (dir.state_test(CDir::STATE_IMPORTBOUND)) out << "|importbound";
 133   if (dir.state_test(CDir::STATE_BADFRAG)) out << "|badfrag";
 134   if (dir.state_test(CDir::STATE_FRAGMENTING)) out << "|fragmenting";
 135   if (dir.state_test(CDir::STATE_CREATING)) out << "|creating";
 136   if (dir.state_test(CDir::STATE_COMMITTING)) out << "|committing";
 137   if (dir.state_test(CDir::STATE_FETCHING)) out << "|fetching";
 138   if (dir.state_test(CDir::STATE_EXPORTING)) out << "|exporting";
 139   if (dir.state_test(CDir::STATE_IMPORTING)) out << "|importing";
 140   if (dir.state_test(CDir::STATE_STICKY)) out << "|sticky";
 141   if (dir.state_test(CDir::STATE_DNPINNEDFRAG)) out << "|dnpinnedfrag";
 142   if (dir.state_test(CDir::STATE_ASSIMRSTAT)) out << "|assimrstat";
 143
 144   // fragstat
 145   out << " " << dir.get_fnode()->fragstat;
 146   if (!(dir.get_fnode()->fragstat == dir.get_fnode()->accounted_fragstat))
 147     out << "/" << dir.get_fnode()->accounted_fragstat;
 148   if (g_conf()->mds_debug_scatterstat && dir.is_projected()) {
 149     const auto& pf = dir.get_projected_fnode();
 150     out << "->" << pf->fragstat;
 151     if (!(pf->fragstat == pf->accounted_fragstat))
 152       out << "/" << pf->accounted_fragstat;
 153   }
 154
 155   // rstat
 156   out << " " << dir.get_fnode()->rstat;
 157   if (!(dir.get_fnode()->rstat == dir.get_fnode()->accounted_rstat))
 158     out << "/" << dir.get_fnode()->accounted_rstat;
 159   if (g_conf()->mds_debug_scatterstat && dir.is_projected()) {
 160     const auto& pf = dir.get_projected_fnode();
 161     out << "->" << pf->rstat;
 162     if (!(pf->rstat == pf->accounted_rstat))
 163       out << "/" << pf->accounted_rstat;
 164  }
 165
 166   out << " hs=" << dir.get_num_head_items() << "+" << dir.get_num_head_null();
 167   out << ",ss=" << dir.get_num_snap_items() << "+" << dir.get_num_snap_null();
 168   if (dir.get_num_dirty())
 169     out << " dirty=" << dir.get_num_dirty();
 170
 171   if (dir.get_num_ref()) {
 172     out << " |";
 173     dir.print_pin_set(out);
 174   }
 175
 176   out << " " << &dir;
 177   return out << "]";
 178 }
 179
 180
 181 void CDir::print(ostream& out)
 182 {
 183   out << *this;
 184 }
 185
 186
 187
 188
 189 ostream& CDir::print_db_line_prefix(ostream& out)
 190 {
 191   return out << ceph_clock_now() << " mds." << mdcache->mds->get_nodeid() << ".cache.dir(" << this->dirfrag() << ") ";
 192 }
 193
 194
 195
 196 // -------------------------------------------------------------------
 197 // CDir
 198
 199 CDir::CDir(CInode *in, frag_t fg, MDCache *mdc, bool auth) :
 200   mdcache(mdc), inode(in), frag(fg),
 201   dirty_rstat_inodes(member_offset(CInode, dirty_rstat_item)),
 202   dirty_dentries(member_offset(CDentry, item_dir_dirty)),
 203   item_dirty(this), item_new(this),
 204   lock_caches_with_auth_pins(member_offset(MDLockCache::DirItem, item_dir)),
 205   freezing_inodes(member_offset(CInode, item_freezing_inode)),
 206   dir_rep(REP_NONE),
 207   pop_me(mdc->decayrate),
 208   pop_nested(mdc->decayrate),
 209   pop_auth_subtree(mdc->decayrate),
 210   pop_auth_subtree_nested(mdc->decayrate),
 211   pop_spread(mdc->decayrate),
 212   pop_lru_subdirs(member_offset(CInode, item_pop_lru)),
 213   dir_auth(CDIR_AUTH_DEFAULT)
 214 {
 215   // auth
 216   ceph_assert(in->is_dir());
 217   if (auth)
 218     state_set(STATE_AUTH);
 219 }
 220
 221 /**
 222  * Check the recursive statistics on size for consistency.
 223  * If mds_debug_scatterstat is enabled, assert for correctness,
 224  * otherwise just print out the mismatch and continue.
 225  */
 226 bool CDir::check_rstats(bool scrub)
 227 {
 228   if (!g_conf()->mds_debug_scatterstat && !scrub)
 229     return true;
 230
 231   dout(25) << "check_rstats on " << this << dendl;
 232   if (!is_complete() || !is_auth() || is_frozen()) {
 233     dout(3) << "check_rstats " << (scrub ? "(scrub) " : "")
 234             << "bailing out -- incomplete or non-auth or frozen dir on "
 235             << *this << dendl;
 236     return !scrub;
 237   }
 238
 239   frag_info_t frag_info;
 240   nest_info_t nest_info;
 241   for (auto i = items.begin(); i != items.end(); ++i) {
 242     if (i->second->last != CEPH_NOSNAP)
 243       continue;
 244     CDentry::linkage_t *dnl = i->second->get_linkage();
 245     if (dnl->is_primary()) {
 246       CInode *in = dnl->get_inode();
 247       nest_info.add(in->get_inode()->accounted_rstat);
 248       if (in->is_dir())
 249         frag_info.nsubdirs++;
 250       else
 251         frag_info.nfiles++;
 252     } else if (dnl->is_remote())
 253       frag_info.nfiles++;
 254   }
 255
 256   bool good = true;
 257   // fragstat
 258   if(!frag_info.same_sums(fnode->fragstat)) {
 259     dout(1) << "mismatch between head items and fnode.fragstat! printing dentries" << dendl;
 260     dout(1) << "get_num_head_items() = " << get_num_head_items()
 261              << "; fnode.fragstat.nfiles=" << fnode->fragstat.nfiles
 262              << " fnode.fragstat.nsubdirs=" << fnode->fragstat.nsubdirs << dendl;
 263     good = false;
 264   } else {
 265     dout(20) << "get_num_head_items() = " << get_num_head_items()
 266              << "; fnode.fragstat.nfiles=" << fnode->fragstat.nfiles
 267              << " fnode.fragstat.nsubdirs=" << fnode->fragstat.nsubdirs << dendl;
 268   }
 269
 270   // rstat
 271   if (!nest_info.same_sums(fnode->rstat)) {
 272     dout(1) << "mismatch between child accounted_rstats and my rstats!" << dendl;
 273     dout(1) << "total of child dentries: " << nest_info << dendl;
 274     dout(1) << "my rstats:              " << fnode->rstat << dendl;
 275     good = false;
 276   } else {
 277     dout(20) << "total of child dentries: " << nest_info << dendl;
 278     dout(20) << "my rstats:              " << fnode->rstat << dendl;
 279   }
 280
 281   if (!good) {
 282     if (!scrub) {
 283       for (auto i = items.begin(); i != items.end(); ++i) {
 284         CDentry *dn = i->second;
 285         if (dn->get_linkage()->is_primary()) {
 286           CInode *in = dn->get_linkage()->inode;
 287           dout(1) << *dn << " rstat " << in->get_inode()->accounted_rstat << dendl;
 288         } else {
 289           dout(1) << *dn << dendl;
 290         }
 291       }
 292
 293       ceph_assert(frag_info.nfiles == fnode->fragstat.nfiles);
 294       ceph_assert(frag_info.nsubdirs == fnode->fragstat.nsubdirs);
 295       ceph_assert(nest_info.rbytes == fnode->rstat.rbytes);
 296       ceph_assert(nest_info.rfiles == fnode->rstat.rfiles);
 297       ceph_assert(nest_info.rsubdirs == fnode->rstat.rsubdirs);
 298     }
 299   }
 300   dout(10) << "check_rstats complete on " << this << dendl;
 301   return good;
 302 }
 303
 304 void CDir::adjust_num_inodes_with_caps(int d)
 305 {
 306   // FIXME: smarter way to decide if adding 'this' to open file table
 307   if (num_inodes_with_caps == 0 && d > 0)
 308     mdcache->open_file_table.add_dirfrag(this);
 309   else if (num_inodes_with_caps > 0 && num_inodes_with_caps == -d)
 310     mdcache->open_file_table.remove_dirfrag(this);
 311
 312   num_inodes_with_caps += d;
 313   ceph_assert(num_inodes_with_caps >= 0);
 314 }
 315
 316 CDentry *CDir::lookup(std::string_view name, snapid_t snap)
 317 {
 318   dout(20) << "lookup (" << snap << ", '" << name << "')" << dendl;
 319   auto iter = items.lower_bound(dentry_key_t(snap, name, inode->hash_dentry_name(name)));
 320   if (iter == items.end())
 321     return 0;
 322   if (iter->second->get_name() == name &&
 323       iter->second->first <= snap &&
 324       iter->second->last >= snap) {
 325     dout(20) << "  hit -> " << iter->first << dendl;
 326     return iter->second;
 327   }
 328   dout(20) << "  miss -> " << iter->first << dendl;
 329   return 0;
 330 }
 331
 332 CDentry *CDir::lookup_exact_snap(std::string_view name, snapid_t last) {
 333   dout(20) << __func__ << " (" << last << ", '" << name << "')" << dendl;
 334   auto p = items.find(dentry_key_t(last, name, inode->hash_dentry_name(name)));
 335   if (p == items.end())
 336     return NULL;
 337   return p->second;
 338 }
 339
 340 /***
 341  * linking fun
 342  */
 343
 344 CDentry* CDir::add_null_dentry(std::string_view dname,
 345                                snapid_t first, snapid_t last)
 346 {
 347   // foreign
 348   ceph_assert(lookup_exact_snap(dname, last) == 0);
 349
 350   // create dentry
 351   CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), "", first, last);
 352   if (is_auth())
 353     dn->state_set(CDentry::STATE_AUTH);
 354
 355   mdcache->bottom_lru.lru_insert_mid(dn);
 356   dn->state_set(CDentry::STATE_BOTTOMLRU);
 357
 358   dn->dir = this;
 359   dn->version = get_projected_version();
 360
 361   // add to dir
 362   ceph_assert(items.count(dn->key()) == 0);
 363   //assert(null_items.count(dn->get_name()) == 0);
 364
 365   items[dn->key()] = dn;
 366   if (last == CEPH_NOSNAP)
 367     num_head_null++;
 368   else
 369     num_snap_null++;
 370
 371   if (state_test(CDir::STATE_DNPINNEDFRAG)) {
 372     dn->get(CDentry::PIN_FRAGMENTING);
 373     dn->state_set(CDentry::STATE_FRAGMENTING);
 374   }
 375
 376   dout(12) << __func__ << " " << *dn << dendl;
 377
 378   // pin?
 379   if (get_num_any() == 1)
 380     get(PIN_CHILD);
 381
 382   ceph_assert(get_num_any() == items.size());
 383   return dn;
 384 }
 385
 386
 387 CDentry* CDir::add_primary_dentry(std::string_view dname, CInode *in,
 388                                   mempool::mds_co::string alternate_name,
 389                                   snapid_t first, snapid_t last)
 390 {
 391   // primary
 392   ceph_assert(lookup_exact_snap(dname, last) == 0);
 393
 394   // create dentry
 395   CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), std::move(alternate_name), first, last);
 396   if (is_auth())
 397     dn->state_set(CDentry::STATE_AUTH);
 398   if (is_auth() || !inode->is_stray()) {
 399     mdcache->lru.lru_insert_mid(dn);
 400   } else {
 401     mdcache->bottom_lru.lru_insert_mid(dn);
 402     dn->state_set(CDentry::STATE_BOTTOMLRU);
 403   }
 404
 405   dn->dir = this;
 406   dn->version = get_projected_version();
 407
 408   // add to dir
 409   ceph_assert(items.count(dn->key()) == 0);
 410   //assert(null_items.count(dn->get_name()) == 0);
 411
 412   items[dn->key()] = dn;
 413
 414   dn->get_linkage()->inode = in;
 415
 416   link_inode_work(dn, in);
 417
 418   if (dn->last == CEPH_NOSNAP)
 419     num_head_items++;
 420   else
 421     num_snap_items++;
 422
 423   if (state_test(CDir::STATE_DNPINNEDFRAG)) {
 424     dn->get(CDentry::PIN_FRAGMENTING);
 425     dn->state_set(CDentry::STATE_FRAGMENTING);
 426   }
 427
 428   dout(12) << __func__ << " " << *dn << dendl;
 429
 430   // pin?
 431   if (get_num_any() == 1)
 432     get(PIN_CHILD);
 433   ceph_assert(get_num_any() == items.size());
 434   return dn;
 435 }
 436
 437 CDentry* CDir::add_remote_dentry(std::string_view dname, inodeno_t ino, unsigned char d_type,
 438                                  mempool::mds_co::string alternate_name,
 439                                  snapid_t first, snapid_t last)
 440 {
 441   // foreign
 442   ceph_assert(lookup_exact_snap(dname, last) == 0);
 443
 444   // create dentry
 445   CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), std::move(alternate_name), ino, d_type, first, last);
 446   if (is_auth())
 447     dn->state_set(CDentry::STATE_AUTH);
 448   mdcache->lru.lru_insert_mid(dn);
 449
 450   dn->dir = this;
 451   dn->version = get_projected_version();
 452
 453   // add to dir
 454   ceph_assert(items.count(dn->key()) == 0);
 455   //assert(null_items.count(dn->get_name()) == 0);
 456
 457   items[dn->key()] = dn;
 458   if (last == CEPH_NOSNAP)
 459     num_head_items++;
 460   else
 461     num_snap_items++;
 462
 463   if (state_test(CDir::STATE_DNPINNEDFRAG)) {
 464     dn->get(CDentry::PIN_FRAGMENTING);
 465     dn->state_set(CDentry::STATE_FRAGMENTING);
 466   }
 467
 468   dout(12) << __func__ << " " << *dn << dendl;
 469
 470   // pin?
 471   if (get_num_any() == 1)
 472     get(PIN_CHILD);
 473
 474   ceph_assert(get_num_any() == items.size());
 475   return dn;
 476 }
 477
 478
 479
 480 void CDir::remove_dentry(CDentry *dn)
 481 {
 482   dout(12) << __func__ << " " << *dn << dendl;
 483
 484   // there should be no client leases at this point!
 485   ceph_assert(dn->client_lease_map.empty());
 486
 487   if (state_test(CDir::STATE_DNPINNEDFRAG)) {
 488     dn->put(CDentry::PIN_FRAGMENTING);
 489     dn->state_clear(CDentry::STATE_FRAGMENTING);
 490   }
 491
 492   if (dn->get_linkage()->is_null()) {
 493     if (dn->last == CEPH_NOSNAP)
 494       num_head_null--;
 495     else
 496       num_snap_null--;
 497   } else {
 498     if (dn->last == CEPH_NOSNAP)
 499       num_head_items--;
 500     else
 501       num_snap_items--;
 502   }
 503
 504   if (!dn->get_linkage()->is_null())
 505     // detach inode and dentry
 506     unlink_inode_work(dn);
 507
 508   // remove from list
 509   ceph_assert(items.count(dn->key()) == 1);
 510   items.erase(dn->key());
 511
 512   // clean?
 513   if (dn->is_dirty())
 514     dn->mark_clean();
 515
 516   if (dn->state_test(CDentry::STATE_BOTTOMLRU))
 517     mdcache->bottom_lru.lru_remove(dn);
 518   else
 519     mdcache->lru.lru_remove(dn);
 520   delete dn;
 521
 522   // unpin?
 523   if (get_num_any() == 0)
 524     put(PIN_CHILD);
 525   ceph_assert(get_num_any() == items.size());
 526 }
 527
 528 void CDir::link_remote_inode(CDentry *dn, CInode *in)
 529 {
 530   link_remote_inode(dn, in->ino(), IFTODT(in->get_projected_inode()->mode));
 531 }
 532
 533 void CDir::link_remote_inode(CDentry *dn, inodeno_t ino, unsigned char d_type)
 534 {
 535   dout(12) << __func__ << " " << *dn << " remote " << ino << dendl;
 536   ceph_assert(dn->get_linkage()->is_null());
 537
 538   dn->get_linkage()->set_remote(ino, d_type);
 539
 540   if (dn->state_test(CDentry::STATE_BOTTOMLRU)) {
 541     mdcache->bottom_lru.lru_remove(dn);
 542     mdcache->lru.lru_insert_mid(dn);
 543     dn->state_clear(CDentry::STATE_BOTTOMLRU);
 544   }
 545
 546   if (dn->last == CEPH_NOSNAP) {
 547     num_head_items++;
 548     num_head_null--;
 549   } else {
 550     num_snap_items++;
 551     num_snap_null--;
 552   }
 553   ceph_assert(get_num_any() == items.size());
 554 }
 555
 556 void CDir::link_primary_inode(CDentry *dn, CInode *in)
 557 {
 558   dout(12) << __func__ << " " << *dn << " " << *in << dendl;
 559   ceph_assert(dn->get_linkage()->is_null());
 560
 561   dn->get_linkage()->inode = in;
 562
 563   link_inode_work(dn, in);
 564
 565   if (dn->state_test(CDentry::STATE_BOTTOMLRU) &&
 566       (is_auth() || !inode->is_stray())) {
 567     mdcache->bottom_lru.lru_remove(dn);
 568     mdcache->lru.lru_insert_mid(dn);
 569     dn->state_clear(CDentry::STATE_BOTTOMLRU);
 570   }
 571
 572   if (dn->last == CEPH_NOSNAP) {
 573     num_head_items++;
 574     num_head_null--;
 575   } else {
 576     num_snap_items++;
 577     num_snap_null--;
 578   }
 579
 580   ceph_assert(get_num_any() == items.size());
 581 }
 582
 583 void CDir::link_inode_work( CDentry *dn, CInode *in)
 584 {
 585   ceph_assert(dn->get_linkage()->get_inode() == in);
 586   in->set_primary_parent(dn);
 587
 588   // set inode version
 589   //in->inode.version = dn->get_version();
 590
 591   // pin dentry?
 592   if (in->get_num_ref())
 593     dn->get(CDentry::PIN_INODEPIN);
 594
 595   if (in->state_test(CInode::STATE_TRACKEDBYOFT))
 596     mdcache->open_file_table.notify_link(in);
 597   if (in->is_any_caps())
 598     adjust_num_inodes_with_caps(1);
 599
 600   // adjust auth pin count
 601   if (in->auth_pins)
 602     dn->adjust_nested_auth_pins(in->auth_pins, NULL);
 603
 604   if (in->is_freezing_inode())
 605     freezing_inodes.push_back(&in->item_freezing_inode);
 606   else if (in->is_frozen_inode() || in->is_frozen_auth_pin())
 607     num_frozen_inodes++;
 608
 609   // verify open snaprealm parent
 610   if (in->snaprealm)
 611     in->snaprealm->adjust_parent();
 612   else if (in->is_any_caps())
 613     in->move_to_realm(inode->find_snaprealm());
 614 }
 615
 616 void CDir::unlink_inode(CDentry *dn, bool adjust_lru)
 617 {
 618   if (dn->get_linkage()->is_primary()) {
 619     dout(12) << __func__ << " " << *dn << " " << *dn->get_linkage()->get_inode() << dendl;
 620   } else {
 621     dout(12) << __func__ << " " << *dn << dendl;
 622   }
 623
 624   unlink_inode_work(dn);
 625
 626   if (adjust_lru && !dn->state_test(CDentry::STATE_BOTTOMLRU)) {
 627     mdcache->lru.lru_remove(dn);
 628     mdcache->bottom_lru.lru_insert_mid(dn);
 629     dn->state_set(CDentry::STATE_BOTTOMLRU);
 630   }
 631
 632   if (dn->last == CEPH_NOSNAP) {
 633     num_head_items--;
 634     num_head_null++;
 635   } else {
 636     num_snap_items--;
 637     num_snap_null++;
 638   }
 639   ceph_assert(get_num_any() == items.size());
 640 }
 641
 642
 643 void CDir::try_remove_unlinked_dn(CDentry *dn)
 644 {
 645   ceph_assert(dn->dir == this);
 646   ceph_assert(dn->get_linkage()->is_null());
 647
 648   // no pins (besides dirty)?
 649   if (dn->get_num_ref() != dn->is_dirty())
 650     return;
 651
 652   // was the dn new?
 653   if (dn->is_new()) {
 654     dout(10) << __func__ << " " << *dn << " in " << *this << dendl;
 655     if (dn->is_dirty())
 656       dn->mark_clean();
 657     remove_dentry(dn);
 658
 659     // NOTE: we may not have any more dirty dentries, but the fnode
 660     // still changed, so the directory must remain dirty.
 661   }
 662 }
 663
 664
 665 void CDir::unlink_inode_work(CDentry *dn)
 666 {
 667   CInode *in = dn->get_linkage()->get_inode();
 668
 669   if (dn->get_linkage()->is_remote()) {
 670     // remote
 671     if (in)
 672       dn->unlink_remote(dn->get_linkage());
 673
 674     dn->get_linkage()->set_remote(0, 0);
 675   } else if (dn->get_linkage()->is_primary()) {
 676     // primary
 677     // unpin dentry?
 678     if (in->get_num_ref())
 679       dn->put(CDentry::PIN_INODEPIN);
 680
 681     if (in->state_test(CInode::STATE_TRACKEDBYOFT))
 682       mdcache->open_file_table.notify_unlink(in);
 683     if (in->is_any_caps())
 684       adjust_num_inodes_with_caps(-1);
 685
 686     // unlink auth_pin count
 687     if (in->auth_pins)
 688       dn->adjust_nested_auth_pins(-in->auth_pins, nullptr);
 689
 690     if (in->is_freezing_inode())
 691       in->item_freezing_inode.remove_myself();
 692     else if (in->is_frozen_inode() || in->is_frozen_auth_pin())
 693       num_frozen_inodes--;
 694
 695     // detach inode
 696     in->remove_primary_parent(dn);
 697     if (in->is_dir())
 698       in->item_pop_lru.remove_myself();
 699     dn->get_linkage()->inode = 0;
 700   } else {
 701     ceph_assert(!dn->get_linkage()->is_null());
 702   }
 703 }
 704
 705 void CDir::add_to_bloom(CDentry *dn)
 706 {
 707   ceph_assert(dn->last == CEPH_NOSNAP);
 708   if (!bloom) {
 709     /* not create bloom filter for incomplete dir that was added by log replay */
 710     if (!is_complete())
 711       return;
 712
 713     /* don't maintain bloom filters in standby replay (saves cycles, and also
 714      * avoids need to implement clearing it in EExport for #16924) */
 715     if (mdcache->mds->is_standby_replay()) {
 716       return;
 717     }
 718
 719     unsigned size = get_num_head_items() + get_num_snap_items();
 720     if (size < 100) size = 100;
 721     bloom.reset(new bloom_filter(size, 1.0 / size, 0));
 722   }
 723   /* This size and false positive probability is completely random.*/
 724   bloom->insert(dn->get_name().data(), dn->get_name().size());
 725 }
 726
 727 bool CDir::is_in_bloom(std::string_view name)
 728 {
 729   if (!bloom)
 730     return false;
 731   return bloom->contains(name.data(), name.size());
 732 }
 733
 734 void CDir::remove_null_dentries() {
 735   dout(12) << __func__ << " " << *this << dendl;
 736
 737   auto p = items.begin();
 738   while (p != items.end()) {
 739     CDentry *dn = p->second;
 740     ++p;
 741     if (dn->get_linkage()->is_null() && !dn->is_projected())
 742       remove_dentry(dn);
 743   }
 744
 745   ceph_assert(num_snap_null == 0);
 746   ceph_assert(num_head_null == 0);
 747   ceph_assert(get_num_any() == items.size());
 748 }
 749
 750 /** remove dirty null dentries for deleted directory. the dirfrag will be
 751  *  deleted soon, so it's safe to not commit dirty dentries.
 752  *
 753  *  This is called when a directory is being deleted, a prerequisite
 754  *  of which is that its children have been unlinked: we expect to only see
 755  *  null, unprojected dentries here.
 756  */
 757 void CDir::try_remove_dentries_for_stray()
 758 {
 759   dout(10) << __func__ << dendl;
 760   ceph_assert(get_parent_dir()->inode->is_stray());
 761
 762   // clear dirty only when the directory was not snapshotted
 763   bool clear_dirty = !inode->snaprealm;
 764
 765   auto p = items.begin();
 766   while (p != items.end()) {
 767     CDentry *dn = p->second;
 768     ++p;
 769     if (dn->last == CEPH_NOSNAP) {
 770       ceph_assert(!dn->is_projected());
 771       ceph_assert(dn->get_linkage()->is_null());
 772       if (clear_dirty && dn->is_dirty())
 773         dn->mark_clean();
 774       // It's OK to remove lease prematurely because we will never link
 775       // the dentry to inode again.
 776       if (dn->is_any_leases())
 777         dn->remove_client_leases(mdcache->mds->locker);
 778       if (dn->get_num_ref() == 0)
 779         remove_dentry(dn);
 780     } else {
 781       ceph_assert(!dn->is_projected());
 782       CDentry::linkage_t *dnl= dn->get_linkage();
 783       CInode *in = NULL;
 784       if (dnl->is_primary()) {
 785         in = dnl->get_inode();
 786         if (clear_dirty && in->is_dirty())
 787           in->mark_clean();
 788       }
 789       if (clear_dirty && dn->is_dirty())
 790         dn->mark_clean();
 791       if (dn->get_num_ref() == 0) {
 792         remove_dentry(dn);
 793         if (in)
 794           mdcache->remove_inode(in);
 795       }
 796     }
 797   }
 798
 799   if (clear_dirty && is_dirty())
 800     mark_clean();
 801 }
 802
 803 bool CDir::try_trim_snap_dentry(CDentry *dn, const set<snapid_t>& snaps)
 804 {
 805   ceph_assert(dn->last != CEPH_NOSNAP);
 806   set<snapid_t>::const_iterator p = snaps.lower_bound(dn->first);
 807   CDentry::linkage_t *dnl= dn->get_linkage();
 808   CInode *in = 0;
 809   if (dnl->is_primary())
 810     in = dnl->get_inode();
 811   if ((p == snaps.end() || *p > dn->last) &&
 812       (dn->get_num_ref() == dn->is_dirty()) &&
 813       (!in || in->get_num_ref() == in->is_dirty())) {
 814     dout(10) << " purging snapped " << *dn << dendl;
 815     if (in && in->is_dirty())
 816       in->mark_clean();
 817     remove_dentry(dn);
 818     if (in) {
 819       dout(10) << " purging snapped " << *in << dendl;
 820       mdcache->remove_inode(in);
 821     }
 822     return true;
 823   }
 824   return false;
 825 }
 826
 827
 828 void CDir::purge_stale_snap_data(const set<snapid_t>& snaps)
 829 {
 830   dout(10) << __func__ << " " << snaps << dendl;
 831
 832   auto p = items.begin();
 833   while (p != items.end()) {
 834     CDentry *dn = p->second;
 835     ++p;
 836
 837     if (dn->last == CEPH_NOSNAP)
 838       continue;
 839
 840     try_trim_snap_dentry(dn, snaps);
 841   }
 842 }
 843
 844
 845 /**
 846  * steal_dentry -- semi-violently move a dentry from one CDir to another
 847  * (*) violently, in that nitems, most pins, etc. are not correctly maintained
 848  * on the old CDir corpse; must call finish_old_fragment() when finished.
 849  */
 850 void CDir::steal_dentry(CDentry *dn)
 851 {
 852   dout(15) << __func__ << " " << *dn << dendl;
 853
 854   items[dn->key()] = dn;
 855
 856   dn->dir->items.erase(dn->key());
 857   if (dn->dir->items.empty())
 858     dn->dir->put(PIN_CHILD);
 859
 860   if (get_num_any() == 0)
 861     get(PIN_CHILD);
 862   if (dn->get_linkage()->is_null()) {
 863     if (dn->last == CEPH_NOSNAP)
 864       num_head_null++;
 865     else
 866       num_snap_null++;
 867   } else if (dn->last == CEPH_NOSNAP) {
 868       num_head_items++;
 869
 870     auto _fnode = _get_fnode();
 871
 872     if (dn->get_linkage()->is_primary()) {
 873       CInode *in = dn->get_linkage()->get_inode();
 874       const auto& pi = in->get_projected_inode();
 875       if (in->is_dir()) {
 876         _fnode->fragstat.nsubdirs++;
 877         if (in->item_pop_lru.is_on_list())
 878           pop_lru_subdirs.push_back(&in->item_pop_lru);
 879       } else {
 880         _fnode->fragstat.nfiles++;
 881       }
 882       _fnode->rstat.rbytes += pi->accounted_rstat.rbytes;
 883       _fnode->rstat.rfiles += pi->accounted_rstat.rfiles;
 884       _fnode->rstat.rsubdirs += pi->accounted_rstat.rsubdirs;
 885       _fnode->rstat.rsnaps += pi->accounted_rstat.rsnaps;
 886       if (pi->accounted_rstat.rctime > fnode->rstat.rctime)
 887         _fnode->rstat.rctime = pi->accounted_rstat.rctime;
 888
 889       if (in->is_any_caps())
 890         adjust_num_inodes_with_caps(1);
 891
 892       // move dirty inode rstat to new dirfrag
 893       if (in->is_dirty_rstat())
 894         dirty_rstat_inodes.push_back(&in->dirty_rstat_item);
 895     } else if (dn->get_linkage()->is_remote()) {
 896       if (dn->get_linkage()->get_remote_d_type() == DT_DIR)
 897         _fnode->fragstat.nsubdirs++;
 898       else
 899         _fnode->fragstat.nfiles++;
 900     }
 901   } else {
 902     num_snap_items++;
 903     if (dn->get_linkage()->is_primary()) {
 904       CInode *in = dn->get_linkage()->get_inode();
 905       if (in->is_dirty_rstat())
 906         dirty_rstat_inodes.push_back(&in->dirty_rstat_item);
 907     }
 908   }
 909
 910   {
 911     int dap = dn->get_num_dir_auth_pins();
 912     if (dap) {
 913       adjust_nested_auth_pins(dap, NULL);
 914       dn->dir->adjust_nested_auth_pins(-dap, NULL);
 915     }
 916   }
 917
 918   if (dn->is_dirty()) {
 919     dirty_dentries.push_back(&dn->item_dir_dirty);
 920     num_dirty++;
 921   }
 922
 923   dn->dir = this;
 924 }
 925
 926 void CDir::prepare_old_fragment(map<string_snap_t, MDSContext::vec >& dentry_waiters, bool replay)
 927 {
 928   // auth_pin old fragment for duration so that any auth_pinning
 929   // during the dentry migration doesn't trigger side effects
 930   if (!replay && is_auth())
 931     auth_pin(this);
 932
 933   if (!waiting_on_dentry.empty()) {
 934     for (const auto &p : waiting_on_dentry) {
 935       auto &e = dentry_waiters[p.first];
 936       for (const auto &waiter : p.second) {
 937         e.push_back(waiter);
 938       }
 939     }
 940     waiting_on_dentry.clear();
 941     put(PIN_DNWAITER);
 942   }
 943 }
 944
 945 void CDir::prepare_new_fragment(bool replay)
 946 {
 947   if (!replay && is_auth()) {
 948     _freeze_dir();
 949     mark_complete();
 950   }
 951   inode->add_dirfrag(this);
 952 }
 953
 954 void CDir::finish_old_fragment(MDSContext::vec& waiters, bool replay)
 955 {
 956   // take waiters _before_ unfreeze...
 957   if (!replay) {
 958     take_waiting(WAIT_ANY_MASK, waiters);
 959     if (is_auth()) {
 960       auth_unpin(this);  // pinned in prepare_old_fragment
 961       ceph_assert(is_frozen_dir());
 962       unfreeze_dir();
 963     }
 964   }
 965
 966   ceph_assert(dir_auth_pins == 0);
 967   ceph_assert(auth_pins == 0);
 968
 969   num_head_items = num_head_null = 0;
 970   num_snap_items = num_snap_null = 0;
 971   adjust_num_inodes_with_caps(-num_inodes_with_caps);
 972
 973   // this mirrors init_fragment_pins()
 974   if (is_auth())
 975     clear_replica_map();
 976   if (is_dirty())
 977     mark_clean();
 978   if (state_test(STATE_IMPORTBOUND))
 979     put(PIN_IMPORTBOUND);
 980   if (state_test(STATE_EXPORTBOUND))
 981     put(PIN_EXPORTBOUND);
 982   if (is_subtree_root())
 983     put(PIN_SUBTREE);
 984
 985   if (auth_pins > 0)
 986     put(PIN_AUTHPIN);
 987
 988   ceph_assert(get_num_ref() == (state_test(STATE_STICKY) ? 1:0));
 989 }
 990
 991 void CDir::init_fragment_pins()
 992 {
 993   if (is_replicated())
 994     get(PIN_REPLICATED);
 995   if (state_test(STATE_DIRTY))
 996     get(PIN_DIRTY);
 997   if (state_test(STATE_EXPORTBOUND))
 998     get(PIN_EXPORTBOUND);
 999   if (state_test(STATE_IMPORTBOUND))
1000     get(PIN_IMPORTBOUND);
1001   if (is_subtree_root())
1002     get(PIN_SUBTREE);
1003 }
1004
1005 void CDir::split(int bits, std::vector<CDir*>* subs, MDSContext::vec& waiters, bool replay)
1006 {
1007   dout(10) << "split by " << bits << " bits on " << *this << dendl;
1008
1009   ceph_assert(replay || is_complete() || !is_auth());
1010
1011   frag_vec_t frags;
1012   frag.split(bits, frags);
1013
1014   vector<CDir*> subfrags(1 << bits);
1015
1016   double fac = 1.0 / (double)(1 << bits);  // for scaling load vecs
1017
1018   version_t rstat_version = inode->get_projected_inode()->rstat.version;
1019   version_t dirstat_version = inode->get_projected_inode()->dirstat.version;
1020
1021   nest_info_t rstatdiff;
1022   frag_info_t fragstatdiff;
1023   if (fnode->accounted_rstat.version == rstat_version)
1024     rstatdiff.add_delta(fnode->accounted_rstat, fnode->rstat);
1025   if (fnode->accounted_fragstat.version == dirstat_version)
1026     fragstatdiff.add_delta(fnode->accounted_fragstat, fnode->fragstat);
1027   dout(10) << " rstatdiff " << rstatdiff << " fragstatdiff " << fragstatdiff << dendl;
1028
1029   map<string_snap_t, MDSContext::vec > dentry_waiters;
1030   prepare_old_fragment(dentry_waiters, replay);
1031
1032   // create subfrag dirs
1033   int n = 0;
1034   for (const auto& fg : frags) {
1035     CDir *f = new CDir(inode, fg, mdcache, is_auth());
1036     f->state_set(state & (MASK_STATE_FRAGMENT_KEPT | STATE_COMPLETE));
1037     f->get_replicas() = get_replicas();
1038     f->pop_me = pop_me;
1039     f->pop_me.scale(fac);
1040
1041     // FIXME; this is an approximation
1042     f->pop_nested = pop_nested;
1043     f->pop_nested.scale(fac);
1044     f->pop_auth_subtree = pop_auth_subtree;
1045     f->pop_auth_subtree.scale(fac);
1046     f->pop_auth_subtree_nested = pop_auth_subtree_nested;
1047     f->pop_auth_subtree_nested.scale(fac);
1048
1049     dout(10) << " subfrag " << fg << " " << *f << dendl;
1050     subfrags[n++] = f;
1051     subs->push_back(f);
1052
1053     f->set_dir_auth(get_dir_auth());
1054     f->freeze_tree_state = freeze_tree_state;
1055     f->prepare_new_fragment(replay);
1056     f->init_fragment_pins();
1057   }
1058
1059   // repartition dentries
1060   while (!items.empty()) {
1061     auto p = items.begin();
1062
1063     CDentry *dn = p->second;
1064     frag_t subfrag = inode->pick_dirfrag(dn->get_name());
1065     int n = (subfrag.value() & (subfrag.mask() ^ frag.mask())) >> subfrag.mask_shift();
1066     dout(15) << " subfrag " << subfrag << " n=" << n << " for " << p->first << dendl;
1067     CDir *f = subfrags[n];
1068     f->steal_dentry(dn);
1069   }
1070
1071   for (const auto &p : dentry_waiters) {
1072     frag_t subfrag = inode->pick_dirfrag(p.first.name);
1073     int n = (subfrag.value() & (subfrag.mask() ^ frag.mask())) >> subfrag.mask_shift();
1074     CDir *f = subfrags[n];
1075
1076     if (f->waiting_on_dentry.empty())
1077       f->get(PIN_DNWAITER);
1078     auto &e = f->waiting_on_dentry[p.first];
1079     for (const auto &waiter : p.second) {
1080       e.push_back(waiter);
1081     }
1082   }
1083
1084   // FIXME: handle dirty old rstat
1085
1086   // fix up new frag fragstats
1087   for (int i = 0; i < n; i++) {
1088     CDir *f = subfrags[i];
1089     auto _fnode = f->_get_fnode();
1090     _fnode->version = f->projected_version = get_version();
1091     _fnode->rstat.version = rstat_version;
1092     _fnode->accounted_rstat = _fnode->rstat;
1093     _fnode->fragstat.version = dirstat_version;
1094     _fnode->accounted_fragstat = _fnode->fragstat;
1095     dout(10) << " rstat " << _fnode->rstat << " fragstat " << _fnode->fragstat
1096              << " on " << *f << dendl;
1097
1098     if (i == 0) {
1099       // give any outstanding frag stat differential to first frag
1100       dout(10) << " giving rstatdiff " << rstatdiff << " fragstatdiff" << fragstatdiff
1101                << " to " << *subfrags[0] << dendl;
1102       _fnode->accounted_rstat.add(rstatdiff);
1103       _fnode->accounted_fragstat.add(fragstatdiff);
1104     }
1105   }
1106
1107   finish_old_fragment(waiters, replay);
1108 }
1109
1110 void CDir::merge(const std::vector<CDir*>& subs, MDSContext::vec& waiters, bool replay)
1111 {
1112   dout(10) << "merge " << subs << dendl;
1113
1114   ceph_assert(subs.size() > 0);
1115
1116   set_dir_auth(subs.front()->get_dir_auth());
1117   freeze_tree_state = subs.front()->freeze_tree_state;
1118
1119   for (const auto& dir : subs) {
1120     ceph_assert(get_dir_auth() == dir->get_dir_auth());
1121     ceph_assert(freeze_tree_state == dir->freeze_tree_state);
1122   }
1123
1124   prepare_new_fragment(replay);
1125
1126   auto _fnode = _get_fnode();
1127
1128   nest_info_t rstatdiff;
1129   frag_info_t fragstatdiff;
1130   bool touched_mtime, touched_chattr;
1131   version_t rstat_version = inode->get_projected_inode()->rstat.version;
1132   version_t dirstat_version = inode->get_projected_inode()->dirstat.version;
1133
1134   map<string_snap_t, MDSContext::vec > dentry_waiters;
1135
1136   for (const auto& dir : subs) {
1137     dout(10) << " subfrag " << dir->get_frag() << " " << *dir << dendl;
1138     ceph_assert(!dir->is_auth() || dir->is_complete() || replay);
1139
1140     if (dir->get_fnode()->accounted_rstat.version == rstat_version)
1141       rstatdiff.add_delta(dir->get_fnode()->accounted_rstat, dir->get_fnode()->rstat);
1142     if (dir->get_fnode()->accounted_fragstat.version == dirstat_version)
1143       fragstatdiff.add_delta(dir->get_fnode()->accounted_fragstat, dir->get_fnode()->fragstat,
1144                              &touched_mtime, &touched_chattr);
1145
1146     dir->prepare_old_fragment(dentry_waiters, replay);
1147
1148     // steal dentries
1149     while (!dir->items.empty())
1150       steal_dentry(dir->items.begin()->second);
1151
1152     // merge replica map
1153     for (const auto &p : dir->get_replicas()) {
1154       unsigned cur = get_replicas()[p.first];
1155       if (p.second > cur)
1156         get_replicas()[p.first] = p.second;
1157     }
1158
1159     // merge version
1160     if (dir->get_version() > _fnode->version)
1161       _fnode->version = projected_version = dir->get_version();
1162
1163     // merge state
1164     state_set(dir->get_state() & MASK_STATE_FRAGMENT_KEPT);
1165
1166     dir->finish_old_fragment(waiters, replay);
1167     inode->close_dirfrag(dir->get_frag());
1168   }
1169
1170   if (!dentry_waiters.empty()) {
1171     get(PIN_DNWAITER);
1172     for (const auto &p : dentry_waiters) {
1173       auto &e = waiting_on_dentry[p.first];
1174       for (const auto &waiter : p.second) {
1175         e.push_back(waiter);
1176       }
1177     }
1178   }
1179
1180   if (is_auth() && !replay)
1181     mark_complete();
1182
1183   // FIXME: merge dirty old rstat
1184   _fnode->rstat.version = rstat_version;
1185   _fnode->accounted_rstat = _fnode->rstat;
1186   _fnode->accounted_rstat.add(rstatdiff);
1187
1188   _fnode->fragstat.version = dirstat_version;
1189   _fnode->accounted_fragstat = _fnode->fragstat;
1190   _fnode->accounted_fragstat.add(fragstatdiff);
1191
1192   init_fragment_pins();
1193 }
1194
1195
1196
1197
1198 void CDir::resync_accounted_fragstat()
1199 {
1200   auto pf = _get_projected_fnode();
1201   const auto& pi = inode->get_projected_inode();
1202
1203   if (pf->accounted_fragstat.version != pi->dirstat.version) {
1204     pf->fragstat.version = pi->dirstat.version;
1205     dout(10) << __func__ << " " << pf->accounted_fragstat << " -> " << pf->fragstat << dendl;
1206     pf->accounted_fragstat = pf->fragstat;
1207   }
1208 }
1209
1210 /*
1211  * resync rstat and accounted_rstat with inode
1212  */
1213 void CDir::resync_accounted_rstat()
1214 {
1215   auto pf = _get_projected_fnode();
1216   const auto& pi = inode->get_projected_inode();
1217
1218   if (pf->accounted_rstat.version != pi->rstat.version) {
1219     pf->rstat.version = pi->rstat.version;
1220     dout(10) << __func__ << " " << pf->accounted_rstat << " -> " << pf->rstat << dendl;
1221     pf->accounted_rstat = pf->rstat;
1222     dirty_old_rstat.clear();
1223   }
1224 }
1225
1226 void CDir::assimilate_dirty_rstat_inodes(MutationRef& mut)
1227 {
1228   dout(10) << __func__ << dendl;
1229   for (elist<CInode*>::iterator p = dirty_rstat_inodes.begin_use_current();
1230        !p.end(); ++p) {
1231     CInode *in = *p;
1232     ceph_assert(in->is_auth());
1233     if (in->is_frozen())
1234       continue;
1235
1236     mut->auth_pin(in);
1237
1238     auto pi = in->project_inode(mut);
1239     pi.inode->version = in->pre_dirty();
1240
1241     mdcache->project_rstat_inode_to_frag(mut, in, this, 0, 0, nullptr);
1242   }
1243   state_set(STATE_ASSIMRSTAT);
1244   dout(10) << __func__ << " done" << dendl;
1245 }
1246
1247 void CDir::assimilate_dirty_rstat_inodes_finish(EMetaBlob *blob)
1248 {
1249   if (!state_test(STATE_ASSIMRSTAT))
1250     return;
1251   state_clear(STATE_ASSIMRSTAT);
1252   dout(10) << __func__ << dendl;
1253   elist<CInode*>::iterator p = dirty_rstat_inodes.begin_use_current();
1254   while (!p.end()) {
1255     CInode *in = *p;
1256     ++p;
1257
1258     if (in->is_frozen())
1259       continue;
1260
1261     CDentry *dn = in->get_projected_parent_dn();
1262
1263     in->clear_dirty_rstat();
1264     blob->add_primary_dentry(dn, in, true);
1265   }
1266
1267   if (!dirty_rstat_inodes.empty())
1268     mdcache->mds->locker->mark_updated_scatterlock(&inode->nestlock);
1269 }
1270
1271
1272
1273
1274 /****************************************
1275  * WAITING
1276  */
1277
1278 void CDir::add_dentry_waiter(std::string_view dname, snapid_t snapid, MDSContext *c)
1279 {
1280   if (waiting_on_dentry.empty())
1281     get(PIN_DNWAITER);
1282   waiting_on_dentry[string_snap_t(dname, snapid)].push_back(c);
1283   dout(10) << __func__ << " dentry " << dname
1284            << " snap " << snapid
1285            << " " << c << " on " << *this << dendl;
1286 }
1287
1288 void CDir::take_dentry_waiting(std::string_view dname, snapid_t first, snapid_t last,
1289                                MDSContext::vec& ls)
1290 {
1291   if (waiting_on_dentry.empty())
1292     return;
1293
1294   string_snap_t lb(dname, first);
1295   string_snap_t ub(dname, last);
1296   auto it = waiting_on_dentry.lower_bound(lb);
1297   while (it != waiting_on_dentry.end() &&
1298          !(ub < it->first)) {
1299     dout(10) << __func__ << " " << dname
1300              << " [" << first << "," << last << "] found waiter on snap "
1301              << it->first.snapid
1302              << " on " << *this << dendl;
1303     for (const auto &waiter : it->second) {
1304       ls.push_back(waiter);
1305     }
1306     waiting_on_dentry.erase(it++);
1307   }
1308
1309   if (waiting_on_dentry.empty())
1310     put(PIN_DNWAITER);
1311 }
1312
1313 void CDir::take_sub_waiting(MDSContext::vec& ls)
1314 {
1315   dout(10) << __func__ << dendl;
1316   if (!waiting_on_dentry.empty()) {
1317     for (const auto &p : waiting_on_dentry) {
1318       for (const auto &waiter : p.second) {
1319         ls.push_back(waiter);
1320       }
1321     }
1322     waiting_on_dentry.clear();
1323     put(PIN_DNWAITER);
1324   }
1325 }
1326
1327
1328
1329 void CDir::add_waiter(uint64_t tag, MDSContext *c)
1330 {
1331   // hierarchical?
1332
1333   // at subtree root?
1334   if (tag & WAIT_ATSUBTREEROOT) {
1335     if (!is_subtree_root()) {
1336       // try parent
1337       dout(10) << "add_waiter " << std::hex << tag << std::dec << " " << c << " should be ATSUBTREEROOT, " << *this << " is not root, trying parent" << dendl;
1338       inode->parent->dir->add_waiter(tag, c);
1339       return;
1340     }
1341   }
1342
1343   ceph_assert(!(tag & WAIT_CREATED) || state_test(STATE_CREATING));
1344
1345   MDSCacheObject::add_waiter(tag, c);
1346 }
1347
1348
1349
1350 /* NOTE: this checks dentry waiters too */
1351 void CDir::take_waiting(uint64_t mask, MDSContext::vec& ls)
1352 {
1353   if ((mask & WAIT_DENTRY) && !waiting_on_dentry.empty()) {
1354     // take all dentry waiters
1355     for (const auto &p : waiting_on_dentry) {
1356       dout(10) << "take_waiting dentry " << p.first.name
1357                << " snap " << p.first.snapid << " on " << *this << dendl;
1358       for (const auto &waiter : p.second) {
1359         ls.push_back(waiter);
1360       }
1361     }
1362     waiting_on_dentry.clear();
1363     put(PIN_DNWAITER);
1364   }
1365
1366   // waiting
1367   MDSCacheObject::take_waiting(mask, ls);
1368 }
1369
1370
1371 void CDir::finish_waiting(uint64_t mask, int result)
1372 {
1373   dout(11) << __func__ << " mask " << hex << mask << dec << " result " << result << " on " << *this << dendl;
1374
1375   MDSContext::vec finished;
1376   take_waiting(mask, finished);
1377   if (result < 0)
1378     finish_contexts(g_ceph_context, finished, result);
1379   else
1380     mdcache->mds->queue_waiters(finished);
1381 }
1382
1383
1384
1385 // dirty/clean
1386
1387 CDir::fnode_ptr CDir::project_fnode(const MutationRef& mut)
1388 {
1389   ceph_assert(get_version() != 0);
1390
1391   if (mut && mut->is_projected(this))
1392     return std::const_pointer_cast<fnode_t>(projected_fnode.back());
1393
1394   auto pf = allocate_fnode(*get_projected_fnode());
1395
1396   if (scrub_infop && scrub_infop->last_scrub_dirty) {
1397     pf->localized_scrub_stamp = scrub_infop->last_local.time;
1398     pf->localized_scrub_version = scrub_infop->last_local.version;
1399     pf->recursive_scrub_stamp = scrub_infop->last_recursive.time;
1400     pf->recursive_scrub_version = scrub_infop->last_recursive.version;
1401     scrub_infop->last_scrub_dirty = false;
1402     scrub_maybe_delete_info();
1403   }
1404
1405   projected_fnode.emplace_back(pf);
1406   if (mut)
1407     mut->add_projected_node(this);
1408   dout(10) << __func__ <<  " " << pf.get() << dendl;
1409   return pf;
1410 }
1411
1412 void CDir::pop_and_dirty_projected_fnode(LogSegment *ls, const MutationRef& mut)
1413 {
1414   ceph_assert(!projected_fnode.empty());
1415   auto pf = std::move(projected_fnode.front());
1416   dout(15) << __func__ << " " << pf.get() << " v" << pf->version << dendl;
1417
1418   projected_fnode.pop_front();
1419   if (mut)
1420     mut->remove_projected_node(this);
1421
1422   reset_fnode(std::move(pf));
1423   _mark_dirty(ls);
1424 }
1425
1426 version_t CDir::pre_dirty(version_t min)
1427 {
1428   if (min > projected_version)
1429     projected_version = min;
1430   ++projected_version;
1431   dout(10) << __func__ << " " << projected_version << dendl;
1432   return projected_version;
1433 }
1434
1435 void CDir::mark_dirty(LogSegment *ls, version_t pv)
1436 {
1437   ceph_assert(is_auth());
1438
1439   if (pv) {
1440     ceph_assert(get_version() < pv);
1441     ceph_assert(pv <= projected_version);
1442     ceph_assert(!projected_fnode.empty() &&
1443                 pv <= projected_fnode.front()->version);
1444   }
1445
1446   _mark_dirty(ls);
1447 }
1448
1449 void CDir::_mark_dirty(LogSegment *ls)
1450 {
1451   if (!state_test(STATE_DIRTY)) {
1452     dout(10) << __func__ << " (was clean) " << *this << " version " << get_version() << dendl;
1453     _set_dirty_flag();
1454     ceph_assert(ls);
1455   } else {
1456     dout(10) << __func__ << " (already dirty) " << *this << " version " << get_version() << dendl;
1457   }
1458   if (ls) {
1459     ls->dirty_dirfrags.push_back(&item_dirty);
1460
1461     // if i've never committed, i need to be before _any_ mention of me is trimmed from the journal.
1462     if (committed_version == 0 && !item_new.is_on_list())
1463       ls->new_dirfrags.push_back(&item_new);
1464   }
1465 }
1466
1467 void CDir::mark_new(LogSegment *ls)
1468 {
1469   ls->new_dirfrags.push_back(&item_new);
1470   state_clear(STATE_CREATING);
1471
1472   MDSContext::vec waiters;
1473   take_waiting(CDir::WAIT_CREATED, waiters);
1474   mdcache->mds->queue_waiters(waiters);
1475 }
1476
1477 void CDir::set_fresh_fnode(fnode_const_ptr&& ptr) {
1478   ceph_assert(inode->is_auth());
1479   ceph_assert(!is_projected());
1480   ceph_assert(!state_test(STATE_COMMITTING));
1481   reset_fnode(std::move(ptr));
1482   projected_version = committing_version = committed_version = get_version();
1483
1484   if (state_test(STATE_REJOINUNDEF)) {
1485     ceph_assert(mdcache->mds->is_rejoin());
1486     state_clear(STATE_REJOINUNDEF);
1487     mdcache->opened_undef_dirfrag(this);
1488   }
1489 }
1490
1491 void CDir::mark_clean()
1492 {
1493   dout(10) << __func__ << " " << *this << " version " << get_version() << dendl;
1494   if (state_test(STATE_DIRTY)) {
1495     item_dirty.remove_myself();
1496     item_new.remove_myself();
1497
1498     state_clear(STATE_DIRTY);
1499     put(PIN_DIRTY);
1500   }
1501 }
1502
1503 // caller should hold auth pin of this
1504 void CDir::log_mark_dirty()
1505 {
1506   if (is_dirty() || projected_version > get_version())
1507     return; // noop if it is already dirty or will be dirty
1508
1509   auto _fnode = allocate_fnode(*get_fnode());
1510   _fnode->version = pre_dirty();
1511   reset_fnode(std::move(_fnode));
1512   mark_dirty(mdcache->mds->mdlog->get_current_segment());
1513 }
1514
1515 void CDir::mark_complete() {
1516   state_set(STATE_COMPLETE);
1517   bloom.reset();
1518 }
1519
1520 void CDir::first_get()
1521 {
1522   inode->get(CInode::PIN_DIRFRAG);
1523 }
1524
1525 void CDir::last_put()
1526 {
1527   inode->put(CInode::PIN_DIRFRAG);
1528 }
1529
1530
1531
1532 /******************************************************************************
1533  * FETCH and COMMIT
1534  */
1535
1536 // -----------------------
1537 // FETCH
1538 void CDir::fetch(MDSContext *c, bool ignore_authpinnability)
1539 {
1540   string want;
1541   return fetch(c, want, ignore_authpinnability);
1542 }
1543
1544 void CDir::fetch(MDSContext *c, std::string_view want_dn, bool ignore_authpinnability)
1545 {
1546   dout(10) << "fetch on " << *this << dendl;
1547
1548   ceph_assert(is_auth());
1549   ceph_assert(!is_complete());
1550
1551   if (!can_auth_pin() && !ignore_authpinnability) {
1552     if (c) {
1553       dout(7) << "fetch waiting for authpinnable" << dendl;
1554       add_waiter(WAIT_UNFREEZE, c);
1555     } else
1556       dout(7) << "fetch not authpinnable and no context" << dendl;
1557     return;
1558   }
1559
1560   // unlinked directory inode shouldn't have any entry
1561   if (!inode->is_base() && get_parent_dir()->inode->is_stray() &&
1562       !inode->snaprealm) {
1563     dout(7) << "fetch dirfrag for unlinked directory, mark complete" << dendl;
1564     if (get_version() == 0) {
1565       auto _fnode = allocate_fnode();
1566       _fnode->version = 1;
1567       set_fresh_fnode(std::move(_fnode));
1568     }
1569     mark_complete();
1570
1571     if (c)
1572       mdcache->mds->queue_waiter(c);
1573     return;
1574   }
1575
1576   if (c) add_waiter(WAIT_COMPLETE, c);
1577   if (!want_dn.empty()) wanted_items.insert(mempool::mds_co::string(want_dn));
1578
1579   // already fetching?
1580   if (state_test(CDir::STATE_FETCHING)) {
1581     dout(7) << "already fetching; waiting" << dendl;
1582     return;
1583   }
1584
1585   auth_pin(this);
1586   state_set(CDir::STATE_FETCHING);
1587
1588   if (mdcache->mds->logger) mdcache->mds->logger->inc(l_mds_dir_fetch);
1589
1590   mdcache->mds->balancer->hit_dir(this, META_POP_FETCH);
1591
1592   std::set<dentry_key_t> empty;
1593   _omap_fetch(NULL, empty);
1594 }
1595
1596 void CDir::fetch(MDSContext *c, const std::set<dentry_key_t>& keys)
1597 {
1598   dout(10) << "fetch " << keys.size() << " keys on " << *this << dendl;
1599
1600   ceph_assert(is_auth());
1601   ceph_assert(!is_complete());
1602
1603   if (!can_auth_pin()) {
1604     dout(7) << "fetch keys waiting for authpinnable" << dendl;
1605     add_waiter(WAIT_UNFREEZE, c);
1606     return;
1607   }
1608   if (state_test(CDir::STATE_FETCHING)) {
1609     dout(7) << "fetch keys waiting for full fetch" << dendl;
1610     add_waiter(WAIT_COMPLETE, c);
1611     return;
1612   }
1613
1614   auth_pin(this);
1615   if (mdcache->mds->logger) mdcache->mds->logger->inc(l_mds_dir_fetch);
1616
1617   mdcache->mds->balancer->hit_dir(this, META_POP_FETCH);
1618
1619   _omap_fetch(c, keys);
1620 }
1621
1622 class C_IO_Dir_OMAP_FetchedMore : public CDirIOContext {
1623   MDSContext *fin;
1624 public:
1625   const version_t omap_version;
1626   bufferlist hdrbl;
1627   bool more = false;
1628   map<string, bufferlist> omap;      ///< carry-over from before
1629   map<string, bufferlist> omap_more; ///< new batch
1630   int ret;
1631   C_IO_Dir_OMAP_FetchedMore(CDir *d, version_t v, MDSContext *f) :
1632     CDirIOContext(d), fin(f), omap_version(v), ret(0) { }
1633   void finish(int r) {
1634     if (omap_version < dir->get_committed_version()) {
1635       omap.clear();
1636       dir->_omap_fetch(fin, {});
1637       return;
1638     }
1639
1640     // merge results
1641     if (omap.empty()) {
1642       omap.swap(omap_more);
1643     } else {
1644       omap.insert(omap_more.begin(), omap_more.end());
1645     }
1646     if (more) {
1647       dir->_omap_fetch_more(omap_version, hdrbl, omap, fin);
1648     } else {
1649       dir->_omap_fetched(hdrbl, omap, !fin, r);
1650       if (fin)
1651         fin->complete(r);
1652     }
1653   }
1654   void print(ostream& out) const override {
1655     out << "dirfrag_fetch_more(" << dir->dirfrag() << ")";
1656   }
1657 };
1658
1659 class C_IO_Dir_OMAP_Fetched : public CDirIOContext {
1660   MDSContext *fin;
1661 public:
1662   const version_t omap_version;
1663   bufferlist hdrbl;
1664   bool more = false;
1665   map<string, bufferlist> omap;
1666   bufferlist btbl;
1667   int ret1, ret2, ret3;
1668
1669   C_IO_Dir_OMAP_Fetched(CDir *d, MDSContext *f) :
1670     CDirIOContext(d), fin(f),
1671     omap_version(d->get_committing_version()),
1672     ret1(0), ret2(0), ret3(0) { }
1673   void finish(int r) override {
1674     // check the correctness of backtrace
1675     if (r >= 0 && ret3 != -CEPHFS_ECANCELED)
1676       dir->inode->verify_diri_backtrace(btbl, ret3);
1677     if (r >= 0) r = ret1;
1678     if (r >= 0) r = ret2;
1679
1680     if (more) {
1681       if (omap_version < dir->get_committed_version()) {
1682         omap.clear();
1683         dir->_omap_fetch(fin, {});
1684       } else {
1685         dir->_omap_fetch_more(omap_version, hdrbl, omap, fin);
1686       }
1687       return;
1688     }
1689
1690     dir->_omap_fetched(hdrbl, omap, !fin, r);
1691     if (fin)
1692       fin->complete(r);
1693
1694   }
1695   void print(ostream& out) const override {
1696     out << "dirfrag_fetch(" << dir->dirfrag() << ")";
1697   }
1698 };
1699
1700 void CDir::_omap_fetch(MDSContext *c, const std::set<dentry_key_t>& keys)
1701 {
1702   C_IO_Dir_OMAP_Fetched *fin = new C_IO_Dir_OMAP_Fetched(this, c);
1703   object_t oid = get_ondisk_object();
1704   object_locator_t oloc(mdcache->mds->mdsmap->get_metadata_pool());
1705   ObjectOperation rd;
1706   rd.omap_get_header(&fin->hdrbl, &fin->ret1);
1707   if (keys.empty()) {
1708     ceph_assert(!c);
1709     rd.omap_get_vals("", "", g_conf()->mds_dir_keys_per_op,
1710                      &fin->omap, &fin->more, &fin->ret2);
1711   } else {
1712     ceph_assert(c);
1713     std::set<std::string> str_keys;
1714     for (auto p : keys) {
1715       string str;
1716       p.encode(str);
1717       str_keys.insert(str);
1718     }
1719     rd.omap_get_vals_by_keys(str_keys, &fin->omap, &fin->ret2);
1720   }
1721   // check the correctness of backtrace
1722   if (g_conf()->mds_verify_backtrace > 0 && frag == frag_t()) {
1723     rd.getxattr("parent", &fin->btbl, &fin->ret3);
1724     rd.set_last_op_flags(CEPH_OSD_OP_FLAG_FAILOK);
1725   } else {
1726     fin->ret3 = -CEPHFS_ECANCELED;
1727   }
1728
1729   mdcache->mds->objecter->read(oid, oloc, rd, CEPH_NOSNAP, NULL, 0,
1730                              new C_OnFinisher(fin, mdcache->mds->finisher));
1731 }
1732
1733 void CDir::_omap_fetch_more(version_t omap_version, bufferlist& hdrbl,
1734                             map<string, bufferlist>& omap, MDSContext *c)
1735 {
1736   // we have more omap keys to fetch!
1737   object_t oid = get_ondisk_object();
1738   object_locator_t oloc(mdcache->mds->mdsmap->get_metadata_pool());
1739   auto fin = new C_IO_Dir_OMAP_FetchedMore(this, omap_version, c);
1740   fin->hdrbl = std::move(hdrbl);
1741   fin->omap.swap(omap);
1742   ObjectOperation rd;
1743   rd.omap_get_vals(fin->omap.rbegin()->first,
1744                    "", /* filter prefix */
1745                    g_conf()->mds_dir_keys_per_op,
1746                    &fin->omap_more,
1747                    &fin->more,
1748                    &fin->ret);
1749   mdcache->mds->objecter->read(oid, oloc, rd, CEPH_NOSNAP, NULL, 0,
1750                              new C_OnFinisher(fin, mdcache->mds->finisher));
1751 }
1752
1753 CDentry *CDir::_load_dentry(
1754     std::string_view key,
1755     std::string_view dname,
1756     const snapid_t last,
1757     bufferlist &bl,
1758     const int pos,
1759     const std::set<snapid_t> *snaps,
1760     double rand_threshold,
1761     bool *force_dirty)
1762 {
1763   auto q = bl.cbegin();
1764
1765   snapid_t first;
1766   decode(first, q);
1767
1768   // marker
1769   char type;
1770   decode(type, q);
1771
1772   dout(20) << "_fetched pos " << pos << " marker '" << type << "' dname '" << dname
1773            << " [" << first << "," << last << "]"
1774            << dendl;
1775
1776   bool stale = false;
1777   if (snaps && last != CEPH_NOSNAP) {
1778     set<snapid_t>::const_iterator p = snaps->lower_bound(first);
1779     if (p == snaps->end() || *p > last) {
1780       dout(10) << " skipping stale dentry on [" << first << "," << last << "]" << dendl;
1781       stale = true;
1782     }
1783   }
1784
1785   /*
1786    * look for existing dentry for _last_ snap, because unlink +
1787    * create may leave a "hole" (epochs during which the dentry
1788    * doesn't exist) but for which no explicit negative dentry is in
1789    * the cache.
1790    */
1791   CDentry *dn;
1792   if (stale)
1793     dn = lookup_exact_snap(dname, last);
1794   else
1795     dn = lookup(dname, last);
1796
1797   if (type == 'L' || type == 'l') {
1798     // hard link
1799     inodeno_t ino;
1800     unsigned char d_type;
1801     mempool::mds_co::string alternate_name;
1802
1803     CDentry::decode_remote(type, ino, d_type, alternate_name, q);
1804
1805     if (stale) {
1806       if (!dn) {
1807         stale_items.insert(mempool::mds_co::string(key));
1808         *force_dirty = true;
1809       }
1810       return dn;
1811     }
1812
1813     if (dn) {
1814       CDentry::linkage_t *dnl = dn->get_linkage();
1815       dout(12) << "_fetched had " << (dnl->is_null() ? "NEG" : "") << " dentry " << *dn << dendl;
1816       if (committed_version == 0 &&
1817           dnl->is_remote() &&
1818           dn->is_dirty() &&
1819           ino == dnl->get_remote_ino() &&
1820           d_type == dnl->get_remote_d_type() &&
1821           alternate_name == dn->get_alternate_name()) {
1822         // see comment below
1823         dout(10) << "_fetched  had underwater dentry " << *dn << ", marking clean" << dendl;
1824         dn->mark_clean();
1825       }
1826     } else {
1827       // (remote) link
1828       dn = add_remote_dentry(dname, ino, d_type, std::move(alternate_name), first, last);
1829
1830       // link to inode?
1831       CInode *in = mdcache->get_inode(ino);   // we may or may not have it.
1832       if (in) {
1833         dn->link_remote(dn->get_linkage(), in);
1834         dout(12) << "_fetched  got remote link " << ino << " which we have " << *in << dendl;
1835       } else {
1836         dout(12) << "_fetched  got remote link " << ino << " (don't have it)" << dendl;
1837       }
1838     }
1839   }
1840   else if (type == 'I' || type == 'i') {
1841     InodeStore inode_data;
1842     mempool::mds_co::string alternate_name;
1843     // inode
1844     // Load inode data before looking up or constructing CInode
1845     if (type == 'i') {
1846       DECODE_START(2, q);
1847       if (struct_v >= 2) {
1848         decode(alternate_name, q);
1849       }
1850       inode_data.decode(q);
1851       DECODE_FINISH(q);
1852     } else {
1853       inode_data.decode_bare(q);
1854     }
1855
1856     if (stale) {
1857       if (!dn) {
1858         stale_items.insert(mempool::mds_co::string(key));
1859         *force_dirty = true;
1860       }
1861       return dn;
1862     }
1863
1864     bool undef_inode = false;
1865     if (dn) {
1866       CDentry::linkage_t *dnl = dn->get_linkage();
1867       dout(12) << "_fetched had " << (dnl->is_null() ? "NEG" : "") << " dentry " << *dn << dendl;
1868
1869       if (dnl->is_primary()) {
1870         CInode *in = dnl->get_inode();
1871         if (in->state_test(CInode::STATE_REJOINUNDEF)) {
1872           undef_inode = true;
1873         } else if (committed_version == 0 &&
1874                    dn->is_dirty() &&
1875                    inode_data.inode->ino == in->ino() &&
1876                    inode_data.inode->version == in->get_version()) {
1877           /* clean underwater item?
1878            * Underwater item is something that is dirty in our cache from
1879            * journal replay, but was previously flushed to disk before the
1880            * mds failed.
1881            *
1882            * We only do this is committed_version == 0. that implies either
1883            * - this is a fetch after from a clean/empty CDir is created
1884            *   (and has no effect, since the dn won't exist); or
1885            * - this is a fetch after _recovery_, which is what we're worried
1886            *   about.  Items that are marked dirty from the journal should be
1887            *   marked clean if they appear on disk.
1888            */
1889           dout(10) << "_fetched  had underwater dentry " << *dn << ", marking clean" << dendl;
1890           dn->mark_clean();
1891           dout(10) << "_fetched  had underwater inode " << *dnl->get_inode() << ", marking clean" << dendl;
1892           in->mark_clean();
1893         }
1894       }
1895     }
1896
1897     if (!dn || undef_inode) {
1898       // add inode
1899       CInode *in = mdcache->get_inode(inode_data.inode->ino, last);
1900       if (!in || undef_inode) {
1901         if (undef_inode && in)
1902           in->first = first;
1903         else
1904           in = new CInode(mdcache, true, first, last);
1905
1906         in->reset_inode(std::move(inode_data.inode));
1907         in->reset_xattrs(std::move(inode_data.xattrs));
1908         // symlink?
1909         if (in->is_symlink())
1910           in->symlink = inode_data.symlink;
1911
1912         in->dirfragtree.swap(inode_data.dirfragtree);
1913         in->reset_old_inodes(std::move(inode_data.old_inodes));
1914         if (in->is_any_old_inodes()) {
1915           snapid_t min_first = in->get_old_inodes()->rbegin()->first + 1;
1916           if (min_first > in->first)
1917             in->first = min_first;
1918         }
1919
1920         in->oldest_snap = inode_data.oldest_snap;
1921         in->decode_snap_blob(inode_data.snap_blob);
1922         if (snaps && !in->snaprealm)
1923           in->purge_stale_snap_data(*snaps);
1924
1925         if (!undef_inode) {
1926           mdcache->add_inode(in); // add
1927           dn = add_primary_dentry(dname, in, std::move(alternate_name), first, last); // link
1928         }
1929         dout(12) << "_fetched  got " << *dn << " " << *in << dendl;
1930
1931         if (in->get_inode()->is_dirty_rstat())
1932           in->mark_dirty_rstat();
1933
1934         in->maybe_ephemeral_rand(rand_threshold);
1935         //in->hack_accessed = false;
1936         //in->hack_load_stamp = ceph_clock_now();
1937         //num_new_inodes_loaded++;
1938       } else if (g_conf().get_val<bool>("mds_hack_allow_loading_invalid_metadata")) {
1939         dout(20) << "hack: adding duplicate dentry for " << *in << dendl;
1940         dn = add_primary_dentry(dname, in, std::move(alternate_name), first, last);
1941       } else {
1942         dout(0) << "_fetched  badness: got (but i already had) " << *in
1943                 << " mode " << in->get_inode()->mode
1944                 << " mtime " << in->get_inode()->mtime << dendl;
1945         string dirpath, inopath;
1946         this->inode->make_path_string(dirpath);
1947         in->make_path_string(inopath);
1948         mdcache->mds->clog->error() << "loaded dup inode " << inode_data.inode->ino
1949           << " [" << first << "," << last << "] v" << inode_data.inode->version
1950           << " at " << dirpath << "/" << dname
1951           << ", but inode " << in->vino() << " v" << in->get_version()
1952           << " already exists at " << inopath;
1953         return dn;
1954       }
1955     }
1956   } else {
1957     CachedStackStringStream css;
1958     *css << "Invalid tag char '" << type << "' pos " << pos;
1959     throw buffer::malformed_input(css->str());
1960   }
1961
1962   return dn;
1963 }
1964
1965 void CDir::_omap_fetched(bufferlist& hdrbl, map<string, bufferlist>& omap,
1966                          bool complete, int r)
1967 {
1968   LogChannelRef clog = mdcache->mds->clog;
1969   dout(10) << "_fetched header " << hdrbl.length() << " bytes "
1970            << omap.size() << " keys for " << *this << dendl;
1971
1972   ceph_assert(r == 0 || r == -CEPHFS_ENOENT || r == -CEPHFS_ENODATA);
1973   ceph_assert(is_auth());
1974   ceph_assert(!is_frozen());
1975
1976   if (hdrbl.length() == 0) {
1977     dout(0) << "_fetched missing object for " << *this << dendl;
1978
1979     clog->error() << "dir " << dirfrag() << " object missing on disk; some "
1980                      "files may be lost (" << get_path() << ")";
1981
1982     go_bad(complete);
1983     return;
1984   }
1985
1986   fnode_t got_fnode;
1987   {
1988     auto p = hdrbl.cbegin();
1989     try {
1990       decode(got_fnode, p);
1991     } catch (const buffer::error &err) {
1992       derr << "Corrupt fnode in dirfrag " << dirfrag()
1993            << ": " << err.what() << dendl;
1994       clog->warn() << "Corrupt fnode header in " << dirfrag() << ": "
1995                    << err.what() << " (" << get_path() << ")";
1996       go_bad(complete);
1997       return;
1998     }
1999     if (!p.end()) {
2000       clog->warn() << "header buffer of dir " << dirfrag() << " has "
2001                   << hdrbl.length() - p.get_off() << " extra bytes ("
2002                   << get_path() << ")";
2003       go_bad(complete);
2004       return;
2005     }
2006   }
2007
2008   dout(10) << "_fetched version " << got_fnode.version << dendl;
2009
2010   // take the loaded fnode?
2011   // only if we are a fresh CDir* with no prior state.
2012   if (get_version() == 0) {
2013     set_fresh_fnode(allocate_fnode(got_fnode));
2014   }
2015
2016   list<CInode*> undef_inodes;
2017
2018   // purge stale snaps?
2019   bool force_dirty = false;
2020   const set<snapid_t> *snaps = NULL;
2021   SnapRealm *realm = inode->find_snaprealm();
2022   if (fnode->snap_purged_thru < realm->get_last_destroyed()) {
2023     snaps = &realm->get_snaps();
2024     dout(10) << " snap_purged_thru " << fnode->snap_purged_thru
2025              << " < " << realm->get_last_destroyed()
2026              << ", snap purge based on " << *snaps << dendl;
2027     if (get_num_snap_items() == 0) {
2028       const_cast<snapid_t&>(fnode->snap_purged_thru) = realm->get_last_destroyed();
2029       force_dirty = true;
2030     }
2031   }
2032
2033   int count = 0;
2034   unsigned pos = omap.size() - 1;
2035   double rand_threshold = get_inode()->get_ephemeral_rand();
2036   for (map<string, bufferlist>::reverse_iterator p = omap.rbegin();
2037        p != omap.rend();
2038        ++p, --pos) {
2039     string dname;
2040     snapid_t last;
2041     dentry_key_t::decode_helper(p->first, dname, last);
2042
2043     if (!(++count % mdcache->mds->heartbeat_reset_grace(2)))
2044       mdcache->mds->heartbeat_reset();
2045
2046     CDentry *dn = NULL;
2047     try {
2048       dn = _load_dentry(
2049             p->first, dname, last, p->second, pos, snaps,
2050             rand_threshold, &force_dirty);
2051     } catch (const buffer::error &err) {
2052       mdcache->mds->clog->warn() << "Corrupt dentry '" << dname << "' in "
2053                                   "dir frag " << dirfrag() << ": "
2054                                << err.what() << "(" << get_path() << ")";
2055
2056       // Remember that this dentry is damaged.  Subsequent operations
2057       // that try to act directly on it will get their CEPHFS_EIOs, but this
2058       // dirfrag as a whole will continue to look okay (minus the
2059       // mysteriously-missing dentry)
2060       go_bad_dentry(last, dname);
2061
2062       // Anyone who was WAIT_DENTRY for this guy will get kicked
2063       // to RetryRequest, and hit the DamageTable-interrogating path.
2064       // Stats will now be bogus because we will think we're complete,
2065       // but have 1 or more missing dentries.
2066       continue;
2067     }
2068
2069     if (!dn)
2070       continue;
2071
2072     CDentry::linkage_t *dnl = dn->get_linkage();
2073     if (dnl->is_primary() && dnl->get_inode()->state_test(CInode::STATE_REJOINUNDEF))
2074       undef_inodes.push_back(dnl->get_inode());
2075
2076     if (wanted_items.count(mempool::mds_co::string(dname)) > 0 || !complete) {
2077       dout(10) << " touching wanted dn " << *dn << dendl;
2078       mdcache->touch_dentry(dn);
2079
2080       if (!(++count % mdcache->mds->heartbeat_reset_grace(2)))
2081         mdcache->mds->heartbeat_reset();
2082     }
2083   }
2084
2085   //cache->mds->logger->inc("newin", num_new_inodes_loaded);
2086
2087   // mark complete, !fetching
2088   if (complete) {
2089     wanted_items.clear();
2090     mark_complete();
2091     state_clear(STATE_FETCHING);
2092   }
2093
2094   // open & force frags
2095   while (!undef_inodes.empty()) {
2096     CInode *in = undef_inodes.front();
2097
2098     undef_inodes.pop_front();
2099     in->state_clear(CInode::STATE_REJOINUNDEF);
2100     mdcache->opened_undef_inode(in);
2101
2102     if (!(++count % mdcache->mds->heartbeat_reset_grace()))
2103       mdcache->mds->heartbeat_reset();
2104   }
2105
2106   // dirty myself to remove stale snap dentries
2107   if (force_dirty && !mdcache->is_readonly())
2108     log_mark_dirty();
2109
2110   auth_unpin(this);
2111
2112   if (complete) {
2113     // kick waiters
2114     finish_waiting(WAIT_COMPLETE, 0);
2115   }
2116 }
2117
2118 void CDir::go_bad_dentry(snapid_t last, std::string_view dname)
2119 {
2120   dout(10) << __func__ << " " << dname << dendl;
2121   std::string path(get_path());
2122   path += "/";
2123   path += dname;
2124   const bool fatal = mdcache->mds->damage_table.notify_dentry(
2125       inode->ino(), frag, last, dname, path);
2126   if (fatal) {
2127     mdcache->mds->damaged();
2128     ceph_abort();  // unreachable, damaged() respawns us
2129   }
2130 }
2131
2132 void CDir::go_bad(bool complete)
2133 {
2134   dout(10) << __func__ << " " << frag << dendl;
2135   const bool fatal = mdcache->mds->damage_table.notify_dirfrag(
2136       inode->ino(), frag, get_path());
2137   if (fatal) {
2138     mdcache->mds->damaged();
2139     ceph_abort();  // unreachable, damaged() respawns us
2140   }
2141
2142   if (complete) {
2143     if (get_version() == 0) {
2144       auto _fnode = allocate_fnode();
2145       _fnode->version = 1;
2146       reset_fnode(std::move(_fnode));
2147     }
2148
2149     state_set(STATE_BADFRAG);
2150     mark_complete();
2151   }
2152
2153   state_clear(STATE_FETCHING);
2154   auth_unpin(this);
2155   finish_waiting(WAIT_COMPLETE, -CEPHFS_EIO);
2156 }
2157
2158 // -----------------------
2159 // COMMIT
2160
2161 /**
2162  * commit
2163  *
2164  * @param want - min version i want committed
2165  * @param c - callback for completion
2166  */
2167 void CDir::commit(version_t want, MDSContext *c, bool ignore_authpinnability, int op_prio)
2168 {
2169   dout(10) << "commit want " << want << " on " << *this << dendl;
2170   if (want == 0) want = get_version();
2171
2172   // preconditions
2173   ceph_assert(want <= get_version() || get_version() == 0);    // can't commit the future
2174   ceph_assert(want > committed_version); // the caller is stupid
2175   ceph_assert(is_auth());
2176   ceph_assert(ignore_authpinnability || can_auth_pin());
2177
2178   // note: queue up a noop if necessary, so that we always
2179   // get an auth_pin.
2180   if (!c)
2181     c = new C_MDSInternalNoop;
2182
2183   // auth_pin on first waiter
2184   if (waiting_for_commit.empty())
2185     auth_pin(this);
2186   waiting_for_commit[want].push_back(c);
2187
2188   // ok.
2189   _commit(want, op_prio);
2190 }
2191
2192 class C_IO_Dir_Committed : public CDirIOContext {
2193   version_t version;
2194 public:
2195   C_IO_Dir_Committed(CDir *d, version_t v) : CDirIOContext(d), version(v) { }
2196   void finish(int r) override {
2197     dir->_committed(r, version);
2198   }
2199   void print(ostream& out) const override {
2200     out << "dirfrag_committed(" << dir->dirfrag() << ")";
2201   }
2202 };
2203
2204 class C_IO_Dir_Commit_Ops : public Context {
2205 public:
2206   C_IO_Dir_Commit_Ops(CDir *d, int pr,
2207                       vector<CDir::dentry_commit_item> &&s, bufferlist &&bl,
2208                       vector<string> &&r,
2209                       mempool::mds_co::compact_set<mempool::mds_co::string> &&stales) :
2210     dir(d), op_prio(pr) {
2211     metapool = dir->mdcache->mds->get_metadata_pool();
2212     version = dir->get_version();
2213     is_new = dir->is_new();
2214     to_set.swap(s);
2215     dfts.swap(bl);
2216     to_remove.swap(r);
2217     stale_items.swap(stales);
2218   }
2219
2220   void finish(int r) override {
2221     dir->_omap_commit_ops(r, op_prio, metapool, version, is_new, to_set, dfts,
2222                           to_remove, stale_items);
2223   }
2224
2225 private:
2226   CDir *dir;
2227   int op_prio;
2228   int64_t metapool;
2229   version_t version;
2230   bool is_new;
2231   vector<CDir::dentry_commit_item> to_set;
2232   bufferlist dfts;
2233   vector<string> to_remove;
2234   mempool::mds_co::compact_set<mempool::mds_co::string> stale_items;
2235 };
2236
2237 // This is doing the same thing with the InodeStoreBase::encode()
2238 void CDir::_encode_primary_inode_base(dentry_commit_item &item, bufferlist &dfts,
2239                                       bufferlist &bl)
2240 {
2241   ENCODE_START(6, 4, bl);
2242   encode(*item.inode, bl, item.features);
2243
2244   if (!item.symlink.empty())
2245     encode(item.symlink, bl);
2246
2247   // dirfragtree
2248   dfts.splice(0, item.dft_len, &bl);
2249
2250   if (item.xattrs)
2251     encode(*item.xattrs, bl);
2252   else
2253     encode((__u32)0, bl);
2254
2255   if (item.snaprealm) {
2256     bufferlist snapr_bl;
2257     encode(item.srnode, snapr_bl);
2258     encode(snapr_bl, bl);
2259   } else {
2260     encode(bufferlist(), bl);
2261   }
2262
2263   if (item.old_inodes)
2264     encode(*item.old_inodes, bl, item.features);
2265   else
2266     encode((__u32)0, bl);
2267
2268   encode(item.oldest_snap, bl);
2269   encode(item.damage_flags, bl);
2270   ENCODE_FINISH(bl);
2271 }
2272
2273 // This is not locked by mds_lock
2274 void CDir::_omap_commit_ops(int r, int op_prio, int64_t metapool, version_t version, bool _new,
2275                             vector<dentry_commit_item> &to_set, bufferlist &dfts,
2276                             vector<string>& to_remove,
2277                             mempool::mds_co::compact_set<mempool::mds_co::string> &stales)
2278 {
2279   dout(10) << __func__ << dendl;
2280
2281   if (r < 0) {
2282     mdcache->mds->handle_write_error_with_lock(r);
2283     return;
2284   }
2285
2286   C_GatherBuilder gather(g_ceph_context,
2287                          new C_OnFinisher(new C_IO_Dir_Committed(this, version),
2288                          mdcache->mds->finisher));
2289
2290   SnapContext snapc;
2291   object_t oid = get_ondisk_object();
2292   object_locator_t oloc(metapool);
2293
2294   map<string, bufferlist> _set;
2295   set<string> _rm;
2296
2297   unsigned max_write_size = mdcache->max_dir_commit_size;
2298   unsigned write_size = 0;
2299
2300   auto commit_one = [&](bool header=false) {
2301     ObjectOperation op;
2302
2303     // don't create new dirfrag blindly
2304     if (!_new)
2305       op.stat(nullptr, nullptr, nullptr);
2306
2307     /*
2308      * save the header at the last moment.. If we were to send it off before
2309      * other updates, but die before sending them all, we'd think that the
2310      * on-disk state was fully committed even though it wasn't! However, since
2311      * the messages are strictly ordered between the MDS and the OSD, and
2312      * since messages to a given PG are strictly ordered, if we simply send
2313      * the message containing the header off last, we cannot get our header
2314      * into an incorrect state.
2315      */
2316     if (header) {
2317       bufferlist header;
2318       encode(*fnode, header);
2319       op.omap_set_header(header);
2320     }
2321
2322     op.priority = op_prio;
2323     if (!_set.empty())
2324       op.omap_set(_set);
2325     if (!_rm.empty())
2326       op.omap_rm_keys(_rm);
2327     mdcache->mds->objecter->mutate(oid, oloc, op, snapc,
2328                                    ceph::real_clock::now(),
2329                                    0, gather.new_sub());
2330     write_size = 0;
2331     _set.clear();
2332     _rm.clear();
2333   };
2334
2335   int count = 0;
2336   for (auto &key : stales) {
2337     unsigned size = key.length() + sizeof(__u32);
2338     if (write_size + size > max_write_size)
2339       commit_one();
2340
2341     write_size += size;
2342     _rm.emplace(key);
2343
2344     if (!(++count % mdcache->mds->heartbeat_reset_grace(2)))
2345       mdcache->mds->heartbeat_reset();
2346   }
2347
2348   for (auto &key : to_remove) {
2349     unsigned size = key.length() + sizeof(__u32);
2350     if (write_size + size > max_write_size)
2351       commit_one();
2352
2353     write_size += size;
2354     _rm.emplace(std::move(key));
2355
2356     if (!(++count % mdcache->mds->heartbeat_reset_grace(2)))
2357       mdcache->mds->heartbeat_reset();
2358   }
2359
2360   bufferlist bl;
2361   using ceph::encode;
2362   for (auto &item : to_set) {
2363     encode(item.first, bl);
2364     if (item.is_remote) {
2365       // remote link
2366       CDentry::encode_remote(item.ino, item.d_type, item.alternate_name, bl);
2367     } else {
2368       // marker, name, inode, [symlink string]
2369       bl.append('i');         // inode
2370
2371       ENCODE_START(2, 1, bl);
2372       encode(item.alternate_name, bl);
2373       _encode_primary_inode_base(item, dfts, bl);
2374       ENCODE_FINISH(bl);
2375     }
2376
2377     unsigned size = item.key.length() + bl.length() + 2 * sizeof(__u32);
2378     if (write_size + size > max_write_size)
2379       commit_one();
2380
2381     write_size += size;
2382     _set[std::move(item.key)].swap(bl);
2383
2384     if (!(++count % mdcache->mds->heartbeat_reset_grace()))
2385       mdcache->mds->heartbeat_reset();
2386   }
2387
2388   commit_one(true);
2389   gather.activate();
2390 }
2391
2392 /**
2393  * Flush out the modified dentries in this dir. Keep the bufferlist
2394  * below max_write_size;
2395  */
2396 void CDir::_omap_commit(int op_prio)
2397 {
2398   dout(10) << __func__ << dendl;
2399
2400   if (op_prio < 0)
2401     op_prio = CEPH_MSG_PRIO_DEFAULT;
2402
2403   // snap purge?
2404   const set<snapid_t> *snaps = NULL;
2405   SnapRealm *realm = inode->find_snaprealm();
2406   if (fnode->snap_purged_thru < realm->get_last_destroyed()) {
2407     snaps = &realm->get_snaps();
2408     dout(10) << " snap_purged_thru " << fnode->snap_purged_thru
2409              << " < " << realm->get_last_destroyed()
2410              << ", snap purge based on " << *snaps << dendl;
2411     // fnode.snap_purged_thru = realm->get_last_destroyed();
2412   }
2413
2414   size_t items_count = 0;
2415   if (state_test(CDir::STATE_FRAGMENTING) && is_new()) {
2416     items_count = get_num_head_items() + get_num_snap_items();
2417   } else {
2418     for (elist<CDentry*>::iterator it = dirty_dentries.begin(); !it.end(); ++it)
2419       ++items_count;
2420   }
2421
2422   vector<string> to_remove;
2423   // reverve enough memories, which maybe larger than the actually needed
2424   to_remove.reserve(items_count);
2425
2426   vector<dentry_commit_item> to_set;
2427   // reverve enough memories, which maybe larger than the actually needed
2428   to_set.reserve(items_count);
2429
2430   // for dir fragtrees
2431   bufferlist dfts(CEPH_PAGE_SIZE);
2432
2433   auto write_one = [&](CDentry *dn) {
2434     string key;
2435     dn->key().encode(key);
2436
2437     if (dn->last != CEPH_NOSNAP &&
2438         snaps && try_trim_snap_dentry(dn, *snaps)) {
2439       dout(10) << " rm " << key << dendl;
2440       to_remove.emplace_back(std::move(key));
2441       return;
2442     }
2443
2444     if (dn->get_linkage()->is_null()) {
2445       dout(10) << " rm " << dn->get_name() << " " << *dn << dendl;
2446       to_remove.emplace_back(std::move(key));
2447     } else {
2448       dout(10) << " set " << dn->get_name() << " " << *dn << dendl;
2449
2450       uint64_t off = dfts.length();
2451       // try to reserve new size if there has less
2452       // than 1/8 page space
2453       uint64_t left = CEPH_PAGE_SIZE - off % CEPH_PAGE_SIZE;
2454       if (left < CEPH_PAGE_SIZE / 8)
2455         dfts.reserve(left + CEPH_PAGE_SIZE);
2456
2457       auto& item = to_set.emplace_back();
2458       item.key = std::move(key);
2459       _parse_dentry(dn, item, snaps, dfts);
2460       item.dft_len = dfts.length() - off;
2461     }
2462   };
2463
2464   int count = 0;
2465   if (state_test(CDir::STATE_FRAGMENTING) && is_new()) {
2466     ceph_assert(committed_version == 0);
2467     for (auto p = items.begin(); p != items.end(); ) {
2468       CDentry *dn = p->second;
2469       ++p;
2470       if (dn->get_linkage()->is_null())
2471         continue;
2472       write_one(dn);
2473
2474       if (!(++count % mdcache->mds->heartbeat_reset_grace()))
2475         mdcache->mds->heartbeat_reset();
2476     }
2477   } else {
2478     for (auto p = dirty_dentries.begin(); !p.end(); ) {
2479       CDentry *dn = *p;
2480       ++p;
2481       write_one(dn);
2482
2483       if (!(++count % mdcache->mds->heartbeat_reset_grace()))
2484         mdcache->mds->heartbeat_reset();
2485     }
2486   }
2487
2488   auto c = new C_IO_Dir_Commit_Ops(this, op_prio, std::move(to_set), std::move(dfts),
2489                                    std::move(to_remove), std::move(stale_items));
2490   stale_items.clear();
2491   mdcache->mds->finisher->queue(c);
2492 }
2493
2494 void CDir::_parse_dentry(CDentry *dn, dentry_commit_item &item,
2495                          const set<snapid_t> *snaps, bufferlist &bl)
2496 {
2497   // clear dentry NEW flag, if any.  we can no longer silently drop it.
2498   dn->clear_new();
2499
2500   item.first = dn->first;
2501
2502   // primary or remote?
2503   auto& linkage = dn->linkage;
2504   item.alternate_name = dn->get_alternate_name();
2505   if (linkage.is_remote()) {
2506     item.is_remote = true;
2507     item.ino = linkage.get_remote_ino();
2508     item.d_type = linkage.get_remote_d_type();
2509     dout(14) << " dn '" << dn->get_name() << "' remote ino " << item.ino << dendl;
2510   } else if (linkage.is_primary()) {
2511     // primary link
2512     CInode *in = linkage.get_inode();
2513     ceph_assert(in);
2514
2515     dout(14) << " dn '" << dn->get_name() << "' inode " << *in << dendl;
2516
2517     if (in->is_multiversion()) {
2518       if (!in->snaprealm) {
2519         if (snaps)
2520           in->purge_stale_snap_data(*snaps);
2521       } else {
2522         in->purge_stale_snap_data(in->snaprealm->get_snaps());
2523       }
2524     }
2525
2526     if (in->snaprealm) {
2527       item.snaprealm = true;
2528       item.srnode = in->snaprealm->srnode;
2529     }
2530     item.features = mdcache->mds->mdsmap->get_up_features();
2531     item.inode = in->inode;
2532     if (in->inode->is_symlink())
2533       item.symlink = in->symlink;
2534     using ceph::encode;
2535     encode(in->dirfragtree, bl);
2536     item.xattrs = in->xattrs;
2537     item.old_inodes = in->old_inodes;
2538     item.oldest_snap = in->oldest_snap;
2539     item.damage_flags = in->damage_flags;
2540   } else {
2541     ceph_assert(!linkage.is_null());
2542   }
2543 }
2544
2545 void CDir::_commit(version_t want, int op_prio)
2546 {
2547   dout(10) << "_commit want " << want << " on " << *this << dendl;
2548
2549   // we can't commit things in the future.
2550   // (even the projected future.)
2551   ceph_assert(want <= get_version() || get_version() == 0);
2552
2553   // check pre+postconditions.
2554   ceph_assert(is_auth());
2555
2556   // already committed?
2557   if (committed_version >= want) {
2558     dout(10) << "already committed " << committed_version << " >= " << want << dendl;
2559     return;
2560   }
2561   // already committing >= want?
2562   if (committing_version >= want) {
2563     dout(10) << "already committing " << committing_version << " >= " << want << dendl;
2564     ceph_assert(state_test(STATE_COMMITTING));
2565     return;
2566   }
2567
2568   // alrady committed an older version?
2569   if (committing_version > committed_version) {
2570     dout(10) << "already committing older " << committing_version << ", waiting for that to finish" << dendl;
2571     return;
2572   }
2573
2574   // commit.
2575   committing_version = get_version();
2576
2577   // mark committing (if not already)
2578   if (!state_test(STATE_COMMITTING)) {
2579     dout(10) << "marking committing" << dendl;
2580     state_set(STATE_COMMITTING);
2581   }
2582
2583   if (mdcache->mds->logger) mdcache->mds->logger->inc(l_mds_dir_commit);
2584
2585   mdcache->mds->balancer->hit_dir(this, META_POP_STORE);
2586
2587   _omap_commit(op_prio);
2588 }
2589
2590
2591 /**
2592  * _committed
2593  *
2594  * @param v version i just committed
2595  */
2596 void CDir::_committed(int r, version_t v)
2597 {
2598   if (r < 0) {
2599     // the directory could be partly purged during MDS failover
2600     if (r == -CEPHFS_ENOENT && committed_version == 0 &&
2601         !inode->is_base() && get_parent_dir()->inode->is_stray()) {
2602       r = 0;
2603       if (inode->snaprealm)
2604         inode->state_set(CInode::STATE_MISSINGOBJS);
2605     }
2606     if (r < 0) {
2607       dout(1) << "commit error " << r << " v " << v << dendl;
2608       mdcache->mds->clog->error() << "failed to commit dir " << dirfrag() << " object,"
2609                                 << " errno " << r;
2610       mdcache->mds->handle_write_error(r);
2611       return;
2612     }
2613   }
2614
2615   dout(10) << "_committed v " << v << " on " << *this << dendl;
2616   ceph_assert(is_auth());
2617
2618   bool stray = inode->is_stray();
2619
2620   // take note.
2621   ceph_assert(v > committed_version);
2622   ceph_assert(v <= committing_version);
2623   committed_version = v;
2624
2625   // _all_ commits done?
2626   if (committing_version == committed_version)
2627     state_clear(CDir::STATE_COMMITTING);
2628
2629   // _any_ commit, even if we've been redirtied, means we're no longer new.
2630   item_new.remove_myself();
2631
2632   // dir clean?
2633   if (committed_version == get_version())
2634     mark_clean();
2635
2636   int count = 0;
2637
2638   // dentries clean?
2639   for (auto p = dirty_dentries.begin(); !p.end(); ) {
2640     CDentry *dn = *p;
2641     ++p;
2642
2643     // inode?
2644     if (dn->linkage.is_primary()) {
2645       CInode *in = dn->linkage.get_inode();
2646       ceph_assert(in);
2647       ceph_assert(in->is_auth());
2648
2649       if (committed_version >= in->get_version()) {
2650         if (in->is_dirty()) {
2651           dout(15) << " dir " << committed_version << " >= inode " << in->get_version() << " now clean " << *in << dendl;
2652           in->mark_clean();
2653         }
2654       } else {
2655         dout(15) << " dir " << committed_version << " < inode " << in->get_version() << " still dirty " << *in << dendl;
2656         ceph_assert(in->is_dirty() || in->last < CEPH_NOSNAP);  // special case for cow snap items (not predirtied)
2657       }
2658     }
2659
2660     // dentry
2661     if (committed_version >= dn->get_version()) {
2662       dout(15) << " dir " << committed_version << " >= dn " << dn->get_version() << " now clean " << *dn << dendl;
2663       dn->mark_clean();
2664
2665       // drop clean null stray dentries immediately
2666       if (stray &&
2667           dn->get_num_ref() == 0 &&
2668           !dn->is_projected() &&
2669           dn->get_linkage()->is_null())
2670         remove_dentry(dn);
2671     } else {
2672       dout(15) << " dir " << committed_version << " < dn " << dn->get_version() << " still dirty " << *dn << dendl;
2673       ceph_assert(dn->is_dirty());
2674     }
2675
2676     if (!(++count % mdcache->mds->heartbeat_reset_grace()))
2677       mdcache->mds->heartbeat_reset();
2678   }
2679
2680   // finishers?
2681   bool were_waiters = !waiting_for_commit.empty();
2682
2683   auto it = waiting_for_commit.begin();
2684   while (it != waiting_for_commit.end()) {
2685     auto _it = it;
2686     ++_it;
2687     if (it->first > committed_version) {
2688       dout(10) << " there are waiters for " << it->first << ", committing again" << dendl;
2689       _commit(it->first, -1);
2690       break;
2691     }
2692     MDSContext::vec t;
2693     for (const auto &waiter : it->second)
2694       t.push_back(waiter);
2695     mdcache->mds->queue_waiters(t);
2696     waiting_for_commit.erase(it);
2697     it = _it;
2698
2699     if (!(++count % mdcache->mds->heartbeat_reset_grace()))
2700       mdcache->mds->heartbeat_reset();
2701   }
2702
2703   // try drop dentries in this dirfrag if it's about to be purged
2704   if (!inode->is_base() && get_parent_dir()->inode->is_stray() &&
2705       inode->snaprealm)
2706     mdcache->maybe_eval_stray(inode, true);
2707
2708   // unpin if we kicked the last waiter.
2709   if (were_waiters &&
2710       waiting_for_commit.empty())
2711     auth_unpin(this);
2712 }
2713
2714
2715
2716
2717 // IMPORT/EXPORT
2718
2719 mds_rank_t CDir::get_export_pin(bool inherit) const
2720 {
2721   mds_rank_t export_pin = inode->get_export_pin(inherit);
2722   if (export_pin == MDS_RANK_EPHEMERAL_DIST)
2723     export_pin = mdcache->hash_into_rank_bucket(ino(), get_frag());
2724   else if (export_pin == MDS_RANK_EPHEMERAL_RAND)
2725     export_pin = mdcache->hash_into_rank_bucket(ino());
2726   return export_pin;
2727 }
2728
2729 bool CDir::is_exportable(mds_rank_t dest) const
2730 {
2731   mds_rank_t export_pin = get_export_pin();
2732   if (export_pin == dest)
2733     return true;
2734   if (export_pin >= 0)
2735     return false;
2736   return true;
2737 }
2738
2739 void CDir::encode_export(bufferlist& bl)
2740 {
2741   ENCODE_START(1, 1, bl);
2742   ceph_assert(!is_projected());
2743   encode(first, bl);
2744   encode(*fnode, bl);
2745   encode(dirty_old_rstat, bl);
2746   encode(committed_version, bl);
2747
2748   encode(state, bl);
2749   encode(dir_rep, bl);
2750
2751   encode(pop_me, bl);
2752   encode(pop_auth_subtree, bl);
2753
2754   encode(dir_rep_by, bl);
2755   encode(get_replicas(), bl);
2756
2757   get(PIN_TEMPEXPORTING);
2758   ENCODE_FINISH(bl);
2759 }
2760
2761 void CDir::finish_export()
2762 {
2763   state &= MASK_STATE_EXPORT_KEPT;
2764   pop_nested.sub(pop_auth_subtree);
2765   pop_auth_subtree_nested.sub(pop_auth_subtree);
2766   pop_me.zero();
2767   pop_auth_subtree.zero();
2768   put(PIN_TEMPEXPORTING);
2769   dirty_old_rstat.clear();
2770 }
2771
2772 void CDir::decode_import(bufferlist::const_iterator& blp, LogSegment *ls)
2773 {
2774   DECODE_START(1, blp);
2775   decode(first, blp);
2776   {
2777     auto _fnode = allocate_fnode();
2778     decode(*_fnode, blp);
2779     reset_fnode(std::move(_fnode));
2780   }
2781   update_projected_version();
2782
2783   decode(dirty_old_rstat, blp);
2784   decode(committed_version, blp);
2785   committing_version = committed_version;
2786
2787   unsigned s;
2788   decode(s, blp);
2789   state &= MASK_STATE_IMPORT_KEPT;
2790   state_set(STATE_AUTH | (s & MASK_STATE_EXPORTED));
2791
2792   if (is_dirty()) {
2793     get(PIN_DIRTY);
2794     _mark_dirty(ls);
2795   }
2796
2797   decode(dir_rep, blp);
2798
2799   decode(pop_me, blp);
2800   decode(pop_auth_subtree, blp);
2801   pop_nested.add(pop_auth_subtree);
2802   pop_auth_subtree_nested.add(pop_auth_subtree);
2803
2804   decode(dir_rep_by, blp);
2805   decode(get_replicas(), blp);
2806   if (is_replicated()) get(PIN_REPLICATED);
2807
2808   replica_nonce = 0;  // no longer defined
2809
2810   // did we import some dirty scatterlock data?
2811   if (dirty_old_rstat.size() ||
2812       !(fnode->rstat == fnode->accounted_rstat)) {
2813     mdcache->mds->locker->mark_updated_scatterlock(&inode->nestlock);
2814     ls->dirty_dirfrag_nest.push_back(&inode->item_dirty_dirfrag_nest);
2815   }
2816   if (!(fnode->fragstat == fnode->accounted_fragstat)) {
2817     mdcache->mds->locker->mark_updated_scatterlock(&inode->filelock);
2818     ls->dirty_dirfrag_dir.push_back(&inode->item_dirty_dirfrag_dir);
2819   }
2820   if (is_dirty_dft()) {
2821     if (inode->dirfragtreelock.get_state() != LOCK_MIX &&
2822         inode->dirfragtreelock.is_stable()) {
2823       // clear stale dirtydft
2824       state_clear(STATE_DIRTYDFT);
2825     } else {
2826       mdcache->mds->locker->mark_updated_scatterlock(&inode->dirfragtreelock);
2827       ls->dirty_dirfrag_dirfragtree.push_back(&inode->item_dirty_dirfrag_dirfragtree);
2828     }
2829   }
2830   DECODE_FINISH(blp);
2831 }
2832
2833 void CDir::abort_import()
2834 {
2835   ceph_assert(is_auth());
2836   state_clear(CDir::STATE_AUTH);
2837   remove_bloom();
2838   clear_replica_map();
2839   set_replica_nonce(CDir::EXPORT_NONCE);
2840   if (is_dirty())
2841     mark_clean();
2842
2843   pop_nested.sub(pop_auth_subtree);
2844   pop_auth_subtree_nested.sub(pop_auth_subtree);
2845   pop_me.zero();
2846   pop_auth_subtree.zero();
2847 }
2848
2849 void CDir::encode_dirstat(bufferlist& bl, const session_info_t& info, const DirStat& ds) {
2850   if (info.has_feature(CEPHFS_FEATURE_REPLY_ENCODING)) {
2851     ENCODE_START(1, 1, bl);
2852     encode(ds.frag, bl);
2853     encode(ds.auth, bl);
2854     encode(ds.dist, bl);
2855     ENCODE_FINISH(bl);
2856   }
2857   else {
2858     encode(ds.frag, bl);
2859     encode(ds.auth, bl);
2860     encode(ds.dist, bl);
2861   }
2862 }
2863
2864 /********************************
2865  * AUTHORITY
2866  */
2867
2868 /*
2869  * if dir_auth.first == parent, auth is same as inode.
2870  * unless .second != unknown, in which case that sticks.
2871  */
2872 mds_authority_t CDir::authority() const
2873 {
2874   if (is_subtree_root())
2875     return dir_auth;
2876   else
2877     return inode->authority();
2878 }
2879
2880 /** is_subtree_root()
2881  * true if this is an auth delegation point.
2882  * that is, dir_auth != default (parent,unknown)
2883  *
2884  * some key observations:
2885  *  if i am auth:
2886  *    - any region bound will be an export, or frozen.
2887  *
2888  * note that this DOES heed dir_auth.pending
2889  */
2890 /*
2891 bool CDir::is_subtree_root()
2892 {
2893   if (dir_auth == CDIR_AUTH_DEFAULT) {
2894     //dout(10) << "is_subtree_root false " << dir_auth << " != " << CDIR_AUTH_DEFAULT
2895     //<< " on " << ino() << dendl;
2896     return false;
2897   } else {
2898     //dout(10) << "is_subtree_root true " << dir_auth << " != " << CDIR_AUTH_DEFAULT
2899     //<< " on " << ino() << dendl;
2900     return true;
2901   }
2902 }
2903 */
2904
2905 /** contains(x)
2906  * true if we are x, or an ancestor of x
2907  */
2908 bool CDir::contains(CDir *x)
2909 {
2910   while (1) {
2911     if (x == this)
2912       return true;
2913     x = x->get_inode()->get_projected_parent_dir();
2914     if (x == 0)
2915       return false;
2916   }
2917 }
2918
2919 bool CDir::can_rep() const
2920 {
2921   if (!is_rep())
2922     return true;
2923
2924   unsigned mds_num = mdcache->mds->get_mds_map()->get_num_mds(MDSMap::STATE_ACTIVE);
2925   if ((mds_num - 1) > get_replicas().size())
2926     return true;
2927
2928   return false;
2929 }
2930
2931
2932 /** set_dir_auth
2933  */
2934 void CDir::set_dir_auth(const mds_authority_t &a)
2935 {
2936   dout(10) << "setting dir_auth=" << a
2937            << " from " << dir_auth
2938            << " on " << *this << dendl;
2939
2940   bool was_subtree = is_subtree_root();
2941   bool was_ambiguous = dir_auth.second >= 0;
2942
2943   // set it.
2944   dir_auth = a;
2945
2946   // new subtree root?
2947   if (!was_subtree && is_subtree_root()) {
2948     dout(10) << " new subtree root, adjusting auth_pins" << dendl;
2949
2950     if (freeze_tree_state) {
2951       // only by CDir::_freeze_tree()
2952       ceph_assert(is_freezing_tree_root());
2953     }
2954
2955     inode->num_subtree_roots++;
2956
2957     // unpin parent of frozen dir/tree?
2958     if (inode->is_auth()) {
2959       ceph_assert(!is_frozen_tree_root());
2960       if (is_frozen_dir())
2961         inode->auth_unpin(this);
2962     }
2963   }
2964   if (was_subtree && !is_subtree_root()) {
2965     dout(10) << " old subtree root, adjusting auth_pins" << dendl;
2966
2967     inode->num_subtree_roots--;
2968
2969     // pin parent of frozen dir/tree?
2970     if (inode->is_auth()) {
2971       ceph_assert(!is_frozen_tree_root());
2972       if (is_frozen_dir())
2973         inode->auth_pin(this);
2974     }
2975   }
2976
2977   // newly single auth?
2978   if (was_ambiguous && dir_auth.second == CDIR_AUTH_UNKNOWN) {
2979     MDSContext::vec ls;
2980     take_waiting(WAIT_SINGLEAUTH, ls);
2981     mdcache->mds->queue_waiters(ls);
2982   }
2983 }
2984
2985 /*****************************************
2986  * AUTH PINS and FREEZING
2987  *
2988  * the basic plan is that auth_pins only exist in auth regions, and they
2989  * prevent a freeze (and subsequent auth change).
2990  *
2991  * however, we also need to prevent a parent from freezing if a child is frozen.
2992  * for that reason, the parent inode of a frozen directory is auth_pinned.
2993  *
2994  * the oddity is when the frozen directory is a subtree root.  if that's the case,
2995  * the parent inode isn't frozen.  which means that when subtree authority is adjusted
2996  * at the bounds, inodes for any frozen bound directories need to get auth_pins at that
2997  * time.
2998  *
2999  */
3000
3001 void CDir::auth_pin(void *by)
3002 {
3003   if (auth_pins == 0)
3004     get(PIN_AUTHPIN);
3005   auth_pins++;
3006
3007 #ifdef MDS_AUTHPIN_SET
3008   auth_pin_set.insert(by);
3009 #endif
3010
3011   dout(10) << "auth_pin by " << by << " on " << *this << " count now " << auth_pins << dendl;
3012
3013   if (freeze_tree_state)
3014     freeze_tree_state->auth_pins += 1;
3015 }
3016
3017 void CDir::auth_unpin(void *by)
3018 {
3019   auth_pins--;
3020
3021 #ifdef MDS_AUTHPIN_SET
3022   {
3023     auto it = auth_pin_set.find(by);
3024     ceph_assert(it != auth_pin_set.end());
3025     auth_pin_set.erase(it);
3026   }
3027 #endif
3028   if (auth_pins == 0)
3029     put(PIN_AUTHPIN);
3030
3031   dout(10) << "auth_unpin by " << by << " on " << *this << " count now " << auth_pins << dendl;
3032   ceph_assert(auth_pins >= 0);
3033
3034   if (freeze_tree_state)
3035     freeze_tree_state->auth_pins -= 1;
3036
3037   maybe_finish_freeze();  // pending freeze?
3038 }
3039
3040 void CDir::adjust_nested_auth_pins(int dirinc, void *by)
3041 {
3042   ceph_assert(dirinc);
3043   dir_auth_pins += dirinc;
3044
3045   dout(15) << __func__ << " " << dirinc << " on " << *this
3046            << " by " << by << " count now "
3047            << auth_pins << "/" << dir_auth_pins << dendl;
3048   ceph_assert(dir_auth_pins >= 0);
3049
3050   if (freeze_tree_state)
3051     freeze_tree_state->auth_pins += dirinc;
3052
3053   if (dirinc < 0)
3054     maybe_finish_freeze();  // pending freeze?
3055 }
3056
3057 #ifdef MDS_VERIFY_FRAGSTAT
3058 void CDir::verify_fragstat()
3059 {
3060   ceph_assert(is_complete());
3061   if (inode->is_stray())
3062     return;
3063
3064   frag_info_t c;
3065   memset(&c, 0, sizeof(c));
3066
3067   for (auto it = items.begin();
3068        it != items.end();
3069        ++it) {
3070     CDentry *dn = it->second;
3071     if (dn->is_null())
3072       continue;
3073
3074     dout(10) << " " << *dn << dendl;
3075     if (dn->is_primary())
3076       dout(10) << "     " << *dn->inode << dendl;
3077
3078     if (dn->is_primary()) {
3079       if (dn->inode->is_dir())
3080         c.nsubdirs++;
3081       else
3082         c.nfiles++;
3083     }
3084     if (dn->is_remote()) {
3085       if (dn->get_remote_d_type() == DT_DIR)
3086         c.nsubdirs++;
3087       else
3088         c.nfiles++;
3089     }
3090   }
3091
3092   if (c.nsubdirs != fnode->fragstat.nsubdirs ||
3093       c.nfiles != fnode->fragstat.nfiles) {
3094     dout(0) << "verify_fragstat failed " << fnode->fragstat << " on " << *this << dendl;
3095     dout(0) << "               i count " << c << dendl;
3096     ceph_abort();
3097   } else {
3098     dout(0) << "verify_fragstat ok " << fnode->fragstat << " on " << *this << dendl;
3099   }
3100 }
3101 #endif
3102
3103 /*****************************************************************************
3104  * FREEZING
3105  */
3106
3107 // FREEZE TREE
3108
3109 void CDir::_walk_tree(std::function<bool(CDir*)> callback)
3110 {
3111   deque<CDir*> dfq;
3112   dfq.push_back(this);
3113
3114   while (!dfq.empty()) {
3115     CDir *dir = dfq.front();
3116     dfq.pop_front();
3117
3118     for (auto& p : *dir) {
3119       CDentry *dn = p.second;
3120       if (!dn->get_linkage()->is_primary())
3121         continue;
3122       CInode *in = dn->get_linkage()->get_inode();
3123       if (!in->is_dir())
3124         continue;
3125
3126       auto&& dfv = in->get_nested_dirfrags();
3127       for (auto& dir : dfv) {
3128         auto ret = callback(dir);
3129         if (ret)
3130           dfq.push_back(dir);
3131       }
3132     }
3133   }
3134 }
3135
3136 bool CDir::freeze_tree()
3137 {
3138   ceph_assert(!is_frozen());
3139   ceph_assert(!is_freezing());
3140   ceph_assert(!freeze_tree_state);
3141
3142   auth_pin(this);
3143
3144   // Travese the subtree to mark dirfrags as 'freezing' (set freeze_tree_state)
3145   // and to accumulate auth pins and record total count in freeze_tree_state.
3146   // when auth unpin an 'freezing' object, the counter in freeze_tree_state also
3147   // gets decreased. Subtree become 'frozen' when the counter reaches zero.
3148   freeze_tree_state = std::make_shared<freeze_tree_state_t>(this);
3149   freeze_tree_state->auth_pins += get_auth_pins() + get_dir_auth_pins();
3150   if (!lock_caches_with_auth_pins.empty())
3151     mdcache->mds->locker->invalidate_lock_caches(this);
3152
3153   _walk_tree([this](CDir *dir) {
3154       if (dir->freeze_tree_state)
3155         return false;
3156       dir->freeze_tree_state = freeze_tree_state;
3157       freeze_tree_state->auth_pins += dir->get_auth_pins() + dir->get_dir_auth_pins();
3158       if (!dir->lock_caches_with_auth_pins.empty())
3159         mdcache->mds->locker->invalidate_lock_caches(dir);
3160       return true;
3161     }
3162   );
3163
3164   if (is_freezeable(true)) {
3165     _freeze_tree();
3166     auth_unpin(this);
3167     return true;
3168   } else {
3169     state_set(STATE_FREEZINGTREE);
3170     ++num_freezing_trees;
3171     dout(10) << "freeze_tree waiting " << *this << dendl;
3172     return false;
3173   }
3174 }
3175
3176 void CDir::_freeze_tree()
3177 {
3178   dout(10) << __func__ << " " << *this << dendl;
3179   ceph_assert(is_freezeable(true));
3180
3181   if (freeze_tree_state) {
3182     ceph_assert(is_auth());
3183   } else {
3184     ceph_assert(!is_auth());
3185     freeze_tree_state = std::make_shared<freeze_tree_state_t>(this);
3186   }
3187   freeze_tree_state->frozen = true;
3188
3189   if (is_auth()) {
3190     mds_authority_t auth;
3191     bool was_subtree = is_subtree_root();
3192     if (was_subtree) {
3193       auth = get_dir_auth();
3194     } else {
3195       // temporarily prevent parent subtree from becoming frozen.
3196       inode->auth_pin(this);
3197       // create new subtree
3198       auth = authority();
3199     }
3200
3201     _walk_tree([this, &auth] (CDir *dir) {
3202         if (dir->freeze_tree_state != freeze_tree_state) {
3203           mdcache->adjust_subtree_auth(dir, auth);
3204           return false;
3205         }
3206         return true;
3207       }
3208     );
3209
3210     ceph_assert(auth.first >= 0);
3211     ceph_assert(auth.second == CDIR_AUTH_UNKNOWN);
3212     auth.second = auth.first;
3213     mdcache->adjust_subtree_auth(this, auth);
3214     if (!was_subtree)
3215       inode->auth_unpin(this);
3216   } else {
3217     // importing subtree ?
3218     _walk_tree([this] (CDir *dir) {
3219         ceph_assert(!dir->freeze_tree_state);
3220         dir->freeze_tree_state = freeze_tree_state;
3221         return true;
3222       }
3223     );
3224   }
3225
3226   // twiddle state
3227   if (state_test(STATE_FREEZINGTREE)) {
3228     state_clear(STATE_FREEZINGTREE);
3229     --num_freezing_trees;
3230   }
3231
3232   state_set(STATE_FROZENTREE);
3233   ++num_frozen_trees;
3234   get(PIN_FROZEN);
3235 }
3236
3237 void CDir::unfreeze_tree()
3238 {
3239   dout(10) << __func__ << " " << *this << dendl;
3240
3241   MDSContext::vec unfreeze_waiters;
3242   take_waiting(WAIT_UNFREEZE, unfreeze_waiters);
3243
3244   if (freeze_tree_state) {
3245     _walk_tree([this, &unfreeze_waiters](CDir *dir) {
3246         if (dir->freeze_tree_state != freeze_tree_state)
3247           return false;
3248         dir->freeze_tree_state.reset();
3249         dir->take_waiting(WAIT_UNFREEZE, unfreeze_waiters);
3250         return true;
3251       }
3252     );
3253   }
3254
3255   if (state_test(STATE_FROZENTREE)) {
3256     // frozen.  unfreeze.
3257     state_clear(STATE_FROZENTREE);
3258     --num_frozen_trees;
3259
3260     put(PIN_FROZEN);
3261
3262     if (is_auth()) {
3263       // must be subtree
3264       ceph_assert(is_subtree_root());
3265       // for debug purpose, caller should ensure 'dir_auth.second == dir_auth.first'
3266       mds_authority_t auth = get_dir_auth();
3267       ceph_assert(auth.first >= 0);
3268       ceph_assert(auth.second == auth.first);
3269       auth.second = CDIR_AUTH_UNKNOWN;
3270       mdcache->adjust_subtree_auth(this, auth);
3271     }
3272     freeze_tree_state.reset();
3273   } else {
3274     ceph_assert(state_test(STATE_FREEZINGTREE));
3275
3276     // freezing.  stop it.
3277     state_clear(STATE_FREEZINGTREE);
3278     --num_freezing_trees;
3279     freeze_tree_state.reset();
3280
3281     finish_waiting(WAIT_FROZEN, -1);
3282     auth_unpin(this);
3283   }
3284
3285   mdcache->mds->queue_waiters(unfreeze_waiters);
3286 }
3287
3288 void CDir::adjust_freeze_after_rename(CDir *dir)
3289 {
3290   if (!freeze_tree_state || dir->freeze_tree_state != freeze_tree_state)
3291     return;
3292   CDir *newdir = dir->get_inode()->get_parent_dir();
3293   if (newdir == this || newdir->freeze_tree_state == freeze_tree_state)
3294     return;
3295
3296   ceph_assert(!freeze_tree_state->frozen);
3297   ceph_assert(get_dir_auth_pins() > 0);
3298
3299   MDSContext::vec unfreeze_waiters;
3300
3301   auto unfreeze = [this, &unfreeze_waiters](CDir *dir) {
3302     if (dir->freeze_tree_state != freeze_tree_state)
3303       return false;
3304     int dec = dir->get_auth_pins() + dir->get_dir_auth_pins();
3305     // shouldn't become zero because srcdn of rename was auth pinned
3306     ceph_assert(freeze_tree_state->auth_pins > dec);
3307     freeze_tree_state->auth_pins -= dec;
3308     dir->freeze_tree_state.reset();
3309     dir->take_waiting(WAIT_UNFREEZE, unfreeze_waiters);
3310     return true;
3311   };
3312
3313   unfreeze(dir);
3314   dir->_walk_tree(unfreeze);
3315
3316   mdcache->mds->queue_waiters(unfreeze_waiters);
3317 }
3318
3319 bool CDir::can_auth_pin(int *err_ret) const
3320 {
3321   int err;
3322   if (!is_auth()) {
3323     err = ERR_NOT_AUTH;
3324   } else if (is_freezing_dir() || is_frozen_dir()) {
3325     err = ERR_FRAGMENTING_DIR;
3326   } else {
3327     auto p = is_freezing_or_frozen_tree();
3328     if (p.first || p.second) {
3329       err = ERR_EXPORTING_TREE;
3330     } else {
3331       err = 0;
3332     }
3333   }
3334   if (err && err_ret)
3335     *err_ret = err;
3336   return !err;
3337 }
3338
3339 class C_Dir_AuthUnpin : public CDirContext {
3340   public:
3341   explicit C_Dir_AuthUnpin(CDir *d) : CDirContext(d) {}
3342   void finish(int r) override {
3343     dir->auth_unpin(dir->get_inode());
3344   }
3345 };
3346
3347 void CDir::maybe_finish_freeze()
3348 {
3349   if (dir_auth_pins != 0)
3350     return;
3351
3352   // we can freeze the _dir_ even with nested pins...
3353   if (state_test(STATE_FREEZINGDIR)) {
3354     if (auth_pins == 1) {
3355       _freeze_dir();
3356       auth_unpin(this);
3357       finish_waiting(WAIT_FROZEN);
3358     }
3359   }
3360
3361   if (freeze_tree_state) {
3362     if (freeze_tree_state->frozen ||
3363         freeze_tree_state->auth_pins != 1)
3364       return;
3365
3366     if (freeze_tree_state->dir != this) {
3367       freeze_tree_state->dir->maybe_finish_freeze();
3368       return;
3369     }
3370
3371     ceph_assert(state_test(STATE_FREEZINGTREE));
3372
3373     if (!is_subtree_root() && inode->is_frozen()) {
3374       dout(10) << __func__ << " !subtree root and frozen inode, waiting for unfreeze on " << inode << dendl;
3375       // retake an auth_pin...
3376       auth_pin(inode);
3377       // and release it when the parent inode unfreezes
3378       inode->add_waiter(WAIT_UNFREEZE, new C_Dir_AuthUnpin(this));
3379       return;
3380     }
3381
3382     _freeze_tree();
3383     auth_unpin(this);
3384     finish_waiting(WAIT_FROZEN);
3385   }
3386 }
3387
3388
3389
3390 // FREEZE DIR
3391
3392 bool CDir::freeze_dir()
3393 {
3394   ceph_assert(!is_frozen());
3395   ceph_assert(!is_freezing());
3396
3397   auth_pin(this);
3398   if (is_freezeable_dir(true)) {
3399     _freeze_dir();
3400     auth_unpin(this);
3401     return true;
3402   } else {
3403     state_set(STATE_FREEZINGDIR);
3404     if (!lock_caches_with_auth_pins.empty())
3405       mdcache->mds->locker->invalidate_lock_caches(this);
3406     dout(10) << "freeze_dir + wait " << *this << dendl;
3407     return false;
3408   }
3409 }
3410
3411 void CDir::_freeze_dir()
3412 {
3413   dout(10) << __func__ << " " << *this << dendl;
3414   //assert(is_freezeable_dir(true));
3415   // not always true during split because the original fragment may have frozen a while
3416   // ago and we're just now getting around to breaking it up.
3417
3418   state_clear(STATE_FREEZINGDIR);
3419   state_set(STATE_FROZENDIR);
3420   get(PIN_FROZEN);
3421
3422   if (is_auth() && !is_subtree_root())
3423     inode->auth_pin(this);  // auth_pin for duration of freeze
3424 }
3425
3426
3427 void CDir::unfreeze_dir()
3428 {
3429   dout(10) << __func__ << " " << *this << dendl;
3430
3431   if (state_test(STATE_FROZENDIR)) {
3432     state_clear(STATE_FROZENDIR);
3433     put(PIN_FROZEN);
3434
3435     // unpin  (may => FREEZEABLE)   FIXME: is this order good?
3436     if (is_auth() && !is_subtree_root())
3437       inode->auth_unpin(this);
3438
3439     finish_waiting(WAIT_UNFREEZE);
3440   } else {
3441     finish_waiting(WAIT_FROZEN, -1);
3442
3443     // still freezing. stop.
3444     ceph_assert(state_test(STATE_FREEZINGDIR));
3445     state_clear(STATE_FREEZINGDIR);
3446     auth_unpin(this);
3447
3448     finish_waiting(WAIT_UNFREEZE);
3449   }
3450 }
3451
3452 void CDir::enable_frozen_inode()
3453 {
3454   ceph_assert(frozen_inode_suppressed > 0);
3455   if (--frozen_inode_suppressed == 0) {
3456     for (auto p = freezing_inodes.begin(); !p.end(); ) {
3457       CInode *in = *p;
3458       ++p;
3459       ceph_assert(in->is_freezing_inode());
3460       in->maybe_finish_freeze_inode();
3461     }
3462   }
3463 }
3464
3465 /**
3466  * Slightly less complete than operator<<, because this is intended
3467  * for identifying a directory and its state rather than for dumping
3468  * debug output.
3469  */
3470 void CDir::dump(Formatter *f, int flags) const
3471 {
3472   ceph_assert(f != NULL);
3473   if (flags & DUMP_PATH) {
3474     f->dump_stream("path") << get_path();
3475   }
3476   if (flags & DUMP_DIRFRAG) {
3477     f->dump_stream("dirfrag") << dirfrag();
3478   }
3479   if (flags & DUMP_SNAPID_FIRST) {
3480     f->dump_int("snapid_first", first);
3481   }
3482   if (flags & DUMP_VERSIONS) {
3483     f->dump_stream("projected_version") << get_projected_version();
3484     f->dump_stream("version") << get_version();
3485     f->dump_stream("committing_version") << get_committing_version();
3486     f->dump_stream("committed_version") << get_committed_version();
3487   }
3488   if (flags & DUMP_REP) {
3489     f->dump_bool("is_rep", is_rep());
3490   }
3491   if (flags & DUMP_DIR_AUTH) {
3492     if (get_dir_auth() != CDIR_AUTH_DEFAULT) {
3493       if (get_dir_auth().second == CDIR_AUTH_UNKNOWN) {
3494         f->dump_stream("dir_auth") << get_dir_auth().first;
3495       } else {
3496         f->dump_stream("dir_auth") << get_dir_auth();
3497       }
3498     } else {
3499       f->dump_string("dir_auth", "");
3500     }
3501   }
3502   if (flags & DUMP_STATES) {
3503     f->open_array_section("states");
3504     MDSCacheObject::dump_states(f);
3505     if (state_test(CDir::STATE_COMPLETE)) f->dump_string("state", "complete");
3506     if (state_test(CDir::STATE_FREEZINGTREE)) f->dump_string("state", "freezingtree");
3507     if (state_test(CDir::STATE_FROZENTREE)) f->dump_string("state", "frozentree");
3508     if (state_test(CDir::STATE_FROZENDIR)) f->dump_string("state", "frozendir");
3509     if (state_test(CDir::STATE_FREEZINGDIR)) f->dump_string("state", "freezingdir");
3510     if (state_test(CDir::STATE_EXPORTBOUND)) f->dump_string("state", "exportbound");
3511     if (state_test(CDir::STATE_IMPORTBOUND)) f->dump_string("state", "importbound");
3512     if (state_test(CDir::STATE_BADFRAG)) f->dump_string("state", "badfrag");
3513     f->close_section();
3514   }
3515   if (flags & DUMP_MDS_CACHE_OBJECT) {
3516     MDSCacheObject::dump(f);
3517   }
3518   if (flags & DUMP_ITEMS) {
3519     f->open_array_section("dentries");
3520     for (auto &p : items) {
3521       CDentry *dn = p.second;
3522       f->open_object_section("dentry");
3523       dn->dump(f);
3524       f->close_section();
3525     }
3526     f->close_section();
3527   }
3528 }
3529
3530 void CDir::dump_load(Formatter *f)
3531 {
3532   f->dump_stream("path") << get_path();
3533   f->dump_stream("dirfrag") << dirfrag();
3534
3535   f->open_object_section("pop_me");
3536   pop_me.dump(f);
3537   f->close_section();
3538
3539   f->open_object_section("pop_nested");
3540   pop_nested.dump(f);
3541   f->close_section();
3542
3543   f->open_object_section("pop_auth_subtree");
3544   pop_auth_subtree.dump(f);
3545   f->close_section();
3546
3547   f->open_object_section("pop_auth_subtree_nested");
3548   pop_auth_subtree_nested.dump(f);
3549   f->close_section();
3550 }
3551
3552 /****** Scrub Stuff *******/
3553
3554 void CDir::scrub_info_create() const
3555 {
3556   ceph_assert(!scrub_infop);
3557
3558   // break out of const-land to set up implicit initial state
3559   CDir *me = const_cast<CDir*>(this);
3560   const auto& pf = me->get_projected_fnode();
3561
3562   std::unique_ptr<scrub_info_t> si(new scrub_info_t());
3563
3564   si->last_recursive.version = pf->recursive_scrub_version;
3565   si->last_recursive.time = pf->recursive_scrub_stamp;
3566
3567   si->last_local.version = pf->localized_scrub_version;
3568   si->last_local.time = pf->localized_scrub_stamp;
3569
3570   me->scrub_infop.swap(si);
3571 }
3572
3573 void CDir::scrub_initialize(const ScrubHeaderRef& header)
3574 {
3575   ceph_assert(header);
3576   // FIXME: weird implicit construction, is someone else meant
3577   // to be calling scrub_info_create first?
3578   scrub_info();
3579   scrub_infop->directory_scrubbing = true;
3580   scrub_infop->header = header;
3581   header->inc_num_pending();
3582 }
3583
3584 void CDir::scrub_aborted() {
3585   dout(20) << __func__ << dendl;
3586   ceph_assert(scrub_is_in_progress());
3587
3588   scrub_infop->last_scrub_dirty = false;
3589   scrub_infop->directory_scrubbing = false;
3590   scrub_infop->header->dec_num_pending();
3591   scrub_infop.reset();
3592 }
3593
3594 void CDir::scrub_finished()
3595 {
3596   dout(20) << __func__ << dendl;
3597   ceph_assert(scrub_is_in_progress());
3598
3599   scrub_infop->last_local.time = ceph_clock_now();
3600   scrub_infop->last_local.version = get_version();
3601   if (scrub_infop->header->get_recursive())
3602     scrub_infop->last_recursive = scrub_infop->last_local;
3603
3604   scrub_infop->last_scrub_dirty = true;
3605
3606   scrub_infop->directory_scrubbing = false;
3607   scrub_infop->header->dec_num_pending();
3608 }
3609
3610 void CDir::scrub_maybe_delete_info()
3611 {
3612   if (scrub_infop &&
3613       !scrub_infop->directory_scrubbing &&
3614       !scrub_infop->last_scrub_dirty)
3615     scrub_infop.reset();
3616 }
3617
3618 bool CDir::scrub_local()
3619 {
3620   ceph_assert(is_complete());
3621   bool good = check_rstats(true);
3622   if (!good && scrub_infop->header->get_repair()) {
3623     mdcache->repair_dirfrag_stats(this);
3624     scrub_infop->header->set_repaired();
3625   }
3626   return good;
3627 }
3628
3629 std::string CDir::get_path() const
3630 {
3631   std::string path;
3632   get_inode()->make_path_string(path, true);
3633   return path;
3634 }
3635
3636 bool CDir::should_split_fast() const
3637 {
3638   // Max size a fragment can be before trigger fast splitting
3639   int fast_limit = g_conf()->mds_bal_split_size * g_conf()->mds_bal_fragment_fast_factor;
3640
3641   // Fast path: the sum of accounted size and null dentries does not
3642   // exceed threshold: we definitely are not over it.
3643   if (get_frag_size() + get_num_head_null() <= fast_limit) {
3644     return false;
3645   }
3646
3647   // Fast path: the accounted size of the frag exceeds threshold: we
3648   // definitely are over it
3649   if (get_frag_size() > fast_limit) {
3650     return true;
3651   }
3652
3653   int64_t effective_size = 0;
3654
3655   for (const auto &p : items) {
3656     const CDentry *dn = p.second;
3657     if (!dn->get_projected_linkage()->is_null()) {
3658       effective_size++;
3659     }
3660   }
3661
3662   return effective_size > fast_limit;
3663 }
3664
3665 bool CDir::should_merge() const
3666 {
3667   if (get_frag() == frag_t())
3668     return false;
3669
3670   if (inode->is_ephemeral_dist()) {
3671     unsigned min_frag_bits = mdcache->get_ephemeral_dist_frag_bits();
3672     if (min_frag_bits > 0 && get_frag().bits() < min_frag_bits + 1)
3673       return false;
3674   }
3675
3676   return (int)get_frag_size() < g_conf()->mds_bal_merge_size;
3677 }
3678
3679 MEMPOOL_DEFINE_OBJECT_FACTORY(CDir, co_dir, mds_co);
3680 MEMPOOL_DEFINE_OBJECT_FACTORY(CDir::scrub_info_t, scrub_info_t, mds_co)