sched.c 27 KB

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