RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2024-12-15 16:07:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

*** empty log message ***

Browse Source 
git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@201 29311d96-e01e-0410-9327-a35deaab8ce9
This commit is contained in:
sage 2005-05-04 05:51:39 +00:00
parent ae62cbb6f7
commit fb43a5401c
7 changed files with 189 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
ceph/mds/MDCache.cc
View File

@ -3610,12 +3610,13 @@ public:
}
};
void MDCache::dentry_xlock_request(CDir *dir, string& dname, Message *req, Context *onfinish)
void MDCache::dentry_xlock_request(CDir *dir, string& dname, bool create,
Message *req, Context *onfinish)
{
dout(10) << "dentry_xlock_request on dn " << dname << " in " << *dir << endl;
dout(10) << "dentry_xlock_request on dn " << dname << " create=" << create << " in " << *dir << endl;
// send request
int dauth = dir->dentry_authority(dname);
MLock *m = new MLock(LOCK_AC_REQXLOCK, dauth);
MLock *m = new MLock(create ? LOCK_AC_REQXLOCKC:LOCK_AC_REQXLOCK, dauth);
m->set_dn(dir->ino(), dname);
mds->messenger->send_message(m,
MSG_ADDR_MDS(dauth), MDS_PORT_CACHE,
@ -3672,7 +3673,6 @@ void MDCache::handle_lock_dn(MLock *m)
}
dn = dir->lookup(dname);
assert(dn);
}
// except with.. an xlock request?
@ -3793,6 +3793,7 @@ void MDCache::handle_lock_dn(MLock *m)
case LOCK_AC_REQXLOCKACK:
case LOCK_AC_REQXLOCKNAK:
{
dout(10) << "handle_lock_dn got ack/nak on a reqxlock for " << *dn << endl;
list<Context*> finished;
dir->take_waiting(CDIR_WAIT_DNREQXLOCK, m->get_dn(), finished);
finish_contexts(finished,
@ -3835,14 +3836,15 @@ void MDCache::handle_lock_dn(MLock *m)
dn = dir->add_dentry(dname);
}
if (!dentry_xlock_start(dn, m, dir->inode, true)) {
// hose null dn if we're waiting on something
if (dn->is_clean() && dn->is_null() && dn->is_sync()) dir->remove_dentry(dn);
return; // waiting for xlock
}
{
// ACK
if (dn->xlockedby != m) {
if (!dentry_xlock_start(dn, m, dir->inode, true)) {
// hose null dn if we're waiting on something
if (dn->is_clean() && dn->is_null() && dn->is_sync()) dir->remove_dentry(dn);
return; // waiting for xlock
}
} else {
// locked by me! (well, bc of request.)
// ACK xlock request
MLock *reply = new MLock(LOCK_AC_REQXLOCKACK, mds->get_nodeid());
reply->set_dn(dir->ino(), dname);
mds->messenger->send_message(reply,

2
ceph/mds/MDCache.h
View File

@ -342,7 +342,7 @@ class MDCache {
bool allnodes=false);
void dentry_xlock_finish(CDentry *dn, bool quiet=false);
void handle_lock_dn(MLock *m);
void dentry_xlock_request(CDir *dir, string& dname,
void dentry_xlock_request(CDir *dir, string& dname, bool create,
Message *req, Context *onfinish);

43
ceph/mds/MDS.cc
View File

@ -316,15 +316,19 @@ void MDS::dispatch(Message *m)
balancer->send_heartbeat();
}
*/
if (whoami == 0 &&
stat_ops == 100) {
dout(1) << "FORCING EXPORTS" << endl;
if (whoami == 0 && false) {
static bool didit = false;
// 7 to 1
CInode *in = mdcache->get_inode(7457);
mdcache->export_dir(in->dir,1);
// 6 to 2
in = mdcache->get_inode(6259);
mdcache->export_dir(in->dir,2);
CInode *in = mdcache->get_inode(2);
if (in && !didit) {
CDir *dir = in->get_or_open_dir(this);
if (dir->is_auth()) {
dout(1) << "FORCING EXPORT" << endl;
mdcache->export_dir(dir,1);
didit = true;
}
}
}
// shut down?
@ -499,10 +503,12 @@ void MDS::dispatch_request(Message *m, CInode *ref)
switch (m->get_type()) {
case MSG_CLIENT_REQUEST:
req = (MClientRequest*)m;
break;
break; // continue below!
case MSG_MDS_LOCK:
mdcache->handle_lock_dn((MLock*)m);
break;
return; // done
default:
assert(0); // shouldn't get here
}
@ -1506,14 +1512,14 @@ void MDS::handle_client_rename_2(MClientRequest *req,
if (destauth != get_nodeid()) {
dout(7) << "rename has remote dest " << destauth << endl;
dout(7) << "FOREIGN RENAME" << endl;
reply_request(req, -EINVAL);
return;
//return;
} else {
dout(7) << "rename is local" << endl;
handle_client_rename_local(req, ref,
srcpath, srcdiri, srcdn,
destpath.get_path(), destdir, destdn, destname);
}
handle_client_rename_local(req, ref,
srcpath, srcdiri, srcdn,
destpath.get_path(), destdir, destdn, destname);
return;
}
@ -1576,7 +1582,8 @@ void MDS::handle_client_rename_local(MClientRequest *req,
dout(7) << "handle_client_rename_local: dest local=" << destlocal << " dn dne yet" << endl;
}
// pick lock ordering
/* lock source and dest dentries, in lexicographic order.
*/
bool dosrc = srcpath < destpath;
for (int i=0; i<2; i++) {
if (dosrc) {
@ -1588,7 +1595,7 @@ void MDS::handle_client_rename_local(MClientRequest *req,
return;
} else {
if (!srcdn || srcdn->xlockedby != req) {
mdcache->dentry_xlock_request(srcdn->dir, srcdn->name, req, new C_MDS_RetryRequest(this, req, ref));
mdcache->dentry_xlock_request(srcdn->dir, srcdn->name, false, req, new C_MDS_RetryRequest(this, req, ref));
return;
}
}
@ -1606,7 +1613,7 @@ void MDS::handle_client_rename_local(MClientRequest *req,
}
} else {
if (!destdn || destdn->xlockedby != req) {
mdcache->dentry_xlock_request(destdir, destname, req, new C_MDS_RetryRequest(this, req, ref));
mdcache->dentry_xlock_request(destdir, destname, true, req, new C_MDS_RetryRequest(this, req, ref));
return;
}
}

58
ceph/messages/MRename.h Normal file
View File

@ -0,0 +1,58 @@
#ifndef __MRENAME_H
#define __MRENAME_H
class MRename : public Message {
inodeno_t srcdirino;
string srcname;
inodeno_t destdirino;
string destname;
crope inode_state;
public:
inodeno_t get_srcdirino() { return srcdirino; }
string& get_srcname() { return srcname; }
inodeno_t get_destdirino() { return destdirino; }
string& get_destname() { return destname; }
MRename() {}
MRename(inodeno_t srcdirino,
const string& srcname,
inodeno_t destdirino,
const string& destname,
crope& inode_state) :
Message(MSG_MDS_RENAME) {
this->srcdirino = srcdirino;
this->srcname = srcname;
this->destdirino = destdirino;
this->destname = destname;
this->inode_state = inode_state;
}
virtual char *get_type_name() { return "Rnot";}
crope& get_inode_state() { return inode_state; }
virtual void decode_payload(crope& s) {
int off = 0;
s.copy(off, sizeof(srcdirino), (char*)&srcdirino);
off += sizeof(srcdirino);
s.copy(off, sizeof(destdirino), (char*)&destdirino);
off += sizeof(destdirino);
srcname = s.c_str() + off;
off += srcname.length() + 1;
destname = s.c_str() + off;
off += destname.length() + 1;
inode_state = s.substr(off, s.length()-off);
}
virtual void encode_payload(crope& s) {
s.append((char*)&srcdirino,sizeof(srcdirino));
s.append((char*)&destdirino,sizeof(destdirino));
s.append((char*)srcname.c_str());
s.append((char)0);
s.append((char*)destname.c_str());
s.append((char)0);
s.append(inode_state);
}
};
#endif

37
ceph/messages/MRenameAck.h Normal file
View File

@ -0,0 +1,37 @@
#ifndef __MRENAMEACK_H
#define __MRENAMEACK_H
class MRename : public Message {
inodeno_t srcdirino;
string srcname;
inodeno_t destdirino;
string destname;
public:
inodeno_t get_srcdirino() { return srcdirino; }
string& get_srcname() { return srcname; }
MRenameAck() {}
MRenameAck(inodeno_t srcdirino,
const string& srcname) :
Message(MSG_MDS_RENAMEACK) {
this->srcdirino = srcdirino;
this->srcname = srcname;
}
virtual char *get_type_name() { return "RnotA";}
virtual void decode_payload(crope& s) {
int off = 0;
s.copy(off, sizeof(srcdirino), (char*)&srcdirino);
off += sizeof(srcdirino);
srcname = s.c_str() + off;
off += srcname.length() + 1;
}
virtual void encode_payload(crope& s) {
s.append((char*)&srcdirino,sizeof(srcdirino));
s.append((char*)srcname.c_str());
s.append((char)0);
}
};
#endif

50
ceph/messages/MRenameReq.h Normal file
View File

@ -0,0 +1,50 @@
#ifndef __MRENAMEREQ_H
#define __MRENAMEREQ_H
class MRenameReq : public Message {
inodeno_t srcdirino;
string srcname;
inodeno_t destdirino;
string destname;
public:
inodeno_t get_srcdirino() { return srcdirino; }
string& get_srcname() { return srcname; }
inodeno_t get_destdirino() { return destdirino; }
string& get_destname() { return destname; }
MRenameReq() {}
MRenameReq(inodeno_t srcdirino,
const string& srcname,
inodeno_t destdirino,
const string& destname) :
Message(MSG_MDS_RENAMEREQ) {
this->srcdirino = srcdirino;
this->srcname = srcname;
this->destdirino = destdirino;
this->destname = destname;
}
virtual char *get_type_name() { return "Rnot";}
virtual void decode_payload(crope& s) {
int off = 0;
s.copy(off, sizeof(srcdirino), (char*)&srcdirino);
off += sizeof(srcdirino);
s.copy(off, sizeof(destdirino), (char*)&destdirino);
off += sizeof(destdirino);
srcname = s.c_str() + off;
off += srcname.length() + 1;
destname = s.c_str() + off;
off += destname.length() + 1;
}
virtual void encode_payload(crope& s) {
s.append((char*)&srcdirino,sizeof(srcdirino));
s.append((char*)&destdirino,sizeof(destdirino));
s.append((char*)srcname.c_str());
s.append((char)0);
s.append((char*)destname.c_str());
s.append((char)0);
}
};
#endif

5
ceph/msg/Message.h
View File

@ -51,7 +51,10 @@
#define MSG_MDS_DENTRYUNLINK 200
#define MSG_MDS_RENAMENOTIFY 300
#define MSG_MDS_RENAMENOTIFY 300 // sent to bystanders
#define MSG_MDS_RENAMEREQ 301 // sent from initiator to src auth (rare)
#define MSG_MDS_RENAME 302 // sent from src to dest, includes inode
#define MSG_MDS_RENAMEACK 303 // sent from dest to src, to xlock_finish
#define MSG_MDS_LOCK 500