[PATCH 0/5] (v2) rpcauth_create() returns -EEXIST

[PATCH 0/5] (v2) rpcauth_create() returns -EEXIST

[Chuck Lever]

A few weeks ago I reported a problem with calling rpcauth_create()
in a loop to serially replace an RPC transport's with different
flavors.  When the loop gets up to the GSS flavors, the second
GSS flavor in the list can't be created.  rpcauth_create() fails
with -EEXIST.  See:

 http://marc.info/?t=134073471800002&r=1&w=2

rpcauth_create() invokes the rpc_auth ->create operation first then
releases the old rpc_auth.  But the old rpc_auth is holding onto
upcall pipes, which prevents the new GSS rpc_auth from creating its
own upcall pipes.

This breaks our client's SECINFO implementation, and will also break
UCS server trunking discovery when it appears.

Trond proposed a solution in the same e-mail thread, but it was
archived uuencoded.  Here is his suggestion in human-readable form:

> The solution here would be to create a per-rpc_client, per-pipename
> shared 'rpcsec_gss_pipe' object that holds the upcall pipe data.

This version of the patch series includes a couple of additional
clean-up patches, and proper Acked-by: lines.  I've confirmed that
the series addresses the issue, and thus should be ready to be
considered for merging.  (I tested on top of UCS patches, have not
tested with the SECINFO use case).

We may also want to consider applying the last three patches to 3.6
and 3.5-stable.

---

Chuck Lever (5):
      SUNRPC: Share upcall pipes among an rpc_clnt's rpc_auth objects
      SUNRPC: Insert a shim under gss_create()
      SUNRPC: Prepare gss_pipes_dentries_create() for sharing pipe data
      SUNRPC: Use __func__ in dprintk() in auth_gss.c
      SUNRPC: Clean up dprintk messages in rpc_pipe.c


 include/linux/sunrpc/clnt.h    |    1 
 net/sunrpc/auth_gss/auth_gss.c |  290 ++++++++++++++++++++++++++++++----------
 net/sunrpc/clnt.c              |    1 
 net/sunrpc/rpc_pipe.c          |    8 +
 4 files changed, 222 insertions(+), 78 deletions(-)

--
Chuck Lever

[PATCH 1/5] SUNRPC: Clean up dprintk messages in rpc_pipe.c

[Chuck Lever]

Clean up: The blank space in front of the message must be spaces.
Tabs show up on the console as a graphical character.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---

 net/sunrpc/rpc_pipe.c |    8 ++++----
 1 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/net/sunrpc/rpc_pipe.c b/net/sunrpc/rpc_pipe.c
index 21fde99..80f5dd2 100644
--- a/net/sunrpc/rpc_pipe.c
+++ b/net/sunrpc/rpc_pipe.c
@@ -1119,8 +1119,8 @@ rpc_fill_super(struct super_block *sb, void *data, int silent)
 		return -ENOMEM;
 	if (rpc_populate(root, files, RPCAUTH_lockd, RPCAUTH_RootEOF, NULL))
 		return -ENOMEM;
-	dprintk("RPC:	sending pipefs MOUNT notification for net %p%s\n", net,
-								NET_NAME(net));
+	dprintk("RPC:       sending pipefs MOUNT notification for net %p%s\n",
+		net, NET_NAME(net));
 	sn->pipefs_sb = sb;
 	err = blocking_notifier_call_chain(&rpc_pipefs_notifier_list,
 					   RPC_PIPEFS_MOUNT,
@@ -1155,8 +1155,8 @@ static void rpc_kill_sb(struct super_block *sb)
 	sn->pipefs_sb = NULL;
 	mutex_unlock(&sn->pipefs_sb_lock);
 	put_net(net);
-	dprintk("RPC:	sending pipefs UMOUNT notification for net %p%s\n", net,
-								NET_NAME(net));
+	dprintk("RPC:       sending pipefs UMOUNT notification for net %p%s\n",
+		net, NET_NAME(net));
 	blocking_notifier_call_chain(&rpc_pipefs_notifier_list,
 					   RPC_PIPEFS_UMOUNT,
 					   sb);

[PATCH 2/5] SUNRPC: Use __func__ in dprintk() in auth_gss.c

[Chuck Lever]

Clean up: Some function names have changed, but debugging messages
were never updated.  Automate the construction of the function name
in debugging messages.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---

 net/sunrpc/auth_gss/auth_gss.c |   58 ++++++++++++++++++++--------------------
 1 files changed, 29 insertions(+), 29 deletions(-)

diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index 34c5220..909dc0c 100644
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -239,7 +239,7 @@ gss_fill_context(const void *p, const void *end, struct gss_cl_ctx *ctx, struct
 	}
 	return q;
 err:
-	dprintk("RPC:       gss_fill_context returning %ld\n", -PTR_ERR(p));
+	dprintk("RPC:       %s returning %ld\n", __func__, -PTR_ERR(p));
 	return p;
 }
 
