kclient: async_create

Need to eventually write these back, and take cap references to ensure that
happens before, say, readdir.

6 files changed, 129 insertions, 31 deletions

diff --git a/src/include/ceph_fs.h b/src/include/ceph_fs.h
index 285a9c75992..e583a276b29 100644
--- a/src/include/ceph_fs.h
+++ b/src/include/ceph_fs.h
@@ -972,6 +972,8 @@ static inline int ceph_flags_to_mode(int flags)
 #define CEPH_CAP_ANY_FILE_WR (CEPH_CAP_FILE_WR|CEPH_CAP_FILE_WRBUFFER)
 #define CEPH_CAP_ANY_WR   (CEPH_CAP_ANY_EXCL | CEPH_CAP_ANY_FILE_WR)
 
+#define CEPH_CAP_ANY      (CEPH_CAP_ANY_WR|CEPH_CAP_ANY_RD)
+
 /*
  * these cap bits time out, if no others are held and nothing is
  * registered as 'wanted' by the client.
diff --git a/src/kernel/caps.c b/src/kernel/caps.c
index 25340319b68..1e4850dd371 100644
--- a/src/kernel/caps.c
+++ b/src/kernel/caps.c
@@ -319,6 +319,12 @@ int __ceph_caps_issued(struct ceph_inode_info *ci, int *implemented)
 	struct ceph_cap *cap;
 	struct rb_node *p;
 
+	if (ci->i_ceph_flags & CEPH_I_NEW) {
+		if (implemented)
+			*implemented = CEPH_CAP_ANY;
+		return CEPH_CAP_ANY;
+	}
+
 	if (implemented)
 		*implemented = 0;
 	for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
diff --git a/src/kernel/dir.c b/src/kernel/dir.c
index 918f933eec9..136350ed94b 100644
--- a/src/kernel/dir.c
+++ b/src/kernel/dir.c
@@ -349,18 +349,24 @@ static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
 }
 
 static int ceph_mknod(struct inode *dir, struct dentry *dentry,
-			  int mode, dev_t rdev)
+		      int mode, dev_t rdev)
 {
 	struct ceph_client *client = ceph_sb_to_client(dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
 	int err;
+	int issued = ceph_caps_issued(ceph_inode(dir));
 
 	if (ceph_snap(dir) != CEPH_NOSNAP)
 		return -EROFS;
 
 	dout(5, "mknod in dir %p dentry %p mode 0%o rdev %d\n",
 	     dir, dentry, mode, rdev);
+	if (ceph_async_create(dir, dentry, issued, mode, NULL) == 0) {
+		dentry->d_inode->i_rdev = rdev;
+		return 0;
+	}
+
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_MKNOD, dentry, NULL,
 				       NULL, NULL, USE_AUTH_MDS);
 	if (IS_ERR(req)) {
@@ -370,7 +376,7 @@ static int ceph_mknod(struct inode *dir, struct dentry *dentry,
 	req->r_locked_dir = dir;
 	req->r_args.mknod.mode = cpu_to_le32(mode);
 	req->r_args.mknod.rdev = cpu_to_le32(rdev);
-	if ((ceph_caps_issued(ceph_inode(dir)) & CEPH_CAP_FILE_EXCL) == 0)
+	if ((issued & CEPH_CAP_FILE_EXCL) == 0)
 		ceph_release_caps(dir, CEPH_CAP_FILE_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, dir, req);
 	if (!err && req->r_reply_info.trace_numd == 0) {
@@ -419,17 +425,22 @@ static int ceph_create(struct inode *dir, struct dentry *dentry, int mode,
 }
 
 static int ceph_symlink(struct inode *dir, struct dentry *dentry,
-			    const char *dest)
+			const char *dest)
 {
 	struct ceph_client *client = ceph_sb_to_client(dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct ceph_mds_request *req;
 	int err;
+	int issued = ceph_caps_issued(ceph_inode(dir));
 
 	if (ceph_snap(dir) != CEPH_NOSNAP)
 		return -EROFS;
 
 	dout(5, "symlink in dir %p dentry %p to '%s'\n", dir, dentry, dest);
+	if (ceph_async_create(dir, dentry, issued, S_IFLNK|S_IRWXUGO,
+			      dest) == 0)
+		return 0;
+
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_SYMLINK,
 				       dentry, NULL, NULL, dest, USE_AUTH_MDS);
 	if (IS_ERR(req)) {
@@ -437,7 +448,7 @@ static int ceph_symlink(struct inode *dir, struct dentry *dentry,
 		return PTR_ERR(req);
 	}
 	req->r_locked_dir = dir;
-	if ((ceph_caps_issued(ceph_inode(dir)) & CEPH_CAP_FILE_EXCL) == 0)
+	if ((issued & CEPH_CAP_FILE_EXCL) == 0)
 		ceph_release_caps(dir, CEPH_CAP_FILE_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, dir, req);
 	ceph_mdsc_put_request(req);
@@ -456,6 +467,7 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, int mode)
 	int snaplen;
 	struct dentry *pathdentry = dentry;
 	int op = CEPH_MDS_OP_MKDIR;
+	int issued = ceph_caps_issued(ceph_inode(dir));
 
 	if (ceph_snap(dir) == CEPH_SNAPDIR) {
 		/* mkdir .snap/foo is a MKSNAP */
