some bugfixes

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@123 29311d96-e01e-0410-9327-a35deaab8ce9

ceph/TODO

@@ -51,19 +51,21 @@ DOCUMENT
 
 
 TEST
-- test hash/unhash operation
-- hash+export: encode list of replicated dir inodes so they can be discovered before import is procesed.
-- test nauthitems (wrt hashing?)
+- hashing
+ - test hash/unhash operation
+ - hash+export: encode list of replicated dir inodes so they can be discovered before import is procesed.
+ - test nauthitems (wrt hashing?)
 
 
 IMPLEMENT
 
-- hash + unhash!
 - rmdir
 - mkdir
 - rename
 - truncate
 
+- hash + unhash!
+
 - dir sync
  - stat of a dir should return dir size????
  - readdir of hashed dir

ceph/client/Client.cc

@@ -461,6 +461,24 @@ void Client::issue_request()
 	  else if (!g_conf.client_deterministic &&
 			   r < 35 && !cwd->isdir)
 		op = MDS_OP_UNLINK;
+	  else if (false && !g_conf.client_deterministic &&
+			   r < 37 && cwd->isdir) {
+		op = MDS_OP_MKDIR;
+		string dn = "client_dir.";
+		char num[10];
+		sprintf(num,"%d",rand()%100);
+		dn += num;
+
+		dout(10) << "trying mkdir on " << p << " / " << dn << endl;
+		if (cwd->lookup(dn))
+		  op = MDS_OP_STAT; // nevermind, exists
+		else {
+		  // do it
+		  p += "/";
+		  p += dn;
+		  last_req_dn.push_back(dn);
+		}
+	  }
 	  else if (r < 41 + open_files.size() && open_files.size() > 0)
 		return close_a_file();  // close file
 	}

ceph/mds/MDCache.cc