@@ -301,10 +301,10 @@ __gss_find_upcall(struct rpc_pipe *pipe, uid_t uid)
 		if (pos->uid != uid)
 			continue;
 		atomic_inc(&pos->count);
-		dprintk("RPC:       gss_find_upcall found msg %p\n", pos);
+		dprintk("RPC:       %s found msg %p\n", __func__, pos);
 		return pos;
 	}
-	dprintk("RPC:       gss_find_upcall found nothing\n");
+	dprintk("RPC:       %s found nothing\n", __func__);
 	return NULL;
 }
 
@@ -507,8 +507,8 @@ gss_refresh_upcall(struct rpc_task *task)
 	struct rpc_pipe *pipe;
 	int err = 0;
 
-	dprintk("RPC: %5u gss_refresh_upcall for uid %u\n", task->tk_pid,
-								cred->cr_uid);
+	dprintk("RPC: %5u %s for uid %u\n",
+		task->tk_pid, __func__, cred->cr_uid);
 	gss_msg = gss_setup_upcall(task->tk_client, gss_auth, cred);
 	if (PTR_ERR(gss_msg) == -EAGAIN) {
 		/* XXX: warning on the first, under the assumption we
@@ -539,8 +539,8 @@ gss_refresh_upcall(struct rpc_task *task)
 	spin_unlock(&pipe->lock);
 	gss_release_msg(gss_msg);
 out:
-	dprintk("RPC: %5u gss_refresh_upcall for uid %u result %d\n",
-			task->tk_pid, cred->cr_uid, err);
+	dprintk("RPC: %5u %s for uid %u result %d\n",
+		task->tk_pid, __func__, cred->cr_uid, err);
 	return err;
 }
 
@@ -553,7 +553,7 @@ gss_create_upcall(struct gss_auth *gss_auth, struct gss_cred *gss_cred)
 	DEFINE_WAIT(wait);
 	int err = 0;
 
-	dprintk("RPC:       gss_upcall for uid %u\n", cred->cr_uid);
+	dprintk("RPC:       %s for uid %u\n", __func__, cred->cr_uid);
 retry:
 	gss_msg = gss_setup_upcall(gss_auth->client, gss_auth, cred);
 	if (PTR_ERR(gss_msg) == -EAGAIN) {
@@ -594,8 +594,8 @@ out_intr:
 	finish_wait(&gss_msg->waitqueue, &wait);
 	gss_release_msg(gss_msg);
 out:
-	dprintk("RPC:       gss_create_upcall for uid %u result %d\n",
-			cred->cr_uid, err);
+	dprintk("RPC:       %s for uid %u result %d\n",
+		__func__, cred->cr_uid, err);
 	return err;
 }
 
@@ -681,7 +681,7 @@ err_put_ctx:
 err:
 	kfree(buf);
 out:
-	dprintk("RPC:       gss_pipe_downcall returning %Zd\n", err);
+	dprintk("RPC:       %s returning %Zd\n", __func__, err);
 	return err;
 }
 
@@ -747,8 +747,8 @@ gss_pipe_destroy_msg(struct rpc_pipe_msg *msg)
 	struct gss_upcall_msg *gss_msg = container_of(msg, struct gss_upcall_msg, msg);
 
 	if (msg->errno < 0) {
-		dprintk("RPC:       gss_pipe_destroy_msg releasing msg %p\n",
-				gss_msg);
+		dprintk("RPC:       %s releasing msg %p\n",
+			__func__, gss_msg);
 		atomic_inc(&gss_msg->count);
 		gss_unhash_msg(gss_msg);
 		if (msg->errno == -ETIMEDOUT)
@@ -976,7 +976,7 @@ gss_destroying_context(struct rpc_cred *cred)
 static void
 gss_do_free_ctx(struct gss_cl_ctx *ctx)
 {
-	dprintk("RPC:       gss_free_ctx\n");
+	dprintk("RPC:       %s\n", __func__);
 
 	gss_delete_sec_context(&ctx->gc_gss_ctx);
 	kfree(ctx->gc_wire_ctx.data);
@@ -999,7 +999,7 @@ gss_free_ctx(struct gss_cl_ctx *ctx)
 static void
 gss_free_cred(struct gss_cred *gss_cred)
 {
-	dprintk("RPC:       gss_free_cred %p\n", gss_cred);
+	dprintk("RPC:       %s cred=%p\n", __func__, gss_cred);
 	kfree(gss_cred);
 }
 
@@ -1049,8 +1049,8 @@ gss_create_cred(struct rpc_auth *auth, struct auth_cred *acred, int flags)
 	struct gss_cred	*cred = NULL;
 	int err = -ENOMEM;
 
-	dprintk("RPC:       gss_create_cred for uid %d, flavor %d\n",
-		acred->uid, auth->au_flavor);
+	dprintk("RPC:       %s for uid %d, flavor %d\n",
+		__func__, acred->uid, auth->au_flavor);
 
 	if (!(cred = kzalloc(sizeof(*cred), GFP_NOFS)))
 		goto out_err;
@@ -1069,7 +1069,7 @@ gss_create_cred(struct rpc_auth *auth, struct auth_cred *acred, int flags)
 	return &cred->gc_base;
 
 out_err:
-	dprintk("RPC:       gss_create_cred failed with error %d\n", err);
+	dprintk("RPC:       %s failed with error %d\n", __func__, err);
 	return ERR_PTR(err);
 }
 
@@ -1127,7 +1127,7 @@ gss_marshal(struct rpc_task *task, __be32 *p)
 	struct kvec	iov;
 	struct xdr_buf	verf_buf;
 
-	dprintk("RPC: %5u gss_marshal\n", task->tk_pid);
+	dprintk("RPC: %5u %s\n", task->tk_pid, __func__);
 
 	*p++ = htonl(RPC_AUTH_GSS);
 	cred_len = p++;
@@ -1253,7 +1253,7 @@ gss_validate(struct rpc_task *task, __be32 *p)
 	u32		flav,len;
 	u32		maj_stat;
 
-	dprintk("RPC: %5u gss_validate\n", task->tk_pid);
+	dprintk("RPC: %5u %s\n", task->tk_pid, __func__);
 
 	flav = ntohl(*p++);
 	if ((len = ntohl(*p++)) > RPC_MAX_AUTH_SIZE)
@@ -1271,20 +1271,20 @@ gss_validate(struct rpc_task *task, __be32 *p)
 	if (maj_stat == GSS_S_CONTEXT_EXPIRED)
 		clear_bit(RPCAUTH_CRED_UPTODATE, &cred->cr_flags);
 	if (maj_stat) {
-		dprintk("RPC: %5u gss_validate: gss_verify_mic returned "
-				"error 0x%08x\n", task->tk_pid, maj_stat);
+		dprintk("RPC: %5u %s: gss_verify_mic returned error 0x%08x\n",
+			task->tk_pid, __func__, maj_stat);
 		goto out_bad;
 	}
 	/* We leave it to unwrap to calculate au_rslack. For now we just
 	 * calculate the length of the verifier: */
 	cred->cr_auth->au_verfsize = XDR_QUADLEN(len) + 2;
 	gss_put_ctx(ctx);
-	dprintk("RPC: %5u gss_validate: gss_verify_mic succeeded.\n",
-			task->tk_pid);
+	dprintk("RPC: %5u %s: gss_verify_mic succeeded.\n",
+			task->tk_pid, __func__);
 	return p + XDR_QUADLEN(len);
 out_bad:
 	gss_put_ctx(ctx);
-	dprintk("RPC: %5u gss_validate failed.\n", task->tk_pid);
+	dprintk("RPC: %5u %s failed.\n", task->tk_pid, __func__);
 	return NULL;
 }
 
