sched.c 28 KB

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