sched.c 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155
  1. /*
  2. * linux/net/sunrpc/sched.c
  3. *
  4. * Scheduling for synchronous and asynchronous RPC requests.
  5. *
  6. * Copyright (C) 1996 Olaf Kirch, <okir@monad.swb.de>
  7. *
  8. * TCP NFS related read + write fixes
  9. * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
  10. */
  11. #include <linux/module.h>
  12. #include <linux/sched.h>
  13. #include <linux/interrupt.h>
  14. #include <linux/slab.h>
  15. #include <linux/mempool.h>
  16. #include <linux/smp.h>
  17. #include <linux/spinlock.h>
  18. #include <linux/mutex.h>
  19. #include <linux/freezer.h>
  20. #include <linux/sunrpc/clnt.h>
  21. #include "sunrpc.h"
  22. #ifdef RPC_DEBUG
  23. #define RPCDBG_FACILITY RPCDBG_SCHED
  24. #endif
  25. #define CREATE_TRACE_POINTS
  26. #include <trace/events/sunrpc.h>
  27. /*
  28. * RPC slabs and memory pools
  29. */
  30. #define RPC_BUFFER_MAXSIZE (2048)
  31. #define RPC_BUFFER_POOLSIZE (8)
  32. #define RPC_TASK_POOLSIZE (8)
  33. static struct kmem_cache *rpc_task_slabp __read_mostly;
  34. static struct kmem_cache *rpc_buffer_slabp __read_mostly;
  35. static mempool_t *rpc_task_mempool __read_mostly;
  36. static mempool_t *rpc_buffer_mempool __read_mostly;
  37. static void rpc_async_schedule(struct work_struct *);
  38. static void rpc_release_task(struct rpc_task *task);
  39. static void __rpc_queue_timer_fn(unsigned long ptr);
  40. /*
  41. * RPC tasks sit here while waiting for conditions to improve.
  42. */
  43. static struct rpc_wait_queue delay_queue;
  44. /*
  45. * rpciod-related stuff
  46. */
  47. struct workqueue_struct *rpciod_workqueue;
  48. /*
  49. * Disable the timer for a given RPC task. Should be called with
  50. * queue->lock and bh_disabled in order to avoid races within
  51. * rpc_run_timer().
  52. */
  53. static void
  54. __rpc_disable_timer(struct rpc_wait_queue *queue, struct rpc_task *task)
  55. {
  56. if (task->tk_timeout == 0)
  57. return;
  58. dprintk("RPC: %5u disabling timer\n", task->tk_pid);
  59. task->tk_timeout = 0;
  60. list_del(&task->u.tk_wait.timer_list);
  61. if (list_empty(&queue->timer_list.list))
  62. del_timer(&queue->timer_list.timer);
  63. }
  64. static void
  65. rpc_set_queue_timer(struct rpc_wait_queue *queue, unsigned long expires)
  66. {
  67. queue->timer_list.expires = expires;
  68. mod_timer(&queue->timer_list.timer, expires);
  69. }
  70. /*
  71. * Set up a timer for the current task.
  72. */
  73. static void
  74. __rpc_add_timer(struct rpc_wait_queue *queue, struct rpc_task *task)
  75. {
  76. if (!task->tk_timeout)
  77. return;
  78. dprintk("RPC: %5u setting alarm for %lu ms\n",
  79. task->tk_pid, task->tk_timeout * 1000 / HZ);
  80. task->u.tk_wait.expires = jiffies + task->tk_timeout;
  81. if (list_empty(&queue->timer_list.list) || time_before(task->u.tk_wait.expires, queue->timer_list.expires))
  82. rpc_set_queue_timer(queue, task->u.tk_wait.expires);
  83. list_add(&task->u.tk_wait.timer_list, &queue->timer_list.list);
  84. }
  85. static void rpc_rotate_queue_owner(struct rpc_wait_queue *queue)
  86. {
  87. struct list_head *q = &queue->tasks[queue->priority];
  88. struct rpc_task *task;
  89. if (!list_empty(q)) {
  90. task = list_first_entry(q, struct rpc_task, u.tk_wait.list);
  91. if (task->tk_owner == queue->owner)
  92. list_move_tail(&task->u.tk_wait.list, q);
  93. }
  94. }
  95. static void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority)
  96. {
  97. if (queue->priority != priority) {
  98. /* Fairness: rotate the list when changing priority */
  99. rpc_rotate_queue_owner(queue);
  100. queue->priority = priority;
  101. }
  102. }
  103. static void rpc_set_waitqueue_owner(struct rpc_wait_queue *queue, pid_t pid)
  104. {
  105. queue->owner = pid;
  106. queue->nr = RPC_BATCH_COUNT;
  107. }
  108. static void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
  109. {
  110. rpc_set_waitqueue_priority(queue, queue->maxpriority);
  111. rpc_set_waitqueue_owner(queue, 0);
  112. }
  113. /*
  114. * Add new request to a priority queue.
  115. */
  116. static void __rpc_add_wait_queue_priority(struct rpc_wait_queue *queue,
  117. struct rpc_task *task,
  118. unsigned char queue_priority)
  119. {
  120. struct list_head *q;
  121. struct rpc_task *t;
  122. INIT_LIST_HEAD(&task->u.tk_wait.links);
  123. if (unlikely(queue_priority > queue->maxpriority))
  124. queue_priority = queue->maxpriority;
  125. if (queue_priority > queue->priority)
  126. rpc_set_waitqueue_priority(queue, queue_priority);
  127. q = &queue->tasks[queue_priority];
  128. list_for_each_entry(t, q, u.tk_wait.list) {
  129. if (t->tk_owner == task->tk_owner) {
  130. list_add_tail(&task->u.tk_wait.list, &t->u.tk_wait.links);
  131. return;
  132. }
  133. }
  134. list_add_tail(&task->u.tk_wait.list, q);
  135. }
  136. /*
  137. * Add new request to wait queue.
  138. *
  139. * Swapper tasks always get inserted at the head of the queue.
  140. * This should avoid many nasty memory deadlocks and hopefully
  141. * improve overall performance.
  142. * Everyone else gets appended to the queue to ensure proper FIFO behavior.
  143. */
  144. static void __rpc_add_wait_queue(struct rpc_wait_queue *queue,
  145. struct rpc_task *task,
  146. unsigned char queue_priority)
  147. {
  148. WARN_ON_ONCE(RPC_IS_QUEUED(task));
  149. if (RPC_IS_QUEUED(task))
  150. return;
  151. if (RPC_IS_PRIORITY(queue))
  152. __rpc_add_wait_queue_priority(queue, task, queue_priority);
  153. else if (RPC_IS_SWAPPER(task))
  154. list_add(&task->u.tk_wait.list, &queue->tasks[0]);
  155. else
  156. list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
  157. task->tk_waitqueue = queue;
  158. queue->qlen++;
  159. /* barrier matches the read in rpc_wake_up_task_queue_locked() */
  160. smp_wmb();
  161. rpc_set_queued(task);
  162. dprintk("RPC: %5u added to queue %p \"%s\"\n",
  163. task->tk_pid, queue, rpc_qname(queue));
  164. }
  165. /*
  166. * Remove request from a priority queue.
  167. */
  168. static void __rpc_remove_wait_queue_priority(struct rpc_task *task)
  169. {
  170. struct rpc_task *t;
  171. if (!list_empty(&task->u.tk_wait.links)) {
  172. t = list_entry(task->u.tk_wait.links.next, struct rpc_task, u.tk_wait.list);
  173. list_move(&t->u.tk_wait.list, &task->u.tk_wait.list);
  174. list_splice_init(&task->u.tk_wait.links, &t->u.tk_wait.links);
  175. }
  176. }
  177. /*
  178. * Remove request from queue.
  179. * Note: must be called with spin lock held.
  180. */
  181. static void __rpc_remove_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
  182. {
  183. __rpc_disable_timer(queue, task);
  184. if (RPC_IS_PRIORITY(queue))
  185. __rpc_remove_wait_queue_priority(task);
  186. list_del(&task->u.tk_wait.list);
  187. queue->qlen--;
  188. dprintk("RPC: %5u removed from queue %p \"%s\"\n",
  189. task->tk_pid, queue, rpc_qname(queue));
  190. }
  191. static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, unsigned char nr_queues)
  192. {
  193. int i;
  194. spin_lock_init(&queue->lock);
  195. for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
  196. INIT_LIST_HEAD(&queue->tasks[i]);
  197. queue->maxpriority = nr_queues - 1;
  198. rpc_reset_waitqueue_priority(queue);
  199. queue->qlen = 0;
  200. setup_timer(&queue->timer_list.timer, __rpc_queue_timer_fn, (unsigned long)queue);
  201. INIT_LIST_HEAD(&queue->timer_list.list);
  202. rpc_assign_waitqueue_name(queue, qname);
  203. }
  204. void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
  205. {
  206. __rpc_init_priority_wait_queue(queue, qname, RPC_NR_PRIORITY);
  207. }
  208. EXPORT_SYMBOL_GPL(rpc_init_priority_wait_queue);
  209. void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
  210. {
  211. __rpc_init_priority_wait_queue(queue, qname, 1);
  212. }
  213. EXPORT_SYMBOL_GPL(rpc_init_wait_queue);
  214. void rpc_destroy_wait_queue(struct rpc_wait_queue *queue)
  215. {
  216. del_timer_sync(&queue->timer_list.timer);
  217. }
  218. EXPORT_SYMBOL_GPL(rpc_destroy_wait_queue);
  219. static int rpc_wait_bit_killable(void *word)
  220. {
  221. if (fatal_signal_pending(current))
  222. return -ERESTARTSYS;
  223. freezable_schedule();
  224. return 0;
  225. }
  226. #ifdef RPC_DEBUG
  227. static void rpc_task_set_debuginfo(struct rpc_task *task)
  228. {
  229. static atomic_t rpc_pid;
  230. task->tk_pid = atomic_inc_return(&rpc_pid);
  231. }
  232. #else
  233. static inline void rpc_task_set_debuginfo(struct rpc_task *task)
  234. {
  235. }
  236. #endif
  237. static void rpc_set_active(struct rpc_task *task)
  238. {
  239. trace_rpc_task_begin(task->tk_client, task, NULL);
  240. rpc_task_set_debuginfo(task);
  241. set_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
  242. }
  243. /*
  244. * Mark an RPC call as having completed by clearing the 'active' bit
  245. * and then waking up all tasks that were sleeping.
  246. */
  247. static int rpc_complete_task(struct rpc_task *task)
  248. {
  249. void *m = &task->tk_runstate;
  250. wait_queue_head_t *wq = bit_waitqueue(m, RPC_TASK_ACTIVE);
  251. struct wait_bit_key k = __WAIT_BIT_KEY_INITIALIZER(m, RPC_TASK_ACTIVE);
  252. unsigned long flags;
  253. int ret;
  254. trace_rpc_task_complete(task->tk_client, task, NULL);
  255. spin_lock_irqsave(&wq->lock, flags);
  256. clear_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
  257. ret = atomic_dec_and_test(&task->tk_count);
  258. if (waitqueue_active(wq))
  259. __wake_up_locked_key(wq, TASK_NORMAL, &k);
  260. spin_unlock_irqrestore(&wq->lock, flags);
  261. return ret;
  262. }
  263. /*
  264. * Allow callers to wait for completion of an RPC call
  265. *
  266. * Note the use of out_of_line_wait_on_bit() rather than wait_on_bit()
  267. * to enforce taking of the wq->lock and hence avoid races with
  268. * rpc_complete_task().
  269. */
  270. int __rpc_wait_for_completion_task(struct rpc_task *task, int (*action)(void *))
  271. {
  272. if (action == NULL)
  273. action = rpc_wait_bit_killable;
  274. return out_of_line_wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
  275. action, TASK_KILLABLE);
  276. }
  277. EXPORT_SYMBOL_GPL(__rpc_wait_for_completion_task);
  278. /*
  279. * Make an RPC task runnable.
  280. *
  281. * Note: If the task is ASYNC, and is being made runnable after sitting on an
  282. * rpc_wait_queue, this must be called with the queue spinlock held to protect
  283. * the wait queue operation.
  284. * Note the ordering of rpc_test_and_set_running() and rpc_clear_queued(),
  285. * which is needed to ensure that __rpc_execute() doesn't loop (due to the
  286. * lockless RPC_IS_QUEUED() test) before we've had a chance to test
  287. * the RPC_TASK_RUNNING flag.
  288. */
  289. static void rpc_make_runnable(struct rpc_task *task)
  290. {
  291. bool need_wakeup = !rpc_test_and_set_running(task);
  292. rpc_clear_queued(task);
  293. if (!need_wakeup)
  294. return;
  295. if (RPC_IS_ASYNC(task)) {
  296. INIT_WORK(&task->u.tk_work, rpc_async_schedule);
  297. queue_work(rpciod_workqueue, &task->u.tk_work);
  298. } else
  299. wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
  300. }
  301. /*
  302. * Prepare for sleeping on a wait queue.
  303. * By always appending tasks to the list we ensure FIFO behavior.
  304. * NB: An RPC task will only receive interrupt-driven events as long
  305. * as it's on a wait queue.
  306. */
  307. static void __rpc_sleep_on_priority(struct rpc_wait_queue *q,
  308. struct rpc_task *task,
  309. rpc_action action,
  310. unsigned char queue_priority)
  311. {
  312. dprintk("RPC: %5u sleep_on(queue \"%s\" time %lu)\n",
  313. task->tk_pid, rpc_qname(q), jiffies);
  314. trace_rpc_task_sleep(task->tk_client, task, q);
  315. __rpc_add_wait_queue(q, task, queue_priority);
  316. WARN_ON_ONCE(task->tk_callback != NULL);
  317. task->tk_callback = action;
  318. __rpc_add_timer(q, task);
  319. }
  320. void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
  321. rpc_action action)
  322. {
  323. /* We shouldn't ever put an inactive task to sleep */
  324. WARN_ON_ONCE(!RPC_IS_ACTIVATED(task));
  325. if (!RPC_IS_ACTIVATED(task)) {
  326. task->tk_status = -EIO;
  327. rpc_put_task_async(task);
  328. return;
  329. }
  330. /*
  331. * Protect the queue operations.
  332. */
  333. spin_lock_bh(&q->lock);
  334. __rpc_sleep_on_priority(q, task, action, task->tk_priority);
  335. spin_unlock_bh(&q->lock);
  336. }
  337. EXPORT_SYMBOL_GPL(rpc_sleep_on);
  338. void rpc_sleep_on_priority(struct rpc_wait_queue *q, struct rpc_task *task,
  339. rpc_action action, int priority)
  340. {
  341. /* We shouldn't ever put an inactive task to sleep */
  342. WARN_ON_ONCE(!RPC_IS_ACTIVATED(task));
  343. if (!RPC_IS_ACTIVATED(task)) {
  344. task->tk_status = -EIO;
  345. rpc_put_task_async(task);
  346. return;
  347. }
  348. /*
  349. * Protect the queue operations.
  350. */
  351. spin_lock_bh(&q->lock);
  352. __rpc_sleep_on_priority(q, task, action, priority - RPC_PRIORITY_LOW);
  353. spin_unlock_bh(&q->lock);
  354. }
  355. EXPORT_SYMBOL_GPL(rpc_sleep_on_priority);
  356. /**
  357. * __rpc_do_wake_up_task - wake up a single rpc_task
  358. * @queue: wait queue
  359. * @task: task to be woken up
  360. *
  361. * Caller must hold queue->lock, and have cleared the task queued flag.
  362. */
  363. static void __rpc_do_wake_up_task(struct rpc_wait_queue *queue, struct rpc_task *task)
  364. {
  365. dprintk("RPC: %5u __rpc_wake_up_task (now %lu)\n",
  366. task->tk_pid, jiffies);
  367. /* Has the task been executed yet? If not, we cannot wake it up! */
  368. if (!RPC_IS_ACTIVATED(task)) {
  369. printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
  370. return;
  371. }
  372. trace_rpc_task_wakeup(task->tk_client, task, queue);
  373. __rpc_remove_wait_queue(queue, task);
  374. rpc_make_runnable(task);
  375. dprintk("RPC: __rpc_wake_up_task done\n");
  376. }
  377. /*
  378. * Wake up a queued task while the queue lock is being held
  379. */
  380. static void rpc_wake_up_task_queue_locked(struct rpc_wait_queue *queue, struct rpc_task *task)
  381. {
  382. if (RPC_IS_QUEUED(task)) {
  383. smp_rmb();
  384. if (task->tk_waitqueue == queue)
  385. __rpc_do_wake_up_task(queue, task);
  386. }
  387. }
  388. /*
  389. * Tests whether rpc queue is empty
  390. */
  391. int rpc_queue_empty(struct rpc_wait_queue *queue)
  392. {
  393. int res;
  394. spin_lock_bh(&queue->lock);
  395. res = queue->qlen;
  396. spin_unlock_bh(&queue->lock);
  397. return res == 0;
  398. }
  399. EXPORT_SYMBOL_GPL(rpc_queue_empty);
  400. /*
  401. * Wake up a task on a specific queue
  402. */
  403. void rpc_wake_up_queued_task(struct rpc_wait_queue *queue, struct rpc_task *task)
  404. {
  405. spin_lock_bh(&queue->lock);
  406. rpc_wake_up_task_queue_locked(queue, task);
  407. spin_unlock_bh(&queue->lock);
  408. }
  409. EXPORT_SYMBOL_GPL(rpc_wake_up_queued_task);
  410. /*
  411. * Wake up the next task on a priority queue.
  412. */
  413. static struct rpc_task *__rpc_find_next_queued_priority(struct rpc_wait_queue *queue)
  414. {
  415. struct list_head *q;
  416. struct rpc_task *task;
  417. /*
  418. * Service a batch of tasks from a single owner.
  419. */
  420. q = &queue->tasks[queue->priority];
  421. if (!list_empty(q)) {
  422. task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
  423. if (queue->owner == task->tk_owner) {
  424. if (--queue->nr)
  425. goto out;
  426. list_move_tail(&task->u.tk_wait.list, q);
  427. }
  428. /*
  429. * Check if we need to switch queues.
  430. */
  431. goto new_owner;
  432. }
  433. /*
  434. * Service the next queue.
  435. */
  436. do {
  437. if (q == &queue->tasks[0])
  438. q = &queue->tasks[queue->maxpriority];
  439. else
  440. q = q - 1;
  441. if (!list_empty(q)) {
  442. task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
  443. goto new_queue;
  444. }
  445. } while (q != &queue->tasks[queue->priority]);
  446. rpc_reset_waitqueue_priority(queue);
  447. return NULL;
  448. new_queue:
  449. rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0]));
  450. new_owner:
  451. rpc_set_waitqueue_owner(queue, task->tk_owner);
  452. out:
  453. return task;
  454. }
  455. static struct rpc_task *__rpc_find_next_queued(struct rpc_wait_queue *queue)
  456. {
  457. if (RPC_IS_PRIORITY(queue))
  458. return __rpc_find_next_queued_priority(queue);
  459. if (!list_empty(&queue->tasks[0]))
  460. return list_first_entry(&queue->tasks[0], struct rpc_task, u.tk_wait.list);
  461. return NULL;
  462. }
  463. /*
  464. * Wake up the first task on the wait queue.
  465. */
  466. struct rpc_task *rpc_wake_up_first(struct rpc_wait_queue *queue,
  467. bool (*func)(struct rpc_task *, void *), void *data)
  468. {
  469. struct rpc_task *task = NULL;
  470. dprintk("RPC: wake_up_first(%p \"%s\")\n",
  471. queue, rpc_qname(queue));
  472. spin_lock_bh(&queue->lock);
  473. task = __rpc_find_next_queued(queue);
  474. if (task != NULL) {
  475. if (func(task, data))
  476. rpc_wake_up_task_queue_locked(queue, task);
  477. else
  478. task = NULL;
  479. }
  480. spin_unlock_bh(&queue->lock);
  481. return task;
  482. }
  483. EXPORT_SYMBOL_GPL(rpc_wake_up_first);
  484. static bool rpc_wake_up_next_func(struct rpc_task *task, void *data)
  485. {
  486. return true;
  487. }
  488. /*
  489. * Wake up the next task on the wait queue.
  490. */
  491. struct rpc_task *rpc_wake_up_next(struct rpc_wait_queue *queue)
  492. {
  493. return rpc_wake_up_first(queue, rpc_wake_up_next_func, NULL);
  494. }
  495. EXPORT_SYMBOL_GPL(rpc_wake_up_next);
  496. /**
  497. * rpc_wake_up - wake up all rpc_tasks
  498. * @queue: rpc_wait_queue on which the tasks are sleeping
  499. *
  500. * Grabs queue->lock
  501. */
  502. void rpc_wake_up(struct rpc_wait_queue *queue)
  503. {
  504. struct list_head *head;
  505. spin_lock_bh(&queue->lock);
  506. head = &queue->tasks[queue->maxpriority];
  507. for (;;) {
  508. while (!list_empty(head)) {
  509. struct rpc_task *task;
  510. task = list_first_entry(head,
  511. struct rpc_task,
  512. u.tk_wait.list);
  513. rpc_wake_up_task_queue_locked(queue, task);
  514. }
  515. if (head == &queue->tasks[0])
  516. break;
  517. head--;
  518. }
  519. spin_unlock_bh(&queue->lock);
  520. }
  521. EXPORT_SYMBOL_GPL(rpc_wake_up);
  522. /**
  523. * rpc_wake_up_status - wake up all rpc_tasks and set their status value.
  524. * @queue: rpc_wait_queue on which the tasks are sleeping
  525. * @status: status value to set
  526. *
  527. * Grabs queue->lock
  528. */
  529. void rpc_wake_up_status(struct rpc_wait_queue *queue, int status)
  530. {
  531. struct list_head *head;
  532. spin_lock_bh(&queue->lock);
  533. head = &queue->tasks[queue->maxpriority];
  534. for (;;) {
  535. while (!list_empty(head)) {
  536. struct rpc_task *task;
  537. task = list_first_entry(head,
  538. struct rpc_task,
  539. u.tk_wait.list);
  540. task->tk_status = status;
  541. rpc_wake_up_task_queue_locked(queue, task);
  542. }
  543. if (head == &queue->tasks[0])
  544. break;
  545. head--;
  546. }
  547. spin_unlock_bh(&queue->lock);
  548. }
  549. EXPORT_SYMBOL_GPL(rpc_wake_up_status);
  550. static void __rpc_queue_timer_fn(unsigned long ptr)
  551. {
  552. struct rpc_wait_queue *queue = (struct rpc_wait_queue *)ptr;
  553. struct rpc_task *task, *n;
  554. unsigned long expires, now, timeo;
  555. spin_lock(&queue->lock);
  556. expires = now = jiffies;
  557. list_for_each_entry_safe(task, n, &queue->timer_list.list, u.tk_wait.timer_list) {
  558. timeo = task->u.tk_wait.expires;
  559. if (time_after_eq(now, timeo)) {
  560. dprintk("RPC: %5u timeout\n", task->tk_pid);
  561. task->tk_status = -ETIMEDOUT;
  562. rpc_wake_up_task_queue_locked(queue, task);
  563. continue;
  564. }
  565. if (expires == now || time_after(expires, timeo))
  566. expires = timeo;
  567. }
  568. if (!list_empty(&queue->timer_list.list))
  569. rpc_set_queue_timer(queue, expires);
  570. spin_unlock(&queue->lock);
  571. }
  572. static void __rpc_atrun(struct rpc_task *task)
  573. {
  574. task->tk_status = 0;
  575. }
  576. /*
  577. * Run a task at a later time
  578. */
  579. void rpc_delay(struct rpc_task *task, unsigned long delay)
  580. {
  581. task->tk_timeout = delay;
  582. rpc_sleep_on(&delay_queue, task, __rpc_atrun);
  583. }
  584. EXPORT_SYMBOL_GPL(rpc_delay);
  585. /*
  586. * Helper to call task->tk_ops->rpc_call_prepare
  587. */
  588. void rpc_prepare_task(struct rpc_task *task)
  589. {
  590. task->tk_ops->rpc_call_prepare(task, task->tk_calldata);
  591. }
  592. static void
  593. rpc_init_task_statistics(struct rpc_task *task)
  594. {
  595. /* Initialize retry counters */
  596. task->tk_garb_retry = 2;
  597. task->tk_cred_retry = 2;
  598. task->tk_rebind_retry = 2;
  599. /* starting timestamp */
  600. task->tk_start = ktime_get();
  601. }
  602. static void
  603. rpc_reset_task_statistics(struct rpc_task *task)
  604. {
  605. task->tk_timeouts = 0;
  606. task->tk_flags &= ~(RPC_CALL_MAJORSEEN|RPC_TASK_KILLED|RPC_TASK_SENT);
  607. rpc_init_task_statistics(task);
  608. }
  609. /*
  610. * Helper that calls task->tk_ops->rpc_call_done if it exists
  611. */
  612. void rpc_exit_task(struct rpc_task *task)
  613. {
  614. task->tk_action = NULL;
  615. if (task->tk_ops->rpc_call_done != NULL) {
  616. task->tk_ops->rpc_call_done(task, task->tk_calldata);
  617. if (task->tk_action != NULL) {
  618. WARN_ON(RPC_ASSASSINATED(task));
  619. /* Always release the RPC slot and buffer memory */
  620. xprt_release(task);
  621. rpc_reset_task_statistics(task);
  622. }
  623. }
  624. }
  625. void rpc_exit(struct rpc_task *task, int status)
  626. {
  627. task->tk_status = status;
  628. task->tk_action = rpc_exit_task;
  629. if (RPC_IS_QUEUED(task))
  630. rpc_wake_up_queued_task(task->tk_waitqueue, task);
  631. }
  632. EXPORT_SYMBOL_GPL(rpc_exit);
  633. void rpc_release_calldata(const struct rpc_call_ops *ops, void *calldata)
  634. {
  635. if (ops->rpc_release != NULL)
  636. ops->rpc_release(calldata);
  637. }
  638. /*
  639. * This is the RPC `scheduler' (or rather, the finite state machine).
  640. */
  641. static void __rpc_execute(struct rpc_task *task)
  642. {
  643. struct rpc_wait_queue *queue;
  644. int task_is_async = RPC_IS_ASYNC(task);
  645. int status = 0;
  646. dprintk("RPC: %5u __rpc_execute flags=0x%x\n",
  647. task->tk_pid, task->tk_flags);
  648. WARN_ON_ONCE(RPC_IS_QUEUED(task));
  649. if (RPC_IS_QUEUED(task))
  650. return;
  651. for (;;) {
  652. void (*do_action)(struct rpc_task *);
  653. /*
  654. * Execute any pending callback first.
  655. */
  656. do_action = task->tk_callback;
  657. task->tk_callback = NULL;
  658. if (do_action == NULL) {
  659. /*
  660. * Perform the next FSM step.
  661. * tk_action may be NULL if the task has been killed.
  662. * In particular, note that rpc_killall_tasks may
  663. * do this at any time, so beware when dereferencing.
  664. */
  665. do_action = task->tk_action;
  666. if (do_action == NULL)
  667. break;
  668. }
  669. trace_rpc_task_run_action(task->tk_client, task, task->tk_action);
  670. do_action(task);
  671. /*
  672. * Lockless check for whether task is sleeping or not.
  673. */
  674. if (!RPC_IS_QUEUED(task))
  675. continue;
  676. /*
  677. * The queue->lock protects against races with
  678. * rpc_make_runnable().
  679. *
  680. * Note that once we clear RPC_TASK_RUNNING on an asynchronous
  681. * rpc_task, rpc_make_runnable() can assign it to a
  682. * different workqueue. We therefore cannot assume that the
  683. * rpc_task pointer may still be dereferenced.
  684. */
  685. queue = task->tk_waitqueue;
  686. spin_lock_bh(&queue->lock);
  687. if (!RPC_IS_QUEUED(task)) {
  688. spin_unlock_bh(&queue->lock);
  689. continue;
  690. }
  691. rpc_clear_running(task);
  692. spin_unlock_bh(&queue->lock);
  693. if (task_is_async)
  694. return;
  695. /* sync task: sleep here */
  696. dprintk("RPC: %5u sync task going to sleep\n", task->tk_pid);
  697. status = out_of_line_wait_on_bit(&task->tk_runstate,
  698. RPC_TASK_QUEUED, rpc_wait_bit_killable,
  699. TASK_KILLABLE);
  700. if (status == -ERESTARTSYS) {
  701. /*
  702. * When a sync task receives a signal, it exits with
  703. * -ERESTARTSYS. In order to catch any callbacks that
  704. * clean up after sleeping on some queue, we don't
  705. * break the loop here, but go around once more.
  706. */
  707. dprintk("RPC: %5u got signal\n", task->tk_pid);
  708. task->tk_flags |= RPC_TASK_KILLED;
  709. rpc_exit(task, -ERESTARTSYS);
  710. }
  711. rpc_set_running(task);
  712. dprintk("RPC: %5u sync task resuming\n", task->tk_pid);
  713. }
  714. dprintk("RPC: %5u return %d, status %d\n", task->tk_pid, status,
  715. task->tk_status);
  716. /* Release all resources associated with the task */
  717. rpc_release_task(task);
  718. }
  719. /*
  720. * User-visible entry point to the scheduler.
  721. *
  722. * This may be called recursively if e.g. an async NFS task updates
  723. * the attributes and finds that dirty pages must be flushed.
  724. * NOTE: Upon exit of this function the task is guaranteed to be
  725. * released. In particular note that tk_release() will have
  726. * been called, so your task memory may have been freed.
  727. */
  728. void rpc_execute(struct rpc_task *task)
  729. {
  730. bool is_async = RPC_IS_ASYNC(task);
  731. rpc_set_active(task);
  732. rpc_make_runnable(task);
  733. if (!is_async)
  734. __rpc_execute(task);
  735. }
  736. static void rpc_async_schedule(struct work_struct *work)
  737. {
  738. current->flags |= PF_FSTRANS;
  739. __rpc_execute(container_of(work, struct rpc_task, u.tk_work));
  740. current->flags &= ~PF_FSTRANS;
  741. }
  742. /**
  743. * rpc_malloc - allocate an RPC buffer
  744. * @task: RPC task that will use this buffer
  745. * @size: requested byte size
  746. *
  747. * To prevent rpciod from hanging, this allocator never sleeps,
  748. * returning NULL if the request cannot be serviced immediately.
  749. * The caller can arrange to sleep in a way that is safe for rpciod.
  750. *
  751. * Most requests are 'small' (under 2KiB) and can be serviced from a
  752. * mempool, ensuring that NFS reads and writes can always proceed,
  753. * and that there is good locality of reference for these buffers.
  754. *
  755. * In order to avoid memory starvation triggering more writebacks of
  756. * NFS requests, we avoid using GFP_KERNEL.
  757. */
  758. void *rpc_malloc(struct rpc_task *task, size_t size)
  759. {
  760. struct rpc_buffer *buf;
  761. gfp_t gfp = GFP_NOWAIT;
  762. if (RPC_IS_SWAPPER(task))
  763. gfp |= __GFP_MEMALLOC;
  764. size += sizeof(struct rpc_buffer);
  765. if (size <= RPC_BUFFER_MAXSIZE)
  766. buf = mempool_alloc(rpc_buffer_mempool, gfp);
  767. else
  768. buf = kmalloc(size, gfp);
  769. if (!buf)
  770. return NULL;
  771. buf->len = size;
  772. dprintk("RPC: %5u allocated buffer of size %zu at %p\n",
  773. task->tk_pid, size, buf);
  774. return &buf->data;
  775. }
  776. EXPORT_SYMBOL_GPL(rpc_malloc);
  777. /**
  778. * rpc_free - free buffer allocated via rpc_malloc
  779. * @buffer: buffer to free
  780. *
  781. */
  782. void rpc_free(void *buffer)
  783. {
  784. size_t size;
  785. struct rpc_buffer *buf;
  786. if (!buffer)
  787. return;
  788. buf = container_of(buffer, struct rpc_buffer, data);
  789. size = buf->len;
  790. dprintk("RPC: freeing buffer of size %zu at %p\n",
  791. size, buf);
  792. if (size <= RPC_BUFFER_MAXSIZE)
  793. mempool_free(buf, rpc_buffer_mempool);
  794. else
  795. kfree(buf);
  796. }
  797. EXPORT_SYMBOL_GPL(rpc_free);
  798. /*
  799. * Creation and deletion of RPC task structures
  800. */
  801. static void rpc_init_task(struct rpc_task *task, const struct rpc_task_setup *task_setup_data)
  802. {
  803. memset(task, 0, sizeof(*task));
  804. atomic_set(&task->tk_count, 1);
  805. task->tk_flags = task_setup_data->flags;
  806. task->tk_ops = task_setup_data->callback_ops;
  807. task->tk_calldata = task_setup_data->callback_data;
  808. INIT_LIST_HEAD(&task->tk_task);
  809. task->tk_priority = task_setup_data->priority - RPC_PRIORITY_LOW;
  810. task->tk_owner = current->tgid;
  811. /* Initialize workqueue for async tasks */
  812. task->tk_workqueue = task_setup_data->workqueue;
  813. if (task->tk_ops->rpc_call_prepare != NULL)
  814. task->tk_action = rpc_prepare_task;
  815. rpc_init_task_statistics(task);
  816. dprintk("RPC: new task initialized, procpid %u\n",
  817. task_pid_nr(current));
  818. }
  819. static struct rpc_task *
  820. rpc_alloc_task(void)
  821. {
  822. return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOIO);
  823. }
  824. /*
  825. * Create a new task for the specified client.
  826. */
  827. struct rpc_task *rpc_new_task(const struct rpc_task_setup *setup_data)
  828. {
  829. struct rpc_task *task = setup_data->task;
  830. unsigned short flags = 0;
  831. if (task == NULL) {
  832. task = rpc_alloc_task();
  833. if (task == NULL) {
  834. rpc_release_calldata(setup_data->callback_ops,
  835. setup_data->callback_data);
  836. return ERR_PTR(-ENOMEM);
  837. }
  838. flags = RPC_TASK_DYNAMIC;
  839. }
  840. rpc_init_task(task, setup_data);
  841. task->tk_flags |= flags;
  842. dprintk("RPC: allocated task %p\n", task);
  843. return task;
  844. }
  845. /*
  846. * rpc_free_task - release rpc task and perform cleanups
  847. *
  848. * Note that we free up the rpc_task _after_ rpc_release_calldata()
  849. * in order to work around a workqueue dependency issue.
  850. *
  851. * Tejun Heo states:
  852. * "Workqueue currently considers two work items to be the same if they're
  853. * on the same address and won't execute them concurrently - ie. it
  854. * makes a work item which is queued again while being executed wait
  855. * for the previous execution to complete.
  856. *
  857. * If a work function frees the work item, and then waits for an event
  858. * which should be performed by another work item and *that* work item
  859. * recycles the freed work item, it can create a false dependency loop.
  860. * There really is no reliable way to detect this short of verifying
  861. * every memory free."
  862. *
  863. */
  864. static void rpc_free_task(struct rpc_task *task)
  865. {
  866. unsigned short tk_flags = task->tk_flags;
  867. rpc_release_calldata(task->tk_ops, task->tk_calldata);
  868. if (tk_flags & RPC_TASK_DYNAMIC) {
  869. dprintk("RPC: %5u freeing task\n", task->tk_pid);
  870. mempool_free(task, rpc_task_mempool);
  871. }
  872. }
  873. static void rpc_async_release(struct work_struct *work)
  874. {
  875. rpc_free_task(container_of(work, struct rpc_task, u.tk_work));
  876. }
  877. static void rpc_release_resources_task(struct rpc_task *task)
  878. {
  879. xprt_release(task);
  880. if (task->tk_msg.rpc_cred) {
  881. put_rpccred(task->tk_msg.rpc_cred);
  882. task->tk_msg.rpc_cred = NULL;
  883. }
  884. rpc_task_release_client(task);
  885. }
  886. static void rpc_final_put_task(struct rpc_task *task,
  887. struct workqueue_struct *q)
  888. {
  889. if (q != NULL) {
  890. INIT_WORK(&task->u.tk_work, rpc_async_release);
  891. queue_work(q, &task->u.tk_work);
  892. } else
  893. rpc_free_task(task);
  894. }
  895. static void rpc_do_put_task(struct rpc_task *task, struct workqueue_struct *q)
  896. {
  897. if (atomic_dec_and_test(&task->tk_count)) {
  898. rpc_release_resources_task(task);
  899. rpc_final_put_task(task, q);
  900. }
  901. }
  902. void rpc_put_task(struct rpc_task *task)
  903. {
  904. rpc_do_put_task(task, NULL);
  905. }
  906. EXPORT_SYMBOL_GPL(rpc_put_task);
  907. void rpc_put_task_async(struct rpc_task *task)
  908. {
  909. rpc_do_put_task(task, task->tk_workqueue);
  910. }
  911. EXPORT_SYMBOL_GPL(rpc_put_task_async);
  912. static void rpc_release_task(struct rpc_task *task)
  913. {
  914. dprintk("RPC: %5u release task\n", task->tk_pid);
  915. WARN_ON_ONCE(RPC_IS_QUEUED(task));
  916. rpc_release_resources_task(task);
  917. /*
  918. * Note: at this point we have been removed from rpc_clnt->cl_tasks,
  919. * so it should be safe to use task->tk_count as a test for whether
  920. * or not any other processes still hold references to our rpc_task.
  921. */
  922. if (atomic_read(&task->tk_count) != 1 + !RPC_IS_ASYNC(task)) {
  923. /* Wake up anyone who may be waiting for task completion */
  924. if (!rpc_complete_task(task))
  925. return;
  926. } else {
  927. if (!atomic_dec_and_test(&task->tk_count))
  928. return;
  929. }
  930. rpc_final_put_task(task, task->tk_workqueue);
  931. }
  932. int rpciod_up(void)
  933. {
  934. return try_module_get(THIS_MODULE) ? 0 : -EINVAL;
  935. }
  936. void rpciod_down(void)
  937. {
  938. module_put(THIS_MODULE);
  939. }
  940. /*
  941. * Start up the rpciod workqueue.
  942. */
  943. static int rpciod_start(void)
  944. {
  945. struct workqueue_struct *wq;
  946. /*
  947. * Create the rpciod thread and wait for it to start.
  948. */
  949. dprintk("RPC: creating workqueue rpciod\n");
  950. wq = alloc_workqueue("rpciod", WQ_MEM_RECLAIM, 1);
  951. rpciod_workqueue = wq;
  952. return rpciod_workqueue != NULL;
  953. }
  954. static void rpciod_stop(void)
  955. {
  956. struct workqueue_struct *wq = NULL;
  957. if (rpciod_workqueue == NULL)
  958. return;
  959. dprintk("RPC: destroying workqueue rpciod\n");
  960. wq = rpciod_workqueue;
  961. rpciod_workqueue = NULL;
  962. destroy_workqueue(wq);
  963. }
  964. void
  965. rpc_destroy_mempool(void)
  966. {
  967. rpciod_stop();
  968. if (rpc_buffer_mempool)
  969. mempool_destroy(rpc_buffer_mempool);
  970. if (rpc_task_mempool)
  971. mempool_destroy(rpc_task_mempool);
  972. if (rpc_task_slabp)
  973. kmem_cache_destroy(rpc_task_slabp);
  974. if (rpc_buffer_slabp)
  975. kmem_cache_destroy(rpc_buffer_slabp);
  976. rpc_destroy_wait_queue(&delay_queue);
  977. }
  978. int
  979. rpc_init_mempool(void)
  980. {
  981. /*
  982. * The following is not strictly a mempool initialisation,
  983. * but there is no harm in doing it here
  984. */
  985. rpc_init_wait_queue(&delay_queue, "delayq");
  986. if (!rpciod_start())
  987. goto err_nomem;
  988. rpc_task_slabp = kmem_cache_create("rpc_tasks",
  989. sizeof(struct rpc_task),
  990. 0, SLAB_HWCACHE_ALIGN,
  991. NULL);
  992. if (!rpc_task_slabp)
  993. goto err_nomem;
  994. rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
  995. RPC_BUFFER_MAXSIZE,
  996. 0, SLAB_HWCACHE_ALIGN,
  997. NULL);
  998. if (!rpc_buffer_slabp)
  999. goto err_nomem;
  1000. rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
  1001. rpc_task_slabp);
  1002. if (!rpc_task_mempool)
  1003. goto err_nomem;
  1004. rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
  1005. rpc_buffer_slabp);
  1006. if (!rpc_buffer_mempool)
  1007. goto err_nomem;
  1008. return 0;
  1009. err_nomem:
  1010. rpc_destroy_mempool();
  1011. return -ENOMEM;
  1012. }