@@ -470,6 +482,9 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, int mode)
 		return -EROFS;
 	} else {
 		dout(5, "mkdir dir %p dn %p mode 0%o\n", dir, dentry, mode);
+		mode |= S_IFDIR;
+		if (ceph_async_create(dir, dentry, issued, mode, NULL) == 0)
+			return 0;
 	}
 	req = ceph_mdsc_create_request(mdsc, op, pathdentry, NULL, 
 				       NULL, NULL, USE_AUTH_MDS);
@@ -483,7 +498,7 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, int mode)
 	req->r_locked_dir = dir;
 	req->r_args.mkdir.mode = cpu_to_le32(mode);
 
-	if ((ceph_caps_issued(ceph_inode(dir)) & CEPH_CAP_FILE_EXCL) == 0)
+	if ((issued & CEPH_CAP_FILE_EXCL) == 0)
 		ceph_release_caps(dir, CEPH_CAP_FILE_RDCACHE);
 	err = ceph_mdsc_do_request(mdsc, dir, req);
 	ceph_mdsc_put_request(req);
@@ -493,7 +508,7 @@ static int ceph_mkdir(struct inode *dir, struct dentry *dentry, int mode)
 }
 
 static int ceph_link(struct dentry *old_dentry, struct inode *dir,
-			 struct dentry *dentry)
+		     struct dentry *dentry)
 {
 	struct ceph_client *client = ceph_sb_to_client(dir->i_sb);
 	struct ceph_mds_client *mdsc = &client->mdsc;
diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index 3a1139c91de..bbd3730efb1 100644
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -301,6 +301,10 @@ struct inode *ceph_alloc_inode(struct super_block *sb)
 	INIT_LIST_HEAD(&ci->i_listener_list);
 	spin_lock_init(&ci->i_listener_lock);
 
+	ci->vfs_inode.i_mapping->a_ops = &ceph_aops;
+	ci->vfs_inode.i_mapping->backing_dev_info =
+		&ceph_client(sb)->backing_dev_info;
+
 	return &ci->vfs_inode;
 }
 
@@ -403,6 +407,96 @@ int ceph_fill_file_bits(struct inode *inode, int issued,
 	return queue_trunc;
 }
 
