Merge tag 'nfs-for-4.6-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs

Pull NFS client updates from Trond Myklebust:
 "Highlights include:

  Features:
   - Add support for multiple NFSv4.1 callbacks in flight
   - Initial patchset for RPC multipath support
   - Adapt RPC/RDMA to use the new completion queue API

  Bugfixes and cleanups:
   - nfs4: nfs4_ff_layout_prepare_ds should return NULL if connection failed
   - Cleanups to remove nfs_inode_dio_wait and nfs4_file_fsync
   - Fix RPC/RDMA credit accounting
   - Properly handle RDMA_ERROR replies
   - xprtrdma: Do not wait if ib_post_send() fails
   - xprtrdma: Segment head and tail XDR buffers on page boundaries
   - xprtrdma cleanups for dprintk, physical_op_map and unused macros"

* tag 'nfs-for-4.6-1' of git://git.linux-nfs.org/projects/trondmy/linux-nfs: (35 commits)
  nfs/blocklayout: make sure making a aligned read request
  nfs4: nfs4_ff_layout_prepare_ds should return NULL if connection failed
  nfs: remove nfs_inode_dio_wait
  nfs: remove nfs4_file_fsync
  xprtrdma: Use new CQ API for RPC-over-RDMA client send CQs
  xprtrdma: Use an anonymous union in struct rpcrdma_mw
  xprtrdma: Use new CQ API for RPC-over-RDMA client receive CQs
  xprtrdma: Serialize credit accounting again
  xprtrdma: Properly handle RDMA_ERROR replies
  rpcrdma: Add RPCRDMA_HDRLEN_ERR
  xprtrdma: Do not wait if ib_post_send() fails
  xprtrdma: Segment head and tail XDR buffers on page boundaries
  xprtrdma: Clean up dprintk format string containing a newline
  xprtrdma: Clean up physical_op_map()
  xprtrdma: Clean up unused RPCRDMA_INLINE_PAD_THRESH macro
  NFS add callback_ops to nfs4_proc_bind_conn_to_session_callback
  pnfs/NFSv4.1: Add multipath capabilities to pNFS flexfiles servers over NFSv3
  SUNRPC: Allow addition of new transports to a struct rpc_clnt
  NFSv4.1: nfs4_proc_bind_conn_to_session must iterate over all connections
  SUNRPC: Make NFS swap work with multipath
  ...

6 files changed, 120 insertions, 35 deletions

diff --git a/include/linux/sunrpc/clnt.h b/include/linux/sunrpc/clnt.h
index 131032f15cc1..9a7ddbaf116e 100644
--- a/include/linux/sunrpc/clnt.h
+++ b/include/linux/sunrpc/clnt.h
@@ -25,6 +25,7 @@
 #include <asm/signal.h>
 #include <linux/path.h>
 #include <net/ipv6.h>
+#include <linux/sunrpc/xprtmultipath.h>
 
 struct rpc_inode;
 
@@ -67,6 +68,7 @@ struct rpc_clnt {
 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
         struct dentry           *cl_debugfs;    /* debugfs directory */
 #endif
+        struct rpc_xprt_iter    cl_xpi;
 };
 
 /*
@@ -139,7 +141,6 @@ struct rpc_clnt *rpc_create_xprt(struct rpc_create_args *args,
                                         struct rpc_xprt *xprt);
 struct rpc_clnt *rpc_bind_new_program(struct rpc_clnt *,
                                 const struct rpc_program *, u32);
-void rpc_task_reset_client(struct rpc_task *task, struct rpc_clnt *clnt);
 struct rpc_clnt *rpc_clone_client(struct rpc_clnt *);
 struct rpc_clnt *rpc_clone_client_set_auth(struct rpc_clnt *,
                                 rpc_authflavor_t);
@@ -181,6 +182,21 @@ size_t		rpc_peeraddr(struct rpc_clnt *, struct sockaddr *, size_t);
 const char      *rpc_peeraddr2str(struct rpc_clnt *, enum rpc_display_format_t);
 int             rpc_localaddr(struct rpc_clnt *, struct sockaddr *, size_t);
 
+int             rpc_clnt_iterate_for_each_xprt(struct rpc_clnt *clnt,
+                        int (*fn)(struct rpc_clnt *, struct rpc_xprt *, void *),
+                        void *data);
+
+int             rpc_clnt_test_and_add_xprt(struct rpc_clnt *clnt,
+                        struct rpc_xprt_switch *xps,
+                        struct rpc_xprt *xprt,
+                        void *dummy);
+int             rpc_clnt_add_xprt(struct rpc_clnt *, struct xprt_create *,
+                        int (*setup)(struct rpc_clnt *,
+                                struct rpc_xprt_switch *,
+                                struct rpc_xprt *,
+                                void *),
+                        void *data);
+
 const char *rpc_proc_name(const struct rpc_task *task);
 #endif /* __KERNEL__ */
 #endif /* _LINUX_SUNRPC_CLNT_H */
