Back to home page

OSCL-LXR
 
 

     

0001 // SPDX-License-Identifier: GPL-2.0-only
0002 /*
0003  * linux/net/sunrpc/svc_xprt.c
0004  *
0005  * Author: Tom Tucker <tom@opengridcomputing.com>
0006  */
0007 
0008 #include <linux/sched.h>
0009 #include <linux/sched/mm.h>
0010 #include <linux/errno.h>
0011 #include <linux/freezer.h>
0012 #include <linux/kthread.h>
0013 #include <linux/slab.h>
0014 #include <net/sock.h>
0015 #include <linux/sunrpc/addr.h>
0016 #include <linux/sunrpc/stats.h>
0017 #include <linux/sunrpc/svc_xprt.h>
0018 #include <linux/sunrpc/svcsock.h>
0019 #include <linux/sunrpc/xprt.h>
0020 #include <linux/module.h>
0021 #include <linux/netdevice.h>
0022 #include <trace/events/sunrpc.h>
0023 
0024 #define RPCDBG_FACILITY RPCDBG_SVCXPRT
0025 
0026 static unsigned int svc_rpc_per_connection_limit __read_mostly;
0027 module_param(svc_rpc_per_connection_limit, uint, 0644);
0028 
0029 
0030 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt);
0031 static int svc_deferred_recv(struct svc_rqst *rqstp);
0032 static struct cache_deferred_req *svc_defer(struct cache_req *req);
0033 static void svc_age_temp_xprts(struct timer_list *t);
0034 static void svc_delete_xprt(struct svc_xprt *xprt);
0035 
0036 /* apparently the "standard" is that clients close
0037  * idle connections after 5 minutes, servers after
0038  * 6 minutes
0039  *   http://nfsv4bat.org/Documents/ConnectAThon/1996/nfstcp.pdf
0040  */
0041 static int svc_conn_age_period = 6*60;
0042 
0043 /* List of registered transport classes */
0044 static DEFINE_SPINLOCK(svc_xprt_class_lock);
0045 static LIST_HEAD(svc_xprt_class_list);
0046 
0047 /* SMP locking strategy:
0048  *
0049  *  svc_pool->sp_lock protects most of the fields of that pool.
0050  *  svc_serv->sv_lock protects sv_tempsocks, sv_permsocks, sv_tmpcnt.
0051  *  when both need to be taken (rare), svc_serv->sv_lock is first.
0052  *  The "service mutex" protects svc_serv->sv_nrthread.
0053  *  svc_sock->sk_lock protects the svc_sock->sk_deferred list
0054  *             and the ->sk_info_authunix cache.
0055  *
0056  *  The XPT_BUSY bit in xprt->xpt_flags prevents a transport being
0057  *  enqueued multiply. During normal transport processing this bit
0058  *  is set by svc_xprt_enqueue and cleared by svc_xprt_received.
0059  *  Providers should not manipulate this bit directly.
0060  *
0061  *  Some flags can be set to certain values at any time
0062  *  providing that certain rules are followed:
0063  *
0064  *  XPT_CONN, XPT_DATA:
0065  *      - Can be set or cleared at any time.
0066  *      - After a set, svc_xprt_enqueue must be called to enqueue
0067  *        the transport for processing.
0068  *      - After a clear, the transport must be read/accepted.
0069  *        If this succeeds, it must be set again.
0070  *  XPT_CLOSE:
0071  *      - Can set at any time. It is never cleared.
0072  *      XPT_DEAD:
0073  *      - Can only be set while XPT_BUSY is held which ensures
0074  *        that no other thread will be using the transport or will
0075  *        try to set XPT_DEAD.
0076  */
0077 int svc_reg_xprt_class(struct svc_xprt_class *xcl)
0078 {
0079     struct svc_xprt_class *cl;
0080     int res = -EEXIST;
0081 
0082     dprintk("svc: Adding svc transport class '%s'\n", xcl->xcl_name);
0083 
0084     INIT_LIST_HEAD(&xcl->xcl_list);
0085     spin_lock(&svc_xprt_class_lock);
0086     /* Make sure there isn't already a class with the same name */
0087     list_for_each_entry(cl, &svc_xprt_class_list, xcl_list) {
0088         if (strcmp(xcl->xcl_name, cl->xcl_name) == 0)
0089             goto out;
0090     }
0091     list_add_tail(&xcl->xcl_list, &svc_xprt_class_list);
0092     res = 0;
0093 out:
0094     spin_unlock(&svc_xprt_class_lock);
0095     return res;
0096 }
0097 EXPORT_SYMBOL_GPL(svc_reg_xprt_class);
0098 
0099 void svc_unreg_xprt_class(struct svc_xprt_class *xcl)
0100 {
0101     dprintk("svc: Removing svc transport class '%s'\n", xcl->xcl_name);
0102     spin_lock(&svc_xprt_class_lock);
0103     list_del_init(&xcl->xcl_list);
0104     spin_unlock(&svc_xprt_class_lock);
0105 }
0106 EXPORT_SYMBOL_GPL(svc_unreg_xprt_class);
0107 
0108 /**
0109  * svc_print_xprts - Format the transport list for printing
0110  * @buf: target buffer for formatted address
0111  * @maxlen: length of target buffer
0112  *
0113  * Fills in @buf with a string containing a list of transport names, each name
0114  * terminated with '\n'. If the buffer is too small, some entries may be
0115  * missing, but it is guaranteed that all lines in the output buffer are
0116  * complete.
0117  *
0118  * Returns positive length of the filled-in string.
0119  */
0120 int svc_print_xprts(char *buf, int maxlen)
0121 {
0122     struct svc_xprt_class *xcl;
0123     char tmpstr[80];
0124     int len = 0;
0125     buf[0] = '\0';
0126 
0127     spin_lock(&svc_xprt_class_lock);
0128     list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) {
0129         int slen;
0130 
0131         slen = snprintf(tmpstr, sizeof(tmpstr), "%s %d\n",
0132                 xcl->xcl_name, xcl->xcl_max_payload);
0133         if (slen >= sizeof(tmpstr) || len + slen >= maxlen)
0134             break;
0135         len += slen;
0136         strcat(buf, tmpstr);
0137     }
0138     spin_unlock(&svc_xprt_class_lock);
0139 
0140     return len;
0141 }
0142 
0143 /**
0144  * svc_xprt_deferred_close - Close a transport
0145  * @xprt: transport instance
0146  *
0147  * Used in contexts that need to defer the work of shutting down
0148  * the transport to an nfsd thread.
0149  */
0150 void svc_xprt_deferred_close(struct svc_xprt *xprt)
0151 {
0152     if (!test_and_set_bit(XPT_CLOSE, &xprt->xpt_flags))
0153         svc_xprt_enqueue(xprt);
0154 }
0155 EXPORT_SYMBOL_GPL(svc_xprt_deferred_close);
0156 
0157 static void svc_xprt_free(struct kref *kref)
0158 {
0159     struct svc_xprt *xprt =
0160         container_of(kref, struct svc_xprt, xpt_ref);
0161     struct module *owner = xprt->xpt_class->xcl_owner;
0162     if (test_bit(XPT_CACHE_AUTH, &xprt->xpt_flags))
0163         svcauth_unix_info_release(xprt);
0164     put_cred(xprt->xpt_cred);
0165     put_net_track(xprt->xpt_net, &xprt->ns_tracker);
0166     /* See comment on corresponding get in xs_setup_bc_tcp(): */
0167     if (xprt->xpt_bc_xprt)
0168         xprt_put(xprt->xpt_bc_xprt);
0169     if (xprt->xpt_bc_xps)
0170         xprt_switch_put(xprt->xpt_bc_xps);
0171     trace_svc_xprt_free(xprt);
0172     xprt->xpt_ops->xpo_free(xprt);
0173     module_put(owner);
0174 }
0175 
0176 void svc_xprt_put(struct svc_xprt *xprt)
0177 {
0178     kref_put(&xprt->xpt_ref, svc_xprt_free);
0179 }
0180 EXPORT_SYMBOL_GPL(svc_xprt_put);
0181 
0182 /*
0183  * Called by transport drivers to initialize the transport independent
0184  * portion of the transport instance.
0185  */
0186 void svc_xprt_init(struct net *net, struct svc_xprt_class *xcl,
0187            struct svc_xprt *xprt, struct svc_serv *serv)
0188 {
0189     memset(xprt, 0, sizeof(*xprt));
0190     xprt->xpt_class = xcl;
0191     xprt->xpt_ops = xcl->xcl_ops;
0192     kref_init(&xprt->xpt_ref);
0193     xprt->xpt_server = serv;
0194     INIT_LIST_HEAD(&xprt->xpt_list);
0195     INIT_LIST_HEAD(&xprt->xpt_ready);
0196     INIT_LIST_HEAD(&xprt->xpt_deferred);
0197     INIT_LIST_HEAD(&xprt->xpt_users);
0198     mutex_init(&xprt->xpt_mutex);
0199     spin_lock_init(&xprt->xpt_lock);
0200     set_bit(XPT_BUSY, &xprt->xpt_flags);
0201     xprt->xpt_net = get_net_track(net, &xprt->ns_tracker, GFP_ATOMIC);
0202     strcpy(xprt->xpt_remotebuf, "uninitialized");
0203 }
0204 EXPORT_SYMBOL_GPL(svc_xprt_init);
0205 
0206 static struct svc_xprt *__svc_xpo_create(struct svc_xprt_class *xcl,
0207                      struct svc_serv *serv,
0208                      struct net *net,
0209                      const int family,
0210                      const unsigned short port,
0211                      int flags)
0212 {
0213     struct sockaddr_in sin = {
0214         .sin_family     = AF_INET,
0215         .sin_addr.s_addr    = htonl(INADDR_ANY),
0216         .sin_port       = htons(port),
0217     };
0218 #if IS_ENABLED(CONFIG_IPV6)
0219     struct sockaddr_in6 sin6 = {
0220         .sin6_family        = AF_INET6,
0221         .sin6_addr      = IN6ADDR_ANY_INIT,
0222         .sin6_port      = htons(port),
0223     };
0224 #endif
0225     struct svc_xprt *xprt;
0226     struct sockaddr *sap;
0227     size_t len;
0228 
0229     switch (family) {
0230     case PF_INET:
0231         sap = (struct sockaddr *)&sin;
0232         len = sizeof(sin);
0233         break;
0234 #if IS_ENABLED(CONFIG_IPV6)
0235     case PF_INET6:
0236         sap = (struct sockaddr *)&sin6;
0237         len = sizeof(sin6);
0238         break;
0239 #endif
0240     default:
0241         return ERR_PTR(-EAFNOSUPPORT);
0242     }
0243 
0244     xprt = xcl->xcl_ops->xpo_create(serv, net, sap, len, flags);
0245     if (IS_ERR(xprt))
0246         trace_svc_xprt_create_err(serv->sv_program->pg_name,
0247                       xcl->xcl_name, sap, len, xprt);
0248     return xprt;
0249 }
0250 
0251 /**
0252  * svc_xprt_received - start next receiver thread
0253  * @xprt: controlling transport
0254  *
0255  * The caller must hold the XPT_BUSY bit and must
0256  * not thereafter touch transport data.
0257  *
0258  * Note: XPT_DATA only gets cleared when a read-attempt finds no (or
0259  * insufficient) data.
0260  */
0261 void svc_xprt_received(struct svc_xprt *xprt)
0262 {
0263     if (!test_bit(XPT_BUSY, &xprt->xpt_flags)) {
0264         WARN_ONCE(1, "xprt=0x%p already busy!", xprt);
0265         return;
0266     }
0267 
0268     /* As soon as we clear busy, the xprt could be closed and
0269      * 'put', so we need a reference to call svc_xprt_enqueue with:
0270      */
0271     svc_xprt_get(xprt);
0272     smp_mb__before_atomic();
0273     clear_bit(XPT_BUSY, &xprt->xpt_flags);
0274     svc_xprt_enqueue(xprt);
0275     svc_xprt_put(xprt);
0276 }
0277 EXPORT_SYMBOL_GPL(svc_xprt_received);
0278 
0279 void svc_add_new_perm_xprt(struct svc_serv *serv, struct svc_xprt *new)
0280 {
0281     clear_bit(XPT_TEMP, &new->xpt_flags);
0282     spin_lock_bh(&serv->sv_lock);
0283     list_add(&new->xpt_list, &serv->sv_permsocks);
0284     spin_unlock_bh(&serv->sv_lock);
0285     svc_xprt_received(new);
0286 }
0287 
0288 static int _svc_xprt_create(struct svc_serv *serv, const char *xprt_name,
0289                 struct net *net, const int family,
0290                 const unsigned short port, int flags,
0291                 const struct cred *cred)
0292 {
0293     struct svc_xprt_class *xcl;
0294 
0295     spin_lock(&svc_xprt_class_lock);
0296     list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) {
0297         struct svc_xprt *newxprt;
0298         unsigned short newport;
0299 
0300         if (strcmp(xprt_name, xcl->xcl_name))
0301             continue;
0302 
0303         if (!try_module_get(xcl->xcl_owner))
0304             goto err;
0305 
0306         spin_unlock(&svc_xprt_class_lock);
0307         newxprt = __svc_xpo_create(xcl, serv, net, family, port, flags);
0308         if (IS_ERR(newxprt)) {
0309             module_put(xcl->xcl_owner);
0310             return PTR_ERR(newxprt);
0311         }
0312         newxprt->xpt_cred = get_cred(cred);
0313         svc_add_new_perm_xprt(serv, newxprt);
0314         newport = svc_xprt_local_port(newxprt);
0315         return newport;
0316     }
0317  err:
0318     spin_unlock(&svc_xprt_class_lock);
0319     /* This errno is exposed to user space.  Provide a reasonable
0320      * perror msg for a bad transport. */
0321     return -EPROTONOSUPPORT;
0322 }
0323 
0324 /**
0325  * svc_xprt_create - Add a new listener to @serv
0326  * @serv: target RPC service
0327  * @xprt_name: transport class name
0328  * @net: network namespace
0329  * @family: network address family
0330  * @port: listener port
0331  * @flags: SVC_SOCK flags
0332  * @cred: credential to bind to this transport
0333  *
0334  * Return values:
0335  *   %0: New listener added successfully
0336  *   %-EPROTONOSUPPORT: Requested transport type not supported
0337  */
0338 int svc_xprt_create(struct svc_serv *serv, const char *xprt_name,
0339             struct net *net, const int family,
0340             const unsigned short port, int flags,
0341             const struct cred *cred)
0342 {
0343     int err;
0344 
0345     err = _svc_xprt_create(serv, xprt_name, net, family, port, flags, cred);
0346     if (err == -EPROTONOSUPPORT) {
0347         request_module("svc%s", xprt_name);
0348         err = _svc_xprt_create(serv, xprt_name, net, family, port, flags, cred);
0349     }
0350     return err;
0351 }
0352 EXPORT_SYMBOL_GPL(svc_xprt_create);
0353 
0354 /*
0355  * Copy the local and remote xprt addresses to the rqstp structure
0356  */
0357 void svc_xprt_copy_addrs(struct svc_rqst *rqstp, struct svc_xprt *xprt)
0358 {
0359     memcpy(&rqstp->rq_addr, &xprt->xpt_remote, xprt->xpt_remotelen);
0360     rqstp->rq_addrlen = xprt->xpt_remotelen;
0361 
0362     /*
0363      * Destination address in request is needed for binding the
0364      * source address in RPC replies/callbacks later.
0365      */
0366     memcpy(&rqstp->rq_daddr, &xprt->xpt_local, xprt->xpt_locallen);
0367     rqstp->rq_daddrlen = xprt->xpt_locallen;
0368 }
0369 EXPORT_SYMBOL_GPL(svc_xprt_copy_addrs);
0370 
0371 /**
0372  * svc_print_addr - Format rq_addr field for printing
0373  * @rqstp: svc_rqst struct containing address to print
0374  * @buf: target buffer for formatted address
0375  * @len: length of target buffer
0376  *
0377  */
0378 char *svc_print_addr(struct svc_rqst *rqstp, char *buf, size_t len)
0379 {
0380     return __svc_print_addr(svc_addr(rqstp), buf, len);
0381 }
0382 EXPORT_SYMBOL_GPL(svc_print_addr);
0383 
0384 static bool svc_xprt_slots_in_range(struct svc_xprt *xprt)
0385 {
0386     unsigned int limit = svc_rpc_per_connection_limit;
0387     int nrqsts = atomic_read(&xprt->xpt_nr_rqsts);
0388 
0389     return limit == 0 || (nrqsts >= 0 && nrqsts < limit);
0390 }
0391 
0392 static bool svc_xprt_reserve_slot(struct svc_rqst *rqstp, struct svc_xprt *xprt)
0393 {
0394     if (!test_bit(RQ_DATA, &rqstp->rq_flags)) {
0395         if (!svc_xprt_slots_in_range(xprt))
0396             return false;
0397         atomic_inc(&xprt->xpt_nr_rqsts);
0398         set_bit(RQ_DATA, &rqstp->rq_flags);
0399     }
0400     return true;
0401 }
0402 
0403 static void svc_xprt_release_slot(struct svc_rqst *rqstp)
0404 {
0405     struct svc_xprt *xprt = rqstp->rq_xprt;
0406     if (test_and_clear_bit(RQ_DATA, &rqstp->rq_flags)) {
0407         atomic_dec(&xprt->xpt_nr_rqsts);
0408         smp_wmb(); /* See smp_rmb() in svc_xprt_ready() */
0409         svc_xprt_enqueue(xprt);
0410     }
0411 }
0412 
0413 static bool svc_xprt_ready(struct svc_xprt *xprt)
0414 {
0415     unsigned long xpt_flags;
0416 
0417     /*
0418      * If another cpu has recently updated xpt_flags,
0419      * sk_sock->flags, xpt_reserved, or xpt_nr_rqsts, we need to
0420      * know about it; otherwise it's possible that both that cpu and
0421      * this one could call svc_xprt_enqueue() without either
0422      * svc_xprt_enqueue() recognizing that the conditions below
0423      * are satisfied, and we could stall indefinitely:
0424      */
0425     smp_rmb();
0426     xpt_flags = READ_ONCE(xprt->xpt_flags);
0427 
0428     if (xpt_flags & BIT(XPT_BUSY))
0429         return false;
0430     if (xpt_flags & (BIT(XPT_CONN) | BIT(XPT_CLOSE)))
0431         return true;
0432     if (xpt_flags & (BIT(XPT_DATA) | BIT(XPT_DEFERRED))) {
0433         if (xprt->xpt_ops->xpo_has_wspace(xprt) &&
0434             svc_xprt_slots_in_range(xprt))
0435             return true;
0436         trace_svc_xprt_no_write_space(xprt);
0437         return false;
0438     }
0439     return false;
0440 }
0441 
0442 /**
0443  * svc_xprt_enqueue - Queue a transport on an idle nfsd thread
0444  * @xprt: transport with data pending
0445  *
0446  */
0447 void svc_xprt_enqueue(struct svc_xprt *xprt)
0448 {
0449     struct svc_pool *pool;
0450     struct svc_rqst *rqstp = NULL;
0451 
0452     if (!svc_xprt_ready(xprt))
0453         return;
0454 
0455     /* Mark transport as busy. It will remain in this state until
0456      * the provider calls svc_xprt_received. We update XPT_BUSY
0457      * atomically because it also guards against trying to enqueue
0458      * the transport twice.
0459      */
0460     if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags))
0461         return;
0462 
0463     pool = svc_pool_for_cpu(xprt->xpt_server);
0464 
0465     atomic_long_inc(&pool->sp_stats.packets);
0466 
0467     spin_lock_bh(&pool->sp_lock);
0468     list_add_tail(&xprt->xpt_ready, &pool->sp_sockets);
0469     pool->sp_stats.sockets_queued++;
0470     spin_unlock_bh(&pool->sp_lock);
0471 
0472     /* find a thread for this xprt */
0473     rcu_read_lock();
0474     list_for_each_entry_rcu(rqstp, &pool->sp_all_threads, rq_all) {
0475         if (test_and_set_bit(RQ_BUSY, &rqstp->rq_flags))
0476             continue;
0477         atomic_long_inc(&pool->sp_stats.threads_woken);
0478         rqstp->rq_qtime = ktime_get();
0479         wake_up_process(rqstp->rq_task);
0480         goto out_unlock;
0481     }
0482     set_bit(SP_CONGESTED, &pool->sp_flags);
0483     rqstp = NULL;
0484 out_unlock:
0485     rcu_read_unlock();
0486     trace_svc_xprt_enqueue(xprt, rqstp);
0487 }
0488 EXPORT_SYMBOL_GPL(svc_xprt_enqueue);
0489 
0490 /*
0491  * Dequeue the first transport, if there is one.
0492  */
0493 static struct svc_xprt *svc_xprt_dequeue(struct svc_pool *pool)
0494 {
0495     struct svc_xprt *xprt = NULL;
0496 
0497     if (list_empty(&pool->sp_sockets))
0498         goto out;
0499 
0500     spin_lock_bh(&pool->sp_lock);
0501     if (likely(!list_empty(&pool->sp_sockets))) {
0502         xprt = list_first_entry(&pool->sp_sockets,
0503                     struct svc_xprt, xpt_ready);
0504         list_del_init(&xprt->xpt_ready);
0505         svc_xprt_get(xprt);
0506     }
0507     spin_unlock_bh(&pool->sp_lock);
0508 out:
0509     return xprt;
0510 }
0511 
0512 /**
0513  * svc_reserve - change the space reserved for the reply to a request.
0514  * @rqstp:  The request in question
0515  * @space: new max space to reserve
0516  *
0517  * Each request reserves some space on the output queue of the transport
0518  * to make sure the reply fits.  This function reduces that reserved
0519  * space to be the amount of space used already, plus @space.
0520  *
0521  */
0522 void svc_reserve(struct svc_rqst *rqstp, int space)
0523 {
0524     struct svc_xprt *xprt = rqstp->rq_xprt;
0525 
0526     space += rqstp->rq_res.head[0].iov_len;
0527 
0528     if (xprt && space < rqstp->rq_reserved) {
0529         atomic_sub((rqstp->rq_reserved - space), &xprt->xpt_reserved);
0530         rqstp->rq_reserved = space;
0531         smp_wmb(); /* See smp_rmb() in svc_xprt_ready() */
0532         svc_xprt_enqueue(xprt);
0533     }
0534 }
0535 EXPORT_SYMBOL_GPL(svc_reserve);
0536 
0537 static void svc_xprt_release(struct svc_rqst *rqstp)
0538 {
0539     struct svc_xprt *xprt = rqstp->rq_xprt;
0540 
0541     xprt->xpt_ops->xpo_release_rqst(rqstp);
0542 
0543     kfree(rqstp->rq_deferred);
0544     rqstp->rq_deferred = NULL;
0545 
0546     pagevec_release(&rqstp->rq_pvec);
0547     svc_free_res_pages(rqstp);
0548     rqstp->rq_res.page_len = 0;
0549     rqstp->rq_res.page_base = 0;
0550 
0551     /* Reset response buffer and release
0552      * the reservation.
0553      * But first, check that enough space was reserved
0554      * for the reply, otherwise we have a bug!
0555      */
0556     if ((rqstp->rq_res.len) >  rqstp->rq_reserved)
0557         printk(KERN_ERR "RPC request reserved %d but used %d\n",
0558                rqstp->rq_reserved,
0559                rqstp->rq_res.len);
0560 
0561     rqstp->rq_res.head[0].iov_len = 0;
0562     svc_reserve(rqstp, 0);
0563     svc_xprt_release_slot(rqstp);
0564     rqstp->rq_xprt = NULL;
0565     svc_xprt_put(xprt);
0566 }
0567 
0568 /*
0569  * Some svc_serv's will have occasional work to do, even when a xprt is not
0570  * waiting to be serviced. This function is there to "kick" a task in one of
0571  * those services so that it can wake up and do that work. Note that we only
0572  * bother with pool 0 as we don't need to wake up more than one thread for
0573  * this purpose.
0574  */
0575 void svc_wake_up(struct svc_serv *serv)
0576 {
0577     struct svc_rqst *rqstp;
0578     struct svc_pool *pool;
0579 
0580     pool = &serv->sv_pools[0];
0581 
0582     rcu_read_lock();
0583     list_for_each_entry_rcu(rqstp, &pool->sp_all_threads, rq_all) {
0584         /* skip any that aren't queued */
0585         if (test_bit(RQ_BUSY, &rqstp->rq_flags))
0586             continue;
0587         rcu_read_unlock();
0588         wake_up_process(rqstp->rq_task);
0589         trace_svc_wake_up(rqstp->rq_task->pid);
0590         return;
0591     }
0592     rcu_read_unlock();
0593 
0594     /* No free entries available */
0595     set_bit(SP_TASK_PENDING, &pool->sp_flags);
0596     smp_wmb();
0597     trace_svc_wake_up(0);
0598 }
0599 EXPORT_SYMBOL_GPL(svc_wake_up);
0600 
0601 int svc_port_is_privileged(struct sockaddr *sin)
0602 {
0603     switch (sin->sa_family) {
0604     case AF_INET:
0605         return ntohs(((struct sockaddr_in *)sin)->sin_port)
0606             < PROT_SOCK;
0607     case AF_INET6:
0608         return ntohs(((struct sockaddr_in6 *)sin)->sin6_port)
0609             < PROT_SOCK;
0610     default:
0611         return 0;
0612     }
0613 }
0614 
0615 /*
0616  * Make sure that we don't have too many active connections. If we have,
0617  * something must be dropped. It's not clear what will happen if we allow
0618  * "too many" connections, but when dealing with network-facing software,
0619  * we have to code defensively. Here we do that by imposing hard limits.
0620  *
0621  * There's no point in trying to do random drop here for DoS
0622  * prevention. The NFS clients does 1 reconnect in 15 seconds. An
0623  * attacker can easily beat that.
0624  *
0625  * The only somewhat efficient mechanism would be if drop old
0626  * connections from the same IP first. But right now we don't even
0627  * record the client IP in svc_sock.
0628  *
0629  * single-threaded services that expect a lot of clients will probably
0630  * need to set sv_maxconn to override the default value which is based
0631  * on the number of threads
0632  */
0633 static void svc_check_conn_limits(struct svc_serv *serv)
0634 {
0635     unsigned int limit = serv->sv_maxconn ? serv->sv_maxconn :
0636                 (serv->sv_nrthreads+3) * 20;
0637 
0638     if (serv->sv_tmpcnt > limit) {
0639         struct svc_xprt *xprt = NULL;
0640         spin_lock_bh(&serv->sv_lock);
0641         if (!list_empty(&serv->sv_tempsocks)) {
0642             /* Try to help the admin */
0643             net_notice_ratelimited("%s: too many open connections, consider increasing the %s\n",
0644                            serv->sv_name, serv->sv_maxconn ?
0645                            "max number of connections" :
0646                            "number of threads");
0647             /*
0648              * Always select the oldest connection. It's not fair,
0649              * but so is life
0650              */
0651             xprt = list_entry(serv->sv_tempsocks.prev,
0652                       struct svc_xprt,
0653                       xpt_list);
0654             set_bit(XPT_CLOSE, &xprt->xpt_flags);
0655             svc_xprt_get(xprt);
0656         }
0657         spin_unlock_bh(&serv->sv_lock);
0658 
0659         if (xprt) {
0660             svc_xprt_enqueue(xprt);
0661             svc_xprt_put(xprt);
0662         }
0663     }
0664 }
0665 
0666 static int svc_alloc_arg(struct svc_rqst *rqstp)
0667 {
0668     struct svc_serv *serv = rqstp->rq_server;
0669     struct xdr_buf *arg = &rqstp->rq_arg;
0670     unsigned long pages, filled, ret;
0671 
0672     pagevec_init(&rqstp->rq_pvec);
0673 
0674     pages = (serv->sv_max_mesg + 2 * PAGE_SIZE) >> PAGE_SHIFT;
0675     if (pages > RPCSVC_MAXPAGES) {
0676         pr_warn_once("svc: warning: pages=%lu > RPCSVC_MAXPAGES=%lu\n",
0677                  pages, RPCSVC_MAXPAGES);
0678         /* use as many pages as possible */
0679         pages = RPCSVC_MAXPAGES;
0680     }
0681 
0682     for (filled = 0; filled < pages; filled = ret) {
0683         ret = alloc_pages_bulk_array(GFP_KERNEL, pages,
0684                          rqstp->rq_pages);
0685         if (ret > filled)
0686             /* Made progress, don't sleep yet */
0687             continue;
0688 
0689         set_current_state(TASK_INTERRUPTIBLE);
0690         if (signalled() || kthread_should_stop()) {
0691             set_current_state(TASK_RUNNING);
0692             return -EINTR;
0693         }
0694         trace_svc_alloc_arg_err(pages, ret);
0695         memalloc_retry_wait(GFP_KERNEL);
0696     }
0697     rqstp->rq_page_end = &rqstp->rq_pages[pages];
0698     rqstp->rq_pages[pages] = NULL; /* this might be seen in nfsd_splice_actor() */
0699 
0700     /* Make arg->head point to first page and arg->pages point to rest */
0701     arg->head[0].iov_base = page_address(rqstp->rq_pages[0]);
0702     arg->head[0].iov_len = PAGE_SIZE;
0703     arg->pages = rqstp->rq_pages + 1;
0704     arg->page_base = 0;
0705     /* save at least one page for response */
0706     arg->page_len = (pages-2)*PAGE_SIZE;
0707     arg->len = (pages-1)*PAGE_SIZE;
0708     arg->tail[0].iov_len = 0;
0709     return 0;
0710 }
0711 
0712 static bool
0713 rqst_should_sleep(struct svc_rqst *rqstp)
0714 {
0715     struct svc_pool     *pool = rqstp->rq_pool;
0716 
0717     /* did someone call svc_wake_up? */
0718     if (test_and_clear_bit(SP_TASK_PENDING, &pool->sp_flags))
0719         return false;
0720 
0721     /* was a socket queued? */
0722     if (!list_empty(&pool->sp_sockets))
0723         return false;
0724 
0725     /* are we shutting down? */
0726     if (signalled() || kthread_should_stop())
0727         return false;
0728 
0729     /* are we freezing? */
0730     if (freezing(current))
0731         return false;
0732 
0733     return true;
0734 }
0735 
0736 static struct svc_xprt *svc_get_next_xprt(struct svc_rqst *rqstp, long timeout)
0737 {
0738     struct svc_pool     *pool = rqstp->rq_pool;
0739     long            time_left = 0;
0740 
0741     /* rq_xprt should be clear on entry */
0742     WARN_ON_ONCE(rqstp->rq_xprt);
0743 
0744     rqstp->rq_xprt = svc_xprt_dequeue(pool);
0745     if (rqstp->rq_xprt)
0746         goto out_found;
0747 
0748     /*
0749      * We have to be able to interrupt this wait
0750      * to bring down the daemons ...
0751      */
0752     set_current_state(TASK_INTERRUPTIBLE);
0753     smp_mb__before_atomic();
0754     clear_bit(SP_CONGESTED, &pool->sp_flags);
0755     clear_bit(RQ_BUSY, &rqstp->rq_flags);
0756     smp_mb__after_atomic();
0757 
0758     if (likely(rqst_should_sleep(rqstp)))
0759         time_left = schedule_timeout(timeout);
0760     else
0761         __set_current_state(TASK_RUNNING);
0762 
0763     try_to_freeze();
0764 
0765     set_bit(RQ_BUSY, &rqstp->rq_flags);
0766     smp_mb__after_atomic();
0767     rqstp->rq_xprt = svc_xprt_dequeue(pool);
0768     if (rqstp->rq_xprt)
0769         goto out_found;
0770 
0771     if (!time_left)
0772         atomic_long_inc(&pool->sp_stats.threads_timedout);
0773 
0774     if (signalled() || kthread_should_stop())
0775         return ERR_PTR(-EINTR);
0776     return ERR_PTR(-EAGAIN);
0777 out_found:
0778     /* Normally we will wait up to 5 seconds for any required
0779      * cache information to be provided.
0780      */
0781     if (!test_bit(SP_CONGESTED, &pool->sp_flags))
0782         rqstp->rq_chandle.thread_wait = 5*HZ;
0783     else
0784         rqstp->rq_chandle.thread_wait = 1*HZ;
0785     trace_svc_xprt_dequeue(rqstp);
0786     return rqstp->rq_xprt;
0787 }
0788 
0789 static void svc_add_new_temp_xprt(struct svc_serv *serv, struct svc_xprt *newxpt)
0790 {
0791     spin_lock_bh(&serv->sv_lock);
0792     set_bit(XPT_TEMP, &newxpt->xpt_flags);
0793     list_add(&newxpt->xpt_list, &serv->sv_tempsocks);
0794     serv->sv_tmpcnt++;
0795     if (serv->sv_temptimer.function == NULL) {
0796         /* setup timer to age temp transports */
0797         serv->sv_temptimer.function = svc_age_temp_xprts;
0798         mod_timer(&serv->sv_temptimer,
0799               jiffies + svc_conn_age_period * HZ);
0800     }
0801     spin_unlock_bh(&serv->sv_lock);
0802     svc_xprt_received(newxpt);
0803 }
0804 
0805 static int svc_handle_xprt(struct svc_rqst *rqstp, struct svc_xprt *xprt)
0806 {
0807     struct svc_serv *serv = rqstp->rq_server;
0808     int len = 0;
0809 
0810     if (test_bit(XPT_CLOSE, &xprt->xpt_flags)) {
0811         if (test_and_clear_bit(XPT_KILL_TEMP, &xprt->xpt_flags))
0812             xprt->xpt_ops->xpo_kill_temp_xprt(xprt);
0813         svc_delete_xprt(xprt);
0814         /* Leave XPT_BUSY set on the dead xprt: */
0815         goto out;
0816     }
0817     if (test_bit(XPT_LISTENER, &xprt->xpt_flags)) {
0818         struct svc_xprt *newxpt;
0819         /*
0820          * We know this module_get will succeed because the
0821          * listener holds a reference too
0822          */
0823         __module_get(xprt->xpt_class->xcl_owner);
0824         svc_check_conn_limits(xprt->xpt_server);
0825         newxpt = xprt->xpt_ops->xpo_accept(xprt);
0826         if (newxpt) {
0827             newxpt->xpt_cred = get_cred(xprt->xpt_cred);
0828             svc_add_new_temp_xprt(serv, newxpt);
0829             trace_svc_xprt_accept(newxpt, serv->sv_name);
0830         } else {
0831             module_put(xprt->xpt_class->xcl_owner);
0832         }
0833         svc_xprt_received(xprt);
0834     } else if (svc_xprt_reserve_slot(rqstp, xprt)) {
0835         /* XPT_DATA|XPT_DEFERRED case: */
0836         dprintk("svc: server %p, pool %u, transport %p, inuse=%d\n",
0837             rqstp, rqstp->rq_pool->sp_id, xprt,
0838             kref_read(&xprt->xpt_ref));
0839         rqstp->rq_deferred = svc_deferred_dequeue(xprt);
0840         if (rqstp->rq_deferred)
0841             len = svc_deferred_recv(rqstp);
0842         else
0843             len = xprt->xpt_ops->xpo_recvfrom(rqstp);
0844         rqstp->rq_stime = ktime_get();
0845         rqstp->rq_reserved = serv->sv_max_mesg;
0846         atomic_add(rqstp->rq_reserved, &xprt->xpt_reserved);
0847     } else
0848         svc_xprt_received(xprt);
0849 
0850 out:
0851     return len;
0852 }
0853 
0854 /*
0855  * Receive the next request on any transport.  This code is carefully
0856  * organised not to touch any cachelines in the shared svc_serv
0857  * structure, only cachelines in the local svc_pool.
0858  */
0859 int svc_recv(struct svc_rqst *rqstp, long timeout)
0860 {
0861     struct svc_xprt     *xprt = NULL;
0862     struct svc_serv     *serv = rqstp->rq_server;
0863     int         len, err;
0864 
0865     err = svc_alloc_arg(rqstp);
0866     if (err)
0867         goto out;
0868 
0869     try_to_freeze();
0870     cond_resched();
0871     err = -EINTR;
0872     if (signalled() || kthread_should_stop())
0873         goto out;
0874 
0875     xprt = svc_get_next_xprt(rqstp, timeout);
0876     if (IS_ERR(xprt)) {
0877         err = PTR_ERR(xprt);
0878         goto out;
0879     }
0880 
0881     len = svc_handle_xprt(rqstp, xprt);
0882 
0883     /* No data, incomplete (TCP) read, or accept() */
0884     err = -EAGAIN;
0885     if (len <= 0)
0886         goto out_release;
0887     trace_svc_xdr_recvfrom(&rqstp->rq_arg);
0888 
0889     clear_bit(XPT_OLD, &xprt->xpt_flags);
0890 
0891     xprt->xpt_ops->xpo_secure_port(rqstp);
0892     rqstp->rq_chandle.defer = svc_defer;
0893     rqstp->rq_xid = svc_getu32(&rqstp->rq_arg.head[0]);
0894 
0895     if (serv->sv_stats)
0896         serv->sv_stats->netcnt++;
0897     return len;
0898 out_release:
0899     rqstp->rq_res.len = 0;
0900     svc_xprt_release(rqstp);
0901 out:
0902     return err;
0903 }
0904 EXPORT_SYMBOL_GPL(svc_recv);
0905 
0906 /*
0907  * Drop request
0908  */
0909 void svc_drop(struct svc_rqst *rqstp)
0910 {
0911     trace_svc_drop(rqstp);
0912     svc_xprt_release(rqstp);
0913 }
0914 EXPORT_SYMBOL_GPL(svc_drop);
0915 
0916 /*
0917  * Return reply to client.
0918  */
0919 int svc_send(struct svc_rqst *rqstp)
0920 {
0921     struct svc_xprt *xprt;
0922     int     len = -EFAULT;
0923     struct xdr_buf  *xb;
0924 
0925     xprt = rqstp->rq_xprt;
0926     if (!xprt)
0927         goto out;
0928 
0929     /* calculate over-all length */
0930     xb = &rqstp->rq_res;
0931     xb->len = xb->head[0].iov_len +
0932         xb->page_len +
0933         xb->tail[0].iov_len;
0934     trace_svc_xdr_sendto(rqstp->rq_xid, xb);
0935     trace_svc_stats_latency(rqstp);
0936 
0937     len = xprt->xpt_ops->xpo_sendto(rqstp);
0938 
0939     trace_svc_send(rqstp, len);
0940     svc_xprt_release(rqstp);
0941 
0942     if (len == -ECONNREFUSED || len == -ENOTCONN || len == -EAGAIN)
0943         len = 0;
0944 out:
0945     return len;
0946 }
0947 
0948 /*
0949  * Timer function to close old temporary transports, using
0950  * a mark-and-sweep algorithm.
0951  */
0952 static void svc_age_temp_xprts(struct timer_list *t)
0953 {
0954     struct svc_serv *serv = from_timer(serv, t, sv_temptimer);
0955     struct svc_xprt *xprt;
0956     struct list_head *le, *next;
0957 
0958     dprintk("svc_age_temp_xprts\n");
0959 
0960     if (!spin_trylock_bh(&serv->sv_lock)) {
0961         /* busy, try again 1 sec later */
0962         dprintk("svc_age_temp_xprts: busy\n");
0963         mod_timer(&serv->sv_temptimer, jiffies + HZ);
0964         return;
0965     }
0966 
0967     list_for_each_safe(le, next, &serv->sv_tempsocks) {
0968         xprt = list_entry(le, struct svc_xprt, xpt_list);
0969 
0970         /* First time through, just mark it OLD. Second time
0971          * through, close it. */
0972         if (!test_and_set_bit(XPT_OLD, &xprt->xpt_flags))
0973             continue;
0974         if (kref_read(&xprt->xpt_ref) > 1 ||
0975             test_bit(XPT_BUSY, &xprt->xpt_flags))
0976             continue;
0977         list_del_init(le);
0978         set_bit(XPT_CLOSE, &xprt->xpt_flags);
0979         dprintk("queuing xprt %p for closing\n", xprt);
0980 
0981         /* a thread will dequeue and close it soon */
0982         svc_xprt_enqueue(xprt);
0983     }
0984     spin_unlock_bh(&serv->sv_lock);
0985 
0986     mod_timer(&serv->sv_temptimer, jiffies + svc_conn_age_period * HZ);
0987 }
0988 
0989 /* Close temporary transports whose xpt_local matches server_addr immediately
0990  * instead of waiting for them to be picked up by the timer.
0991  *
0992  * This is meant to be called from a notifier_block that runs when an ip
0993  * address is deleted.
0994  */
0995 void svc_age_temp_xprts_now(struct svc_serv *serv, struct sockaddr *server_addr)
0996 {
0997     struct svc_xprt *xprt;
0998     struct list_head *le, *next;
0999     LIST_HEAD(to_be_closed);
1000 
1001     spin_lock_bh(&serv->sv_lock);
1002     list_for_each_safe(le, next, &serv->sv_tempsocks) {
1003         xprt = list_entry(le, struct svc_xprt, xpt_list);
1004         if (rpc_cmp_addr(server_addr, (struct sockaddr *)
1005                 &xprt->xpt_local)) {
1006             dprintk("svc_age_temp_xprts_now: found %p\n", xprt);
1007             list_move(le, &to_be_closed);
1008         }
1009     }
1010     spin_unlock_bh(&serv->sv_lock);
1011 
1012     while (!list_empty(&to_be_closed)) {
1013         le = to_be_closed.next;
1014         list_del_init(le);
1015         xprt = list_entry(le, struct svc_xprt, xpt_list);
1016         set_bit(XPT_CLOSE, &xprt->xpt_flags);
1017         set_bit(XPT_KILL_TEMP, &xprt->xpt_flags);
1018         dprintk("svc_age_temp_xprts_now: queuing xprt %p for closing\n",
1019                 xprt);
1020         svc_xprt_enqueue(xprt);
1021     }
1022 }
1023 EXPORT_SYMBOL_GPL(svc_age_temp_xprts_now);
1024 
1025 static void call_xpt_users(struct svc_xprt *xprt)
1026 {
1027     struct svc_xpt_user *u;
1028 
1029     spin_lock(&xprt->xpt_lock);
1030     while (!list_empty(&xprt->xpt_users)) {
1031         u = list_first_entry(&xprt->xpt_users, struct svc_xpt_user, list);
1032         list_del_init(&u->list);
1033         u->callback(u);
1034     }
1035     spin_unlock(&xprt->xpt_lock);
1036 }
1037 
1038 /*
1039  * Remove a dead transport
1040  */
1041 static void svc_delete_xprt(struct svc_xprt *xprt)
1042 {
1043     struct svc_serv *serv = xprt->xpt_server;
1044     struct svc_deferred_req *dr;
1045 
1046     if (test_and_set_bit(XPT_DEAD, &xprt->xpt_flags))
1047         return;
1048 
1049     trace_svc_xprt_detach(xprt);
1050     xprt->xpt_ops->xpo_detach(xprt);
1051     if (xprt->xpt_bc_xprt)
1052         xprt->xpt_bc_xprt->ops->close(xprt->xpt_bc_xprt);
1053 
1054     spin_lock_bh(&serv->sv_lock);
1055     list_del_init(&xprt->xpt_list);
1056     WARN_ON_ONCE(!list_empty(&xprt->xpt_ready));
1057     if (test_bit(XPT_TEMP, &xprt->xpt_flags))
1058         serv->sv_tmpcnt--;
1059     spin_unlock_bh(&serv->sv_lock);
1060 
1061     while ((dr = svc_deferred_dequeue(xprt)) != NULL)
1062         kfree(dr);
1063 
1064     call_xpt_users(xprt);
1065     svc_xprt_put(xprt);
1066 }
1067 
1068 /**
1069  * svc_xprt_close - Close a client connection
1070  * @xprt: transport to disconnect
1071  *
1072  */
1073 void svc_xprt_close(struct svc_xprt *xprt)
1074 {
1075     trace_svc_xprt_close(xprt);
1076     set_bit(XPT_CLOSE, &xprt->xpt_flags);
1077     if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags))
1078         /* someone else will have to effect the close */
1079         return;
1080     /*
1081      * We expect svc_close_xprt() to work even when no threads are
1082      * running (e.g., while configuring the server before starting
1083      * any threads), so if the transport isn't busy, we delete
1084      * it ourself:
1085      */
1086     svc_delete_xprt(xprt);
1087 }
1088 EXPORT_SYMBOL_GPL(svc_xprt_close);
1089 
1090 static int svc_close_list(struct svc_serv *serv, struct list_head *xprt_list, struct net *net)
1091 {
1092     struct svc_xprt *xprt;
1093     int ret = 0;
1094 
1095     spin_lock_bh(&serv->sv_lock);
1096     list_for_each_entry(xprt, xprt_list, xpt_list) {
1097         if (xprt->xpt_net != net)
1098             continue;
1099         ret++;
1100         set_bit(XPT_CLOSE, &xprt->xpt_flags);
1101         svc_xprt_enqueue(xprt);
1102     }
1103     spin_unlock_bh(&serv->sv_lock);
1104     return ret;
1105 }
1106 
1107 static struct svc_xprt *svc_dequeue_net(struct svc_serv *serv, struct net *net)
1108 {
1109     struct svc_pool *pool;
1110     struct svc_xprt *xprt;
1111     struct svc_xprt *tmp;
1112     int i;
1113 
1114     for (i = 0; i < serv->sv_nrpools; i++) {
1115         pool = &serv->sv_pools[i];
1116 
1117         spin_lock_bh(&pool->sp_lock);
1118         list_for_each_entry_safe(xprt, tmp, &pool->sp_sockets, xpt_ready) {
1119             if (xprt->xpt_net != net)
1120                 continue;
1121             list_del_init(&xprt->xpt_ready);
1122             spin_unlock_bh(&pool->sp_lock);
1123             return xprt;
1124         }
1125         spin_unlock_bh(&pool->sp_lock);
1126     }
1127     return NULL;
1128 }
1129 
1130 static void svc_clean_up_xprts(struct svc_serv *serv, struct net *net)
1131 {
1132     struct svc_xprt *xprt;
1133 
1134     while ((xprt = svc_dequeue_net(serv, net))) {
1135         set_bit(XPT_CLOSE, &xprt->xpt_flags);
1136         svc_delete_xprt(xprt);
1137     }
1138 }
1139 
1140 /**
1141  * svc_xprt_destroy_all - Destroy transports associated with @serv
1142  * @serv: RPC service to be shut down
1143  * @net: target network namespace
1144  *
1145  * Server threads may still be running (especially in the case where the
1146  * service is still running in other network namespaces).
1147  *
1148  * So we shut down sockets the same way we would on a running server, by
1149  * setting XPT_CLOSE, enqueuing, and letting a thread pick it up to do
1150  * the close.  In the case there are no such other threads,
1151  * threads running, svc_clean_up_xprts() does a simple version of a
1152  * server's main event loop, and in the case where there are other
1153  * threads, we may need to wait a little while and then check again to
1154  * see if they're done.
1155  */
1156 void svc_xprt_destroy_all(struct svc_serv *serv, struct net *net)
1157 {
1158     int delay = 0;
1159 
1160     while (svc_close_list(serv, &serv->sv_permsocks, net) +
1161            svc_close_list(serv, &serv->sv_tempsocks, net)) {
1162 
1163         svc_clean_up_xprts(serv, net);
1164         msleep(delay++);
1165     }
1166 }
1167 EXPORT_SYMBOL_GPL(svc_xprt_destroy_all);
1168 
1169 /*
1170  * Handle defer and revisit of requests
1171  */
1172 
1173 static void svc_revisit(struct cache_deferred_req *dreq, int too_many)
1174 {
1175     struct svc_deferred_req *dr =
1176         container_of(dreq, struct svc_deferred_req, handle);
1177     struct svc_xprt *xprt = dr->xprt;
1178 
1179     spin_lock(&xprt->xpt_lock);
1180     set_bit(XPT_DEFERRED, &xprt->xpt_flags);
1181     if (too_many || test_bit(XPT_DEAD, &xprt->xpt_flags)) {
1182         spin_unlock(&xprt->xpt_lock);
1183         trace_svc_defer_drop(dr);
1184         svc_xprt_put(xprt);
1185         kfree(dr);
1186         return;
1187     }
1188     dr->xprt = NULL;
1189     list_add(&dr->handle.recent, &xprt->xpt_deferred);
1190     spin_unlock(&xprt->xpt_lock);
1191     trace_svc_defer_queue(dr);
1192     svc_xprt_enqueue(xprt);
1193     svc_xprt_put(xprt);
1194 }
1195 
1196 /*
1197  * Save the request off for later processing. The request buffer looks
1198  * like this:
1199  *
1200  * <xprt-header><rpc-header><rpc-pagelist><rpc-tail>
1201  *
1202  * This code can only handle requests that consist of an xprt-header
1203  * and rpc-header.
1204  */
1205 static struct cache_deferred_req *svc_defer(struct cache_req *req)
1206 {
1207     struct svc_rqst *rqstp = container_of(req, struct svc_rqst, rq_chandle);
1208     struct svc_deferred_req *dr;
1209 
1210     if (rqstp->rq_arg.page_len || !test_bit(RQ_USEDEFERRAL, &rqstp->rq_flags))
1211         return NULL; /* if more than a page, give up FIXME */
1212     if (rqstp->rq_deferred) {
1213         dr = rqstp->rq_deferred;
1214         rqstp->rq_deferred = NULL;
1215     } else {
1216         size_t skip;
1217         size_t size;
1218         /* FIXME maybe discard if size too large */
1219         size = sizeof(struct svc_deferred_req) + rqstp->rq_arg.len;
1220         dr = kmalloc(size, GFP_KERNEL);
1221         if (dr == NULL)
1222             return NULL;
1223 
1224         dr->handle.owner = rqstp->rq_server;
1225         dr->prot = rqstp->rq_prot;
1226         memcpy(&dr->addr, &rqstp->rq_addr, rqstp->rq_addrlen);
1227         dr->addrlen = rqstp->rq_addrlen;
1228         dr->daddr = rqstp->rq_daddr;
1229         dr->argslen = rqstp->rq_arg.len >> 2;
1230         dr->xprt_ctxt = rqstp->rq_xprt_ctxt;
1231         rqstp->rq_xprt_ctxt = NULL;
1232 
1233         /* back up head to the start of the buffer and copy */
1234         skip = rqstp->rq_arg.len - rqstp->rq_arg.head[0].iov_len;
1235         memcpy(dr->args, rqstp->rq_arg.head[0].iov_base - skip,
1236                dr->argslen << 2);
1237     }
1238     trace_svc_defer(rqstp);
1239     svc_xprt_get(rqstp->rq_xprt);
1240     dr->xprt = rqstp->rq_xprt;
1241     __set_bit(RQ_DROPME, &rqstp->rq_flags);
1242 
1243     dr->handle.revisit = svc_revisit;
1244     return &dr->handle;
1245 }
1246 
1247 /*
1248  * recv data from a deferred request into an active one
1249  */
1250 static noinline int svc_deferred_recv(struct svc_rqst *rqstp)
1251 {
1252     struct svc_deferred_req *dr = rqstp->rq_deferred;
1253 
1254     trace_svc_defer_recv(dr);
1255 
1256     /* setup iov_base past transport header */
1257     rqstp->rq_arg.head[0].iov_base = dr->args;
1258     /* The iov_len does not include the transport header bytes */
1259     rqstp->rq_arg.head[0].iov_len = dr->argslen << 2;
1260     rqstp->rq_arg.page_len = 0;
1261     /* The rq_arg.len includes the transport header bytes */
1262     rqstp->rq_arg.len     = dr->argslen << 2;
1263     rqstp->rq_prot        = dr->prot;
1264     memcpy(&rqstp->rq_addr, &dr->addr, dr->addrlen);
1265     rqstp->rq_addrlen     = dr->addrlen;
1266     /* Save off transport header len in case we get deferred again */
1267     rqstp->rq_daddr       = dr->daddr;
1268     rqstp->rq_respages    = rqstp->rq_pages;
1269     rqstp->rq_xprt_ctxt   = dr->xprt_ctxt;
1270     svc_xprt_received(rqstp->rq_xprt);
1271     return dr->argslen << 2;
1272 }
1273 
1274 
1275 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt)
1276 {
1277     struct svc_deferred_req *dr = NULL;
1278 
1279     if (!test_bit(XPT_DEFERRED, &xprt->xpt_flags))
1280         return NULL;
1281     spin_lock(&xprt->xpt_lock);
1282     if (!list_empty(&xprt->xpt_deferred)) {
1283         dr = list_entry(xprt->xpt_deferred.next,
1284                 struct svc_deferred_req,
1285                 handle.recent);
1286         list_del_init(&dr->handle.recent);
1287     } else
1288         clear_bit(XPT_DEFERRED, &xprt->xpt_flags);
1289     spin_unlock(&xprt->xpt_lock);
1290     return dr;
1291 }
1292 
1293 /**
1294  * svc_find_xprt - find an RPC transport instance
1295  * @serv: pointer to svc_serv to search
1296  * @xcl_name: C string containing transport's class name
1297  * @net: owner net pointer
1298  * @af: Address family of transport's local address
1299  * @port: transport's IP port number
1300  *
1301  * Return the transport instance pointer for the endpoint accepting
1302  * connections/peer traffic from the specified transport class,
1303  * address family and port.
1304  *
1305  * Specifying 0 for the address family or port is effectively a
1306  * wild-card, and will result in matching the first transport in the
1307  * service's list that has a matching class name.
1308  */
1309 struct svc_xprt *svc_find_xprt(struct svc_serv *serv, const char *xcl_name,
1310                    struct net *net, const sa_family_t af,
1311                    const unsigned short port)
1312 {
1313     struct svc_xprt *xprt;
1314     struct svc_xprt *found = NULL;
1315 
1316     /* Sanity check the args */
1317     if (serv == NULL || xcl_name == NULL)
1318         return found;
1319 
1320     spin_lock_bh(&serv->sv_lock);
1321     list_for_each_entry(xprt, &serv->sv_permsocks, xpt_list) {
1322         if (xprt->xpt_net != net)
1323             continue;
1324         if (strcmp(xprt->xpt_class->xcl_name, xcl_name))
1325             continue;
1326         if (af != AF_UNSPEC && af != xprt->xpt_local.ss_family)
1327             continue;
1328         if (port != 0 && port != svc_xprt_local_port(xprt))
1329             continue;
1330         found = xprt;
1331         svc_xprt_get(xprt);
1332         break;
1333     }
1334     spin_unlock_bh(&serv->sv_lock);
1335     return found;
1336 }
1337 EXPORT_SYMBOL_GPL(svc_find_xprt);
1338 
1339 static int svc_one_xprt_name(const struct svc_xprt *xprt,
1340                  char *pos, int remaining)
1341 {
1342     int len;
1343 
1344     len = snprintf(pos, remaining, "%s %u\n",
1345             xprt->xpt_class->xcl_name,
1346             svc_xprt_local_port(xprt));
1347     if (len >= remaining)
1348         return -ENAMETOOLONG;
1349     return len;
1350 }
1351 
1352 /**
1353  * svc_xprt_names - format a buffer with a list of transport names
1354  * @serv: pointer to an RPC service
1355  * @buf: pointer to a buffer to be filled in
1356  * @buflen: length of buffer to be filled in
1357  *
1358  * Fills in @buf with a string containing a list of transport names,
1359  * each name terminated with '\n'.
1360  *
1361  * Returns positive length of the filled-in string on success; otherwise
1362  * a negative errno value is returned if an error occurs.
1363  */
1364 int svc_xprt_names(struct svc_serv *serv, char *buf, const int buflen)
1365 {
1366     struct svc_xprt *xprt;
1367     int len, totlen;
1368     char *pos;
1369 
1370     /* Sanity check args */
1371     if (!serv)
1372         return 0;
1373 
1374     spin_lock_bh(&serv->sv_lock);
1375 
1376     pos = buf;
1377     totlen = 0;
1378     list_for_each_entry(xprt, &serv->sv_permsocks, xpt_list) {
1379         len = svc_one_xprt_name(xprt, pos, buflen - totlen);
1380         if (len < 0) {
1381             *buf = '\0';
1382             totlen = len;
1383         }
1384         if (len <= 0)
1385             break;
1386 
1387         pos += len;
1388         totlen += len;
1389     }
1390 
1391     spin_unlock_bh(&serv->sv_lock);
1392     return totlen;
1393 }
1394 EXPORT_SYMBOL_GPL(svc_xprt_names);
1395 
1396 
1397 /*----------------------------------------------------------------------------*/
1398 
1399 static void *svc_pool_stats_start(struct seq_file *m, loff_t *pos)
1400 {
1401     unsigned int pidx = (unsigned int)*pos;
1402     struct svc_serv *serv = m->private;
1403 
1404     dprintk("svc_pool_stats_start, *pidx=%u\n", pidx);
1405 
1406     if (!pidx)
1407         return SEQ_START_TOKEN;
1408     return (pidx > serv->sv_nrpools ? NULL : &serv->sv_pools[pidx-1]);
1409 }
1410 
1411 static void *svc_pool_stats_next(struct seq_file *m, void *p, loff_t *pos)
1412 {
1413     struct svc_pool *pool = p;
1414     struct svc_serv *serv = m->private;
1415 
1416     dprintk("svc_pool_stats_next, *pos=%llu\n", *pos);
1417 
1418     if (p == SEQ_START_TOKEN) {
1419         pool = &serv->sv_pools[0];
1420     } else {
1421         unsigned int pidx = (pool - &serv->sv_pools[0]);
1422         if (pidx < serv->sv_nrpools-1)
1423             pool = &serv->sv_pools[pidx+1];
1424         else
1425             pool = NULL;
1426     }
1427     ++*pos;
1428     return pool;
1429 }
1430 
1431 static void svc_pool_stats_stop(struct seq_file *m, void *p)
1432 {
1433 }
1434 
1435 static int svc_pool_stats_show(struct seq_file *m, void *p)
1436 {
1437     struct svc_pool *pool = p;
1438 
1439     if (p == SEQ_START_TOKEN) {
1440         seq_puts(m, "# pool packets-arrived sockets-enqueued threads-woken threads-timedout\n");
1441         return 0;
1442     }
1443 
1444     seq_printf(m, "%u %lu %lu %lu %lu\n",
1445         pool->sp_id,
1446         (unsigned long)atomic_long_read(&pool->sp_stats.packets),
1447         pool->sp_stats.sockets_queued,
1448         (unsigned long)atomic_long_read(&pool->sp_stats.threads_woken),
1449         (unsigned long)atomic_long_read(&pool->sp_stats.threads_timedout));
1450 
1451     return 0;
1452 }
1453 
1454 static const struct seq_operations svc_pool_stats_seq_ops = {
1455     .start  = svc_pool_stats_start,
1456     .next   = svc_pool_stats_next,
1457     .stop   = svc_pool_stats_stop,
1458     .show   = svc_pool_stats_show,
1459 };
1460 
1461 int svc_pool_stats_open(struct svc_serv *serv, struct file *file)
1462 {
1463     int err;
1464 
1465     err = seq_open(file, &svc_pool_stats_seq_ops);
1466     if (!err)
1467         ((struct seq_file *) file->private_data)->private = serv;
1468     return err;
1469 }
1470 EXPORT_SYMBOL(svc_pool_stats_open);
1471 
1472 /*----------------------------------------------------------------------------*/
[ Source navigation ] [ Diff markup ] [ Identifier search ] [ general search ]

This page was automatically generated by the 2.1.0 LXR engine.

The LXR team
Valid CSS 2.1! Valid HTML 4.01!