#include "BatchOp.h"
#include "MDSCacheObject.h"
#include "MDSContext.h"
+#include "Mutation.h"
#include "SimpleLock.h"
#include "LocalLockC.h"
#include "ScrubHeader.h"
static const int STATE_EVALUATINGSTRAY = (1<<4);
static const int STATE_PURGINGPINNED = (1<<5);
static const int STATE_BOTTOMLRU = (1<<6);
+ static const int STATE_UNLINKING = (1<<7);
// stray dentry needs notification of releasing reference
static const int STATE_STRAY = STATE_NOTIFYREF;
static const int MASK_STATE_IMPORT_KEPT = STATE_BOTTOMLRU;
// -- pins --
- static const int PIN_INODEPIN = 1; // linked inode is pinned
- static const int PIN_FRAGMENTING = -2; // containing dir is refragmenting
- static const int PIN_PURGING = 3;
- static const int PIN_SCRUBPARENT = 4;
+ static const int PIN_INODEPIN = 1; // linked inode is pinned
+ static const int PIN_FRAGMENTING = -2; // containing dir is refragmenting
+ static const int PIN_PURGING = 3;
+ static const int PIN_SCRUBPARENT = 4;
+ static const int PIN_WAITUNLINKSTATE = 5;
static const unsigned EXPORT_NONCE = 1;
+ const static uint64_t WAIT_UNLINK_STATE = (1<<0);
+ const static uint64_t WAIT_UNLINK_FINISH = (1<<1);
+ uint32_t replica_unlinking_ref = 0;
CDentry(std::string_view n, __u32 h,
mempool::mds_co::string alternate_name,
case PIN_FRAGMENTING: return "fragmenting";
case PIN_PURGING: return "purging";
case PIN_SCRUBPARENT: return "scrubparent";
+ case PIN_WAITUNLINKSTATE: return "waitunlinkstate";
default: return generic_pin_name(p);
}
}
case MSG_MDS_DENTRYUNLINK:
handle_dentry_unlink(ref_cast<MDentryUnlink>(m));
break;
+ case MSG_MDS_DENTRYUNLINK_ACK:
+ handle_dentry_unlink_ack(ref_cast<MDentryUnlinkAck>(m));
+ break;
+
case MSG_MDS_FRAGMENTNOTIFY:
handle_fragment_notify(ref_cast<MMDSFragmentNotify>(m));
// UNLINK
-void MDCache::send_dentry_unlink(CDentry *dn, CDentry *straydn, MDRequestRef& mdr)
+void MDCache::send_dentry_unlink(CDentry *dn, CDentry *straydn,
+ MDRequestRef& mdr, bool unlinking)
{
dout(10) << __func__ << " " << *dn << dendl;
// share unlink news with replicas
CInode *strayin = straydn->get_linkage()->get_inode();
strayin->encode_snap_blob(snapbl);
}
+
+ if (unlinking) {
+ ceph_assert(!straydn);
+ dn->replica_unlinking_ref = 0;
+ }
for (set<mds_rank_t>::iterator it = replicas.begin();
it != replicas.end();
++it) {
rejoin_gather.count(*it)))
continue;
- auto unlink = make_message<MDentryUnlink>(dn->get_dir()->dirfrag(), dn->get_name());
+ auto unlink = make_message<MDentryUnlink>(dn->get_dir()->dirfrag(),
+ dn->get_name(), unlinking);
if (straydn) {
encode_replica_stray(straydn, *it, unlink->straybl);
unlink->snapbl = snapbl;
}
mds->send_message_mds(unlink, *it);
+ if (unlinking) {
+ dn->replica_unlinking_ref++;
+ dn->get(CDentry::PIN_WAITUNLINKSTATE);
+ }
+ }
+
+ if (unlinking && dn->replica_unlinking_ref) {
+ dn->add_waiter(CDentry::WAIT_UNLINK_STATE, new C_MDS_RetryRequest(this, mdr));
}
}
// straydn
CDentry *straydn = nullptr;
CInode *strayin = nullptr;
+
if (m->straybl.length())
decode_replica_stray(straydn, &strayin, m->straybl, mds_rank_t(m->get_source().num()));
+ boost::intrusive_ptr<MDentryUnlinkAck> ack;
+ CDentry::linkage_t *dnl;
+ CDentry *dn;
+ CInode *in;
+ bool hadrealm;
+
CDir *dir = get_dirfrag(m->get_dirfrag());
if (!dir) {
dout(7) << __func__ << " don't have dirfrag " << m->get_dirfrag() << dendl;
+ if (m->is_unlinking())
+ goto ack;
} else {
- CDentry *dn = dir->lookup(m->get_dn());
+ dn = dir->lookup(m->get_dn());
if (!dn) {
dout(7) << __func__ << " don't have dentry " << *dir << " dn " << m->get_dn() << dendl;
+ if (m->is_unlinking())
+ goto ack;
} else {
dout(7) << __func__ << " on " << *dn << dendl;
- CDentry::linkage_t *dnl = dn->get_linkage();
+
+ if (m->is_unlinking()) {
+ dn->state_set(CDentry::STATE_UNLINKING);
+ goto ack;
+ }
+
+ dnl = dn->get_linkage();
// open inode?
if (dnl->is_primary()) {
- CInode *in = dnl->get_inode();
+ in = dnl->get_inode();
dn->dir->unlink_inode(dn);
ceph_assert(straydn);
straydn->dir->link_primary_inode(straydn, in);
in->first = straydn->first;
// update subtree map?
- if (in->is_dir())
+ if (in->is_dir()) {
adjust_subtree_after_rename(in, dir, false);
+ }
if (m->snapbl.length()) {
- bool hadrealm = (in->snaprealm ? true : false);
+ hadrealm = (in->snaprealm ? true : false);
in->decode_snap_blob(m->snapbl);
ceph_assert(in->snaprealm);
if (!hadrealm)
if (in->is_any_caps() &&
!in->state_test(CInode::STATE_EXPORTINGCAPS))
migrator->export_caps(in);
-
+
straydn = NULL;
} else {
ceph_assert(!straydn);
dn->dir->unlink_inode(dn);
}
ceph_assert(dnl->is_null());
+ dn->state_clear(CDentry::STATE_UNLINKING);
}
}
trim_dentry(straydn, ex);
send_expire_messages(ex);
}
+ return;
+
+ack:
+ ack = make_message<MDentryUnlinkAck>(m->get_dirfrag(), m->get_dn());
+ mds->send_message(ack, m->get_connection());
}
+void MDCache::handle_dentry_unlink_ack(const cref_t<MDentryUnlinkAck> &m)
+{
+ CDir *dir = get_dirfrag(m->get_dirfrag());
+ if (!dir) {
+ dout(7) << __func__ << " don't have dirfrag " << m->get_dirfrag() << dendl;
+ } else {
+ CDentry *dn = dir->lookup(m->get_dn());
+ if (!dn) {
+ dout(7) << __func__ << " don't have dentry " << *dir << " dn " << m->get_dn() << dendl;
+ } else {
+ dout(7) << __func__ << " on " << *dn << " ref "
+ << dn->replica_unlinking_ref << " -> "
+ << dn->replica_unlinking_ref - 1 << dendl;
+ dn->replica_unlinking_ref--;
+ if (!dn->replica_unlinking_ref) {
+ MDSContext::vec finished;
+ dn->take_waiting(CDentry::WAIT_UNLINK_STATE, finished);
+ mds->queue_waiters(finished);
+ }
+ dn->put(CDentry::PIN_WAITUNLINKSTATE);
+ }
+ }
+}
void encode_remote_dentry_link(CDentry::linkage_t *dnl, bufferlist& bl);
void decode_remote_dentry_link(CDir *dir, CDentry *dn, bufferlist::const_iterator& p);
void send_dentry_link(CDentry *dn, MDRequestRef& mdr);
- void send_dentry_unlink(CDentry *dn, CDentry *straydn, MDRequestRef& mdr);
+ void send_dentry_unlink(CDentry *dn, CDentry *straydn, MDRequestRef& mdr, bool unlinking=false);
void wait_for_uncommitted_fragment(dirfrag_t dirfrag, MDSContext *c) {
uncommitted_fragments.at(dirfrag).waiters.push_back(c);
void handle_discover_reply(const cref_t<MDiscoverReply> &m);
void handle_dentry_link(const cref_t<MDentryLink> &m);
void handle_dentry_unlink(const cref_t<MDentryUnlink> &m);
+ void handle_dentry_unlink_ack(const cref_t<MDentryUnlinkAck> &m);
int dump_cache(std::string_view fn, Formatter *f);
mdr->pin(dn);
early_reply(mdr, in, dn);
-
+
mdr->committing = true;
submit_mdlog_entry(le, fin, mdr, __func__);
-
+
if (mdr->client_request && mdr->client_request->is_queued_for_replay()) {
if (mds->queue_one_replay()) {
dout(10) << " queued next replay op" << dendl;
} else {
dout(10) << " journaled last replay op" << dendl;
}
- } else if (mdr->did_early_reply)
+ } else if (mdr->did_early_reply) {
mds->locker->drop_rdlocks_for_early_reply(mdr.get());
- else
+ if (dn && dn->is_waiter_for(CDentry::WAIT_UNLINK_FINISH))
+ mdlog->flush();
+ } else {
mdlog->flush();
+ }
}
void Server::submit_mdlog_entry(LogEvent *le, MDSLogContextBase *fin, MDRequestRef& mdr,
if (!dn)
return;
+ if (is_unlink_pending(dn)) {
+ wait_for_pending_unlink(dn, mdr);
+ return;
+ }
+
CDentry::linkage_t *dnl = dn->get_projected_linkage();
if (!excl && !dnl->is_null()) {
// it existed.
// ------------------------------------------------
+struct C_WaitUnlinkToFinish : public MDSContext {
+protected:
+ MDCache *mdcache;
+ CDentry *dn;
+ MDSContext *fin;
+
+ MDSRank *get_mds() override
+ {
+ ceph_assert(mdcache != NULL);
+ return mdcache->mds;
+ }
+
+public:
+ C_WaitUnlinkToFinish(MDCache *m, CDentry *d, MDSContext *f) :
+ mdcache(m), dn(d), fin(f) {}
+ void finish(int r) override {
+ fin->complete(r);
+ dn->put(CDentry::PIN_PURGING);
+ }
+};
+
+bool Server::is_unlink_pending(CDentry *dn)
+{
+ CDentry::linkage_t *dnl = dn->get_projected_linkage();
+ if (!dnl->is_null() && dn->state_test(CDentry::STATE_UNLINKING)) {
+ return true;
+ }
+ return false;
+}
+
+void Server::wait_for_pending_unlink(CDentry *dn, MDRequestRef& mdr)
+{
+ mds->locker->drop_locks(mdr.get());
+ auto fin = new C_MDS_RetryRequest(mdcache, mdr);
+ dn->get(CDentry::PIN_PURGING);
+ dn->add_waiter(CDentry::WAIT_UNLINK_FINISH, new C_WaitUnlinkToFinish(mdcache, dn, fin));
+}
+
// MKNOD
class C_MDS_mknod_finish : public ServerLogContext {
if (!dn)
return;
+ if (is_unlink_pending(dn)) {
+ wait_for_pending_unlink(dn, mdr);
+ return;
+ }
+
CDir *dir = dn->get_dir();
CInode *diri = dir->get_inode();
if (!check_access(mdr, diri, MAY_WRITE))
if (!dn)
return;
+ if (is_unlink_pending(dn)) {
+ wait_for_pending_unlink(dn, mdr);
+ return;
+ }
+
CDir *dir = dn->get_dir();
CInode *diri = dir->get_inode();
if (!dn)
return;
+ if (is_unlink_pending(dn)) {
+ wait_for_pending_unlink(dn, mdr);
+ return;
+ }
+
CDir *dir = dn->get_dir();
CInode *diri = dir->get_inode();
targeti = ret.second->get_projected_linkage()->get_inode();
}
+ if (is_unlink_pending(destdn)) {
+ wait_for_pending_unlink(destdn, mdr);
+ return;
+ }
+
ceph_assert(destdn->get_projected_linkage()->is_null());
if (req->get_alternate_name().size() > alternate_name_max) {
dout(10) << " alternate_name longer than " << alternate_name_max << dendl;
mdr->apply();
MDRequestRef null_ref;
- if (inc)
+ if (inc) {
mdcache->send_dentry_link(dn, null_ref);
- else
+ } else {
+ dn->state_clear(CDentry::STATE_UNLINKING);
mdcache->send_dentry_unlink(dn, NULL, null_ref);
-
+
+ MDSContext::vec finished;
+ dn->take_waiting(CDentry::WAIT_UNLINK_FINISH, finished);
+ mdcache->mds->queue_waiters(finished);
+ }
+
// bump target popularity
mds->balancer->hit_inode(targeti, META_POP_IWR);
mds->balancer->hit_dir(dn->get_dir(), META_POP_IWR);
if (rmdir)
mdr->disable_lock_cache();
+
CDentry *dn = rdlock_path_xlock_dentry(mdr, false, true);
if (!dn)
return;
+ // notify replica MDSes the dentry is under unlink
+ if (!dn->state_test(CDentry::STATE_UNLINKING)) {
+ dn->state_set(CDentry::STATE_UNLINKING);
+ mdcache->send_dentry_unlink(dn, nullptr, mdr, true);
+ if (dn->replica_unlinking_ref) {
+ return;
+ }
+ }
+
CDentry::linkage_t *dnl = dn->get_linkage(client, mdr);
ceph_assert(!dnl->is_null());
CInode *in = dnl->get_inode();
}
mdr->apply();
-
+
+ dn->state_clear(CDentry::STATE_UNLINKING);
mdcache->send_dentry_unlink(dn, straydn, mdr);
-
+
+ MDSContext::vec finished;
+ dn->take_waiting(CDentry::WAIT_UNLINK_FINISH, finished);
+ mdcache->mds->queue_waiters(finished);
+
if (straydn) {
// update subtree map?
if (strayin->is_dir())
// reply
respond_to_request(mdr, 0);
-
+
// removing a new dn?
dn->get_dir()->try_remove_unlinked_dn(dn);
if (!destdn)
return;
+ if (is_unlink_pending(destdn)) {
+ wait_for_pending_unlink(destdn, mdr);
+ return;
+ }
+
+ if (is_unlink_pending(srcdn)) {
+ wait_for_pending_unlink(srcdn, mdr);
+ return;
+ }
+
dout(10) << " destdn " << *destdn << dendl;
CDir *destdir = destdn->get_dir();
ceph_assert(destdir->is_auth());
void handle_client_fsync(MDRequestRef& mdr);
+ bool is_unlink_pending(CDentry *dn);
+ void wait_for_pending_unlink(CDentry *dn, MDRequestRef& mdr);
+
// open
void handle_client_open(MDRequestRef& mdr);
void handle_client_openc(MDRequestRef& mdr); // O_CREAT variant.
class MDentryUnlink final : public MMDSOp {
private:
- static constexpr int HEAD_VERSION = 1;
+ static constexpr int HEAD_VERSION = 2;
static constexpr int COMPAT_VERSION = 1;
-
+
dirfrag_t dirfrag;
std::string dn;
+ bool unlinking = false;
public:
dirfrag_t get_dirfrag() const { return dirfrag; }
const std::string& get_dn() const { return dn; }
+ bool is_unlinking() const { return unlinking; }
ceph::buffer::list straybl;
ceph::buffer::list snapbl;
protected:
MDentryUnlink() :
MMDSOp(MSG_MDS_DENTRYUNLINK, HEAD_VERSION, COMPAT_VERSION) { }
- MDentryUnlink(dirfrag_t df, std::string_view n) :
+ MDentryUnlink(dirfrag_t df, std::string_view n, bool u=false) :
MMDSOp(MSG_MDS_DENTRYUNLINK, HEAD_VERSION, COMPAT_VERSION),
- dirfrag(df),
- dn(n) {}
+ dirfrag(df), dn(n), unlinking(u) {}
~MDentryUnlink() final {}
public:
void print(std::ostream& o) const override {
o << "dentry_unlink(" << dirfrag << " " << dn << ")";
}
-
+
void decode_payload() override {
using ceph::decode;
auto p = payload.cbegin();
decode(dirfrag, p);
decode(dn, p);
decode(straybl, p);
+ if (header.version >= 2)
+ decode(unlinking, p);
}
void encode_payload(uint64_t features) override {
using ceph::encode;
encode(dirfrag, payload);
encode(dn, payload);
encode(straybl, payload);
+ encode(unlinking, payload);
+ }
+private:
+ template<class T, typename... Args>
+ friend boost::intrusive_ptr<T> ceph::make_message(Args&&... args);
+};
+
+class MDentryUnlinkAck final : public MMDSOp {
+private:
+ static constexpr int HEAD_VERSION = 1;
+ static constexpr int COMPAT_VERSION = 1;
+
+ dirfrag_t dirfrag;
+ std::string dn;
+
+ public:
+ dirfrag_t get_dirfrag() const { return dirfrag; }
+ const std::string& get_dn() const { return dn; }
+
+protected:
+ MDentryUnlinkAck() :
+ MMDSOp(MSG_MDS_DENTRYUNLINK_ACK, HEAD_VERSION, COMPAT_VERSION) { }
+ MDentryUnlinkAck(dirfrag_t df, std::string_view n) :
+ MMDSOp(MSG_MDS_DENTRYUNLINK_ACK, HEAD_VERSION, COMPAT_VERSION),
+ dirfrag(df), dn(n) {}
+ ~MDentryUnlinkAck() final {}
+
+public:
+ std::string_view get_type_name() const override { return "dentry_unlink_ack";}
+ void print(std::ostream& o) const override {
+ o << "dentry_unlink_ack(" << dirfrag << " " << dn << ")";
+ }
+
+ void decode_payload() override {
+ using ceph::decode;
+ auto p = payload.cbegin();
+ decode(dirfrag, p);
+ decode(dn, p);
+ }
+ void encode_payload(uint64_t features) override {
+ using ceph::encode;
+ encode(dirfrag, payload);
+ encode(dn, payload);
}
private:
template<class T, typename... Args>
break;
+ case MSG_MDS_DENTRYUNLINK_ACK:
+ m = make_message<MDentryUnlinkAck>();
+ break;
case MSG_MDS_DENTRYUNLINK:
m = make_message<MDentryUnlink>();
break;
#define MSG_MDS_OPENINOREPLY 0x210
#define MSG_MDS_SNAPUPDATE 0x211
#define MSG_MDS_FRAGMENTNOTIFYACK 0x212
+#define MSG_MDS_DENTRYUNLINK_ACK 0x213
#define MSG_MDS_LOCK 0x300 // 0x3xx are for locker of mds
#define MSG_MDS_INODEFILECAPS 0x301
projects / ceph.git / commitdiff