kclient: move readdir cache prepopulation inside reply handler, out of caller process context

src/TODO

@@ -27,12 +27,9 @@ userspace client
 - reference count lease validations on path lookup?
 
 kernel client
-- make sure create() is setting mode properly.. untar shouldn't need to chmod!
 - use list_for_each_safe for caps removal?
   - revisit cap removal locking, make sure it's okay....
-- fix dentry locking on lease addition
 - trim expired leases so we don't indefinitely hold dcache refs...
-- move readdir result prepopulation inside reply handler (not in caller context!)
 - carry wrbuffer/rdcache caps until data is flushed
   - this should make the utimes bit kick in
 - make sure link/unlink results reflected by inode/dentry cache  (let fill_trace do it?  invalidate?  do actual update?)

src/kernel/dir.c

@@ -94,77 +94,6 @@ static unsigned fpos_off(loff_t p)
 	return p & 0xffffffff;
 }
 
-
-static int prepopulate_dir(struct dentry *parent,
-			   struct ceph_mds_reply_info *rinfo,
-			   struct ceph_mds_session *session, 
-			   unsigned long from_time)
-{
-	struct qstr dname;
-	struct dentry *dn;
-	struct inode *in;
-	int i;
-
-	for (i = 0; i < rinfo->dir_nr; i++) {
-		/* dentry */
-		dname.name = rinfo->dir_dname[i];
-		dname.len = rinfo->dir_dname_len[i];
-		dname.hash = full_name_hash(dname.name, dname.len);
-
-		dn = d_lookup(parent, &dname);
-		dout(30, "calling d_lookup on parent=%p name=%.*s"
-		     " returned %p\n", parent, dname.len, dname.name, dn);
-
-		if (!dn) {
-			dn = d_alloc(parent, &dname);
-			if (dn == NULL) {
-				dout(30, "d_alloc badness\n");
-				return -1;
-			}
-			ceph_init_dentry(dn);
-		}
-		ceph_update_dentry_lease(dn, rinfo->dir_dlease[i], 
-					 session, from_time);
-
-		/* inode */
-		if (dn->d_inode == NULL) {
-			in = new_inode(parent->d_sb);
-			if (in == NULL) {
-				dout(30, "new_inode badness\n");
-				d_delete(dn);
-				return -1;
-			}
-		} else {
-			in = dn->d_inode;
-		}
-
-		if (ceph_ino(in) !=
-		    le64_to_cpu(rinfo->dir_in[i].in->ino)) {
-			if (ceph_fill_inode(in, rinfo->dir_in[i].in) < 0) {
-				dout(30, "ceph_fill_inode badness\n");
-				iput(in);
-				d_delete(dn);
-				return -1;
-			}
-			d_instantiate(dn, in);
-			if (d_unhashed(dn))
-				d_rehash(dn);
-			dout(10, "dir_readdir added dentry %p ino %llx %d/%d\n",
-			     dn, ceph_ino(in), i, rinfo->dir_nr);
-		} else {
-			if (ceph_fill_inode(in, rinfo->dir_in[i].in) < 0) {
-				dout(30, "ceph_fill_inode badness\n");
-				return -1;
-			}
-		}
-		ceph_update_inode_lease(in, rinfo->dir_ilease[i], session,
-					from_time);
-
-		dput(dn);
-	}
-	return 0;
-}
-
 static int ceph_dir_readdir(struct file *filp, void *dirent, filldir_t filldir)
 {
 	struct ceph_file_info *fi = filp->private_data;
@@ -204,11 +133,6 @@ nextfrag:
 		dout(10, "dir_readdir got and parsed readdir result=%d"
 		     " on frag %u\n", err, frag);
 		fi->last_readdir = req;
-
-		/* pre-populate dentry cache */
-		prepopulate_dir(filp->f_dentry, &req->r_reply_info, 
-				req->r_session,
-				req->r_from_time);
 	}
 
 	/* include . and .. with first fragment */

src/kernel/inode.c

@@ -337,7 +337,6 @@ void ceph_revoke_dentry_lease(struct dentry *dentry)
 
 
 
-
 int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req, 
 		    struct ceph_mds_session *session)
 {
@@ -479,6 +478,81 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
 	return err;
 }
 
