kclient: atomically handle fmode refcounts (untested)

src/TODO

@@ -20,7 +20,6 @@ yehuda:
 - apply_truncate needs to truncate page cache, or something?
 
 kernel client
-- take file_mode refs on inode safely/atomically on open
 - fix session close seq interaction with mds (sessions don't currently close all the way)
 - file_data_version stuff!
 - deal with CAP_RDCACHE properly: invalidate cache pages?

src/kernel/dir.c

@@ -190,15 +190,6 @@ nextfrag:
 	return 0;
 }
 
-
-const struct file_operations ceph_dir_fops = {
-	.read = generic_read_dir,
-	.readdir = ceph_dir_readdir,
-	.open = ceph_open,
-	.release = ceph_release,
-};
-
-
 int ceph_do_lookup(struct super_block *sb, struct dentry *dentry, int mask)
 {
 	struct ceph_client *client = ceph_sb_to_client(sb);
@@ -538,6 +529,12 @@ static void ceph_dentry_release(struct dentry *dentry)
 	}
 }
 
+const struct file_operations ceph_dir_fops = {
+	.read = generic_read_dir,
+	.readdir = ceph_dir_readdir,
+	.open = ceph_open,
+	.release = ceph_release,
+};
 
 const struct inode_operations ceph_dir_iops = {
 	.lookup = ceph_dir_lookup,

src/kernel/file.c

@@ -15,7 +15,7 @@ int ceph_debug_file = -1;
 /*
  * if err==0, caller is responsible for a put_session on *psession
  */
-struct ceph_mds_request *
+static struct ceph_mds_request *
 prepare_open_request(struct super_block *sb, struct dentry *dentry,
 		     int flags, int create_mode)
 {
@@ -36,6 +36,7 @@ prepare_open_request(struct super_block *sb, struct dentry *dentry,
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_OPEN, pathbase, path,
 				       0, 0);
 	req->r_expects_cap = 1;
+	req->r_fmode = ceph_file_mode(flags);
 	kfree(path);
 	if (!IS_ERR(req)) {
 		rhead = req->r_request->front.iov_base;
@@ -45,20 +46,21 @@ prepare_open_request(struct super_block *sb, struct dentry *dentry,
 	return req;
 }
 
-
-static int ceph_init_file(struct inode *inode, struct file *file, int flags)
+/*
+ * initialize private struct file data.
+ * if we fail, clean up by dropping fmode reference on the ceph_inode
+ */
+static int ceph_init_file(struct inode *inode, struct file *file, int fmode)
 {
-	struct ceph_inode_info *ci = ceph_inode(inode);
 	struct ceph_file_info *cf;
-	int fmode = ceph_file_mode(flags);
 
 	cf = kzalloc(sizeof(*cf), GFP_KERNEL);
-	if (cf == NULL)
+	if (cf == NULL) {
+		ceph_put_fmode(ceph_inode(inode), fmode);  /* clean up */
 		return -ENOMEM;
+	}
 	file->private_data = cf;
-
 	cf->mode = fmode;
-	ceph_get_mode(ci, fmode);
 	return 0;
 }
 
@@ -94,10 +96,9 @@ int ceph_open(struct inode *inode, struct file *file)
 	if ((__ceph_caps_issued(ci) & wantcaps) == wantcaps) {
 		dout(10, "open fmode %d caps %d using existing on %p\n", 
 		     fmode, wantcaps, inode);
+		__ceph_get_fmode(ci, fmode);
 		spin_unlock(&inode->i_lock);
-		err = ceph_init_file(inode, file, flags);
-		BUG_ON(err); /* fixme */
-		return 0;
+		return ceph_init_file(inode, file, fmode);
 	} 
 	spin_unlock(&inode->i_lock);
 	dout(10, "open mode %d, don't have caps %d\n", fmode, wantcaps);
@@ -106,8 +107,8 @@ int ceph_open(struct inode *inode, struct file *file)
 	if (IS_ERR(req))
 		return PTR_ERR(req);
 	err = ceph_mdsc_do_request(mdsc, req);
-	if (err == 0) 
-		err = ceph_init_file(inode, file, flags);
+	if (err == 0)
+		err = ceph_init_file(inode, file, req->r_fmode);
 	ceph_mdsc_put_request(req);
 	dout(5, "ceph_open result=%d on %llx\n", err, ceph_ino(inode));
 	return err;
@@ -146,7 +147,7 @@ int ceph_lookup_open(struct inode *dir, struct dentry *dentry,
 	req->r_last_dentry = dentry; /* use this dentry in fill_trace */
 	err = ceph_mdsc_do_request(mdsc, req);
 	if (err == 0)
-		err = ceph_init_file(req->r_last_inode, file, flags);
+		err = ceph_init_file(req->r_last_inode, file, req->r_fmode);
 	else if (err == -ENOENT) {
 		ceph_init_dentry(dentry);
 		d_add(dentry, NULL);
@@ -173,7 +174,7 @@ int ceph_release(struct inode *inode, struct file *file)
 	 * for now, store the open mode in ceph_file_info.
 	 */
 
-	ceph_put_mode(ci, cf->mode);
+	ceph_put_fmode(ci, cf->mode);
 	if (cf->last_readdir)
 		ceph_mdsc_put_request(cf->last_readdir);
 	kfree(cf);

src/kernel/inode.c

@@ -651,6 +651,7 @@ static struct ceph_inode_cap *__get_cap_for_mds(struct inode *inode, int mds)
  */
 struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
 				    struct ceph_mds_session *session, 
+				    int fmode,
 				    u32 issued, u32 seq)
 {
 	int mds = session->s_mds;
@@ -701,6 +702,7 @@ struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
 	cap->issued |= issued;
 	cap->implemented |= issued;
 	cap->seq = seq;
+	__ceph_get_fmode(ci, fmode);
 	spin_unlock(&inode->i_lock);
 	if (is_new)
 		igrab(inode);
@@ -929,25 +931,14 @@ void ceph_inode_set_size(struct inode *inode, loff_t size)
 		spin_unlock(&inode->i_lock);
 }
 
-/*
- * called on struct file init and release, to safely track which
- * capabilities we want due to currently open files
- */
-void ceph_get_mode(struct ceph_inode_info *ci, int mode)
-{
-	spin_lock(&ci->vfs_inode.i_lock);
-	ci->i_nr_by_mode[mode]++;
-	spin_unlock(&ci->vfs_inode.i_lock);
-}
-
-void ceph_put_mode(struct ceph_inode_info *ci, int mode)
+void ceph_put_fmode(struct ceph_inode_info *ci, int fmode)
 {
 	int last = 0;
 
 	spin_lock(&ci->vfs_inode.i_lock);
-	dout(20, "put_mode %p mode %d %d -> %d\n", &ci->vfs_inode, mode,
-	     ci->i_nr_by_mode[mode], ci->i_nr_by_mode[mode]-1);
-	if (--ci->i_nr_by_mode[mode] == 0)
+	dout(20, "put_mode %p fmode %d %d -> %d\n", &ci->vfs_inode, fmode,
+	     ci->i_nr_by_mode[fmode], ci->i_nr_by_mode[fmode]-1);
+	if (--ci->i_nr_by_mode[fmode] == 0)
 		last++;
 	spin_unlock(&ci->vfs_inode.i_lock);
 

src/kernel/mds_client.c

@@ -406,6 +406,7 @@ static struct ceph_mds_request *new_request(struct ceph_msg *msg)
 	req->r_last_dentry = 0;
 	req->r_old_dentry = 0;
 	req->r_expects_cap = 0;
+	req->r_fmode = 0;
 	req->r_cap = 0;
 	req->r_session = 0;
 	req->r_num_mds = 0;
@@ -930,6 +931,7 @@ void ceph_mdsc_handle_reply(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
 			capseq = le32_to_cpu(rinfo->head->file_caps_seq);
 			req->r_cap = ceph_add_cap(req->r_last_inode,
 						  req->r_session,
+						  req->r_fmode,
 						  cap, capseq);
 			if (IS_ERR(req->r_cap)) {
 				err = PTR_ERR(req->r_cap);

src/kernel/mds_client.h

@@ -76,6 +76,7 @@ struct ceph_mds_request {
 	struct dentry    *r_last_dentry;
 	struct dentry    *r_old_dentry;   /* for rename */
 	int			r_expects_cap;
+	int                     r_fmode;  /* if expecting cap */
 	unsigned long           r_from_time;
 	struct ceph_inode_cap  *r_cap;
 	struct ceph_mds_session *r_session;

src/kernel/super.h

@@ -326,6 +326,12 @@ static inline int ceph_file_mode(int flags)
 	return FILE_MODE_RDWR;  /* not -EINVAL under Linux, strangely */
 }
 
+static inline void __ceph_get_fmode(struct ceph_inode_info *ci, int mode) 
+{
+	ci->i_nr_by_mode[mode]++;
+}
+extern void ceph_put_fmode(struct ceph_inode_info *ci, int mode);
+
 static inline struct ceph_client *ceph_inode_to_client(struct inode *inode)
 {
 	return (struct ceph_client *)inode->i_sb->s_fs_info;
@@ -399,6 +405,7 @@ extern int ceph_dentry_lease_valid(struct dentry *dentry);
 
 extern struct ceph_inode_cap *ceph_add_cap(struct inode *inode,
 					   struct ceph_mds_session *session,
+					   int fmode,
 					   u32 cap, u32 seq);
 extern void ceph_remove_cap(struct ceph_inode_cap *cap);
 extern void ceph_remove_all_caps(struct ceph_inode_info *ci);
@@ -414,8 +421,6 @@ extern void ceph_put_cap_refs(struct ceph_inode_info *ci, int had);
 extern void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr);
 extern void ceph_cap_delayed_work(struct work_struct *work);
 extern void ceph_check_caps(struct ceph_inode_info *ci, int was_last);
-extern void ceph_get_mode(struct ceph_inode_info *ci, int mode);
-extern void ceph_put_mode(struct ceph_inode_info *ci, int mode);
 extern void ceph_inode_set_size(struct inode *inode, loff_t size);
 extern void ceph_inode_writeback(struct work_struct *work);
 

src/start.sh

@@ -43,7 +43,7 @@ do
 done
 
 # mds
-$CEPH_BIN/cmds $ARGS --debug_mds 20 --debug_ms 1
+$CEPH_BIN/cmds $ARGS --debug_mds 20 --debug_ms 20
 
 echo "started.  stop.sh to stop.  see out/* (e.g. 'tail -f out/????') for debug output."