1 // SPDX-License-Identifier: GPL-2.0-only
2 /*
3  * linux/net/sunrpc/svc_xprt.c
4  *
5  * Author: Tom Tucker <tom@opengridcomputing.com>
6  */
7 
8 #include <linux/sched.h>
9 #include <linux/errno.h>
10 #include <linux/freezer.h>
11 #include <linux/kthread.h>
12 #include <linux/slab.h>
13 #include <net/sock.h>
14 #include <linux/sunrpc/addr.h>
15 #include <linux/sunrpc/stats.h>
16 #include <linux/sunrpc/svc_xprt.h>
17 #include <linux/sunrpc/svcsock.h>
18 #include <linux/sunrpc/xprt.h>
19 #include <linux/module.h>
20 #include <linux/netdevice.h>
21 #include <trace/events/sunrpc.h>
22 
23 #define RPCDBG_FACILITY	RPCDBG_SVCXPRT
24 
25 static unsigned int svc_rpc_per_connection_limit __read_mostly;
26 module_param(svc_rpc_per_connection_limit, uint, 0644);
27 
28 
29 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt);
30 static int svc_deferred_recv(struct svc_rqst *rqstp);
31 static struct cache_deferred_req *svc_defer(struct cache_req *req);
32 static void svc_age_temp_xprts(struct timer_list *t);
33 static void svc_delete_xprt(struct svc_xprt *xprt);
34 
35 /* apparently the "standard" is that clients close
36  * idle connections after 5 minutes, servers after
37  * 6 minutes
38  *   http://nfsv4bat.org/Documents/ConnectAThon/1996/nfstcp.pdf
39  */
40 static int svc_conn_age_period = 6*60;
41 
42 /* List of registered transport classes */
43 static DEFINE_SPINLOCK(svc_xprt_class_lock);
44 static LIST_HEAD(svc_xprt_class_list);
45 
46 /* SMP locking strategy:
47  *
48  *	svc_pool->sp_lock protects most of the fields of that pool.
49  *	svc_serv->sv_lock protects sv_tempsocks, sv_permsocks, sv_tmpcnt.
50  *	when both need to be taken (rare), svc_serv->sv_lock is first.
51  *	The "service mutex" protects svc_serv->sv_nrthread.
52  *	svc_sock->sk_lock protects the svc_sock->sk_deferred list
53  *             and the ->sk_info_authunix cache.
54  *
55  *	The XPT_BUSY bit in xprt->xpt_flags prevents a transport being
56  *	enqueued multiply. During normal transport processing this bit
57  *	is set by svc_xprt_enqueue and cleared by svc_xprt_received.
58  *	Providers should not manipulate this bit directly.
59  *
60  *	Some flags can be set to certain values at any time
61  *	providing that certain rules are followed:
62  *
63  *	XPT_CONN, XPT_DATA:
64  *		- Can be set or cleared at any time.
65  *		- After a set, svc_xprt_enqueue must be called to enqueue
66  *		  the transport for processing.
67  *		- After a clear, the transport must be read/accepted.
68  *		  If this succeeds, it must be set again.
69  *	XPT_CLOSE:
70  *		- Can set at any time. It is never cleared.
71  *      XPT_DEAD:
72  *		- Can only be set while XPT_BUSY is held which ensures
73  *		  that no other thread will be using the transport or will
74  *		  try to set XPT_DEAD.
75  */
svc_reg_xprt_class(struct svc_xprt_class * xcl)76 int svc_reg_xprt_class(struct svc_xprt_class *xcl)
77 {
78 	struct svc_xprt_class *cl;
79 	int res = -EEXIST;
80 
81 	dprintk("svc: Adding svc transport class '%s'\n", xcl->xcl_name);
82 
83 	INIT_LIST_HEAD(&xcl->xcl_list);
84 	spin_lock(&svc_xprt_class_lock);
85 	/* Make sure there isn't already a class with the same name */
86 	list_for_each_entry(cl, &svc_xprt_class_list, xcl_list) {
87 		if (strcmp(xcl->xcl_name, cl->xcl_name) == 0)
88 			goto out;
89 	}
90 	list_add_tail(&xcl->xcl_list, &svc_xprt_class_list);
91 	res = 0;
92 out:
93 	spin_unlock(&svc_xprt_class_lock);
94 	return res;
95 }
96 EXPORT_SYMBOL_GPL(svc_reg_xprt_class);
97 
svc_unreg_xprt_class(struct svc_xprt_class * xcl)98 void svc_unreg_xprt_class(struct svc_xprt_class *xcl)
99 {
100 	dprintk("svc: Removing svc transport class '%s'\n", xcl->xcl_name);
101 	spin_lock(&svc_xprt_class_lock);
102 	list_del_init(&xcl->xcl_list);
103 	spin_unlock(&svc_xprt_class_lock);
104 }
105 EXPORT_SYMBOL_GPL(svc_unreg_xprt_class);
106 
107 /**
108  * svc_print_xprts - Format the transport list for printing
109  * @buf: target buffer for formatted address
110  * @maxlen: length of target buffer
111  *
112  * Fills in @buf with a string containing a list of transport names, each name
113  * terminated with '\n'. If the buffer is too small, some entries may be
114  * missing, but it is guaranteed that all lines in the output buffer are
115  * complete.
116  *
117  * Returns positive length of the filled-in string.
118  */
svc_print_xprts(char * buf,int maxlen)119 int svc_print_xprts(char *buf, int maxlen)
120 {
121 	struct svc_xprt_class *xcl;
122 	char tmpstr[80];
123 	int len = 0;
124 	buf[0] = '\0';
125 
126 	spin_lock(&svc_xprt_class_lock);
127 	list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) {
128 		int slen;
129 
130 		slen = snprintf(tmpstr, sizeof(tmpstr), "%s %d\n",
131 				xcl->xcl_name, xcl->xcl_max_payload);
132 		if (slen >= sizeof(tmpstr) || len + slen >= maxlen)
133 			break;
134 		len += slen;
135 		strcat(buf, tmpstr);
136 	}
137 	spin_unlock(&svc_xprt_class_lock);
138 
139 	return len;
140 }
141 
142 /**
143  * svc_xprt_deferred_close - Close a transport
144  * @xprt: transport instance
145  *
146  * Used in contexts that need to defer the work of shutting down
147  * the transport to an nfsd thread.
148  */
svc_xprt_deferred_close(struct svc_xprt * xprt)149 void svc_xprt_deferred_close(struct svc_xprt *xprt)
150 {
151 	if (!test_and_set_bit(XPT_CLOSE, &xprt->xpt_flags))
152 		svc_xprt_enqueue(xprt);
153 }
154 EXPORT_SYMBOL_GPL(svc_xprt_deferred_close);
155 
svc_xprt_free(struct kref * kref)156 static void svc_xprt_free(struct kref *kref)
157 {
158 	struct svc_xprt *xprt =
159 		container_of(kref, struct svc_xprt, xpt_ref);
160 	struct module *owner = xprt->xpt_class->xcl_owner;
161 	if (test_bit(XPT_CACHE_AUTH, &xprt->xpt_flags))
162 		svcauth_unix_info_release(xprt);
163 	put_cred(xprt->xpt_cred);
164 	put_net(xprt->xpt_net);
165 	/* See comment on corresponding get in xs_setup_bc_tcp(): */
166 	if (xprt->xpt_bc_xprt)
167 		xprt_put(xprt->xpt_bc_xprt);
168 	if (xprt->xpt_bc_xps)
169 		xprt_switch_put(xprt->xpt_bc_xps);
170 	trace_svc_xprt_free(xprt);
171 	xprt->xpt_ops->xpo_free(xprt);
172 	module_put(owner);
173 }
174 
svc_xprt_put(struct svc_xprt * xprt)175 void svc_xprt_put(struct svc_xprt *xprt)
176 {
177 	kref_put(&xprt->xpt_ref, svc_xprt_free);
178 }
179 EXPORT_SYMBOL_GPL(svc_xprt_put);
180 
181 /*
182  * Called by transport drivers to initialize the transport independent
183  * portion of the transport instance.
184  */
svc_xprt_init(struct net * net,struct svc_xprt_class * xcl,struct svc_xprt * xprt,struct svc_serv * serv)185 void svc_xprt_init(struct net *net, struct svc_xprt_class *xcl,
186 		   struct svc_xprt *xprt, struct svc_serv *serv)
187 {
188 	memset(xprt, 0, sizeof(*xprt));
189 	xprt->xpt_class = xcl;
190 	xprt->xpt_ops = xcl->xcl_ops;
191 	kref_init(&xprt->xpt_ref);
192 	xprt->xpt_server = serv;
193 	INIT_LIST_HEAD(&xprt->xpt_list);
194 	INIT_LIST_HEAD(&xprt->xpt_ready);
195 	INIT_LIST_HEAD(&xprt->xpt_deferred);
196 	INIT_LIST_HEAD(&xprt->xpt_users);
197 	mutex_init(&xprt->xpt_mutex);
198 	spin_lock_init(&xprt->xpt_lock);
199 	set_bit(XPT_BUSY, &xprt->xpt_flags);
200 	xprt->xpt_net = get_net(net);
201 	strcpy(xprt->xpt_remotebuf, "uninitialized");
202 }
203 EXPORT_SYMBOL_GPL(svc_xprt_init);
204 
__svc_xpo_create(struct svc_xprt_class * xcl,struct svc_serv * serv,struct net * net,const int family,const unsigned short port,int flags)205 static struct svc_xprt *__svc_xpo_create(struct svc_xprt_class *xcl,
206 					 struct svc_serv *serv,
207 					 struct net *net,
208 					 const int family,
209 					 const unsigned short port,
210 					 int flags)
211 {
212 	struct sockaddr_in sin = {
213 		.sin_family		= AF_INET,
214 		.sin_addr.s_addr	= htonl(INADDR_ANY),
215 		.sin_port		= htons(port),
216 	};
217 #if IS_ENABLED(CONFIG_IPV6)
218 	struct sockaddr_in6 sin6 = {
219 		.sin6_family		= AF_INET6,
220 		.sin6_addr		= IN6ADDR_ANY_INIT,
221 		.sin6_port		= htons(port),
222 	};
223 #endif
224 	struct svc_xprt *xprt;
225 	struct sockaddr *sap;
226 	size_t len;
227 
228 	switch (family) {
229 	case PF_INET:
230 		sap = (struct sockaddr *)&sin;
231 		len = sizeof(sin);
232 		break;
233 #if IS_ENABLED(CONFIG_IPV6)
234 	case PF_INET6:
235 		sap = (struct sockaddr *)&sin6;
236 		len = sizeof(sin6);
237 		break;
238 #endif
239 	default:
240 		return ERR_PTR(-EAFNOSUPPORT);
241 	}
242 
243 	xprt = xcl->xcl_ops->xpo_create(serv, net, sap, len, flags);
244 	if (IS_ERR(xprt))
245 		trace_svc_xprt_create_err(serv->sv_program->pg_name,
246 					  xcl->xcl_name, sap, xprt);
247 	return xprt;
248 }
249 
250 /**
251  * svc_xprt_received - start next receiver thread
252  * @xprt: controlling transport
253  *
254  * The caller must hold the XPT_BUSY bit and must
255  * not thereafter touch transport data.
256  *
257  * Note: XPT_DATA only gets cleared when a read-attempt finds no (or
258  * insufficient) data.
259  */
svc_xprt_received(struct svc_xprt * xprt)260 void svc_xprt_received(struct svc_xprt *xprt)
261 {
262 	if (!test_bit(XPT_BUSY, &xprt->xpt_flags)) {
263 		WARN_ONCE(1, "xprt=0x%p already busy!", xprt);
264 		return;
265 	}
266 
267 	trace_svc_xprt_received(xprt);
268 
269 	/* As soon as we clear busy, the xprt could be closed and
270 	 * 'put', so we need a reference to call svc_enqueue_xprt with:
271 	 */
272 	svc_xprt_get(xprt);
273 	smp_mb__before_atomic();
274 	clear_bit(XPT_BUSY, &xprt->xpt_flags);
275 	xprt->xpt_server->sv_ops->svo_enqueue_xprt(xprt);
276 	svc_xprt_put(xprt);
277 }
278 EXPORT_SYMBOL_GPL(svc_xprt_received);
279 
svc_add_new_perm_xprt(struct svc_serv * serv,struct svc_xprt * new)280 void svc_add_new_perm_xprt(struct svc_serv *serv, struct svc_xprt *new)
281 {
282 	clear_bit(XPT_TEMP, &new->xpt_flags);
283 	spin_lock_bh(&serv->sv_lock);
284 	list_add(&new->xpt_list, &serv->sv_permsocks);
285 	spin_unlock_bh(&serv->sv_lock);
286 	svc_xprt_received(new);
287 }
288 
_svc_create_xprt(struct svc_serv * serv,const char * xprt_name,struct net * net,const int family,const unsigned short port,int flags,const struct cred * cred)289 static int _svc_create_xprt(struct svc_serv *serv, const char *xprt_name,
290 			    struct net *net, const int family,
291 			    const unsigned short port, int flags,
292 			    const struct cred *cred)
293 {
294 	struct svc_xprt_class *xcl;
295 
296 	spin_lock(&svc_xprt_class_lock);
297 	list_for_each_entry(xcl, &svc_xprt_class_list, xcl_list) {
298 		struct svc_xprt *newxprt;
299 		unsigned short newport;
300 
301 		if (strcmp(xprt_name, xcl->xcl_name))
302 			continue;
303 
304 		if (!try_module_get(xcl->xcl_owner))
305 			goto err;
306 
307 		spin_unlock(&svc_xprt_class_lock);
308 		newxprt = __svc_xpo_create(xcl, serv, net, family, port, flags);
309 		if (IS_ERR(newxprt)) {
310 			module_put(xcl->xcl_owner);
311 			return PTR_ERR(newxprt);
312 		}
313 		newxprt->xpt_cred = get_cred(cred);
314 		svc_add_new_perm_xprt(serv, newxprt);
315 		newport = svc_xprt_local_port(newxprt);
316 		return newport;
317 	}
318  err:
319 	spin_unlock(&svc_xprt_class_lock);
320 	/* This errno is exposed to user space.  Provide a reasonable
321 	 * perror msg for a bad transport. */
322 	return -EPROTONOSUPPORT;
323 }
324 
svc_create_xprt(struct svc_serv * serv,const char * xprt_name,struct net * net,const int family,const unsigned short port,int flags,const struct cred * cred)325 int svc_create_xprt(struct svc_serv *serv, const char *xprt_name,
326 		    struct net *net, const int family,
327 		    const unsigned short port, int flags,
328 		    const struct cred *cred)
329 {
330 	int err;
331 
332 	err = _svc_create_xprt(serv, xprt_name, net, family, port, flags, cred);
333 	if (err == -EPROTONOSUPPORT) {
334 		request_module("svc%s", xprt_name);
335 		err = _svc_create_xprt(serv, xprt_name, net, family, port, flags, cred);
336 	}
337 	return err;
338 }
339 EXPORT_SYMBOL_GPL(svc_create_xprt);
340 
341 /*
342  * Copy the local and remote xprt addresses to the rqstp structure
343  */
svc_xprt_copy_addrs(struct svc_rqst * rqstp,struct svc_xprt * xprt)344 void svc_xprt_copy_addrs(struct svc_rqst *rqstp, struct svc_xprt *xprt)
345 {
346 	memcpy(&rqstp->rq_addr, &xprt->xpt_remote, xprt->xpt_remotelen);
347 	rqstp->rq_addrlen = xprt->xpt_remotelen;
348 
349 	/*
350 	 * Destination address in request is needed for binding the
351 	 * source address in RPC replies/callbacks later.
352 	 */
353 	memcpy(&rqstp->rq_daddr, &xprt->xpt_local, xprt->xpt_locallen);
354 	rqstp->rq_daddrlen = xprt->xpt_locallen;
355 }
356 EXPORT_SYMBOL_GPL(svc_xprt_copy_addrs);
357 
358 /**
359  * svc_print_addr - Format rq_addr field for printing
360  * @rqstp: svc_rqst struct containing address to print
361  * @buf: target buffer for formatted address
362  * @len: length of target buffer
363  *
364  */
svc_print_addr(struct svc_rqst * rqstp,char * buf,size_t len)365 char *svc_print_addr(struct svc_rqst *rqstp, char *buf, size_t len)
366 {
367 	return __svc_print_addr(svc_addr(rqstp), buf, len);
368 }
369 EXPORT_SYMBOL_GPL(svc_print_addr);
370 
svc_xprt_slots_in_range(struct svc_xprt * xprt)371 static bool svc_xprt_slots_in_range(struct svc_xprt *xprt)
372 {
373 	unsigned int limit = svc_rpc_per_connection_limit;
374 	int nrqsts = atomic_read(&xprt->xpt_nr_rqsts);
375 
376 	return limit == 0 || (nrqsts >= 0 && nrqsts < limit);
377 }
378 
svc_xprt_reserve_slot(struct svc_rqst * rqstp,struct svc_xprt * xprt)379 static bool svc_xprt_reserve_slot(struct svc_rqst *rqstp, struct svc_xprt *xprt)
380 {
381 	if (!test_bit(RQ_DATA, &rqstp->rq_flags)) {
382 		if (!svc_xprt_slots_in_range(xprt))
383 			return false;
384 		atomic_inc(&xprt->xpt_nr_rqsts);
385 		set_bit(RQ_DATA, &rqstp->rq_flags);
386 	}
387 	return true;
388 }
389 
svc_xprt_release_slot(struct svc_rqst * rqstp)390 static void svc_xprt_release_slot(struct svc_rqst *rqstp)
391 {
392 	struct svc_xprt	*xprt = rqstp->rq_xprt;
393 	if (test_and_clear_bit(RQ_DATA, &rqstp->rq_flags)) {
394 		atomic_dec(&xprt->xpt_nr_rqsts);
395 		smp_wmb(); /* See smp_rmb() in svc_xprt_ready() */
396 		svc_xprt_enqueue(xprt);
397 	}
398 }
399 
svc_xprt_ready(struct svc_xprt * xprt)400 static bool svc_xprt_ready(struct svc_xprt *xprt)
401 {
402 	unsigned long xpt_flags;
403 
404 	/*
405 	 * If another cpu has recently updated xpt_flags,
406 	 * sk_sock->flags, xpt_reserved, or xpt_nr_rqsts, we need to
407 	 * know about it; otherwise it's possible that both that cpu and
408 	 * this one could call svc_xprt_enqueue() without either
409 	 * svc_xprt_enqueue() recognizing that the conditions below
410 	 * are satisfied, and we could stall indefinitely:
411 	 */
412 	smp_rmb();
413 	xpt_flags = READ_ONCE(xprt->xpt_flags);
414 
415 	if (xpt_flags & (BIT(XPT_CONN) | BIT(XPT_CLOSE)))
416 		return true;
417 	if (xpt_flags & (BIT(XPT_DATA) | BIT(XPT_DEFERRED))) {
418 		if (xprt->xpt_ops->xpo_has_wspace(xprt) &&
419 		    svc_xprt_slots_in_range(xprt))
420 			return true;
421 		trace_svc_xprt_no_write_space(xprt);
422 		return false;
423 	}
424 	return false;
425 }
426 
svc_xprt_do_enqueue(struct svc_xprt * xprt)427 void svc_xprt_do_enqueue(struct svc_xprt *xprt)
428 {
429 	struct svc_pool *pool;
430 	struct svc_rqst	*rqstp = NULL;
431 	int cpu;
432 
433 	if (!svc_xprt_ready(xprt))
434 		return;
435 
436 	/* Mark transport as busy. It will remain in this state until
437 	 * the provider calls svc_xprt_received. We update XPT_BUSY
438 	 * atomically because it also guards against trying to enqueue
439 	 * the transport twice.
440 	 */
441 	if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags))
442 		return;
443 
444 	cpu = get_cpu();
445 	pool = svc_pool_for_cpu(xprt->xpt_server, cpu);
446 
447 	atomic_long_inc(&pool->sp_stats.packets);
448 
449 	spin_lock_bh(&pool->sp_lock);
450 	list_add_tail(&xprt->xpt_ready, &pool->sp_sockets);
451 	pool->sp_stats.sockets_queued++;
452 	spin_unlock_bh(&pool->sp_lock);
453 
454 	/* find a thread for this xprt */
455 	rcu_read_lock();
456 	list_for_each_entry_rcu(rqstp, &pool->sp_all_threads, rq_all) {
457 		if (test_and_set_bit(RQ_BUSY, &rqstp->rq_flags))
458 			continue;
459 		atomic_long_inc(&pool->sp_stats.threads_woken);
460 		rqstp->rq_qtime = ktime_get();
461 		wake_up_process(rqstp->rq_task);
462 		goto out_unlock;
463 	}
464 	set_bit(SP_CONGESTED, &pool->sp_flags);
465 	rqstp = NULL;
466 out_unlock:
467 	rcu_read_unlock();
468 	put_cpu();
469 	trace_svc_xprt_do_enqueue(xprt, rqstp);
470 }
471 EXPORT_SYMBOL_GPL(svc_xprt_do_enqueue);
472 
473 /*
474  * Queue up a transport with data pending. If there are idle nfsd
475  * processes, wake 'em up.
476  *
477  */
svc_xprt_enqueue(struct svc_xprt * xprt)478 void svc_xprt_enqueue(struct svc_xprt *xprt)
479 {
480 	if (test_bit(XPT_BUSY, &xprt->xpt_flags))
481 		return;
482 	xprt->xpt_server->sv_ops->svo_enqueue_xprt(xprt);
483 }
484 EXPORT_SYMBOL_GPL(svc_xprt_enqueue);
485 
486 /*
487  * Dequeue the first transport, if there is one.
488  */
svc_xprt_dequeue(struct svc_pool * pool)489 static struct svc_xprt *svc_xprt_dequeue(struct svc_pool *pool)
490 {
491 	struct svc_xprt	*xprt = NULL;
492 
493 	if (list_empty(&pool->sp_sockets))
494 		goto out;
495 
496 	spin_lock_bh(&pool->sp_lock);
497 	if (likely(!list_empty(&pool->sp_sockets))) {
498 		xprt = list_first_entry(&pool->sp_sockets,
499 					struct svc_xprt, xpt_ready);
500 		list_del_init(&xprt->xpt_ready);
501 		svc_xprt_get(xprt);
502 	}
503 	spin_unlock_bh(&pool->sp_lock);
504 out:
505 	return xprt;
506 }
507 
508 /**
509  * svc_reserve - change the space reserved for the reply to a request.
510  * @rqstp:  The request in question
511  * @space: new max space to reserve
512  *
513  * Each request reserves some space on the output queue of the transport
514  * to make sure the reply fits.  This function reduces that reserved
515  * space to be the amount of space used already, plus @space.
516  *
517  */
svc_reserve(struct svc_rqst * rqstp,int space)518 void svc_reserve(struct svc_rqst *rqstp, int space)
519 {
520 	struct svc_xprt *xprt = rqstp->rq_xprt;
521 
522 	space += rqstp->rq_res.head[0].iov_len;
523 
524 	if (xprt && space < rqstp->rq_reserved) {
525 		atomic_sub((rqstp->rq_reserved - space), &xprt->xpt_reserved);
526 		rqstp->rq_reserved = space;
527 		smp_wmb(); /* See smp_rmb() in svc_xprt_ready() */
528 		svc_xprt_enqueue(xprt);
529 	}
530 }
531 EXPORT_SYMBOL_GPL(svc_reserve);
532 
svc_xprt_release(struct svc_rqst * rqstp)533 static void svc_xprt_release(struct svc_rqst *rqstp)
534 {
535 	struct svc_xprt	*xprt = rqstp->rq_xprt;
536 
537 	xprt->xpt_ops->xpo_release_rqst(rqstp);
538 
539 	kfree(rqstp->rq_deferred);
540 	rqstp->rq_deferred = NULL;
541 
542 	pagevec_release(&rqstp->rq_pvec);
543 	svc_free_res_pages(rqstp);
544 	rqstp->rq_res.page_len = 0;
545 	rqstp->rq_res.page_base = 0;
546 
547 	/* Reset response buffer and release
548 	 * the reservation.
549 	 * But first, check that enough space was reserved
550 	 * for the reply, otherwise we have a bug!
551 	 */
552 	if ((rqstp->rq_res.len) >  rqstp->rq_reserved)
553 		printk(KERN_ERR "RPC request reserved %d but used %d\n",
554 		       rqstp->rq_reserved,
555 		       rqstp->rq_res.len);
556 
557 	rqstp->rq_res.head[0].iov_len = 0;
558 	svc_reserve(rqstp, 0);
559 	svc_xprt_release_slot(rqstp);
560 	rqstp->rq_xprt = NULL;
561 	svc_xprt_put(xprt);
562 }
563 
564 /*
565  * Some svc_serv's will have occasional work to do, even when a xprt is not
566  * waiting to be serviced. This function is there to "kick" a task in one of
567  * those services so that it can wake up and do that work. Note that we only
568  * bother with pool 0 as we don't need to wake up more than one thread for
569  * this purpose.
570  */
svc_wake_up(struct svc_serv * serv)571 void svc_wake_up(struct svc_serv *serv)
572 {
573 	struct svc_rqst	*rqstp;
574 	struct svc_pool *pool;
575 
576 	pool = &serv->sv_pools[0];
577 
578 	rcu_read_lock();
579 	list_for_each_entry_rcu(rqstp, &pool->sp_all_threads, rq_all) {
580 		/* skip any that aren't queued */
581 		if (test_bit(RQ_BUSY, &rqstp->rq_flags))
582 			continue;
583 		rcu_read_unlock();
584 		wake_up_process(rqstp->rq_task);
585 		trace_svc_wake_up(rqstp->rq_task->pid);
586 		return;
587 	}
588 	rcu_read_unlock();
589 
590 	/* No free entries available */
591 	set_bit(SP_TASK_PENDING, &pool->sp_flags);
592 	smp_wmb();
593 	trace_svc_wake_up(0);
594 }
595 EXPORT_SYMBOL_GPL(svc_wake_up);
596 
svc_port_is_privileged(struct sockaddr * sin)597 int svc_port_is_privileged(struct sockaddr *sin)
598 {
599 	switch (sin->sa_family) {
600 	case AF_INET:
601 		return ntohs(((struct sockaddr_in *)sin)->sin_port)
602 			< PROT_SOCK;
603 	case AF_INET6:
604 		return ntohs(((struct sockaddr_in6 *)sin)->sin6_port)
605 			< PROT_SOCK;
606 	default:
607 		return 0;
608 	}
609 }
610 
611 /*
612  * Make sure that we don't have too many active connections. If we have,
613  * something must be dropped. It's not clear what will happen if we allow
614  * "too many" connections, but when dealing with network-facing software,
615  * we have to code defensively. Here we do that by imposing hard limits.
616  *
617  * There's no point in trying to do random drop here for DoS
618  * prevention. The NFS clients does 1 reconnect in 15 seconds. An
619  * attacker can easily beat that.
620  *
621  * The only somewhat efficient mechanism would be if drop old
622  * connections from the same IP first. But right now we don't even
623  * record the client IP in svc_sock.
624  *
625  * single-threaded services that expect a lot of clients will probably
626  * need to set sv_maxconn to override the default value which is based
627  * on the number of threads
628  */
svc_check_conn_limits(struct svc_serv * serv)629 static void svc_check_conn_limits(struct svc_serv *serv)
630 {
631 	unsigned int limit = serv->sv_maxconn ? serv->sv_maxconn :
632 				(serv->sv_nrthreads+3) * 20;
633 
634 	if (serv->sv_tmpcnt > limit) {
635 		struct svc_xprt *xprt = NULL;
636 		spin_lock_bh(&serv->sv_lock);
637 		if (!list_empty(&serv->sv_tempsocks)) {
638 			/* Try to help the admin */
639 			net_notice_ratelimited("%s: too many open connections, consider increasing the %s\n",
640 					       serv->sv_name, serv->sv_maxconn ?
641 					       "max number of connections" :
642 					       "number of threads");
643 			/*
644 			 * Always select the oldest connection. It's not fair,
645 			 * but so is life
646 			 */
647 			xprt = list_entry(serv->sv_tempsocks.prev,
648 					  struct svc_xprt,
649 					  xpt_list);
650 			set_bit(XPT_CLOSE, &xprt->xpt_flags);
651 			svc_xprt_get(xprt);
652 		}
653 		spin_unlock_bh(&serv->sv_lock);
654 
655 		if (xprt) {
656 			svc_xprt_enqueue(xprt);
657 			svc_xprt_put(xprt);
658 		}
659 	}
660 }
661 
svc_alloc_arg(struct svc_rqst * rqstp)662 static int svc_alloc_arg(struct svc_rqst *rqstp)
663 {
664 	struct svc_serv *serv = rqstp->rq_server;
665 	struct xdr_buf *arg = &rqstp->rq_arg;
666 	unsigned long pages, filled, ret;
667 
668 	pagevec_init(&rqstp->rq_pvec);
669 
670 	pages = (serv->sv_max_mesg + 2 * PAGE_SIZE) >> PAGE_SHIFT;
671 	if (pages > RPCSVC_MAXPAGES) {
672 		pr_warn_once("svc: warning: pages=%lu > RPCSVC_MAXPAGES=%lu\n",
673 			     pages, RPCSVC_MAXPAGES);
674 		/* use as many pages as possible */
675 		pages = RPCSVC_MAXPAGES;
676 	}
677 
678 	for (filled = 0; filled < pages; filled = ret) {
679 		ret = alloc_pages_bulk_array(GFP_KERNEL, pages,
680 					     rqstp->rq_pages);
681 		if (ret > filled)
682 			/* Made progress, don't sleep yet */
683 			continue;
684 
685 		set_current_state(TASK_INTERRUPTIBLE);
686 		if (signalled() || kthread_should_stop()) {
687 			set_current_state(TASK_RUNNING);
688 			return -EINTR;
689 		}
690 		trace_svc_alloc_arg_err(pages);
691 		schedule_timeout(msecs_to_jiffies(500));
692 	}
693 	rqstp->rq_page_end = &rqstp->rq_pages[pages];
694 	rqstp->rq_pages[pages] = NULL; /* this might be seen in nfsd_splice_actor() */
695 
696 	/* Make arg->head point to first page and arg->pages point to rest */
697 	arg->head[0].iov_base = page_address(rqstp->rq_pages[0]);
698 	arg->head[0].iov_len = PAGE_SIZE;
699 	arg->pages = rqstp->rq_pages + 1;
700 	arg->page_base = 0;
701 	/* save at least one page for response */
702 	arg->page_len = (pages-2)*PAGE_SIZE;
703 	arg->len = (pages-1)*PAGE_SIZE;
704 	arg->tail[0].iov_len = 0;
705 	return 0;
706 }
707 
708 static bool
rqst_should_sleep(struct svc_rqst * rqstp)709 rqst_should_sleep(struct svc_rqst *rqstp)
710 {
711 	struct svc_pool		*pool = rqstp->rq_pool;
712 
713 	/* did someone call svc_wake_up? */
714 	if (test_and_clear_bit(SP_TASK_PENDING, &pool->sp_flags))
715 		return false;
716 
717 	/* was a socket queued? */
718 	if (!list_empty(&pool->sp_sockets))
719 		return false;
720 
721 	/* are we shutting down? */
722 	if (signalled() || kthread_should_stop())
723 		return false;
724 
725 	/* are we freezing? */
726 	if (freezing(current))
727 		return false;
728 
729 	return true;
730 }
731 
svc_get_next_xprt(struct svc_rqst * rqstp,long timeout)732 static struct svc_xprt *svc_get_next_xprt(struct svc_rqst *rqstp, long timeout)
733 {
734 	struct svc_pool		*pool = rqstp->rq_pool;
735 	long			time_left = 0;
736 
737 	/* rq_xprt should be clear on entry */
738 	WARN_ON_ONCE(rqstp->rq_xprt);
739 
740 	rqstp->rq_xprt = svc_xprt_dequeue(pool);
741 	if (rqstp->rq_xprt)
742 		goto out_found;
743 
744 	/*
745 	 * We have to be able to interrupt this wait
746 	 * to bring down the daemons ...
747 	 */
748 	set_current_state(TASK_INTERRUPTIBLE);
749 	smp_mb__before_atomic();
750 	clear_bit(SP_CONGESTED, &pool->sp_flags);
751 	clear_bit(RQ_BUSY, &rqstp->rq_flags);
752 	smp_mb__after_atomic();
753 
754 	if (likely(rqst_should_sleep(rqstp)))
755 		time_left = schedule_timeout(timeout);
756 	else
757 		__set_current_state(TASK_RUNNING);
758 
759 	try_to_freeze();
760 
761 	set_bit(RQ_BUSY, &rqstp->rq_flags);
762 	smp_mb__after_atomic();
763 	rqstp->rq_xprt = svc_xprt_dequeue(pool);
764 	if (rqstp->rq_xprt)
765 		goto out_found;
766 
767 	if (!time_left)
768 		atomic_long_inc(&pool->sp_stats.threads_timedout);
769 
770 	if (signalled() || kthread_should_stop())
771 		return ERR_PTR(-EINTR);
772 	return ERR_PTR(-EAGAIN);
773 out_found:
774 	/* Normally we will wait up to 5 seconds for any required
775 	 * cache information to be provided.
776 	 */
777 	if (!test_bit(SP_CONGESTED, &pool->sp_flags))
778 		rqstp->rq_chandle.thread_wait = 5*HZ;
779 	else
780 		rqstp->rq_chandle.thread_wait = 1*HZ;
781 	trace_svc_xprt_dequeue(rqstp);
782 	return rqstp->rq_xprt;
783 }
784 
svc_add_new_temp_xprt(struct svc_serv * serv,struct svc_xprt * newxpt)785 static void svc_add_new_temp_xprt(struct svc_serv *serv, struct svc_xprt *newxpt)
786 {
787 	spin_lock_bh(&serv->sv_lock);
788 	set_bit(XPT_TEMP, &newxpt->xpt_flags);
789 	list_add(&newxpt->xpt_list, &serv->sv_tempsocks);
790 	serv->sv_tmpcnt++;
791 	if (serv->sv_temptimer.function == NULL) {
792 		/* setup timer to age temp transports */
793 		serv->sv_temptimer.function = svc_age_temp_xprts;
794 		mod_timer(&serv->sv_temptimer,
795 			  jiffies + svc_conn_age_period * HZ);
796 	}
797 	spin_unlock_bh(&serv->sv_lock);
798 	svc_xprt_received(newxpt);
799 }
800 
svc_handle_xprt(struct svc_rqst * rqstp,struct svc_xprt * xprt)801 static int svc_handle_xprt(struct svc_rqst *rqstp, struct svc_xprt *xprt)
802 {
803 	struct svc_serv *serv = rqstp->rq_server;
804 	int len = 0;
805 
806 	if (test_bit(XPT_CLOSE, &xprt->xpt_flags)) {
807 		if (test_and_clear_bit(XPT_KILL_TEMP, &xprt->xpt_flags))
808 			xprt->xpt_ops->xpo_kill_temp_xprt(xprt);
809 		svc_delete_xprt(xprt);
810 		/* Leave XPT_BUSY set on the dead xprt: */
811 		goto out;
812 	}
813 	if (test_bit(XPT_LISTENER, &xprt->xpt_flags)) {
814 		struct svc_xprt *newxpt;
815 		/*
816 		 * We know this module_get will succeed because the
817 		 * listener holds a reference too
818 		 */
819 		__module_get(xprt->xpt_class->xcl_owner);
820 		svc_check_conn_limits(xprt->xpt_server);
821 		newxpt = xprt->xpt_ops->xpo_accept(xprt);
822 		if (newxpt) {
823 			newxpt->xpt_cred = get_cred(xprt->xpt_cred);
824 			svc_add_new_temp_xprt(serv, newxpt);
825 			trace_svc_xprt_accept(newxpt, serv->sv_name);
826 		} else {
827 			module_put(xprt->xpt_class->xcl_owner);
828 		}
829 		svc_xprt_received(xprt);
830 	} else if (svc_xprt_reserve_slot(rqstp, xprt)) {
831 		/* XPT_DATA|XPT_DEFERRED case: */
832 		dprintk("svc: server %p, pool %u, transport %p, inuse=%d\n",
833 			rqstp, rqstp->rq_pool->sp_id, xprt,
834 			kref_read(&xprt->xpt_ref));
835 		rqstp->rq_deferred = svc_deferred_dequeue(xprt);
836 		if (rqstp->rq_deferred)
837 			len = svc_deferred_recv(rqstp);
838 		else
839 			len = xprt->xpt_ops->xpo_recvfrom(rqstp);
840 		rqstp->rq_stime = ktime_get();
841 		rqstp->rq_reserved = serv->sv_max_mesg;
842 		atomic_add(rqstp->rq_reserved, &xprt->xpt_reserved);
843 	} else
844 		svc_xprt_received(xprt);
845 out:
846 	trace_svc_handle_xprt(xprt, len);
847 	return len;
848 }
849 
850 /*
851  * Receive the next request on any transport.  This code is carefully
852  * organised not to touch any cachelines in the shared svc_serv
853  * structure, only cachelines in the local svc_pool.
854  */
svc_recv(struct svc_rqst * rqstp,long timeout)855 int svc_recv(struct svc_rqst *rqstp, long timeout)
856 {
857 	struct svc_xprt		*xprt = NULL;
858 	struct svc_serv		*serv = rqstp->rq_server;
859 	int			len, err;
860 
861 	err = svc_alloc_arg(rqstp);
862 	if (err)
863 		goto out;
864 
865 	try_to_freeze();
866 	cond_resched();
867 	err = -EINTR;
868 	if (signalled() || kthread_should_stop())
869 		goto out;
870 
871 	xprt = svc_get_next_xprt(rqstp, timeout);
872 	if (IS_ERR(xprt)) {
873 		err = PTR_ERR(xprt);
874 		goto out;
875 	}
876 
877 	len = svc_handle_xprt(rqstp, xprt);
878 
879 	/* No data, incomplete (TCP) read, or accept() */
880 	err = -EAGAIN;
881 	if (len <= 0)
882 		goto out_release;
883 	trace_svc_xdr_recvfrom(&rqstp->rq_arg);
884 
885 	clear_bit(XPT_OLD, &xprt->xpt_flags);
886 
887 	xprt->xpt_ops->xpo_secure_port(rqstp);
888 	rqstp->rq_chandle.defer = svc_defer;
889 	rqstp->rq_xid = svc_getu32(&rqstp->rq_arg.head[0]);
890 
891 	if (serv->sv_stats)
892 		serv->sv_stats->netcnt++;
893 	return len;
894 out_release:
895 	rqstp->rq_res.len = 0;
896 	svc_xprt_release(rqstp);
897 out:
898 	return err;
899 }
900 EXPORT_SYMBOL_GPL(svc_recv);
901 
902 /*
903  * Drop request
904  */
svc_drop(struct svc_rqst * rqstp)905 void svc_drop(struct svc_rqst *rqstp)
906 {
907 	trace_svc_drop(rqstp);
908 	svc_xprt_release(rqstp);
909 }
910 EXPORT_SYMBOL_GPL(svc_drop);
911 
912 /*
913  * Return reply to client.
914  */
svc_send(struct svc_rqst * rqstp)915 int svc_send(struct svc_rqst *rqstp)
916 {
917 	struct svc_xprt	*xprt;
918 	int		len = -EFAULT;
919 	struct xdr_buf	*xb;
920 
921 	xprt = rqstp->rq_xprt;
922 	if (!xprt)
923 		goto out;
924 
925 	/* calculate over-all length */
926 	xb = &rqstp->rq_res;
927 	xb->len = xb->head[0].iov_len +
928 		xb->page_len +
929 		xb->tail[0].iov_len;
930 	trace_svc_xdr_sendto(rqstp->rq_xid, xb);
931 	trace_svc_stats_latency(rqstp);
932 
933 	len = xprt->xpt_ops->xpo_sendto(rqstp);
934 
935 	trace_svc_send(rqstp, len);
936 	svc_xprt_release(rqstp);
937 
938 	if (len == -ECONNREFUSED || len == -ENOTCONN || len == -EAGAIN)
939 		len = 0;
940 out:
941 	return len;
942 }
943 
944 /*
945  * Timer function to close old temporary transports, using
946  * a mark-and-sweep algorithm.
947  */
svc_age_temp_xprts(struct timer_list * t)948 static void svc_age_temp_xprts(struct timer_list *t)
949 {
950 	struct svc_serv *serv = from_timer(serv, t, sv_temptimer);
951 	struct svc_xprt *xprt;
952 	struct list_head *le, *next;
953 
954 	dprintk("svc_age_temp_xprts\n");
955 
956 	if (!spin_trylock_bh(&serv->sv_lock)) {
957 		/* busy, try again 1 sec later */
958 		dprintk("svc_age_temp_xprts: busy\n");
959 		mod_timer(&serv->sv_temptimer, jiffies + HZ);
960 		return;
961 	}
962 
963 	list_for_each_safe(le, next, &serv->sv_tempsocks) {
964 		xprt = list_entry(le, struct svc_xprt, xpt_list);
965 
966 		/* First time through, just mark it OLD. Second time
967 		 * through, close it. */
968 		if (!test_and_set_bit(XPT_OLD, &xprt->xpt_flags))
969 			continue;
970 		if (kref_read(&xprt->xpt_ref) > 1 ||
971 		    test_bit(XPT_BUSY, &xprt->xpt_flags))
972 			continue;
973 		list_del_init(le);
974 		set_bit(XPT_CLOSE, &xprt->xpt_flags);
975 		dprintk("queuing xprt %p for closing\n", xprt);
976 
977 		/* a thread will dequeue and close it soon */
978 		svc_xprt_enqueue(xprt);
979 	}
980 	spin_unlock_bh(&serv->sv_lock);
981 
982 	mod_timer(&serv->sv_temptimer, jiffies + svc_conn_age_period * HZ);
983 }
984 
985 /* Close temporary transports whose xpt_local matches server_addr immediately
986  * instead of waiting for them to be picked up by the timer.
987  *
988  * This is meant to be called from a notifier_block that runs when an ip
989  * address is deleted.
990  */
svc_age_temp_xprts_now(struct svc_serv * serv,struct sockaddr * server_addr)991 void svc_age_temp_xprts_now(struct svc_serv *serv, struct sockaddr *server_addr)
992 {
993 	struct svc_xprt *xprt;
994 	struct list_head *le, *next;
995 	LIST_HEAD(to_be_closed);
996 
997 	spin_lock_bh(&serv->sv_lock);
998 	list_for_each_safe(le, next, &serv->sv_tempsocks) {
999 		xprt = list_entry(le, struct svc_xprt, xpt_list);
1000 		if (rpc_cmp_addr(server_addr, (struct sockaddr *)
1001 				&xprt->xpt_local)) {
1002 			dprintk("svc_age_temp_xprts_now: found %p\n", xprt);
1003 			list_move(le, &to_be_closed);
1004 		}
1005 	}
1006 	spin_unlock_bh(&serv->sv_lock);
1007 
1008 	while (!list_empty(&to_be_closed)) {
1009 		le = to_be_closed.next;
1010 		list_del_init(le);
1011 		xprt = list_entry(le, struct svc_xprt, xpt_list);
1012 		set_bit(XPT_CLOSE, &xprt->xpt_flags);
1013 		set_bit(XPT_KILL_TEMP, &xprt->xpt_flags);
1014 		dprintk("svc_age_temp_xprts_now: queuing xprt %p for closing\n",
1015 				xprt);
1016 		svc_xprt_enqueue(xprt);
1017 	}
1018 }
1019 EXPORT_SYMBOL_GPL(svc_age_temp_xprts_now);
1020 
call_xpt_users(struct svc_xprt * xprt)1021 static void call_xpt_users(struct svc_xprt *xprt)
1022 {
1023 	struct svc_xpt_user *u;
1024 
1025 	spin_lock(&xprt->xpt_lock);
1026 	while (!list_empty(&xprt->xpt_users)) {
1027 		u = list_first_entry(&xprt->xpt_users, struct svc_xpt_user, list);
1028 		list_del_init(&u->list);
1029 		u->callback(u);
1030 	}
1031 	spin_unlock(&xprt->xpt_lock);
1032 }
1033 
1034 /*
1035  * Remove a dead transport
1036  */
svc_delete_xprt(struct svc_xprt * xprt)1037 static void svc_delete_xprt(struct svc_xprt *xprt)
1038 {
1039 	struct svc_serv	*serv = xprt->xpt_server;
1040 	struct svc_deferred_req *dr;
1041 
1042 	if (test_and_set_bit(XPT_DEAD, &xprt->xpt_flags))
1043 		return;
1044 
1045 	trace_svc_xprt_detach(xprt);
1046 	xprt->xpt_ops->xpo_detach(xprt);
1047 	if (xprt->xpt_bc_xprt)
1048 		xprt->xpt_bc_xprt->ops->close(xprt->xpt_bc_xprt);
1049 
1050 	spin_lock_bh(&serv->sv_lock);
1051 	list_del_init(&xprt->xpt_list);
1052 	WARN_ON_ONCE(!list_empty(&xprt->xpt_ready));
1053 	if (test_bit(XPT_TEMP, &xprt->xpt_flags))
1054 		serv->sv_tmpcnt--;
1055 	spin_unlock_bh(&serv->sv_lock);
1056 
1057 	while ((dr = svc_deferred_dequeue(xprt)) != NULL)
1058 		kfree(dr);
1059 
1060 	call_xpt_users(xprt);
1061 	svc_xprt_put(xprt);
1062 }
1063 
svc_close_xprt(struct svc_xprt * xprt)1064 void svc_close_xprt(struct svc_xprt *xprt)
1065 {
1066 	trace_svc_xprt_close(xprt);
1067 	set_bit(XPT_CLOSE, &xprt->xpt_flags);
1068 	if (test_and_set_bit(XPT_BUSY, &xprt->xpt_flags))
1069 		/* someone else will have to effect the close */
1070 		return;
1071 	/*
1072 	 * We expect svc_close_xprt() to work even when no threads are
1073 	 * running (e.g., while configuring the server before starting
1074 	 * any threads), so if the transport isn't busy, we delete
1075 	 * it ourself:
1076 	 */
1077 	svc_delete_xprt(xprt);
1078 }
1079 EXPORT_SYMBOL_GPL(svc_close_xprt);
1080 
svc_close_list(struct svc_serv * serv,struct list_head * xprt_list,struct net * net)1081 static int svc_close_list(struct svc_serv *serv, struct list_head *xprt_list, struct net *net)
1082 {
1083 	struct svc_xprt *xprt;
1084 	int ret = 0;
1085 
1086 	spin_lock_bh(&serv->sv_lock);
1087 	list_for_each_entry(xprt, xprt_list, xpt_list) {
1088 		if (xprt->xpt_net != net)
1089 			continue;
1090 		ret++;
1091 		set_bit(XPT_CLOSE, &xprt->xpt_flags);
1092 		svc_xprt_enqueue(xprt);
1093 	}
1094 	spin_unlock_bh(&serv->sv_lock);
1095 	return ret;
1096 }
1097 
svc_dequeue_net(struct svc_serv * serv,struct net * net)1098 static struct svc_xprt *svc_dequeue_net(struct svc_serv *serv, struct net *net)
1099 {
1100 	struct svc_pool *pool;
1101 	struct svc_xprt *xprt;
1102 	struct svc_xprt *tmp;
1103 	int i;
1104 
1105 	for (i = 0; i < serv->sv_nrpools; i++) {
1106 		pool = &serv->sv_pools[i];
1107 
1108 		spin_lock_bh(&pool->sp_lock);
1109 		list_for_each_entry_safe(xprt, tmp, &pool->sp_sockets, xpt_ready) {
1110 			if (xprt->xpt_net != net)
1111 				continue;
1112 			list_del_init(&xprt->xpt_ready);
1113 			spin_unlock_bh(&pool->sp_lock);
1114 			return xprt;
1115 		}
1116 		spin_unlock_bh(&pool->sp_lock);
1117 	}
1118 	return NULL;
1119 }
1120 
svc_clean_up_xprts(struct svc_serv * serv,struct net * net)1121 static void svc_clean_up_xprts(struct svc_serv *serv, struct net *net)
1122 {
1123 	struct svc_xprt *xprt;
1124 
1125 	while ((xprt = svc_dequeue_net(serv, net))) {
1126 		set_bit(XPT_CLOSE, &xprt->xpt_flags);
1127 		svc_delete_xprt(xprt);
1128 	}
1129 }
1130 
1131 /*
1132  * Server threads may still be running (especially in the case where the
1133  * service is still running in other network namespaces).
1134  *
1135  * So we shut down sockets the same way we would on a running server, by
1136  * setting XPT_CLOSE, enqueuing, and letting a thread pick it up to do
1137  * the close.  In the case there are no such other threads,
1138  * threads running, svc_clean_up_xprts() does a simple version of a
1139  * server's main event loop, and in the case where there are other
1140  * threads, we may need to wait a little while and then check again to
1141  * see if they're done.
1142  */
svc_close_net(struct svc_serv * serv,struct net * net)1143 void svc_close_net(struct svc_serv *serv, struct net *net)
1144 {
1145 	int delay = 0;
1146 
1147 	while (svc_close_list(serv, &serv->sv_permsocks, net) +
1148 	       svc_close_list(serv, &serv->sv_tempsocks, net)) {
1149 
1150 		svc_clean_up_xprts(serv, net);
1151 		msleep(delay++);
1152 	}
1153 }
1154 
1155 /*
1156  * Handle defer and revisit of requests
1157  */
1158 
svc_revisit(struct cache_deferred_req * dreq,int too_many)1159 static void svc_revisit(struct cache_deferred_req *dreq, int too_many)
1160 {
1161 	struct svc_deferred_req *dr =
1162 		container_of(dreq, struct svc_deferred_req, handle);
1163 	struct svc_xprt *xprt = dr->xprt;
1164 
1165 	spin_lock(&xprt->xpt_lock);
1166 	set_bit(XPT_DEFERRED, &xprt->xpt_flags);
1167 	if (too_many || test_bit(XPT_DEAD, &xprt->xpt_flags)) {
1168 		spin_unlock(&xprt->xpt_lock);
1169 		trace_svc_defer_drop(dr);
1170 		svc_xprt_put(xprt);
1171 		kfree(dr);
1172 		return;
1173 	}
1174 	dr->xprt = NULL;
1175 	list_add(&dr->handle.recent, &xprt->xpt_deferred);
1176 	spin_unlock(&xprt->xpt_lock);
1177 	trace_svc_defer_queue(dr);
1178 	svc_xprt_enqueue(xprt);
1179 	svc_xprt_put(xprt);
1180 }
1181 
1182 /*
1183  * Save the request off for later processing. The request buffer looks
1184  * like this:
1185  *
1186  * <xprt-header><rpc-header><rpc-pagelist><rpc-tail>
1187  *
1188  * This code can only handle requests that consist of an xprt-header
1189  * and rpc-header.
1190  */
svc_defer(struct cache_req * req)1191 static struct cache_deferred_req *svc_defer(struct cache_req *req)
1192 {
1193 	struct svc_rqst *rqstp = container_of(req, struct svc_rqst, rq_chandle);
1194 	struct svc_deferred_req *dr;
1195 
1196 	if (rqstp->rq_arg.page_len || !test_bit(RQ_USEDEFERRAL, &rqstp->rq_flags))
1197 		return NULL; /* if more than a page, give up FIXME */
1198 	if (rqstp->rq_deferred) {
1199 		dr = rqstp->rq_deferred;
1200 		rqstp->rq_deferred = NULL;
1201 	} else {
1202 		size_t skip;
1203 		size_t size;
1204 		/* FIXME maybe discard if size too large */
1205 		size = sizeof(struct svc_deferred_req) + rqstp->rq_arg.len;
1206 		dr = kmalloc(size, GFP_KERNEL);
1207 		if (dr == NULL)
1208 			return NULL;
1209 
1210 		dr->handle.owner = rqstp->rq_server;
1211 		dr->prot = rqstp->rq_prot;
1212 		memcpy(&dr->addr, &rqstp->rq_addr, rqstp->rq_addrlen);
1213 		dr->addrlen = rqstp->rq_addrlen;
1214 		dr->daddr = rqstp->rq_daddr;
1215 		dr->argslen = rqstp->rq_arg.len >> 2;
1216 		dr->xprt_hlen = rqstp->rq_xprt_hlen;
1217 
1218 		/* back up head to the start of the buffer and copy */
1219 		skip = rqstp->rq_arg.len - rqstp->rq_arg.head[0].iov_len;
1220 		memcpy(dr->args, rqstp->rq_arg.head[0].iov_base - skip,
1221 		       dr->argslen << 2);
1222 	}
1223 	trace_svc_defer(rqstp);
1224 	svc_xprt_get(rqstp->rq_xprt);
1225 	dr->xprt = rqstp->rq_xprt;
1226 	set_bit(RQ_DROPME, &rqstp->rq_flags);
1227 
1228 	dr->handle.revisit = svc_revisit;
1229 	return &dr->handle;
1230 }
1231 
1232 /*
1233  * recv data from a deferred request into an active one
1234  */
svc_deferred_recv(struct svc_rqst * rqstp)1235 static noinline int svc_deferred_recv(struct svc_rqst *rqstp)
1236 {
1237 	struct svc_deferred_req *dr = rqstp->rq_deferred;
1238 
1239 	trace_svc_defer_recv(dr);
1240 
1241 	/* setup iov_base past transport header */
1242 	rqstp->rq_arg.head[0].iov_base = dr->args + (dr->xprt_hlen>>2);
1243 	/* The iov_len does not include the transport header bytes */
1244 	rqstp->rq_arg.head[0].iov_len = (dr->argslen<<2) - dr->xprt_hlen;
1245 	rqstp->rq_arg.page_len = 0;
1246 	/* The rq_arg.len includes the transport header bytes */
1247 	rqstp->rq_arg.len     = dr->argslen<<2;
1248 	rqstp->rq_prot        = dr->prot;
1249 	memcpy(&rqstp->rq_addr, &dr->addr, dr->addrlen);
1250 	rqstp->rq_addrlen     = dr->addrlen;
1251 	/* Save off transport header len in case we get deferred again */
1252 	rqstp->rq_xprt_hlen   = dr->xprt_hlen;
1253 	rqstp->rq_daddr       = dr->daddr;
1254 	rqstp->rq_respages    = rqstp->rq_pages;
1255 	svc_xprt_received(rqstp->rq_xprt);
1256 	return (dr->argslen<<2) - dr->xprt_hlen;
1257 }
1258 
1259 
svc_deferred_dequeue(struct svc_xprt * xprt)1260 static struct svc_deferred_req *svc_deferred_dequeue(struct svc_xprt *xprt)
1261 {
1262 	struct svc_deferred_req *dr = NULL;
1263 
1264 	if (!test_bit(XPT_DEFERRED, &xprt->xpt_flags))
1265 		return NULL;
1266 	spin_lock(&xprt->xpt_lock);
1267 	if (!list_empty(&xprt->xpt_deferred)) {
1268 		dr = list_entry(xprt->xpt_deferred.next,
1269 				struct svc_deferred_req,
1270 				handle.recent);
1271 		list_del_init(&dr->handle.recent);
1272 	} else
1273 		clear_bit(XPT_DEFERRED, &xprt->xpt_flags);
1274 	spin_unlock(&xprt->xpt_lock);
1275 	return dr;
1276 }
1277 
1278 /**
1279  * svc_find_xprt - find an RPC transport instance
1280  * @serv: pointer to svc_serv to search
1281  * @xcl_name: C string containing transport's class name
1282  * @net: owner net pointer
1283  * @af: Address family of transport's local address
1284  * @port: transport's IP port number
1285  *
1286  * Return the transport instance pointer for the endpoint accepting
1287  * connections/peer traffic from the specified transport class,
1288  * address family and port.
1289  *
1290  * Specifying 0 for the address family or port is effectively a
1291  * wild-card, and will result in matching the first transport in the
1292  * service's list that has a matching class name.
1293  */
svc_find_xprt(struct svc_serv * serv,const char * xcl_name,struct net * net,const sa_family_t af,const unsigned short port)1294 struct svc_xprt *svc_find_xprt(struct svc_serv *serv, const char *xcl_name,
1295 			       struct net *net, const sa_family_t af,
1296 			       const unsigned short port)
1297 {
1298 	struct svc_xprt *xprt;
1299 	struct svc_xprt *found = NULL;
1300 
1301 	/* Sanity check the args */
1302 	if (serv == NULL || xcl_name == NULL)
1303 		return found;
1304 
1305 	spin_lock_bh(&serv->sv_lock);
1306 	list_for_each_entry(xprt, &serv->sv_permsocks, xpt_list) {
1307 		if (xprt->xpt_net != net)
1308 			continue;
1309 		if (strcmp(xprt->xpt_class->xcl_name, xcl_name))
1310 			continue;
1311 		if (af != AF_UNSPEC && af != xprt->xpt_local.ss_family)
1312 			continue;
1313 		if (port != 0 && port != svc_xprt_local_port(xprt))
1314 			continue;
1315 		found = xprt;
1316 		svc_xprt_get(xprt);
1317 		break;
1318 	}
1319 	spin_unlock_bh(&serv->sv_lock);
1320 	return found;
1321 }
1322 EXPORT_SYMBOL_GPL(svc_find_xprt);
1323 
svc_one_xprt_name(const struct svc_xprt * xprt,char * pos,int remaining)1324 static int svc_one_xprt_name(const struct svc_xprt *xprt,
1325 			     char *pos, int remaining)
1326 {
1327 	int len;
1328 
1329 	len = snprintf(pos, remaining, "%s %u\n",
1330 			xprt->xpt_class->xcl_name,
1331 			svc_xprt_local_port(xprt));
1332 	if (len >= remaining)
1333 		return -ENAMETOOLONG;
1334 	return len;
1335 }
1336 
1337 /**
1338  * svc_xprt_names - format a buffer with a list of transport names
1339  * @serv: pointer to an RPC service
1340  * @buf: pointer to a buffer to be filled in
1341  * @buflen: length of buffer to be filled in
1342  *
1343  * Fills in @buf with a string containing a list of transport names,
1344  * each name terminated with '\n'.
1345  *
1346  * Returns positive length of the filled-in string on success; otherwise
1347  * a negative errno value is returned if an error occurs.
1348  */
svc_xprt_names(struct svc_serv * serv,char * buf,const int buflen)1349 int svc_xprt_names(struct svc_serv *serv, char *buf, const int buflen)
1350 {
1351 	struct svc_xprt *xprt;
1352 	int len, totlen;
1353 	char *pos;
1354 
1355 	/* Sanity check args */
1356 	if (!serv)
1357 		return 0;
1358 
1359 	spin_lock_bh(&serv->sv_lock);
1360 
1361 	pos = buf;
1362 	totlen = 0;
1363 	list_for_each_entry(xprt, &serv->sv_permsocks, xpt_list) {
1364 		len = svc_one_xprt_name(xprt, pos, buflen - totlen);
1365 		if (len < 0) {
1366 			*buf = '\0';
1367 			totlen = len;
1368 		}
1369 		if (len <= 0)
1370 			break;
1371 
1372 		pos += len;
1373 		totlen += len;
1374 	}
1375 
1376 	spin_unlock_bh(&serv->sv_lock);
1377 	return totlen;
1378 }
1379 EXPORT_SYMBOL_GPL(svc_xprt_names);
1380 
1381 
1382 /*----------------------------------------------------------------------------*/
1383 
svc_pool_stats_start(struct seq_file * m,loff_t * pos)1384 static void *svc_pool_stats_start(struct seq_file *m, loff_t *pos)
1385 {
1386 	unsigned int pidx = (unsigned int)*pos;
1387 	struct svc_serv *serv = m->private;
1388 
1389 	dprintk("svc_pool_stats_start, *pidx=%u\n", pidx);
1390 
1391 	if (!pidx)
1392 		return SEQ_START_TOKEN;
1393 	return (pidx > serv->sv_nrpools ? NULL : &serv->sv_pools[pidx-1]);
1394 }
1395 
svc_pool_stats_next(struct seq_file * m,void * p,loff_t * pos)1396 static void *svc_pool_stats_next(struct seq_file *m, void *p, loff_t *pos)
1397 {
1398 	struct svc_pool *pool = p;
1399 	struct svc_serv *serv = m->private;
1400 
1401 	dprintk("svc_pool_stats_next, *pos=%llu\n", *pos);
1402 
1403 	if (p == SEQ_START_TOKEN) {
1404 		pool = &serv->sv_pools[0];
1405 	} else {
1406 		unsigned int pidx = (pool - &serv->sv_pools[0]);
1407 		if (pidx < serv->sv_nrpools-1)
1408 			pool = &serv->sv_pools[pidx+1];
1409 		else
1410 			pool = NULL;
1411 	}
1412 	++*pos;
1413 	return pool;
1414 }
1415 
svc_pool_stats_stop(struct seq_file * m,void * p)1416 static void svc_pool_stats_stop(struct seq_file *m, void *p)
1417 {
1418 }
1419 
svc_pool_stats_show(struct seq_file * m,void * p)1420 static int svc_pool_stats_show(struct seq_file *m, void *p)
1421 {
1422 	struct svc_pool *pool = p;
1423 
1424 	if (p == SEQ_START_TOKEN) {
1425 		seq_puts(m, "# pool packets-arrived sockets-enqueued threads-woken threads-timedout\n");
1426 		return 0;
1427 	}
1428 
1429 	seq_printf(m, "%u %lu %lu %lu %lu\n",
1430 		pool->sp_id,
1431 		(unsigned long)atomic_long_read(&pool->sp_stats.packets),
1432 		pool->sp_stats.sockets_queued,
1433 		(unsigned long)atomic_long_read(&pool->sp_stats.threads_woken),
1434 		(unsigned long)atomic_long_read(&pool->sp_stats.threads_timedout));
1435 
1436 	return 0;
1437 }
1438 
1439 static const struct seq_operations svc_pool_stats_seq_ops = {
1440 	.start	= svc_pool_stats_start,
1441 	.next	= svc_pool_stats_next,
1442 	.stop	= svc_pool_stats_stop,
1443 	.show	= svc_pool_stats_show,
1444 };
1445 
svc_pool_stats_open(struct svc_serv * serv,struct file * file)1446 int svc_pool_stats_open(struct svc_serv *serv, struct file *file)
1447 {
1448 	int err;
1449 
1450 	err = seq_open(file, &svc_pool_stats_seq_ops);
1451 	if (!err)
1452 		((struct seq_file *) file->private_data)->private = serv;
1453 	return err;
1454 }
1455 EXPORT_SYMBOL(svc_pool_stats_open);
1456 
1457 /*----------------------------------------------------------------------------*/
1458