RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2025-03-06 16:28:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

*** empty log message ***

Browse Source 
git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@128 29311d96-e01e-0410-9327-a35deaab8ce9
This commit is contained in:
sage 2005-02-11 06:08:46 +00:00
parent 50103faaa8
commit 9dfe61b489
13 changed files with 510 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
ceph/client/ClNode.h
View File

@ -27,6 +27,7 @@ class ClNode : public LRUObject {
ClNode() {
parent = 0;
refs = 0;
isdir = havedircontents = false;
dangling = false;
}
@ -38,11 +39,16 @@ class ClNode : public LRUObject {
return 1 + parent->depth();
}
int refs; // reference count
void get() {
refs++;
lru_expireable = false;
}
void put() {
lru_expireable = true;
refs--;
assert(refs >= 0);
if (!refs) lru_expireable = true;
}
void detach() {

24
ceph/client/Client.cc
View File

@ -76,7 +76,6 @@ void Client::dispatch(Message *m)
// basic stuff
case MSG_CLIENT_REPLY:
dout(9) << "got reply" << endl;
assim_reply((MClientReply*)m);
if (tid < max_requests)
@ -187,10 +186,11 @@ void Client::done() {
void Client::assim_reply(MClientReply *r)
{
if (r->get_result() < 0) {
dout(12) << "error" << endl;
if (r->get_result() != 0) {
dout(12) << "error " << r->get_result() << endl;
return;
}
dout(10) << "success" << endl;
// closed a file?
@ -363,6 +363,7 @@ void Client::trim_cache()
break; // out of things to expire!
dout(12) << "expiring inode " << i->ino << endl;
assert(i->refs == 0);
detach_node(i);
remove_node(i);
expired++;
@ -427,6 +428,7 @@ void Client::issue_request()
cwd->full_path(p,last_req_dn);
}
string arg;
if (!op) {
int r = rand() % 100;
op = MDS_OP_STAT;
@ -461,6 +463,17 @@ void Client::issue_request()
else if (!g_conf.client_deterministic &&
r < 35 && !cwd->isdir)
op = MDS_OP_UNLINK;
else if (!g_conf.client_deterministic &&
r < 100 && !cwd->isdir) {
op = MDS_OP_RENAME;
char s[100];
sprintf(s,"rename.%d", rand() % 50);
string ss = s;
filepath p1 = p;
filepath p2 = p1.subpath(p1.length() - 1);
p2.add_dentry(ss);
arg = p2.get_path();
}
else if (false && !g_conf.client_deterministic &&
r < 37 && cwd->isdir) {
op = MDS_OP_MKDIR;
@ -484,7 +497,7 @@ void Client::issue_request()
}
}
send_request(p, op); // root, if !cwd
send_request(p, op, arg); // root, if !cwd
}
@ -520,12 +533,13 @@ void Client::close_a_file()
0);
}
void Client::send_request(string& path, int op)
void Client::send_request(string& path, int op, string& arg)
{
MClientRequest *req = new MClientRequest(tid++, op, whoami);
req->set_ino(1);
req->set_path(path);
req->set_arg(arg);
// direct it
int mds = 0;

6
ceph/client/Client.h
View File

@ -50,8 +50,8 @@ class Client : public Dispatcher {
return it->second;
}
void add_node(ClNode *n) {
assert(node_map.count(n->ino) == 0);
node_map.insert(pair<inodeno_t,ClNode*>(n->ino, n));
if (node_map.count(n->ino) == 0)
node_map.insert(pair<inodeno_t,ClNode*>(n->ino, n));
}
void remove_node(ClNode *n) {
if (cwd == n) cwd = root;
@ -71,7 +71,7 @@ class Client : public Dispatcher {
virtual void assim_reply(MClientReply*);
virtual void issue_request();
virtual void send_request(string& p, int op);
virtual void send_request(string& p, int op, string& arg);
void close_a_file();
bool is_open(ClNode *n);
bool is_sync(ClNode *n);

4
ceph/config.cc
View File

@ -33,7 +33,7 @@ md_config_t g_conf = {
fake_clock: true,
fakemessenger_serialize: true,
debug: 10,
debug: 13,
mdcache_size: MDS_CACHE_SIZE,
mdcache_mid: .8,
@ -44,7 +44,7 @@ md_config_t g_conf = {
mdbal_replicate_threshold: 500,
mdbal_unreplicate_threshold: 200,
mds_heartbeat_op_interval: 200,
mds_heartbeat_op_interval: 20000,
mds_verify_export_dirauth: true
};

9
ceph/mds/CInode.h
View File

@ -126,6 +126,10 @@ static char *cinode_pin_names[CINODE_NUM_PINS] = {
// waiters: handle_export_dir_warning
// triggers: handle_export_dir_notify
#define CINODE_WAIT_RENAME 32768
// waiters: file_rename
// triggers: file_rename_finish
#define CINODE_WAIT_ANY 0xffffff
@ -138,6 +142,9 @@ static char *cinode_pin_names[CINODE_NUM_PINS] = {
#define CINODE_STATE_PROXY 32 // can't expire yet
#define CINODE_STATE_EXPORTING 64 // on nonauth bystander.
#define CINODE_STATE_RENAMING 128 // moving me
#define CINODE_STATE_RENAMINGTO 256 // rename target (will be unlinked)
// misc
#define CINODE_EXPORT_NONCE 1 // nonce given to replicas created by export
@ -199,9 +206,11 @@ class CInode : LRUObject {
bool sync_replicawantback; // avoids sticky sync
set<int> unlink_waiting_for_ack;
set<int> rename_waiting_for_ack;
int dangling_auth; // explicit auth when dangling.
// waiters
multimap<int,Context*> waiting;

214
ceph/mds/MDCache.cc
View File

@ -57,6 +57,8 @@
#include "messages/MDirSyncRelease.h"
//#include "messages/MDirSyncRecall.h"
#include "messages/MRenameLocalFile.h"
#include "InoAllocator.h"
#include <assert.h>
@ -703,7 +705,7 @@ int MDCache::path_traverse(filepath& path,
// don't have it.
int dauth = cur->dir->dentry_authority( path[depth] );
dout(12) << " dentry " << path[depth] << " dauth is " << dauth << endl;
dout(12) << " miss dentry " << path[depth] << " dauth is " << dauth << endl;
if (dauth == whoami) {
@ -781,6 +783,86 @@ int MDCache::path_traverse(filepath& path,
// rename crap
void MDCache::rename_file(CInode *from,
CDir *destdir,
string name)
{
assert(destdir->lookup(name) == 0);
// unlink, re-link
unlink_inode(from);
link_inode( destdir, name, from );
}
void MDCache::rename_dir(CInode *from,
CDir *destdir,
string name)
{
CDir *old_parent_dir = from->get_parent_dir();
assert(destdir->lookup(name) == 0);
// unlink, re-link
unlink_inode(from);
link_inode( destdir, name, from );
// fix nested exports?
CDir *containing_import = get_containing_import( destdir );
if (from->dir->is_import()) {
// all my exports now belong to new containing_import
dout(7) << "moving exports of renamed import " << *from << " to " << *containing_import << endl;
for (pair<multimap<CDir*,CDir*>::iterator, multimap<CDir*,CDir*>::iterator> p =
nested_exports.equal_range(from->dir);
p.first != p.second;
p.first++) {
CDir *nested = (*p.first).second;
dout(7) << " moving " << *nested << endl;
nested_exports.insert(pair<CDir*,CDir*>(containing_import, nested));
}
// i'm not an import anymore
nested_exports.erase( from->dir );
imports.erase( from->dir );
from->dir->state_clear(CDIR_STATE_IMPORT);
from->dir->put(CDIR_PIN_IMPORT);
} else {
// check for exports under me
CDir *old_containing = get_containing_import( old_parent_dir );
for (pair<multimap<CDir*,CDir*>::iterator, multimap<CDir*,CDir*>::iterator> p =
nested_exports.equal_range(old_containing);
p.first != p.second;
) {
CDir *nested = (*p.first).second;
// NOTE/FIXME/WARNING: possible concurrent modification weirdness!
// trace back to import, or dir
CDir *cur = nested->get_parent_dir();
while (!cur->is_import()) {
if (cur == from->dir) {
dout(7) << " moving " << *nested << endl;
nested_exports.insert(pair<CDir*,CDir*>(containing_import, nested));
nested_exports.erase( p.first++ ); // tricky
break;
} else {
cur = cur->get_parent_dir();
p.first++;
}
}
}
}
show_imports();
}
// REPLICAS
@ -919,7 +1001,7 @@ void MDCache::handle_discover(MDiscover *dis)
if (cur->dir->is_complete()) {
// set error flag in reply
dout(7) << "mds" << whoami << " dentry " << dis->get_dentry(i) << " not found in " << *cur->dir << ", returning error" << endl;
reply->set_flag_error();
reply->set_flag_error( dis->get_dentry(i) );
break;
} else {
delete reply;
@ -992,6 +1074,21 @@ void MDCache::handle_discover_reply(MDiscoverReply *m)
dout(7) << "discover_reply " << *cur << " + " << m->get_path() << ", have " << m->get_num_inodes() << " inodes" << endl;
}
if (m->is_flag_error()) {
// error!
assert(cur->is_dir());
if (cur->dir) {
dout(7) << " flag_error on dentry " << m->get_error_dentry() << ", triggering dentry?" << endl;
cur->dir->take_waiting(CDIR_WAIT_DENTRY,
m->get_error_dentry(),
finished);
} else {
dout(7) << " flag_error on dentry " << m->get_error_dentry() << ", triggering dir?" << endl;
cur->take_waiting(CINODE_WAIT_DIR, finished);
}
}
// start this loop even if we have no inodes, but just the base_dir
int plus_root_dir = (m->get_num_inodes() == 1 && m->has_root()) ? 1:0; //ugly, sorry, see handle_discover
@ -1454,6 +1551,117 @@ void MDCache::handle_inode_unlink_ack(MInodeUnlinkAck *m)
// renaming!
class C_MDC_RenameLocalFile : public Context {
MDCache *mdc;
CInode *from, *oldin;
CDir *destdir;
Context *c;
public:
C_MDC_RenameLocalFile(MDCache *mdc,
CInode *from,
CDir *destdir,
CInode *oldin,
Context *c) {
this->mdc = mdc;
this->from = from;
this->destdir = destdir;
this->oldin = oldin;
this->c = c;
}
void finish(int r) {
mdc->file_rename_finish(from, destdir, oldin, c);
}
};
void MDCache::file_rename(CInode *from, CDir *destdir, string& name, CInode *oldin, Context *c)
{
assert(from->is_auth());
assert(!from->is_cached_by_anyone() || from->is_lockbyme());
assert(!oldin || !oldin->is_cached_by_anyone() || oldin->is_lockbyme());
// log it
from->mark_unsafe();
// mds->mdlog->submit_entry(new EInodeUnlink(in, in->get_parent_dir()),
// new C_MDC_InodeLog(this,in));
// mark dirty
from->mark_dirty();
// rename locally
if (oldin) {
unlink_inode(oldin);
oldin->mark_clean();
oldin->state_set(CINODE_STATE_DANGLING);
}
rename_file(from, destdir, name);
// tell replicas?
from->rename_waiting_for_ack = from->cached_by;
if (oldin) {
for (set<int>::iterator it = oldin->cached_by_begin();
it != oldin->cached_by_end();
it++)
from->rename_waiting_for_ack.insert( *it );
}
if (!from->rename_waiting_for_ack.empty()) {
from->state_set(CINODE_STATE_RENAMING);
oldin->state_set(CINODE_STATE_RENAMINGTO);
// send notify
for (set<int>::iterator it = from->rename_waiting_for_ack.begin();
it != from->rename_waiting_for_ack.end();
it++) {
dout(7) << "rename_file sending rename to " << *it << endl;
mds->messenger->send_message(new MRenameLocalFile(from->ino(),
destdir->ino(),
name,
oldin ? oldin->ino():0),
MSG_ADDR_MDS(*it), MDS_PORT_CACHE, MDS_PORT_CACHE);
}
// waiter
from->add_waiter(CINODE_WAIT_RENAME,
new C_MDC_RenameLocalFile(this, from, destdir, oldin, c));
}
else {
// done
file_rename_finish(from, destdir, oldin, c);
}
}
void MDCache::file_rename_finish(CInode *from, CDir *destdir, CInode *oldin, Context *c)
{
dout(7) << "file_rename_finish on " << *from << endl;
// drop locks?
from->state_clear(CINODE_STATE_RENAMING);
write_hard_finish(from);
if (oldin) {
write_hard_finish(oldin);
oldin->state_clear(CINODE_STATE_RENAMINGTO);
}
// finish
c->finish(0);
delete c;
}
// locks ----------------------------------------------------------------
@ -1765,9 +1973,11 @@ void MDCache::inode_sync_start(CInode *in)
}
// sync clients
int last = -1;
for (multiset<int>::iterator it = in->get_open_write().begin();
it != in->get_open_write().end();
it++) {
if (*it == last) continue; last = *it; // only 1 per client (even if open multiple times)
in->sync_waiting_for_ack.insert(MSG_ADDR_CLIENT(*it));
mds->messenger->send_message(new MInodeSyncStart(in->ino(), mds->get_nodeid()),
MSG_ADDR_CLIENT(*it), 0,

13
ceph/mds/MDCache.h
View File

@ -45,6 +45,7 @@ class MInodeLockRelease;
class MDirSyncStart;
class MDirSyncAck;
class MDirSyncRelease;
class MRenameLocalFile;
class C_MDS_ExportFinish;
class InoAllocator;
@ -155,6 +156,14 @@ class MDCache {
Message *req,
int onfail);
void rename_file(CInode *from,
CDir *destdir,
string name);
void rename_dir(CInode *from,
CDir *destdir,
string name);
// == messages ==
int proc_message(Message *m);
@ -174,6 +183,10 @@ class MDCache {
void handle_inode_unlink(MInodeUnlink *m);
void handle_inode_unlink_ack(MInodeUnlinkAck *m);
void file_rename(CInode *in, CDir *destdir, string& name, CInode *oldin, Context *c);
void file_rename_finish(CInode *in, CDir *destdir, CInode *oldin, Context *c);
// -- misc auth --
int ino_proxy_auth(inodeno_t ino,

279
ceph/mds/MDS.cc
View File

@ -322,6 +322,17 @@ void MDS::handle_ping(MPing *m)
}
void MDS::reply_request(MClientRequest *req, int r)
{
// send error
messenger->send_message(new MClientReply(req, r),
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
// discard request
delete req;
}
int MDS::handle_client_request(MClientRequest *req)
{
dout(10) << "req " << *req << endl;
@ -365,14 +376,8 @@ int MDS::handle_client_request(MClientRequest *req)
r == -ENOTDIR ||
r == -EISDIR) {
// error!
dout(10) << "error, replying" << endl;
MClientReply *reply = new MClientReply(req, r);
messenger->send_message(reply,
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
// discard request
delete req;
dout(10) << " path traverse error " << r << ", replying" << endl;
reply_request(req, r);
return 0;
}
@ -829,11 +834,7 @@ void MDS::handle_client_openwrc(MClientRequest *req)
// send error response
dout(7) << "handle_client_openwrc error " << r << " replying to client" << endl;
MClientReply *reply = new MClientReply(req, r);
messenger->send_message(reply,
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
delete req;
reply_request(req, r);
return;
}
@ -879,9 +880,7 @@ void MDS::handle_client_unlink(MClientRequest *req,
// regular files only
if (in->is_dir()) {
dout(7) << "handle_client_unlink on dir " << *in << ", returning error" << endl;
messenger->send_message(new MClientReply(req, -EISDIR),
MSG_ADDR_CLIENT(req->get_client()), 0, MDS_PORT_SERVER);
delete req;
reply_request(req, -EISDIR);
return;
}
@ -942,13 +941,7 @@ void MDS::handle_client_mkdir(MClientRequest *req)
if (r > 0) return; // forwarded
if (r < 0) {
dout(10) << "error, replying" << endl;
MClientReply *reply = new MClientReply(req, r);
messenger->send_message(reply,
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
// discard request
delete req;
reply_request(req, r);
return;
}
@ -962,13 +955,7 @@ void MDS::handle_client_mkdir(MClientRequest *req)
if (dir->lookup(name) != 0) {
// name already exists
dout(10) << "name " << name << " exists in " << *dir << endl;
MClientReply *reply = new MClientReply(req, -EEXIST);
messenger->send_message(reply,
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
delete req;
reply_request(req, -EEXIST);
return;
}
@ -986,15 +973,14 @@ void MDS::handle_client_mkdir(MClientRequest *req)
new C_MDS_RetryMessage(this, req));
// reply
MClientReply *reply = new MClientReply(req);
messenger->send_message(reply,
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
delete req;
reply_request(req, 0);
return;
}
/*
@ -1009,86 +995,138 @@ void MDS::handle_client_rename(MClientRequest *req,
// forward
dout(10) << "handle_client_rename " << *cur << " not auth, forwarding" << endl;
int auth = cur->authority();
assert(auth != mdcluster->get_nodeid());
assert(auth != get_nodeid());
messenger->send_message(req,
MSG_ADDR_MDS(auth), MDS_PORT_SERVER,
MDS_PORT_SERVER);
return;
}
dout(10) << "handle_client_rename " << *cur << " to " << cur->get_arg() << endl;
dout(10) << "handle_client_rename " << *cur << " to " << req->get_arg() << endl;
// find the destination.
// discover, etc. on the way.. get just it on the local node.
filepath dirpath = cur->get_arg();
// ** FIXME? relative destination path???
filepath destpath = req->get_arg();
// FIXME? relative destination path???
vector<CInode*> trace;
int r = mdcache->path_traverse(dirpath, trace, req, MDS_TRAVERSE_DISCOVER);
int r = mdcache->path_traverse(destpath, trace, req, MDS_TRAVERSE_DISCOVER);
dout(12) << " r = " << r << " trace size " << trace.size() << " destpath depth " << destpath.depth() << endl;
string name;
CDir *destdir = 0;
// what is the dest? (dir or file or complete filename)
if (trace.size() == dirpath.depth()) {
// note: trace includes root, destpath doesn't (include leading /)
if (trace.size() == destpath.depth()+1) {
CInode *d = trace[trace.size()-1];
if (d->is_dir()) {
// mv some/thing /to/some/dir
destdir = d->get_or_open_dir(this); // /to/some/dir
name = cur->get_file_path().last(); // thing
name = req->get_filepath().last_bit(); // thing
} else {
// mv some/thing /to/some/existing_filename
destdir = trace[trace.size()-2]->get_or_open_dir(); // /to/some
name = dirpath.last(); // existing_filename
destdir = trace[trace.size()-2]->get_or_open_dir(this); // /to/some
name = destpath.last_bit(); // existing_filename
}
}
else if (trace.size() == dirpath.depth() - 1) {
else if (trace.size() == destpath.depth()) {
CInode *d = trace[trace.size()-1];
if (d->is_dir()) {
// mv some/thing /to/some/place_that_dne
destdir = d->get_or_open_dir(this); // /to/some
name = cur->get_file_path().last(); // place_that_dne
name = destpath.last_bit(); // place_that_dne
}
}
else {
assert(trace.size() < dirpath.depth()-1);
assert(trace.size() < destpath.depth());
// check traverse return value
if (r > 0) return; // discover, readdir, etc.
assert(r < 0); // musta been an error
assert(r < 0 || trace.size() <= 1); // musta been an error
// error out
dout(7) << "rename " << *cur << " dest " << dirpath << " dne" << end;
MClientReply *reply = new MClientReply(req, -ENODEST);
messenger->send_message(reply,
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
delete req;
dout(7) << " rename dest " << destpath << " dne" << endl;
reply_request(req, -EINVAL);
return;
}
// local or remote?
assert(cur->is_auth());
int dauth = destdir->dentry_authority(name);
if (dauth != mdcluster->get_nodeid()) {
if (dauth != get_nodeid()) {
dout(7) << "rename has remote dest " << dauth << endl;
// implement me
assert(0);
return 0;
//assert(0);
// *** IMPLEMENT ME ***
dout(7) << "rename " << *cur << " NOT IMPLEMENTED" << endl;
reply_request(req, -EINVAL);
return;
}
// file or dir?
if (cur->is_dir()) {
handle_client_rename_dir(req,cur);
//handle_client_rename_dir(req,cur);
reply_request(req, -EINVAL); // *** IMPLEMENT ME ****
} else {
handle_client_rename_file(req,cur);
handle_client_rename_file(req, cur, destdir, name);
}
}
/*
locking
if (!locked && flag=renaming)
(maybe) if (!locked && flag=renamingto)
basic protocol with replicas:
> Lock (possibly x2?)
< LockAck (possibly x2?)
> Rename
src ino
dst dir
either dst ino (is unlinked)
or dst name
< RenameAck
(implicitly unlocks, unlinks, etc.)
*/
class C_MDS_RenameFinish : public Context{
MDS *mds;
MClientRequest *req;
public:
C_MDS_RenameFinish(MDS *mds, MClientRequest *req) {
this->mds = mds;
this->req = req;
}
virtual void finish(int r) {
MClientReply *reply = new MClientReply(req, r);
// include trace?
mds->messenger->send_message(reply,
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
delete req;
}
};
void MDS::handle_client_rename_file(MClientRequest *req,
CInode *from,
CDir *destdir,
string name)
string& name)
{
bool must_wait_for_lock = false;
// does destination exist? (is this an overwrite?)
CDentry *dn = destdir->lookup(name);
CInode *oldin = 0;
@ -1098,76 +1136,107 @@ void MDS::handle_client_rename_file(MClientRequest *req,
if (oldin->is_dir()) {
// fail!
dout(7) << "dest exists and is dir" << endl;
MClientReply *reply = new MClientReply(req, -EISDIR);
messenger->send_message(reply,
MSG_ADDR_CLIENT(req->get_client()), 0,
MDS_PORT_SERVER);
delete req;
reply_request(req, -EISDIR);
return;
}
dout(7) << "dest " << name << " exists " << *oldin << endl;
// lock dest?
if (!oldin->test_state(CINODE_STATE_RENAMINGTO)) {
if (write_hard_start(oldin, req) == false) {
// wait
dout(7) << "dest/overwrite " << *oldin << " locking, waiting" << endl;
if (!oldin->state_test(CINODE_STATE_RENAMINGTO)) {
// can't be presync/lock
if (oldin->is_presync() || oldin->is_prelock()) {
dout(7) << "dest/overwrite " << *oldin << " presync/prelock, waiting" << endl;
oldin->add_waiter(oldin->is_presync() ? CINODE_WAIT_SYNC:CINODE_WAIT_LOCK,
new C_MDS_RetryMessage(this, req));
return;
}
oldin->set_state(CINODE_STATE_RENAMINGTO); // only lock once!
cout << "locked destinode " << *oldin << endl;
if (mdcache->write_hard_start(oldin, req) == false) {
// wait
dout(7) << "dest/overwrite " << *oldin << " locking, waiting" << endl;
must_wait_for_lock = true;
} else {
oldin->state_set(CINODE_STATE_RENAMINGTO); // only lock once!
cout << "locked destinode " << *oldin << endl;
}
} else {
cout << "already locked destinode " << *oldin << endl;
if (oldin->is_prelock())
must_wait_for_lock = true;
else
cout << "already locked destinode " << *oldin << endl;
}
}
// lock source?
if (!from->test_state(CINODE_STATE_RENAMING)) {
if (write_hard_state(from, req) == false) {
// wait
dout(7) << "from " << *from << " locking, waiting" << endl;
from->add_waiter(CINODE_WAIT_LOCK,
new C_MDS_RetryMessage(this, req));
if (!from->state_test(CINODE_STATE_RENAMING)) {
// can't be presync/lock
if (from->is_presync() || from->is_prelock()) {
dout(7) << "dest/overwrite " << *from << " presync/prelock, waiting" << endl;
from->add_waiter(from->is_presync() ? CINODE_WAIT_SYNC:CINODE_WAIT_LOCK,
new C_MDS_RetryMessage(this, req));
return;
}
from->set_state(CINODE_STATE_RENAMING); // only lock once
cout << "locked srcinode " << *from << endl;
} else {
cout << "already locked srcinode " << *from << endl;
}
// ...
if (from->is_cached_by_anyone()) {
} else {
// src not replicated
if (oldin && oldin->is_cached_by_anyone()) {
// dst replicated
if (mdcache->write_hard_start(from, req) == false) {
// wait
dout(7) << "from " << *from << " locking, waiting" << endl;
must_wait_for_lock = true;
} else {
// dst not replicated, or dne
// local rename, easy!
cout << "local rename" << endl;
if (oldin) {
// unlink old guy first
inode_unlink( oldin, NULL ); // waiter unnec; this is imm
// bc oldin not replicated
// *FIXME use waiter, combine w/ dst replicated case ******
}
mdcache->rename_file(from, destdir, name);
from->state_set(CINODE_STATE_RENAMING); // only lock once
cout << "locked srcinode " << *from << endl;
}
} else {
if (from->is_prelock())
must_wait_for_lock = true;
else
cout << "already locked srcinode " << *from << endl;
}
if (must_wait_for_lock)
return;
// ok go!
mdcache->file_rename(from, destdir, name, oldin,
new C_MDS_RenameFinish(this, req));
}
/*
void MDS::handle_client_rename_dir(MClientRequest *req,
CInode *from,
CDir *destdir,
string name)
{
}
// ...
// make sure dest isn't nested under src
CDir *td = destdir;
while (td) {
if (from->dir == td) {
// error out
dout(7) << " dest " << *destdir << " nested under src " << *from << endl;
reply_request(req, -EINVAL);
return;
}
td = td->get_parent_dir();
}
// ..
// ok do it
mdcache->rename_dir(from, destdir, name);
}
*/
// WRITE ME

7
ceph/mds/MDS.h
View File

@ -54,6 +54,7 @@ using namespace std;
class MDCluster;
class CInode;
class CDir;
class MDCache;
class MDStore;
class MDLog;
@ -146,6 +147,8 @@ class MDS : public Dispatcher {
bool open_root(Context *c);
bool open_root_2(int result, Context *c);
void reply_request(MClientRequest *req, int r = 0);
void handle_ping(class MPing *m);
void handle_client_done(Message *m);
@ -187,8 +190,8 @@ class MDS : public Dispatcher {
void handle_client_rmdir(MClientRequest *req, CInode *cur);
void handle_client_rename(MClientRequest *req, CInode *cur);
void handle_client_rename_mvlocal(MClientRequest *req, CInode *cur);
void handle_client_rename_mvremote(MClientRequest *req, CInode *cur);
void handle_client_rename_file(MClientRequest *req, CInode *cur, CDir *destdir, string& name);
void handle_client_rename_dir(MClientRequest *req, CInode *cur, CDir *destdir, string& name);
int do_stat(MClientRequest *m);

4
ceph/messages/MClientRequest.h
View File

@ -30,6 +30,7 @@ class MClientRequest : public Message {
void set_path(string& p) { path.set_path(p); }
void set_ino(inodeno_t ino) { st.ino = ino; }
void set_arg(string& arg) { this->arg = arg; }
long get_tid() { return st.tid; }
int get_op() { return st.op; }
@ -37,12 +38,13 @@ class MClientRequest : public Message {
inodeno_t get_ino() { return st.ino; }
string& get_path() { return path.get_path(); }
filepath& get_filepath() { return path; }
string& get_arg() { return arg; }
virtual int decode_payload(crope s) {
s.copy(0, sizeof(st), (char*)&st);
path.set_path( s.c_str() + sizeof(st) );
int off = sizeof(st) + path.length() + 0;
int off = sizeof(st) + path.length() + 1;
arg = s.c_str() + off;
return 0;
}

8
ceph/messages/MDiscoverReply.h
View File

@ -22,6 +22,7 @@ class MDiscoverReply : public Message {
bool no_base_dentry;
bool flag_forward;
bool flag_error;
string error_dentry; // dentry that was not found (to trigger waiters on asker)
// ... + dir + dentry + inode
// inode [ + ... ], base_ino = 0 : discover base_ino=0, start w/ root ino
@ -49,6 +50,7 @@ class MDiscoverReply : public Message {
bool is_flag_forward() { return flag_forward; }
bool is_flag_error() { return flag_error; }
string& get_error_dentry() { return error_dentry; }
// these index _arguments_ are aligned to the inodes.
CDirDiscover& get_dir(int n) { return *(dirs[n - no_base_dir]); }
@ -96,7 +98,7 @@ class MDiscoverReply : public Message {
}
void set_flag_forward() { flag_forward = true; }
void set_flag_error() { flag_error = true; }
void set_flag_error(string& dn) { flag_error = true; error_dentry = dn; }
// ...
@ -112,6 +114,8 @@ class MDiscoverReply : public Message {
off += sizeof(bool);
r.copy(off, sizeof(bool), (char*)&flag_error);
off += sizeof(bool);
error_dentry = r.c_str() + off;
off += error_dentry.length() + 1;
// dirs
int n;
@ -145,6 +149,8 @@ class MDiscoverReply : public Message {
r.append((char*)&no_base_dentry, sizeof(bool));
r.append((char*)&flag_forward, sizeof(bool));
r.append((char*)&flag_forward, sizeof(bool));
r.append((char*)error_dentry.c_str());
r.append((char)0);
// dirs
int n = dirs.size();

55
ceph/messages/MRenameLocalFile.h Normal file
View File

@ -0,0 +1,55 @@
#ifndef __MRENAMELOCALFILE_H
#define __MRENAMELOCALFILE_H
class MRenameLocalFile : public Message {
inodeno_t fromino;
// and (
inodeno_t destdirino;
string name;
// or
inodeno_t oldino;
// )
public:
inodeno_t get_fromino() { return fromino; }
inodeno_t get_destdirino() { return destdirino; }
inodeno_t get_oldino() { return oldino; }
string& get_name() { return name; }
MRenameLocalFile() {}
MRenameLocalFile(inodeno_t fromino,
inodeno_t destdirino,
string name,
inodeno_t oldino) :
Message(MSG_MDS_RENAMELOCALFILE) {
this->fromino = fromino;
this->destdirino = destdirino;
this->name = name;
this->oldino = oldino;
}
virtual char *get_type_name() { return "Rlf";}
virtual int decode_payload(crope s) {
int off = 0;
s.copy(off, sizeof(fromino), (char*)&fromino);
off += sizeof(fromino);
s.copy(off, sizeof(destdirino), (char*)&destdirino);
off += sizeof(destdirino);
s.copy(off, sizeof(oldino), (char*)&oldino);
off += sizeof(oldino);
name = s.c_str() + off;
off += name.length() + 1;
return off;
}
virtual crope get_payload() {
crope s;
s.append((char*)&fromino,sizeof(fromino));
s.append((char*)&destdirino,sizeof(destdirino));
s.append((char*)&oldino,sizeof(oldino));
s.append((char*)name.c_str());
s.append((char)0);
return s;
}
};
#endif

1
ceph/msg/Message.h
View File

@ -64,6 +64,7 @@
#define MSG_MDS_INODEUNLINK 200
#define MSG_MDS_INODEUNLINKACK 201
#define MSG_MDS_RENAMELOCALFILE 300
#define MSG_MDS_SHUTDOWNSTART 900
#define MSG_MDS_SHUTDOWNFINISH 901