diff --git a/include/linux/sunrpc/rpc_rdma.h b/include/linux/sunrpc/rpc_rdma.h
index f33c5a4d6fe4..3b1ff38f0c37 100644
--- a/include/linux/sunrpc/rpc_rdma.h
+++ b/include/linux/sunrpc/rpc_rdma.h
@@ -93,6 +93,12 @@ struct rpcrdma_msg {
                         __be32 rm_pempty[3];    /* 3 empty chunk lists */
                 } rm_padded;
 
+                struct {
+                        __be32 rm_err;
+                        __be32 rm_vers_low;
+                        __be32 rm_vers_high;
+                } rm_error;
+
                 __be32 rm_chunks[0];    /* read, write and reply chunks */
 
         } rm_body;
@@ -102,17 +108,13 @@ struct rpcrdma_msg {
  * Smallest RPC/RDMA header: rm_xid through rm_type, then rm_nochunks
  */
 #define RPCRDMA_HDRLEN_MIN      (sizeof(__be32) * 7)
+#define RPCRDMA_HDRLEN_ERR      (sizeof(__be32) * 5)
 
 enum rpcrdma_errcode {
         ERR_VERS = 1,
         ERR_CHUNK = 2
 };
 
-struct rpcrdma_err_vers {
-        uint32_t rdma_vers_low; /* Version range supported by peer */
-        uint32_t rdma_vers_high;
-};
-
 enum rpcrdma_proc {
         RDMA_MSG = 0,           /* An RPC call or reply msg */
         RDMA_NOMSG = 1,         /* An RPC call or reply msg - separate body */
diff --git a/include/linux/sunrpc/sched.h b/include/linux/sunrpc/sched.h
index d703f0ef37d8..05a1809c44d9 100644
--- a/include/linux/sunrpc/sched.h
+++ b/include/linux/sunrpc/sched.h
@@ -42,40 +42,43 @@ struct rpc_wait {
  */
 struct rpc_task {
         atomic_t                tk_count;       /* Reference count */
+        int                     tk_status;      /* result of last operation */
         struct list_head        tk_task;        /* global list of tasks */
-        struct rpc_clnt *       tk_client;      /* RPC client */
-        struct rpc_rqst *       tk_rqstp;       /* RPC request */
-
-        /*
-         * RPC call state
-         */
-        struct rpc_message      tk_msg;         /* RPC call info */
 
         /*
          * callback     to be executed after waking up
          * action       next procedure for async tasks
-         * tk_ops       caller callbacks
          */
         void                    (*tk_callback)(struct rpc_task *);
         void                    (*tk_action)(struct rpc_task *);
-        const struct rpc_call_ops *tk_ops;
-        void *                  tk_calldata;
 
         unsigned long           tk_timeout;     /* timeout for rpc_sleep() */
         unsigned long           tk_runstate;    /* Task run status */
-        struct workqueue_struct *tk_workqueue;  /* Normally rpciod, but could
-                                                 * be any workqueue
-                                                 */
+
         struct rpc_wait_queue   *tk_waitqueue;  /* RPC wait queue we're on */
         union {
                 struct work_struct      tk_work;        /* Async task work queue */
                 struct rpc_wait         tk_wait;        /* RPC wait */
         } u;
 
+        /*
+         * RPC call state
+         */
+        struct rpc_message      tk_msg;         /* RPC call info */
+        void *                  tk_calldata;    /* Caller private data */
+        const struct rpc_call_ops *tk_ops;      /* Caller callbacks */
+
+        struct rpc_clnt *       tk_client;      /* RPC client */
+        struct rpc_xprt *       tk_xprt;        /* Transport */
+
+        struct rpc_rqst *       tk_rqstp;       /* RPC request */
+
+        struct workqueue_struct *tk_workqueue;  /* Normally rpciod, but could
+                                                 * be any workqueue
+                                                 */
         ktime_t                 tk_start;       /* RPC task init timestamp */
 
         pid_t                   tk_owner;       /* Process id for batching tasks */
-        int                     tk_status;      /* result of last operation */
         unsigned short          tk_flags;       /* misc flags */
         unsigned short          tk_timeouts;    /* maj timeouts */
 
@@ -100,6 +103,7 @@ struct rpc_call_ops {
 struct rpc_task_setup {
         struct rpc_task *task;
         struct rpc_clnt *rpc_client;
+        struct rpc_xprt *rpc_xprt;
         const struct rpc_message *rpc_message;
         const struct rpc_call_ops *callback_ops;
         void *callback_data;
diff --git a/include/linux/sunrpc/xprt.h b/include/linux/sunrpc/xprt.h
index 69ef5b3ab038..fb0d212e0d3a 100644
--- a/include/linux/sunrpc/xprt.h
+++ b/include/linux/sunrpc/xprt.h
@@ -13,6 +13,7 @@
 #include <linux/socket.h>
 #include <linux/in.h>
 #include <linux/ktime.h>
+#include <linux/kref.h>
 #include <linux/sunrpc/sched.h>
 #include <linux/sunrpc/xdr.h>
 #include <linux/sunrpc/msg_prot.h>
@@ -166,7 +167,7 @@ enum xprt_transports {
 };
 
 struct rpc_xprt {
-        atomic_t                count;          /* Reference count */
+        struct kref             kref;           /* Reference count */
         struct rpc_xprt_ops *   ops;            /* transport methods */
 
         const struct rpc_timeout *timeout;      /* timeout parms */
@@ -197,6 +198,11 @@ struct rpc_xprt {
         unsigned int            bind_index;     /* bind function index */
 
         /*
+         * Multipath
+         */
+        struct list_head        xprt_switch;
+
+        /*
          * Connection of transports
          */
         unsigned long           bind_timeout,
@@ -256,6 +262,7 @@ struct rpc_xprt {
         struct dentry           *debugfs;               /* debugfs directory */
         atomic_t                inject_disconnect;
 #endif
+        struct rcu_head         rcu;
 };
 
 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
@@ -318,24 +325,13 @@ int			xprt_adjust_timeout(struct rpc_rqst *req);
 void                    xprt_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task);
 void                    xprt_release_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task);
 void                    xprt_release(struct rpc_task *task);
+struct rpc_xprt *       xprt_get(struct rpc_xprt *xprt);
 void                    xprt_put(struct rpc_xprt *xprt);
 struct rpc_xprt *       xprt_alloc(struct net *net, size_t size,
                                 unsigned int num_prealloc,
                                 unsigned int max_req);
 void                    xprt_free(struct rpc_xprt *);
 
-/**
- * xprt_get - return a reference to an RPC transport.
- * @xprt: pointer to the transport
- *
- */
-static inline struct rpc_xprt *xprt_get(struct rpc_xprt *xprt)
-{
-        if (atomic_inc_not_zero(&xprt->count))
-                return xprt;
-        return NULL;
-}
-
 static inline __be32 *xprt_skip_transport_header(struct rpc_xprt *xprt, __be32 *p)
 {
         return p + xprt->tsh_size;
diff --git a/include/linux/sunrpc/xprtmultipath.h b/include/linux/sunrpc/xprtmultipath.h
new file mode 100644
index 000000000000..5a9acffa41be
--- /dev/null
+++ b/include/linux/sunrpc/xprtmultipath.h
@@ -0,0 +1,69 @@
+/*
+ * RPC client multipathing definitions
+ *
+ * Copyright (c) 2015, 2016, Primary Data, Inc. All rights reserved.
+ *
+ * Trond Myklebust <trond.myklebust@primarydata.com>
+ */
+#ifndef _NET_SUNRPC_XPRTMULTIPATH_H
+#define _NET_SUNRPC_XPRTMULTIPATH_H
+
+struct rpc_xprt_iter_ops;
+struct rpc_xprt_switch {
+        spinlock_t              xps_lock;
+        struct kref             xps_kref;
+
+        unsigned int            xps_nxprts;
+        struct list_head        xps_xprt_list;
+
+        struct net *            xps_net;
+
+        const struct rpc_xprt_iter_ops *xps_iter_ops;
+
+        struct rcu_head         xps_rcu;
+};
+
+struct rpc_xprt_iter {
+        struct rpc_xprt_switch __rcu *xpi_xpswitch;
+        struct rpc_xprt *       xpi_cursor;
+
+        const struct rpc_xprt_iter_ops *xpi_ops;
+};
+
+
+struct rpc_xprt_iter_ops {
+        void (*xpi_rewind)(struct rpc_xprt_iter *);
+        struct rpc_xprt *(*xpi_xprt)(struct rpc_xprt_iter *);
+        struct rpc_xprt *(*xpi_next)(struct rpc_xprt_iter *);
+};
+
+extern struct rpc_xprt_switch *xprt_switch_alloc(struct rpc_xprt *xprt,
+                gfp_t gfp_flags);
+
+extern struct rpc_xprt_switch *xprt_switch_get(struct rpc_xprt_switch *xps);
+extern void xprt_switch_put(struct rpc_xprt_switch *xps);
+
+extern void rpc_xprt_switch_set_roundrobin(struct rpc_xprt_switch *xps);
+
+extern void rpc_xprt_switch_add_xprt(struct rpc_xprt_switch *xps,
+                struct rpc_xprt *xprt);
+extern void rpc_xprt_switch_remove_xprt(struct rpc_xprt_switch *xps,
+                struct rpc_xprt *xprt);
+
+extern void xprt_iter_init(struct rpc_xprt_iter *xpi,
+                struct rpc_xprt_switch *xps);
+
+extern void xprt_iter_init_listall(struct rpc_xprt_iter *xpi,
+                struct rpc_xprt_switch *xps);
+
+extern void xprt_iter_destroy(struct rpc_xprt_iter *xpi);
+
+extern struct rpc_xprt_switch *xprt_iter_xchg_switch(
+                struct rpc_xprt_iter *xpi,
+                struct rpc_xprt_switch *newswitch);
+
+extern struct rpc_xprt *xprt_iter_xprt(struct rpc_xprt_iter *xpi);
+extern struct rpc_xprt *xprt_iter_get_xprt(struct rpc_xprt_iter *xpi);
+extern struct rpc_xprt *xprt_iter_get_next(struct rpc_xprt_iter *xpi);
+
+#endif
diff --git a/include/linux/sunrpc/xprtrdma.h b/include/linux/sunrpc/xprtrdma.h
index b7b279b54504..767190b01363 100644
--- a/include/linux/sunrpc/xprtrdma.h
+++ b/include/linux/sunrpc/xprtrdma.h
@@ -54,8 +54,6 @@
 
 #define RPCRDMA_DEF_INLINE  (1024)      /* default inline max */
 
-#define RPCRDMA_INLINE_PAD_THRESH  (512)/* payload threshold to pad (bytes) */
-
 /* Memory registration strategies, by number.
  * This is part of a kernel / user space API. Do not remove. */
 enum rpcrdma_memreg {