@@ -2798,6 +2798,7 @@ void MDCache::export_dir_go(CDir *dir,
 	dir->put(CDIR_PIN_IMPORT);                  // unpin, no longer an import
 	
 	// discard nested exports (that we're handing off
+	// NOTE: possible concurrent modification bug?
 	pair<multimap<CDir*,CDir*>::iterator, multimap<CDir*,CDir*>::iterator> p =
 	  nested_exports.equal_range(dir);
 	while (p.first != p.second) {
@@ -2820,6 +2821,7 @@ void MDCache::export_dir_go(CDir *dir,
 	dir->get(CDIR_PIN_EXPORT);                  // i must keep it pinned
 	
 	// discard nested exports (that we're handing off)
+	// NOTE: possible concurrent modification bug?
 	pair<multimap<CDir*,CDir*>::iterator, multimap<CDir*,CDir*>::iterator> p =
 	  nested_exports.equal_range(containing_import);
 	while (p.first != p.second) {
@@ -3396,14 +3398,20 @@ void MDCache::handle_export_dir(MExportDir *m)
       mds->logger->inc("immyex");
 
       // move nested exports under containing_import
+	  list<CDir*> to_move;
       for (pair<multimap<CDir*,CDir*>::iterator, multimap<CDir*,CDir*>::iterator> p =
              nested_exports.equal_range(ex);
            p.first != p.second;
            p.first++) {
         CDir *nested = (*p.first).second;
-        dout(7) << "     moving nested export " << nested << " under " << containing_import << endl;
-        nested_exports.insert(pair<CDir*,CDir*>(containing_import, nested));
+        dout(7) << "     moving nested export " << *nested << " under " << *containing_import << endl;
+		to_move.push_back(nested);
       }
+	  for (list<CDir*>::iterator it = to_move.begin();
+		   it != to_move.end();
+		   it++) {
+        nested_exports.insert(pair<CDir*,CDir*>(containing_import, *it));
+	  }
 
       // de-list under old import
       nested_exports.erase(ex);	

ceph/mds/MDS.cc

@@ -349,6 +349,10 @@ int MDS::handle_client_request(MClientRequest *req)
   case MDS_OP_OPENWRC:
 	handle_client_openwrc(req);
 	return 0;
+
+  case MDS_OP_MKDIR:
+	handle_client_mkdir(req);
+	return 0;
   }
 
   
@@ -400,6 +404,7 @@ int MDS::handle_client_request(MClientRequest *req)
   case MDS_OP_OPENRD:
 	reply = handle_client_openrd(req, cur);
 	break;
+
   case MDS_OP_OPENWR:
 	reply = handle_client_openwr(req, cur);
 	break;
@@ -408,6 +413,16 @@ int MDS::handle_client_request(MClientRequest *req)
 	handle_client_unlink(req, cur);
 	break;
 
+	/*
+  case MDS_OP_RMDIR:
+	handle_client_rmdir(req, cur);
+	break;
+
+  case MDS_OP_RENAME:
+	handle_client_rename(req, cur);
+	break;
+	*/
+
   default:
 	dout(1) << " unknown mop " << req->get_op() << endl;
 	assert(0);
@@ -915,9 +930,119 @@ void MDS::handle_client_unlink_2(MClientRequest *req,
 }
 
 
+void MDS::handle_client_mkdir(MClientRequest *req) 
+{
+  // get containing directory
+  filepath dirpath = req->get_filepath().subpath(req->get_filepath().depth());
+  string name = req->get_filepath().last_bit();
+
+  vector<CInode*> trace;
+  int r = mdcache->path_traverse(dirpath, trace, req, MDS_TRAVERSE_FORWARD);
+  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;
+	return;
+  }
+
+  // ok!
+  CInode *diri = trace[trace.size()-1];
+  assert(diri->is_dir());
+  assert(diri->dir_is_auth());
+  CDir *dir = diri->get_or_open_dir(this);
+  
+  // make sure name doesn't already exist
+  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;
+	return;
+  }
+
+  // create!
+  CInode *newi = mdcache->create_inode();
+  mdcache->link_inode(dir, name, newi);
+  
+  newi->inode.isdir = 1;
+
+  newi->mark_dirty();
+
+  // log it
+  dout(10) << "log for " << *req << " create " << newi->ino() << endl;
+  mdlog->submit_entry(new EInodeUpdate(newi),                    // FIXME should be differnet log entry
+					  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;
+  return;
+}
 
 
 
+// WRITE ME
+
+
+/*
+void MDS::handle_client_rmdir(MClientRequest *req,
+							  CInode *cur)
+{
+  assert(cur->is_auth());
+  assert(cur->is_dir());
+  
+  if (cur->dir_is_auth()) {
+	// ok good, i have the dir and the inode.  
+	CDir *dir = cur->get_or_open_dir(this);
+	
+	// empty?
+	if (dir->get_size() > 0) {
+	  // not empty
+	  MClientReply *reply = new MClientReply(req, -ENOTEMPTY);
+	  return reply;
+	} else {
+	  if (!dir->is_complete()) {
+		// fetch
+		mdstore->fetch_dir(cur->dir, new C_MDS_RetryMessage(this, req));
+		return 0;
+	  }
+	  
+	  // else... complete and empty!
+	}
+	assert(dir->is_complete());
+	assert(dir->get_size() == 0);
+
+	// close any dups?
+	if (dir->is_open_by_anyone()) {
+	  // ***
+	}
+
+	// remove
+		
+	
+  } else {
+	// ugh: reimport, but only if empty, etc.
+	
+  }
+
+  
+}
+
+*/
 
 
 

ceph/mds/MDS.h

@@ -184,7 +184,11 @@ class MDS : public Dispatcher {
 							  CInode *cur);
 
   void handle_client_mkdir(MClientRequest *req);
-  void handle_client_rmdir(MClientRequest *req);
+  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);
 
 
   int do_stat(MClientRequest *m);

ceph/messages/MDirExpireReq.h

@@ -0,0 +1,37 @@
+#ifndef __MDIREXPIREREQ_H
+#define __MDIREXPIREREQ_H
+
+typedef struct {
+  inodeno_t ino;
+  int nonce;
+  int from;
+} MDirExpireReq_st;
+
+class MDirExpire : public Message {
+  MDirExpireReq_st st;
+
+ public:
+  inodeno_t get_ino() { return st.ino; }
+  int get_from() { return st.from; }
+  int get_nonce() { return st.nonce; }
+
+  MDirExpire() {}
+  MDirExpire(inodeno_t ino, int from, int nonce) :
+	Message(MSG_MDS_DIREXPIREREQ) {
+	st.ino = ino;
+	st.from = from;
+	st.nonce = nonce;
+  }
+  virtual char *get_type_name() { return "DirExR";}
+  
+  virtual int decode_payload(crope s) {
+	s.copy(0, sizeof(st), (char*)&st);
+  }
+  virtual crope get_payload() {
+	crope s;
+	s.append((char*)&st,sizeof(st));
+	return s;
+  }
+};
+
+#endif

ceph/msg/Message.h

@@ -25,7 +25,9 @@
 #define MSG_MDS_INODEUPDATE  120
 #define MSG_MDS_DIRUPDATE    121
 #define MSG_MDS_INODEEXPIRE  122
-#define MSG_MDS_DIREXPIRE  123
+#define MSG_MDS_DIREXPIRE    123
+
+#define MSG_MDS_DIREXPIREREQ 124
 
 #define MSG_MDS_CACHEEXPIRE  125