Merge branch 'mpi-master' into wip-k-fmlpwip-k-fmlp

Conflicts:
	litmus/sched_cedf.c

1 files changed, 92 insertions, 74 deletions

diff --git a/net/sunrpc/svc_xprt.c b/net/sunrpc/svc_xprt.c
index cbc084939dd8..ab86b7927f84 100644
--- a/net/sunrpc/svc_xprt.c
+++ b/net/sunrpc/svc_xprt.c
@@ -5,7 +5,6 @@
  */
 
 #include <linux/sched.h>
-#include <linux/smp_lock.h>
 #include <linux/errno.h>
 #include <linux/freezer.h>
 #include <linux/kthread.h>
@@ -14,6 +13,7 @@
 #include <linux/sunrpc/stats.h>
 #include <linux/sunrpc/svc_xprt.h>
 #include <linux/sunrpc/svcsock.h>
+#include <linux/sunrpc/xprt.h>
 
 #define RPCDBG_FACILITY RPCDBG_SVCXPRT
 
@@ -100,16 +100,14 @@ EXPORT_SYMBOL_GPL(svc_unreg_xprt_class);
  */
 int svc_print_xprts(char *buf, int maxlen)
 {
-        struct list_head *le;
+        struct svc_xprt_class *xcl;
         char tmpstr[80];
         int len = 0;
         buf[0] = '\0';
 
         spin_lock(&svc_xprt_class_lock);
-        list_for_each(le, &svc_xprt_class_list) {
+        list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) {
                 int slen;
-                struct svc_xprt_class *xcl =
-                        list_entry(le, struct svc_xprt_class, xcl_list);
 
                 sprintf(tmpstr, "%s %d\n", xcl->xcl_name, xcl->xcl_max_payload);
                 slen = strlen(tmpstr);
@@ -128,9 +126,12 @@ static void svc_xprt_free(struct kref *kref)
         struct svc_xprt *xprt =
                 container_of(kref, struct svc_xprt, xpt_ref);
         struct module *owner = xprt->xpt_class->xcl_owner;
-        if (test_bit(XPT_CACHE_AUTH, &xprt->xpt_flags) &&
-            xprt->xpt_auth_cache != NULL)
-                svcauth_unix_info_release(xprt->xpt_auth_cache);
+        if (test_bit(XPT_CACHE_AUTH, &xprt->xpt_flags))
+                svcauth_unix_info_release(xprt);
+        put_net(xprt->xpt_net);
+        /* See comment on corresponding get in xs_setup_bc_tcp(): */
+        if (xprt->xpt_bc_xprt)
+                xprt_put(xprt->xpt_bc_xprt);
         xprt->xpt_ops->xpo_free(xprt);
         module_put(owner);
 }
@@ -156,15 +157,18 @@ void svc_xprt_init(struct svc_xprt_class *xcl, struct svc_xprt *xprt,
         INIT_LIST_HEAD(&xprt->xpt_list);
         INIT_LIST_HEAD(&xprt->xpt_ready);
         INIT_LIST_HEAD(&xprt->xpt_deferred);
+        INIT_LIST_HEAD(&xprt->xpt_users);
         mutex_init(&xprt->xpt_mutex);
         spin_lock_init(&xprt->xpt_lock);
         set_bit(XPT_BUSY, &xprt->xpt_flags);
         rpc_init_wait_queue(&xprt->xpt_bc_pending, "xpt_bc_pending");
+        xprt->xpt_net = get_net(&init_net);
 }
 EXPORT_SYMBOL_GPL(svc_xprt_init);
 
 static struct svc_xprt *__svc_xpo_create(struct svc_xprt_class *xcl,
                                          struct svc_serv *serv,
+                                         struct net *net,
                                          const int family,
                                          const unsigned short port,
                                          int flags)
@@ -199,12 +203,12 @@ static struct svc_xprt *__svc_xpo_create(struct svc_xprt_class *xcl,
                 return ERR_PTR(-EAFNOSUPPORT);
         }
 
-        return xcl->xcl_ops->xpo_create(serv, sap, len, flags);
+        return xcl->xcl_ops->xpo_create(serv, net, sap, len, flags);
 }
 
 int svc_create_xprt(struct svc_serv *serv, const char *xprt_name,
-                    const int family, const unsigned short port,
-                    int flags)
+                    struct net *net, const int family,
+                    const unsigned short port, int flags)
 {
         struct svc_xprt_class *xcl;
 
@@ -212,6 +216,7 @@ int svc_create_xprt(struct svc_serv *serv, const char *xprt_name,
         spin_lock(&svc_xprt_class_lock);
         list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) {
                 struct svc_xprt *newxprt;
+                unsigned short newport;
 
                 if (strcmp(xprt_name, xcl->xcl_name))
                         continue;
@@ -220,7 +225,7 @@ int svc_create_xprt(struct svc_serv *serv, const char *xprt_name,
                         goto err;
 
                 spin_unlock(&svc_xprt_class_lock);
-                newxprt = __svc_xpo_create(xcl, serv, family, port, flags);
+                newxprt = __svc_xpo_create(xcl, serv, net, family, port, flags);
                 if (IS_ERR(newxprt)) {
                         module_put(xcl->xcl_owner);
                         return PTR_ERR(newxprt);
@@ -230,8 +235,9 @@ int svc_create_xprt(struct svc_serv *serv, const char *xprt_name,
                 spin_lock_bh(&serv->sv_lock);
                 list_add(&newxprt->xpt_list, &serv->sv_permsocks);
                 spin_unlock_bh(&serv->sv_lock);
+                newport = svc_xprt_local_port(newxprt);
                 clear_bit(XPT_BUSY, &newxprt->xpt_flags);
-                return svc_xprt_local_port(newxprt);
+                return newport;
         }
  err:
         spin_unlock(&svc_xprt_class_lock);
@@ -301,6 +307,15 @@ static void svc_thread_dequeue(struct svc_pool *pool, struct svc_rqst *rqstp)
         list_del(&rqstp->rq_list);
 }
 
+static bool svc_xprt_has_something_to_do(struct svc_xprt *xprt)
+{
+        if (xprt->xpt_flags & ((1<<XPT_CONN)|(1<<XPT_CLOSE)))
+                return true;
+        if (xprt->xpt_flags & ((1<<XPT_DATA)|(1<<XPT_DEFERRED)))
+                return xprt->xpt_ops->xpo_has_wspace(xprt);
+        return false;
+}
+
 /*
  * Queue up a transport with data pending. If there are idle nfsd
  * processes, wake 'em up.
@@ -313,8 +328,7 @@ void svc_xprt_enqueue(struct svc_xprt *xprt)
         struct svc_rqst *rqstp;
         int cpu;
 
-        if (!(xprt->xpt_flags &
-              ((1<<XPT_CONN)|(1<<XPT_DATA)|(1<<XPT_CLOSE)|(1<<XPT_DEFERRED))))
+        if (!svc_xprt_has_something_to_do(xprt))
                 return;
 
         cpu = get_cpu();
@@ -329,12 +343,6 @@ void svc_xprt_enqueue(struct svc_xprt *xprt)
                        "svc_xprt_enqueue: "
                        "threads and transports both waiting??\n");
 
-        if (test_bit(XPT_DEAD, &xprt->xpt_flags)) {
-                /* Don't enqueue dead transports */
-                dprintk("svc: transport %p is dead, not enqueued\n", xprt);
-                goto out_unlock;
-        }
-
         pool->sp_stats.packets++;
 
         /* Mark transport as busy. It will remain in this state until
@@ -347,28 +355,7 @@ void svc_xprt_enqueue(struct svc_xprt *xprt)
                 dprintk("svc: transport %p busy, not enqueued\n", xprt);
                 goto out_unlock;
         }
-        BUG_ON(xprt->xpt_pool != NULL);
-        xprt->xpt_pool = pool;
-
-        /* Handle pending connection */
-        if (test_bit(XPT_CONN, &xprt->xpt_flags))
-                goto process;
-
-        /* Handle close in-progress */
-        if (test_bit(XPT_CLOSE, &xprt->xpt_flags))
-                goto process;
-
-        /* Check if we have space to reply to a request */
-        if (!xprt->xpt_ops->xpo_has_wspace(xprt)) {
-                /* Don't enqueue while not enough space for reply */
-                dprintk("svc: no write space, transport %p  not enqueued\n",
-                        xprt);
-                xprt->xpt_pool = NULL;
-                clear_bit(XPT_BUSY, &xprt->xpt_flags);
-                goto out_unlock;
-        }
 
- process:
         if (!list_empty(&pool->sp_threads)) {
                 rqstp = list_entry(pool->sp_threads.next,
                                    struct svc_rqst,
@@ -385,13 +372,11 @@ void svc_xprt_enqueue(struct svc_xprt *xprt)
                 rqstp->rq_reserved = serv->sv_max_mesg;
                 atomic_add(rqstp->rq_reserved, &xprt->xpt_reserved);
                 pool->sp_stats.threads_woken++;
-                BUG_ON(xprt->xpt_pool != pool);
                 wake_up(&rqstp->rq_wait);
         } else {
                 dprintk("svc: transport %p put into queue\n", xprt);
                 list_add_tail(&xprt->xpt_ready, &pool->sp_sockets);
                 pool->sp_stats.sockets_queued++;
-                BUG_ON(xprt->xpt_pool != pool);
         }
 
 out_unlock:
@@ -430,9 +415,13 @@ static struct svc_xprt *svc_xprt_dequeue(struct svc_pool *pool)
 void svc_xprt_received(struct svc_xprt *xprt)
 {
         BUG_ON(!test_bit(XPT_BUSY, &xprt->xpt_flags));
-        xprt->xpt_pool = NULL;
+        /* As soon as we clear busy, the xprt could be closed and
+         * 'put', so we need a reference to call svc_xprt_enqueue with:
+         */
+        svc_xprt_get(xprt);
         clear_bit(XPT_BUSY, &xprt->xpt_flags);
         svc_xprt_enqueue(xprt);
+        svc_xprt_put(xprt);
 }
 EXPORT_SYMBOL_GPL(svc_xprt_received);
 
@@ -651,6 +640,11 @@ int svc_recv(struct svc_rqst *rqstp, long timeout)
         if (signalled() || kthread_should_stop())
                 return -EINTR;
 
+        /* Normally we will wait up to 5 seconds for any required
+         * cache information to be provided.
+         */
+        rqstp->rq_chandle.thread_wait = 5*HZ;
+
         spin_lock_bh(&pool->sp_lock);
         xprt = svc_xprt_dequeue(pool);
         if (xprt) {
@@ -658,6 +652,12 @@ int svc_recv(struct svc_rqst *rqstp, long timeout)
                 svc_xprt_get(xprt);
                 rqstp->rq_reserved = serv->sv_max_mesg;
                 atomic_add(rqstp->rq_reserved, &xprt->xpt_reserved);
+
+                /* As there is a shortage of threads and this request
+                 * had to be queued, don't allow the thread to wait so
+                 * long for cache updates.
+                 */
+                rqstp->rq_chandle.thread_wait = 1*HZ;
         } else {
                 /* No data pending. Go to sleep */
                 svc_thread_enqueue(pool, rqstp);
@@ -710,7 +710,10 @@ int svc_recv(struct svc_rqst *rqstp, long timeout)
         if (test_bit(XPT_CLOSE, &xprt->xpt_flags)) {
                 dprintk("svc_recv: found XPT_CLOSE\n");
                 svc_delete_xprt(xprt);
-        } else if (test_bit(XPT_LISTENER, &xprt->xpt_flags)) {
+                /* Leave XPT_BUSY set on the dead xprt: */
+                goto out;
+        }
+        if (test_bit(XPT_LISTENER, &xprt->xpt_flags)) {
                 struct svc_xprt *newxpt;
                 newxpt = xprt->xpt_ops->xpo_accept(xprt);
                 if (newxpt) {
@@ -735,28 +738,23 @@ int svc_recv(struct svc_rqst *rqstp, long timeout)
                         spin_unlock_bh(&serv->sv_lock);
                         svc_xprt_received(newxpt);
                 }
-                svc_xprt_received(xprt);
-        } else {
+        } else if (xprt->xpt_ops->xpo_has_wspace(xprt)) {
                 dprintk("svc: server %p, pool %u, transport %p, inuse=%d\n",
                         rqstp, pool->sp_id, xprt,
                         atomic_read(&xprt->xpt_ref.refcount));
                 rqstp->rq_deferred = svc_deferred_dequeue(xprt);
-                if (rqstp->rq_deferred) {
-                        svc_xprt_received(xprt);
+                if (rqstp->rq_deferred)
                         len = svc_deferred_recv(rqstp);
-                } else {
+                else
                         len = xprt->xpt_ops->xpo_recvfrom(rqstp);
-                        svc_xprt_received(xprt);
-                }
                 dprintk("svc: got len=%d\n", len);
         }
+        svc_xprt_received(xprt);
 
         /* No data, incomplete (TCP) read, or accept() */
-        if (len == 0 || len == -EAGAIN) {
-                rqstp->rq_res.len = 0;
-                svc_xprt_release(rqstp);
-                return -EAGAIN;
-        }
+        if (len == 0 || len == -EAGAIN)
+                goto out;
+
         clear_bit(XPT_OLD, &xprt->xpt_flags);
 
         rqstp->rq_secure = svc_port_is_privileged(svc_addr(rqstp));
@@ -765,6 +763,10 @@ int svc_recv(struct svc_rqst *rqstp, long timeout)
         if (serv->sv_stats)
                 serv->sv_stats->netcnt++;
         return len;
+out:
+        rqstp->rq_res.len = 0;
+        svc_xprt_release(rqstp);
+        return -EAGAIN;
 }
 EXPORT_SYMBOL_GPL(svc_recv);
 
@@ -868,6 +870,19 @@ static void svc_age_temp_xprts(unsigned long closure)
         mod_timer(&serv->sv_temptimer, jiffies + svc_conn_age_period * HZ);
 }
 
+static void call_xpt_users(struct svc_xprt *xprt)
+{
+        struct svc_xpt_user *u;
+
+        spin_lock(&xprt->xpt_lock);
+        while (!list_empty(&xprt->xpt_users)) {
+                u = list_first_entry(&xprt->xpt_users, struct svc_xpt_user, list);
+                list_del(&u->list);
+                u->callback(u);
+        }
+        spin_unlock(&xprt->xpt_lock);
+}
+
 /*
  * Remove a dead transport
  */
@@ -878,7 +893,7 @@ void svc_delete_xprt(struct svc_xprt *xprt)
 
         /* Only do this once */
         if (test_and_set_bit(XPT_DEAD, &xprt->xpt_flags))
-                return;
+                BUG();
 
         dprintk("svc: svc_delete_xprt(%p)\n", xprt);
         xprt->xpt_ops->xpo_detach(xprt);
@@ -900,6 +915,7 @@ void svc_delete_xprt(struct svc_xprt *xprt)
         while ((dr = svc_deferred_dequeue(xprt)) != NULL)
                 kfree(dr);
 
+        call_xpt_users(xprt);
         svc_xprt_put(xprt);
 }
 
@@ -909,11 +925,13 @@ void svc_close_xprt(struct svc_xprt *xprt)
         if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags))
                 /* someone else will have to effect the close */
                 return;