+static void init_inode_ops(struct inode *inode)
+{
+	switch (inode->i_mode & S_IFMT) {
+	case S_IFIFO:
+	case S_IFBLK:
+	case S_IFCHR:
+	case S_IFSOCK:
+		init_special_inode(inode, inode->i_mode, inode->i_rdev);
+		inode->i_op = &ceph_file_iops;
+		break;
+	case S_IFREG:
+		inode->i_op = &ceph_file_iops;
+		inode->i_fop = &ceph_file_fops;
+		break;
+	case S_IFLNK:
+		inode->i_op = &ceph_symlink_iops;
+		break;
+	case S_IFDIR:
+		inode->i_op = &ceph_dir_iops;
+		inode->i_fop = &ceph_dir_fops;
+		break;
+	default:
+		derr(0, "%p BAD mode 0%o S_IFMT 0%o\n", inode, inode->i_mode,
+		     inode->i_mode & S_IFMT);
+	}
+}
+
+int ceph_async_create(struct inode *dir, struct dentry *dentry,
+		      int issued, int mode, const char *symdest)
+{
+	struct ceph_vino vino;
+	struct inode *inode;
+	struct ceph_inode_info *ci;
+	struct ceph_mds_client *mdsc = &ceph_client(dir->i_sb)->mdsc;
+
+	dout(10, "async_create %p dn %p issued %s mode 0%o\n",
+	     dir, dentry, ceph_cap_string(issued), mode);
+	if ((issued & CEPH_CAP_FILE_EXCL) == 0 ||
+	    (issued & CEPH_CAP_AUTH_RDCACHE) == 0)
+		return -EPERM;
+	if ((ceph_inode(dir)->i_ceph_flags & CEPH_I_COMPLETE) == 0)
+		return -EPERM;
+
+	vino.ino = ceph_mdsc_prealloc_dequeue(mdsc);
+	if (!vino.ino)
+		return -EAGAIN;
+	vino.snap = CEPH_NOSNAP;
+
+	inode = ceph_get_inode(dir->i_sb, vino);
+	if (!inode)
+		return -EIO;
+	ci = ceph_inode(inode);
+
+	if (symdest) {
+		inode->i_size = strlen(symdest);
+		ci->i_symlink = kmalloc(inode->i_size + 1, GFP_NOFS);
+		if (!ci->i_symlink)
+			return -ENOMEM;
+		strcpy(ci->i_symlink, symdest);
+	}
+
+	inode->i_uid = current->fsuid;
+	if (dir->i_mode & S_ISGID) {
+		inode->i_gid = dir->i_gid;
+		if (S_ISDIR(mode))
+			mode |= S_ISGID;
+	} else
+		inode->i_gid = current->fsgid;
+	inode->i_mode = mode;
+
+	inode->i_mtime = inode->i_ctime = CURRENT_TIME;
+	inode->i_nlink = 1;
+
+	ci->i_version = 1;
+	ci->i_ceph_flags = CEPH_I_NEW | CEPH_I_COMPLETE;
+
+	if (S_ISDIR(mode)) 
+		ci->i_rsubdirs = 1;
+	else
+		ci->i_rfiles = 1;
+
+	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);
+	return 0;
+}
+
 /*
  * populate an inode based on info from mds.
  * may be called on new or existing inodes.
@@ -504,10 +598,6 @@ static int fill_inode(struct inode *inode,
 
 	ci->i_old_atime = inode->i_atime;
 
-	inode->i_mapping->a_ops = &ceph_aops;
-	inode->i_mapping->backing_dev_info =
-		&ceph_client(inode->i_sb)->backing_dev_info;
-
 no_change:
 	spin_unlock(&inode->i_lock);
 
@@ -559,20 +649,10 @@ no_change:
 	if (dirinfo)
 		ceph_fill_dirfrag(inode, dirinfo);
 
+	init_inode_ops(inode);
+
 	switch (inode->i_mode & S_IFMT) {
-	case S_IFIFO:
-	case S_IFBLK:
-	case S_IFCHR:
-	case S_IFSOCK:
-		init_special_inode(inode, inode->i_mode, inode->i_rdev);
-		inode->i_op = &ceph_file_iops;
-		break;
-	case S_IFREG:
-		inode->i_op = &ceph_file_iops;
-		inode->i_fop = &ceph_file_fops;
-		break;
 	case S_IFLNK:
-		inode->i_op = &ceph_symlink_iops;
 		if (!ci->i_symlink) {
 			int symlen = iinfo->symlink_len;
 
@@ -586,9 +666,6 @@ no_change:
 		}
 		break;
 	case S_IFDIR:
-		inode->i_op = &ceph_dir_iops;
-		inode->i_fop = &ceph_dir_fops;
-
 		ci->i_files = le64_to_cpu(info->files);
 		ci->i_subdirs = le64_to_cpu(info->subdirs);
 		ci->i_rbytes = le64_to_cpu(info->rbytes);
@@ -609,11 +686,6 @@ no_change:
 			ci->i_ceph_flags |= CEPH_I_COMPLETE;
 		}
 		break;
-	default:
-		derr(0, "BAD mode 0%o S_IFMT 0%o\n", inode->i_mode,
-		     inode->i_mode & S_IFMT);
-		err = -EINVAL;
-		goto out;
 	}
 	err = 0;
 
diff --git a/src/kernel/mds_client.h b/src/kernel/mds_client.h
index dbd424ff2f2..b19227f7836 100644
--- a/src/kernel/mds_client.h
+++ b/src/kernel/mds_client.h
@@ -312,6 +312,6 @@ extern void ceph_mdsc_flushed_all_caps(struct ceph_mds_client *mdsc,
 extern struct ceph_mds_request *ceph_mdsc_get_listener_req(struct inode *inode,
 						    u64 tid);
 
-extern u64 ceph_mdsc_prealloc_ino(struct ceph_mds_client *mdsc);
+extern u64 ceph_mdsc_prealloc_dequeue(struct ceph_mds_client *mdsc);
 
 #endif
diff --git a/src/kernel/super.h b/src/kernel/super.h
index f979e87ae8a..82a3b909f9c 100644
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -214,6 +214,7 @@ struct ceph_inode_frag {
  */
 #define CEPH_I_COMPLETE  1  /* we have complete directory cached */
 #define CEPH_I_READDIR   2  /* no dentries trimmed since readdir start */
+#define CEPH_I_NEW       4  /* not yet created on mds */
 
 struct ceph_inode_info {
 	struct ceph_vino i_vino;   /* ceph ino + snap */
@@ -677,6 +678,8 @@ extern struct kmem_cache *ceph_inode_cachep;
 
 extern struct inode *ceph_alloc_inode(struct super_block *sb);
 extern void ceph_destroy_inode(struct inode *inode);
+extern int ceph_async_create(struct inode *dir, struct dentry *dentry,
+			     int issued, int mode, const char *symdest);
 
 extern struct inode *ceph_get_inode(struct super_block *sb,
 				    struct ceph_vino vino);