Make symlinks actually work:

all - leave room for trailing NUL, for simplicity
mds_getattr_internal - check reqbody->valid instead of repbody->valid
ll_symlink - use lli_open_sem to protect lli_symlink_name allocation/data

diff --git a/lustre/llite/namei.c b/lustre/llite/namei.c
index 5e6000c..893b457 100644 (file)
--- a/lustre/llite/namei.c
+++ b/lustre/llite/namei.c
@@ -145,7 +145,7 @@ static struct dentry *ll_lookup2(struct inode *dir, struct dentry *dentry,
         struct ll_inode_md md;
         struct lustre_handle lockh;
         struct lookup_intent lookup_it = { IT_LOOKUP };
-        int err, offset, mode;
+        int err, offset;
         obd_id ino = 0;
 
         ENTRY;
@@ -168,49 +168,59 @@ static struct dentry *ll_lookup2(struct inode *dir, struct dentry *dentry,
 
         request = (struct ptlrpc_request *)it->it_data;
         if (it->it_disposition) {
+                int mode, easize = 0;
+                obd_flag valid;
+
                 offset = 1;
                 if (it->it_op & (IT_CREAT | IT_MKDIR | IT_SYMLINK | IT_MKNOD)) {
                         /* For create ops, we want the lookup to be negative */
                         if (!it->it_status)
                                 GOTO(negative, NULL);
-                } else if (it->it_op & (IT_GETATTR | IT_UNLINK |
-                                        IT_RMDIR | IT_SETATTR | IT_LOOKUP)) {
-                        /* For remove/check, we want the lookup to succeed */
+                } else if (it->it_op & (IT_GETATTR | IT_SETATTR | IT_LOOKUP)) {
+                        /* For check ops, we want the lookup to succeed */
                         it->it_data = NULL;
                         if (it->it_status)
                                 GOTO(neg_req, NULL);
                 } else if (it->it_op & IT_RENAME) {
+                        /* For rename, we want the lookup to succeed */
                         if (it->it_status) {
                                 it->it_data = NULL;
                                 GOTO(neg_req, NULL);
                         }
                         it->it_data = dentry;
+                } else if (it->it_op & (IT_UNLINK | IT_RMDIR)) {
+                        /* For remove ops, we want the lookup to succeed */
+                        it->it_data = NULL;
+                        if (it->it_status)
+                                GOTO(neg_req, NULL);
+                        goto iget;
                 } else if (it->it_op == IT_OPEN) {
                         it->it_data = NULL;
                         if (it->it_status && it->it_status != -EEXIST)
                                 GOTO(neg_req, NULL);
                 } else if (it->it_op == IT_RENAME2) {
-                        struct mds_body *body = 
+                        struct mds_body *body =
                                 lustre_msg_buf(request->rq_repmsg, offset);
                         it->it_data = NULL;
-                        if (body->valid == 0) 
+                        /* For rename2, this means the lookup is negative */
+                        if (body->valid == 0)
                                 GOTO(neg_req, NULL);
-                        GOTO(iget, NULL);
+                        goto iget; /* XXX not sure about this */
                 }
 
                 /* Do a getattr now that we have the lock */
-                if ((it->it_op == IT_UNLINK || it->it_op == IT_RMDIR) &&
-                    it->it_status == 0)
-                        /* the unlink/rmdir succeeded, there's nothing to
-                         * lookup */
-                        goto iget;
                 md.body = lustre_msg_buf(request->rq_repmsg, offset);
                 ino = md.body->fid1.id;
                 mode = md.body->mode;
+                valid = OBD_MD_FLNOTOBD | OBD_MD_FLEASIZE;
+                if (it->it_op == IT_READLINK) {
+                        valid |= OBD_MD_LINKNAME;
+                        easize = md.body->size;
+                }
                 ptlrpc_free_req(request);
                 request = NULL;
                 err = mdc_getattr(&sbi->ll_mdc_conn, ino, mode,
-                                  OBD_MD_FLNOTOBD|OBD_MD_FLEASIZE, 0, &request);
+                                  valid, easize, &request);
                 if (err) {
                         CERROR("failure %d inode %Ld\n", err, (long long)ino);
                         ptlrpc_free_req(request);

diff --git a/lustre/llite/symlink.c b/lustre/llite/symlink.c
index 35941a8..ad50fab 100644 (file)
--- a/lustre/llite/symlink.c
+++ b/lustre/llite/symlink.c
@@ -1,4 +1,5 @@
-/*
+/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
+ * vim:expandtab:shiftwidth=8:tabstop=8:
  *  linux/fs/ext2/symlink.c
  *
  * This code is issued under the GNU General Public License.
@@ -13,7 +14,7 @@
  *
  *  ext2 symlink handling code
  *
- * Modified for OBDFS: 
+ * Modified for OBDFS:
  *  Copyright (C) 1999 Seagate Technology Inc. (author: braam@stelias.com)
  */
 
@@ -30,27 +31,44 @@
 static int ll_readlink(struct dentry *dentry, char *buffer, int buflen)
 {
         struct ll_sb_info *sbi = ll_i2sbi(dentry->d_inode);
-        struct ptlrpc_request *request;
-        char *tmp;
-        int rc, size;
+        struct ptlrpc_request *request = NULL;
+        struct inode *inode = dentry->d_inode;
+        struct ll_inode_info *lli = ll_i2info(inode);
+        int len = inode->i_size + 1;
+        char *symname;
+        int rc;
         ENTRY;
 
-        rc = mdc_getattr(&sbi->ll_mdc_conn, dentry->d_inode->i_ino, S_IFLNK,
-                         OBD_MD_LINKNAME, dentry->d_inode->i_size, &request);
+        /* on symlinks lli_open_sem protects lli_symlink_name allocation/data */
+        down(&lli->lli_open_sem);
+        if (lli->lli_symlink_name) {
+                symname = lli->lli_symlink_name;
+                CDEBUG(D_INODE, "using cached symlink %s\n", symname);
+                GOTO(out_readlink, rc = 0);
+        }
+
+        rc = mdc_getattr(&sbi->ll_mdc_conn, inode->i_ino, S_IFLNK,
+                         OBD_MD_LINKNAME, len, &request);
+
         if (rc) {
-                CERROR("failure %d inode %ld\n", rc,
-                       (long)dentry->d_inode->i_ino);
-                ptlrpc_free_req(request);
-                RETURN(rc);
+                CERROR("inode %d readlink: rc = %ld\n", rc, inode->i_ino);
+                GOTO(out_readlink_sem, rc);
         }
 
-        tmp = lustre_msg_buf(request->rq_repmsg, 1);
-        size = MIN(request->rq_repmsg->buflens[1], buflen);
-        rc = copy_to_user(buffer, tmp, size);
-        if (rc == 0)
-                rc = size;
+        symname = lustre_msg_buf(request->rq_repmsg, 1);
 
+        OBD_ALLOC(lli->lli_symlink_name, len);
+        /* do not return an error if we cannot cache the symlink locally */
+        if (lli->lli_symlink_name)
+                memcpy(lli->lli_symlink_name, symname, len);
+
+out_readlink:
+        rc = vfs_readlink(dentry, buffer, buflen, symname);
+
+out_readlink_sem:
+        up(&lli->lli_open_sem);
         ptlrpc_free_req(request);
+
         RETURN(rc);
 }
 

diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c
index 223987e..f5a0193 100644 (file)
--- a/lustre/mds/handler.c
+++ b/lustre/mds/handler.c
@@ -443,7 +443,7 @@ int mds_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 
 static int mds_getattr_internal(struct mds_obd *mds, struct dentry *dentry,
                                 struct ptlrpc_request *req,
-                                int request_off, int reply_off)
+                                struct mds_body *reqbody, int reply_off)
 {
         struct mds_body *body;
         struct inode *inode = dentry->d_inode;
@@ -472,7 +472,7 @@ static int mds_getattr_internal(struct mds_obd *mds, struct dentry *dentry,
                         RETURN(rc);
                 }
                 body->valid |= OBD_MD_FLEASIZE;
-        } else if ((body->valid & OBD_MD_LINKNAME) && S_ISLNK(inode->i_mode)) {
+        } else if (S_ISLNK(inode->i_mode) && reqbody->valid & OBD_MD_LINKNAME) {
                 char *symname = lustre_msg_buf(req->rq_repmsg, reply_off + 1);
                 int len = req->rq_repmsg->buflens[reply_off + 1];
 
@@ -498,7 +498,7 @@ static int mds_getattr_name(int offset, struct ptlrpc_request *req)
         struct inode *dir;
         struct lustre_handle lockh;
         char *name;
-        int namelen, flags, lock_mode, rc = 0, old_offset = offset;
+        int namelen, flags, lock_mode, rc = 0;
         __u64 res_id[3] = {0, 0, 0};
         ENTRY;
 
@@ -553,7 +553,7 @@ static int mds_getattr_name(int offset, struct ptlrpc_request *req)
                 GOTO(out_create_dchild, rc = -ESTALE);
         }
 
-        rc = mds_getattr_internal(mds, dchild, req, old_offset, offset);
+        rc = mds_getattr_internal(mds, dchild, req, body, offset);
 
         EXIT;
 out_create_dchild:
@@ -592,7 +592,7 @@ static int mds_getattr(int offset, struct ptlrpc_request *req)
                 size[1] = mds->mds_max_mdsize;
         } else if (body->valid & OBD_MD_LINKNAME) {
                 bufcount = 2;
-                size[1] = inode->i_size;
+                size[1] = inode->i_size + 1;
         }
 
         rc = lustre_pack_msg(bufcount, size, NULL, &req->rq_replen,
@@ -603,7 +603,7 @@ static int mds_getattr(int offset, struct ptlrpc_request *req)
                 GOTO(out, rc = 0);
         }
 
-        req->rq_status = mds_getattr_internal(mds, de, req, offset, 0);
+        req->rq_status = mds_getattr_internal(mds, de, req, body, 0);
 
 out:
         l_dput(de);