1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
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.
15 #include <string_view>
18 #include "include/types.h"
30 #include "LogSegment.h"
31 #include "MDBalancer.h"
33 #include "common/bloom_filter.hpp"
34 #include "include/Context.h"
35 #include "common/Clock.h"
37 #include "osdc/Objecter.h"
39 #include "common/config.h"
40 #include "include/ceph_assert.h"
41 #include "include/compat.h"
43 #define dout_context g_ceph_context
44 #define dout_subsys ceph_subsys_mds
46 #define dout_prefix *_dout << "mds." << mdcache->mds->get_nodeid() << ".cache.dir(" << this->dirfrag() << ") "
50 int CDir::num_frozen_trees
= 0;
51 int CDir::num_freezing_trees
= 0;
53 CDir::fnode_const_ptr
CDir::empty_fnode
= CDir::allocate_fnode();
55 class CDirContext
: public MDSContext
59 MDSRank
* get_mds() override
{return dir
->mdcache
->mds
;}
62 explicit CDirContext(CDir
*d
) : dir(d
) {
63 ceph_assert(dir
!= NULL
);
68 class CDirIOContext
: public MDSIOContextBase
72 MDSRank
* get_mds() override
{return dir
->mdcache
->mds
;}
75 explicit CDirIOContext(CDir
*d
) : dir(d
) {
76 ceph_assert(dir
!= NULL
);
82 //int cdir_pins[CDIR_NUM_PINS] = { 0,0,0,0,0,0,0,0,0,0,0,0,0,0 };
85 ostream
& operator<<(ostream
& out
, const CDir
& dir
)
87 out
<< "[dir " << dir
.dirfrag() << " " << dir
.get_path() << "/"
88 << " [" << dir
.first
<< ",head]";
91 if (dir
.is_replicated())
92 out
<< dir
.get_replicas();
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();
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();
107 if (dir
.is_rep()) out
<< " REP";
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
;
113 out
<< " dir_auth=" << dir
.get_dir_auth();
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
);
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";
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
;
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
;
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();
171 if (dir
.get_num_ref()) {
173 dir
.print_pin_set(out
);
181 void CDir::print(ostream
& out
)
189 ostream
& CDir::print_db_line_prefix(ostream
& out
)
191 return out
<< ceph_clock_now() << " mds." << mdcache
->mds
->get_nodeid() << ".cache.dir(" << this->dirfrag() << ") ";
196 // -------------------------------------------------------------------
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
)),
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
)
216 ceph_assert(in
->is_dir());
218 state_set(STATE_AUTH
);
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.
226 bool CDir::check_rstats(bool scrub
)
228 if (!g_conf()->mds_debug_scatterstat
&& !scrub
)
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 "
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
)
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
);
249 frag_info
.nsubdirs
++;
252 } else if (dnl
->is_remote())
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
;
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
;
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
;
277 dout(20) << "total of child dentries: " << nest_info
<< dendl
;
278 dout(20) << "my rstats: " << fnode
->rstat
<< dendl
;
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
;
289 dout(1) << *dn
<< dendl
;
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
);
300 dout(10) << "check_rstats complete on " << this << dendl
;
304 void CDir::adjust_num_inodes_with_caps(int d
)
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);
312 num_inodes_with_caps
+= d
;
313 ceph_assert(num_inodes_with_caps
>= 0);
316 CDentry
*CDir::lookup(std::string_view name
, snapid_t snap
)
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())
322 if (iter
->second
->get_name() == name
&&
323 iter
->second
->first
<= snap
&&
324 iter
->second
->last
>= snap
) {
325 dout(20) << " hit -> " << iter
->first
<< dendl
;
328 dout(20) << " miss -> " << iter
->first
<< dendl
;
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())
344 CDentry
* CDir::add_null_dentry(std::string_view dname
,
345 snapid_t first
, snapid_t last
)
348 ceph_assert(lookup_exact_snap(dname
, last
) == 0);
351 CDentry
* dn
= new CDentry(dname
, inode
->hash_dentry_name(dname
), "", first
, last
);
353 dn
->state_set(CDentry::STATE_AUTH
);
355 mdcache
->bottom_lru
.lru_insert_mid(dn
);
356 dn
->state_set(CDentry::STATE_BOTTOMLRU
);
359 dn
->version
= get_projected_version();
362 ceph_assert(items
.count(dn
->key()) == 0);
363 //assert(null_items.count(dn->get_name()) == 0);
365 items
[dn
->key()] = dn
;
366 if (last
== CEPH_NOSNAP
)
371 if (state_test(CDir::STATE_DNPINNEDFRAG
)) {
372 dn
->get(CDentry::PIN_FRAGMENTING
);
373 dn
->state_set(CDentry::STATE_FRAGMENTING
);
376 dout(12) << __func__
<< " " << *dn
<< dendl
;
379 if (get_num_any() == 1)
382 ceph_assert(get_num_any() == items
.size());
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
)
392 ceph_assert(lookup_exact_snap(dname
, last
) == 0);
395 CDentry
* dn
= new CDentry(dname
, inode
->hash_dentry_name(dname
), std::move(alternate_name
), first
, last
);
397 dn
->state_set(CDentry::STATE_AUTH
);
398 if (is_auth() || !inode
->is_stray()) {
399 mdcache
->lru
.lru_insert_mid(dn
);
401 mdcache
->bottom_lru
.lru_insert_mid(dn
);
402 dn
->state_set(CDentry::STATE_BOTTOMLRU
);
406 dn
->version
= get_projected_version();
409 ceph_assert(items
.count(dn
->key()) == 0);
410 //assert(null_items.count(dn->get_name()) == 0);
412 items
[dn
->key()] = dn
;
414 dn
->get_linkage()->inode
= in
;
416 link_inode_work(dn
, in
);
418 if (dn
->last
== CEPH_NOSNAP
)
423 if (state_test(CDir::STATE_DNPINNEDFRAG
)) {
424 dn
->get(CDentry::PIN_FRAGMENTING
);
425 dn
->state_set(CDentry::STATE_FRAGMENTING
);
428 dout(12) << __func__
<< " " << *dn
<< dendl
;
431 if (get_num_any() == 1)
433 ceph_assert(get_num_any() == items
.size());
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
)
442 ceph_assert(lookup_exact_snap(dname
, last
) == 0);
445 CDentry
* dn
= new CDentry(dname
, inode
->hash_dentry_name(dname
), std::move(alternate_name
), ino
, d_type
, first
, last
);
447 dn
->state_set(CDentry::STATE_AUTH
);
448 mdcache
->lru
.lru_insert_mid(dn
);
451 dn
->version
= get_projected_version();
454 ceph_assert(items
.count(dn
->key()) == 0);
455 //assert(null_items.count(dn->get_name()) == 0);
457 items
[dn
->key()] = dn
;
458 if (last
== CEPH_NOSNAP
)
463 if (state_test(CDir::STATE_DNPINNEDFRAG
)) {
464 dn
->get(CDentry::PIN_FRAGMENTING
);
465 dn
->state_set(CDentry::STATE_FRAGMENTING
);
468 dout(12) << __func__
<< " " << *dn
<< dendl
;
471 if (get_num_any() == 1)
474 ceph_assert(get_num_any() == items
.size());
480 void CDir::remove_dentry(CDentry
*dn
)
482 dout(12) << __func__
<< " " << *dn
<< dendl
;
484 // there should be no client leases at this point!
485 ceph_assert(dn
->client_lease_map
.empty());
487 if (state_test(CDir::STATE_DNPINNEDFRAG
)) {
488 dn
->put(CDentry::PIN_FRAGMENTING
);
489 dn
->state_clear(CDentry::STATE_FRAGMENTING
);
492 if (dn
->get_linkage()->is_null()) {
493 if (dn
->last
== CEPH_NOSNAP
)
498 if (dn
->last
== CEPH_NOSNAP
)
504 if (!dn
->get_linkage()->is_null())
505 // detach inode and dentry
506 unlink_inode_work(dn
);
509 ceph_assert(items
.count(dn
->key()) == 1);
510 items
.erase(dn
->key());
516 if (dn
->state_test(CDentry::STATE_BOTTOMLRU
))
517 mdcache
->bottom_lru
.lru_remove(dn
);
519 mdcache
->lru
.lru_remove(dn
);
523 if (get_num_any() == 0)
525 ceph_assert(get_num_any() == items
.size());
528 void CDir::link_remote_inode(CDentry
*dn
, CInode
*in
)
530 link_remote_inode(dn
, in
->ino(), IFTODT(in
->get_projected_inode()->mode
));
533 void CDir::link_remote_inode(CDentry
*dn
, inodeno_t ino
, unsigned char d_type
)
535 dout(12) << __func__
<< " " << *dn
<< " remote " << ino
<< dendl
;
536 ceph_assert(dn
->get_linkage()->is_null());
538 dn
->get_linkage()->set_remote(ino
, d_type
);
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
);
546 if (dn
->last
== CEPH_NOSNAP
) {
553 ceph_assert(get_num_any() == items
.size());
556 void CDir::link_primary_inode(CDentry
*dn
, CInode
*in
)
558 dout(12) << __func__
<< " " << *dn
<< " " << *in
<< dendl
;
559 ceph_assert(dn
->get_linkage()->is_null());
561 dn
->get_linkage()->inode
= in
;
563 link_inode_work(dn
, in
);
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
);
572 if (dn
->last
== CEPH_NOSNAP
) {
580 ceph_assert(get_num_any() == items
.size());
583 void CDir::link_inode_work( CDentry
*dn
, CInode
*in
)
585 ceph_assert(dn
->get_linkage()->get_inode() == in
);
586 in
->set_primary_parent(dn
);
589 //in->inode.version = dn->get_version();
592 if (in
->get_num_ref())
593 dn
->get(CDentry::PIN_INODEPIN
);
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);
600 // adjust auth pin count
602 dn
->adjust_nested_auth_pins(in
->auth_pins
, NULL
);
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())
609 // verify open snaprealm parent
611 in
->snaprealm
->adjust_parent();
612 else if (in
->is_any_caps())
613 in
->move_to_realm(inode
->find_snaprealm());
616 void CDir::unlink_inode(CDentry
*dn
, bool adjust_lru
)
618 if (dn
->get_linkage()->is_primary()) {
619 dout(12) << __func__
<< " " << *dn
<< " " << *dn
->get_linkage()->get_inode() << dendl
;
621 dout(12) << __func__
<< " " << *dn
<< dendl
;
624 unlink_inode_work(dn
);
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
);
632 if (dn
->last
== CEPH_NOSNAP
) {
639 ceph_assert(get_num_any() == items
.size());
643 void CDir::try_remove_unlinked_dn(CDentry
*dn
)
645 ceph_assert(dn
->dir
== this);
646 ceph_assert(dn
->get_linkage()->is_null());
648 // no pins (besides dirty)?
649 if (dn
->get_num_ref() != dn
->is_dirty())
654 dout(10) << __func__
<< " " << *dn
<< " in " << *this << dendl
;
659 // NOTE: we may not have any more dirty dentries, but the fnode
660 // still changed, so the directory must remain dirty.
665 void CDir::unlink_inode_work(CDentry
*dn
)
667 CInode
*in
= dn
->get_linkage()->get_inode();
669 if (dn
->get_linkage()->is_remote()) {
672 dn
->unlink_remote(dn
->get_linkage());
674 dn
->get_linkage()->set_remote(0, 0);
675 } else if (dn
->get_linkage()->is_primary()) {
678 if (in
->get_num_ref())
679 dn
->put(CDentry::PIN_INODEPIN
);
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);
686 // unlink auth_pin count
688 dn
->adjust_nested_auth_pins(-in
->auth_pins
, nullptr);
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())
696 in
->remove_primary_parent(dn
);
698 in
->item_pop_lru
.remove_myself();
699 dn
->get_linkage()->inode
= 0;
701 ceph_assert(!dn
->get_linkage()->is_null());
705 void CDir::add_to_bloom(CDentry
*dn
)
707 ceph_assert(dn
->last
== CEPH_NOSNAP
);
709 /* not create bloom filter for incomplete dir that was added by log replay */
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()) {
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));
723 /* This size and false positive probability is completely random.*/
724 bloom
->insert(dn
->get_name().data(), dn
->get_name().size());
727 bool CDir::is_in_bloom(std::string_view name
)
731 return bloom
->contains(name
.data(), name
.size());
734 void CDir::remove_null_dentries() {
735 dout(12) << __func__
<< " " << *this << dendl
;
737 auto p
= items
.begin();
738 while (p
!= items
.end()) {
739 CDentry
*dn
= p
->second
;
741 if (dn
->get_linkage()->is_null() && !dn
->is_projected())
745 ceph_assert(num_snap_null
== 0);
746 ceph_assert(num_head_null
== 0);
747 ceph_assert(get_num_any() == items
.size());
750 /** remove dirty null dentries for deleted directory. the dirfrag will be
751 * deleted soon, so it's safe to not commit dirty dentries.
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.
757 void CDir::try_remove_dentries_for_stray()
759 dout(10) << __func__
<< dendl
;
760 ceph_assert(get_parent_dir()->inode
->is_stray());
762 // clear dirty only when the directory was not snapshotted
763 bool clear_dirty
= !inode
->snaprealm
;
765 auto p
= items
.begin();
766 while (p
!= items
.end()) {
767 CDentry
*dn
= p
->second
;
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())
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)
781 ceph_assert(!dn
->is_projected());
782 CDentry::linkage_t
*dnl
= dn
->get_linkage();
784 if (dnl
->is_primary()) {
785 in
= dnl
->get_inode();
786 if (clear_dirty
&& in
->is_dirty())
789 if (clear_dirty
&& dn
->is_dirty())
791 if (dn
->get_num_ref() == 0) {
794 mdcache
->remove_inode(in
);
799 if (clear_dirty
&& is_dirty())
803 bool CDir::try_trim_snap_dentry(CDentry
*dn
, const set
<snapid_t
>& snaps
)
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();
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())
819 dout(10) << " purging snapped " << *in
<< dendl
;
820 mdcache
->remove_inode(in
);
828 void CDir::purge_stale_snap_data(const set
<snapid_t
>& snaps
)
830 dout(10) << __func__
<< " " << snaps
<< dendl
;
832 auto p
= items
.begin();
833 while (p
!= items
.end()) {
834 CDentry
*dn
= p
->second
;
837 if (dn
->last
== CEPH_NOSNAP
)
840 try_trim_snap_dentry(dn
, snaps
);
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.
850 void CDir::steal_dentry(CDentry
*dn
)
852 dout(15) << __func__
<< " " << *dn
<< dendl
;
854 items
[dn
->key()] = dn
;
856 dn
->dir
->items
.erase(dn
->key());
857 if (dn
->dir
->items
.empty())
858 dn
->dir
->put(PIN_CHILD
);
860 if (get_num_any() == 0)
862 if (dn
->get_linkage()->is_null()) {
863 if (dn
->last
== CEPH_NOSNAP
)
867 } else if (dn
->last
== CEPH_NOSNAP
) {
870 auto _fnode
= _get_fnode();
872 if (dn
->get_linkage()->is_primary()) {
873 CInode
*in
= dn
->get_linkage()->get_inode();
874 const auto& pi
= in
->get_projected_inode();
876 _fnode
->fragstat
.nsubdirs
++;
877 if (in
->item_pop_lru
.is_on_list())
878 pop_lru_subdirs
.push_back(&in
->item_pop_lru
);
880 _fnode
->fragstat
.nfiles
++;
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
;
889 if (in
->is_any_caps())
890 adjust_num_inodes_with_caps(1);
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
++;
899 _fnode
->fragstat
.nfiles
++;
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
);
911 int dap
= dn
->get_num_dir_auth_pins();
913 adjust_nested_auth_pins(dap
, NULL
);
914 dn
->dir
->adjust_nested_auth_pins(-dap
, NULL
);
918 if (dn
->is_dirty()) {
919 dirty_dentries
.push_back(&dn
->item_dir_dirty
);
926 void CDir::prepare_old_fragment(map
<string_snap_t
, MDSContext::vec
>& dentry_waiters
, bool replay
)
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())
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
) {
940 waiting_on_dentry
.clear();
945 void CDir::prepare_new_fragment(bool replay
)
947 if (!replay
&& is_auth()) {
951 inode
->add_dirfrag(this);
954 void CDir::finish_old_fragment(MDSContext::vec
& waiters
, bool replay
)
956 // take waiters _before_ unfreeze...
958 take_waiting(WAIT_ANY_MASK
, waiters
);
960 auth_unpin(this); // pinned in prepare_old_fragment
961 ceph_assert(is_frozen_dir());
966 ceph_assert(dir_auth_pins
== 0);
967 ceph_assert(auth_pins
== 0);
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
);
973 // this mirrors init_fragment_pins()
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())
988 ceph_assert(get_num_ref() == (state_test(STATE_STICKY
) ? 1:0));
991 void CDir::init_fragment_pins()
995 if (state_test(STATE_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())
1005 void CDir::split(int bits
, std::vector
<CDir
*>* subs
, MDSContext::vec
& waiters
, bool replay
)
1007 dout(10) << "split by " << bits
<< " bits on " << *this << dendl
;
1009 ceph_assert(replay
|| is_complete() || !is_auth());
1012 frag
.split(bits
, frags
);
1014 vector
<CDir
*> subfrags(1 << bits
);
1016 double fac
= 1.0 / (double)(1 << bits
); // for scaling load vecs
1018 version_t rstat_version
= inode
->get_projected_inode()->rstat
.version
;
1019 version_t dirstat_version
= inode
->get_projected_inode()->dirstat
.version
;
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
;
1029 map
<string_snap_t
, MDSContext::vec
> dentry_waiters
;
1030 prepare_old_fragment(dentry_waiters
, replay
);
1032 // create subfrag dirs
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();
1039 f
->pop_me
.scale(fac
);
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
);
1049 dout(10) << " subfrag " << fg
<< " " << *f
<< dendl
;
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();
1059 // repartition dentries
1060 while (!items
.empty()) {
1061 auto p
= items
.begin();
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
);
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
];
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
);
1084 // FIXME: handle dirty old rstat
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
;
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
);
1107 finish_old_fragment(waiters
, replay
);
1110 void CDir::merge(const std::vector
<CDir
*>& subs
, MDSContext::vec
& waiters
, bool replay
)
1112 dout(10) << "merge " << subs
<< dendl
;
1114 ceph_assert(subs
.size() > 0);
1116 set_dir_auth(subs
.front()->get_dir_auth());
1117 freeze_tree_state
= subs
.front()->freeze_tree_state
;
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
);
1124 prepare_new_fragment(replay
);
1126 auto _fnode
= _get_fnode();
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
;
1134 map
<string_snap_t
, MDSContext::vec
> dentry_waiters
;
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
);
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
);
1146 dir
->prepare_old_fragment(dentry_waiters
, replay
);
1149 while (!dir
->items
.empty())
1150 steal_dentry(dir
->items
.begin()->second
);
1152 // merge replica map
1153 for (const auto &p
: dir
->get_replicas()) {
1154 unsigned cur
= get_replicas()[p
.first
];
1156 get_replicas()[p
.first
] = p
.second
;
1160 if (dir
->get_version() > _fnode
->version
)
1161 _fnode
->version
= projected_version
= dir
->get_version();
1164 state_set(dir
->get_state() & MASK_STATE_FRAGMENT_KEPT
);
1166 dir
->finish_old_fragment(waiters
, replay
);
1167 inode
->close_dirfrag(dir
->get_frag());
1170 if (!dentry_waiters
.empty()) {
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
);
1180 if (is_auth() && !replay
)
1183 // FIXME: merge dirty old rstat
1184 _fnode
->rstat
.version
= rstat_version
;
1185 _fnode
->accounted_rstat
= _fnode
->rstat
;
1186 _fnode
->accounted_rstat
.add(rstatdiff
);
1188 _fnode
->fragstat
.version
= dirstat_version
;
1189 _fnode
->accounted_fragstat
= _fnode
->fragstat
;
1190 _fnode
->accounted_fragstat
.add(fragstatdiff
);
1192 init_fragment_pins();
1198 void CDir::resync_accounted_fragstat()
1200 auto pf
= _get_projected_fnode();
1201 const auto& pi
= inode
->get_projected_inode();
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
;
1211 * resync rstat and accounted_rstat with inode
1213 void CDir::resync_accounted_rstat()
1215 auto pf
= _get_projected_fnode();
1216 const auto& pi
= inode
->get_projected_inode();
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();
1226 void CDir::assimilate_dirty_rstat_inodes(MutationRef
& mut
)
1228 dout(10) << __func__
<< dendl
;
1229 for (elist
<CInode
*>::iterator p
= dirty_rstat_inodes
.begin_use_current();
1232 ceph_assert(in
->is_auth());
1233 if (in
->is_frozen())
1238 auto pi
= in
->project_inode(mut
);
1239 pi
.inode
->version
= in
->pre_dirty();
1241 mdcache
->project_rstat_inode_to_frag(mut
, in
, this, 0, 0, nullptr);
1243 state_set(STATE_ASSIMRSTAT
);
1244 dout(10) << __func__
<< " done" << dendl
;
1247 void CDir::assimilate_dirty_rstat_inodes_finish(EMetaBlob
*blob
)
1249 if (!state_test(STATE_ASSIMRSTAT
))
1251 state_clear(STATE_ASSIMRSTAT
);
1252 dout(10) << __func__
<< dendl
;
1253 elist
<CInode
*>::iterator p
= dirty_rstat_inodes
.begin_use_current();
1258 if (in
->is_frozen())
1261 CDentry
*dn
= in
->get_projected_parent_dn();
1263 in
->clear_dirty_rstat();
1264 blob
->add_primary_dentry(dn
, in
, true);
1267 if (!dirty_rstat_inodes
.empty())
1268 mdcache
->mds
->locker
->mark_updated_scatterlock(&inode
->nestlock
);
1274 /****************************************
1278 void CDir::add_dentry_waiter(std::string_view dname
, snapid_t snapid
, MDSContext
*c
)
1280 if (waiting_on_dentry
.empty())
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
;
1288 void CDir::take_dentry_waiting(std::string_view dname
, snapid_t first
, snapid_t last
,
1289 MDSContext::vec
& ls
)
1291 if (waiting_on_dentry
.empty())
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 "
1302 << " on " << *this << dendl
;
1303 for (const auto &waiter
: it
->second
) {
1304 ls
.push_back(waiter
);
1306 waiting_on_dentry
.erase(it
++);
1309 if (waiting_on_dentry
.empty())
1313 void CDir::take_sub_waiting(MDSContext::vec
& ls
)
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
);
1322 waiting_on_dentry
.clear();
1329 void CDir::add_waiter(uint64_t tag
, MDSContext
*c
)
1334 if (tag
& WAIT_ATSUBTREEROOT
) {
1335 if (!is_subtree_root()) {
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
);
1343 ceph_assert(!(tag
& WAIT_CREATED
) || state_test(STATE_CREATING
));
1345 MDSCacheObject::add_waiter(tag
, c
);
1350 /* NOTE: this checks dentry waiters too */
1351 void CDir::take_waiting(uint64_t mask
, MDSContext::vec
& ls
)
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
);
1362 waiting_on_dentry
.clear();
1367 MDSCacheObject::take_waiting(mask
, ls
);
1371 void CDir::finish_waiting(uint64_t mask
, int result
)
1373 dout(11) << __func__
<< " mask " << hex
<< mask
<< dec
<< " result " << result
<< " on " << *this << dendl
;
1375 MDSContext::vec finished
;
1376 take_waiting(mask
, finished
);
1378 finish_contexts(g_ceph_context
, finished
, result
);
1380 mdcache
->mds
->queue_waiters(finished
);
1387 CDir::fnode_ptr
CDir::project_fnode(const MutationRef
& mut
)
1389 ceph_assert(get_version() != 0);
1391 if (mut
&& mut
->is_projected(this))
1392 return std::const_pointer_cast
<fnode_t
>(projected_fnode
.back());
1394 auto pf
= allocate_fnode(*get_projected_fnode());
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();
1405 projected_fnode
.emplace_back(pf
);
1407 mut
->add_projected_node(this);
1408 dout(10) << __func__
<< " " << pf
.get() << dendl
;
1412 void CDir::pop_and_dirty_projected_fnode(LogSegment
*ls
, const MutationRef
& mut
)
1414 ceph_assert(!projected_fnode
.empty());
1415 auto pf
= std::move(projected_fnode
.front());
1416 dout(15) << __func__
<< " " << pf
.get() << " v" << pf
->version
<< dendl
;
1418 projected_fnode
.pop_front();
1420 mut
->remove_projected_node(this);
1422 reset_fnode(std::move(pf
));
1426 version_t
CDir::pre_dirty(version_t min
)
1428 if (min
> projected_version
)
1429 projected_version
= min
;
1430 ++projected_version
;
1431 dout(10) << __func__
<< " " << projected_version
<< dendl
;
1432 return projected_version
;
1435 void CDir::mark_dirty(LogSegment
*ls
, version_t pv
)
1437 ceph_assert(is_auth());
1440 ceph_assert(get_version() < pv
);
1441 ceph_assert(pv
<= projected_version
);
1442 ceph_assert(!projected_fnode
.empty() &&
1443 pv
<= projected_fnode
.front()->version
);
1449 void CDir::_mark_dirty(LogSegment
*ls
)
1451 if (!state_test(STATE_DIRTY
)) {
1452 dout(10) << __func__
<< " (was clean) " << *this << " version " << get_version() << dendl
;
1456 dout(10) << __func__
<< " (already dirty) " << *this << " version " << get_version() << dendl
;
1459 ls
->dirty_dirfrags
.push_back(&item_dirty
);
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
);
1467 void CDir::mark_new(LogSegment
*ls
)
1469 ls
->new_dirfrags
.push_back(&item_new
);
1470 state_clear(STATE_CREATING
);
1472 MDSContext::vec waiters
;
1473 take_waiting(CDir::WAIT_CREATED
, waiters
);
1474 mdcache
->mds
->queue_waiters(waiters
);
1477 void CDir::mark_clean()
1479 dout(10) << __func__
<< " " << *this << " version " << get_version() << dendl
;
1480 if (state_test(STATE_DIRTY
)) {
1481 item_dirty
.remove_myself();
1482 item_new
.remove_myself();
1484 state_clear(STATE_DIRTY
);
1489 // caller should hold auth pin of this
1490 void CDir::log_mark_dirty()
1492 if (is_dirty() || projected_version
> get_version())
1493 return; // noop if it is already dirty or will be dirty
1495 auto _fnode
= allocate_fnode(*get_fnode());
1496 _fnode
->version
= pre_dirty();
1497 reset_fnode(std::move(_fnode
));
1498 mark_dirty(mdcache
->mds
->mdlog
->get_current_segment());
1501 void CDir::mark_complete() {
1502 state_set(STATE_COMPLETE
);
1506 void CDir::first_get()
1508 inode
->get(CInode::PIN_DIRFRAG
);
1511 void CDir::last_put()
1513 inode
->put(CInode::PIN_DIRFRAG
);
1518 /******************************************************************************
1522 // -----------------------
1524 void CDir::fetch(MDSContext
*c
, bool ignore_authpinnability
)
1527 return fetch(c
, want
, ignore_authpinnability
);
1530 void CDir::fetch(MDSContext
*c
, std::string_view want_dn
, bool ignore_authpinnability
)
1532 dout(10) << "fetch on " << *this << dendl
;
1534 ceph_assert(is_auth());
1535 ceph_assert(!is_complete());
1537 if (!can_auth_pin() && !ignore_authpinnability
) {
1539 dout(7) << "fetch waiting for authpinnable" << dendl
;
1540 add_waiter(WAIT_UNFREEZE
, c
);
1542 dout(7) << "fetch not authpinnable and no context" << dendl
;
1546 // unlinked directory inode shouldn't have any entry
1547 if (!inode
->is_base() && get_parent_dir()->inode
->is_stray() &&
1548 !inode
->snaprealm
) {
1549 dout(7) << "fetch dirfrag for unlinked directory, mark complete" << dendl
;
1550 if (get_version() == 0) {
1551 ceph_assert(inode
->is_auth());
1552 auto _fnode
= allocate_fnode();
1553 _fnode
->version
= 1;
1554 reset_fnode(std::move(_fnode
));
1556 if (state_test(STATE_REJOINUNDEF
)) {
1557 ceph_assert(mdcache
->mds
->is_rejoin());
1558 state_clear(STATE_REJOINUNDEF
);
1559 mdcache
->opened_undef_dirfrag(this);
1565 mdcache
->mds
->queue_waiter(c
);
1569 if (c
) add_waiter(WAIT_COMPLETE
, c
);
1570 if (!want_dn
.empty()) wanted_items
.insert(mempool::mds_co::string(want_dn
));
1572 // already fetching?
1573 if (state_test(CDir::STATE_FETCHING
)) {
1574 dout(7) << "already fetching; waiting" << dendl
;
1579 state_set(CDir::STATE_FETCHING
);
1581 if (mdcache
->mds
->logger
) mdcache
->mds
->logger
->inc(l_mds_dir_fetch
);
1583 mdcache
->mds
->balancer
->hit_dir(this, META_POP_FETCH
);
1585 std::set
<dentry_key_t
> empty
;
1586 _omap_fetch(NULL
, empty
);
1589 void CDir::fetch(MDSContext
*c
, const std::set
<dentry_key_t
>& keys
)
1591 dout(10) << "fetch " << keys
.size() << " keys on " << *this << dendl
;
1593 ceph_assert(is_auth());
1594 ceph_assert(!is_complete());
1596 if (!can_auth_pin()) {
1597 dout(7) << "fetch keys waiting for authpinnable" << dendl
;
1598 add_waiter(WAIT_UNFREEZE
, c
);
1601 if (state_test(CDir::STATE_FETCHING
)) {
1602 dout(7) << "fetch keys waiting for full fetch" << dendl
;
1603 add_waiter(WAIT_COMPLETE
, c
);
1608 if (mdcache
->mds
->logger
) mdcache
->mds
->logger
->inc(l_mds_dir_fetch
);
1610 mdcache
->mds
->balancer
->hit_dir(this, META_POP_FETCH
);
1612 _omap_fetch(c
, keys
);
1615 class C_IO_Dir_OMAP_FetchedMore
: public CDirIOContext
{
1618 const version_t omap_version
;
1621 map
<string
, bufferlist
> omap
; ///< carry-over from before
1622 map
<string
, bufferlist
> omap_more
; ///< new batch
1624 C_IO_Dir_OMAP_FetchedMore(CDir
*d
, version_t v
, MDSContext
*f
) :
1625 CDirIOContext(d
), fin(f
), omap_version(v
), ret(0) { }
1626 void finish(int r
) {
1627 if (omap_version
< dir
->get_committed_version()) {
1629 dir
->_omap_fetch(fin
, {});
1635 omap
.swap(omap_more
);
1637 omap
.insert(omap_more
.begin(), omap_more
.end());
1640 dir
->_omap_fetch_more(omap_version
, hdrbl
, omap
, fin
);
1642 dir
->_omap_fetched(hdrbl
, omap
, !fin
, r
);
1647 void print(ostream
& out
) const override
{
1648 out
<< "dirfrag_fetch_more(" << dir
->dirfrag() << ")";
1652 class C_IO_Dir_OMAP_Fetched
: public CDirIOContext
{
1655 const version_t omap_version
;
1658 map
<string
, bufferlist
> omap
;
1660 int ret1
, ret2
, ret3
;
1662 C_IO_Dir_OMAP_Fetched(CDir
*d
, MDSContext
*f
) :
1663 CDirIOContext(d
), fin(f
),
1664 omap_version(d
->get_committing_version()),
1665 ret1(0), ret2(0), ret3(0) { }
1666 void finish(int r
) override
{
1667 // check the correctness of backtrace
1668 if (r
>= 0 && ret3
!= -CEPHFS_ECANCELED
)
1669 dir
->inode
->verify_diri_backtrace(btbl
, ret3
);
1670 if (r
>= 0) r
= ret1
;
1671 if (r
>= 0) r
= ret2
;
1674 if (omap_version
< dir
->get_committed_version()) {
1676 dir
->_omap_fetch(fin
, {});
1678 dir
->_omap_fetch_more(omap_version
, hdrbl
, omap
, fin
);
1683 dir
->_omap_fetched(hdrbl
, omap
, !fin
, r
);
1688 void print(ostream
& out
) const override
{
1689 out
<< "dirfrag_fetch(" << dir
->dirfrag() << ")";
1693 void CDir::_omap_fetch(MDSContext
*c
, const std::set
<dentry_key_t
>& keys
)
1695 C_IO_Dir_OMAP_Fetched
*fin
= new C_IO_Dir_OMAP_Fetched(this, c
);
1696 object_t oid
= get_ondisk_object();
1697 object_locator_t
oloc(mdcache
->mds
->mdsmap
->get_metadata_pool());
1699 rd
.omap_get_header(&fin
->hdrbl
, &fin
->ret1
);
1702 rd
.omap_get_vals("", "", g_conf()->mds_dir_keys_per_op
,
1703 &fin
->omap
, &fin
->more
, &fin
->ret2
);
1706 std::set
<std::string
> str_keys
;
1707 for (auto p
: keys
) {
1710 str_keys
.insert(str
);
1712 rd
.omap_get_vals_by_keys(str_keys
, &fin
->omap
, &fin
->ret2
);
1714 // check the correctness of backtrace
1715 if (g_conf()->mds_verify_backtrace
> 0 && frag
== frag_t()) {
1716 rd
.getxattr("parent", &fin
->btbl
, &fin
->ret3
);
1717 rd
.set_last_op_flags(CEPH_OSD_OP_FLAG_FAILOK
);
1719 fin
->ret3
= -CEPHFS_ECANCELED
;
1722 mdcache
->mds
->objecter
->read(oid
, oloc
, rd
, CEPH_NOSNAP
, NULL
, 0,
1723 new C_OnFinisher(fin
, mdcache
->mds
->finisher
));
1726 void CDir::_omap_fetch_more(version_t omap_version
, bufferlist
& hdrbl
,
1727 map
<string
, bufferlist
>& omap
, MDSContext
*c
)
1729 // we have more omap keys to fetch!
1730 object_t oid
= get_ondisk_object();
1731 object_locator_t
oloc(mdcache
->mds
->mdsmap
->get_metadata_pool());
1732 auto fin
= new C_IO_Dir_OMAP_FetchedMore(this, omap_version
, c
);
1733 fin
->hdrbl
= std::move(hdrbl
);
1734 fin
->omap
.swap(omap
);
1736 rd
.omap_get_vals(fin
->omap
.rbegin()->first
,
1737 "", /* filter prefix */
1738 g_conf()->mds_dir_keys_per_op
,
1742 mdcache
->mds
->objecter
->read(oid
, oloc
, rd
, CEPH_NOSNAP
, NULL
, 0,
1743 new C_OnFinisher(fin
, mdcache
->mds
->finisher
));
1746 CDentry
*CDir::_load_dentry(
1747 std::string_view key
,
1748 std::string_view dname
,
1749 const snapid_t last
,
1752 const std::set
<snapid_t
> *snaps
,
1753 double rand_threshold
,
1756 auto q
= bl
.cbegin();
1765 dout(20) << "_fetched pos " << pos
<< " marker '" << type
<< "' dname '" << dname
1766 << " [" << first
<< "," << last
<< "]"
1770 if (snaps
&& last
!= CEPH_NOSNAP
) {
1771 set
<snapid_t
>::const_iterator p
= snaps
->lower_bound(first
);
1772 if (p
== snaps
->end() || *p
> last
) {
1773 dout(10) << " skipping stale dentry on [" << first
<< "," << last
<< "]" << dendl
;
1779 * look for existing dentry for _last_ snap, because unlink +
1780 * create may leave a "hole" (epochs during which the dentry
1781 * doesn't exist) but for which no explicit negative dentry is in
1786 dn
= lookup_exact_snap(dname
, last
);
1788 dn
= lookup(dname
, last
);
1790 if (type
== 'L' || type
== 'l') {
1793 unsigned char d_type
;
1794 mempool::mds_co::string alternate_name
;
1796 CDentry::decode_remote(type
, ino
, d_type
, alternate_name
, q
);
1800 stale_items
.insert(mempool::mds_co::string(key
));
1801 *force_dirty
= true;
1807 CDentry::linkage_t
*dnl
= dn
->get_linkage();
1808 dout(12) << "_fetched had " << (dnl
->is_null() ? "NEG" : "") << " dentry " << *dn
<< dendl
;
1809 if (committed_version
== 0 &&
1812 ino
== dnl
->get_remote_ino() &&
1813 d_type
== dnl
->get_remote_d_type() &&
1814 alternate_name
== dn
->get_alternate_name()) {
1815 // see comment below
1816 dout(10) << "_fetched had underwater dentry " << *dn
<< ", marking clean" << dendl
;
1821 dn
= add_remote_dentry(dname
, ino
, d_type
, std::move(alternate_name
), first
, last
);
1824 CInode
*in
= mdcache
->get_inode(ino
); // we may or may not have it.
1826 dn
->link_remote(dn
->get_linkage(), in
);
1827 dout(12) << "_fetched got remote link " << ino
<< " which we have " << *in
<< dendl
;
1829 dout(12) << "_fetched got remote link " << ino
<< " (don't have it)" << dendl
;
1833 else if (type
== 'I' || type
== 'i') {
1834 InodeStore inode_data
;
1835 mempool::mds_co::string alternate_name
;
1837 // Load inode data before looking up or constructing CInode
1840 if (struct_v
>= 2) {
1841 decode(alternate_name
, q
);
1843 inode_data
.decode(q
);
1846 inode_data
.decode_bare(q
);
1851 stale_items
.insert(mempool::mds_co::string(key
));
1852 *force_dirty
= true;
1857 bool undef_inode
= false;
1859 CDentry::linkage_t
*dnl
= dn
->get_linkage();
1860 dout(12) << "_fetched had " << (dnl
->is_null() ? "NEG" : "") << " dentry " << *dn
<< dendl
;
1862 if (dnl
->is_primary()) {
1863 CInode
*in
= dnl
->get_inode();
1864 if (in
->state_test(CInode::STATE_REJOINUNDEF
)) {
1866 } else if (committed_version
== 0 &&
1868 inode_data
.inode
->ino
== in
->ino() &&
1869 inode_data
.inode
->version
== in
->get_version()) {
1870 /* clean underwater item?
1871 * Underwater item is something that is dirty in our cache from
1872 * journal replay, but was previously flushed to disk before the
1875 * We only do this is committed_version == 0. that implies either
1876 * - this is a fetch after from a clean/empty CDir is created
1877 * (and has no effect, since the dn won't exist); or
1878 * - this is a fetch after _recovery_, which is what we're worried
1879 * about. Items that are marked dirty from the journal should be
1880 * marked clean if they appear on disk.
1882 dout(10) << "_fetched had underwater dentry " << *dn
<< ", marking clean" << dendl
;
1884 dout(10) << "_fetched had underwater inode " << *dnl
->get_inode() << ", marking clean" << dendl
;
1890 if (!dn
|| undef_inode
) {
1892 CInode
*in
= mdcache
->get_inode(inode_data
.inode
->ino
, last
);
1893 if (!in
|| undef_inode
) {
1894 if (undef_inode
&& in
)
1897 in
= new CInode(mdcache
, true, first
, last
);
1899 in
->reset_inode(std::move(inode_data
.inode
));
1900 in
->reset_xattrs(std::move(inode_data
.xattrs
));
1902 if (in
->is_symlink())
1903 in
->symlink
= inode_data
.symlink
;
1905 in
->dirfragtree
.swap(inode_data
.dirfragtree
);
1906 in
->reset_old_inodes(std::move(inode_data
.old_inodes
));
1907 if (in
->is_any_old_inodes()) {
1908 snapid_t min_first
= in
->get_old_inodes()->rbegin()->first
+ 1;
1909 if (min_first
> in
->first
)
1910 in
->first
= min_first
;
1913 in
->oldest_snap
= inode_data
.oldest_snap
;
1914 in
->decode_snap_blob(inode_data
.snap_blob
);
1915 if (snaps
&& !in
->snaprealm
)
1916 in
->purge_stale_snap_data(*snaps
);
1919 mdcache
->add_inode(in
); // add
1920 dn
= add_primary_dentry(dname
, in
, std::move(alternate_name
), first
, last
); // link
1922 dout(12) << "_fetched got " << *dn
<< " " << *in
<< dendl
;
1924 if (in
->get_inode()->is_dirty_rstat())
1925 in
->mark_dirty_rstat();
1927 in
->maybe_ephemeral_rand(rand_threshold
);
1928 //in->hack_accessed = false;
1929 //in->hack_load_stamp = ceph_clock_now();
1930 //num_new_inodes_loaded++;
1931 } else if (g_conf().get_val
<bool>("mds_hack_allow_loading_invalid_metadata")) {
1932 dout(20) << "hack: adding duplicate dentry for " << *in
<< dendl
;
1933 dn
= add_primary_dentry(dname
, in
, std::move(alternate_name
), first
, last
);
1935 dout(0) << "_fetched badness: got (but i already had) " << *in
1936 << " mode " << in
->get_inode()->mode
1937 << " mtime " << in
->get_inode()->mtime
<< dendl
;
1938 string dirpath
, inopath
;
1939 this->inode
->make_path_string(dirpath
);
1940 in
->make_path_string(inopath
);
1941 mdcache
->mds
->clog
->error() << "loaded dup inode " << inode_data
.inode
->ino
1942 << " [" << first
<< "," << last
<< "] v" << inode_data
.inode
->version
1943 << " at " << dirpath
<< "/" << dname
1944 << ", but inode " << in
->vino() << " v" << in
->get_version()
1945 << " already exists at " << inopath
;
1950 CachedStackStringStream css
;
1951 *css
<< "Invalid tag char '" << type
<< "' pos " << pos
;
1952 throw buffer::malformed_input(css
->str());
1958 void CDir::_omap_fetched(bufferlist
& hdrbl
, map
<string
, bufferlist
>& omap
,
1959 bool complete
, int r
)
1961 LogChannelRef clog
= mdcache
->mds
->clog
;
1962 dout(10) << "_fetched header " << hdrbl
.length() << " bytes "
1963 << omap
.size() << " keys for " << *this << dendl
;
1965 ceph_assert(r
== 0 || r
== -CEPHFS_ENOENT
|| r
== -CEPHFS_ENODATA
);
1966 ceph_assert(is_auth());
1967 ceph_assert(!is_frozen());
1969 if (hdrbl
.length() == 0) {
1970 dout(0) << "_fetched missing object for " << *this << dendl
;
1972 clog
->error() << "dir " << dirfrag() << " object missing on disk; some "
1973 "files may be lost (" << get_path() << ")";
1981 auto p
= hdrbl
.cbegin();
1983 decode(got_fnode
, p
);
1984 } catch (const buffer::error
&err
) {
1985 derr
<< "Corrupt fnode in dirfrag " << dirfrag()
1986 << ": " << err
.what() << dendl
;
1987 clog
->warn() << "Corrupt fnode header in " << dirfrag() << ": "
1988 << err
.what() << " (" << get_path() << ")";
1993 clog
->warn() << "header buffer of dir " << dirfrag() << " has "
1994 << hdrbl
.length() - p
.get_off() << " extra bytes ("
1995 << get_path() << ")";
2001 dout(10) << "_fetched version " << got_fnode
.version
<< dendl
;
2003 // take the loaded fnode?
2004 // only if we are a fresh CDir* with no prior state.
2005 if (get_version() == 0) {
2006 ceph_assert(!is_projected());
2007 ceph_assert(!state_test(STATE_COMMITTING
));
2008 auto _fnode
= allocate_fnode(got_fnode
);
2009 reset_fnode(std::move(_fnode
));
2010 projected_version
= committing_version
= committed_version
= get_version();
2012 if (state_test(STATE_REJOINUNDEF
)) {
2013 ceph_assert(mdcache
->mds
->is_rejoin());
2014 state_clear(STATE_REJOINUNDEF
);
2015 mdcache
->opened_undef_dirfrag(this);
2019 list
<CInode
*> undef_inodes
;
2021 // purge stale snaps?
2022 bool force_dirty
= false;
2023 const set
<snapid_t
> *snaps
= NULL
;
2024 SnapRealm
*realm
= inode
->find_snaprealm();
2025 if (fnode
->snap_purged_thru
< realm
->get_last_destroyed()) {
2026 snaps
= &realm
->get_snaps();
2027 dout(10) << " snap_purged_thru " << fnode
->snap_purged_thru
2028 << " < " << realm
->get_last_destroyed()
2029 << ", snap purge based on " << *snaps
<< dendl
;
2030 if (get_num_snap_items() == 0) {
2031 const_cast<snapid_t
&>(fnode
->snap_purged_thru
) = realm
->get_last_destroyed();
2036 unsigned pos
= omap
.size() - 1;
2037 double rand_threshold
= get_inode()->get_ephemeral_rand();
2038 for (map
<string
, bufferlist
>::reverse_iterator p
= omap
.rbegin();
2043 dentry_key_t::decode_helper(p
->first
, dname
, last
);
2048 p
->first
, dname
, last
, p
->second
, pos
, snaps
,
2049 rand_threshold
, &force_dirty
);
2050 } catch (const buffer::error
&err
) {
2051 mdcache
->mds
->clog
->warn() << "Corrupt dentry '" << dname
<< "' in "
2052 "dir frag " << dirfrag() << ": "
2053 << err
.what() << "(" << get_path() << ")";
2055 // Remember that this dentry is damaged. Subsequent operations
2056 // that try to act directly on it will get their CEPHFS_EIOs, but this
2057 // dirfrag as a whole will continue to look okay (minus the
2058 // mysteriously-missing dentry)
2059 go_bad_dentry(last
, dname
);
2061 // Anyone who was WAIT_DENTRY for this guy will get kicked
2062 // to RetryRequest, and hit the DamageTable-interrogating path.
2063 // Stats will now be bogus because we will think we're complete,
2064 // but have 1 or more missing dentries.
2071 CDentry::linkage_t
*dnl
= dn
->get_linkage();
2072 if (dnl
->is_primary() && dnl
->get_inode()->state_test(CInode::STATE_REJOINUNDEF
))
2073 undef_inodes
.push_back(dnl
->get_inode());
2075 if (wanted_items
.count(mempool::mds_co::string(dname
)) > 0 || !complete
) {
2076 dout(10) << " touching wanted dn " << *dn
<< dendl
;
2077 mdcache
->touch_dentry(dn
);
2081 //cache->mds->logger->inc("newin", num_new_inodes_loaded);
2083 // mark complete, !fetching
2085 wanted_items
.clear();
2087 state_clear(STATE_FETCHING
);
2090 // open & force frags
2091 while (!undef_inodes
.empty()) {
2092 CInode
*in
= undef_inodes
.front();
2093 undef_inodes
.pop_front();
2094 in
->state_clear(CInode::STATE_REJOINUNDEF
);
2095 mdcache
->opened_undef_inode(in
);
2098 // dirty myself to remove stale snap dentries
2099 if (force_dirty
&& !mdcache
->is_readonly())
2106 finish_waiting(WAIT_COMPLETE
, 0);
2110 void CDir::go_bad_dentry(snapid_t last
, std::string_view dname
)
2112 dout(10) << __func__
<< " " << dname
<< dendl
;
2113 std::string
path(get_path());
2116 const bool fatal
= mdcache
->mds
->damage_table
.notify_dentry(
2117 inode
->ino(), frag
, last
, dname
, path
);
2119 mdcache
->mds
->damaged();
2120 ceph_abort(); // unreachable, damaged() respawns us
2124 void CDir::go_bad(bool complete
)
2126 dout(10) << __func__
<< " " << frag
<< dendl
;
2127 const bool fatal
= mdcache
->mds
->damage_table
.notify_dirfrag(
2128 inode
->ino(), frag
, get_path());
2130 mdcache
->mds
->damaged();
2131 ceph_abort(); // unreachable, damaged() respawns us
2135 if (get_version() == 0) {
2136 auto _fnode
= allocate_fnode();
2137 _fnode
->version
= 1;
2138 reset_fnode(std::move(_fnode
));
2141 state_set(STATE_BADFRAG
);
2145 state_clear(STATE_FETCHING
);
2147 finish_waiting(WAIT_COMPLETE
, -CEPHFS_EIO
);
2150 // -----------------------
2156 * @param want - min version i want committed
2157 * @param c - callback for completion
2159 void CDir::commit(version_t want
, MDSContext
*c
, bool ignore_authpinnability
, int op_prio
)
2161 dout(10) << "commit want " << want
<< " on " << *this << dendl
;
2162 if (want
== 0) want
= get_version();
2165 ceph_assert(want
<= get_version() || get_version() == 0); // can't commit the future
2166 ceph_assert(want
> committed_version
); // the caller is stupid
2167 ceph_assert(is_auth());
2168 ceph_assert(ignore_authpinnability
|| can_auth_pin());
2170 // note: queue up a noop if necessary, so that we always
2173 c
= new C_MDSInternalNoop
;
2175 // auth_pin on first waiter
2176 if (waiting_for_commit
.empty())
2178 waiting_for_commit
[want
].push_back(c
);
2181 _commit(want
, op_prio
);
2184 class C_IO_Dir_Committed
: public CDirIOContext
{
2187 C_IO_Dir_Committed(CDir
*d
, version_t v
) : CDirIOContext(d
), version(v
) { }
2188 void finish(int r
) override
{
2189 dir
->_committed(r
, version
);
2191 void print(ostream
& out
) const override
{
2192 out
<< "dirfrag_committed(" << dir
->dirfrag() << ")";
2196 class C_IO_Dir_Commit_Ops
: public Context
{
2198 C_IO_Dir_Commit_Ops(CDir
*d
, int pr
,
2199 vector
<CDir::dentry_commit_item
> &&s
, bufferlist
&&bl
,
2201 mempool::mds_co::compact_set
<mempool::mds_co::string
> &&stales
) :
2202 dir(d
), op_prio(pr
) {
2203 metapool
= dir
->mdcache
->mds
->get_metadata_pool();
2204 version
= dir
->get_version();
2205 is_new
= dir
->is_new();
2209 stale_items
.swap(stales
);
2212 void finish(int r
) override
{
2213 dir
->_omap_commit_ops(r
, op_prio
, metapool
, version
, is_new
, to_set
, dfts
,
2214 to_remove
, stale_items
);
2223 vector
<CDir::dentry_commit_item
> to_set
;
2225 vector
<string
> to_remove
;
2226 mempool::mds_co::compact_set
<mempool::mds_co::string
> stale_items
;
2229 // This is doing the same thing with the InodeStoreBase::encode()
2230 void CDir::_encode_primary_inode_base(dentry_commit_item
&item
, bufferlist
&dfts
,
2233 ENCODE_START(6, 4, bl
);
2234 encode(*item
.inode
, bl
, item
.features
);
2236 if (!item
.symlink
.empty())
2237 encode(item
.symlink
, bl
);
2240 dfts
.splice(0, item
.dft_len
, &bl
);
2243 encode(*item
.xattrs
, bl
);
2245 encode((__u32
)0, bl
);
2247 if (item
.snaprealm
) {
2248 bufferlist snapr_bl
;
2249 encode(item
.srnode
, snapr_bl
);
2250 encode(snapr_bl
, bl
);
2252 encode(bufferlist(), bl
);
2255 if (item
.old_inodes
)
2256 encode(*item
.old_inodes
, bl
, item
.features
);
2258 encode((__u32
)0, bl
);
2260 encode(item
.oldest_snap
, bl
);
2261 encode(item
.damage_flags
, bl
);
2265 // This is not locked by mds_lock
2266 void CDir::_omap_commit_ops(int r
, int op_prio
, int64_t metapool
, version_t version
, bool _new
,
2267 vector
<dentry_commit_item
> &to_set
, bufferlist
&dfts
,
2268 vector
<string
>& to_remove
,
2269 mempool::mds_co::compact_set
<mempool::mds_co::string
> &stales
)
2271 dout(10) << __func__
<< dendl
;
2274 mdcache
->mds
->handle_write_error_with_lock(r
);
2278 C_GatherBuilder
gather(g_ceph_context
,
2279 new C_OnFinisher(new C_IO_Dir_Committed(this, version
),
2280 mdcache
->mds
->finisher
));
2283 object_t oid
= get_ondisk_object();
2284 object_locator_t
oloc(metapool
);
2286 map
<string
, bufferlist
> _set
;
2289 unsigned max_write_size
= mdcache
->max_dir_commit_size
;
2290 unsigned write_size
= 0;
2292 auto commit_one
= [&](bool header
=false) {
2295 // don't create new dirfrag blindly
2297 op
.stat(nullptr, nullptr, nullptr);
2300 * save the header at the last moment.. If we were to send it off before
2301 * other updates, but die before sending them all, we'd think that the
2302 * on-disk state was fully committed even though it wasn't! However, since
2303 * the messages are strictly ordered between the MDS and the OSD, and
2304 * since messages to a given PG are strictly ordered, if we simply send
2305 * the message containing the header off last, we cannot get our header
2306 * into an incorrect state.
2310 encode(*fnode
, header
);
2311 op
.omap_set_header(header
);
2314 op
.priority
= op_prio
;
2318 op
.omap_rm_keys(_rm
);
2319 mdcache
->mds
->objecter
->mutate(oid
, oloc
, op
, snapc
,
2320 ceph::real_clock::now(),
2321 0, gather
.new_sub());
2327 for (auto &key
: stales
) {
2328 unsigned size
= key
.length() + sizeof(__u32
);
2329 if (write_size
+ size
> max_write_size
)
2336 for (auto &key
: to_remove
) {
2337 unsigned size
= key
.length() + sizeof(__u32
);
2338 if (write_size
+ size
> max_write_size
)
2342 _rm
.emplace(std::move(key
));
2347 for (auto &item
: to_set
) {
2348 encode(item
.first
, bl
);
2349 if (item
.is_remote
) {
2351 CDentry::encode_remote(item
.ino
, item
.d_type
, item
.alternate_name
, bl
);
2353 // marker, name, inode, [symlink string]
2354 bl
.append('i'); // inode
2356 ENCODE_START(2, 1, bl
);
2357 encode(item
.alternate_name
, bl
);
2358 _encode_primary_inode_base(item
, dfts
, bl
);
2362 unsigned size
= item
.key
.length() + bl
.length() + 2 * sizeof(__u32
);
2363 if (write_size
+ size
> max_write_size
)
2367 _set
[std::move(item
.key
)].swap(bl
);
2375 * Flush out the modified dentries in this dir. Keep the bufferlist
2376 * below max_write_size;
2378 void CDir::_omap_commit(int op_prio
)
2380 dout(10) << __func__
<< dendl
;
2383 op_prio
= CEPH_MSG_PRIO_DEFAULT
;
2386 const set
<snapid_t
> *snaps
= NULL
;
2387 SnapRealm
*realm
= inode
->find_snaprealm();
2388 if (fnode
->snap_purged_thru
< realm
->get_last_destroyed()) {
2389 snaps
= &realm
->get_snaps();
2390 dout(10) << " snap_purged_thru " << fnode
->snap_purged_thru
2391 << " < " << realm
->get_last_destroyed()
2392 << ", snap purge based on " << *snaps
<< dendl
;
2393 // fnode.snap_purged_thru = realm->get_last_destroyed();
2397 if (state_test(CDir::STATE_FRAGMENTING
) && is_new()) {
2398 count
= get_num_head_items() + get_num_snap_items();
2400 for (elist
<CDentry
*>::iterator it
= dirty_dentries
.begin(); !it
.end(); ++it
)
2404 vector
<string
> to_remove
;
2405 // reverve enough memories, which maybe larger than the actually needed
2406 to_remove
.reserve(count
);
2408 vector
<dentry_commit_item
> to_set
;
2409 // reverve enough memories, which maybe larger than the actually needed
2410 to_set
.reserve(count
);
2412 // for dir fragtrees
2413 bufferlist
dfts(CEPH_PAGE_SIZE
);
2415 auto write_one
= [&](CDentry
*dn
) {
2417 dn
->key().encode(key
);
2419 if (dn
->last
!= CEPH_NOSNAP
&&
2420 snaps
&& try_trim_snap_dentry(dn
, *snaps
)) {
2421 dout(10) << " rm " << key
<< dendl
;
2422 to_remove
.emplace_back(std::move(key
));
2426 if (dn
->get_linkage()->is_null()) {
2427 dout(10) << " rm " << dn
->get_name() << " " << *dn
<< dendl
;
2428 to_remove
.emplace_back(std::move(key
));
2430 dout(10) << " set " << dn
->get_name() << " " << *dn
<< dendl
;
2432 uint64_t off
= dfts
.length();
2433 // try to reserve new size if there has less
2434 // than 1/8 page space
2435 uint64_t left
= CEPH_PAGE_SIZE
- off
% CEPH_PAGE_SIZE
;
2436 if (left
< CEPH_PAGE_SIZE
/ 8)
2437 dfts
.reserve(left
+ CEPH_PAGE_SIZE
);
2439 auto& item
= to_set
.emplace_back();
2440 item
.key
= std::move(key
);
2441 _parse_dentry(dn
, item
, snaps
, dfts
);
2442 item
.dft_len
= dfts
.length() - off
;
2446 if (state_test(CDir::STATE_FRAGMENTING
) && is_new()) {
2447 ceph_assert(committed_version
== 0);
2448 for (auto p
= items
.begin(); p
!= items
.end(); ) {
2449 CDentry
*dn
= p
->second
;
2451 if (dn
->get_linkage()->is_null())
2456 for (auto p
= dirty_dentries
.begin(); !p
.end(); ) {
2463 auto c
= new C_IO_Dir_Commit_Ops(this, op_prio
, std::move(to_set
), std::move(dfts
),
2464 std::move(to_remove
), std::move(stale_items
));
2465 stale_items
.clear();
2466 mdcache
->mds
->finisher
->queue(c
);
2469 void CDir::_parse_dentry(CDentry
*dn
, dentry_commit_item
&item
,
2470 const set
<snapid_t
> *snaps
, bufferlist
&bl
)
2472 // clear dentry NEW flag, if any. we can no longer silently drop it.
2475 item
.first
= dn
->first
;
2477 // primary or remote?
2478 auto& linkage
= dn
->linkage
;
2479 item
.alternate_name
= dn
->get_alternate_name();
2480 if (linkage
.is_remote()) {
2481 item
.is_remote
= true;
2482 item
.ino
= linkage
.get_remote_ino();
2483 item
.d_type
= linkage
.get_remote_d_type();
2484 dout(14) << " dn '" << dn
->get_name() << "' remote ino " << item
.ino
<< dendl
;
2485 } else if (linkage
.is_primary()) {
2487 CInode
*in
= linkage
.get_inode();
2490 dout(14) << " dn '" << dn
->get_name() << "' inode " << *in
<< dendl
;
2492 if (in
->is_multiversion()) {
2493 if (!in
->snaprealm
) {
2495 in
->purge_stale_snap_data(*snaps
);
2497 in
->purge_stale_snap_data(in
->snaprealm
->get_snaps());
2501 if (in
->snaprealm
) {
2502 item
.snaprealm
= true;
2503 item
.srnode
= in
->snaprealm
->srnode
;
2505 item
.features
= mdcache
->mds
->mdsmap
->get_up_features();
2506 item
.inode
= in
->inode
;
2507 if (in
->inode
->is_symlink())
2508 item
.symlink
= in
->symlink
;
2510 encode(in
->dirfragtree
, bl
);
2511 item
.xattrs
= in
->xattrs
;
2512 item
.old_inodes
= in
->old_inodes
;
2513 item
.oldest_snap
= in
->oldest_snap
;
2514 item
.damage_flags
= in
->damage_flags
;
2516 ceph_assert(!linkage
.is_null());
2520 void CDir::_commit(version_t want
, int op_prio
)
2522 dout(10) << "_commit want " << want
<< " on " << *this << dendl
;
2524 // we can't commit things in the future.
2525 // (even the projected future.)
2526 ceph_assert(want
<= get_version() || get_version() == 0);
2528 // check pre+postconditions.
2529 ceph_assert(is_auth());
2531 // already committed?
2532 if (committed_version
>= want
) {
2533 dout(10) << "already committed " << committed_version
<< " >= " << want
<< dendl
;
2536 // already committing >= want?
2537 if (committing_version
>= want
) {
2538 dout(10) << "already committing " << committing_version
<< " >= " << want
<< dendl
;
2539 ceph_assert(state_test(STATE_COMMITTING
));
2543 // alrady committed an older version?
2544 if (committing_version
> committed_version
) {
2545 dout(10) << "already committing older " << committing_version
<< ", waiting for that to finish" << dendl
;
2550 committing_version
= get_version();
2552 // mark committing (if not already)
2553 if (!state_test(STATE_COMMITTING
)) {
2554 dout(10) << "marking committing" << dendl
;
2555 state_set(STATE_COMMITTING
);
2558 if (mdcache
->mds
->logger
) mdcache
->mds
->logger
->inc(l_mds_dir_commit
);
2560 mdcache
->mds
->balancer
->hit_dir(this, META_POP_STORE
);
2562 _omap_commit(op_prio
);
2569 * @param v version i just committed
2571 void CDir::_committed(int r
, version_t v
)
2574 // the directory could be partly purged during MDS failover
2575 if (r
== -CEPHFS_ENOENT
&& committed_version
== 0 &&
2576 !inode
->is_base() && get_parent_dir()->inode
->is_stray()) {
2578 if (inode
->snaprealm
)
2579 inode
->state_set(CInode::STATE_MISSINGOBJS
);
2582 dout(1) << "commit error " << r
<< " v " << v
<< dendl
;
2583 mdcache
->mds
->clog
->error() << "failed to commit dir " << dirfrag() << " object,"
2585 mdcache
->mds
->handle_write_error(r
);
2590 dout(10) << "_committed v " << v
<< " on " << *this << dendl
;
2591 ceph_assert(is_auth());
2593 bool stray
= inode
->is_stray();
2596 ceph_assert(v
> committed_version
);
2597 ceph_assert(v
<= committing_version
);
2598 committed_version
= v
;
2600 // _all_ commits done?
2601 if (committing_version
== committed_version
)
2602 state_clear(CDir::STATE_COMMITTING
);
2604 // _any_ commit, even if we've been redirtied, means we're no longer new.
2605 item_new
.remove_myself();
2608 if (committed_version
== get_version())
2612 for (auto p
= dirty_dentries
.begin(); !p
.end(); ) {
2617 if (dn
->linkage
.is_primary()) {
2618 CInode
*in
= dn
->linkage
.get_inode();
2620 ceph_assert(in
->is_auth());
2622 if (committed_version
>= in
->get_version()) {
2623 if (in
->is_dirty()) {
2624 dout(15) << " dir " << committed_version
<< " >= inode " << in
->get_version() << " now clean " << *in
<< dendl
;
2628 dout(15) << " dir " << committed_version
<< " < inode " << in
->get_version() << " still dirty " << *in
<< dendl
;
2629 ceph_assert(in
->is_dirty() || in
->last
< CEPH_NOSNAP
); // special case for cow snap items (not predirtied)
2634 if (committed_version
>= dn
->get_version()) {
2635 dout(15) << " dir " << committed_version
<< " >= dn " << dn
->get_version() << " now clean " << *dn
<< dendl
;
2638 // drop clean null stray dentries immediately
2640 dn
->get_num_ref() == 0 &&
2641 !dn
->is_projected() &&
2642 dn
->get_linkage()->is_null())
2645 dout(15) << " dir " << committed_version
<< " < dn " << dn
->get_version() << " still dirty " << *dn
<< dendl
;
2646 ceph_assert(dn
->is_dirty());
2651 bool were_waiters
= !waiting_for_commit
.empty();
2653 auto it
= waiting_for_commit
.begin();
2654 while (it
!= waiting_for_commit
.end()) {
2657 if (it
->first
> committed_version
) {
2658 dout(10) << " there are waiters for " << it
->first
<< ", committing again" << dendl
;
2659 _commit(it
->first
, -1);
2663 for (const auto &waiter
: it
->second
)
2664 t
.push_back(waiter
);
2665 mdcache
->mds
->queue_waiters(t
);
2666 waiting_for_commit
.erase(it
);
2670 // try drop dentries in this dirfrag if it's about to be purged
2671 if (!inode
->is_base() && get_parent_dir()->inode
->is_stray() &&
2673 mdcache
->maybe_eval_stray(inode
, true);
2675 // unpin if we kicked the last waiter.
2677 waiting_for_commit
.empty())
2686 mds_rank_t
CDir::get_export_pin(bool inherit
) const
2688 mds_rank_t export_pin
= inode
->get_export_pin(inherit
);
2689 if (export_pin
== MDS_RANK_EPHEMERAL_DIST
)
2690 export_pin
= mdcache
->hash_into_rank_bucket(ino(), get_frag());
2691 else if (export_pin
== MDS_RANK_EPHEMERAL_RAND
)
2692 export_pin
= mdcache
->hash_into_rank_bucket(ino());
2696 bool CDir::is_exportable(mds_rank_t dest
) const
2698 mds_rank_t export_pin
= get_export_pin();
2699 if (export_pin
== dest
)
2701 if (export_pin
>= 0)
2706 void CDir::encode_export(bufferlist
& bl
)
2708 ENCODE_START(1, 1, bl
);
2709 ceph_assert(!is_projected());
2712 encode(dirty_old_rstat
, bl
);
2713 encode(committed_version
, bl
);
2716 encode(dir_rep
, bl
);
2719 encode(pop_auth_subtree
, bl
);
2721 encode(dir_rep_by
, bl
);
2722 encode(get_replicas(), bl
);
2724 get(PIN_TEMPEXPORTING
);
2728 void CDir::finish_export()
2730 state
&= MASK_STATE_EXPORT_KEPT
;
2731 pop_nested
.sub(pop_auth_subtree
);
2732 pop_auth_subtree_nested
.sub(pop_auth_subtree
);
2734 pop_auth_subtree
.zero();
2735 put(PIN_TEMPEXPORTING
);
2736 dirty_old_rstat
.clear();
2739 void CDir::decode_import(bufferlist::const_iterator
& blp
, LogSegment
*ls
)
2741 DECODE_START(1, blp
);
2744 auto _fnode
= allocate_fnode();
2745 decode(*_fnode
, blp
);
2746 reset_fnode(std::move(_fnode
));
2748 update_projected_version();
2750 decode(dirty_old_rstat
, blp
);
2751 decode(committed_version
, blp
);
2752 committing_version
= committed_version
;
2756 state
&= MASK_STATE_IMPORT_KEPT
;
2757 state_set(STATE_AUTH
| (s
& MASK_STATE_EXPORTED
));
2764 decode(dir_rep
, blp
);
2766 decode(pop_me
, blp
);
2767 decode(pop_auth_subtree
, blp
);
2768 pop_nested
.add(pop_auth_subtree
);
2769 pop_auth_subtree_nested
.add(pop_auth_subtree
);
2771 decode(dir_rep_by
, blp
);
2772 decode(get_replicas(), blp
);
2773 if (is_replicated()) get(PIN_REPLICATED
);
2775 replica_nonce
= 0; // no longer defined
2777 // did we import some dirty scatterlock data?
2778 if (dirty_old_rstat
.size() ||
2779 !(fnode
->rstat
== fnode
->accounted_rstat
)) {
2780 mdcache
->mds
->locker
->mark_updated_scatterlock(&inode
->nestlock
);
2781 ls
->dirty_dirfrag_nest
.push_back(&inode
->item_dirty_dirfrag_nest
);
2783 if (!(fnode
->fragstat
== fnode
->accounted_fragstat
)) {
2784 mdcache
->mds
->locker
->mark_updated_scatterlock(&inode
->filelock
);
2785 ls
->dirty_dirfrag_dir
.push_back(&inode
->item_dirty_dirfrag_dir
);
2787 if (is_dirty_dft()) {
2788 if (inode
->dirfragtreelock
.get_state() != LOCK_MIX
&&
2789 inode
->dirfragtreelock
.is_stable()) {
2790 // clear stale dirtydft
2791 state_clear(STATE_DIRTYDFT
);
2793 mdcache
->mds
->locker
->mark_updated_scatterlock(&inode
->dirfragtreelock
);
2794 ls
->dirty_dirfrag_dirfragtree
.push_back(&inode
->item_dirty_dirfrag_dirfragtree
);
2800 void CDir::abort_import()
2802 ceph_assert(is_auth());
2803 state_clear(CDir::STATE_AUTH
);
2805 clear_replica_map();
2806 set_replica_nonce(CDir::EXPORT_NONCE
);
2810 pop_nested
.sub(pop_auth_subtree
);
2811 pop_auth_subtree_nested
.sub(pop_auth_subtree
);
2813 pop_auth_subtree
.zero();
2816 void CDir::encode_dirstat(bufferlist
& bl
, const session_info_t
& info
, const DirStat
& ds
) {
2817 if (info
.has_feature(CEPHFS_FEATURE_REPLY_ENCODING
)) {
2818 ENCODE_START(1, 1, bl
);
2819 encode(ds
.frag
, bl
);
2820 encode(ds
.auth
, bl
);
2821 encode(ds
.dist
, bl
);
2825 encode(ds
.frag
, bl
);
2826 encode(ds
.auth
, bl
);
2827 encode(ds
.dist
, bl
);
2831 /********************************
2836 * if dir_auth.first == parent, auth is same as inode.
2837 * unless .second != unknown, in which case that sticks.
2839 mds_authority_t
CDir::authority() const
2841 if (is_subtree_root())
2844 return inode
->authority();
2847 /** is_subtree_root()
2848 * true if this is an auth delegation point.
2849 * that is, dir_auth != default (parent,unknown)
2851 * some key observations:
2853 * - any region bound will be an export, or frozen.
2855 * note that this DOES heed dir_auth.pending
2858 bool CDir::is_subtree_root()
2860 if (dir_auth == CDIR_AUTH_DEFAULT) {
2861 //dout(10) << "is_subtree_root false " << dir_auth << " != " << CDIR_AUTH_DEFAULT
2862 //<< " on " << ino() << dendl;
2865 //dout(10) << "is_subtree_root true " << dir_auth << " != " << CDIR_AUTH_DEFAULT
2866 //<< " on " << ino() << dendl;
2873 * true if we are x, or an ancestor of x
2875 bool CDir::contains(CDir
*x
)
2880 x
= x
->get_inode()->get_projected_parent_dir();
2886 bool CDir::can_rep() const
2891 unsigned mds_num
= mdcache
->mds
->get_mds_map()->get_num_mds(MDSMap::STATE_ACTIVE
);
2892 if ((mds_num
- 1) > get_replicas().size())
2901 void CDir::set_dir_auth(const mds_authority_t
&a
)
2903 dout(10) << "setting dir_auth=" << a
2904 << " from " << dir_auth
2905 << " on " << *this << dendl
;
2907 bool was_subtree
= is_subtree_root();
2908 bool was_ambiguous
= dir_auth
.second
>= 0;
2913 // new subtree root?
2914 if (!was_subtree
&& is_subtree_root()) {
2915 dout(10) << " new subtree root, adjusting auth_pins" << dendl
;
2917 if (freeze_tree_state
) {
2918 // only by CDir::_freeze_tree()
2919 ceph_assert(is_freezing_tree_root());
2922 inode
->num_subtree_roots
++;
2924 // unpin parent of frozen dir/tree?
2925 if (inode
->is_auth()) {
2926 ceph_assert(!is_frozen_tree_root());
2927 if (is_frozen_dir())
2928 inode
->auth_unpin(this);
2931 if (was_subtree
&& !is_subtree_root()) {
2932 dout(10) << " old subtree root, adjusting auth_pins" << dendl
;
2934 inode
->num_subtree_roots
--;
2936 // pin parent of frozen dir/tree?
2937 if (inode
->is_auth()) {
2938 ceph_assert(!is_frozen_tree_root());
2939 if (is_frozen_dir())
2940 inode
->auth_pin(this);
2944 // newly single auth?
2945 if (was_ambiguous
&& dir_auth
.second
== CDIR_AUTH_UNKNOWN
) {
2947 take_waiting(WAIT_SINGLEAUTH
, ls
);
2948 mdcache
->mds
->queue_waiters(ls
);
2952 /*****************************************
2953 * AUTH PINS and FREEZING
2955 * the basic plan is that auth_pins only exist in auth regions, and they
2956 * prevent a freeze (and subsequent auth change).
2958 * however, we also need to prevent a parent from freezing if a child is frozen.
2959 * for that reason, the parent inode of a frozen directory is auth_pinned.
2961 * the oddity is when the frozen directory is a subtree root. if that's the case,
2962 * the parent inode isn't frozen. which means that when subtree authority is adjusted
2963 * at the bounds, inodes for any frozen bound directories need to get auth_pins at that
2968 void CDir::auth_pin(void *by
)
2974 #ifdef MDS_AUTHPIN_SET
2975 auth_pin_set
.insert(by
);
2978 dout(10) << "auth_pin by " << by
<< " on " << *this << " count now " << auth_pins
<< dendl
;
2980 if (freeze_tree_state
)
2981 freeze_tree_state
->auth_pins
+= 1;
2984 void CDir::auth_unpin(void *by
)
2988 #ifdef MDS_AUTHPIN_SET
2990 auto it
= auth_pin_set
.find(by
);
2991 ceph_assert(it
!= auth_pin_set
.end());
2992 auth_pin_set
.erase(it
);
2998 dout(10) << "auth_unpin by " << by
<< " on " << *this << " count now " << auth_pins
<< dendl
;
2999 ceph_assert(auth_pins
>= 0);
3001 if (freeze_tree_state
)
3002 freeze_tree_state
->auth_pins
-= 1;
3004 maybe_finish_freeze(); // pending freeze?
3007 void CDir::adjust_nested_auth_pins(int dirinc
, void *by
)
3009 ceph_assert(dirinc
);
3010 dir_auth_pins
+= dirinc
;
3012 dout(15) << __func__
<< " " << dirinc
<< " on " << *this
3013 << " by " << by
<< " count now "
3014 << auth_pins
<< "/" << dir_auth_pins
<< dendl
;
3015 ceph_assert(dir_auth_pins
>= 0);
3017 if (freeze_tree_state
)
3018 freeze_tree_state
->auth_pins
+= dirinc
;
3021 maybe_finish_freeze(); // pending freeze?
3024 #ifdef MDS_VERIFY_FRAGSTAT
3025 void CDir::verify_fragstat()
3027 ceph_assert(is_complete());
3028 if (inode
->is_stray())
3032 memset(&c
, 0, sizeof(c
));
3034 for (auto it
= items
.begin();
3037 CDentry
*dn
= it
->second
;
3041 dout(10) << " " << *dn
<< dendl
;
3042 if (dn
->is_primary())
3043 dout(10) << " " << *dn
->inode
<< dendl
;
3045 if (dn
->is_primary()) {
3046 if (dn
->inode
->is_dir())
3051 if (dn
->is_remote()) {
3052 if (dn
->get_remote_d_type() == DT_DIR
)
3059 if (c
.nsubdirs
!= fnode
->fragstat
.nsubdirs
||
3060 c
.nfiles
!= fnode
->fragstat
.nfiles
) {
3061 dout(0) << "verify_fragstat failed " << fnode
->fragstat
<< " on " << *this << dendl
;
3062 dout(0) << " i count " << c
<< dendl
;
3065 dout(0) << "verify_fragstat ok " << fnode
->fragstat
<< " on " << *this << dendl
;
3070 /*****************************************************************************
3076 void CDir::_walk_tree(std::function
<bool(CDir
*)> callback
)
3079 dfq
.push_back(this);
3081 while (!dfq
.empty()) {
3082 CDir
*dir
= dfq
.front();
3085 for (auto& p
: *dir
) {
3086 CDentry
*dn
= p
.second
;
3087 if (!dn
->get_linkage()->is_primary())
3089 CInode
*in
= dn
->get_linkage()->get_inode();
3093 auto&& dfv
= in
->get_nested_dirfrags();
3094 for (auto& dir
: dfv
) {
3095 auto ret
= callback(dir
);
3103 bool CDir::freeze_tree()
3105 ceph_assert(!is_frozen());
3106 ceph_assert(!is_freezing());
3107 ceph_assert(!freeze_tree_state
);
3111 // Travese the subtree to mark dirfrags as 'freezing' (set freeze_tree_state)
3112 // and to accumulate auth pins and record total count in freeze_tree_state.
3113 // when auth unpin an 'freezing' object, the counter in freeze_tree_state also
3114 // gets decreased. Subtree become 'frozen' when the counter reaches zero.
3115 freeze_tree_state
= std::make_shared
<freeze_tree_state_t
>(this);
3116 freeze_tree_state
->auth_pins
+= get_auth_pins() + get_dir_auth_pins();
3117 if (!lock_caches_with_auth_pins
.empty())
3118 mdcache
->mds
->locker
->invalidate_lock_caches(this);
3120 _walk_tree([this](CDir
*dir
) {
3121 if (dir
->freeze_tree_state
)
3123 dir
->freeze_tree_state
= freeze_tree_state
;
3124 freeze_tree_state
->auth_pins
+= dir
->get_auth_pins() + dir
->get_dir_auth_pins();
3125 if (!dir
->lock_caches_with_auth_pins
.empty())
3126 mdcache
->mds
->locker
->invalidate_lock_caches(dir
);
3131 if (is_freezeable(true)) {
3136 state_set(STATE_FREEZINGTREE
);
3137 ++num_freezing_trees
;
3138 dout(10) << "freeze_tree waiting " << *this << dendl
;
3143 void CDir::_freeze_tree()
3145 dout(10) << __func__
<< " " << *this << dendl
;
3146 ceph_assert(is_freezeable(true));
3148 if (freeze_tree_state
) {
3149 ceph_assert(is_auth());
3151 ceph_assert(!is_auth());
3152 freeze_tree_state
= std::make_shared
<freeze_tree_state_t
>(this);
3154 freeze_tree_state
->frozen
= true;
3157 mds_authority_t auth
;
3158 bool was_subtree
= is_subtree_root();
3160 auth
= get_dir_auth();
3162 // temporarily prevent parent subtree from becoming frozen.
3163 inode
->auth_pin(this);
3164 // create new subtree
3168 _walk_tree([this, &auth
] (CDir
*dir
) {
3169 if (dir
->freeze_tree_state
!= freeze_tree_state
) {
3170 mdcache
->adjust_subtree_auth(dir
, auth
);
3177 ceph_assert(auth
.first
>= 0);
3178 ceph_assert(auth
.second
== CDIR_AUTH_UNKNOWN
);
3179 auth
.second
= auth
.first
;
3180 mdcache
->adjust_subtree_auth(this, auth
);
3182 inode
->auth_unpin(this);
3184 // importing subtree ?
3185 _walk_tree([this] (CDir
*dir
) {
3186 ceph_assert(!dir
->freeze_tree_state
);
3187 dir
->freeze_tree_state
= freeze_tree_state
;
3194 if (state_test(STATE_FREEZINGTREE
)) {
3195 state_clear(STATE_FREEZINGTREE
);
3196 --num_freezing_trees
;
3199 state_set(STATE_FROZENTREE
);
3204 void CDir::unfreeze_tree()
3206 dout(10) << __func__
<< " " << *this << dendl
;
3208 MDSContext::vec unfreeze_waiters
;
3209 take_waiting(WAIT_UNFREEZE
, unfreeze_waiters
);
3211 if (freeze_tree_state
) {
3212 _walk_tree([this, &unfreeze_waiters
](CDir
*dir
) {
3213 if (dir
->freeze_tree_state
!= freeze_tree_state
)
3215 dir
->freeze_tree_state
.reset();
3216 dir
->take_waiting(WAIT_UNFREEZE
, unfreeze_waiters
);
3222 if (state_test(STATE_FROZENTREE
)) {
3223 // frozen. unfreeze.
3224 state_clear(STATE_FROZENTREE
);
3231 ceph_assert(is_subtree_root());
3232 // for debug purpose, caller should ensure 'dir_auth.second == dir_auth.first'
3233 mds_authority_t auth
= get_dir_auth();
3234 ceph_assert(auth
.first
>= 0);
3235 ceph_assert(auth
.second
== auth
.first
);
3236 auth
.second
= CDIR_AUTH_UNKNOWN
;
3237 mdcache
->adjust_subtree_auth(this, auth
);
3239 freeze_tree_state
.reset();
3241 ceph_assert(state_test(STATE_FREEZINGTREE
));
3243 // freezing. stop it.
3244 state_clear(STATE_FREEZINGTREE
);
3245 --num_freezing_trees
;
3246 freeze_tree_state
.reset();
3248 finish_waiting(WAIT_FROZEN
, -1);
3252 mdcache
->mds
->queue_waiters(unfreeze_waiters
);
3255 void CDir::adjust_freeze_after_rename(CDir
*dir
)
3257 if (!freeze_tree_state
|| dir
->freeze_tree_state
!= freeze_tree_state
)
3259 CDir
*newdir
= dir
->get_inode()->get_parent_dir();
3260 if (newdir
== this || newdir
->freeze_tree_state
== freeze_tree_state
)
3263 ceph_assert(!freeze_tree_state
->frozen
);
3264 ceph_assert(get_dir_auth_pins() > 0);
3266 MDSContext::vec unfreeze_waiters
;
3268 auto unfreeze
= [this, &unfreeze_waiters
](CDir
*dir
) {
3269 if (dir
->freeze_tree_state
!= freeze_tree_state
)
3271 int dec
= dir
->get_auth_pins() + dir
->get_dir_auth_pins();
3272 // shouldn't become zero because srcdn of rename was auth pinned
3273 ceph_assert(freeze_tree_state
->auth_pins
> dec
);
3274 freeze_tree_state
->auth_pins
-= dec
;
3275 dir
->freeze_tree_state
.reset();
3276 dir
->take_waiting(WAIT_UNFREEZE
, unfreeze_waiters
);
3281 dir
->_walk_tree(unfreeze
);
3283 mdcache
->mds
->queue_waiters(unfreeze_waiters
);
3286 bool CDir::can_auth_pin(int *err_ret
) const
3291 } else if (is_freezing_dir() || is_frozen_dir()) {
3292 err
= ERR_FRAGMENTING_DIR
;
3294 auto p
= is_freezing_or_frozen_tree();
3295 if (p
.first
|| p
.second
) {
3296 err
= ERR_EXPORTING_TREE
;
3306 class C_Dir_AuthUnpin
: public CDirContext
{
3308 explicit C_Dir_AuthUnpin(CDir
*d
) : CDirContext(d
) {}
3309 void finish(int r
) override
{
3310 dir
->auth_unpin(dir
->get_inode());
3314 void CDir::maybe_finish_freeze()
3316 if (dir_auth_pins
!= 0)
3319 // we can freeze the _dir_ even with nested pins...
3320 if (state_test(STATE_FREEZINGDIR
)) {
3321 if (auth_pins
== 1) {
3324 finish_waiting(WAIT_FROZEN
);
3328 if (freeze_tree_state
) {
3329 if (freeze_tree_state
->frozen
||
3330 freeze_tree_state
->auth_pins
!= 1)
3333 if (freeze_tree_state
->dir
!= this) {
3334 freeze_tree_state
->dir
->maybe_finish_freeze();
3338 ceph_assert(state_test(STATE_FREEZINGTREE
));
3340 if (!is_subtree_root() && inode
->is_frozen()) {
3341 dout(10) << __func__
<< " !subtree root and frozen inode, waiting for unfreeze on " << inode
<< dendl
;
3342 // retake an auth_pin...
3344 // and release it when the parent inode unfreezes
3345 inode
->add_waiter(WAIT_UNFREEZE
, new C_Dir_AuthUnpin(this));
3351 finish_waiting(WAIT_FROZEN
);
3359 bool CDir::freeze_dir()
3361 ceph_assert(!is_frozen());
3362 ceph_assert(!is_freezing());
3365 if (is_freezeable_dir(true)) {
3370 state_set(STATE_FREEZINGDIR
);
3371 if (!lock_caches_with_auth_pins
.empty())
3372 mdcache
->mds
->locker
->invalidate_lock_caches(this);
3373 dout(10) << "freeze_dir + wait " << *this << dendl
;
3378 void CDir::_freeze_dir()
3380 dout(10) << __func__
<< " " << *this << dendl
;
3381 //assert(is_freezeable_dir(true));
3382 // not always true during split because the original fragment may have frozen a while
3383 // ago and we're just now getting around to breaking it up.
3385 state_clear(STATE_FREEZINGDIR
);
3386 state_set(STATE_FROZENDIR
);
3389 if (is_auth() && !is_subtree_root())
3390 inode
->auth_pin(this); // auth_pin for duration of freeze
3394 void CDir::unfreeze_dir()
3396 dout(10) << __func__
<< " " << *this << dendl
;
3398 if (state_test(STATE_FROZENDIR
)) {
3399 state_clear(STATE_FROZENDIR
);
3402 // unpin (may => FREEZEABLE) FIXME: is this order good?
3403 if (is_auth() && !is_subtree_root())
3404 inode
->auth_unpin(this);
3406 finish_waiting(WAIT_UNFREEZE
);
3408 finish_waiting(WAIT_FROZEN
, -1);
3410 // still freezing. stop.
3411 ceph_assert(state_test(STATE_FREEZINGDIR
));
3412 state_clear(STATE_FREEZINGDIR
);
3415 finish_waiting(WAIT_UNFREEZE
);
3419 void CDir::enable_frozen_inode()
3421 ceph_assert(frozen_inode_suppressed
> 0);
3422 if (--frozen_inode_suppressed
== 0) {
3423 for (auto p
= freezing_inodes
.begin(); !p
.end(); ) {
3426 ceph_assert(in
->is_freezing_inode());
3427 in
->maybe_finish_freeze_inode();
3433 * Slightly less complete than operator<<, because this is intended
3434 * for identifying a directory and its state rather than for dumping
3437 void CDir::dump(Formatter
*f
, int flags
) const
3439 ceph_assert(f
!= NULL
);
3440 if (flags
& DUMP_PATH
) {
3441 f
->dump_stream("path") << get_path();
3443 if (flags
& DUMP_DIRFRAG
) {
3444 f
->dump_stream("dirfrag") << dirfrag();
3446 if (flags
& DUMP_SNAPID_FIRST
) {
3447 f
->dump_int("snapid_first", first
);
3449 if (flags
& DUMP_VERSIONS
) {
3450 f
->dump_stream("projected_version") << get_projected_version();
3451 f
->dump_stream("version") << get_version();
3452 f
->dump_stream("committing_version") << get_committing_version();
3453 f
->dump_stream("committed_version") << get_committed_version();
3455 if (flags
& DUMP_REP
) {
3456 f
->dump_bool("is_rep", is_rep());
3458 if (flags
& DUMP_DIR_AUTH
) {
3459 if (get_dir_auth() != CDIR_AUTH_DEFAULT
) {
3460 if (get_dir_auth().second
== CDIR_AUTH_UNKNOWN
) {
3461 f
->dump_stream("dir_auth") << get_dir_auth().first
;
3463 f
->dump_stream("dir_auth") << get_dir_auth();
3466 f
->dump_string("dir_auth", "");
3469 if (flags
& DUMP_STATES
) {
3470 f
->open_array_section("states");
3471 MDSCacheObject::dump_states(f
);
3472 if (state_test(CDir::STATE_COMPLETE
)) f
->dump_string("state", "complete");
3473 if (state_test(CDir::STATE_FREEZINGTREE
)) f
->dump_string("state", "freezingtree");
3474 if (state_test(CDir::STATE_FROZENTREE
)) f
->dump_string("state", "frozentree");
3475 if (state_test(CDir::STATE_FROZENDIR
)) f
->dump_string("state", "frozendir");
3476 if (state_test(CDir::STATE_FREEZINGDIR
)) f
->dump_string("state", "freezingdir");
3477 if (state_test(CDir::STATE_EXPORTBOUND
)) f
->dump_string("state", "exportbound");
3478 if (state_test(CDir::STATE_IMPORTBOUND
)) f
->dump_string("state", "importbound");
3479 if (state_test(CDir::STATE_BADFRAG
)) f
->dump_string("state", "badfrag");
3482 if (flags
& DUMP_MDS_CACHE_OBJECT
) {
3483 MDSCacheObject::dump(f
);
3485 if (flags
& DUMP_ITEMS
) {
3486 f
->open_array_section("dentries");
3487 for (auto &p
: items
) {
3488 CDentry
*dn
= p
.second
;
3489 f
->open_object_section("dentry");
3497 void CDir::dump_load(Formatter
*f
)
3499 f
->dump_stream("path") << get_path();
3500 f
->dump_stream("dirfrag") << dirfrag();
3502 f
->open_object_section("pop_me");
3506 f
->open_object_section("pop_nested");
3510 f
->open_object_section("pop_auth_subtree");
3511 pop_auth_subtree
.dump(f
);
3514 f
->open_object_section("pop_auth_subtree_nested");
3515 pop_auth_subtree_nested
.dump(f
);
3519 /****** Scrub Stuff *******/
3521 void CDir::scrub_info_create() const
3523 ceph_assert(!scrub_infop
);
3525 // break out of const-land to set up implicit initial state
3526 CDir
*me
= const_cast<CDir
*>(this);
3527 const auto& pf
= me
->get_projected_fnode();
3529 std::unique_ptr
<scrub_info_t
> si(new scrub_info_t());
3531 si
->last_recursive
.version
= pf
->recursive_scrub_version
;
3532 si
->last_recursive
.time
= pf
->recursive_scrub_stamp
;
3534 si
->last_local
.version
= pf
->localized_scrub_version
;
3535 si
->last_local
.time
= pf
->localized_scrub_stamp
;
3537 me
->scrub_infop
.swap(si
);
3540 void CDir::scrub_initialize(const ScrubHeaderRef
& header
)
3542 ceph_assert(header
);
3543 // FIXME: weird implicit construction, is someone else meant
3544 // to be calling scrub_info_create first?
3546 scrub_infop
->directory_scrubbing
= true;
3547 scrub_infop
->header
= header
;
3548 header
->inc_num_pending();
3551 void CDir::scrub_aborted() {
3552 dout(20) << __func__
<< dendl
;
3553 ceph_assert(scrub_is_in_progress());
3555 scrub_infop
->last_scrub_dirty
= false;
3556 scrub_infop
->directory_scrubbing
= false;
3557 scrub_infop
->header
->dec_num_pending();
3558 scrub_infop
.reset();
3561 void CDir::scrub_finished()
3563 dout(20) << __func__
<< dendl
;
3564 ceph_assert(scrub_is_in_progress());
3566 scrub_infop
->last_local
.time
= ceph_clock_now();
3567 scrub_infop
->last_local
.version
= get_version();
3568 if (scrub_infop
->header
->get_recursive())
3569 scrub_infop
->last_recursive
= scrub_infop
->last_local
;
3571 scrub_infop
->last_scrub_dirty
= true;
3573 scrub_infop
->directory_scrubbing
= false;
3574 scrub_infop
->header
->dec_num_pending();
3577 void CDir::scrub_maybe_delete_info()
3580 !scrub_infop
->directory_scrubbing
&&
3581 !scrub_infop
->last_scrub_dirty
)
3582 scrub_infop
.reset();
3585 bool CDir::scrub_local()
3587 ceph_assert(is_complete());
3588 bool good
= check_rstats(true);
3589 if (!good
&& scrub_infop
->header
->get_repair()) {
3590 mdcache
->repair_dirfrag_stats(this);
3591 scrub_infop
->header
->set_repaired();
3596 std::string
CDir::get_path() const
3599 get_inode()->make_path_string(path
, true);
3603 bool CDir::should_split_fast() const
3605 // Max size a fragment can be before trigger fast splitting
3606 int fast_limit
= g_conf()->mds_bal_split_size
* g_conf()->mds_bal_fragment_fast_factor
;
3608 // Fast path: the sum of accounted size and null dentries does not
3609 // exceed threshold: we definitely are not over it.
3610 if (get_frag_size() + get_num_head_null() <= fast_limit
) {
3614 // Fast path: the accounted size of the frag exceeds threshold: we
3615 // definitely are over it
3616 if (get_frag_size() > fast_limit
) {
3620 int64_t effective_size
= 0;
3622 for (const auto &p
: items
) {
3623 const CDentry
*dn
= p
.second
;
3624 if (!dn
->get_projected_linkage()->is_null()) {
3629 return effective_size
> fast_limit
;
3632 bool CDir::should_merge() const
3634 if (get_frag() == frag_t())
3637 if (inode
->is_ephemeral_dist()) {
3638 unsigned min_frag_bits
= mdcache
->get_ephemeral_dist_frag_bits();
3639 if (min_frag_bits
> 0 && get_frag().bits() < min_frag_bits
+ 1)
3643 return (int)get_frag_size() < g_conf()->mds_bal_merge_size
;
3646 MEMPOOL_DEFINE_OBJECT_FACTORY(CDir
, co_dir
, mds_co
);
3647 MEMPOOL_DEFINE_OBJECT_FACTORY(CDir::scrub_info_t
, scrub_info_t
, mds_co
)