sched.c 25 KB

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