sched.c 27 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103
  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. rpc_set_queued(task);
  128. dprintk("RPC: %5u added to queue %p \"%s\"\n",
  129. task->tk_pid, queue, rpc_qname(queue));
  130. }
  131. /*
  132. * Remove request from a priority queue.
  133. */
  134. static void __rpc_remove_wait_queue_priority(struct rpc_task *task)
  135. {
  136. struct rpc_task *t;
  137. if (!list_empty(&task->u.tk_wait.links)) {
  138. t = list_entry(task->u.tk_wait.links.next, struct rpc_task, u.tk_wait.list);
  139. list_move(&t->u.tk_wait.list, &task->u.tk_wait.list);
  140. list_splice_init(&task->u.tk_wait.links, &t->u.tk_wait.links);
  141. }
  142. }
  143. /*
  144. * Remove request from queue.
  145. * Note: must be called with spin lock held.
  146. */
  147. static void __rpc_remove_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
  148. {
  149. __rpc_disable_timer(queue, task);
  150. if (RPC_IS_PRIORITY(queue))
  151. __rpc_remove_wait_queue_priority(task);
  152. list_del(&task->u.tk_wait.list);
  153. queue->qlen--;
  154. dprintk("RPC: %5u removed from queue %p \"%s\"\n",
  155. task->tk_pid, queue, rpc_qname(queue));
  156. }
  157. static inline void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority)
  158. {
  159. queue->priority = priority;
  160. queue->count = 1 << (priority * 2);
  161. }
  162. static inline void rpc_set_waitqueue_owner(struct rpc_wait_queue *queue, pid_t pid)
  163. {
  164. queue->owner = pid;
  165. queue->nr = RPC_BATCH_COUNT;
  166. }
  167. static inline void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
  168. {
  169. rpc_set_waitqueue_priority(queue, queue->maxpriority);
  170. rpc_set_waitqueue_owner(queue, 0);
  171. }
  172. static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, unsigned char nr_queues)
  173. {
  174. int i;
  175. spin_lock_init(&queue->lock);
  176. for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
  177. INIT_LIST_HEAD(&queue->tasks[i]);
  178. queue->maxpriority = nr_queues - 1;
  179. rpc_reset_waitqueue_priority(queue);
  180. queue->qlen = 0;
  181. setup_timer(&queue->timer_list.timer, __rpc_queue_timer_fn, (unsigned long)queue);
  182. INIT_LIST_HEAD(&queue->timer_list.list);
  183. rpc_assign_waitqueue_name(queue, qname);
  184. }
  185. void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
  186. {
  187. __rpc_init_priority_wait_queue(queue, qname, RPC_NR_PRIORITY);
  188. }
  189. EXPORT_SYMBOL_GPL(rpc_init_priority_wait_queue);
  190. void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
  191. {
  192. __rpc_init_priority_wait_queue(queue, qname, 1);
  193. }
  194. EXPORT_SYMBOL_GPL(rpc_init_wait_queue);
  195. void rpc_destroy_wait_queue(struct rpc_wait_queue *queue)
  196. {
  197. del_timer_sync(&queue->timer_list.timer);
  198. }
  199. EXPORT_SYMBOL_GPL(rpc_destroy_wait_queue);
  200. static int rpc_wait_bit_killable(void *word)
  201. {
  202. if (fatal_signal_pending(current))
  203. return -ERESTARTSYS;
  204. freezable_schedule();
  205. return 0;
  206. }
  207. #ifdef RPC_DEBUG
  208. static void rpc_task_set_debuginfo(struct rpc_task *task)
  209. {
  210. static atomic_t rpc_pid;
  211. task->tk_pid = atomic_inc_return(&rpc_pid);
  212. }
  213. #else
  214. static inline void rpc_task_set_debuginfo(struct rpc_task *task)
  215. {
  216. }
  217. #endif
  218. static void rpc_set_active(struct rpc_task *task)
  219. {
  220. trace_rpc_task_begin(task->tk_client, task, NULL);
  221. rpc_task_set_debuginfo(task);
  222. set_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
  223. }
  224. /*
  225. * Mark an RPC call as having completed by clearing the 'active' bit
  226. * and then waking up all tasks that were sleeping.
  227. */
  228. static int rpc_complete_task(struct rpc_task *task)
  229. {
  230. void *m = &task->tk_runstate;
  231. wait_queue_head_t *wq = bit_waitqueue(m, RPC_TASK_ACTIVE);
  232. struct wait_bit_key k = __WAIT_BIT_KEY_INITIALIZER(m, RPC_TASK_ACTIVE);
  233. unsigned long flags;
  234. int ret;
  235. trace_rpc_task_complete(task->tk_client, task, NULL);
  236. spin_lock_irqsave(&wq->lock, flags);
  237. clear_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
  238. ret = atomic_dec_and_test(&task->tk_count);
  239. if (waitqueue_active(wq))
  240. __wake_up_locked_key(wq, TASK_NORMAL, &k);
  241. spin_unlock_irqrestore(&wq->lock, flags);
  242. return ret;
  243. }
  244. /*
  245. * Allow callers to wait for completion of an RPC call
  246. *
  247. * Note the use of out_of_line_wait_on_bit() rather than wait_on_bit()
  248. * to enforce taking of the wq->lock and hence avoid races with
  249. * rpc_complete_task().
  250. */
  251. int __rpc_wait_for_completion_task(struct rpc_task *task, int (*action)(void *))
  252. {
  253. if (action == NULL)
  254. action = rpc_wait_bit_killable;
  255. return out_of_line_wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
  256. action, TASK_KILLABLE);
  257. }
  258. EXPORT_SYMBOL_GPL(__rpc_wait_for_completion_task);
  259. /*
  260. * Make an RPC task runnable.
  261. *
  262. * Note: If the task is ASYNC, and is being made runnable after sitting on an
  263. * rpc_wait_queue, this must be called with the queue spinlock held to protect
  264. * the wait queue operation.
  265. */
  266. static void rpc_make_runnable(struct rpc_task *task)
  267. {
  268. rpc_clear_queued(task);
  269. if (rpc_test_and_set_running(task))
  270. return;
  271. if (RPC_IS_ASYNC(task)) {
  272. INIT_WORK(&task->u.tk_work, rpc_async_schedule);
  273. queue_work(rpciod_workqueue, &task->u.tk_work);
  274. } else
  275. wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
  276. }
  277. /*
  278. * Prepare for sleeping on a wait queue.
  279. * By always appending tasks to the list we ensure FIFO behavior.
  280. * NB: An RPC task will only receive interrupt-driven events as long
  281. * as it's on a wait queue.
  282. */
  283. static void __rpc_sleep_on_priority(struct rpc_wait_queue *q,
  284. struct rpc_task *task,
  285. rpc_action action,
  286. unsigned char queue_priority)
  287. {
  288. dprintk("RPC: %5u sleep_on(queue \"%s\" time %lu)\n",
  289. task->tk_pid, rpc_qname(q), jiffies);
  290. trace_rpc_task_sleep(task->tk_client, task, q);
  291. __rpc_add_wait_queue(q, task, queue_priority);
  292. BUG_ON(task->tk_callback != NULL);
  293. task->tk_callback = action;
  294. __rpc_add_timer(q, task);
  295. }
  296. void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
  297. rpc_action action)
  298. {
  299. /* We shouldn't ever put an inactive task to sleep */
  300. WARN_ON_ONCE(!RPC_IS_ACTIVATED(task));
  301. if (!RPC_IS_ACTIVATED(task)) {
  302. task->tk_status = -EIO;
  303. rpc_put_task_async(task);
  304. return;
  305. }
  306. /*
  307. * Protect the queue operations.
  308. */
  309. spin_lock_bh(&q->lock);
  310. __rpc_sleep_on_priority(q, task, action, task->tk_priority);
  311. spin_unlock_bh(&q->lock);
  312. }
  313. EXPORT_SYMBOL_GPL(rpc_sleep_on);
  314. void rpc_sleep_on_priority(struct rpc_wait_queue *q, struct rpc_task *task,
  315. rpc_action action, int priority)
  316. {
  317. /* We shouldn't ever put an inactive task to sleep */
  318. WARN_ON_ONCE(!RPC_IS_ACTIVATED(task));
  319. if (!RPC_IS_ACTIVATED(task)) {
  320. task->tk_status = -EIO;
  321. rpc_put_task_async(task);
  322. return;
  323. }
  324. /*
  325. * Protect the queue operations.
  326. */
  327. spin_lock_bh(&q->lock);
  328. __rpc_sleep_on_priority(q, task, action, priority - RPC_PRIORITY_LOW);
  329. spin_unlock_bh(&q->lock);
  330. }
  331. /**
  332. * __rpc_do_wake_up_task - wake up a single rpc_task
  333. * @queue: wait queue
  334. * @task: task to be woken up
  335. *
  336. * Caller must hold queue->lock, and have cleared the task queued flag.
  337. */
  338. static void __rpc_do_wake_up_task(struct rpc_wait_queue *queue, struct rpc_task *task)
  339. {
  340. dprintk("RPC: %5u __rpc_wake_up_task (now %lu)\n",
  341. task->tk_pid, jiffies);
  342. /* Has the task been executed yet? If not, we cannot wake it up! */
  343. if (!RPC_IS_ACTIVATED(task)) {
  344. printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
  345. return;
  346. }
  347. trace_rpc_task_wakeup(task->tk_client, task, queue);
  348. __rpc_remove_wait_queue(queue, task);
  349. rpc_make_runnable(task);
  350. dprintk("RPC: __rpc_wake_up_task done\n");
  351. }
  352. /*
  353. * Wake up a queued task while the queue lock is being held
  354. */
  355. static void rpc_wake_up_task_queue_locked(struct rpc_wait_queue *queue, struct rpc_task *task)
  356. {
  357. if (RPC_IS_QUEUED(task) && task->tk_waitqueue == queue)
  358. __rpc_do_wake_up_task(queue, task);
  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 = GFP_NOWAIT;
  732. if (RPC_IS_SWAPPER(task))
  733. gfp |= __GFP_MEMALLOC;
  734. size += sizeof(struct rpc_buffer);
  735. if (size <= RPC_BUFFER_MAXSIZE)
  736. buf = mempool_alloc(rpc_buffer_mempool, gfp);
  737. else
  738. buf = kmalloc(size, gfp);
  739. if (!buf)
  740. return NULL;
  741. buf->len = size;
  742. dprintk("RPC: %5u allocated buffer of size %zu at %p\n",
  743. task->tk_pid, size, buf);
  744. return &buf->data;
  745. }
  746. EXPORT_SYMBOL_GPL(rpc_malloc);
  747. /**
  748. * rpc_free - free buffer allocated via rpc_malloc
  749. * @buffer: buffer to free
  750. *
  751. */
  752. void rpc_free(void *buffer)
  753. {
  754. size_t size;
  755. struct rpc_buffer *buf;
  756. if (!buffer)
  757. return;
  758. buf = container_of(buffer, struct rpc_buffer, data);
  759. size = buf->len;
  760. dprintk("RPC: freeing buffer of size %zu at %p\n",
  761. size, buf);
  762. if (size <= RPC_BUFFER_MAXSIZE)
  763. mempool_free(buf, rpc_buffer_mempool);
  764. else
  765. kfree(buf);
  766. }
  767. EXPORT_SYMBOL_GPL(rpc_free);
  768. /*
  769. * Creation and deletion of RPC task structures
  770. */
  771. static void rpc_init_task(struct rpc_task *task, const struct rpc_task_setup *task_setup_data)
  772. {
  773. memset(task, 0, sizeof(*task));
  774. atomic_set(&task->tk_count, 1);
  775. task->tk_flags = task_setup_data->flags;
  776. task->tk_ops = task_setup_data->callback_ops;
  777. task->tk_calldata = task_setup_data->callback_data;
  778. INIT_LIST_HEAD(&task->tk_task);
  779. task->tk_priority = task_setup_data->priority - RPC_PRIORITY_LOW;
  780. task->tk_owner = current->tgid;
  781. /* Initialize workqueue for async tasks */
  782. task->tk_workqueue = task_setup_data->workqueue;
  783. if (task->tk_ops->rpc_call_prepare != NULL)
  784. task->tk_action = rpc_prepare_task;
  785. rpc_init_task_statistics(task);
  786. dprintk("RPC: new task initialized, procpid %u\n",
  787. task_pid_nr(current));
  788. }
  789. static struct rpc_task *
  790. rpc_alloc_task(void)
  791. {
  792. return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOIO);
  793. }
  794. /*
  795. * Create a new task for the specified client.
  796. */
  797. struct rpc_task *rpc_new_task(const struct rpc_task_setup *setup_data)
  798. {
  799. struct rpc_task *task = setup_data->task;
  800. unsigned short flags = 0;
  801. if (task == NULL) {
  802. task = rpc_alloc_task();
  803. if (task == NULL) {
  804. rpc_release_calldata(setup_data->callback_ops,
  805. setup_data->callback_data);
  806. return ERR_PTR(-ENOMEM);
  807. }
  808. flags = RPC_TASK_DYNAMIC;
  809. }
  810. rpc_init_task(task, setup_data);
  811. task->tk_flags |= flags;
  812. dprintk("RPC: allocated task %p\n", task);
  813. return task;
  814. }
  815. static void rpc_free_task(struct rpc_task *task)
  816. {
  817. const struct rpc_call_ops *tk_ops = task->tk_ops;
  818. void *calldata = task->tk_calldata;
  819. if (task->tk_flags & RPC_TASK_DYNAMIC) {
  820. dprintk("RPC: %5u freeing task\n", task->tk_pid);
  821. mempool_free(task, rpc_task_mempool);
  822. }
  823. rpc_release_calldata(tk_ops, calldata);
  824. }
  825. static void rpc_async_release(struct work_struct *work)
  826. {
  827. rpc_free_task(container_of(work, struct rpc_task, u.tk_work));
  828. }
  829. static void rpc_release_resources_task(struct rpc_task *task)
  830. {
  831. if (task->tk_rqstp)
  832. xprt_release(task);
  833. if (task->tk_msg.rpc_cred) {
  834. put_rpccred(task->tk_msg.rpc_cred);
  835. task->tk_msg.rpc_cred = NULL;
  836. }
  837. rpc_task_release_client(task);
  838. }
  839. static void rpc_final_put_task(struct rpc_task *task,
  840. struct workqueue_struct *q)
  841. {
  842. if (q != NULL) {
  843. INIT_WORK(&task->u.tk_work, rpc_async_release);
  844. queue_work(q, &task->u.tk_work);
  845. } else
  846. rpc_free_task(task);
  847. }
  848. static void rpc_do_put_task(struct rpc_task *task, struct workqueue_struct *q)
  849. {
  850. if (atomic_dec_and_test(&task->tk_count)) {
  851. rpc_release_resources_task(task);
  852. rpc_final_put_task(task, q);
  853. }
  854. }
  855. void rpc_put_task(struct rpc_task *task)
  856. {
  857. rpc_do_put_task(task, NULL);
  858. }
  859. EXPORT_SYMBOL_GPL(rpc_put_task);
  860. void rpc_put_task_async(struct rpc_task *task)
  861. {
  862. rpc_do_put_task(task, task->tk_workqueue);
  863. }
  864. EXPORT_SYMBOL_GPL(rpc_put_task_async);
  865. static void rpc_release_task(struct rpc_task *task)
  866. {
  867. dprintk("RPC: %5u release task\n", task->tk_pid);
  868. BUG_ON (RPC_IS_QUEUED(task));
  869. rpc_release_resources_task(task);
  870. /*
  871. * Note: at this point we have been removed from rpc_clnt->cl_tasks,
  872. * so it should be safe to use task->tk_count as a test for whether
  873. * or not any other processes still hold references to our rpc_task.
  874. */
  875. if (atomic_read(&task->tk_count) != 1 + !RPC_IS_ASYNC(task)) {
  876. /* Wake up anyone who may be waiting for task completion */
  877. if (!rpc_complete_task(task))
  878. return;
  879. } else {
  880. if (!atomic_dec_and_test(&task->tk_count))
  881. return;
  882. }
  883. rpc_final_put_task(task, task->tk_workqueue);
  884. }
  885. int rpciod_up(void)
  886. {
  887. return try_module_get(THIS_MODULE) ? 0 : -EINVAL;
  888. }
  889. void rpciod_down(void)
  890. {
  891. module_put(THIS_MODULE);
  892. }
  893. /*
  894. * Start up the rpciod workqueue.
  895. */
  896. static int rpciod_start(void)
  897. {
  898. struct workqueue_struct *wq;
  899. /*
  900. * Create the rpciod thread and wait for it to start.
  901. */
  902. dprintk("RPC: creating workqueue rpciod\n");
  903. wq = alloc_workqueue("rpciod", WQ_MEM_RECLAIM, 1);
  904. rpciod_workqueue = wq;
  905. return rpciod_workqueue != NULL;
  906. }
  907. static void rpciod_stop(void)
  908. {
  909. struct workqueue_struct *wq = NULL;
  910. if (rpciod_workqueue == NULL)
  911. return;
  912. dprintk("RPC: destroying workqueue rpciod\n");
  913. wq = rpciod_workqueue;
  914. rpciod_workqueue = NULL;
  915. destroy_workqueue(wq);
  916. }
  917. void
  918. rpc_destroy_mempool(void)
  919. {
  920. rpciod_stop();
  921. if (rpc_buffer_mempool)
  922. mempool_destroy(rpc_buffer_mempool);
  923. if (rpc_task_mempool)
  924. mempool_destroy(rpc_task_mempool);
  925. if (rpc_task_slabp)
  926. kmem_cache_destroy(rpc_task_slabp);
  927. if (rpc_buffer_slabp)
  928. kmem_cache_destroy(rpc_buffer_slabp);
  929. rpc_destroy_wait_queue(&delay_queue);
  930. }
  931. int
  932. rpc_init_mempool(void)
  933. {
  934. /*
  935. * The following is not strictly a mempool initialisation,
  936. * but there is no harm in doing it here
  937. */
  938. rpc_init_wait_queue(&delay_queue, "delayq");
  939. if (!rpciod_start())
  940. goto err_nomem;
  941. rpc_task_slabp = kmem_cache_create("rpc_tasks",
  942. sizeof(struct rpc_task),
  943. 0, SLAB_HWCACHE_ALIGN,
  944. NULL);
  945. if (!rpc_task_slabp)
  946. goto err_nomem;
  947. rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
  948. RPC_BUFFER_MAXSIZE,
  949. 0, SLAB_HWCACHE_ALIGN,
  950. NULL);
  951. if (!rpc_buffer_slabp)
  952. goto err_nomem;
  953. rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
  954. rpc_task_slabp);
  955. if (!rpc_task_mempool)
  956. goto err_nomem;
  957. rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
  958. rpc_buffer_slabp);
  959. if (!rpc_buffer_mempool)
  960. goto err_nomem;
  961. return 0;
  962. err_nomem:
  963. rpc_destroy_mempool();
  964. return -ENOMEM;
  965. }