sched.c 28 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120
  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. BUG_ON (RPC_IS_QUEUED(task));
  119. if (RPC_IS_PRIORITY(queue))
  120. __rpc_add_wait_queue_priority(queue, task, queue_priority);
  121. else if (RPC_IS_SWAPPER(task))
  122. list_add(&task->u.tk_wait.list, &queue->tasks[0]);
  123. else
  124. list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
  125. task->tk_waitqueue = queue;
  126. queue->qlen++;
  127. /* barrier matches the read in rpc_wake_up_task_queue_locked() */
  128. smp_wmb();
  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_unsafe();
  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. * Note the ordering of rpc_test_and_set_running() and rpc_clear_queued(),
  268. * which is needed to ensure that __rpc_execute() doesn't loop (due to the
  269. * lockless RPC_IS_QUEUED() test) before we've had a chance to test
  270. * the RPC_TASK_RUNNING flag.
  271. */
  272. static void rpc_make_runnable(struct rpc_task *task)
  273. {
  274. bool need_wakeup = !rpc_test_and_set_running(task);
  275. rpc_clear_queued(task);
  276. if (!need_wakeup)
  277. return;
  278. if (RPC_IS_ASYNC(task)) {
  279. INIT_WORK(&task->u.tk_work, rpc_async_schedule);
  280. queue_work(rpciod_workqueue, &task->u.tk_work);
  281. } else
  282. wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
  283. }
  284. /*
  285. * Prepare for sleeping on a wait queue.
  286. * By always appending tasks to the list we ensure FIFO behavior.
  287. * NB: An RPC task will only receive interrupt-driven events as long
  288. * as it's on a wait queue.
  289. */
  290. static void __rpc_sleep_on_priority(struct rpc_wait_queue *q,
  291. struct rpc_task *task,
  292. rpc_action action,
  293. unsigned char queue_priority)
  294. {
  295. dprintk("RPC: %5u sleep_on(queue \"%s\" time %lu)\n",
  296. task->tk_pid, rpc_qname(q), jiffies);
  297. trace_rpc_task_sleep(task->tk_client, task, q);
  298. __rpc_add_wait_queue(q, task, queue_priority);
  299. BUG_ON(task->tk_callback != NULL);
  300. task->tk_callback = action;
  301. __rpc_add_timer(q, task);
  302. }
  303. void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
  304. rpc_action action)
  305. {
  306. /* We shouldn't ever put an inactive task to sleep */
  307. BUG_ON(!RPC_IS_ACTIVATED(task));
  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. BUG_ON(!RPC_IS_ACTIVATED(task));
  321. /*
  322. * Protect the queue operations.
  323. */
  324. spin_lock_bh(&q->lock);
  325. __rpc_sleep_on_priority(q, task, action, priority - RPC_PRIORITY_LOW);
  326. spin_unlock_bh(&q->lock);
  327. }
  328. /**
  329. * __rpc_do_wake_up_task - wake up a single rpc_task
  330. * @queue: wait queue
  331. * @task: task to be woken up
  332. *
  333. * Caller must hold queue->lock, and have cleared the task queued flag.
  334. */
  335. static void __rpc_do_wake_up_task(struct rpc_wait_queue *queue, struct rpc_task *task)
  336. {
  337. dprintk("RPC: %5u __rpc_wake_up_task (now %lu)\n",
  338. task->tk_pid, jiffies);
  339. /* Has the task been executed yet? If not, we cannot wake it up! */
  340. if (!RPC_IS_ACTIVATED(task)) {
  341. printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
  342. return;
  343. }
  344. trace_rpc_task_wakeup(task->tk_client, task, queue);
  345. __rpc_remove_wait_queue(queue, task);
  346. rpc_make_runnable(task);
  347. dprintk("RPC: __rpc_wake_up_task done\n");
  348. }
  349. /*
  350. * Wake up a queued task while the queue lock is being held
  351. */
  352. static void rpc_wake_up_task_queue_locked(struct rpc_wait_queue *queue, struct rpc_task *task)
  353. {
  354. if (RPC_IS_QUEUED(task)) {
  355. smp_rmb();
  356. if (task->tk_waitqueue == queue)
  357. __rpc_do_wake_up_task(queue, task);
  358. }
  359. }
  360. /*
  361. * Tests whether rpc queue is empty
  362. */
  363. int rpc_queue_empty(struct rpc_wait_queue *queue)
  364. {
  365. int res;
  366. spin_lock_bh(&queue->lock);
  367. res = queue->qlen;
  368. spin_unlock_bh(&queue->lock);
  369. return res == 0;
  370. }
  371. EXPORT_SYMBOL_GPL(rpc_queue_empty);
  372. /*
  373. * Wake up a task on a specific queue
  374. */
  375. void rpc_wake_up_queued_task(struct rpc_wait_queue *queue, struct rpc_task *task)
  376. {
  377. spin_lock_bh(&queue->lock);
  378. rpc_wake_up_task_queue_locked(queue, task);
  379. spin_unlock_bh(&queue->lock);
  380. }
  381. EXPORT_SYMBOL_GPL(rpc_wake_up_queued_task);
  382. /*
  383. * Wake up the next task on a priority queue.
  384. */
  385. static struct rpc_task *__rpc_find_next_queued_priority(struct rpc_wait_queue *queue)
  386. {
  387. struct list_head *q;
  388. struct rpc_task *task;
  389. /*
  390. * Service a batch of tasks from a single owner.
  391. */
  392. q = &queue->tasks[queue->priority];
  393. if (!list_empty(q)) {
  394. task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
  395. if (queue->owner == task->tk_owner) {
  396. if (--queue->nr)
  397. goto out;
  398. list_move_tail(&task->u.tk_wait.list, q);
  399. }
  400. /*
  401. * Check if we need to switch queues.
  402. */
  403. if (--queue->count)
  404. goto new_owner;
  405. }
  406. /*
  407. * Service the next queue.
  408. */
  409. do {
  410. if (q == &queue->tasks[0])
  411. q = &queue->tasks[queue->maxpriority];
  412. else
  413. q = q - 1;
  414. if (!list_empty(q)) {
  415. task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
  416. goto new_queue;
  417. }
  418. } while (q != &queue->tasks[queue->priority]);
  419. rpc_reset_waitqueue_priority(queue);
  420. return NULL;
  421. new_queue:
  422. rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0]));
  423. new_owner:
  424. rpc_set_waitqueue_owner(queue, task->tk_owner);
  425. out:
  426. return task;
  427. }
  428. static struct rpc_task *__rpc_find_next_queued(struct rpc_wait_queue *queue)
  429. {
  430. if (RPC_IS_PRIORITY(queue))
  431. return __rpc_find_next_queued_priority(queue);
  432. if (!list_empty(&queue->tasks[0]))
  433. return list_first_entry(&queue->tasks[0], struct rpc_task, u.tk_wait.list);
  434. return NULL;
  435. }
  436. /*
  437. * Wake up the first task on the wait queue.
  438. */
  439. struct rpc_task *rpc_wake_up_first(struct rpc_wait_queue *queue,
  440. bool (*func)(struct rpc_task *, void *), void *data)
  441. {
  442. struct rpc_task *task = NULL;
  443. dprintk("RPC: wake_up_first(%p \"%s\")\n",
  444. queue, rpc_qname(queue));
  445. spin_lock_bh(&queue->lock);
  446. task = __rpc_find_next_queued(queue);
  447. if (task != NULL) {
  448. if (func(task, data))
  449. rpc_wake_up_task_queue_locked(queue, task);
  450. else
  451. task = NULL;
  452. }
  453. spin_unlock_bh(&queue->lock);
  454. return task;
  455. }
  456. EXPORT_SYMBOL_GPL(rpc_wake_up_first);
  457. static bool rpc_wake_up_next_func(struct rpc_task *task, void *data)
  458. {
  459. return true;
  460. }
  461. /*
  462. * Wake up the next task on the wait queue.
  463. */
  464. struct rpc_task *rpc_wake_up_next(struct rpc_wait_queue *queue)
  465. {
  466. return rpc_wake_up_first(queue, rpc_wake_up_next_func, NULL);
  467. }
  468. EXPORT_SYMBOL_GPL(rpc_wake_up_next);
  469. /**
  470. * rpc_wake_up - wake up all rpc_tasks
  471. * @queue: rpc_wait_queue on which the tasks are sleeping
  472. *
  473. * Grabs queue->lock
  474. */
  475. void rpc_wake_up(struct rpc_wait_queue *queue)
  476. {
  477. struct list_head *head;
  478. spin_lock_bh(&queue->lock);
  479. head = &queue->tasks[queue->maxpriority];
  480. for (;;) {
  481. while (!list_empty(head)) {
  482. struct rpc_task *task;
  483. task = list_first_entry(head,
  484. struct rpc_task,
  485. u.tk_wait.list);
  486. rpc_wake_up_task_queue_locked(queue, task);
  487. }
  488. if (head == &queue->tasks[0])
  489. break;
  490. head--;
  491. }
  492. spin_unlock_bh(&queue->lock);
  493. }
  494. EXPORT_SYMBOL_GPL(rpc_wake_up);
  495. /**
  496. * rpc_wake_up_status - wake up all rpc_tasks and set their status value.
  497. * @queue: rpc_wait_queue on which the tasks are sleeping
  498. * @status: status value to set
  499. *
  500. * Grabs queue->lock
  501. */
  502. void rpc_wake_up_status(struct rpc_wait_queue *queue, int status)
  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. task->tk_status = status;
  514. rpc_wake_up_task_queue_locked(queue, task);
  515. }
  516. if (head == &queue->tasks[0])
  517. break;
  518. head--;
  519. }
  520. spin_unlock_bh(&queue->lock);
  521. }
  522. EXPORT_SYMBOL_GPL(rpc_wake_up_status);
  523. static void __rpc_queue_timer_fn(unsigned long ptr)
  524. {
  525. struct rpc_wait_queue *queue = (struct rpc_wait_queue *)ptr;
  526. struct rpc_task *task, *n;
  527. unsigned long expires, now, timeo;
  528. spin_lock(&queue->lock);
  529. expires = now = jiffies;
  530. list_for_each_entry_safe(task, n, &queue->timer_list.list, u.tk_wait.timer_list) {
  531. timeo = task->u.tk_wait.expires;
  532. if (time_after_eq(now, timeo)) {
  533. dprintk("RPC: %5u timeout\n", task->tk_pid);
  534. task->tk_status = -ETIMEDOUT;
  535. rpc_wake_up_task_queue_locked(queue, task);
  536. continue;
  537. }
  538. if (expires == now || time_after(expires, timeo))
  539. expires = timeo;
  540. }
  541. if (!list_empty(&queue->timer_list.list))
  542. rpc_set_queue_timer(queue, expires);
  543. spin_unlock(&queue->lock);
  544. }
  545. static void __rpc_atrun(struct rpc_task *task)
  546. {
  547. task->tk_status = 0;
  548. }
  549. /*
  550. * Run a task at a later time
  551. */
  552. void rpc_delay(struct rpc_task *task, unsigned long delay)
  553. {
  554. task->tk_timeout = delay;
  555. rpc_sleep_on(&delay_queue, task, __rpc_atrun);
  556. }
  557. EXPORT_SYMBOL_GPL(rpc_delay);
  558. /*
  559. * Helper to call task->tk_ops->rpc_call_prepare
  560. */
  561. void rpc_prepare_task(struct rpc_task *task)
  562. {
  563. task->tk_ops->rpc_call_prepare(task, task->tk_calldata);
  564. }
  565. static void
  566. rpc_init_task_statistics(struct rpc_task *task)
  567. {
  568. /* Initialize retry counters */
  569. task->tk_garb_retry = 2;
  570. task->tk_cred_retry = 2;
  571. task->tk_rebind_retry = 2;
  572. /* starting timestamp */
  573. task->tk_start = ktime_get();
  574. }
  575. static void
  576. rpc_reset_task_statistics(struct rpc_task *task)
  577. {
  578. task->tk_timeouts = 0;
  579. task->tk_flags &= ~(RPC_CALL_MAJORSEEN|RPC_TASK_KILLED|RPC_TASK_SENT);
  580. rpc_init_task_statistics(task);
  581. }
  582. /*
  583. * Helper that calls task->tk_ops->rpc_call_done if it exists
  584. */
  585. void rpc_exit_task(struct rpc_task *task)
  586. {
  587. task->tk_action = NULL;
  588. if (task->tk_ops->rpc_call_done != NULL) {
  589. task->tk_ops->rpc_call_done(task, task->tk_calldata);
  590. if (task->tk_action != NULL) {
  591. WARN_ON(RPC_ASSASSINATED(task));
  592. /* Always release the RPC slot and buffer memory */
  593. xprt_release(task);
  594. rpc_reset_task_statistics(task);
  595. }
  596. }
  597. }
  598. void rpc_exit(struct rpc_task *task, int status)
  599. {
  600. task->tk_status = status;
  601. task->tk_action = rpc_exit_task;
  602. if (RPC_IS_QUEUED(task))
  603. rpc_wake_up_queued_task(task->tk_waitqueue, task);
  604. }
  605. EXPORT_SYMBOL_GPL(rpc_exit);
  606. void rpc_release_calldata(const struct rpc_call_ops *ops, void *calldata)
  607. {
  608. if (ops->rpc_release != NULL)
  609. ops->rpc_release(calldata);
  610. }
  611. /*
  612. * This is the RPC `scheduler' (or rather, the finite state machine).
  613. */
  614. static void __rpc_execute(struct rpc_task *task)
  615. {
  616. struct rpc_wait_queue *queue;
  617. int task_is_async = RPC_IS_ASYNC(task);
  618. int status = 0;
  619. dprintk("RPC: %5u __rpc_execute flags=0x%x\n",
  620. task->tk_pid, task->tk_flags);
  621. BUG_ON(RPC_IS_QUEUED(task));
  622. for (;;) {
  623. void (*do_action)(struct rpc_task *);
  624. /*
  625. * Execute any pending callback first.
  626. */
  627. do_action = task->tk_callback;
  628. task->tk_callback = NULL;
  629. if (do_action == NULL) {
  630. /*
  631. * Perform the next FSM step.
  632. * tk_action may be NULL if the task has been killed.
  633. * In particular, note that rpc_killall_tasks may
  634. * do this at any time, so beware when dereferencing.
  635. */
  636. do_action = task->tk_action;
  637. if (do_action == NULL)
  638. break;
  639. }
  640. trace_rpc_task_run_action(task->tk_client, task, task->tk_action);
  641. do_action(task);
  642. /*
  643. * Lockless check for whether task is sleeping or not.
  644. */
  645. if (!RPC_IS_QUEUED(task))
  646. continue;
  647. /*
  648. * The queue->lock protects against races with
  649. * rpc_make_runnable().
  650. *
  651. * Note that once we clear RPC_TASK_RUNNING on an asynchronous
  652. * rpc_task, rpc_make_runnable() can assign it to a
  653. * different workqueue. We therefore cannot assume that the
  654. * rpc_task pointer may still be dereferenced.
  655. */
  656. queue = task->tk_waitqueue;
  657. spin_lock_bh(&queue->lock);
  658. if (!RPC_IS_QUEUED(task)) {
  659. spin_unlock_bh(&queue->lock);
  660. continue;
  661. }
  662. rpc_clear_running(task);
  663. spin_unlock_bh(&queue->lock);
  664. if (task_is_async)
  665. return;
  666. /* sync task: sleep here */
  667. dprintk("RPC: %5u sync task going to sleep\n", task->tk_pid);
  668. status = out_of_line_wait_on_bit(&task->tk_runstate,
  669. RPC_TASK_QUEUED, rpc_wait_bit_killable,
  670. TASK_KILLABLE);
  671. if (status == -ERESTARTSYS) {
  672. /*
  673. * When a sync task receives a signal, it exits with
  674. * -ERESTARTSYS. In order to catch any callbacks that
  675. * clean up after sleeping on some queue, we don't
  676. * break the loop here, but go around once more.
  677. */
  678. dprintk("RPC: %5u got signal\n", task->tk_pid);
  679. task->tk_flags |= RPC_TASK_KILLED;
  680. rpc_exit(task, -ERESTARTSYS);
  681. }
  682. rpc_set_running(task);
  683. dprintk("RPC: %5u sync task resuming\n", task->tk_pid);
  684. }
  685. dprintk("RPC: %5u return %d, status %d\n", task->tk_pid, status,
  686. task->tk_status);
  687. /* Release all resources associated with the task */
  688. rpc_release_task(task);
  689. }
  690. /*
  691. * User-visible entry point to the scheduler.
  692. *
  693. * This may be called recursively if e.g. an async NFS task updates
  694. * the attributes and finds that dirty pages must be flushed.
  695. * NOTE: Upon exit of this function the task is guaranteed to be
  696. * released. In particular note that tk_release() will have
  697. * been called, so your task memory may have been freed.
  698. */
  699. void rpc_execute(struct rpc_task *task)
  700. {
  701. rpc_set_active(task);
  702. rpc_make_runnable(task);
  703. if (!RPC_IS_ASYNC(task))
  704. __rpc_execute(task);
  705. }
  706. static void rpc_async_schedule(struct work_struct *work)
  707. {
  708. current->flags |= PF_FSTRANS;
  709. __rpc_execute(container_of(work, struct rpc_task, u.tk_work));
  710. current->flags &= ~PF_FSTRANS;
  711. }
  712. /**
  713. * rpc_malloc - allocate an RPC buffer
  714. * @task: RPC task that will use this buffer
  715. * @size: requested byte size
  716. *
  717. * To prevent rpciod from hanging, this allocator never sleeps,
  718. * returning NULL if the request cannot be serviced immediately.
  719. * The caller can arrange to sleep in a way that is safe for rpciod.
  720. *
  721. * Most requests are 'small' (under 2KiB) and can be serviced from a
  722. * mempool, ensuring that NFS reads and writes can always proceed,
  723. * and that there is good locality of reference for these buffers.
  724. *
  725. * In order to avoid memory starvation triggering more writebacks of
  726. * NFS requests, we avoid using GFP_KERNEL.
  727. */
  728. void *rpc_malloc(struct rpc_task *task, size_t size)
  729. {
  730. struct rpc_buffer *buf;
  731. gfp_t gfp = RPC_IS_SWAPPER(task) ? GFP_ATOMIC : GFP_NOWAIT;
  732. size += sizeof(struct rpc_buffer);
  733. if (size <= RPC_BUFFER_MAXSIZE)
  734. buf = mempool_alloc(rpc_buffer_mempool, gfp);
  735. else
  736. buf = kmalloc(size, gfp);
  737. if (!buf)
  738. return NULL;
  739. buf->len = size;
  740. dprintk("RPC: %5u allocated buffer of size %zu at %p\n",
  741. task->tk_pid, size, buf);
  742. return &buf->data;
  743. }
  744. EXPORT_SYMBOL_GPL(rpc_malloc);
  745. /**
  746. * rpc_free - free buffer allocated via rpc_malloc
  747. * @buffer: buffer to free
  748. *
  749. */
  750. void rpc_free(void *buffer)
  751. {
  752. size_t size;
  753. struct rpc_buffer *buf;
  754. if (!buffer)
  755. return;
  756. buf = container_of(buffer, struct rpc_buffer, data);
  757. size = buf->len;
  758. dprintk("RPC: freeing buffer of size %zu at %p\n",
  759. size, buf);
  760. if (size <= RPC_BUFFER_MAXSIZE)
  761. mempool_free(buf, rpc_buffer_mempool);
  762. else
  763. kfree(buf);
  764. }
  765. EXPORT_SYMBOL_GPL(rpc_free);
  766. /*
  767. * Creation and deletion of RPC task structures
  768. */
  769. static void rpc_init_task(struct rpc_task *task, const struct rpc_task_setup *task_setup_data)
  770. {
  771. memset(task, 0, sizeof(*task));
  772. atomic_set(&task->tk_count, 1);
  773. task->tk_flags = task_setup_data->flags;
  774. task->tk_ops = task_setup_data->callback_ops;
  775. task->tk_calldata = task_setup_data->callback_data;
  776. INIT_LIST_HEAD(&task->tk_task);
  777. task->tk_priority = task_setup_data->priority - RPC_PRIORITY_LOW;
  778. task->tk_owner = current->tgid;
  779. /* Initialize workqueue for async tasks */
  780. task->tk_workqueue = task_setup_data->workqueue;
  781. if (task->tk_ops->rpc_call_prepare != NULL)
  782. task->tk_action = rpc_prepare_task;
  783. rpc_init_task_statistics(task);
  784. dprintk("RPC: new task initialized, procpid %u\n",
  785. task_pid_nr(current));
  786. }
  787. static struct rpc_task *
  788. rpc_alloc_task(void)
  789. {
  790. return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOFS);
  791. }
  792. /*
  793. * Create a new task for the specified client.
  794. */
  795. struct rpc_task *rpc_new_task(const struct rpc_task_setup *setup_data)
  796. {
  797. struct rpc_task *task = setup_data->task;
  798. unsigned short flags = 0;
  799. if (task == NULL) {
  800. task = rpc_alloc_task();
  801. if (task == NULL) {
  802. rpc_release_calldata(setup_data->callback_ops,
  803. setup_data->callback_data);
  804. return ERR_PTR(-ENOMEM);
  805. }
  806. flags = RPC_TASK_DYNAMIC;
  807. }
  808. rpc_init_task(task, setup_data);
  809. task->tk_flags |= flags;
  810. dprintk("RPC: allocated task %p\n", task);
  811. return task;
  812. }
  813. /*
  814. * rpc_free_task - release rpc task and perform cleanups
  815. *
  816. * Note that we free up the rpc_task _after_ rpc_release_calldata()
  817. * in order to work around a workqueue dependency issue.
  818. *
  819. * Tejun Heo states:
  820. * "Workqueue currently considers two work items to be the same if they're
  821. * on the same address and won't execute them concurrently - ie. it
  822. * makes a work item which is queued again while being executed wait
  823. * for the previous execution to complete.
  824. *
  825. * If a work function frees the work item, and then waits for an event
  826. * which should be performed by another work item and *that* work item
  827. * recycles the freed work item, it can create a false dependency loop.
  828. * There really is no reliable way to detect this short of verifying
  829. * every memory free."
  830. *
  831. */
  832. static void rpc_free_task(struct rpc_task *task)
  833. {
  834. unsigned short tk_flags = task->tk_flags;
  835. rpc_release_calldata(task->tk_ops, task->tk_calldata);
  836. if (tk_flags & RPC_TASK_DYNAMIC) {
  837. dprintk("RPC: %5u freeing task\n", task->tk_pid);
  838. mempool_free(task, rpc_task_mempool);
  839. }
  840. }
  841. static void rpc_async_release(struct work_struct *work)
  842. {
  843. rpc_free_task(container_of(work, struct rpc_task, u.tk_work));
  844. }
  845. static void rpc_release_resources_task(struct rpc_task *task)
  846. {
  847. xprt_release(task);
  848. if (task->tk_msg.rpc_cred) {
  849. put_rpccred(task->tk_msg.rpc_cred);
  850. task->tk_msg.rpc_cred = NULL;
  851. }
  852. rpc_task_release_client(task);
  853. }
  854. static void rpc_final_put_task(struct rpc_task *task,
  855. struct workqueue_struct *q)
  856. {
  857. if (q != NULL) {
  858. INIT_WORK(&task->u.tk_work, rpc_async_release);
  859. queue_work(q, &task->u.tk_work);
  860. } else
  861. rpc_free_task(task);
  862. }
  863. static void rpc_do_put_task(struct rpc_task *task, struct workqueue_struct *q)
  864. {
  865. if (atomic_dec_and_test(&task->tk_count)) {
  866. rpc_release_resources_task(task);
  867. rpc_final_put_task(task, q);
  868. }
  869. }
  870. void rpc_put_task(struct rpc_task *task)
  871. {
  872. rpc_do_put_task(task, NULL);
  873. }
  874. EXPORT_SYMBOL_GPL(rpc_put_task);
  875. void rpc_put_task_async(struct rpc_task *task)
  876. {
  877. rpc_do_put_task(task, task->tk_workqueue);
  878. }
  879. EXPORT_SYMBOL_GPL(rpc_put_task_async);
  880. static void rpc_release_task(struct rpc_task *task)
  881. {
  882. dprintk("RPC: %5u release task\n", task->tk_pid);
  883. BUG_ON (RPC_IS_QUEUED(task));
  884. rpc_release_resources_task(task);
  885. /*
  886. * Note: at this point we have been removed from rpc_clnt->cl_tasks,
  887. * so it should be safe to use task->tk_count as a test for whether
  888. * or not any other processes still hold references to our rpc_task.
  889. */
  890. if (atomic_read(&task->tk_count) != 1 + !RPC_IS_ASYNC(task)) {
  891. /* Wake up anyone who may be waiting for task completion */
  892. if (!rpc_complete_task(task))
  893. return;
  894. } else {
  895. if (!atomic_dec_and_test(&task->tk_count))
  896. return;
  897. }
  898. rpc_final_put_task(task, task->tk_workqueue);
  899. }
  900. int rpciod_up(void)
  901. {
  902. return try_module_get(THIS_MODULE) ? 0 : -EINVAL;
  903. }
  904. void rpciod_down(void)
  905. {
  906. module_put(THIS_MODULE);
  907. }
  908. /*
  909. * Start up the rpciod workqueue.
  910. */
  911. static int rpciod_start(void)
  912. {
  913. struct workqueue_struct *wq;
  914. /*
  915. * Create the rpciod thread and wait for it to start.
  916. */
  917. dprintk("RPC: creating workqueue rpciod\n");
  918. wq = alloc_workqueue("rpciod", WQ_MEM_RECLAIM, 0);
  919. rpciod_workqueue = wq;
  920. return rpciod_workqueue != NULL;
  921. }
  922. static void rpciod_stop(void)
  923. {
  924. struct workqueue_struct *wq = NULL;
  925. if (rpciod_workqueue == NULL)
  926. return;
  927. dprintk("RPC: destroying workqueue rpciod\n");
  928. wq = rpciod_workqueue;
  929. rpciod_workqueue = NULL;
  930. destroy_workqueue(wq);
  931. }
  932. void
  933. rpc_destroy_mempool(void)
  934. {
  935. rpciod_stop();
  936. if (rpc_buffer_mempool)
  937. mempool_destroy(rpc_buffer_mempool);
  938. if (rpc_task_mempool)
  939. mempool_destroy(rpc_task_mempool);
  940. if (rpc_task_slabp)
  941. kmem_cache_destroy(rpc_task_slabp);
  942. if (rpc_buffer_slabp)
  943. kmem_cache_destroy(rpc_buffer_slabp);
  944. rpc_destroy_wait_queue(&delay_queue);
  945. }
  946. int
  947. rpc_init_mempool(void)
  948. {
  949. /*
  950. * The following is not strictly a mempool initialisation,
  951. * but there is no harm in doing it here
  952. */
  953. rpc_init_wait_queue(&delay_queue, "delayq");
  954. if (!rpciod_start())
  955. goto err_nomem;
  956. rpc_task_slabp = kmem_cache_create("rpc_tasks",
  957. sizeof(struct rpc_task),
  958. 0, SLAB_HWCACHE_ALIGN,
  959. NULL);
  960. if (!rpc_task_slabp)
  961. goto err_nomem;
  962. rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
  963. RPC_BUFFER_MAXSIZE,
  964. 0, SLAB_HWCACHE_ALIGN,
  965. NULL);
  966. if (!rpc_buffer_slabp)
  967. goto err_nomem;
  968. rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
  969. rpc_task_slabp);
  970. if (!rpc_task_mempool)
  971. goto err_nomem;
  972. rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
  973. rpc_buffer_slabp);
  974. if (!rpc_buffer_mempool)
  975. goto err_nomem;
  976. return 0;
  977. err_nomem:
  978. rpc_destroy_mempool();
  979. return -ENOMEM;
  980. }