+/*
+ * prepopulate cache with readdir results
+ */
+int ceph_readdir_prepopulate(struct ceph_mds_request *req)
+{
+	struct dentry *parent = req->r_last_dentry;
+	struct ceph_mds_reply_info *rinfo = &req->r_reply_info;
+	struct qstr dname;
+	struct dentry *dn;
+	struct inode *in;
+	int i;
+
+	dout(10, "readdir_prepopulate %d items under dentry %p\n", 
+	     rinfo->dir_nr, parent);
+	for (i = 0; i < rinfo->dir_nr; i++) {
+		/* dentry */
+		dname.name = rinfo->dir_dname[i];
+		dname.len = rinfo->dir_dname_len[i];
+		dname.hash = full_name_hash(dname.name, dname.len);
+
+		dn = d_lookup(parent, &dname);
+		dout(30, "calling d_lookup on parent=%p name=%.*s"
+		     " returned %p\n", parent, dname.len, dname.name, dn);
+
+		if (!dn) {
+			dn = d_alloc(parent, &dname);
+			if (dn == NULL) {
+				dout(30, "d_alloc badness\n");
+				return -1;
+			}
+			ceph_init_dentry(dn);
+		}
+		ceph_update_dentry_lease(dn, rinfo->dir_dlease[i], 
+					 req->r_session, req->r_from_time);
+
+		/* inode */
+		if (dn->d_inode == NULL) {
+			in = new_inode(parent->d_sb);
+			if (in == NULL) {
+				dout(30, "new_inode badness\n");
+				d_delete(dn);
+				return -1;
+			}
+		} else {
+			in = dn->d_inode;
+		}
+
+		if (ceph_ino(in) !=
+		    le64_to_cpu(rinfo->dir_in[i].in->ino)) {
+			if (ceph_fill_inode(in, rinfo->dir_in[i].in) < 0) {
+				dout(30, "ceph_fill_inode badness\n");
+				iput(in);
+				d_delete(dn);
+				return -1;
+			}
+			d_instantiate(dn, in);
+			if (d_unhashed(dn))
+				d_rehash(dn);
+			dout(10, "added dentry %p ino %llx %d/%d\n",
+			     dn, ceph_ino(in), i, rinfo->dir_nr);
+		} else {
+			if (ceph_fill_inode(in, rinfo->dir_in[i].in) < 0) {
+				dout(30, "ceph_fill_inode badness\n");
+				return -1;
+			}
+		}
+		ceph_update_inode_lease(in, rinfo->dir_ilease[i], 
+					req->r_session, req->r_from_time);
+
+		dput(dn);
+	}
+	dout(10, "readdir_prepopulate done\n");
+	return 0;
+}
+
 
 /*
  * capabilities

src/kernel/mds_client.c

@@ -928,17 +928,23 @@ void ceph_mdsc_handle_reply(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
 	err = ceph_fill_trace(mdsc->client->sb, req, req->r_session);
 	if (err)
 		goto done;
-	if (result == 0 && req->r_expects_cap) {
-		cap = le32_to_cpu(rinfo->head->file_caps);
-		capseq = le32_to_cpu(rinfo->head->file_caps_seq);
-		req->r_cap = ceph_add_cap(req->r_last_inode,
-					  req->r_session,
-					  cap, capseq);
-		if (IS_ERR(req->r_cap)) {
-			err = PTR_ERR(req->r_cap);
-			req->r_cap = 0;
-			goto done;
+	if (result == 0) {
+		/* caps? */
+		if (req->r_expects_cap) {
+			cap = le32_to_cpu(rinfo->head->file_caps);
+			capseq = le32_to_cpu(rinfo->head->file_caps_seq);
+			req->r_cap = ceph_add_cap(req->r_last_inode,
+						  req->r_session,
+						  cap, capseq);
+			if (IS_ERR(req->r_cap)) {
+				err = PTR_ERR(req->r_cap);
+				req->r_cap = 0;
+				goto done;
+			}
 		}
+
+		/* readdir result? */
+		ceph_readdir_prepopulate(req);
 	}
 
 done:

src/kernel/super.h

@@ -329,6 +329,10 @@ extern int ceph_get_inode(struct super_block *sb, u64 ino,
 			  struct inode **pinode);
 extern int ceph_fill_inode(struct inode *inode,
 			   struct ceph_mds_reply_inode *info);
+extern int ceph_fill_trace(struct super_block *sb, 
+			   struct ceph_mds_request *req,
+			   struct ceph_mds_session *session);
+extern int ceph_readdir_prepopulate(struct ceph_mds_request *req);
 
 extern void ceph_update_inode_lease(struct inode *inode, 
 				    struct ceph_mds_reply_lease *lease,
@@ -378,9 +382,6 @@ extern const struct file_operations ceph_dir_fops;
 extern struct dentry_operations ceph_dentry_ops;
 
 extern char *ceph_build_dentry_path(struct dentry *dentry, int *len);
-extern int ceph_fill_trace(struct super_block *sb, 
-			   struct ceph_mds_request *req,
-			   struct ceph_mds_session *session);
 extern int ceph_do_lookup(struct super_block *sb, struct dentry *dentry, int m);
 
 static inline void ceph_init_dentry(struct dentry *dentry) {