kclient: async creates now mostly working

5 files changed, 58 insertions, 16 deletions

diff --git a/src/kernel/dir.c b/src/kernel/dir.c
index 54c47b317e3..8a774bdf764 100644
--- a/src/kernel/dir.c
+++ b/src/kernel/dir.c
@@ -273,7 +273,7 @@ struct dentry *ceph_finish_lookup(struct ceph_mds_request *req,
  * path built from @dentry.
  */
 struct dentry *ceph_do_lookup(struct super_block *sb, struct dentry *dentry,
-			      int mask, int on_inode, int locked_dir)
+			      int mask, int on_inode)
 {
 	struct ceph_client *client = ceph_sb_to_client(sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
@@ -339,13 +339,13 @@ static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
 				    (nd->flags & LOOKUP_CONTINUE) == 0 && /* only open last component */
 				    !(nd->intent.open.flags & O_CREAT)) {
 					int mode = nd->intent.open.create_mode & ~current->fs->umask;
-					return ceph_lookup_open(dir, dentry, nd, mode, 1);
+					return ceph_lookup_open(dir, dentry, nd, mode);
 				}
 
 		}
 	}
 
-	return ceph_do_lookup(dir->i_sb, dentry, CEPH_STAT_CAP_INODE, 0, 1);
+	return ceph_do_lookup(dir->i_sb, dentry, CEPH_STAT_CAP_INODE, 0);
 }
 
 static int ceph_mknod(struct inode *dir, struct dentry *dentry,
@@ -386,7 +386,7 @@ static int ceph_mknod(struct inode *dir, struct dentry *dentry,
 		 */
 		struct dentry *d;
 		d = ceph_do_lookup(dir->i_sb, dentry, CEPH_STAT_CAP_INODE_ALL,
-				   0, 0);
+				   0);
 		if (d) {
 			/* ick.  this is untested, and inherently racey... i
 			   suppose we _did_ create the file, but it has since
@@ -413,7 +413,7 @@ static int ceph_create(struct inode *dir, struct dentry *dentry, int mode,
 
 	if (nd) {
 		BUG_ON((nd->flags & LOOKUP_OPEN) == 0);
-		dentry = ceph_lookup_open(dir, dentry, nd, mode, 0);
+		dentry = ceph_lookup_open(dir, dentry, nd, mode);
 		/* hrm, what should i do here if we get aliased? */
 		if (IS_ERR(dentry))
 			return PTR_ERR(dentry);
@@ -518,6 +518,8 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
 	if (ceph_snap(dir) != CEPH_NOSNAP)
 		return -EROFS;
 
+	ceph_pending_flush(dentry);
+
 	dout(5, "link in dir %p old_dentry %p dentry %p\n", dir,
 	     old_dentry, dentry);
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LINK, dentry,
@@ -562,6 +564,8 @@ static int ceph_unlink(struct inode *dir, struct dentry *dentry)
 		CEPH_MDS_OP_RMDIR : CEPH_MDS_OP_UNLINK;
 	struct dentry *pathdentry = dentry;
 
+	ceph_pending_flush(dentry);
+
 	if (ceph_snap(dir) == CEPH_SNAPDIR) {
 		/* rmdir .snap/foo is RMSNAP */
 		op = CEPH_MDS_OP_RMSNAP;
@@ -609,6 +613,9 @@ static int ceph_rename(struct inode *old_dir, struct dentry *old_dentry,
 	    ceph_snap(new_dir) != CEPH_NOSNAP)
 		return -EROFS;
 
+	ceph_pending_flush(old_dentry);
+	ceph_pending_flush(d_find_alias(new_dir));
+
 	dout(5, "dir_rename in dir %p dentry %p to dir %p dentry %p\n",
 	     old_dir, old_dentry, new_dir, new_dentry);
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_RENAME,
diff --git a/src/kernel/file.c b/src/kernel/file.c
index b75368cf91a..37a8cd883d8 100644
--- a/src/kernel/file.c
+++ b/src/kernel/file.c
@@ -151,8 +151,7 @@ out:
  *  path_lookup_create -> LOOKUP_OPEN|LOOKUP_CREATE
  */
 struct dentry *ceph_lookup_open(struct inode *dir, struct dentry *dentry,
-				struct nameidata *nd, int mode,
-				int locked_dir)
+				struct nameidata *nd, int mode)
 {
 	struct ceph_client *client = ceph_sb_to_client(dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
@@ -161,16 +160,21 @@ struct dentry *ceph_lookup_open(struct inode *dir, struct dentry *dentry,
 	struct ceph_mds_request *req;
 	int err;
 	int flags = nd->intent.open.flags - 1;  /* silly vfs! */
+	int issued = ceph_caps_issued(ceph_inode(dir));
 
 	dout(5, "ceph_lookup_open dentry %p '%.*s' flags %d mode 0%o\n",
 	     dentry, dentry->d_name.len, dentry->d_name.name, flags, mode);
 
+	if (ceph_async_create(dir, dentry, issued, mode, NULL) == 0)
+		return 0;
+	ceph_pending_flush(d_find_alias(dir));
+
 	/* do the open */
 	req = prepare_open_request(dir->i_sb, dentry, flags, mode);
 	if (IS_ERR(req))
 		return ERR_PTR(PTR_ERR(req));
 	if ((flags & O_CREAT) &&
-	    (ceph_caps_issued(ceph_inode(dir)) & CEPH_CAP_FILE_EXCL) == 0)
+	    (issued & CEPH_CAP_FILE_EXCL) == 0)
 		ceph_release_caps(dir, CEPH_CAP_FILE_RDCACHE);
 	req->r_locked_dir = dir;           /* caller holds dir->i_mutex */
 	err = ceph_mdsc_do_request(mdsc, req, parent_inode);
diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index 9a9667cd8db..cf5d7d51b5f 100644
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -7,6 +7,7 @@
 #include <linux/kernel.h>
 #include <linux/namei.h>
 #include <linux/writeback.h>
+#include <linux/dcache.h>
 
 #include "ceph_debug.h"
 
@@ -451,6 +452,8 @@ int ceph_async_create(struct inode *dir, struct dentry *dentry,
 	int got;
 	int err;
 
+	BUG_ON(!mutex_is_locked(&dir->i_mutex));
+
 	if ((client->mount_args.flags & CEPH_MOUNT_ASYNCMETA) == 0)
 		return -EPERM;
 	mdsc = &client->mdsc;
@@ -483,6 +486,9 @@ int ceph_async_create(struct inode *dir, struct dentry *dentry,
 		if (!ci->i_symlink)
 			goto out_iput;
 		strcpy(ci->i_symlink, symdest);
+	} else {
+		ci->i_layout = client->mount_args.default_layout;
+		ci->i_max_size = ceph_file_layout_object_size(ci->i_layout);
 	}
 
 	/* add to dir child list */
@@ -497,7 +503,9 @@ int ceph_async_create(struct inode *dir, struct dentry *dentry,
 		inode->i_gid = current->fsgid;
 	inode->i_mode = mode;
 
-	inode->i_mtime = inode->i_ctime = CURRENT_TIME;
+	inode->i_mtime = CURRENT_TIME;
+	inode->i_ctime = CURRENT_TIME;
+	inode->i_atime = CURRENT_TIME;
 	inode->i_nlink = 1;
 
 	ci->i_version = 1;
@@ -508,12 +516,19 @@ int ceph_async_create(struct inode *dir, struct dentry *dentry,
 	else
 		ci->i_rfiles = 1;
 
+	/* inherit snap realm from parent dir */
+	down_read(&mdsc->snap_rwsem);
+	ci->i_snap_realm = dirci->i_snap_realm;
+	atomic_inc(&ci->i_snap_realm->nref);
+	up_read(&mdsc->snap_rwsem);
+
 	dout(10, "async_create %p dn %p issued %s mode 0%o = %p (%llx)\n",
 	     dir, dentry, ceph_cap_string(issued), mode, inode, vino.ino);
 
 	init_inode_ops(inode);
 
-	d_add(dentry, inode);
+	dout(0, "dentry %p d_name.hash %d\n", dentry, dentry->d_name.hash);
+	d_instantiate(dentry, inode);
 	return 0;
 
 out_iput:
@@ -535,10 +550,11 @@ static void ceph_create_completion(struct ceph_mds_client *mds,
 	dout(10, "create_completion %p on %p\n", req, inode);
 	mutex_lock(&mdsc->create_mutex);
 	if (ci->i_ceph_flags & CEPH_I_NEW) {
+		struct inode *dir = d_find_alias(inode)->d_parent->d_inode;
+
 		ci->i_ceph_flags &= ~(CEPH_I_NEW|CEPH_I_CREATING);
 		list_del_init(&ci->i_new_child);
-		ceph_put_cap_refs(ceph_inode(d_find_alias(inode)->d_parent->d_inode),
-				  CEPH_CAP_FILE_EXCL);
+		ceph_put_cap_refs(ceph_inode(dir), CEPH_CAP_FILE_EXCL);
 	}
 	mutex_unlock(&mdsc->create_mutex);
 	ceph_mdsc_put_request(req);
@@ -1956,7 +1972,7 @@ int ceph_do_getattr(struct dentry *dentry, int mask)
 		}
 	}
 	ret = ceph_do_lookup(dentry->d_inode->i_sb, dentry, mask,
-			     on_inode, 0);
+			     on_inode);
 	if (IS_ERR(ret))
 		return PTR_ERR(ret);
 	if (ret)
diff --git a/src/kernel/super.c b/src/kernel/super.c
index ec9688a988b..d7d241b50a0 100644
--- a/src/kernel/super.c
+++ b/src/kernel/super.c
@@ -480,6 +480,14 @@ static int parse_mount_args(int flags, char *options, const char *dev_name,
 	args->prealloc_max = 1024;
 	args->snapdir_name = ".snap";
 
+	args->default_layout.fl_stripe_unit = 4 << 20;
+	args->default_layout.fl_stripe_count = 1;
+	args->default_layout.fl_object_size = 4 << 20;
+	args->default_layout.fl_pg_preferred = -1;
+	args->default_layout.fl_pg_type = CEPH_PG_LAYOUT_CRUSH;
+	args->default_layout.fl_pg_size = 2;
+	args->default_layout.fl_pg_pool = 1;
+
 	/* ip1[:port1][,ip2[:port2]...]:/subdir/in/fs */
 	c = strstr(dev_name, ":/");
 	if (c == NULL)
diff --git a/src/kernel/super.h b/src/kernel/super.h
index 4a8a5cabf95..515230f8161 100644
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -71,6 +71,7 @@ struct ceph_mount_args {
 	int osd_timeout;
 	int prealloc_min, prealloc_max;
 	char *snapdir_name;   /* default ".snap" */
+	struct ceph_file_layout default_layout;
 };
 
 enum {
@@ -688,6 +689,13 @@ extern int ceph_async_create(struct inode *dir, struct dentry *dentry,
 			     int issued, int mode, const char *symdest);
 extern int ceph_flush_create(struct dentry *dentry);
 
+inline static void ceph_pending_flush(struct dentry *dentry)
+{
+	if (dentry && dentry->d_inode &&
+	    ceph_inode(dentry->d_inode)->i_ceph_flags & CEPH_I_NEW)
+		ceph_flush_create(dentry);
+}
+
 extern struct inode *ceph_get_inode(struct super_block *sb,
 				    struct ceph_vino vino);
 extern struct inode *ceph_get_snapdir(struct inode *parent);
@@ -758,8 +766,7 @@ extern const struct file_operations ceph_file_fops;
 extern const struct address_space_operations ceph_aops;
 extern int ceph_open(struct inode *inode, struct file *file);
 extern struct dentry *ceph_lookup_open(struct inode *dir, struct dentry *dentry,
-				       struct nameidata *nd, int mode,
-				       int locked_dir);
+				       struct nameidata *nd, int mode);
 extern int ceph_release(struct inode *inode, struct file *filp);
 
 
@@ -771,7 +778,7 @@ extern struct dentry_operations ceph_dentry_ops, ceph_snap_dentry_ops,
 
 extern struct dentry *ceph_do_lookup(struct super_block *sb,
 				     struct dentry *dentry,
-				     int mask, int on_inode, int locked_dir);
+				     int mask, int on_inode);
 extern struct dentry *ceph_finish_lookup(struct ceph_mds_request *req,
 					 struct dentry *dentry, int err);