-
-        svc_xprt_get(xprt);
+        /*
+         * We expect svc_close_xprt() to work even when no threads are
+         * running (e.g., while configuring the server before starting
+         * any threads), so if the transport isn't busy, we delete
+         * it ourself:
+         */
         svc_delete_xprt(xprt);
-        clear_bit(XPT_BUSY, &xprt->xpt_flags);
-        svc_xprt_put(xprt);
 }
 EXPORT_SYMBOL_GPL(svc_close_xprt);
 
@@ -922,16 +940,16 @@ void svc_close_all(struct list_head *xprt_list)
         struct svc_xprt *xprt;
         struct svc_xprt *tmp;
 
+        /*
+         * The server is shutting down, and no more threads are running.
+         * svc_xprt_enqueue() might still be running, but at worst it
+         * will re-add the xprt to sp_sockets, which will soon get
+         * freed.  So we don't bother with any more locking, and don't
+         * leave the close to the (nonexistent) server threads:
+         */
         list_for_each_entry_safe(xprt, tmp, xprt_list, xpt_list) {
                 set_bit(XPT_CLOSE, &xprt->xpt_flags);
-                if (test_bit(XPT_BUSY, &xprt->xpt_flags)) {
-                        /* Waiting to be processed, but no threads left,
-                         * So just remove it from the waiting list
-                         */
-                        list_del_init(&xprt->xpt_ready);
-                        clear_bit(XPT_BUSY, &xprt->xpt_flags);
-                }
-                svc_close_xprt(xprt);
+                svc_delete_xprt(xprt);
         }
 }
 
@@ -1005,6 +1023,7 @@ static struct cache_deferred_req *svc_defer(struct cache_req *req)
         }
         svc_xprt_get(rqstp->rq_xprt);
         dr->xprt = rqstp->rq_xprt;
+        rqstp->rq_dropme = true;
 
         dr->handle.revisit = svc_revisit;
         return &dr->handle;
@@ -1042,14 +1061,13 @@ static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt)
         if (!test_bit(XPT_DEFERRED, &xprt->xpt_flags))
                 return NULL;
         spin_lock(&xprt->xpt_lock);
-        clear_bit(XPT_DEFERRED, &xprt->xpt_flags);
         if (!list_empty(&xprt->xpt_deferred)) {
                 dr = list_entry(xprt->xpt_deferred.next,
                                 struct svc_deferred_req,
                                 handle.recent);
                 list_del_init(&dr->handle.recent);
-                set_bit(XPT_DEFERRED, &xprt->xpt_flags);
-        }
+        } else
+                clear_bit(XPT_DEFERRED, &xprt->xpt_flags);
         spin_unlock(&xprt->xpt_lock);
         return dr;
 }