transport.c 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793
  1. /*
  2. * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
  3. *
  4. * This software is available to you under a choice of one of two
  5. * licenses. You may choose to be licensed under the terms of the GNU
  6. * General Public License (GPL) Version 2, available from the file
  7. * COPYING in the main directory of this source tree, or the BSD-type
  8. * license below:
  9. *
  10. * Redistribution and use in source and binary forms, with or without
  11. * modification, are permitted provided that the following conditions
  12. * are met:
  13. *
  14. * Redistributions of source code must retain the above copyright
  15. * notice, this list of conditions and the following disclaimer.
  16. *
  17. * Redistributions in binary form must reproduce the above
  18. * copyright notice, this list of conditions and the following
  19. * disclaimer in the documentation and/or other materials provided
  20. * with the distribution.
  21. *
  22. * Neither the name of the Network Appliance, Inc. nor the names of
  23. * its contributors may be used to endorse or promote products
  24. * derived from this software without specific prior written
  25. * permission.
  26. *
  27. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  28. * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  29. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  30. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  31. * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  32. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  33. * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  34. * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  35. * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  36. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  37. * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  38. */
  39. /*
  40. * transport.c
  41. *
  42. * This file contains the top-level implementation of an RPC RDMA
  43. * transport.
  44. *
  45. * Naming convention: functions beginning with xprt_ are part of the
  46. * transport switch. All others are RPC RDMA internal.
  47. */
  48. #include <linux/module.h>
  49. #include <linux/init.h>
  50. #include <linux/slab.h>
  51. #include <linux/seq_file.h>
  52. #include "xprt_rdma.h"
  53. #ifdef RPC_DEBUG
  54. # define RPCDBG_FACILITY RPCDBG_TRANS
  55. #endif
  56. MODULE_LICENSE("Dual BSD/GPL");
  57. MODULE_DESCRIPTION("RPC/RDMA Transport for Linux kernel NFS");
  58. MODULE_AUTHOR("Network Appliance, Inc.");
  59. /*
  60. * tunables
  61. */
  62. static unsigned int xprt_rdma_slot_table_entries = RPCRDMA_DEF_SLOT_TABLE;
  63. static unsigned int xprt_rdma_max_inline_read = RPCRDMA_DEF_INLINE;
  64. static unsigned int xprt_rdma_max_inline_write = RPCRDMA_DEF_INLINE;
  65. static unsigned int xprt_rdma_inline_write_padding;
  66. static unsigned int xprt_rdma_memreg_strategy = RPCRDMA_FRMR;
  67. int xprt_rdma_pad_optimize = 0;
  68. #ifdef RPC_DEBUG
  69. static unsigned int min_slot_table_size = RPCRDMA_MIN_SLOT_TABLE;
  70. static unsigned int max_slot_table_size = RPCRDMA_MAX_SLOT_TABLE;
  71. static unsigned int zero;
  72. static unsigned int max_padding = PAGE_SIZE;
  73. static unsigned int min_memreg = RPCRDMA_BOUNCEBUFFERS;
  74. static unsigned int max_memreg = RPCRDMA_LAST - 1;
  75. static struct ctl_table_header *sunrpc_table_header;
  76. static ctl_table xr_tunables_table[] = {
  77. {
  78. .procname = "rdma_slot_table_entries",
  79. .data = &xprt_rdma_slot_table_entries,
  80. .maxlen = sizeof(unsigned int),
  81. .mode = 0644,
  82. .proc_handler = proc_dointvec_minmax,
  83. .extra1 = &min_slot_table_size,
  84. .extra2 = &max_slot_table_size
  85. },
  86. {
  87. .procname = "rdma_max_inline_read",
  88. .data = &xprt_rdma_max_inline_read,
  89. .maxlen = sizeof(unsigned int),
  90. .mode = 0644,
  91. .proc_handler = proc_dointvec,
  92. },
  93. {
  94. .procname = "rdma_max_inline_write",
  95. .data = &xprt_rdma_max_inline_write,
  96. .maxlen = sizeof(unsigned int),
  97. .mode = 0644,
  98. .proc_handler = proc_dointvec,
  99. },
  100. {
  101. .procname = "rdma_inline_write_padding",
  102. .data = &xprt_rdma_inline_write_padding,
  103. .maxlen = sizeof(unsigned int),
  104. .mode = 0644,
  105. .proc_handler = proc_dointvec_minmax,
  106. .extra1 = &zero,
  107. .extra2 = &max_padding,
  108. },
  109. {
  110. .procname = "rdma_memreg_strategy",
  111. .data = &xprt_rdma_memreg_strategy,
  112. .maxlen = sizeof(unsigned int),
  113. .mode = 0644,
  114. .proc_handler = proc_dointvec_minmax,
  115. .extra1 = &min_memreg,
  116. .extra2 = &max_memreg,
  117. },
  118. {
  119. .procname = "rdma_pad_optimize",
  120. .data = &xprt_rdma_pad_optimize,
  121. .maxlen = sizeof(unsigned int),
  122. .mode = 0644,
  123. .proc_handler = proc_dointvec,
  124. },
  125. { },
  126. };
  127. static ctl_table sunrpc_table[] = {
  128. {
  129. .procname = "sunrpc",
  130. .mode = 0555,
  131. .child = xr_tunables_table
  132. },
  133. { },
  134. };
  135. #endif
  136. static struct rpc_xprt_ops xprt_rdma_procs; /* forward reference */
  137. static void
  138. xprt_rdma_format_addresses(struct rpc_xprt *xprt)
  139. {
  140. struct sockaddr *sap = (struct sockaddr *)
  141. &rpcx_to_rdmad(xprt).addr;
  142. struct sockaddr_in *sin = (struct sockaddr_in *)sap;
  143. char buf[64];
  144. (void)rpc_ntop(sap, buf, sizeof(buf));
  145. xprt->address_strings[RPC_DISPLAY_ADDR] = kstrdup(buf, GFP_KERNEL);
  146. snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
  147. xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
  148. xprt->address_strings[RPC_DISPLAY_PROTO] = "rdma";
  149. snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr));
  150. xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
  151. snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
  152. xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
  153. /* netid */
  154. xprt->address_strings[RPC_DISPLAY_NETID] = "rdma";
  155. }
  156. static void
  157. xprt_rdma_free_addresses(struct rpc_xprt *xprt)
  158. {
  159. unsigned int i;
  160. for (i = 0; i < RPC_DISPLAY_MAX; i++)
  161. switch (i) {
  162. case RPC_DISPLAY_PROTO:
  163. case RPC_DISPLAY_NETID:
  164. continue;
  165. default:
  166. kfree(xprt->address_strings[i]);
  167. }
  168. }
  169. static void
  170. xprt_rdma_connect_worker(struct work_struct *work)
  171. {
  172. struct rpcrdma_xprt *r_xprt =
  173. container_of(work, struct rpcrdma_xprt, rdma_connect.work);
  174. struct rpc_xprt *xprt = &r_xprt->xprt;
  175. int rc = 0;
  176. if (!xprt->shutdown) {
  177. xprt_clear_connected(xprt);
  178. dprintk("RPC: %s: %sconnect\n", __func__,
  179. r_xprt->rx_ep.rep_connected != 0 ? "re" : "");
  180. rc = rpcrdma_ep_connect(&r_xprt->rx_ep, &r_xprt->rx_ia);
  181. if (rc)
  182. goto out;
  183. }
  184. goto out_clear;
  185. out:
  186. xprt_wake_pending_tasks(xprt, rc);
  187. out_clear:
  188. dprintk("RPC: %s: exit\n", __func__);
  189. xprt_clear_connecting(xprt);
  190. }
  191. /*
  192. * xprt_rdma_destroy
  193. *
  194. * Destroy the xprt.
  195. * Free all memory associated with the object, including its own.
  196. * NOTE: none of the *destroy methods free memory for their top-level
  197. * objects, even though they may have allocated it (they do free
  198. * private memory). It's up to the caller to handle it. In this
  199. * case (RDMA transport), all structure memory is inlined with the
  200. * struct rpcrdma_xprt.
  201. */
  202. static void
  203. xprt_rdma_destroy(struct rpc_xprt *xprt)
  204. {
  205. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  206. int rc;
  207. dprintk("RPC: %s: called\n", __func__);
  208. cancel_delayed_work(&r_xprt->rdma_connect);
  209. flush_scheduled_work();
  210. xprt_clear_connected(xprt);
  211. rpcrdma_buffer_destroy(&r_xprt->rx_buf);
  212. rc = rpcrdma_ep_destroy(&r_xprt->rx_ep, &r_xprt->rx_ia);
  213. if (rc)
  214. dprintk("RPC: %s: rpcrdma_ep_destroy returned %i\n",
  215. __func__, rc);
  216. rpcrdma_ia_close(&r_xprt->rx_ia);
  217. xprt_rdma_free_addresses(xprt);
  218. kfree(xprt->slot);
  219. xprt->slot = NULL;
  220. kfree(xprt);
  221. dprintk("RPC: %s: returning\n", __func__);
  222. module_put(THIS_MODULE);
  223. }
  224. static const struct rpc_timeout xprt_rdma_default_timeout = {
  225. .to_initval = 60 * HZ,
  226. .to_maxval = 60 * HZ,
  227. };
  228. /**
  229. * xprt_setup_rdma - Set up transport to use RDMA
  230. *
  231. * @args: rpc transport arguments
  232. */
  233. static struct rpc_xprt *
  234. xprt_setup_rdma(struct xprt_create *args)
  235. {
  236. struct rpcrdma_create_data_internal cdata;
  237. struct rpc_xprt *xprt;
  238. struct rpcrdma_xprt *new_xprt;
  239. struct rpcrdma_ep *new_ep;
  240. struct sockaddr_in *sin;
  241. int rc;
  242. if (args->addrlen > sizeof(xprt->addr)) {
  243. dprintk("RPC: %s: address too large\n", __func__);
  244. return ERR_PTR(-EBADF);
  245. }
  246. xprt = kzalloc(sizeof(struct rpcrdma_xprt), GFP_KERNEL);
  247. if (xprt == NULL) {
  248. dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n",
  249. __func__);
  250. return ERR_PTR(-ENOMEM);
  251. }
  252. xprt->max_reqs = xprt_rdma_slot_table_entries;
  253. xprt->slot = kcalloc(xprt->max_reqs,
  254. sizeof(struct rpc_rqst), GFP_KERNEL);
  255. if (xprt->slot == NULL) {
  256. dprintk("RPC: %s: couldn't allocate %d slots\n",
  257. __func__, xprt->max_reqs);
  258. kfree(xprt);
  259. return ERR_PTR(-ENOMEM);
  260. }
  261. /* 60 second timeout, no retries */
  262. xprt->timeout = &xprt_rdma_default_timeout;
  263. xprt->bind_timeout = (60U * HZ);
  264. xprt->reestablish_timeout = (5U * HZ);
  265. xprt->idle_timeout = (5U * 60 * HZ);
  266. xprt->resvport = 0; /* privileged port not needed */
  267. xprt->tsh_size = 0; /* RPC-RDMA handles framing */
  268. xprt->max_payload = RPCRDMA_MAX_DATA_SEGS * PAGE_SIZE;
  269. xprt->ops = &xprt_rdma_procs;
  270. /*
  271. * Set up RDMA-specific connect data.
  272. */
  273. /* Put server RDMA address in local cdata */
  274. memcpy(&cdata.addr, args->dstaddr, args->addrlen);
  275. /* Ensure xprt->addr holds valid server TCP (not RDMA)
  276. * address, for any side protocols which peek at it */
  277. xprt->prot = IPPROTO_TCP;
  278. xprt->addrlen = args->addrlen;
  279. memcpy(&xprt->addr, &cdata.addr, xprt->addrlen);
  280. sin = (struct sockaddr_in *)&cdata.addr;
  281. if (ntohs(sin->sin_port) != 0)
  282. xprt_set_bound(xprt);
  283. dprintk("RPC: %s: %pI4:%u\n",
  284. __func__, &sin->sin_addr.s_addr, ntohs(sin->sin_port));
  285. /* Set max requests */
  286. cdata.max_requests = xprt->max_reqs;
  287. /* Set some length limits */
  288. cdata.rsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA write max */
  289. cdata.wsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA read max */
  290. cdata.inline_wsize = xprt_rdma_max_inline_write;
  291. if (cdata.inline_wsize > cdata.wsize)
  292. cdata.inline_wsize = cdata.wsize;
  293. cdata.inline_rsize = xprt_rdma_max_inline_read;
  294. if (cdata.inline_rsize > cdata.rsize)
  295. cdata.inline_rsize = cdata.rsize;
  296. cdata.padding = xprt_rdma_inline_write_padding;
  297. /*
  298. * Create new transport instance, which includes initialized
  299. * o ia
  300. * o endpoint
  301. * o buffers
  302. */
  303. new_xprt = rpcx_to_rdmax(xprt);
  304. rc = rpcrdma_ia_open(new_xprt, (struct sockaddr *) &cdata.addr,
  305. xprt_rdma_memreg_strategy);
  306. if (rc)
  307. goto out1;
  308. /*
  309. * initialize and create ep
  310. */
  311. new_xprt->rx_data = cdata;
  312. new_ep = &new_xprt->rx_ep;
  313. new_ep->rep_remote_addr = cdata.addr;
  314. rc = rpcrdma_ep_create(&new_xprt->rx_ep,
  315. &new_xprt->rx_ia, &new_xprt->rx_data);
  316. if (rc)
  317. goto out2;
  318. /*
  319. * Allocate pre-registered send and receive buffers for headers and
  320. * any inline data. Also specify any padding which will be provided
  321. * from a preregistered zero buffer.
  322. */
  323. rc = rpcrdma_buffer_create(&new_xprt->rx_buf, new_ep, &new_xprt->rx_ia,
  324. &new_xprt->rx_data);
  325. if (rc)
  326. goto out3;
  327. /*
  328. * Register a callback for connection events. This is necessary because
  329. * connection loss notification is async. We also catch connection loss
  330. * when reaping receives.
  331. */
  332. INIT_DELAYED_WORK(&new_xprt->rdma_connect, xprt_rdma_connect_worker);
  333. new_ep->rep_func = rpcrdma_conn_func;
  334. new_ep->rep_xprt = xprt;
  335. xprt_rdma_format_addresses(xprt);
  336. if (!try_module_get(THIS_MODULE))
  337. goto out4;
  338. return xprt;
  339. out4:
  340. xprt_rdma_free_addresses(xprt);
  341. rc = -EINVAL;
  342. out3:
  343. (void) rpcrdma_ep_destroy(new_ep, &new_xprt->rx_ia);
  344. out2:
  345. rpcrdma_ia_close(&new_xprt->rx_ia);
  346. out1:
  347. kfree(xprt->slot);
  348. kfree(xprt);
  349. return ERR_PTR(rc);
  350. }
  351. /*
  352. * Close a connection, during shutdown or timeout/reconnect
  353. */
  354. static void
  355. xprt_rdma_close(struct rpc_xprt *xprt)
  356. {
  357. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  358. dprintk("RPC: %s: closing\n", __func__);
  359. if (r_xprt->rx_ep.rep_connected > 0)
  360. xprt->reestablish_timeout = 0;
  361. xprt_disconnect_done(xprt);
  362. (void) rpcrdma_ep_disconnect(&r_xprt->rx_ep, &r_xprt->rx_ia);
  363. }
  364. static void
  365. xprt_rdma_set_port(struct rpc_xprt *xprt, u16 port)
  366. {
  367. struct sockaddr_in *sap;
  368. sap = (struct sockaddr_in *)&xprt->addr;
  369. sap->sin_port = htons(port);
  370. sap = (struct sockaddr_in *)&rpcx_to_rdmad(xprt).addr;
  371. sap->sin_port = htons(port);
  372. dprintk("RPC: %s: %u\n", __func__, port);
  373. }
  374. static void
  375. xprt_rdma_connect(struct rpc_task *task)
  376. {
  377. struct rpc_xprt *xprt = (struct rpc_xprt *)task->tk_xprt;
  378. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  379. if (r_xprt->rx_ep.rep_connected != 0) {
  380. /* Reconnect */
  381. schedule_delayed_work(&r_xprt->rdma_connect,
  382. xprt->reestablish_timeout);
  383. xprt->reestablish_timeout <<= 1;
  384. if (xprt->reestablish_timeout > (30 * HZ))
  385. xprt->reestablish_timeout = (30 * HZ);
  386. else if (xprt->reestablish_timeout < (5 * HZ))
  387. xprt->reestablish_timeout = (5 * HZ);
  388. } else {
  389. schedule_delayed_work(&r_xprt->rdma_connect, 0);
  390. if (!RPC_IS_ASYNC(task))
  391. flush_scheduled_work();
  392. }
  393. }
  394. static int
  395. xprt_rdma_reserve_xprt(struct rpc_task *task)
  396. {
  397. struct rpc_xprt *xprt = task->tk_xprt;
  398. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  399. int credits = atomic_read(&r_xprt->rx_buf.rb_credits);
  400. /* == RPC_CWNDSCALE @ init, but *after* setup */
  401. if (r_xprt->rx_buf.rb_cwndscale == 0UL) {
  402. r_xprt->rx_buf.rb_cwndscale = xprt->cwnd;
  403. dprintk("RPC: %s: cwndscale %lu\n", __func__,
  404. r_xprt->rx_buf.rb_cwndscale);
  405. BUG_ON(r_xprt->rx_buf.rb_cwndscale <= 0);
  406. }
  407. xprt->cwnd = credits * r_xprt->rx_buf.rb_cwndscale;
  408. return xprt_reserve_xprt_cong(task);
  409. }
  410. /*
  411. * The RDMA allocate/free functions need the task structure as a place
  412. * to hide the struct rpcrdma_req, which is necessary for the actual send/recv
  413. * sequence. For this reason, the recv buffers are attached to send
  414. * buffers for portions of the RPC. Note that the RPC layer allocates
  415. * both send and receive buffers in the same call. We may register
  416. * the receive buffer portion when using reply chunks.
  417. */
  418. static void *
  419. xprt_rdma_allocate(struct rpc_task *task, size_t size)
  420. {
  421. struct rpc_xprt *xprt = task->tk_xprt;
  422. struct rpcrdma_req *req, *nreq;
  423. req = rpcrdma_buffer_get(&rpcx_to_rdmax(xprt)->rx_buf);
  424. BUG_ON(NULL == req);
  425. if (size > req->rl_size) {
  426. dprintk("RPC: %s: size %zd too large for buffer[%zd]: "
  427. "prog %d vers %d proc %d\n",
  428. __func__, size, req->rl_size,
  429. task->tk_client->cl_prog, task->tk_client->cl_vers,
  430. task->tk_msg.rpc_proc->p_proc);
  431. /*
  432. * Outgoing length shortage. Our inline write max must have
  433. * been configured to perform direct i/o.
  434. *
  435. * This is therefore a large metadata operation, and the
  436. * allocate call was made on the maximum possible message,
  437. * e.g. containing long filename(s) or symlink data. In
  438. * fact, while these metadata operations *might* carry
  439. * large outgoing payloads, they rarely *do*. However, we
  440. * have to commit to the request here, so reallocate and
  441. * register it now. The data path will never require this
  442. * reallocation.
  443. *
  444. * If the allocation or registration fails, the RPC framework
  445. * will (doggedly) retry.
  446. */
  447. if (rpcx_to_rdmax(xprt)->rx_ia.ri_memreg_strategy ==
  448. RPCRDMA_BOUNCEBUFFERS) {
  449. /* forced to "pure inline" */
  450. dprintk("RPC: %s: too much data (%zd) for inline "
  451. "(r/w max %d/%d)\n", __func__, size,
  452. rpcx_to_rdmad(xprt).inline_rsize,
  453. rpcx_to_rdmad(xprt).inline_wsize);
  454. size = req->rl_size;
  455. rpc_exit(task, -EIO); /* fail the operation */
  456. rpcx_to_rdmax(xprt)->rx_stats.failed_marshal_count++;
  457. goto out;
  458. }
  459. if (task->tk_flags & RPC_TASK_SWAPPER)
  460. nreq = kmalloc(sizeof *req + size, GFP_ATOMIC);
  461. else
  462. nreq = kmalloc(sizeof *req + size, GFP_NOFS);
  463. if (nreq == NULL)
  464. goto outfail;
  465. if (rpcrdma_register_internal(&rpcx_to_rdmax(xprt)->rx_ia,
  466. nreq->rl_base, size + sizeof(struct rpcrdma_req)
  467. - offsetof(struct rpcrdma_req, rl_base),
  468. &nreq->rl_handle, &nreq->rl_iov)) {
  469. kfree(nreq);
  470. goto outfail;
  471. }
  472. rpcx_to_rdmax(xprt)->rx_stats.hardway_register_count += size;
  473. nreq->rl_size = size;
  474. nreq->rl_niovs = 0;
  475. nreq->rl_nchunks = 0;
  476. nreq->rl_buffer = (struct rpcrdma_buffer *)req;
  477. nreq->rl_reply = req->rl_reply;
  478. memcpy(nreq->rl_segments,
  479. req->rl_segments, sizeof nreq->rl_segments);
  480. /* flag the swap with an unused field */
  481. nreq->rl_iov.length = 0;
  482. req->rl_reply = NULL;
  483. req = nreq;
  484. }
  485. dprintk("RPC: %s: size %zd, request 0x%p\n", __func__, size, req);
  486. out:
  487. req->rl_connect_cookie = 0; /* our reserved value */
  488. return req->rl_xdr_buf;
  489. outfail:
  490. rpcrdma_buffer_put(req);
  491. rpcx_to_rdmax(xprt)->rx_stats.failed_marshal_count++;
  492. return NULL;
  493. }
  494. /*
  495. * This function returns all RDMA resources to the pool.
  496. */
  497. static void
  498. xprt_rdma_free(void *buffer)
  499. {
  500. struct rpcrdma_req *req;
  501. struct rpcrdma_xprt *r_xprt;
  502. struct rpcrdma_rep *rep;
  503. int i;
  504. if (buffer == NULL)
  505. return;
  506. req = container_of(buffer, struct rpcrdma_req, rl_xdr_buf[0]);
  507. if (req->rl_iov.length == 0) { /* see allocate above */
  508. r_xprt = container_of(((struct rpcrdma_req *) req->rl_buffer)->rl_buffer,
  509. struct rpcrdma_xprt, rx_buf);
  510. } else
  511. r_xprt = container_of(req->rl_buffer, struct rpcrdma_xprt, rx_buf);
  512. rep = req->rl_reply;
  513. dprintk("RPC: %s: called on 0x%p%s\n",
  514. __func__, rep, (rep && rep->rr_func) ? " (with waiter)" : "");
  515. /*
  516. * Finish the deregistration. When using mw bind, this was
  517. * begun in rpcrdma_reply_handler(). In all other modes, we
  518. * do it here, in thread context. The process is considered
  519. * complete when the rr_func vector becomes NULL - this
  520. * was put in place during rpcrdma_reply_handler() - the wait
  521. * call below will not block if the dereg is "done". If
  522. * interrupted, our framework will clean up.
  523. */
  524. for (i = 0; req->rl_nchunks;) {
  525. --req->rl_nchunks;
  526. i += rpcrdma_deregister_external(
  527. &req->rl_segments[i], r_xprt, NULL);
  528. }
  529. if (rep && wait_event_interruptible(rep->rr_unbind, !rep->rr_func)) {
  530. rep->rr_func = NULL; /* abandon the callback */
  531. req->rl_reply = NULL;
  532. }
  533. if (req->rl_iov.length == 0) { /* see allocate above */
  534. struct rpcrdma_req *oreq = (struct rpcrdma_req *)req->rl_buffer;
  535. oreq->rl_reply = req->rl_reply;
  536. (void) rpcrdma_deregister_internal(&r_xprt->rx_ia,
  537. req->rl_handle,
  538. &req->rl_iov);
  539. kfree(req);
  540. req = oreq;
  541. }
  542. /* Put back request+reply buffers */
  543. rpcrdma_buffer_put(req);
  544. }
  545. /*
  546. * send_request invokes the meat of RPC RDMA. It must do the following:
  547. * 1. Marshal the RPC request into an RPC RDMA request, which means
  548. * putting a header in front of data, and creating IOVs for RDMA
  549. * from those in the request.
  550. * 2. In marshaling, detect opportunities for RDMA, and use them.
  551. * 3. Post a recv message to set up asynch completion, then send
  552. * the request (rpcrdma_ep_post).
  553. * 4. No partial sends are possible in the RPC-RDMA protocol (as in UDP).
  554. */
  555. static int
  556. xprt_rdma_send_request(struct rpc_task *task)
  557. {
  558. struct rpc_rqst *rqst = task->tk_rqstp;
  559. struct rpc_xprt *xprt = task->tk_xprt;
  560. struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
  561. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  562. /* marshal the send itself */
  563. if (req->rl_niovs == 0 && rpcrdma_marshal_req(rqst) != 0) {
  564. r_xprt->rx_stats.failed_marshal_count++;
  565. dprintk("RPC: %s: rpcrdma_marshal_req failed\n",
  566. __func__);
  567. return -EIO;
  568. }
  569. if (req->rl_reply == NULL) /* e.g. reconnection */
  570. rpcrdma_recv_buffer_get(req);
  571. if (req->rl_reply) {
  572. req->rl_reply->rr_func = rpcrdma_reply_handler;
  573. /* this need only be done once, but... */
  574. req->rl_reply->rr_xprt = xprt;
  575. }
  576. /* Must suppress retransmit to maintain credits */
  577. if (req->rl_connect_cookie == xprt->connect_cookie)
  578. goto drop_connection;
  579. req->rl_connect_cookie = xprt->connect_cookie;
  580. if (rpcrdma_ep_post(&r_xprt->rx_ia, &r_xprt->rx_ep, req))
  581. goto drop_connection;
  582. rqst->rq_xmit_bytes_sent += rqst->rq_snd_buf.len;
  583. rqst->rq_bytes_sent = 0;
  584. return 0;
  585. drop_connection:
  586. xprt_disconnect_done(xprt);
  587. return -ENOTCONN; /* implies disconnect */
  588. }
  589. static void xprt_rdma_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
  590. {
  591. struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  592. long idle_time = 0;
  593. if (xprt_connected(xprt))
  594. idle_time = (long)(jiffies - xprt->last_used) / HZ;
  595. seq_printf(seq,
  596. "\txprt:\trdma %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu "
  597. "%lu %lu %lu %Lu %Lu %Lu %Lu %lu %lu %lu\n",
  598. 0, /* need a local port? */
  599. xprt->stat.bind_count,
  600. xprt->stat.connect_count,
  601. xprt->stat.connect_time,
  602. idle_time,
  603. xprt->stat.sends,
  604. xprt->stat.recvs,
  605. xprt->stat.bad_xids,
  606. xprt->stat.req_u,
  607. xprt->stat.bklog_u,
  608. r_xprt->rx_stats.read_chunk_count,
  609. r_xprt->rx_stats.write_chunk_count,
  610. r_xprt->rx_stats.reply_chunk_count,
  611. r_xprt->rx_stats.total_rdma_request,
  612. r_xprt->rx_stats.total_rdma_reply,
  613. r_xprt->rx_stats.pullup_copy_count,
  614. r_xprt->rx_stats.fixup_copy_count,
  615. r_xprt->rx_stats.hardway_register_count,
  616. r_xprt->rx_stats.failed_marshal_count,
  617. r_xprt->rx_stats.bad_reply_count);
  618. }
  619. /*
  620. * Plumbing for rpc transport switch and kernel module
  621. */
  622. static struct rpc_xprt_ops xprt_rdma_procs = {
  623. .reserve_xprt = xprt_rdma_reserve_xprt,
  624. .release_xprt = xprt_release_xprt_cong, /* sunrpc/xprt.c */
  625. .release_request = xprt_release_rqst_cong, /* ditto */
  626. .set_retrans_timeout = xprt_set_retrans_timeout_def, /* ditto */
  627. .rpcbind = rpcb_getport_async, /* sunrpc/rpcb_clnt.c */
  628. .set_port = xprt_rdma_set_port,
  629. .connect = xprt_rdma_connect,
  630. .buf_alloc = xprt_rdma_allocate,
  631. .buf_free = xprt_rdma_free,
  632. .send_request = xprt_rdma_send_request,
  633. .close = xprt_rdma_close,
  634. .destroy = xprt_rdma_destroy,
  635. .print_stats = xprt_rdma_print_stats
  636. };
  637. static struct xprt_class xprt_rdma = {
  638. .list = LIST_HEAD_INIT(xprt_rdma.list),
  639. .name = "rdma",
  640. .owner = THIS_MODULE,
  641. .ident = XPRT_TRANSPORT_RDMA,
  642. .setup = xprt_setup_rdma,
  643. };
  644. static void __exit xprt_rdma_cleanup(void)
  645. {
  646. int rc;
  647. dprintk(KERN_INFO "RPCRDMA Module Removed, deregister RPC RDMA transport\n");
  648. #ifdef RPC_DEBUG
  649. if (sunrpc_table_header) {
  650. unregister_sysctl_table(sunrpc_table_header);
  651. sunrpc_table_header = NULL;
  652. }
  653. #endif
  654. rc = xprt_unregister_transport(&xprt_rdma);
  655. if (rc)
  656. dprintk("RPC: %s: xprt_unregister returned %i\n",
  657. __func__, rc);
  658. }
  659. static int __init xprt_rdma_init(void)
  660. {
  661. int rc;
  662. rc = xprt_register_transport(&xprt_rdma);
  663. if (rc)
  664. return rc;
  665. dprintk(KERN_INFO "RPCRDMA Module Init, register RPC RDMA transport\n");
  666. dprintk(KERN_INFO "Defaults:\n");
  667. dprintk(KERN_INFO "\tSlots %d\n"
  668. "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n",
  669. xprt_rdma_slot_table_entries,
  670. xprt_rdma_max_inline_read, xprt_rdma_max_inline_write);
  671. dprintk(KERN_INFO "\tPadding %d\n\tMemreg %d\n",
  672. xprt_rdma_inline_write_padding, xprt_rdma_memreg_strategy);
  673. #ifdef RPC_DEBUG
  674. if (!sunrpc_table_header)
  675. sunrpc_table_header = register_sysctl_table(sunrpc_table);
  676. #endif
  677. return 0;
  678. }
  679. module_init(xprt_rdma_init);
  680. module_exit(xprt_rdma_cleanup);