xprt.c 33 KB

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