Fix up hangs with the upcall mechanism for RPCSEC_GSS and the

NFSv4 idmapper.

 - Correct bugs in idmapper wait code.
 - Correct layering bugs in RPCSEC_GSS and idmapper
 - Ensure we advance the file pointer in case of partial
   reads/writes of a message on the pipes

diff --git a/fs/nfs/idmap.c b/fs/nfs/idmap.c
index c8747d20b4206ddd405c07dd50b2438db3926486..5373803719576fd0d7941c268b9735774e9218c4 100644 (file)
--- a/fs/nfs/idmap.c
+++ b/fs/nfs/idmap.c
@@ -113,6 +113,7 @@ nfs_idmap_new(struct nfs_server *server)
         init_MUTEX(&idmap->idmap_lock);
         init_MUTEX(&idmap->idmap_im_lock);
         init_MUTEX(&idmap->idmap_hash_lock);
+       init_waitqueue_head(&idmap->idmap_wq);
 
        return (idmap);
 
@@ -126,7 +127,10 @@ nfs_idmap_delete(struct nfs_server *server)
 {
        struct idmap *idmap = server->idmap;
 
+       if (!idmap)
+               return;
        rpc_unlink(idmap->idmap_path);
+       server->idmap = NULL;
        kfree(idmap);
 }
 
