sched.c 26 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097
  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/smp_lock.h>
  18. #include <linux/spinlock.h>
  19. #include <linux/mutex.h>
  20. #include <linux/sunrpc/clnt.h>
  21. #ifdef RPC_DEBUG
  22. #define RPCDBG_FACILITY RPCDBG_SCHED
  23. #define RPC_TASK_MAGIC_ID 0xf00baa
  24. static int rpc_task_id;
  25. #endif
  26. /*
  27. * RPC slabs and memory pools
  28. */
  29. #define RPC_BUFFER_MAXSIZE (2048)
  30. #define RPC_BUFFER_POOLSIZE (8)
  31. #define RPC_TASK_POOLSIZE (8)
  32. static kmem_cache_t *rpc_task_slabp __read_mostly;
  33. static kmem_cache_t *rpc_buffer_slabp __read_mostly;
  34. static mempool_t *rpc_task_mempool __read_mostly;
  35. static mempool_t *rpc_buffer_mempool __read_mostly;
  36. static void __rpc_default_timer(struct rpc_task *task);
  37. static void rpciod_killall(void);
  38. static void rpc_async_schedule(void *);
  39. /*
  40. * RPC tasks sit here while waiting for conditions to improve.
  41. */
  42. static RPC_WAITQ(delay_queue, "delayq");
  43. /*
  44. * All RPC tasks are linked into this list
  45. */
  46. static LIST_HEAD(all_tasks);
  47. /*
  48. * rpciod-related stuff
  49. */
  50. static DEFINE_MUTEX(rpciod_mutex);
  51. static unsigned int rpciod_users;
  52. struct workqueue_struct *rpciod_workqueue;
  53. /*
  54. * Spinlock for other critical sections of code.
  55. */
  56. static DEFINE_SPINLOCK(rpc_sched_lock);
  57. /*
  58. * Disable the timer for a given RPC task. Should be called with
  59. * queue->lock and bh_disabled in order to avoid races within
  60. * rpc_run_timer().
  61. */
  62. static inline void
  63. __rpc_disable_timer(struct rpc_task *task)
  64. {
  65. dprintk("RPC: %4d disabling timer\n", task->tk_pid);
  66. task->tk_timeout_fn = NULL;
  67. task->tk_timeout = 0;
  68. }
  69. /*
  70. * Run a timeout function.
  71. * We use the callback in order to allow __rpc_wake_up_task()
  72. * and friends to disable the timer synchronously on SMP systems
  73. * without calling del_timer_sync(). The latter could cause a
  74. * deadlock if called while we're holding spinlocks...
  75. */
  76. static void rpc_run_timer(struct rpc_task *task)
  77. {
  78. void (*callback)(struct rpc_task *);
  79. callback = task->tk_timeout_fn;
  80. task->tk_timeout_fn = NULL;
  81. if (callback && RPC_IS_QUEUED(task)) {
  82. dprintk("RPC: %4d running timer\n", task->tk_pid);
  83. callback(task);
  84. }
  85. smp_mb__before_clear_bit();
  86. clear_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate);
  87. smp_mb__after_clear_bit();
  88. }
  89. /*
  90. * Set up a timer for the current task.
  91. */
  92. static inline void
  93. __rpc_add_timer(struct rpc_task *task, rpc_action timer)
  94. {
  95. if (!task->tk_timeout)
  96. return;
  97. dprintk("RPC: %4d setting alarm for %lu ms\n",
  98. task->tk_pid, task->tk_timeout * 1000 / HZ);
  99. if (timer)
  100. task->tk_timeout_fn = timer;
  101. else
  102. task->tk_timeout_fn = __rpc_default_timer;
  103. set_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate);
  104. mod_timer(&task->tk_timer, jiffies + task->tk_timeout);
  105. }
  106. /*
  107. * Delete any timer for the current task. Because we use del_timer_sync(),
  108. * this function should never be called while holding queue->lock.
  109. */
  110. static void
  111. rpc_delete_timer(struct rpc_task *task)
  112. {
  113. if (RPC_IS_QUEUED(task))
  114. return;
  115. if (test_and_clear_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate)) {
  116. del_singleshot_timer_sync(&task->tk_timer);
  117. dprintk("RPC: %4d deleting timer\n", task->tk_pid);
  118. }
  119. }
  120. /*
  121. * Add new request to a priority queue.
  122. */
  123. static void __rpc_add_wait_queue_priority(struct rpc_wait_queue *queue, struct rpc_task *task)
  124. {
  125. struct list_head *q;
  126. struct rpc_task *t;
  127. INIT_LIST_HEAD(&task->u.tk_wait.links);
  128. q = &queue->tasks[task->tk_priority];
  129. if (unlikely(task->tk_priority > queue->maxpriority))
  130. q = &queue->tasks[queue->maxpriority];
  131. list_for_each_entry(t, q, u.tk_wait.list) {
  132. if (t->tk_cookie == task->tk_cookie) {
  133. list_add_tail(&task->u.tk_wait.list, &t->u.tk_wait.links);
  134. return;
  135. }
  136. }
  137. list_add_tail(&task->u.tk_wait.list, q);
  138. }
  139. /*
  140. * Add new request to wait queue.
  141. *
  142. * Swapper tasks always get inserted at the head of the queue.
  143. * This should avoid many nasty memory deadlocks and hopefully
  144. * improve overall performance.
  145. * Everyone else gets appended to the queue to ensure proper FIFO behavior.
  146. */
  147. static void __rpc_add_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
  148. {
  149. BUG_ON (RPC_IS_QUEUED(task));
  150. if (RPC_IS_PRIORITY(queue))
  151. __rpc_add_wait_queue_priority(queue, task);
  152. else if (RPC_IS_SWAPPER(task))
  153. list_add(&task->u.tk_wait.list, &queue->tasks[0]);
  154. else
  155. list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
  156. task->u.tk_wait.rpc_waitq = queue;
  157. queue->qlen++;
  158. rpc_set_queued(task);
  159. dprintk("RPC: %4d added to queue %p \"%s\"\n",
  160. task->tk_pid, queue, rpc_qname(queue));
  161. }
  162. /*
  163. * Remove request from a priority queue.
  164. */
  165. static void __rpc_remove_wait_queue_priority(struct rpc_task *task)
  166. {
  167. struct rpc_task *t;
  168. if (!list_empty(&task->u.tk_wait.links)) {
  169. t = list_entry(task->u.tk_wait.links.next, struct rpc_task, u.tk_wait.list);
  170. list_move(&t->u.tk_wait.list, &task->u.tk_wait.list);
  171. list_splice_init(&task->u.tk_wait.links, &t->u.tk_wait.links);
  172. }
  173. list_del(&task->u.tk_wait.list);
  174. }
  175. /*
  176. * Remove request from queue.
  177. * Note: must be called with spin lock held.
  178. */
  179. static void __rpc_remove_wait_queue(struct rpc_task *task)
  180. {
  181. struct rpc_wait_queue *queue;
  182. queue = task->u.tk_wait.rpc_waitq;
  183. if (RPC_IS_PRIORITY(queue))
  184. __rpc_remove_wait_queue_priority(task);
  185. else
  186. list_del(&task->u.tk_wait.list);
  187. queue->qlen--;
  188. dprintk("RPC: %4d removed from queue %p \"%s\"\n",
  189. task->tk_pid, queue, rpc_qname(queue));
  190. }
  191. static inline void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority)
  192. {
  193. queue->priority = priority;
  194. queue->count = 1 << (priority * 2);
  195. }
  196. static inline void rpc_set_waitqueue_cookie(struct rpc_wait_queue *queue, unsigned long cookie)
  197. {
  198. queue->cookie = cookie;
  199. queue->nr = RPC_BATCH_COUNT;
  200. }
  201. static inline void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
  202. {
  203. rpc_set_waitqueue_priority(queue, queue->maxpriority);
  204. rpc_set_waitqueue_cookie(queue, 0);
  205. }
  206. static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, int maxprio)
  207. {
  208. int i;
  209. spin_lock_init(&queue->lock);
  210. for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
  211. INIT_LIST_HEAD(&queue->tasks[i]);
  212. queue->maxpriority = maxprio;
  213. rpc_reset_waitqueue_priority(queue);
  214. #ifdef RPC_DEBUG
  215. queue->name = qname;
  216. #endif
  217. }
  218. void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
  219. {
  220. __rpc_init_priority_wait_queue(queue, qname, RPC_PRIORITY_HIGH);
  221. }
  222. void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
  223. {
  224. __rpc_init_priority_wait_queue(queue, qname, 0);
  225. }
  226. EXPORT_SYMBOL(rpc_init_wait_queue);
  227. static int rpc_wait_bit_interruptible(void *word)
  228. {
  229. if (signal_pending(current))
  230. return -ERESTARTSYS;
  231. schedule();
  232. return 0;
  233. }
  234. /*
  235. * Mark an RPC call as having completed by clearing the 'active' bit
  236. */
  237. static inline void rpc_mark_complete_task(struct rpc_task *task)
  238. {
  239. rpc_clear_active(task);
  240. wake_up_bit(&task->tk_runstate, RPC_TASK_ACTIVE);
  241. }
  242. /*
  243. * Allow callers to wait for completion of an RPC call
  244. */
  245. int __rpc_wait_for_completion_task(struct rpc_task *task, int (*action)(void *))
  246. {
  247. if (action == NULL)
  248. action = rpc_wait_bit_interruptible;
  249. return wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
  250. action, TASK_INTERRUPTIBLE);
  251. }
  252. EXPORT_SYMBOL(__rpc_wait_for_completion_task);
  253. /*
  254. * Make an RPC task runnable.
  255. *
  256. * Note: If the task is ASYNC, this must be called with
  257. * the spinlock held to protect the wait queue operation.
  258. */
  259. static void rpc_make_runnable(struct rpc_task *task)
  260. {
  261. BUG_ON(task->tk_timeout_fn);
  262. rpc_clear_queued(task);
  263. if (rpc_test_and_set_running(task))
  264. return;
  265. /* We might have raced */
  266. if (RPC_IS_QUEUED(task)) {
  267. rpc_clear_running(task);
  268. return;
  269. }
  270. if (RPC_IS_ASYNC(task)) {
  271. int status;
  272. INIT_WORK(&task->u.tk_work, rpc_async_schedule, (void *)task);
  273. status = queue_work(task->tk_workqueue, &task->u.tk_work);
  274. if (status < 0) {
  275. printk(KERN_WARNING "RPC: failed to add task to queue: error: %d!\n", status);
  276. task->tk_status = status;
  277. return;
  278. }
  279. } else
  280. wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
  281. }
  282. /*
  283. * Prepare for sleeping on a wait queue.
  284. * By always appending tasks to the list we ensure FIFO behavior.
  285. * NB: An RPC task will only receive interrupt-driven events as long
  286. * as it's on a wait queue.
  287. */
  288. static void __rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
  289. rpc_action action, rpc_action timer)
  290. {
  291. dprintk("RPC: %4d sleep_on(queue \"%s\" time %ld)\n", task->tk_pid,
  292. rpc_qname(q), jiffies);
  293. if (!RPC_IS_ASYNC(task) && !RPC_IS_ACTIVATED(task)) {
  294. printk(KERN_ERR "RPC: Inactive synchronous task put to sleep!\n");
  295. return;
  296. }
  297. /* Mark the task as being activated if so needed */
  298. rpc_set_active(task);
  299. __rpc_add_wait_queue(q, task);
  300. BUG_ON(task->tk_callback != NULL);
  301. task->tk_callback = action;
  302. __rpc_add_timer(task, timer);
  303. }
  304. void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
  305. rpc_action action, rpc_action timer)
  306. {
  307. /*
  308. * Protect the queue operations.
  309. */
  310. spin_lock_bh(&q->lock);
  311. __rpc_sleep_on(q, task, action, timer);
  312. spin_unlock_bh(&q->lock);
  313. }
  314. /**
  315. * __rpc_do_wake_up_task - wake up a single rpc_task
  316. * @task: task to be woken up
  317. *
  318. * Caller must hold queue->lock, and have cleared the task queued flag.
  319. */
  320. static void __rpc_do_wake_up_task(struct rpc_task *task)
  321. {
  322. dprintk("RPC: %4d __rpc_wake_up_task (now %ld)\n", task->tk_pid, jiffies);
  323. #ifdef RPC_DEBUG
  324. BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID);
  325. #endif
  326. /* Has the task been executed yet? If not, we cannot wake it up! */
  327. if (!RPC_IS_ACTIVATED(task)) {
  328. printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
  329. return;
  330. }
  331. __rpc_disable_timer(task);
  332. __rpc_remove_wait_queue(task);
  333. rpc_make_runnable(task);
  334. dprintk("RPC: __rpc_wake_up_task done\n");
  335. }
  336. /*
  337. * Wake up the specified task
  338. */
  339. static void __rpc_wake_up_task(struct rpc_task *task)
  340. {
  341. if (rpc_start_wakeup(task)) {
  342. if (RPC_IS_QUEUED(task))
  343. __rpc_do_wake_up_task(task);
  344. rpc_finish_wakeup(task);
  345. }
  346. }
  347. /*
  348. * Default timeout handler if none specified by user
  349. */
  350. static void
  351. __rpc_default_timer(struct rpc_task *task)
  352. {
  353. dprintk("RPC: %d timeout (default timer)\n", task->tk_pid);
  354. task->tk_status = -ETIMEDOUT;
  355. rpc_wake_up_task(task);
  356. }
  357. /*
  358. * Wake up the specified task
  359. */
  360. void rpc_wake_up_task(struct rpc_task *task)
  361. {
  362. if (rpc_start_wakeup(task)) {
  363. if (RPC_IS_QUEUED(task)) {
  364. struct rpc_wait_queue *queue = task->u.tk_wait.rpc_waitq;
  365. spin_lock_bh(&queue->lock);
  366. __rpc_do_wake_up_task(task);
  367. spin_unlock_bh(&queue->lock);
  368. }
  369. rpc_finish_wakeup(task);
  370. }
  371. }
  372. /*
  373. * Wake up the next task on a priority queue.
  374. */
  375. static struct rpc_task * __rpc_wake_up_next_priority(struct rpc_wait_queue *queue)
  376. {
  377. struct list_head *q;
  378. struct rpc_task *task;
  379. /*
  380. * Service a batch of tasks from a single cookie.
  381. */
  382. q = &queue->tasks[queue->priority];
  383. if (!list_empty(q)) {
  384. task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
  385. if (queue->cookie == task->tk_cookie) {
  386. if (--queue->nr)
  387. goto out;
  388. list_move_tail(&task->u.tk_wait.list, q);
  389. }
  390. /*
  391. * Check if we need to switch queues.
  392. */
  393. if (--queue->count)
  394. goto new_cookie;
  395. }
  396. /*
  397. * Service the next queue.
  398. */
  399. do {
  400. if (q == &queue->tasks[0])
  401. q = &queue->tasks[queue->maxpriority];
  402. else
  403. q = q - 1;
  404. if (!list_empty(q)) {
  405. task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
  406. goto new_queue;
  407. }
  408. } while (q != &queue->tasks[queue->priority]);
  409. rpc_reset_waitqueue_priority(queue);
  410. return NULL;
  411. new_queue:
  412. rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0]));
  413. new_cookie:
  414. rpc_set_waitqueue_cookie(queue, task->tk_cookie);
  415. out:
  416. __rpc_wake_up_task(task);
  417. return task;
  418. }
  419. /*
  420. * Wake up the next task on the wait queue.
  421. */
  422. struct rpc_task * rpc_wake_up_next(struct rpc_wait_queue *queue)
  423. {
  424. struct rpc_task *task = NULL;
  425. dprintk("RPC: wake_up_next(%p \"%s\")\n", queue, rpc_qname(queue));
  426. spin_lock_bh(&queue->lock);
  427. if (RPC_IS_PRIORITY(queue))
  428. task = __rpc_wake_up_next_priority(queue);
  429. else {
  430. task_for_first(task, &queue->tasks[0])
  431. __rpc_wake_up_task(task);
  432. }
  433. spin_unlock_bh(&queue->lock);
  434. return task;
  435. }
  436. /**
  437. * rpc_wake_up - wake up all rpc_tasks
  438. * @queue: rpc_wait_queue on which the tasks are sleeping
  439. *
  440. * Grabs queue->lock
  441. */
  442. void rpc_wake_up(struct rpc_wait_queue *queue)
  443. {
  444. struct rpc_task *task, *next;
  445. struct list_head *head;
  446. spin_lock_bh(&queue->lock);
  447. head = &queue->tasks[queue->maxpriority];
  448. for (;;) {
  449. list_for_each_entry_safe(task, next, head, u.tk_wait.list)
  450. __rpc_wake_up_task(task);
  451. if (head == &queue->tasks[0])
  452. break;
  453. head--;
  454. }
  455. spin_unlock_bh(&queue->lock);
  456. }
  457. /**
  458. * rpc_wake_up_status - wake up all rpc_tasks and set their status value.
  459. * @queue: rpc_wait_queue on which the tasks are sleeping
  460. * @status: status value to set
  461. *
  462. * Grabs queue->lock
  463. */
  464. void rpc_wake_up_status(struct rpc_wait_queue *queue, int status)
  465. {
  466. struct rpc_task *task, *next;
  467. struct list_head *head;
  468. spin_lock_bh(&queue->lock);
  469. head = &queue->tasks[queue->maxpriority];
  470. for (;;) {
  471. list_for_each_entry_safe(task, next, head, u.tk_wait.list) {
  472. task->tk_status = status;
  473. __rpc_wake_up_task(task);
  474. }
  475. if (head == &queue->tasks[0])
  476. break;
  477. head--;
  478. }
  479. spin_unlock_bh(&queue->lock);
  480. }
  481. static void __rpc_atrun(struct rpc_task *task)
  482. {
  483. rpc_wake_up_task(task);
  484. }
  485. /*
  486. * Run a task at a later time
  487. */
  488. void rpc_delay(struct rpc_task *task, unsigned long delay)
  489. {
  490. task->tk_timeout = delay;
  491. rpc_sleep_on(&delay_queue, task, NULL, __rpc_atrun);
  492. }
  493. /*
  494. * Helper to call task->tk_ops->rpc_call_prepare
  495. */
  496. static void rpc_prepare_task(struct rpc_task *task)
  497. {
  498. task->tk_ops->rpc_call_prepare(task, task->tk_calldata);
  499. }
  500. /*
  501. * Helper that calls task->tk_ops->rpc_call_done if it exists
  502. */
  503. void rpc_exit_task(struct rpc_task *task)
  504. {
  505. task->tk_action = NULL;
  506. if (task->tk_ops->rpc_call_done != NULL) {
  507. task->tk_ops->rpc_call_done(task, task->tk_calldata);
  508. if (task->tk_action != NULL) {
  509. WARN_ON(RPC_ASSASSINATED(task));
  510. /* Always release the RPC slot and buffer memory */
  511. xprt_release(task);
  512. }
  513. }
  514. }
  515. EXPORT_SYMBOL(rpc_exit_task);
  516. /*
  517. * This is the RPC `scheduler' (or rather, the finite state machine).
  518. */
  519. static int __rpc_execute(struct rpc_task *task)
  520. {
  521. int status = 0;
  522. dprintk("RPC: %4d rpc_execute flgs %x\n",
  523. task->tk_pid, task->tk_flags);
  524. BUG_ON(RPC_IS_QUEUED(task));
  525. for (;;) {
  526. /*
  527. * Garbage collection of pending timers...
  528. */
  529. rpc_delete_timer(task);
  530. /*
  531. * Execute any pending callback.
  532. */
  533. if (RPC_DO_CALLBACK(task)) {
  534. /* Define a callback save pointer */
  535. void (*save_callback)(struct rpc_task *);
  536. /*
  537. * If a callback exists, save it, reset it,
  538. * call it.
  539. * The save is needed to stop from resetting
  540. * another callback set within the callback handler
  541. * - Dave
  542. */
  543. save_callback=task->tk_callback;
  544. task->tk_callback=NULL;
  545. lock_kernel();
  546. save_callback(task);
  547. unlock_kernel();
  548. }
  549. /*
  550. * Perform the next FSM step.
  551. * tk_action may be NULL when the task has been killed
  552. * by someone else.
  553. */
  554. if (!RPC_IS_QUEUED(task)) {
  555. if (task->tk_action == NULL)
  556. break;
  557. lock_kernel();
  558. task->tk_action(task);
  559. unlock_kernel();
  560. }
  561. /*
  562. * Lockless check for whether task is sleeping or not.
  563. */
  564. if (!RPC_IS_QUEUED(task))
  565. continue;
  566. rpc_clear_running(task);
  567. if (RPC_IS_ASYNC(task)) {
  568. /* Careful! we may have raced... */
  569. if (RPC_IS_QUEUED(task))
  570. return 0;
  571. if (rpc_test_and_set_running(task))
  572. return 0;
  573. continue;
  574. }
  575. /* sync task: sleep here */
  576. dprintk("RPC: %4d sync task going to sleep\n", task->tk_pid);
  577. /* Note: Caller should be using rpc_clnt_sigmask() */
  578. status = out_of_line_wait_on_bit(&task->tk_runstate,
  579. RPC_TASK_QUEUED, rpc_wait_bit_interruptible,
  580. TASK_INTERRUPTIBLE);
  581. if (status == -ERESTARTSYS) {
  582. /*
  583. * When a sync task receives a signal, it exits with
  584. * -ERESTARTSYS. In order to catch any callbacks that
  585. * clean up after sleeping on some queue, we don't
  586. * break the loop here, but go around once more.
  587. */
  588. dprintk("RPC: %4d got signal\n", task->tk_pid);
  589. task->tk_flags |= RPC_TASK_KILLED;
  590. rpc_exit(task, -ERESTARTSYS);
  591. rpc_wake_up_task(task);
  592. }
  593. rpc_set_running(task);
  594. dprintk("RPC: %4d sync task resuming\n", task->tk_pid);
  595. }
  596. dprintk("RPC: %4d, return %d, status %d\n", task->tk_pid, status, task->tk_status);
  597. /* Wake up anyone who is waiting for task completion */
  598. rpc_mark_complete_task(task);
  599. /* Release all resources associated with the task */
  600. rpc_release_task(task);
  601. return status;
  602. }
  603. /*
  604. * User-visible entry point to the scheduler.
  605. *
  606. * This may be called recursively if e.g. an async NFS task updates
  607. * the attributes and finds that dirty pages must be flushed.
  608. * NOTE: Upon exit of this function the task is guaranteed to be
  609. * released. In particular note that tk_release() will have
  610. * been called, so your task memory may have been freed.
  611. */
  612. int
  613. rpc_execute(struct rpc_task *task)
  614. {
  615. rpc_set_active(task);
  616. rpc_set_running(task);
  617. return __rpc_execute(task);
  618. }
  619. static void rpc_async_schedule(void *arg)
  620. {
  621. __rpc_execute((struct rpc_task *)arg);
  622. }
  623. /**
  624. * rpc_malloc - allocate an RPC buffer
  625. * @task: RPC task that will use this buffer
  626. * @size: requested byte size
  627. *
  628. * We try to ensure that some NFS reads and writes can always proceed
  629. * by using a mempool when allocating 'small' buffers.
  630. * In order to avoid memory starvation triggering more writebacks of
  631. * NFS requests, we use GFP_NOFS rather than GFP_KERNEL.
  632. */
  633. void * rpc_malloc(struct rpc_task *task, size_t size)
  634. {
  635. struct rpc_rqst *req = task->tk_rqstp;
  636. gfp_t gfp;
  637. if (task->tk_flags & RPC_TASK_SWAPPER)
  638. gfp = GFP_ATOMIC;
  639. else
  640. gfp = GFP_NOFS;
  641. if (size > RPC_BUFFER_MAXSIZE) {
  642. req->rq_buffer = kmalloc(size, gfp);
  643. if (req->rq_buffer)
  644. req->rq_bufsize = size;
  645. } else {
  646. req->rq_buffer = mempool_alloc(rpc_buffer_mempool, gfp);
  647. if (req->rq_buffer)
  648. req->rq_bufsize = RPC_BUFFER_MAXSIZE;
  649. }
  650. return req->rq_buffer;
  651. }
  652. /**
  653. * rpc_free - free buffer allocated via rpc_malloc
  654. * @task: RPC task with a buffer to be freed
  655. *
  656. */
  657. void rpc_free(struct rpc_task *task)
  658. {
  659. struct rpc_rqst *req = task->tk_rqstp;
  660. if (req->rq_buffer) {
  661. if (req->rq_bufsize == RPC_BUFFER_MAXSIZE)
  662. mempool_free(req->rq_buffer, rpc_buffer_mempool);
  663. else
  664. kfree(req->rq_buffer);
  665. req->rq_buffer = NULL;
  666. req->rq_bufsize = 0;
  667. }
  668. }
  669. /*
  670. * Creation and deletion of RPC task structures
  671. */
  672. void rpc_init_task(struct rpc_task *task, struct rpc_clnt *clnt, int flags, const struct rpc_call_ops *tk_ops, void *calldata)
  673. {
  674. memset(task, 0, sizeof(*task));
  675. init_timer(&task->tk_timer);
  676. task->tk_timer.data = (unsigned long) task;
  677. task->tk_timer.function = (void (*)(unsigned long)) rpc_run_timer;
  678. atomic_set(&task->tk_count, 1);
  679. task->tk_client = clnt;
  680. task->tk_flags = flags;
  681. task->tk_ops = tk_ops;
  682. if (tk_ops->rpc_call_prepare != NULL)
  683. task->tk_action = rpc_prepare_task;
  684. task->tk_calldata = calldata;
  685. /* Initialize retry counters */
  686. task->tk_garb_retry = 2;
  687. task->tk_cred_retry = 2;
  688. task->tk_priority = RPC_PRIORITY_NORMAL;
  689. task->tk_cookie = (unsigned long)current;
  690. /* Initialize workqueue for async tasks */
  691. task->tk_workqueue = rpciod_workqueue;
  692. if (clnt) {
  693. atomic_inc(&clnt->cl_users);
  694. if (clnt->cl_softrtry)
  695. task->tk_flags |= RPC_TASK_SOFT;
  696. if (!clnt->cl_intr)
  697. task->tk_flags |= RPC_TASK_NOINTR;
  698. }
  699. #ifdef RPC_DEBUG
  700. task->tk_magic = RPC_TASK_MAGIC_ID;
  701. task->tk_pid = rpc_task_id++;
  702. #endif
  703. /* Add to global list of all tasks */
  704. spin_lock(&rpc_sched_lock);
  705. list_add_tail(&task->tk_task, &all_tasks);
  706. spin_unlock(&rpc_sched_lock);
  707. BUG_ON(task->tk_ops == NULL);
  708. /* starting timestamp */
  709. task->tk_start = jiffies;
  710. dprintk("RPC: %4d new task procpid %d\n", task->tk_pid,
  711. current->pid);
  712. }
  713. static struct rpc_task *
  714. rpc_alloc_task(void)
  715. {
  716. return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOFS);
  717. }
  718. static void rpc_free_task(struct rpc_task *task)
  719. {
  720. dprintk("RPC: %4d freeing task\n", task->tk_pid);
  721. mempool_free(task, rpc_task_mempool);
  722. }
  723. /*
  724. * Create a new task for the specified client. We have to
  725. * clean up after an allocation failure, as the client may
  726. * have specified "oneshot".
  727. */
  728. struct rpc_task *rpc_new_task(struct rpc_clnt *clnt, int flags, const struct rpc_call_ops *tk_ops, void *calldata)
  729. {
  730. struct rpc_task *task;
  731. task = rpc_alloc_task();
  732. if (!task)
  733. goto cleanup;
  734. rpc_init_task(task, clnt, flags, tk_ops, calldata);
  735. dprintk("RPC: %4d allocated task\n", task->tk_pid);
  736. task->tk_flags |= RPC_TASK_DYNAMIC;
  737. out:
  738. return task;
  739. cleanup:
  740. /* Check whether to release the client */
  741. if (clnt) {
  742. printk("rpc_new_task: failed, users=%d, oneshot=%d\n",
  743. atomic_read(&clnt->cl_users), clnt->cl_oneshot);
  744. atomic_inc(&clnt->cl_users); /* pretend we were used ... */
  745. rpc_release_client(clnt);
  746. }
  747. goto out;
  748. }
  749. void rpc_release_task(struct rpc_task *task)
  750. {
  751. const struct rpc_call_ops *tk_ops = task->tk_ops;
  752. void *calldata = task->tk_calldata;
  753. #ifdef RPC_DEBUG
  754. BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID);
  755. #endif
  756. if (!atomic_dec_and_test(&task->tk_count))
  757. return;
  758. dprintk("RPC: %4d release task\n", task->tk_pid);
  759. /* Remove from global task list */
  760. spin_lock(&rpc_sched_lock);
  761. list_del(&task->tk_task);
  762. spin_unlock(&rpc_sched_lock);
  763. BUG_ON (RPC_IS_QUEUED(task));
  764. /* Synchronously delete any running timer */
  765. rpc_delete_timer(task);
  766. /* Release resources */
  767. if (task->tk_rqstp)
  768. xprt_release(task);
  769. if (task->tk_msg.rpc_cred)
  770. rpcauth_unbindcred(task);
  771. if (task->tk_client) {
  772. rpc_release_client(task->tk_client);
  773. task->tk_client = NULL;
  774. }
  775. #ifdef RPC_DEBUG
  776. task->tk_magic = 0;
  777. #endif
  778. if (task->tk_flags & RPC_TASK_DYNAMIC)
  779. rpc_free_task(task);
  780. if (tk_ops->rpc_release)
  781. tk_ops->rpc_release(calldata);
  782. }
  783. /**
  784. * rpc_run_task - Allocate a new RPC task, then run rpc_execute against it
  785. * @clnt: pointer to RPC client
  786. * @flags: RPC flags
  787. * @ops: RPC call ops
  788. * @data: user call data
  789. */
  790. struct rpc_task *rpc_run_task(struct rpc_clnt *clnt, int flags,
  791. const struct rpc_call_ops *ops,
  792. void *data)
  793. {
  794. struct rpc_task *task;
  795. task = rpc_new_task(clnt, flags, ops, data);
  796. if (task == NULL) {
  797. if (ops->rpc_release != NULL)
  798. ops->rpc_release(data);
  799. return ERR_PTR(-ENOMEM);
  800. }
  801. atomic_inc(&task->tk_count);
  802. rpc_execute(task);
  803. return task;
  804. }
  805. EXPORT_SYMBOL(rpc_run_task);
  806. /*
  807. * Kill all tasks for the given client.
  808. * XXX: kill their descendants as well?
  809. */
  810. void rpc_killall_tasks(struct rpc_clnt *clnt)
  811. {
  812. struct rpc_task *rovr;
  813. struct list_head *le;
  814. dprintk("RPC: killing all tasks for client %p\n", clnt);
  815. /*
  816. * Spin lock all_tasks to prevent changes...
  817. */
  818. spin_lock(&rpc_sched_lock);
  819. alltask_for_each(rovr, le, &all_tasks) {
  820. if (! RPC_IS_ACTIVATED(rovr))
  821. continue;
  822. if (!clnt || rovr->tk_client == clnt) {
  823. rovr->tk_flags |= RPC_TASK_KILLED;
  824. rpc_exit(rovr, -EIO);
  825. rpc_wake_up_task(rovr);
  826. }
  827. }
  828. spin_unlock(&rpc_sched_lock);
  829. }
  830. static DECLARE_MUTEX_LOCKED(rpciod_running);
  831. static void rpciod_killall(void)
  832. {
  833. unsigned long flags;
  834. while (!list_empty(&all_tasks)) {
  835. clear_thread_flag(TIF_SIGPENDING);
  836. rpc_killall_tasks(NULL);
  837. flush_workqueue(rpciod_workqueue);
  838. if (!list_empty(&all_tasks)) {
  839. dprintk("rpciod_killall: waiting for tasks to exit\n");
  840. yield();
  841. }
  842. }
  843. spin_lock_irqsave(&current->sighand->siglock, flags);
  844. recalc_sigpending();
  845. spin_unlock_irqrestore(&current->sighand->siglock, flags);
  846. }
  847. /*
  848. * Start up the rpciod process if it's not already running.
  849. */
  850. int
  851. rpciod_up(void)
  852. {
  853. struct workqueue_struct *wq;
  854. int error = 0;
  855. mutex_lock(&rpciod_mutex);
  856. dprintk("rpciod_up: users %d\n", rpciod_users);
  857. rpciod_users++;
  858. if (rpciod_workqueue)
  859. goto out;
  860. /*
  861. * If there's no pid, we should be the first user.
  862. */
  863. if (rpciod_users > 1)
  864. printk(KERN_WARNING "rpciod_up: no workqueue, %d users??\n", rpciod_users);
  865. /*
  866. * Create the rpciod thread and wait for it to start.
  867. */
  868. error = -ENOMEM;
  869. wq = create_workqueue("rpciod");
  870. if (wq == NULL) {
  871. printk(KERN_WARNING "rpciod_up: create workqueue failed, error=%d\n", error);
  872. rpciod_users--;
  873. goto out;
  874. }
  875. rpciod_workqueue = wq;
  876. error = 0;
  877. out:
  878. mutex_unlock(&rpciod_mutex);
  879. return error;
  880. }
  881. void
  882. rpciod_down(void)
  883. {
  884. mutex_lock(&rpciod_mutex);
  885. dprintk("rpciod_down sema %d\n", rpciod_users);
  886. if (rpciod_users) {
  887. if (--rpciod_users)
  888. goto out;
  889. } else
  890. printk(KERN_WARNING "rpciod_down: no users??\n");
  891. if (!rpciod_workqueue) {
  892. dprintk("rpciod_down: Nothing to do!\n");
  893. goto out;
  894. }
  895. rpciod_killall();
  896. destroy_workqueue(rpciod_workqueue);
  897. rpciod_workqueue = NULL;
  898. out:
  899. mutex_unlock(&rpciod_mutex);
  900. }
  901. #ifdef RPC_DEBUG
  902. void rpc_show_tasks(void)
  903. {
  904. struct list_head *le;
  905. struct rpc_task *t;
  906. spin_lock(&rpc_sched_lock);
  907. if (list_empty(&all_tasks)) {
  908. spin_unlock(&rpc_sched_lock);
  909. return;
  910. }
  911. printk("-pid- proc flgs status -client- -prog- --rqstp- -timeout "
  912. "-rpcwait -action- ---ops--\n");
  913. alltask_for_each(t, le, &all_tasks) {
  914. const char *rpc_waitq = "none";
  915. if (RPC_IS_QUEUED(t))
  916. rpc_waitq = rpc_qname(t->u.tk_wait.rpc_waitq);
  917. printk("%05d %04d %04x %06d %8p %6d %8p %08ld %8s %8p %8p\n",
  918. t->tk_pid,
  919. (t->tk_msg.rpc_proc ? t->tk_msg.rpc_proc->p_proc : -1),
  920. t->tk_flags, t->tk_status,
  921. t->tk_client,
  922. (t->tk_client ? t->tk_client->cl_prog : 0),
  923. t->tk_rqstp, t->tk_timeout,
  924. rpc_waitq,
  925. t->tk_action, t->tk_ops);
  926. }
  927. spin_unlock(&rpc_sched_lock);
  928. }
  929. #endif
  930. void
  931. rpc_destroy_mempool(void)
  932. {
  933. if (rpc_buffer_mempool)
  934. mempool_destroy(rpc_buffer_mempool);
  935. if (rpc_task_mempool)
  936. mempool_destroy(rpc_task_mempool);
  937. if (rpc_task_slabp)
  938. kmem_cache_destroy(rpc_task_slabp);
  939. if (rpc_buffer_slabp)
  940. kmem_cache_destroy(rpc_buffer_slabp);
  941. }
  942. int
  943. rpc_init_mempool(void)
  944. {
  945. rpc_task_slabp = kmem_cache_create("rpc_tasks",
  946. sizeof(struct rpc_task),
  947. 0, SLAB_HWCACHE_ALIGN,
  948. NULL, NULL);
  949. if (!rpc_task_slabp)
  950. goto err_nomem;
  951. rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
  952. RPC_BUFFER_MAXSIZE,
  953. 0, SLAB_HWCACHE_ALIGN,
  954. NULL, NULL);
  955. if (!rpc_buffer_slabp)
  956. goto err_nomem;
  957. rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
  958. rpc_task_slabp);
  959. if (!rpc_task_mempool)
  960. goto err_nomem;
  961. rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
  962. rpc_buffer_slabp);
  963. if (!rpc_buffer_mempool)
  964. goto err_nomem;
  965. return 0;
  966. err_nomem:
  967. rpc_destroy_mempool();
  968. return -ENOMEM;
  969. }