xref: /openbmc/linux/net/sunrpc/sched.c (revision 55eb9a6c)
1 // SPDX-License-Identifier: GPL-2.0-only
2 /*
3  * linux/net/sunrpc/sched.c
4  *
5  * Scheduling for synchronous and asynchronous RPC requests.
6  *
7  * Copyright (C) 1996 Olaf Kirch, <okir@monad.swb.de>
8  *
9  * TCP NFS related read + write fixes
10  * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
11  */
12 
13 #include <linux/module.h>
14 
15 #include <linux/sched.h>
16 #include <linux/interrupt.h>
17 #include <linux/slab.h>
18 #include <linux/mempool.h>
19 #include <linux/smp.h>
20 #include <linux/spinlock.h>
21 #include <linux/mutex.h>
22 #include <linux/freezer.h>
23 #include <linux/sched/mm.h>
24 
25 #include <linux/sunrpc/clnt.h>
26 #include <linux/sunrpc/metrics.h>
27 
28 #include "sunrpc.h"
29 
30 #define CREATE_TRACE_POINTS
31 #include <trace/events/sunrpc.h>
32 
33 /*
34  * RPC slabs and memory pools
35  */
36 #define RPC_BUFFER_MAXSIZE	(2048)
37 #define RPC_BUFFER_POOLSIZE	(8)
38 #define RPC_TASK_POOLSIZE	(8)
39 static struct kmem_cache	*rpc_task_slabp __read_mostly;
40 static struct kmem_cache	*rpc_buffer_slabp __read_mostly;
41 static mempool_t	*rpc_task_mempool __read_mostly;
42 static mempool_t	*rpc_buffer_mempool __read_mostly;
43 
44 static void			rpc_async_schedule(struct work_struct *);
45 static void			 rpc_release_task(struct rpc_task *task);
46 static void __rpc_queue_timer_fn(struct work_struct *);
47 
48 /*
49  * RPC tasks sit here while waiting for conditions to improve.
50  */
51 static struct rpc_wait_queue delay_queue;
52 
53 /*
54  * rpciod-related stuff
55  */
56 struct workqueue_struct *rpciod_workqueue __read_mostly;
57 struct workqueue_struct *xprtiod_workqueue __read_mostly;
58 EXPORT_SYMBOL_GPL(xprtiod_workqueue);
59 
60 gfp_t rpc_task_gfp_mask(void)
61 {
62 	if (current->flags & PF_WQ_WORKER)
63 		return GFP_KERNEL | __GFP_NORETRY | __GFP_NOWARN;
64 	return GFP_KERNEL;
65 }
66 
67 unsigned long
68 rpc_task_timeout(const struct rpc_task *task)
69 {
70 	unsigned long timeout = READ_ONCE(task->tk_timeout);
71 
72 	if (timeout != 0) {
73 		unsigned long now = jiffies;
74 		if (time_before(now, timeout))
75 			return timeout - now;
76 	}
77 	return 0;
78 }
79 EXPORT_SYMBOL_GPL(rpc_task_timeout);
80 
81 /*
82  * Disable the timer for a given RPC task. Should be called with
83  * queue->lock and bh_disabled in order to avoid races within
84  * rpc_run_timer().
85  */
86 static void
87 __rpc_disable_timer(struct rpc_wait_queue *queue, struct rpc_task *task)
88 {
89 	if (list_empty(&task->u.tk_wait.timer_list))
90 		return;
91 	task->tk_timeout = 0;
92 	list_del(&task->u.tk_wait.timer_list);
93 	if (list_empty(&queue->timer_list.list))
94 		cancel_delayed_work(&queue->timer_list.dwork);
95 }
96 
97 static void
98 rpc_set_queue_timer(struct rpc_wait_queue *queue, unsigned long expires)
99 {
100 	unsigned long now = jiffies;
101 	queue->timer_list.expires = expires;
102 	if (time_before_eq(expires, now))
103 		expires = 0;
104 	else
105 		expires -= now;
106 	mod_delayed_work(rpciod_workqueue, &queue->timer_list.dwork, expires);
107 }
108 
109 /*
110  * Set up a timer for the current task.
111  */
112 static void
113 __rpc_add_timer(struct rpc_wait_queue *queue, struct rpc_task *task,
114 		unsigned long timeout)
115 {
116 	task->tk_timeout = timeout;
117 	if (list_empty(&queue->timer_list.list) || time_before(timeout, queue->timer_list.expires))
118 		rpc_set_queue_timer(queue, timeout);
119 	list_add(&task->u.tk_wait.timer_list, &queue->timer_list.list);
120 }
121 
122 static void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority)
123 {
124 	if (queue->priority != priority) {
125 		queue->priority = priority;
126 		queue->nr = 1U << priority;
127 	}
128 }
129 
130 static void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
131 {
132 	rpc_set_waitqueue_priority(queue, queue->maxpriority);
133 }
134 
135 /*
136  * Add a request to a queue list
137  */
138 static void
139 __rpc_list_enqueue_task(struct list_head *q, struct rpc_task *task)
140 {
141 	struct rpc_task *t;
142 
143 	list_for_each_entry(t, q, u.tk_wait.list) {
144 		if (t->tk_owner == task->tk_owner) {
145 			list_add_tail(&task->u.tk_wait.links,
146 					&t->u.tk_wait.links);
147 			/* Cache the queue head in task->u.tk_wait.list */
148 			task->u.tk_wait.list.next = q;
149 			task->u.tk_wait.list.prev = NULL;
150 			return;
151 		}
152 	}
153 	INIT_LIST_HEAD(&task->u.tk_wait.links);
154 	list_add_tail(&task->u.tk_wait.list, q);
155 }
156 
157 /*
158  * Remove request from a queue list
159  */
160 static void
161 __rpc_list_dequeue_task(struct rpc_task *task)
162 {
163 	struct list_head *q;
164 	struct rpc_task *t;
165 
166 	if (task->u.tk_wait.list.prev == NULL) {
167 		list_del(&task->u.tk_wait.links);
168 		return;
169 	}
170 	if (!list_empty(&task->u.tk_wait.links)) {
171 		t = list_first_entry(&task->u.tk_wait.links,
172 				struct rpc_task,
173 				u.tk_wait.links);
174 		/* Assume __rpc_list_enqueue_task() cached the queue head */
175 		q = t->u.tk_wait.list.next;
176 		list_add_tail(&t->u.tk_wait.list, q);
177 		list_del(&task->u.tk_wait.links);
178 	}
179 	list_del(&task->u.tk_wait.list);
180 }
181 
182 /*
183  * Add new request to a priority queue.
184  */
185 static void __rpc_add_wait_queue_priority(struct rpc_wait_queue *queue,
186 		struct rpc_task *task,
187 		unsigned char queue_priority)
188 {
189 	if (unlikely(queue_priority > queue->maxpriority))
190 		queue_priority = queue->maxpriority;
191 	__rpc_list_enqueue_task(&queue->tasks[queue_priority], task);
192 }
193 
194 /*
195  * Add new request to wait queue.
196  */
197 static void __rpc_add_wait_queue(struct rpc_wait_queue *queue,
198 		struct rpc_task *task,
199 		unsigned char queue_priority)
200 {
201 	INIT_LIST_HEAD(&task->u.tk_wait.timer_list);
202 	if (RPC_IS_PRIORITY(queue))
203 		__rpc_add_wait_queue_priority(queue, task, queue_priority);
204 	else
205 		list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
206 	task->tk_waitqueue = queue;
207 	queue->qlen++;
208 	/* barrier matches the read in rpc_wake_up_task_queue_locked() */
209 	smp_wmb();
210 	rpc_set_queued(task);
211 }
212 
213 /*
214  * Remove request from a priority queue.
215  */
216 static void __rpc_remove_wait_queue_priority(struct rpc_task *task)
217 {
218 	__rpc_list_dequeue_task(task);
219 }
220 
221 /*
222  * Remove request from queue.
223  * Note: must be called with spin lock held.
224  */
225 static void __rpc_remove_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
226 {
227 	__rpc_disable_timer(queue, task);
228 	if (RPC_IS_PRIORITY(queue))
229 		__rpc_remove_wait_queue_priority(task);
230 	else
231 		list_del(&task->u.tk_wait.list);
232 	queue->qlen--;
233 }
234 
235 static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, unsigned char nr_queues)
236 {
237 	int i;
238 
239 	spin_lock_init(&queue->lock);
240 	for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
241 		INIT_LIST_HEAD(&queue->tasks[i]);
242 	queue->maxpriority = nr_queues - 1;
243 	rpc_reset_waitqueue_priority(queue);
244 	queue->qlen = 0;
245 	queue->timer_list.expires = 0;
246 	INIT_DELAYED_WORK(&queue->timer_list.dwork, __rpc_queue_timer_fn);
247 	INIT_LIST_HEAD(&queue->timer_list.list);
248 	rpc_assign_waitqueue_name(queue, qname);
249 }
250 
251 void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
252 {
253 	__rpc_init_priority_wait_queue(queue, qname, RPC_NR_PRIORITY);
254 }
255 EXPORT_SYMBOL_GPL(rpc_init_priority_wait_queue);
256 
257 void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
258 {
259 	__rpc_init_priority_wait_queue(queue, qname, 1);
260 }
261 EXPORT_SYMBOL_GPL(rpc_init_wait_queue);
262 
263 void rpc_destroy_wait_queue(struct rpc_wait_queue *queue)
264 {
265 	cancel_delayed_work_sync(&queue->timer_list.dwork);
266 }
267 EXPORT_SYMBOL_GPL(rpc_destroy_wait_queue);
268 
269 static int rpc_wait_bit_killable(struct wait_bit_key *key, int mode)
270 {
271 	freezable_schedule_unsafe();
272 	if (signal_pending_state(mode, current))
273 		return -ERESTARTSYS;
274 	return 0;
275 }
276 
277 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) || IS_ENABLED(CONFIG_TRACEPOINTS)
278 static void rpc_task_set_debuginfo(struct rpc_task *task)
279 {
280 	struct rpc_clnt *clnt = task->tk_client;
281 
282 	/* Might be a task carrying a reverse-direction operation */
283 	if (!clnt) {
284 		static atomic_t rpc_pid;
285 
286 		task->tk_pid = atomic_inc_return(&rpc_pid);
287 		return;
288 	}
289 
290 	task->tk_pid = atomic_inc_return(&clnt->cl_pid);
291 }
292 #else
293 static inline void rpc_task_set_debuginfo(struct rpc_task *task)
294 {
295 }
296 #endif
297 
298 static void rpc_set_active(struct rpc_task *task)
299 {
300 	rpc_task_set_debuginfo(task);
301 	set_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
302 	trace_rpc_task_begin(task, NULL);
303 }
304 
305 /*
306  * Mark an RPC call as having completed by clearing the 'active' bit
307  * and then waking up all tasks that were sleeping.
308  */
309 static int rpc_complete_task(struct rpc_task *task)
310 {
311 	void *m = &task->tk_runstate;
312 	wait_queue_head_t *wq = bit_waitqueue(m, RPC_TASK_ACTIVE);
313 	struct wait_bit_key k = __WAIT_BIT_KEY_INITIALIZER(m, RPC_TASK_ACTIVE);
314 	unsigned long flags;
315 	int ret;
316 
317 	trace_rpc_task_complete(task, NULL);
318 
319 	spin_lock_irqsave(&wq->lock, flags);
320 	clear_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
321 	ret = atomic_dec_and_test(&task->tk_count);
322 	if (waitqueue_active(wq))
323 		__wake_up_locked_key(wq, TASK_NORMAL, &k);
324 	spin_unlock_irqrestore(&wq->lock, flags);
325 	return ret;
326 }
327 
328 /*
329  * Allow callers to wait for completion of an RPC call
330  *
331  * Note the use of out_of_line_wait_on_bit() rather than wait_on_bit()
332  * to enforce taking of the wq->lock and hence avoid races with
333  * rpc_complete_task().
334  */
335 int __rpc_wait_for_completion_task(struct rpc_task *task, wait_bit_action_f *action)
336 {
337 	if (action == NULL)
338 		action = rpc_wait_bit_killable;
339 	return out_of_line_wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
340 			action, TASK_KILLABLE);
341 }
342 EXPORT_SYMBOL_GPL(__rpc_wait_for_completion_task);
343 
344 /*
345  * Make an RPC task runnable.
346  *
347  * Note: If the task is ASYNC, and is being made runnable after sitting on an
348  * rpc_wait_queue, this must be called with the queue spinlock held to protect
349  * the wait queue operation.
350  * Note the ordering of rpc_test_and_set_running() and rpc_clear_queued(),
351  * which is needed to ensure that __rpc_execute() doesn't loop (due to the
352  * lockless RPC_IS_QUEUED() test) before we've had a chance to test
353  * the RPC_TASK_RUNNING flag.
354  */
355 static void rpc_make_runnable(struct workqueue_struct *wq,
356 		struct rpc_task *task)
357 {
358 	bool need_wakeup = !rpc_test_and_set_running(task);
359 
360 	rpc_clear_queued(task);
361 	if (!need_wakeup)
362 		return;
363 	if (RPC_IS_ASYNC(task)) {
364 		INIT_WORK(&task->u.tk_work, rpc_async_schedule);
365 		queue_work(wq, &task->u.tk_work);
366 	} else
367 		wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
368 }
369 
370 /*
371  * Prepare for sleeping on a wait queue.
372  * By always appending tasks to the list we ensure FIFO behavior.
373  * NB: An RPC task will only receive interrupt-driven events as long
374  * as it's on a wait queue.
375  */
376 static void __rpc_do_sleep_on_priority(struct rpc_wait_queue *q,
377 		struct rpc_task *task,
378 		unsigned char queue_priority)
379 {
380 	trace_rpc_task_sleep(task, q);
381 
382 	__rpc_add_wait_queue(q, task, queue_priority);
383 }
384 
385 static void __rpc_sleep_on_priority(struct rpc_wait_queue *q,
386 		struct rpc_task *task,
387 		unsigned char queue_priority)
388 {
389 	if (WARN_ON_ONCE(RPC_IS_QUEUED(task)))
390 		return;
391 	__rpc_do_sleep_on_priority(q, task, queue_priority);
392 }
393 
394 static void __rpc_sleep_on_priority_timeout(struct rpc_wait_queue *q,
395 		struct rpc_task *task, unsigned long timeout,
396 		unsigned char queue_priority)
397 {
398 	if (WARN_ON_ONCE(RPC_IS_QUEUED(task)))
399 		return;
400 	if (time_is_after_jiffies(timeout)) {
401 		__rpc_do_sleep_on_priority(q, task, queue_priority);
402 		__rpc_add_timer(q, task, timeout);
403 	} else
404 		task->tk_status = -ETIMEDOUT;
405 }
406 
407 static void rpc_set_tk_callback(struct rpc_task *task, rpc_action action)
408 {
409 	if (action && !WARN_ON_ONCE(task->tk_callback != NULL))
410 		task->tk_callback = action;
411 }
412 
413 static bool rpc_sleep_check_activated(struct rpc_task *task)
414 {
415 	/* We shouldn't ever put an inactive task to sleep */
416 	if (WARN_ON_ONCE(!RPC_IS_ACTIVATED(task))) {
417 		task->tk_status = -EIO;
418 		rpc_put_task_async(task);
419 		return false;
420 	}
421 	return true;
422 }
423 
424 void rpc_sleep_on_timeout(struct rpc_wait_queue *q, struct rpc_task *task,
425 				rpc_action action, unsigned long timeout)
426 {
427 	if (!rpc_sleep_check_activated(task))
428 		return;
429 
430 	rpc_set_tk_callback(task, action);
431 
432 	/*
433 	 * Protect the queue operations.
434 	 */
435 	spin_lock(&q->lock);
436 	__rpc_sleep_on_priority_timeout(q, task, timeout, task->tk_priority);
437 	spin_unlock(&q->lock);
438 }
439 EXPORT_SYMBOL_GPL(rpc_sleep_on_timeout);
440 
441 void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
442 				rpc_action action)
443 {
444 	if (!rpc_sleep_check_activated(task))
445 		return;
446 
447 	rpc_set_tk_callback(task, action);
448 
449 	WARN_ON_ONCE(task->tk_timeout != 0);
450 	/*
451 	 * Protect the queue operations.
452 	 */
453 	spin_lock(&q->lock);
454 	__rpc_sleep_on_priority(q, task, task->tk_priority);
455 	spin_unlock(&q->lock);
456 }
457 EXPORT_SYMBOL_GPL(rpc_sleep_on);
458 
459 void rpc_sleep_on_priority_timeout(struct rpc_wait_queue *q,
460 		struct rpc_task *task, unsigned long timeout, int priority)
461 {
462 	if (!rpc_sleep_check_activated(task))
463 		return;
464 
465 	priority -= RPC_PRIORITY_LOW;
466 	/*
467 	 * Protect the queue operations.
468 	 */
469 	spin_lock(&q->lock);
470 	__rpc_sleep_on_priority_timeout(q, task, timeout, priority);
471 	spin_unlock(&q->lock);
472 }
473 EXPORT_SYMBOL_GPL(rpc_sleep_on_priority_timeout);
474 
475 void rpc_sleep_on_priority(struct rpc_wait_queue *q, struct rpc_task *task,
476 		int priority)
477 {
478 	if (!rpc_sleep_check_activated(task))
479 		return;
480 
481 	WARN_ON_ONCE(task->tk_timeout != 0);
482 	priority -= RPC_PRIORITY_LOW;
483 	/*
484 	 * Protect the queue operations.
485 	 */
486 	spin_lock(&q->lock);
487 	__rpc_sleep_on_priority(q, task, priority);
488 	spin_unlock(&q->lock);
489 }
490 EXPORT_SYMBOL_GPL(rpc_sleep_on_priority);
491 
492 /**
493  * __rpc_do_wake_up_task_on_wq - wake up a single rpc_task
494  * @wq: workqueue on which to run task
495  * @queue: wait queue
496  * @task: task to be woken up
497  *
498  * Caller must hold queue->lock, and have cleared the task queued flag.
499  */
500 static void __rpc_do_wake_up_task_on_wq(struct workqueue_struct *wq,
501 		struct rpc_wait_queue *queue,
502 		struct rpc_task *task)
503 {
504 	/* Has the task been executed yet? If not, we cannot wake it up! */
505 	if (!RPC_IS_ACTIVATED(task)) {
506 		printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
507 		return;
508 	}
509 
510 	trace_rpc_task_wakeup(task, queue);
511 
512 	__rpc_remove_wait_queue(queue, task);
513 
514 	rpc_make_runnable(wq, task);
515 }
516 
517 /*
518  * Wake up a queued task while the queue lock is being held
519  */
520 static struct rpc_task *
521 rpc_wake_up_task_on_wq_queue_action_locked(struct workqueue_struct *wq,
522 		struct rpc_wait_queue *queue, struct rpc_task *task,
523 		bool (*action)(struct rpc_task *, void *), void *data)
524 {
525 	if (RPC_IS_QUEUED(task)) {
526 		smp_rmb();
527 		if (task->tk_waitqueue == queue) {
528 			if (action == NULL || action(task, data)) {
529 				__rpc_do_wake_up_task_on_wq(wq, queue, task);
530 				return task;
531 			}
532 		}
533 	}
534 	return NULL;
535 }
536 
537 /*
538  * Wake up a queued task while the queue lock is being held
539  */
540 static void rpc_wake_up_task_queue_locked(struct rpc_wait_queue *queue,
541 					  struct rpc_task *task)
542 {
543 	rpc_wake_up_task_on_wq_queue_action_locked(rpciod_workqueue, queue,
544 						   task, NULL, NULL);
545 }
546 
547 /*
548  * Wake up a task on a specific queue
549  */
550 void rpc_wake_up_queued_task(struct rpc_wait_queue *queue, struct rpc_task *task)
551 {
552 	if (!RPC_IS_QUEUED(task))
553 		return;
554 	spin_lock(&queue->lock);
555 	rpc_wake_up_task_queue_locked(queue, task);
556 	spin_unlock(&queue->lock);
557 }
558 EXPORT_SYMBOL_GPL(rpc_wake_up_queued_task);
559 
560 static bool rpc_task_action_set_status(struct rpc_task *task, void *status)
561 {
562 	task->tk_status = *(int *)status;
563 	return true;
564 }
565 
566 static void
567 rpc_wake_up_task_queue_set_status_locked(struct rpc_wait_queue *queue,
568 		struct rpc_task *task, int status)
569 {
570 	rpc_wake_up_task_on_wq_queue_action_locked(rpciod_workqueue, queue,
571 			task, rpc_task_action_set_status, &status);
572 }
573 
574 /**
575  * rpc_wake_up_queued_task_set_status - wake up a task and set task->tk_status
576  * @queue: pointer to rpc_wait_queue
577  * @task: pointer to rpc_task
578  * @status: integer error value
579  *
580  * If @task is queued on @queue, then it is woken up, and @task->tk_status is
581  * set to the value of @status.
582  */
583 void
584 rpc_wake_up_queued_task_set_status(struct rpc_wait_queue *queue,
585 		struct rpc_task *task, int status)
586 {
587 	if (!RPC_IS_QUEUED(task))
588 		return;
589 	spin_lock(&queue->lock);
590 	rpc_wake_up_task_queue_set_status_locked(queue, task, status);
591 	spin_unlock(&queue->lock);
592 }
593 
594 /*
595  * Wake up the next task on a priority queue.
596  */
597 static struct rpc_task *__rpc_find_next_queued_priority(struct rpc_wait_queue *queue)
598 {
599 	struct list_head *q;
600 	struct rpc_task *task;
601 
602 	/*
603 	 * Service the privileged queue.
604 	 */
605 	q = &queue->tasks[RPC_NR_PRIORITY - 1];
606 	if (queue->maxpriority > RPC_PRIORITY_PRIVILEGED && !list_empty(q)) {
607 		task = list_first_entry(q, struct rpc_task, u.tk_wait.list);
608 		goto out;
609 	}
610 
611 	/*
612 	 * Service a batch of tasks from a single owner.
613 	 */
614 	q = &queue->tasks[queue->priority];
615 	if (!list_empty(q) && queue->nr) {
616 		queue->nr--;
617 		task = list_first_entry(q, struct rpc_task, u.tk_wait.list);
618 		goto out;
619 	}
620 
621 	/*
622 	 * Service the next queue.
623 	 */
624 	do {
625 		if (q == &queue->tasks[0])
626 			q = &queue->tasks[queue->maxpriority];
627 		else
628 			q = q - 1;
629 		if (!list_empty(q)) {
630 			task = list_first_entry(q, struct rpc_task, u.tk_wait.list);
631 			goto new_queue;
632 		}
633 	} while (q != &queue->tasks[queue->priority]);
634 
635 	rpc_reset_waitqueue_priority(queue);
636 	return NULL;
637 
638 new_queue:
639 	rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0]));
640 out:
641 	return task;
642 }
643 
644 static struct rpc_task *__rpc_find_next_queued(struct rpc_wait_queue *queue)
645 {
646 	if (RPC_IS_PRIORITY(queue))
647 		return __rpc_find_next_queued_priority(queue);
648 	if (!list_empty(&queue->tasks[0]))
649 		return list_first_entry(&queue->tasks[0], struct rpc_task, u.tk_wait.list);
650 	return NULL;
651 }
652 
653 /*
654  * Wake up the first task on the wait queue.
655  */
656 struct rpc_task *rpc_wake_up_first_on_wq(struct workqueue_struct *wq,
657 		struct rpc_wait_queue *queue,
658 		bool (*func)(struct rpc_task *, void *), void *data)
659 {
660 	struct rpc_task	*task = NULL;
661 
662 	spin_lock(&queue->lock);
663 	task = __rpc_find_next_queued(queue);
664 	if (task != NULL)
665 		task = rpc_wake_up_task_on_wq_queue_action_locked(wq, queue,
666 				task, func, data);
667 	spin_unlock(&queue->lock);
668 
669 	return task;
670 }
671 
672 /*
673  * Wake up the first task on the wait queue.
674  */
675 struct rpc_task *rpc_wake_up_first(struct rpc_wait_queue *queue,
676 		bool (*func)(struct rpc_task *, void *), void *data)
677 {
678 	return rpc_wake_up_first_on_wq(rpciod_workqueue, queue, func, data);
679 }
680 EXPORT_SYMBOL_GPL(rpc_wake_up_first);
681 
682 static bool rpc_wake_up_next_func(struct rpc_task *task, void *data)
683 {
684 	return true;
685 }
686 
687 /*
688  * Wake up the next task on the wait queue.
689 */
690 struct rpc_task *rpc_wake_up_next(struct rpc_wait_queue *queue)
691 {
692 	return rpc_wake_up_first(queue, rpc_wake_up_next_func, NULL);
693 }
694 EXPORT_SYMBOL_GPL(rpc_wake_up_next);
695 
696 /**
697  * rpc_wake_up_locked - wake up all rpc_tasks
698  * @queue: rpc_wait_queue on which the tasks are sleeping
699  *
700  */
701 static void rpc_wake_up_locked(struct rpc_wait_queue *queue)
702 {
703 	struct rpc_task *task;
704 
705 	for (;;) {
706 		task = __rpc_find_next_queued(queue);
707 		if (task == NULL)
708 			break;
709 		rpc_wake_up_task_queue_locked(queue, task);
710 	}
711 }
712 
713 /**
714  * rpc_wake_up - wake up all rpc_tasks
715  * @queue: rpc_wait_queue on which the tasks are sleeping
716  *
717  * Grabs queue->lock
718  */
719 void rpc_wake_up(struct rpc_wait_queue *queue)
720 {
721 	spin_lock(&queue->lock);
722 	rpc_wake_up_locked(queue);
723 	spin_unlock(&queue->lock);
724 }
725 EXPORT_SYMBOL_GPL(rpc_wake_up);
726 
727 /**
728  * rpc_wake_up_status_locked - wake up all rpc_tasks and set their status value.
729  * @queue: rpc_wait_queue on which the tasks are sleeping
730  * @status: status value to set
731  */
732 static void rpc_wake_up_status_locked(struct rpc_wait_queue *queue, int status)
733 {
734 	struct rpc_task *task;
735 
736 	for (;;) {
737 		task = __rpc_find_next_queued(queue);
738 		if (task == NULL)
739 			break;
740 		rpc_wake_up_task_queue_set_status_locked(queue, task, status);
741 	}
742 }
743 
744 /**
745  * rpc_wake_up_status - wake up all rpc_tasks and set their status value.
746  * @queue: rpc_wait_queue on which the tasks are sleeping
747  * @status: status value to set
748  *
749  * Grabs queue->lock
750  */
751 void rpc_wake_up_status(struct rpc_wait_queue *queue, int status)
752 {
753 	spin_lock(&queue->lock);
754 	rpc_wake_up_status_locked(queue, status);
755 	spin_unlock(&queue->lock);
756 }
757 EXPORT_SYMBOL_GPL(rpc_wake_up_status);
758 
759 static void __rpc_queue_timer_fn(struct work_struct *work)
760 {
761 	struct rpc_wait_queue *queue = container_of(work,
762 			struct rpc_wait_queue,
763 			timer_list.dwork.work);
764 	struct rpc_task *task, *n;
765 	unsigned long expires, now, timeo;
766 
767 	spin_lock(&queue->lock);
768 	expires = now = jiffies;
769 	list_for_each_entry_safe(task, n, &queue->timer_list.list, u.tk_wait.timer_list) {
770 		timeo = task->tk_timeout;
771 		if (time_after_eq(now, timeo)) {
772 			trace_rpc_task_timeout(task, task->tk_action);
773 			task->tk_status = -ETIMEDOUT;
774 			rpc_wake_up_task_queue_locked(queue, task);
775 			continue;
776 		}
777 		if (expires == now || time_after(expires, timeo))
778 			expires = timeo;
779 	}
780 	if (!list_empty(&queue->timer_list.list))
781 		rpc_set_queue_timer(queue, expires);
782 	spin_unlock(&queue->lock);
783 }
784 
785 static void __rpc_atrun(struct rpc_task *task)
786 {
787 	if (task->tk_status == -ETIMEDOUT)
788 		task->tk_status = 0;
789 }
790 
791 /*
792  * Run a task at a later time
793  */
794 void rpc_delay(struct rpc_task *task, unsigned long delay)
795 {
796 	rpc_sleep_on_timeout(&delay_queue, task, __rpc_atrun, jiffies + delay);
797 }
798 EXPORT_SYMBOL_GPL(rpc_delay);
799 
800 /*
801  * Helper to call task->tk_ops->rpc_call_prepare
802  */
803 void rpc_prepare_task(struct rpc_task *task)
804 {
805 	task->tk_ops->rpc_call_prepare(task, task->tk_calldata);
806 }
807 
808 static void
809 rpc_init_task_statistics(struct rpc_task *task)
810 {
811 	/* Initialize retry counters */
812 	task->tk_garb_retry = 2;
813 	task->tk_cred_retry = 2;
814 	task->tk_rebind_retry = 2;
815 
816 	/* starting timestamp */
817 	task->tk_start = ktime_get();
818 }
819 
820 static void
821 rpc_reset_task_statistics(struct rpc_task *task)
822 {
823 	task->tk_timeouts = 0;
824 	task->tk_flags &= ~(RPC_CALL_MAJORSEEN|RPC_TASK_SENT);
825 	rpc_init_task_statistics(task);
826 }
827 
828 /*
829  * Helper that calls task->tk_ops->rpc_call_done if it exists
830  */
831 void rpc_exit_task(struct rpc_task *task)
832 {
833 	trace_rpc_task_end(task, task->tk_action);
834 	task->tk_action = NULL;
835 	if (task->tk_ops->rpc_count_stats)
836 		task->tk_ops->rpc_count_stats(task, task->tk_calldata);
837 	else if (task->tk_client)
838 		rpc_count_iostats(task, task->tk_client->cl_metrics);
839 	if (task->tk_ops->rpc_call_done != NULL) {
840 		trace_rpc_task_call_done(task, task->tk_ops->rpc_call_done);
841 		task->tk_ops->rpc_call_done(task, task->tk_calldata);
842 		if (task->tk_action != NULL) {
843 			/* Always release the RPC slot and buffer memory */
844 			xprt_release(task);
845 			rpc_reset_task_statistics(task);
846 		}
847 	}
848 }
849 
850 void rpc_signal_task(struct rpc_task *task)
851 {
852 	struct rpc_wait_queue *queue;
853 
854 	if (!RPC_IS_ACTIVATED(task))
855 		return;
856 
857 	trace_rpc_task_signalled(task, task->tk_action);
858 	set_bit(RPC_TASK_SIGNALLED, &task->tk_runstate);
859 	smp_mb__after_atomic();
860 	queue = READ_ONCE(task->tk_waitqueue);
861 	if (queue)
862 		rpc_wake_up_queued_task_set_status(queue, task, -ERESTARTSYS);
863 }
864 
865 void rpc_exit(struct rpc_task *task, int status)
866 {
867 	task->tk_status = status;
868 	task->tk_action = rpc_exit_task;
869 	rpc_wake_up_queued_task(task->tk_waitqueue, task);
870 }
871 EXPORT_SYMBOL_GPL(rpc_exit);
872 
873 void rpc_release_calldata(const struct rpc_call_ops *ops, void *calldata)
874 {
875 	if (ops->rpc_release != NULL)
876 		ops->rpc_release(calldata);
877 }
878 
879 static bool xprt_needs_memalloc(struct rpc_xprt *xprt, struct rpc_task *tk)
880 {
881 	if (!xprt)
882 		return false;
883 	if (!atomic_read(&xprt->swapper))
884 		return false;
885 	return test_bit(XPRT_LOCKED, &xprt->state) && xprt->snd_task == tk;
886 }
887 
888 /*
889  * This is the RPC `scheduler' (or rather, the finite state machine).
890  */
891 static void __rpc_execute(struct rpc_task *task)
892 {
893 	struct rpc_wait_queue *queue;
894 	int task_is_async = RPC_IS_ASYNC(task);
895 	int status = 0;
896 	unsigned long pflags = current->flags;
897 
898 	WARN_ON_ONCE(RPC_IS_QUEUED(task));
899 	if (RPC_IS_QUEUED(task))
900 		return;
901 
902 	for (;;) {
903 		void (*do_action)(struct rpc_task *);
904 
905 		/*
906 		 * Perform the next FSM step or a pending callback.
907 		 *
908 		 * tk_action may be NULL if the task has been killed.
909 		 * In particular, note that rpc_killall_tasks may
910 		 * do this at any time, so beware when dereferencing.
911 		 */
912 		do_action = task->tk_action;
913 		if (task->tk_callback) {
914 			do_action = task->tk_callback;
915 			task->tk_callback = NULL;
916 		}
917 		if (!do_action)
918 			break;
919 		if (RPC_IS_SWAPPER(task) ||
920 		    xprt_needs_memalloc(task->tk_xprt, task))
921 			current->flags |= PF_MEMALLOC;
922 
923 		trace_rpc_task_run_action(task, do_action);
924 		do_action(task);
925 
926 		/*
927 		 * Lockless check for whether task is sleeping or not.
928 		 */
929 		if (!RPC_IS_QUEUED(task)) {
930 			cond_resched();
931 			continue;
932 		}
933 
934 		/*
935 		 * Signalled tasks should exit rather than sleep.
936 		 */
937 		if (RPC_SIGNALLED(task)) {
938 			task->tk_rpc_status = -ERESTARTSYS;
939 			rpc_exit(task, -ERESTARTSYS);
940 		}
941 
942 		/*
943 		 * The queue->lock protects against races with
944 		 * rpc_make_runnable().
945 		 *
946 		 * Note that once we clear RPC_TASK_RUNNING on an asynchronous
947 		 * rpc_task, rpc_make_runnable() can assign it to a
948 		 * different workqueue. We therefore cannot assume that the
949 		 * rpc_task pointer may still be dereferenced.
950 		 */
951 		queue = task->tk_waitqueue;
952 		spin_lock(&queue->lock);
953 		if (!RPC_IS_QUEUED(task)) {
954 			spin_unlock(&queue->lock);
955 			continue;
956 		}
957 		rpc_clear_running(task);
958 		spin_unlock(&queue->lock);
959 		if (task_is_async)
960 			goto out;
961 
962 		/* sync task: sleep here */
963 		trace_rpc_task_sync_sleep(task, task->tk_action);
964 		status = out_of_line_wait_on_bit(&task->tk_runstate,
965 				RPC_TASK_QUEUED, rpc_wait_bit_killable,
966 				TASK_KILLABLE);
967 		if (status < 0) {
968 			/*
969 			 * When a sync task receives a signal, it exits with
970 			 * -ERESTARTSYS. In order to catch any callbacks that
971 			 * clean up after sleeping on some queue, we don't
972 			 * break the loop here, but go around once more.
973 			 */
974 			trace_rpc_task_signalled(task, task->tk_action);
975 			set_bit(RPC_TASK_SIGNALLED, &task->tk_runstate);
976 			task->tk_rpc_status = -ERESTARTSYS;
977 			rpc_exit(task, -ERESTARTSYS);
978 		}
979 		trace_rpc_task_sync_wake(task, task->tk_action);
980 	}
981 
982 	/* Release all resources associated with the task */
983 	rpc_release_task(task);
984 out:
985 	current_restore_flags(pflags, PF_MEMALLOC);
986 }
987 
988 /*
989  * User-visible entry point to the scheduler.
990  *
991  * This may be called recursively if e.g. an async NFS task updates
992  * the attributes and finds that dirty pages must be flushed.
993  * NOTE: Upon exit of this function the task is guaranteed to be
994  *	 released. In particular note that tk_release() will have
995  *	 been called, so your task memory may have been freed.
996  */
997 void rpc_execute(struct rpc_task *task)
998 {
999 	bool is_async = RPC_IS_ASYNC(task);
1000 
1001 	rpc_set_active(task);
1002 	rpc_make_runnable(rpciod_workqueue, task);
1003 	if (!is_async) {
1004 		unsigned int pflags = memalloc_nofs_save();
1005 		__rpc_execute(task);
1006 		memalloc_nofs_restore(pflags);
1007 	}
1008 }
1009 
1010 static void rpc_async_schedule(struct work_struct *work)
1011 {
1012 	unsigned int pflags = memalloc_nofs_save();
1013 
1014 	__rpc_execute(container_of(work, struct rpc_task, u.tk_work));
1015 	memalloc_nofs_restore(pflags);
1016 }
1017 
1018 /**
1019  * rpc_malloc - allocate RPC buffer resources
1020  * @task: RPC task
1021  *
1022  * A single memory region is allocated, which is split between the
1023  * RPC call and RPC reply that this task is being used for. When
1024  * this RPC is retired, the memory is released by calling rpc_free.
1025  *
1026  * To prevent rpciod from hanging, this allocator never sleeps,
1027  * returning -ENOMEM and suppressing warning if the request cannot
1028  * be serviced immediately. The caller can arrange to sleep in a
1029  * way that is safe for rpciod.
1030  *
1031  * Most requests are 'small' (under 2KiB) and can be serviced from a
1032  * mempool, ensuring that NFS reads and writes can always proceed,
1033  * and that there is good locality of reference for these buffers.
1034  */
1035 int rpc_malloc(struct rpc_task *task)
1036 {
1037 	struct rpc_rqst *rqst = task->tk_rqstp;
1038 	size_t size = rqst->rq_callsize + rqst->rq_rcvsize;
1039 	struct rpc_buffer *buf;
1040 	gfp_t gfp = rpc_task_gfp_mask();
1041 
1042 	size += sizeof(struct rpc_buffer);
1043 	if (size <= RPC_BUFFER_MAXSIZE) {
1044 		buf = kmem_cache_alloc(rpc_buffer_slabp, gfp);
1045 		/* Reach for the mempool if dynamic allocation fails */
1046 		if (!buf && RPC_IS_ASYNC(task))
1047 			buf = mempool_alloc(rpc_buffer_mempool, GFP_NOWAIT);
1048 	} else
1049 		buf = kmalloc(size, gfp);
1050 
1051 	if (!buf)
1052 		return -ENOMEM;
1053 
1054 	buf->len = size;
1055 	rqst->rq_buffer = buf->data;
1056 	rqst->rq_rbuffer = (char *)rqst->rq_buffer + rqst->rq_callsize;
1057 	return 0;
1058 }
1059 EXPORT_SYMBOL_GPL(rpc_malloc);
1060 
1061 /**
1062  * rpc_free - free RPC buffer resources allocated via rpc_malloc
1063  * @task: RPC task
1064  *
1065  */
1066 void rpc_free(struct rpc_task *task)
1067 {
1068 	void *buffer = task->tk_rqstp->rq_buffer;
1069 	size_t size;
1070 	struct rpc_buffer *buf;
1071 
1072 	buf = container_of(buffer, struct rpc_buffer, data);
1073 	size = buf->len;
1074 
1075 	if (size <= RPC_BUFFER_MAXSIZE)
1076 		mempool_free(buf, rpc_buffer_mempool);
1077 	else
1078 		kfree(buf);
1079 }
1080 EXPORT_SYMBOL_GPL(rpc_free);
1081 
1082 /*
1083  * Creation and deletion of RPC task structures
1084  */
1085 static void rpc_init_task(struct rpc_task *task, const struct rpc_task_setup *task_setup_data)
1086 {
1087 	memset(task, 0, sizeof(*task));
1088 	atomic_set(&task->tk_count, 1);
1089 	task->tk_flags  = task_setup_data->flags;
1090 	task->tk_ops = task_setup_data->callback_ops;
1091 	task->tk_calldata = task_setup_data->callback_data;
1092 	INIT_LIST_HEAD(&task->tk_task);
1093 
1094 	task->tk_priority = task_setup_data->priority - RPC_PRIORITY_LOW;
1095 	task->tk_owner = current->tgid;
1096 
1097 	/* Initialize workqueue for async tasks */
1098 	task->tk_workqueue = task_setup_data->workqueue;
1099 
1100 	task->tk_xprt = rpc_task_get_xprt(task_setup_data->rpc_client,
1101 			xprt_get(task_setup_data->rpc_xprt));
1102 
1103 	task->tk_op_cred = get_rpccred(task_setup_data->rpc_op_cred);
1104 
1105 	if (task->tk_ops->rpc_call_prepare != NULL)
1106 		task->tk_action = rpc_prepare_task;
1107 
1108 	rpc_init_task_statistics(task);
1109 }
1110 
1111 static struct rpc_task *rpc_alloc_task(void)
1112 {
1113 	struct rpc_task *task;
1114 
1115 	task = kmem_cache_alloc(rpc_task_slabp, rpc_task_gfp_mask());
1116 	if (task)
1117 		return task;
1118 	return mempool_alloc(rpc_task_mempool, GFP_NOWAIT);
1119 }
1120 
1121 /*
1122  * Create a new task for the specified client.
1123  */
1124 struct rpc_task *rpc_new_task(const struct rpc_task_setup *setup_data)
1125 {
1126 	struct rpc_task	*task = setup_data->task;
1127 	unsigned short flags = 0;
1128 
1129 	if (task == NULL) {
1130 		task = rpc_alloc_task();
1131 		if (task == NULL) {
1132 			rpc_release_calldata(setup_data->callback_ops,
1133 					     setup_data->callback_data);
1134 			return ERR_PTR(-ENOMEM);
1135 		}
1136 		flags = RPC_TASK_DYNAMIC;
1137 	}
1138 
1139 	rpc_init_task(task, setup_data);
1140 	task->tk_flags |= flags;
1141 	return task;
1142 }
1143 
1144 /*
1145  * rpc_free_task - release rpc task and perform cleanups
1146  *
1147  * Note that we free up the rpc_task _after_ rpc_release_calldata()
1148  * in order to work around a workqueue dependency issue.
1149  *
1150  * Tejun Heo states:
1151  * "Workqueue currently considers two work items to be the same if they're
1152  * on the same address and won't execute them concurrently - ie. it
1153  * makes a work item which is queued again while being executed wait
1154  * for the previous execution to complete.
1155  *
1156  * If a work function frees the work item, and then waits for an event
1157  * which should be performed by another work item and *that* work item
1158  * recycles the freed work item, it can create a false dependency loop.
1159  * There really is no reliable way to detect this short of verifying
1160  * every memory free."
1161  *
1162  */
1163 static void rpc_free_task(struct rpc_task *task)
1164 {
1165 	unsigned short tk_flags = task->tk_flags;
1166 
1167 	put_rpccred(task->tk_op_cred);
1168 	rpc_release_calldata(task->tk_ops, task->tk_calldata);
1169 
1170 	if (tk_flags & RPC_TASK_DYNAMIC)
1171 		mempool_free(task, rpc_task_mempool);
1172 }
1173 
1174 static void rpc_async_release(struct work_struct *work)
1175 {
1176 	unsigned int pflags = memalloc_nofs_save();
1177 
1178 	rpc_free_task(container_of(work, struct rpc_task, u.tk_work));
1179 	memalloc_nofs_restore(pflags);
1180 }
1181 
1182 static void rpc_release_resources_task(struct rpc_task *task)
1183 {
1184 	xprt_release(task);
1185 	if (task->tk_msg.rpc_cred) {
1186 		if (!(task->tk_flags & RPC_TASK_CRED_NOREF))
1187 			put_cred(task->tk_msg.rpc_cred);
1188 		task->tk_msg.rpc_cred = NULL;
1189 	}
1190 	rpc_task_release_client(task);
1191 }
1192 
1193 static void rpc_final_put_task(struct rpc_task *task,
1194 		struct workqueue_struct *q)
1195 {
1196 	if (q != NULL) {
1197 		INIT_WORK(&task->u.tk_work, rpc_async_release);
1198 		queue_work(q, &task->u.tk_work);
1199 	} else
1200 		rpc_free_task(task);
1201 }
1202 
1203 static void rpc_do_put_task(struct rpc_task *task, struct workqueue_struct *q)
1204 {
1205 	if (atomic_dec_and_test(&task->tk_count)) {
1206 		rpc_release_resources_task(task);
1207 		rpc_final_put_task(task, q);
1208 	}
1209 }
1210 
1211 void rpc_put_task(struct rpc_task *task)
1212 {
1213 	rpc_do_put_task(task, NULL);
1214 }
1215 EXPORT_SYMBOL_GPL(rpc_put_task);
1216 
1217 void rpc_put_task_async(struct rpc_task *task)
1218 {
1219 	rpc_do_put_task(task, task->tk_workqueue);
1220 }
1221 EXPORT_SYMBOL_GPL(rpc_put_task_async);
1222 
1223 static void rpc_release_task(struct rpc_task *task)
1224 {
1225 	WARN_ON_ONCE(RPC_IS_QUEUED(task));
1226 
1227 	rpc_release_resources_task(task);
1228 
1229 	/*
1230 	 * Note: at this point we have been removed from rpc_clnt->cl_tasks,
1231 	 * so it should be safe to use task->tk_count as a test for whether
1232 	 * or not any other processes still hold references to our rpc_task.
1233 	 */
1234 	if (atomic_read(&task->tk_count) != 1 + !RPC_IS_ASYNC(task)) {
1235 		/* Wake up anyone who may be waiting for task completion */
1236 		if (!rpc_complete_task(task))
1237 			return;
1238 	} else {
1239 		if (!atomic_dec_and_test(&task->tk_count))
1240 			return;
1241 	}
1242 	rpc_final_put_task(task, task->tk_workqueue);
1243 }
1244 
1245 int rpciod_up(void)
1246 {
1247 	return try_module_get(THIS_MODULE) ? 0 : -EINVAL;
1248 }
1249 
1250 void rpciod_down(void)
1251 {
1252 	module_put(THIS_MODULE);
1253 }
1254 
1255 /*
1256  * Start up the rpciod workqueue.
1257  */
1258 static int rpciod_start(void)
1259 {
1260 	struct workqueue_struct *wq;
1261 
1262 	/*
1263 	 * Create the rpciod thread and wait for it to start.
1264 	 */
1265 	wq = alloc_workqueue("rpciod", WQ_MEM_RECLAIM | WQ_UNBOUND, 0);
1266 	if (!wq)
1267 		goto out_failed;
1268 	rpciod_workqueue = wq;
1269 	wq = alloc_workqueue("xprtiod", WQ_UNBOUND | WQ_MEM_RECLAIM, 0);
1270 	if (!wq)
1271 		goto free_rpciod;
1272 	xprtiod_workqueue = wq;
1273 	return 1;
1274 free_rpciod:
1275 	wq = rpciod_workqueue;
1276 	rpciod_workqueue = NULL;
1277 	destroy_workqueue(wq);
1278 out_failed:
1279 	return 0;
1280 }
1281 
1282 static void rpciod_stop(void)
1283 {
1284 	struct workqueue_struct *wq = NULL;
1285 
1286 	if (rpciod_workqueue == NULL)
1287 		return;
1288 
1289 	wq = rpciod_workqueue;
1290 	rpciod_workqueue = NULL;
1291 	destroy_workqueue(wq);
1292 	wq = xprtiod_workqueue;
1293 	xprtiod_workqueue = NULL;
1294 	destroy_workqueue(wq);
1295 }
1296 
1297 void
1298 rpc_destroy_mempool(void)
1299 {
1300 	rpciod_stop();
1301 	mempool_destroy(rpc_buffer_mempool);
1302 	mempool_destroy(rpc_task_mempool);
1303 	kmem_cache_destroy(rpc_task_slabp);
1304 	kmem_cache_destroy(rpc_buffer_slabp);
1305 	rpc_destroy_wait_queue(&delay_queue);
1306 }
1307 
1308 int
1309 rpc_init_mempool(void)
1310 {
1311 	/*
1312 	 * The following is not strictly a mempool initialisation,
1313 	 * but there is no harm in doing it here
1314 	 */
1315 	rpc_init_wait_queue(&delay_queue, "delayq");
1316 	if (!rpciod_start())
1317 		goto err_nomem;
1318 
1319 	rpc_task_slabp = kmem_cache_create("rpc_tasks",
1320 					     sizeof(struct rpc_task),
1321 					     0, SLAB_HWCACHE_ALIGN,
1322 					     NULL);
1323 	if (!rpc_task_slabp)
1324 		goto err_nomem;
1325 	rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
1326 					     RPC_BUFFER_MAXSIZE,
1327 					     0, SLAB_HWCACHE_ALIGN,
1328 					     NULL);
1329 	if (!rpc_buffer_slabp)
1330 		goto err_nomem;
1331 	rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
1332 						    rpc_task_slabp);
1333 	if (!rpc_task_mempool)
1334 		goto err_nomem;
1335 	rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
1336 						      rpc_buffer_slabp);
1337 	if (!rpc_buffer_mempool)
1338 		goto err_nomem;
1339 	return 0;
1340 err_nomem:
1341 	rpc_destroy_mempool();
1342 	return -ENOMEM;
1343 }
1344