xprt.c 31 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191
  1. /*
  2. * linux/net/sunrpc/xprt.c
  3. *
  4. * This is a generic RPC call interface supporting congestion avoidance,
  5. * and asynchronous calls.
  6. *
  7. * The interface works like this:
  8. *
  9. * - When a process places a call, it allocates a request slot if
  10. * one is available. Otherwise, it sleeps on the backlog queue
  11. * (xprt_reserve).
  12. * - Next, the caller puts together the RPC message, stuffs it into
  13. * the request struct, and calls xprt_transmit().
  14. * - xprt_transmit sends the message and installs the caller on the
  15. * transport's wait list. At the same time, if a reply is expected,
  16. * it installs a timer that is run after the packet's timeout has
  17. * expired.
  18. * - When a packet arrives, the data_ready handler walks the list of
  19. * pending requests for that transport. If a matching XID is found, the
  20. * caller is woken up, and the timer removed.
  21. * - When no reply arrives within the timeout interval, the timer is
  22. * fired by the kernel and runs xprt_timer(). It either adjusts the
  23. * timeout values (minor timeout) or wakes up the caller with a status
  24. * of -ETIMEDOUT.
  25. * - When the caller receives a notification from RPC that a reply arrived,
  26. * it should release the RPC slot, and process the reply.
  27. * If the call timed out, it may choose to retry the operation by
  28. * adjusting the initial timeout value, and simply calling rpc_call
  29. * again.
  30. *
  31. * Support for async RPC is done through a set of RPC-specific scheduling
  32. * primitives that `transparently' work for processes as well as async
  33. * tasks that rely on callbacks.
  34. *
  35. * Copyright (C) 1995-1997, Olaf Kirch <okir@monad.swb.de>
  36. *
  37. * Transport switch API copyright (C) 2005, Chuck Lever <cel@netapp.com>
  38. */
  39. #include <linux/module.h>
  40. #include <linux/types.h>
  41. #include <linux/interrupt.h>
  42. #include <linux/workqueue.h>
  43. #include <linux/net.h>
  44. #include <linux/ktime.h>
  45. #include <linux/sunrpc/clnt.h>
  46. #include <linux/sunrpc/metrics.h>
  47. #include <linux/sunrpc/bc_xprt.h>
  48. #include "sunrpc.h"
  49. /*
  50. * Local variables
  51. */
  52. #ifdef RPC_DEBUG
  53. # define RPCDBG_FACILITY RPCDBG_XPRT
  54. #endif
  55. /*
  56. * Local functions
  57. */
  58. static void xprt_request_init(struct rpc_task *, struct rpc_xprt *);
  59. static void xprt_connect_status(struct rpc_task *task);
  60. static int __xprt_get_cong(struct rpc_xprt *, struct rpc_task *);
  61. static DEFINE_SPINLOCK(xprt_list_lock);
  62. static LIST_HEAD(xprt_list);
  63. /*
  64. * The transport code maintains an estimate on the maximum number of out-
  65. * standing RPC requests, using a smoothed version of the congestion
  66. * avoidance implemented in 44BSD. This is basically the Van Jacobson
  67. * congestion algorithm: If a retransmit occurs, the congestion window is
  68. * halved; otherwise, it is incremented by 1/cwnd when
  69. *
  70. * - a reply is received and
  71. * - a full number of requests are outstanding and
  72. * - the congestion window hasn't been updated recently.
  73. */
  74. #define RPC_CWNDSHIFT (8U)
  75. #define RPC_CWNDSCALE (1U << RPC_CWNDSHIFT)
  76. #define RPC_INITCWND RPC_CWNDSCALE
  77. #define RPC_MAXCWND(xprt) ((xprt)->max_reqs << RPC_CWNDSHIFT)
  78. #define RPCXPRT_CONGESTED(xprt) ((xprt)->cong >= (xprt)->cwnd)
  79. /**
  80. * xprt_register_transport - register a transport implementation
  81. * @transport: transport to register
  82. *
  83. * If a transport implementation is loaded as a kernel module, it can
  84. * call this interface to make itself known to the RPC client.
  85. *
  86. * Returns:
  87. * 0: transport successfully registered
  88. * -EEXIST: transport already registered
  89. * -EINVAL: transport module being unloaded
  90. */
  91. int xprt_register_transport(struct xprt_class *transport)
  92. {
  93. struct xprt_class *t;
  94. int result;
  95. result = -EEXIST;
  96. spin_lock(&xprt_list_lock);
  97. list_for_each_entry(t, &xprt_list, list) {
  98. /* don't register the same transport class twice */
  99. if (t->ident == transport->ident)
  100. goto out;
  101. }
  102. list_add_tail(&transport->list, &xprt_list);
  103. printk(KERN_INFO "RPC: Registered %s transport module.\n",
  104. transport->name);
  105. result = 0;
  106. out:
  107. spin_unlock(&xprt_list_lock);
  108. return result;
  109. }
  110. EXPORT_SYMBOL_GPL(xprt_register_transport);
  111. /**
  112. * xprt_unregister_transport - unregister a transport implementation
  113. * @transport: transport to unregister
  114. *
  115. * Returns:
  116. * 0: transport successfully unregistered
  117. * -ENOENT: transport never registered
  118. */
  119. int xprt_unregister_transport(struct xprt_class *transport)
  120. {
  121. struct xprt_class *t;
  122. int result;
  123. result = 0;
  124. spin_lock(&xprt_list_lock);
  125. list_for_each_entry(t, &xprt_list, list) {
  126. if (t == transport) {
  127. printk(KERN_INFO
  128. "RPC: Unregistered %s transport module.\n",
  129. transport->name);
  130. list_del_init(&transport->list);
  131. goto out;
  132. }
  133. }
  134. result = -ENOENT;
  135. out:
  136. spin_unlock(&xprt_list_lock);
  137. return result;
  138. }
  139. EXPORT_SYMBOL_GPL(xprt_unregister_transport);
  140. /**
  141. * xprt_load_transport - load a transport implementation
  142. * @transport_name: transport to load
  143. *
  144. * Returns:
  145. * 0: transport successfully loaded
  146. * -ENOENT: transport module not available
  147. */
  148. int xprt_load_transport(const char *transport_name)
  149. {
  150. struct xprt_class *t;
  151. int result;
  152. result = 0;
  153. spin_lock(&xprt_list_lock);
  154. list_for_each_entry(t, &xprt_list, list) {
  155. if (strcmp(t->name, transport_name) == 0) {
  156. spin_unlock(&xprt_list_lock);
  157. goto out;
  158. }
  159. }
  160. spin_unlock(&xprt_list_lock);
  161. result = request_module("xprt%s", transport_name);
  162. out:
  163. return result;
  164. }
  165. EXPORT_SYMBOL_GPL(xprt_load_transport);
  166. /**
  167. * xprt_reserve_xprt - serialize write access to transports
  168. * @task: task that is requesting access to the transport
  169. *
  170. * This prevents mixing the payload of separate requests, and prevents
  171. * transport connects from colliding with writes. No congestion control
  172. * is provided.
  173. */
  174. int xprt_reserve_xprt(struct rpc_task *task)
  175. {
  176. struct rpc_rqst *req = task->tk_rqstp;
  177. struct rpc_xprt *xprt = req->rq_xprt;
  178. if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
  179. if (task == xprt->snd_task)
  180. return 1;
  181. goto out_sleep;
  182. }
  183. xprt->snd_task = task;
  184. req->rq_bytes_sent = 0;
  185. req->rq_ntrans++;
  186. return 1;
  187. out_sleep:
  188. dprintk("RPC: %5u failed to lock transport %p\n",
  189. task->tk_pid, xprt);
  190. task->tk_timeout = 0;
  191. task->tk_status = -EAGAIN;
  192. if (req->rq_ntrans)
  193. rpc_sleep_on(&xprt->resend, task, NULL);
  194. else
  195. rpc_sleep_on(&xprt->sending, task, NULL);
  196. return 0;
  197. }
  198. EXPORT_SYMBOL_GPL(xprt_reserve_xprt);
  199. static void xprt_clear_locked(struct rpc_xprt *xprt)
  200. {
  201. xprt->snd_task = NULL;
  202. if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state) || xprt->shutdown) {
  203. smp_mb__before_clear_bit();
  204. clear_bit(XPRT_LOCKED, &xprt->state);
  205. smp_mb__after_clear_bit();
  206. } else
  207. queue_work(rpciod_workqueue, &xprt->task_cleanup);
  208. }
  209. /*
  210. * xprt_reserve_xprt_cong - serialize write access to transports
  211. * @task: task that is requesting access to the transport
  212. *
  213. * Same as xprt_reserve_xprt, but Van Jacobson congestion control is
  214. * integrated into the decision of whether a request is allowed to be
  215. * woken up and given access to the transport.
  216. */
  217. int xprt_reserve_xprt_cong(struct rpc_task *task)
  218. {
  219. struct rpc_xprt *xprt = task->tk_xprt;
  220. struct rpc_rqst *req = task->tk_rqstp;
  221. if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
  222. if (task == xprt->snd_task)
  223. return 1;
  224. goto out_sleep;
  225. }
  226. if (__xprt_get_cong(xprt, task)) {
  227. xprt->snd_task = task;
  228. if (req) {
  229. req->rq_bytes_sent = 0;
  230. req->rq_ntrans++;
  231. }
  232. return 1;
  233. }
  234. xprt_clear_locked(xprt);
  235. out_sleep:
  236. dprintk("RPC: %5u failed to lock transport %p\n", task->tk_pid, xprt);
  237. task->tk_timeout = 0;
  238. task->tk_status = -EAGAIN;
  239. if (req && req->rq_ntrans)
  240. rpc_sleep_on(&xprt->resend, task, NULL);
  241. else
  242. rpc_sleep_on(&xprt->sending, task, NULL);
  243. return 0;
  244. }
  245. EXPORT_SYMBOL_GPL(xprt_reserve_xprt_cong);
  246. static inline int xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
  247. {
  248. int retval;
  249. spin_lock_bh(&xprt->transport_lock);
  250. retval = xprt->ops->reserve_xprt(task);
  251. spin_unlock_bh(&xprt->transport_lock);
  252. return retval;
  253. }
  254. static void __xprt_lock_write_next(struct rpc_xprt *xprt)
  255. {
  256. struct rpc_task *task;
  257. struct rpc_rqst *req;
  258. if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
  259. return;
  260. task = rpc_wake_up_next(&xprt->resend);
  261. if (!task) {
  262. task = rpc_wake_up_next(&xprt->sending);
  263. if (!task)
  264. goto out_unlock;
  265. }
  266. req = task->tk_rqstp;
  267. xprt->snd_task = task;
  268. if (req) {
  269. req->rq_bytes_sent = 0;
  270. req->rq_ntrans++;
  271. }
  272. return;
  273. out_unlock:
  274. xprt_clear_locked(xprt);
  275. }
  276. static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt)
  277. {
  278. struct rpc_task *task;
  279. if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
  280. return;
  281. if (RPCXPRT_CONGESTED(xprt))
  282. goto out_unlock;
  283. task = rpc_wake_up_next(&xprt->resend);
  284. if (!task) {
  285. task = rpc_wake_up_next(&xprt->sending);
  286. if (!task)
  287. goto out_unlock;
  288. }
  289. if (__xprt_get_cong(xprt, task)) {
  290. struct rpc_rqst *req = task->tk_rqstp;
  291. xprt->snd_task = task;
  292. if (req) {
  293. req->rq_bytes_sent = 0;
  294. req->rq_ntrans++;
  295. }
  296. return;
  297. }
  298. out_unlock:
  299. xprt_clear_locked(xprt);
  300. }
  301. /**
  302. * xprt_release_xprt - allow other requests to use a transport
  303. * @xprt: transport with other tasks potentially waiting
  304. * @task: task that is releasing access to the transport
  305. *
  306. * Note that "task" can be NULL. No congestion control is provided.
  307. */
  308. void xprt_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
  309. {
  310. if (xprt->snd_task == task) {
  311. xprt_clear_locked(xprt);
  312. __xprt_lock_write_next(xprt);
  313. }
  314. }
  315. EXPORT_SYMBOL_GPL(xprt_release_xprt);
  316. /**
  317. * xprt_release_xprt_cong - allow other requests to use a transport
  318. * @xprt: transport with other tasks potentially waiting
  319. * @task: task that is releasing access to the transport
  320. *
  321. * Note that "task" can be NULL. Another task is awoken to use the
  322. * transport if the transport's congestion window allows it.
  323. */
  324. void xprt_release_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task)
  325. {
  326. if (xprt->snd_task == task) {
  327. xprt_clear_locked(xprt);
  328. __xprt_lock_write_next_cong(xprt);
  329. }
  330. }
  331. EXPORT_SYMBOL_GPL(xprt_release_xprt_cong);
  332. static inline void xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task)
  333. {
  334. spin_lock_bh(&xprt->transport_lock);
  335. xprt->ops->release_xprt(xprt, task);
  336. spin_unlock_bh(&xprt->transport_lock);
  337. }
  338. /*
  339. * Van Jacobson congestion avoidance. Check if the congestion window
  340. * overflowed. Put the task to sleep if this is the case.
  341. */
  342. static int
  343. __xprt_get_cong(struct rpc_xprt *xprt, struct rpc_task *task)
  344. {
  345. struct rpc_rqst *req = task->tk_rqstp;
  346. if (req->rq_cong)
  347. return 1;
  348. dprintk("RPC: %5u xprt_cwnd_limited cong = %lu cwnd = %lu\n",
  349. task->tk_pid, xprt->cong, xprt->cwnd);
  350. if (RPCXPRT_CONGESTED(xprt))
  351. return 0;
  352. req->rq_cong = 1;
  353. xprt->cong += RPC_CWNDSCALE;
  354. return 1;
  355. }
  356. /*
  357. * Adjust the congestion window, and wake up the next task
  358. * that has been sleeping due to congestion
  359. */
  360. static void
  361. __xprt_put_cong(struct rpc_xprt *xprt, struct rpc_rqst *req)
  362. {
  363. if (!req->rq_cong)
  364. return;
  365. req->rq_cong = 0;
  366. xprt->cong -= RPC_CWNDSCALE;
  367. __xprt_lock_write_next_cong(xprt);
  368. }
  369. /**
  370. * xprt_release_rqst_cong - housekeeping when request is complete
  371. * @task: RPC request that recently completed
  372. *
  373. * Useful for transports that require congestion control.
  374. */
  375. void xprt_release_rqst_cong(struct rpc_task *task)
  376. {
  377. __xprt_put_cong(task->tk_xprt, task->tk_rqstp);
  378. }
  379. EXPORT_SYMBOL_GPL(xprt_release_rqst_cong);
  380. /**
  381. * xprt_adjust_cwnd - adjust transport congestion window
  382. * @task: recently completed RPC request used to adjust window
  383. * @result: result code of completed RPC request
  384. *
  385. * We use a time-smoothed congestion estimator to avoid heavy oscillation.
  386. */
  387. void xprt_adjust_cwnd(struct rpc_task *task, int result)
  388. {
  389. struct rpc_rqst *req = task->tk_rqstp;
  390. struct rpc_xprt *xprt = task->tk_xprt;
  391. unsigned long cwnd = xprt->cwnd;
  392. if (result >= 0 && cwnd <= xprt->cong) {
  393. /* The (cwnd >> 1) term makes sure
  394. * the result gets rounded properly. */
  395. cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd;
  396. if (cwnd > RPC_MAXCWND(xprt))
  397. cwnd = RPC_MAXCWND(xprt);
  398. __xprt_lock_write_next_cong(xprt);
  399. } else if (result == -ETIMEDOUT) {
  400. cwnd >>= 1;
  401. if (cwnd < RPC_CWNDSCALE)
  402. cwnd = RPC_CWNDSCALE;
  403. }
  404. dprintk("RPC: cong %ld, cwnd was %ld, now %ld\n",
  405. xprt->cong, xprt->cwnd, cwnd);
  406. xprt->cwnd = cwnd;
  407. __xprt_put_cong(xprt, req);
  408. }
  409. EXPORT_SYMBOL_GPL(xprt_adjust_cwnd);
  410. /**
  411. * xprt_wake_pending_tasks - wake all tasks on a transport's pending queue
  412. * @xprt: transport with waiting tasks
  413. * @status: result code to plant in each task before waking it
  414. *
  415. */
  416. void xprt_wake_pending_tasks(struct rpc_xprt *xprt, int status)
  417. {
  418. if (status < 0)
  419. rpc_wake_up_status(&xprt->pending, status);
  420. else
  421. rpc_wake_up(&xprt->pending);
  422. }
  423. EXPORT_SYMBOL_GPL(xprt_wake_pending_tasks);
  424. /**
  425. * xprt_wait_for_buffer_space - wait for transport output buffer to clear
  426. * @task: task to be put to sleep
  427. * @action: function pointer to be executed after wait
  428. */
  429. void xprt_wait_for_buffer_space(struct rpc_task *task, rpc_action action)
  430. {
  431. struct rpc_rqst *req = task->tk_rqstp;
  432. struct rpc_xprt *xprt = req->rq_xprt;
  433. task->tk_timeout = req->rq_timeout;
  434. rpc_sleep_on(&xprt->pending, task, action);
  435. }
  436. EXPORT_SYMBOL_GPL(xprt_wait_for_buffer_space);
  437. /**
  438. * xprt_write_space - wake the task waiting for transport output buffer space
  439. * @xprt: transport with waiting tasks
  440. *
  441. * Can be called in a soft IRQ context, so xprt_write_space never sleeps.
  442. */
  443. void xprt_write_space(struct rpc_xprt *xprt)
  444. {
  445. if (unlikely(xprt->shutdown))
  446. return;
  447. spin_lock_bh(&xprt->transport_lock);
  448. if (xprt->snd_task) {
  449. dprintk("RPC: write space: waking waiting task on "
  450. "xprt %p\n", xprt);
  451. rpc_wake_up_queued_task(&xprt->pending, xprt->snd_task);
  452. }
  453. spin_unlock_bh(&xprt->transport_lock);
  454. }
  455. EXPORT_SYMBOL_GPL(xprt_write_space);
  456. /**
  457. * xprt_set_retrans_timeout_def - set a request's retransmit timeout
  458. * @task: task whose timeout is to be set
  459. *
  460. * Set a request's retransmit timeout based on the transport's
  461. * default timeout parameters. Used by transports that don't adjust
  462. * the retransmit timeout based on round-trip time estimation.
  463. */
  464. void xprt_set_retrans_timeout_def(struct rpc_task *task)
  465. {
  466. task->tk_timeout = task->tk_rqstp->rq_timeout;
  467. }
  468. EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_def);
  469. /*
  470. * xprt_set_retrans_timeout_rtt - set a request's retransmit timeout
  471. * @task: task whose timeout is to be set
  472. *
  473. * Set a request's retransmit timeout using the RTT estimator.
  474. */
  475. void xprt_set_retrans_timeout_rtt(struct rpc_task *task)
  476. {
  477. int timer = task->tk_msg.rpc_proc->p_timer;
  478. struct rpc_clnt *clnt = task->tk_client;
  479. struct rpc_rtt *rtt = clnt->cl_rtt;
  480. struct rpc_rqst *req = task->tk_rqstp;
  481. unsigned long max_timeout = clnt->cl_timeout->to_maxval;
  482. task->tk_timeout = rpc_calc_rto(rtt, timer);
  483. task->tk_timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries;
  484. if (task->tk_timeout > max_timeout || task->tk_timeout == 0)
  485. task->tk_timeout = max_timeout;
  486. }
  487. EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_rtt);
  488. static void xprt_reset_majortimeo(struct rpc_rqst *req)
  489. {
  490. const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout;
  491. req->rq_majortimeo = req->rq_timeout;
  492. if (to->to_exponential)
  493. req->rq_majortimeo <<= to->to_retries;
  494. else
  495. req->rq_majortimeo += to->to_increment * to->to_retries;
  496. if (req->rq_majortimeo > to->to_maxval || req->rq_majortimeo == 0)
  497. req->rq_majortimeo = to->to_maxval;
  498. req->rq_majortimeo += jiffies;
  499. }
  500. /**
  501. * xprt_adjust_timeout - adjust timeout values for next retransmit
  502. * @req: RPC request containing parameters to use for the adjustment
  503. *
  504. */
  505. int xprt_adjust_timeout(struct rpc_rqst *req)
  506. {
  507. struct rpc_xprt *xprt = req->rq_xprt;
  508. const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout;
  509. int status = 0;
  510. if (time_before(jiffies, req->rq_majortimeo)) {
  511. if (to->to_exponential)
  512. req->rq_timeout <<= 1;
  513. else
  514. req->rq_timeout += to->to_increment;
  515. if (to->to_maxval && req->rq_timeout >= to->to_maxval)
  516. req->rq_timeout = to->to_maxval;
  517. req->rq_retries++;
  518. } else {
  519. req->rq_timeout = to->to_initval;
  520. req->rq_retries = 0;
  521. xprt_reset_majortimeo(req);
  522. /* Reset the RTT counters == "slow start" */
  523. spin_lock_bh(&xprt->transport_lock);
  524. rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval);
  525. spin_unlock_bh(&xprt->transport_lock);
  526. status = -ETIMEDOUT;
  527. }
  528. if (req->rq_timeout == 0) {
  529. printk(KERN_WARNING "xprt_adjust_timeout: rq_timeout = 0!\n");
  530. req->rq_timeout = 5 * HZ;
  531. }
  532. return status;
  533. }
  534. static void xprt_autoclose(struct work_struct *work)
  535. {
  536. struct rpc_xprt *xprt =
  537. container_of(work, struct rpc_xprt, task_cleanup);
  538. xprt->ops->close(xprt);
  539. clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
  540. xprt_release_write(xprt, NULL);
  541. }
  542. /**
  543. * xprt_disconnect_done - mark a transport as disconnected
  544. * @xprt: transport to flag for disconnect
  545. *
  546. */
  547. void xprt_disconnect_done(struct rpc_xprt *xprt)
  548. {
  549. dprintk("RPC: disconnected transport %p\n", xprt);
  550. spin_lock_bh(&xprt->transport_lock);
  551. xprt_clear_connected(xprt);
  552. xprt_wake_pending_tasks(xprt, -EAGAIN);
  553. spin_unlock_bh(&xprt->transport_lock);
  554. }
  555. EXPORT_SYMBOL_GPL(xprt_disconnect_done);
  556. /**
  557. * xprt_force_disconnect - force a transport to disconnect
  558. * @xprt: transport to disconnect
  559. *
  560. */
  561. void xprt_force_disconnect(struct rpc_xprt *xprt)
  562. {
  563. /* Don't race with the test_bit() in xprt_clear_locked() */
  564. spin_lock_bh(&xprt->transport_lock);
  565. set_bit(XPRT_CLOSE_WAIT, &xprt->state);
  566. /* Try to schedule an autoclose RPC call */
  567. if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
  568. queue_work(rpciod_workqueue, &xprt->task_cleanup);
  569. xprt_wake_pending_tasks(xprt, -EAGAIN);
  570. spin_unlock_bh(&xprt->transport_lock);
  571. }
  572. /**
  573. * xprt_conditional_disconnect - force a transport to disconnect
  574. * @xprt: transport to disconnect
  575. * @cookie: 'connection cookie'
  576. *
  577. * This attempts to break the connection if and only if 'cookie' matches
  578. * the current transport 'connection cookie'. It ensures that we don't
  579. * try to break the connection more than once when we need to retransmit
  580. * a batch of RPC requests.
  581. *
  582. */
  583. void xprt_conditional_disconnect(struct rpc_xprt *xprt, unsigned int cookie)
  584. {
  585. /* Don't race with the test_bit() in xprt_clear_locked() */
  586. spin_lock_bh(&xprt->transport_lock);
  587. if (cookie != xprt->connect_cookie)
  588. goto out;
  589. if (test_bit(XPRT_CLOSING, &xprt->state) || !xprt_connected(xprt))
  590. goto out;
  591. set_bit(XPRT_CLOSE_WAIT, &xprt->state);
  592. /* Try to schedule an autoclose RPC call */
  593. if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
  594. queue_work(rpciod_workqueue, &xprt->task_cleanup);
  595. xprt_wake_pending_tasks(xprt, -EAGAIN);
  596. out:
  597. spin_unlock_bh(&xprt->transport_lock);
  598. }
  599. static void
  600. xprt_init_autodisconnect(unsigned long data)
  601. {
  602. struct rpc_xprt *xprt = (struct rpc_xprt *)data;
  603. spin_lock(&xprt->transport_lock);
  604. if (!list_empty(&xprt->recv) || xprt->shutdown)
  605. goto out_abort;
  606. if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
  607. goto out_abort;
  608. spin_unlock(&xprt->transport_lock);
  609. set_bit(XPRT_CONNECTION_CLOSE, &xprt->state);
  610. queue_work(rpciod_workqueue, &xprt->task_cleanup);
  611. return;
  612. out_abort:
  613. spin_unlock(&xprt->transport_lock);
  614. }
  615. /**
  616. * xprt_connect - schedule a transport connect operation
  617. * @task: RPC task that is requesting the connect
  618. *
  619. */
  620. void xprt_connect(struct rpc_task *task)
  621. {
  622. struct rpc_xprt *xprt = task->tk_xprt;
  623. dprintk("RPC: %5u xprt_connect xprt %p %s connected\n", task->tk_pid,
  624. xprt, (xprt_connected(xprt) ? "is" : "is not"));
  625. if (!xprt_bound(xprt)) {
  626. task->tk_status = -EAGAIN;
  627. return;
  628. }
  629. if (!xprt_lock_write(xprt, task))
  630. return;
  631. if (test_and_clear_bit(XPRT_CLOSE_WAIT, &xprt->state))
  632. xprt->ops->close(xprt);
  633. if (xprt_connected(xprt))
  634. xprt_release_write(xprt, task);
  635. else {
  636. if (task->tk_rqstp)
  637. task->tk_rqstp->rq_bytes_sent = 0;
  638. task->tk_timeout = task->tk_rqstp->rq_timeout;
  639. rpc_sleep_on(&xprt->pending, task, xprt_connect_status);
  640. if (test_bit(XPRT_CLOSING, &xprt->state))
  641. return;
  642. if (xprt_test_and_set_connecting(xprt))
  643. return;
  644. xprt->stat.connect_start = jiffies;
  645. xprt->ops->connect(task);
  646. }
  647. }
  648. static void xprt_connect_status(struct rpc_task *task)
  649. {
  650. struct rpc_xprt *xprt = task->tk_xprt;
  651. if (task->tk_status == 0) {
  652. xprt->stat.connect_count++;
  653. xprt->stat.connect_time += (long)jiffies - xprt->stat.connect_start;
  654. dprintk("RPC: %5u xprt_connect_status: connection established\n",
  655. task->tk_pid);
  656. return;
  657. }
  658. switch (task->tk_status) {
  659. case -EAGAIN:
  660. dprintk("RPC: %5u xprt_connect_status: retrying\n", task->tk_pid);
  661. break;
  662. case -ETIMEDOUT:
  663. dprintk("RPC: %5u xprt_connect_status: connect attempt timed "
  664. "out\n", task->tk_pid);
  665. break;
  666. default:
  667. dprintk("RPC: %5u xprt_connect_status: error %d connecting to "
  668. "server %s\n", task->tk_pid, -task->tk_status,
  669. task->tk_client->cl_server);
  670. xprt_release_write(xprt, task);
  671. task->tk_status = -EIO;
  672. }
  673. }
  674. /**
  675. * xprt_lookup_rqst - find an RPC request corresponding to an XID
  676. * @xprt: transport on which the original request was transmitted
  677. * @xid: RPC XID of incoming reply
  678. *
  679. */
  680. struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, __be32 xid)
  681. {
  682. struct rpc_rqst *entry;
  683. list_for_each_entry(entry, &xprt->recv, rq_list)
  684. if (entry->rq_xid == xid)
  685. return entry;
  686. dprintk("RPC: xprt_lookup_rqst did not find xid %08x\n",
  687. ntohl(xid));
  688. xprt->stat.bad_xids++;
  689. return NULL;
  690. }
  691. EXPORT_SYMBOL_GPL(xprt_lookup_rqst);
  692. static void xprt_update_rtt(struct rpc_task *task)
  693. {
  694. struct rpc_rqst *req = task->tk_rqstp;
  695. struct rpc_rtt *rtt = task->tk_client->cl_rtt;
  696. unsigned timer = task->tk_msg.rpc_proc->p_timer;
  697. long m = usecs_to_jiffies(ktime_to_us(req->rq_rtt));
  698. if (timer) {
  699. if (req->rq_ntrans == 1)
  700. rpc_update_rtt(rtt, timer, m);
  701. rpc_set_timeo(rtt, timer, req->rq_ntrans - 1);
  702. }
  703. }
  704. /**
  705. * xprt_complete_rqst - called when reply processing is complete
  706. * @task: RPC request that recently completed
  707. * @copied: actual number of bytes received from the transport
  708. *
  709. * Caller holds transport lock.
  710. */
  711. void xprt_complete_rqst(struct rpc_task *task, int copied)
  712. {
  713. struct rpc_rqst *req = task->tk_rqstp;
  714. struct rpc_xprt *xprt = req->rq_xprt;
  715. dprintk("RPC: %5u xid %08x complete (%d bytes received)\n",
  716. task->tk_pid, ntohl(req->rq_xid), copied);
  717. xprt->stat.recvs++;
  718. req->rq_rtt = ktime_sub(ktime_get(), req->rq_xtime);
  719. if (xprt->ops->timer != NULL)
  720. xprt_update_rtt(task);
  721. list_del_init(&req->rq_list);
  722. req->rq_private_buf.len = copied;
  723. /* Ensure all writes are done before we update */
  724. /* req->rq_reply_bytes_recvd */
  725. smp_wmb();
  726. req->rq_reply_bytes_recvd = copied;
  727. rpc_wake_up_queued_task(&xprt->pending, task);
  728. }
  729. EXPORT_SYMBOL_GPL(xprt_complete_rqst);
  730. static void xprt_timer(struct rpc_task *task)
  731. {
  732. struct rpc_rqst *req = task->tk_rqstp;
  733. struct rpc_xprt *xprt = req->rq_xprt;
  734. if (task->tk_status != -ETIMEDOUT)
  735. return;
  736. dprintk("RPC: %5u xprt_timer\n", task->tk_pid);
  737. spin_lock_bh(&xprt->transport_lock);
  738. if (!req->rq_reply_bytes_recvd) {
  739. if (xprt->ops->timer)
  740. xprt->ops->timer(task);
  741. } else
  742. task->tk_status = 0;
  743. spin_unlock_bh(&xprt->transport_lock);
  744. }
  745. static inline int xprt_has_timer(struct rpc_xprt *xprt)
  746. {
  747. return xprt->idle_timeout != 0;
  748. }
  749. /**
  750. * xprt_prepare_transmit - reserve the transport before sending a request
  751. * @task: RPC task about to send a request
  752. *
  753. */
  754. int xprt_prepare_transmit(struct rpc_task *task)
  755. {
  756. struct rpc_rqst *req = task->tk_rqstp;
  757. struct rpc_xprt *xprt = req->rq_xprt;
  758. int err = 0;
  759. dprintk("RPC: %5u xprt_prepare_transmit\n", task->tk_pid);
  760. spin_lock_bh(&xprt->transport_lock);
  761. if (req->rq_reply_bytes_recvd && !req->rq_bytes_sent) {
  762. err = req->rq_reply_bytes_recvd;
  763. goto out_unlock;
  764. }
  765. if (!xprt->ops->reserve_xprt(task))
  766. err = -EAGAIN;
  767. out_unlock:
  768. spin_unlock_bh(&xprt->transport_lock);
  769. return err;
  770. }
  771. void xprt_end_transmit(struct rpc_task *task)
  772. {
  773. xprt_release_write(task->tk_rqstp->rq_xprt, task);
  774. }
  775. /**
  776. * xprt_transmit - send an RPC request on a transport
  777. * @task: controlling RPC task
  778. *
  779. * We have to copy the iovec because sendmsg fiddles with its contents.
  780. */
  781. void xprt_transmit(struct rpc_task *task)
  782. {
  783. struct rpc_rqst *req = task->tk_rqstp;
  784. struct rpc_xprt *xprt = req->rq_xprt;
  785. int status;
  786. dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
  787. if (!req->rq_reply_bytes_recvd) {
  788. if (list_empty(&req->rq_list) && rpc_reply_expected(task)) {
  789. /*
  790. * Add to the list only if we're expecting a reply
  791. */
  792. spin_lock_bh(&xprt->transport_lock);
  793. /* Update the softirq receive buffer */
  794. memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
  795. sizeof(req->rq_private_buf));
  796. /* Add request to the receive list */
  797. list_add_tail(&req->rq_list, &xprt->recv);
  798. spin_unlock_bh(&xprt->transport_lock);
  799. xprt_reset_majortimeo(req);
  800. /* Turn off autodisconnect */
  801. del_singleshot_timer_sync(&xprt->timer);
  802. }
  803. } else if (!req->rq_bytes_sent)
  804. return;
  805. req->rq_connect_cookie = xprt->connect_cookie;
  806. req->rq_xtime = ktime_get();
  807. status = xprt->ops->send_request(task);
  808. if (status != 0) {
  809. task->tk_status = status;
  810. return;
  811. }
  812. dprintk("RPC: %5u xmit complete\n", task->tk_pid);
  813. task->tk_flags |= RPC_TASK_SENT;
  814. spin_lock_bh(&xprt->transport_lock);
  815. xprt->ops->set_retrans_timeout(task);
  816. xprt->stat.sends++;
  817. xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs;
  818. xprt->stat.bklog_u += xprt->backlog.qlen;
  819. /* Don't race with disconnect */
  820. if (!xprt_connected(xprt))
  821. task->tk_status = -ENOTCONN;
  822. else if (!req->rq_reply_bytes_recvd && rpc_reply_expected(task)) {
  823. /*
  824. * Sleep on the pending queue since
  825. * we're expecting a reply.
  826. */
  827. rpc_sleep_on(&xprt->pending, task, xprt_timer);
  828. }
  829. spin_unlock_bh(&xprt->transport_lock);
  830. }
  831. static void xprt_alloc_slot(struct rpc_task *task)
  832. {
  833. struct rpc_xprt *xprt = task->tk_xprt;
  834. task->tk_status = 0;
  835. if (task->tk_rqstp)
  836. return;
  837. if (!list_empty(&xprt->free)) {
  838. struct rpc_rqst *req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
  839. list_del_init(&req->rq_list);
  840. task->tk_rqstp = req;
  841. xprt_request_init(task, xprt);
  842. return;
  843. }
  844. dprintk("RPC: waiting for request slot\n");
  845. task->tk_status = -EAGAIN;
  846. task->tk_timeout = 0;
  847. rpc_sleep_on(&xprt->backlog, task, NULL);
  848. }
  849. static void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
  850. {
  851. memset(req, 0, sizeof(*req)); /* mark unused */
  852. spin_lock(&xprt->reserve_lock);
  853. list_add(&req->rq_list, &xprt->free);
  854. rpc_wake_up_next(&xprt->backlog);
  855. spin_unlock(&xprt->reserve_lock);
  856. }
  857. struct rpc_xprt *xprt_alloc(struct net *net, int size, int max_req)
  858. {
  859. struct rpc_xprt *xprt;
  860. xprt = kzalloc(size, GFP_KERNEL);
  861. if (xprt == NULL)
  862. goto out;
  863. atomic_set(&xprt->count, 1);
  864. xprt->max_reqs = max_req;
  865. xprt->slot = kcalloc(max_req, sizeof(struct rpc_rqst), GFP_KERNEL);
  866. if (xprt->slot == NULL)
  867. goto out_free;
  868. xprt->xprt_net = get_net(net);
  869. return xprt;
  870. out_free:
  871. kfree(xprt);
  872. out:
  873. return NULL;
  874. }
  875. EXPORT_SYMBOL_GPL(xprt_alloc);
  876. void xprt_free(struct rpc_xprt *xprt)
  877. {
  878. put_net(xprt->xprt_net);
  879. kfree(xprt->slot);
  880. kfree(xprt);
  881. }
  882. EXPORT_SYMBOL_GPL(xprt_free);
  883. /**
  884. * xprt_reserve - allocate an RPC request slot
  885. * @task: RPC task requesting a slot allocation
  886. *
  887. * If no more slots are available, place the task on the transport's
  888. * backlog queue.
  889. */
  890. void xprt_reserve(struct rpc_task *task)
  891. {
  892. struct rpc_xprt *xprt = task->tk_xprt;
  893. task->tk_status = -EIO;
  894. spin_lock(&xprt->reserve_lock);
  895. xprt_alloc_slot(task);
  896. spin_unlock(&xprt->reserve_lock);
  897. }
  898. static inline __be32 xprt_alloc_xid(struct rpc_xprt *xprt)
  899. {
  900. return (__force __be32)xprt->xid++;
  901. }
  902. static inline void xprt_init_xid(struct rpc_xprt *xprt)
  903. {
  904. xprt->xid = net_random();
  905. }
  906. static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
  907. {
  908. struct rpc_rqst *req = task->tk_rqstp;
  909. req->rq_timeout = task->tk_client->cl_timeout->to_initval;
  910. req->rq_task = task;
  911. req->rq_xprt = xprt;
  912. req->rq_buffer = NULL;
  913. req->rq_xid = xprt_alloc_xid(xprt);
  914. req->rq_release_snd_buf = NULL;
  915. xprt_reset_majortimeo(req);
  916. dprintk("RPC: %5u reserved req %p xid %08x\n", task->tk_pid,
  917. req, ntohl(req->rq_xid));
  918. }
  919. /**
  920. * xprt_release - release an RPC request slot
  921. * @task: task which is finished with the slot
  922. *
  923. */
  924. void xprt_release(struct rpc_task *task)
  925. {
  926. struct rpc_xprt *xprt;
  927. struct rpc_rqst *req;
  928. if (!(req = task->tk_rqstp))
  929. return;
  930. xprt = req->rq_xprt;
  931. rpc_count_iostats(task);
  932. spin_lock_bh(&xprt->transport_lock);
  933. xprt->ops->release_xprt(xprt, task);
  934. if (xprt->ops->release_request)
  935. xprt->ops->release_request(task);
  936. if (!list_empty(&req->rq_list))
  937. list_del(&req->rq_list);
  938. xprt->last_used = jiffies;
  939. if (list_empty(&xprt->recv) && xprt_has_timer(xprt))
  940. mod_timer(&xprt->timer,
  941. xprt->last_used + xprt->idle_timeout);
  942. spin_unlock_bh(&xprt->transport_lock);
  943. if (req->rq_buffer)
  944. xprt->ops->buf_free(req->rq_buffer);
  945. if (req->rq_cred != NULL)
  946. put_rpccred(req->rq_cred);
  947. task->tk_rqstp = NULL;
  948. if (req->rq_release_snd_buf)
  949. req->rq_release_snd_buf(req);
  950. dprintk("RPC: %5u release request %p\n", task->tk_pid, req);
  951. if (likely(!bc_prealloc(req)))
  952. xprt_free_slot(xprt, req);
  953. else
  954. xprt_free_bc_request(req);
  955. }
  956. /**
  957. * xprt_create_transport - create an RPC transport
  958. * @args: rpc transport creation arguments
  959. *
  960. */
  961. struct rpc_xprt *xprt_create_transport(struct xprt_create *args)
  962. {
  963. struct rpc_xprt *xprt;
  964. struct rpc_rqst *req;
  965. struct xprt_class *t;
  966. spin_lock(&xprt_list_lock);
  967. list_for_each_entry(t, &xprt_list, list) {
  968. if (t->ident == args->ident) {
  969. spin_unlock(&xprt_list_lock);
  970. goto found;
  971. }
  972. }
  973. spin_unlock(&xprt_list_lock);
  974. printk(KERN_ERR "RPC: transport (%d) not supported\n", args->ident);
  975. return ERR_PTR(-EIO);
  976. found:
  977. xprt = t->setup(args);
  978. if (IS_ERR(xprt)) {
  979. dprintk("RPC: xprt_create_transport: failed, %ld\n",
  980. -PTR_ERR(xprt));
  981. return xprt;
  982. }
  983. if (test_and_set_bit(XPRT_INITIALIZED, &xprt->state))
  984. /* ->setup returned a pre-initialized xprt: */
  985. return xprt;
  986. spin_lock_init(&xprt->transport_lock);
  987. spin_lock_init(&xprt->reserve_lock);
  988. INIT_LIST_HEAD(&xprt->free);
  989. INIT_LIST_HEAD(&xprt->recv);
  990. #if defined(CONFIG_NFS_V4_1)
  991. spin_lock_init(&xprt->bc_pa_lock);
  992. INIT_LIST_HEAD(&xprt->bc_pa_list);
  993. #endif /* CONFIG_NFS_V4_1 */
  994. INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
  995. if (xprt_has_timer(xprt))
  996. setup_timer(&xprt->timer, xprt_init_autodisconnect,
  997. (unsigned long)xprt);
  998. else
  999. init_timer(&xprt->timer);
  1000. xprt->last_used = jiffies;
  1001. xprt->cwnd = RPC_INITCWND;
  1002. xprt->bind_index = 0;
  1003. rpc_init_wait_queue(&xprt->binding, "xprt_binding");
  1004. rpc_init_wait_queue(&xprt->pending, "xprt_pending");
  1005. rpc_init_wait_queue(&xprt->sending, "xprt_sending");
  1006. rpc_init_wait_queue(&xprt->resend, "xprt_resend");
  1007. rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog");
  1008. /* initialize free list */
  1009. for (req = &xprt->slot[xprt->max_reqs-1]; req >= &xprt->slot[0]; req--)
  1010. list_add(&req->rq_list, &xprt->free);
  1011. xprt_init_xid(xprt);
  1012. dprintk("RPC: created transport %p with %u slots\n", xprt,
  1013. xprt->max_reqs);
  1014. return xprt;
  1015. }
  1016. /**
  1017. * xprt_destroy - destroy an RPC transport, killing off all requests.
  1018. * @xprt: transport to destroy
  1019. *
  1020. */
  1021. static void xprt_destroy(struct rpc_xprt *xprt)
  1022. {
  1023. dprintk("RPC: destroying transport %p\n", xprt);
  1024. xprt->shutdown = 1;
  1025. del_timer_sync(&xprt->timer);
  1026. rpc_destroy_wait_queue(&xprt->binding);
  1027. rpc_destroy_wait_queue(&xprt->pending);
  1028. rpc_destroy_wait_queue(&xprt->sending);
  1029. rpc_destroy_wait_queue(&xprt->resend);
  1030. rpc_destroy_wait_queue(&xprt->backlog);
  1031. cancel_work_sync(&xprt->task_cleanup);
  1032. /*
  1033. * Tear down transport state and free the rpc_xprt
  1034. */
  1035. xprt->ops->destroy(xprt);
  1036. }
  1037. /**
  1038. * xprt_put - release a reference to an RPC transport.
  1039. * @xprt: pointer to the transport
  1040. *
  1041. */
  1042. void xprt_put(struct rpc_xprt *xprt)
  1043. {
  1044. if (atomic_dec_and_test(&xprt->count))
  1045. xprt_destroy(xprt);
  1046. }
  1047. /**
  1048. * xprt_get - return a reference to an RPC transport.
  1049. * @xprt: pointer to the transport
  1050. *
  1051. */
  1052. struct rpc_xprt *xprt_get(struct rpc_xprt *xprt)
  1053. {
  1054. if (atomic_inc_not_zero(&xprt->count))
  1055. return xprt;
  1056. return NULL;
  1057. }