@@ -1466,7 +1466,7 @@ gss_wrap_req(struct rpc_task *task,
 	struct gss_cl_ctx *ctx = gss_cred_get_ctx(cred);
 	int             status = -EIO;
 
-	dprintk("RPC: %5u gss_wrap_req\n", task->tk_pid);
+	dprintk("RPC: %5u %s\n", task->tk_pid, __func__);
 	if (ctx->gc_proc != RPC_GSS_PROC_DATA) {
 		/* The spec seems a little ambiguous here, but I think that not
 		 * wrapping context destruction requests makes the most sense.
@@ -1489,7 +1489,7 @@ gss_wrap_req(struct rpc_task *task,
 	}
 out:
 	gss_put_ctx(ctx);
-	dprintk("RPC: %5u gss_wrap_req returning %d\n", task->tk_pid, status);
+	dprintk("RPC: %5u %s returning %d\n", task->tk_pid, __func__, status);
 	return status;
 }
 
@@ -1604,8 +1604,8 @@ out_decode:
 	status = gss_unwrap_req_decode(decode, rqstp, p, obj);
 out:
 	gss_put_ctx(ctx);
-	dprintk("RPC: %5u gss_unwrap_resp returning %d\n", task->tk_pid,
-			status);
+	dprintk("RPC: %5u %s returning %d\n",
+		task->tk_pid, __func__, status);
 	return status;
 }
 

[PATCH 3/5] SUNRPC: Prepare gss_pipes_dentries_create() for sharing pipe data

[Chuck Lever]

We are about to make pipe data shareable. gss_pipes_dentries_create()
and gss_pipes_dentries_destroy() have to be more careful when dealing
with their dentry pointers.  Namely:

  1. When asked to create a dentry, don't write over possibly valid
     pointers, but instead preserve an existing dentry

  2. When unlinking a dentry, make sure the ->dentry pointer is NULL

  3. If creating a dentry fails, don't leak the ERR_PTR value, make
     sure the ->dentry pointer is NULL

2. and 3. are needed so that 1. can recognize when a new dentry
should be created.

This is clean up, mostly.  But it's necessary because unmounting a
pipe FS can cause pipe dentries, but not the pipe data, to be
released (via ->pipes_destroy).

The functions that manage pipe data for gss_create() have to be
prepared for finding an existing shared rpc_pipe_data object that
has a dentry, or doesn't, and then do the correct thing in either
case.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
Acked-by: Stanislav Kinsbursky <skinsbursky@parallels.com>
---

 net/sunrpc/auth_gss/auth_gss.c |   67 +++++++++++++++++++++++++++++-----------
 1 files changed, 49 insertions(+), 18 deletions(-)

diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index 909dc0c..947ab01 100644
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -757,42 +757,73 @@ gss_pipe_destroy_msg(struct rpc_pipe_msg *msg)
 	}
 }
 
+static void gss_pipe_dentry_destroy(struct rpc_pipe *pipe)
+{
+	if (pipe->dentry != NULL) {
+		rpc_unlink(pipe->dentry);
+		pipe->dentry = NULL;
+	}
+}
+
+/*
+ * Release existing pipe dentries, but leave the rpc_pipe data
+ * ready to receive fresh dentries if subsequently needed.
+ */
 static void gss_pipes_dentries_destroy(struct rpc_auth *auth)
 {
 	struct gss_auth *gss_auth;
 
 	gss_auth = container_of(auth, struct gss_auth, rpc_auth);
-	if (gss_auth->pipe[0]->dentry)
-		rpc_unlink(gss_auth->pipe[0]->dentry);
-	if (gss_auth->pipe[1]->dentry)
-		rpc_unlink(gss_auth->pipe[1]->dentry);
+
+	dprintk("RPC:       %s auth=%p, clnt=%p\n",
+		__func__, auth, gss_auth->client);
+
+	gss_pipe_dentry_destroy(gss_auth->pipe[0]);
+	gss_pipe_dentry_destroy(gss_auth->pipe[1]);
 }
 
+/*
+ * On success, zero is returned and both elements of the rpc_pipe
+ * array are populated with valid dentries.  Otherwise an error value
+ * is returned, the dentries are freed, and both ->dentry pointers are
+ * set to NULL.
+ */
 static int gss_pipes_dentries_create(struct rpc_auth *auth)
 {
-	int err;
+	int err = 0;
 	struct gss_auth *gss_auth;
 	struct rpc_clnt *clnt;
 
 	gss_auth = container_of(auth, struct gss_auth, rpc_auth);
 	clnt = gss_auth->client;
 
-	gss_auth->pipe[1]->dentry = rpc_mkpipe_dentry(clnt->cl_dentry,
-						      "gssd",
-						      clnt, gss_auth->pipe[1]);
-	if (IS_ERR(gss_auth->pipe[1]->dentry))
-		return PTR_ERR(gss_auth->pipe[1]->dentry);
-	gss_auth->pipe[0]->dentry = rpc_mkpipe_dentry(clnt->cl_dentry,
-						      gss_auth->mech->gm_name,
-						      clnt, gss_auth->pipe[0]);
-	if (IS_ERR(gss_auth->pipe[0]->dentry)) {
-		err = PTR_ERR(gss_auth->pipe[0]->dentry);
-		goto err_unlink_pipe_1;
+	if (gss_auth->pipe[1]->dentry == NULL) {
+		gss_auth->pipe[1]->dentry =
+			rpc_mkpipe_dentry(clnt->cl_dentry, "gssd", clnt,
+							gss_auth->pipe[1]);
+		if (IS_ERR(gss_auth->pipe[1]->dentry)) {
+			err = PTR_ERR(gss_auth->pipe[1]->dentry);
+			gss_auth->pipe[1]->dentry = NULL;
+			goto out_err;
+		}
 	}
+
+	if (gss_auth->pipe[0]->dentry == NULL) {
+		gss_auth->pipe[0]->dentry =
+			rpc_mkpipe_dentry(clnt->cl_dentry,
+						gss_auth->mech->gm_name,
+						clnt, gss_auth->pipe[0]);
+		if (IS_ERR(gss_auth->pipe[0]->dentry)) {
+			err = PTR_ERR(gss_auth->pipe[0]->dentry);
+			gss_auth->pipe[0]->dentry = NULL;
+			goto out_err;
+		}
+	}
+
 	return 0;
 
-err_unlink_pipe_1:
-	rpc_unlink(gss_auth->pipe[1]->dentry);
+out_err:
+	gss_pipes_dentries_destroy(auth);
 	return err;
 }
 

[PATCH 4/5] SUNRPC: Insert a shim under gss_create()

[Chuck Lever]

We want to share rpc_pipe data between GSS-flavored rpc_auth objects.
To do this, create a shim in gss_create() and gss_free() where the
pipe data will be looked up, created, and destroyed.  The unused
arguments anticipate a per-RPC client cache of pipes.

This is a refactoring change which shouldn't have a functional
effect.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
Acked-by: Stanislav Kinsbursky <skinsbursky@parallels.com>
---

 net/sunrpc/auth_gss/auth_gss.c |   28 +++++++++++++++++++---------
 1 files changed, 19 insertions(+), 9 deletions(-)

diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index 947ab01..82a3156 100644
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -857,6 +857,18 @@ static int gss_pipes_dentries_create_net(struct rpc_clnt *clnt,
 	return err;
 }
 
+static struct rpc_pipe *gss_mkpipe_data(struct rpc_clnt *clnt,
+					const struct rpc_pipe_ops *ops,
+					char *name)
+{
+	return rpc_mkpipe_data(ops, RPC_PIPE_WAIT_FOR_OPEN);
+}
+
+static void gss_destroy_pipe_data(struct rpc_clnt *clnt, struct rpc_pipe *pipe)
+{
+	rpc_destroy_pipe_data(pipe);
+}
+
 /*
  * NOTE: we have the opportunity to use different
  * parameters based on the input flavor (which must be a pseudoflavor)
@@ -899,15 +911,13 @@ gss_create(struct rpc_clnt *clnt, rpc_authflavor_t flavor)
 	 * that we supported only the old pipe.  So we instead create
 	 * the new pipe first.
 	 */
-	gss_auth->pipe[1] = rpc_mkpipe_data(&gss_upcall_ops_v1,
-					    RPC_PIPE_WAIT_FOR_OPEN);
+	gss_auth->pipe[1] = gss_mkpipe_data(clnt, &gss_upcall_ops_v1, "gssd");
 	if (IS_ERR(gss_auth->pipe[1])) {
 		err = PTR_ERR(gss_auth->pipe[1]);
 		goto err_put_mech;
 	}
-
-	gss_auth->pipe[0] = rpc_mkpipe_data(&gss_upcall_ops_v0,
-					    RPC_PIPE_WAIT_FOR_OPEN);
+	gss_auth->pipe[0] = gss_mkpipe_data(clnt, &gss_upcall_ops_v0,
+						gss_auth->mech->gm_name);
 	if (IS_ERR(gss_auth->pipe[0])) {
 		err = PTR_ERR(gss_auth->pipe[0]);
 		goto err_destroy_pipe_1;
@@ -923,9 +933,9 @@ gss_create(struct rpc_clnt *clnt, rpc_authflavor_t flavor)
 err_unlink_pipes:
 	gss_pipes_dentries_destroy_net(clnt, auth);
 err_destroy_pipe_0:
-	rpc_destroy_pipe_data(gss_auth->pipe[0]);
+	gss_destroy_pipe_data(clnt, gss_auth->pipe[0]);
 err_destroy_pipe_1:
-	rpc_destroy_pipe_data(gss_auth->pipe[1]);
+	gss_destroy_pipe_data(clnt, gss_auth->pipe[1]);
 err_put_mech:
 	gss_mech_put(gss_auth->mech);
 err_free:
@@ -939,8 +949,8 @@ static void
 gss_free(struct gss_auth *gss_auth)
 {
 	gss_pipes_dentries_destroy_net(gss_auth->client, &gss_auth->rpc_auth);
-	rpc_destroy_pipe_data(gss_auth->pipe[0]);
-	rpc_destroy_pipe_data(gss_auth->pipe[1]);
+	gss_destroy_pipe_data(gss_auth->client, gss_auth->pipe[0]);
+	gss_destroy_pipe_data(gss_auth->client, gss_auth->pipe[1]);
 	gss_mech_put(gss_auth->mech);
 
 	kfree(gss_auth);

[PATCH 5/5] SUNRPC: Share upcall pipes among an rpc_clnt's rpc_auth objects

[Chuck Lever]

An ULP is supposed to be able to replace a GSS rpc_auth object with
another GSS rpc_auth object using rpcauth_create().  However,
rpcauth_create() in 3.5 reliably fails with -EEXIST in this case.
This is because when gss_create() attempts to create the upcall pipes,
sometimes they are already there.  For example if a pipe FS mount
event occurs, or a previous GSS flavor was in use for this rpc_clnt.

The ->pipes_destroy method works only on whatever rpc_auth is
associated with the rpc_clnt when it is destroyed.  But there is no
guarantee that this is the same rpc_auth that was created when
the rpc_clnt was created.  Thus cached pipe data must be reference
counted and destroyed when no more gss_auth's are using it.

Pipe dentries appear to be created and deleted independently of the
rpc_pipe data.  We must allow this to continue in order that mounting
and unmounting the pipe FS can work.

In the meantime, we need to provide a way for gss_create() to find
an existing pipe of a particular name so it may be shared.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
Acked-by: Stanislav Kinsbursky <skinsbursky@parallels.com>
---

 include/linux/sunrpc/clnt.h    |    1 
 net/sunrpc/auth_gss/auth_gss.c |  141 ++++++++++++++++++++++++++++++++++------
 net/sunrpc/clnt.c              |    1 
 3 files changed, 123 insertions(+), 20 deletions(-)

diff --git a/include/linux/sunrpc/clnt.h b/include/linux/sunrpc/clnt.h
index 523547e..31857e0 100644
--- a/include/linux/sunrpc/clnt.h
+++ b/include/linux/sunrpc/clnt.h
@@ -62,6 +62,7 @@ struct rpc_clnt {
 	struct rpc_timeout	cl_timeout_default;
 	const struct rpc_program *cl_program;
 	char			*cl_principal;	/* target to authenticate to */
+	struct list_head	cl_pipes;	/* clnt's upcall pipes */
 };
 
 /*
diff --git a/net/sunrpc/auth_gss/auth_gss.c b/net/sunrpc/auth_gss/auth_gss.c
index 82a3156..b85ebe8 100644
--- a/net/sunrpc/auth_gss/auth_gss.c
+++ b/net/sunrpc/auth_gss/auth_gss.c
@@ -84,6 +84,14 @@ struct gss_auth {
 	struct rpc_pipe *pipe[2];
 };
 
+struct gss_cached_pipe {
+	struct list_head	gp_list;
+	unsigned int		gp_ref;
+	char			*gp_name;
+	struct rpc_pipe		*gp_pipe;
+};
+DEFINE_MUTEX(gss_pipe_mutex);
+
 /* pipe_version >= 0 if and only if someone has a pipe open. */
 static int pipe_version = -1;
 static atomic_t pipe_users = ATOMIC_INIT(0);
@@ -765,6 +773,20 @@ static void gss_pipe_dentry_destroy(struct rpc_pipe *pipe)
 	}
 }
 
+static void gss_pipe_dentry_destroy_net(struct rpc_clnt *clnt,
+					struct rpc_pipe *pipe)
+{
+	struct net *net = rpc_net_ns(clnt);
+	struct super_block *sb;
+
+	sb = rpc_get_sb_net(net);
+	if (sb) {
+		if (clnt->cl_dentry)
+			gss_pipe_dentry_destroy(pipe);
+		rpc_put_sb_net(net);
+	}
+}
+
 /*
  * Release existing pipe dentries, but leave the rpc_pipe data
  * ready to receive fresh dentries if subsequently needed.
@@ -827,20 +849,6 @@ out_err:
 	return err;
 }
 
-static void gss_pipes_dentries_destroy_net(struct rpc_clnt *clnt,
-					   struct rpc_auth *auth)
-{
-	struct net *net = rpc_net_ns(clnt);
-	struct super_block *sb;
-
-	sb = rpc_get_sb_net(net);
-	if (sb) {
-		if (clnt->cl_dentry)
-			gss_pipes_dentries_destroy(auth);
-		rpc_put_sb_net(net);
-	}
-}
-
 static int gss_pipes_dentries_create_net(struct rpc_clnt *clnt,
 					 struct rpc_auth *auth)
 {
@@ -857,16 +865,111 @@ static int gss_pipes_dentries_create_net(struct rpc_clnt *clnt,
 	return err;
 }
 
+static struct gss_cached_pipe *gss_alloc_cached_pipe(char *name)
+{
+	struct gss_cached_pipe *cp;
+
+	cp = kmalloc(sizeof(*cp), GFP_KERNEL);
+	if (cp == NULL)
+		return NULL;
+
+	cp->gp_name = kstrdup(name, GFP_KERNEL);
+	if (cp->gp_name == NULL) {
+		kfree(cp);
+		return NULL;
+	}
+
+	INIT_LIST_HEAD(&cp->gp_list);
+	cp->gp_ref = 1;
+	return cp;
+}
+
+static void gss_free_cached_pipe(struct gss_cached_pipe *cp)
+{
+	kfree(cp->gp_name);
+	kfree(cp);
+}
+
+/*
+ * Returns a fresh or cached pipe data object.  A cached pipe
+ * data object may already have a dentry attached to it.  If an
+ * object cannot be found or created, an ERR_PTR is returned.
+ */
 static struct rpc_pipe *gss_mkpipe_data(struct rpc_clnt *clnt,
 					const struct rpc_pipe_ops *ops,
 					char *name)
 {
-	return rpc_mkpipe_data(ops, RPC_PIPE_WAIT_FOR_OPEN);
+	struct gss_cached_pipe *cp;
+	struct rpc_pipe *pipe;
+
+	mutex_lock(&gss_pipe_mutex);
+
+	list_for_each_entry(cp, &clnt->cl_pipes, gp_list) {
+		if (strcmp(cp->gp_name, name) == 0) {
+			dprintk("RPC:       %s found '%s' for clnt %p: %p\n",
+				__func__, name, clnt, cp->gp_pipe);
+			cp->gp_ref++;
+			pipe = cp->gp_pipe;
+			goto out;
+		}
+	}
+
+	pipe = ERR_PTR(-ENOMEM);
+	cp = gss_alloc_cached_pipe(name);
+	if (cp == NULL)
+		goto out;
+
+	pipe = rpc_mkpipe_data(ops, RPC_PIPE_WAIT_FOR_OPEN);
+	if (IS_ERR(pipe)) {
+		gss_free_cached_pipe(cp);
+		goto out;
+	}
+
+	dprintk("RPC:       %s created '%s' for clnt %p: %p\n",
+		__func__, name, clnt, pipe);
+	cp->gp_pipe = pipe;
+	list_add(&cp->gp_list, &clnt->cl_pipes);
+
+out:
+	mutex_unlock(&gss_pipe_mutex);
+	return pipe;
 }
 
+/*
+ * Decrements a cached pipes reference count, and releases it if
+ * the count goes to zero.  Associated dentry is freed if present.
+ */
 static void gss_destroy_pipe_data(struct rpc_clnt *clnt, struct rpc_pipe *pipe)
 {
-	rpc_destroy_pipe_data(pipe);
+	struct gss_cached_pipe *cp;
+
+	mutex_lock(&gss_pipe_mutex);
+
+	list_for_each_entry(cp, &clnt->cl_pipes, gp_list) {
+		if (cp->gp_pipe == pipe)
+			goto found;
+	}
+
+	dprintk("RPC:       %s missing cache for pipe %p for clnt %p\n",
+		__func__, pipe, clnt);
+	WARN_ON(true);
+
+	mutex_unlock(&gss_pipe_mutex);
+	return;
+
+found:
+	if (--cp->gp_ref == 0) {
+		dprintk("RPC:       %s destroying '%s' (%p) for clnt %p\n",
+			__func__, cp->gp_name, pipe, clnt);
+		gss_pipe_dentry_destroy_net(clnt, pipe);
+		rpc_destroy_pipe_data(pipe);
+		list_del(&cp->gp_list);
+		gss_free_cached_pipe(cp);
+	} else
+		dprintk("RPC:       %s leaving '%s' (%p) for clnt %p\n",
+			__func__, cp->gp_name, pipe, clnt);
+
+	mutex_unlock(&gss_pipe_mutex);
 }
 
 /*
@@ -925,13 +1028,12 @@ gss_create(struct rpc_clnt *clnt, rpc_authflavor_t flavor)
 	err = gss_pipes_dentries_create_net(clnt, auth);
 	if (err)
 		goto err_destroy_pipe_0;
+
 	err = rpcauth_init_credcache(auth);
 	if (err)
-		goto err_unlink_pipes;
+		goto err_destroy_pipe_0;
 
 	return auth;
-err_unlink_pipes:
-	gss_pipes_dentries_destroy_net(clnt, auth);
 err_destroy_pipe_0:
 	gss_destroy_pipe_data(clnt, gss_auth->pipe[0]);
 err_destroy_pipe_1:
@@ -948,7 +1050,6 @@ out_dec:
 static void
 gss_free(struct gss_auth *gss_auth)
 {
-	gss_pipes_dentries_destroy_net(gss_auth->client, &gss_auth->rpc_auth);
 	gss_destroy_pipe_data(gss_auth->client, gss_auth->pipe[0]);
 	gss_destroy_pipe_data(gss_auth->client, gss_auth->pipe[1]);
 	gss_mech_put(gss_auth->mech);
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c
index f56f045..7bf6ed1 100644
--- a/net/sunrpc/clnt.c
+++ b/net/sunrpc/clnt.c
@@ -349,6 +349,7 @@ static struct rpc_clnt * rpc_new_client(const struct rpc_create_args *args, stru
 		if (!clnt->cl_principal)
 			goto out_no_principal;
 	}
+	INIT_LIST_HEAD(&clnt->cl_pipes);
 
 	atomic_set(&clnt->cl_count, 1);