@@ -181,17 +185,17 @@ nfs_idmap_id(struct nfs_server *server, u_int8_t type, char *name,
        msg.data = im;
        msg.len = sizeof(*im);
 
-       init_waitqueue_head(&idmap->idmap_wq);
        add_wait_queue(&idmap->idmap_wq, &wq);
-       set_current_state(TASK_UNINTERRUPTIBLE);
-
        if (rpc_queue_upcall(idmap->idmap_dentry->d_inode, &msg) < 0) {
-               set_current_state(TASK_RUNNING);
+               remove_wait_queue(&idmap->idmap_wq, &wq);
                goto out;
        }
 
+       set_current_state(TASK_UNINTERRUPTIBLE);
        up(&idmap->idmap_im_lock);
        schedule();
+       current->state = TASK_RUNNING;
+       remove_wait_queue(&idmap->idmap_wq, &wq);
        down(&idmap->idmap_im_lock);
 
        /*
@@ -252,20 +256,21 @@ nfs_idmap_name(struct nfs_server *server, u_int8_t type, uid_t id,
        msg.data = im;
        msg.len = sizeof(*im);
 
-       init_waitqueue_head(&idmap->idmap_wq);
        add_wait_queue(&idmap->idmap_wq, &wq);
-       set_current_state(TASK_UNINTERRUPTIBLE);
 
        if (rpc_queue_upcall(idmap->idmap_dentry->d_inode, &msg) < 0) {
-               set_current_state(TASK_RUNNING);
+               remove_wait_queue(&idmap->idmap_wq, &wq);
                goto out;
        }
 
        /*
         * XXX add timeouts here
         */
+       set_current_state(TASK_UNINTERRUPTIBLE);
        up(&idmap->idmap_im_lock);
        schedule();
+       current->state = TASK_RUNNING;
+       remove_wait_queue(&idmap->idmap_wq, &wq);
        down(&idmap->idmap_im_lock);
 
        if (im->im_status & IDMAP_STATUS_SUCCESS) {
@@ -298,8 +303,14 @@ idmap_pipe_upcall(struct file *filp, struct rpc_pipe_msg *msg,
                 mlen = buflen;
 
         left = copy_to_user(dst, data, mlen);
-
-        return (mlen - left);
+       if (left < 0) {
+               msg->errno = left;
+               return left;
+       }
+       mlen -= left;
+       msg->copied += mlen;
+       msg->errno = 0;
+        return mlen;
 }
 
 static ssize_t
@@ -342,7 +353,6 @@ idmap_pipe_downcall(struct file *filp, const char *src, size_t mlen)
        if (match) {
                memcpy(im, &im_in, sizeof(*im));
                wake_up(&idmap->idmap_wq);
-               __rpc_purge_current_upcall(filp);
        } else if (!badmsg) {
                hashtype = im_in.im_conv == IDMAP_CONV_IDTONAME ?
                    IDMAP_HASH_TYPE_ID : IDMAP_HASH_TYPE_NAME;
@@ -361,6 +371,8 @@ idmap_pipe_destroy_msg(struct rpc_pipe_msg *msg)
        struct idmap_msg *im = msg->data;
        struct idmap *idmap = container_of(im, struct idmap, idmap_im); 
 
+       if (msg->errno >= 0)
+               return;
        down(&idmap->idmap_im_lock);
        im->im_status = IDMAP_STATUS_LOOKUPFAIL;
        wake_up(&idmap->idmap_wq);

diff --git a/include/linux/sunrpc/rpc_pipe_fs.h b/include/linux/sunrpc/rpc_pipe_fs.h
index d16cb88210b2b530b41d8541265b6e7f7e714d6f..eabb2ebf228933c7e02d97d4733b3c66c8b8afa9 100644 (file)
--- a/include/linux/sunrpc/rpc_pipe_fs.h
+++ b/include/linux/sunrpc/rpc_pipe_fs.h
@@ -43,8 +43,5 @@ extern int rpc_rmdir(char *);
 extern struct dentry *rpc_mkpipe(char *, void *, struct rpc_pipe_ops *, int flags);
 extern int rpc_unlink(char *);
 
-void __rpc_purge_current_upcall(struct file *);
-void __rpc_purge_one_upcall(struct file *filp, struct rpc_pipe_msg *target);
-
 #endif
 #endif

diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index dc742cc61aca725b2bb172abebc049b436d6228b..79dbcca25f1e83af578c6b25b754f19d7723aca0 100644 (file)
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -292,21 +292,59 @@ struct gss_upcall_msg {
 static void
 gss_release_msg(struct gss_upcall_msg *gss_msg)
 {
-       if (atomic_dec_and_test(&gss_msg->count))
-               kfree(gss_msg);
+       struct gss_auth *gss_auth = gss_msg->auth;
+
+       if (!atomic_dec_and_lock(&gss_msg->count, &gss_auth->lock))
+               return;
+       if (!list_empty(&gss_msg->list))
+               list_del(&gss_msg->list);
+       spin_unlock(&gss_auth->lock);
+       kfree(gss_msg);
 }
 
 static struct gss_upcall_msg *
-gss_find_upcall(struct gss_auth *gss_auth, uid_t uid)
+__gss_find_upcall(struct gss_auth *gss_auth, uid_t uid)
 {
        struct gss_upcall_msg *pos;
        list_for_each_entry(pos, &gss_auth->upcalls, list) {
-               if (pos->uid == uid)
-                       return pos;
+               if (pos->uid != uid)
+                       continue;
+               atomic_inc(&pos->count);
+               return pos;
        }
        return NULL;
 }
 
+static struct gss_upcall_msg *
+gss_find_upcall(struct gss_auth *gss_auth, uid_t uid)
+{
+       struct gss_upcall_msg *gss_msg;
+
+       spin_lock(&gss_auth->lock);
+       gss_msg = __gss_find_upcall(gss_auth, uid);
+       spin_unlock(&gss_auth->lock);
+       return gss_msg;
+}
+
+static void
+__gss_unhash_msg(struct gss_upcall_msg *gss_msg)
+{
+       if (list_empty(&gss_msg->list))
+               return;
+       list_del_init(&gss_msg->list);
+       rpc_wake_up(&gss_msg->waitq);
+}
+
+static void
+gss_unhash_msg(struct gss_upcall_msg *gss_msg)
+{
+       struct gss_auth *gss_auth = gss_msg->auth;
+
+       spin_lock(&gss_auth->lock);
+       __gss_unhash_msg(gss_msg);
+       spin_unlock(&gss_auth->lock);
+}
+
 static void
 gss_release_callback(struct rpc_task *task)
 {
@@ -315,12 +353,10 @@ gss_release_callback(struct rpc_task *task)
                        struct gss_auth, rpc_auth);
        struct gss_upcall_msg *gss_msg;
 
-       spin_lock(&gss_auth->lock);
        gss_msg = gss_find_upcall(gss_auth, task->tk_msg.rpc_cred->cr_uid);
-       if (gss_msg) {
-               rpc_wake_up(&gss_msg->waitq);
-       }
-       spin_unlock(&gss_auth->lock);
+       BUG_ON(!gss_msg);
+       atomic_dec(&gss_msg->count);
+       gss_release_msg(gss_msg);
 }
 
 static int
@@ -334,22 +370,22 @@ gss_upcall(struct rpc_clnt *clnt, struct rpc_task *task, uid_t uid)
        int res;
 
 retry:
-       gss_msg = gss_find_upcall(gss_auth, uid);
-       if (gss_msg == NULL && gss_new == NULL) {
+       gss_msg = __gss_find_upcall(gss_auth, uid);
+       if (gss_msg)
+               goto out_sleep;
+       if (gss_new == NULL) {
                spin_unlock(&gss_auth->lock);
                gss_new = kmalloc(sizeof(*gss_new), GFP_KERNEL);
-               spin_lock(&gss_auth->lock);
                if (gss_new)
-                       goto retry;
-               return -ENOMEM;
+                       return -ENOMEM;
+               spin_lock(&gss_auth->lock);
+               goto retry;
        }
-       if (gss_msg)
-               goto out_sleep;
        gss_msg = gss_new;
        memset(gss_new, 0, sizeof(*gss_new));
        INIT_LIST_HEAD(&gss_new->list);
        INIT_RPC_WAITQ(&gss_new->waitq, "RPCSEC_GSS upcall waitq");
-       atomic_set(&gss_new->count, 1);
+       atomic_set(&gss_new->count, 2);
        msg = &gss_new->msg;
        msg->data = &gss_new->uid;
        msg->len = sizeof(gss_new->uid);
@@ -361,15 +397,14 @@ retry:
        rpc_sleep_on(&gss_msg->waitq, task, gss_release_callback, NULL);
        spin_unlock(&gss_auth->lock);
        res = rpc_queue_upcall(dentry->d_inode, msg);
-       spin_lock(&gss_auth->lock);
        if (res) {
-               rpc_wake_up(&gss_msg->waitq);
-               list_del(&gss_msg->list);
+               gss_unhash_msg(gss_msg);
                gss_release_msg(gss_msg);
        }
        return res;
 out_sleep:
-       rpc_sleep_on(&gss_msg->waitq, task, NULL, NULL);
+       rpc_sleep_on(&gss_msg->waitq, task, gss_release_callback, NULL);
+       spin_unlock(&gss_auth->lock);
        if (gss_new)
                kfree(gss_new);
        return 0;
@@ -386,7 +421,14 @@ gss_pipe_upcall(struct file *filp, struct rpc_pipe_msg *msg,
        if (mlen > buflen)
                mlen = buflen;
        left = copy_to_user(dst, data, mlen);
-       return mlen - left;
+       if (left < 0) {
+               msg->errno = left;
+               return left;
+       }
+       mlen -= left;
+       msg->copied += mlen;
+       msg->errno = 0;
+       return mlen;
 }
 
 static ssize_t
@@ -432,14 +474,13 @@ gss_pipe_downcall(struct file *filp, const char *src, size_t mlen)
        else
                gss_cred_set_ctx(cred, ctx);
        spin_lock(&gss_auth->lock);
-       gss_msg = gss_find_upcall(gss_auth, acred.uid);
+       gss_msg = __gss_find_upcall(gss_auth, acred.uid);
        if (gss_msg) {
-               list_del(&gss_msg->list);
-               __rpc_purge_one_upcall(filp, &gss_msg->msg);
-               rpc_wake_up(&gss_msg->waitq);
+               __gss_unhash_msg(gss_msg);
+               spin_unlock(&gss_auth->lock);
                gss_release_msg(gss_msg);
-       }
-       spin_unlock(&gss_auth->lock);
+       } else
+               spin_unlock(&gss_auth->lock);
        rpc_release_client(clnt);
        return mlen;
 err:
@@ -454,13 +495,10 @@ void
 gss_pipe_destroy_msg(struct rpc_pipe_msg *msg)
 {
        struct gss_upcall_msg *gss_msg = container_of(msg, struct gss_upcall_msg, msg);
-       struct gss_auth *gss_auth = gss_msg->auth;
 
-       spin_lock(&gss_auth->lock);
-       list_del(&gss_msg->list);
-       rpc_wake_up(&gss_msg->waitq);
+       if (msg->errno < 0)
+               gss_unhash_msg(gss_msg);
        gss_release_msg(gss_msg);
-       spin_unlock(&gss_auth->lock);
 }
 
 /* 

diff --git a/net/sunrpc/rpc_pipe.c b/net/sunrpc/rpc_pipe.c
index 513797932eb1fbfe3db20900f06f105c10e15d27..d6cf592d6c6cc05841338793e981ec36d367aa5c 100644 (file)
--- a/net/sunrpc/rpc_pipe.c
+++ b/net/sunrpc/rpc_pipe.c
@@ -59,43 +59,6 @@ rpc_purge_upcall(struct inode *inode, int err)
        up(&inode->i_sem);
 }
 
-/*
- * XXX should only be called in ->downcall
- */
-void
-__rpc_purge_current_upcall(struct file *filp)
-{
-       struct rpc_pipe_msg *msg;
-
-       msg = filp->private_data;
-       filp->private_data = NULL;
-
-       if (msg != NULL)
-               msg->errno = 0;
-}
-
-void
-__rpc_purge_one_upcall(struct file *filp, struct rpc_pipe_msg *target)
-{
-       struct rpc_inode *rpci = RPC_I(filp->f_dentry->d_inode);
-       struct rpc_pipe_msg *msg;
-
-       msg = filp->private_data;
-       if (msg == target) {
-               filp->private_data = NULL;
-               goto found;
-       }
-       list_for_each_entry(msg, &rpci->pipe, list) {
-               if (msg == target) {
-                       list_del(&msg->list);
-                       goto found;
-               }
-       }
-       BUG();
-found:
-       return;
-}
-
 int
 rpc_queue_upcall(struct inode *inode, struct rpc_pipe_msg *msg)
 {
@@ -198,14 +161,15 @@ rpc_pipe_read(struct file *filp, char __user *buf, size_t len, loff_t *offset)
                        list_del_init(&msg->list);
                        rpci->pipelen -= msg->len;
                        filp->private_data = msg;
+                       msg->copied = 0;
                }
                if (msg == NULL)
                        goto out_unlock;
        }
+       /* NOTE: it is up to the callback to update msg->copied */
        res = rpci->ops->upcall(filp, msg, buf, len);
        if (res < 0 || msg->len == msg->copied) {
                filp->private_data = NULL;
-               msg->errno = 0;
                rpci->ops->destroy_msg(msg);
        }
 out_unlock:
@@ -685,7 +649,7 @@ rpc_mkpipe(char *path, void *private, struct rpc_pipe_ops *ops, int flags)
        if (IS_ERR(dentry))
                return dentry;
        dir = nd.dentry->d_inode;
-       inode = rpc_get_inode(dir->i_sb, S_IFSOCK | S_IRUSR | S_IXUSR);
+       inode = rpc_get_inode(dir->i_sb, S_IFSOCK | S_IRUSR | S_IWUSR);
        if (!inode)
                goto err_dput;
        inode->i_ino = iunique(dir->i_sb, 100);

diff --git a/net/sunrpc/sunrpc_syms.c b/net/sunrpc/sunrpc_syms.c
index 2a9975fd0306c0851091e03fec9a7953f8bb5824..bb05bd79a133064b419c40c4e991990d58114777 100644 (file)
--- a/net/sunrpc/sunrpc_syms.c
+++ b/net/sunrpc/sunrpc_syms.c
@@ -56,8 +56,6 @@ EXPORT_SYMBOL(rpc_unlink);
 EXPORT_SYMBOL(rpc_wake_up);
 EXPORT_SYMBOL(rpc_queue_upcall);
 EXPORT_SYMBOL(rpc_mkpipe);
-EXPORT_SYMBOL(__rpc_purge_current_upcall);
-EXPORT_SYMBOL(__rpc_purge_one_upcall);
 
 /* Client transport */
 EXPORT_SYMBOL(xprt_create_proto);