~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/net/sunrpc/xprtrdma/transport.c

Version: ~ [ linux-5.6 ] ~ [ linux-5.5.13 ] ~ [ linux-5.4.28 ] ~ [ linux-5.3.18 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.113 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.174 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.217 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.217 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.82 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-3.9.11 ] ~ [ linux-3.8.13 ] ~ [ linux-3.7.10 ] ~ [ linux-3.6.11 ] ~ [ linux-3.5.7 ] ~ [ linux-3.4.113 ] ~ [ linux-3.3.8 ] ~ [ linux-3.2.102 ] ~ [ linux-3.1.10 ] ~ [ linux-3.0.101 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

  1 /*
  2  * Copyright (c) 2014-2017 Oracle.  All rights reserved.
  3  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
  4  *
  5  * This software is available to you under a choice of one of two
  6  * licenses.  You may choose to be licensed under the terms of the GNU
  7  * General Public License (GPL) Version 2, available from the file
  8  * COPYING in the main directory of this source tree, or the BSD-type
  9  * license below:
 10  *
 11  * Redistribution and use in source and binary forms, with or without
 12  * modification, are permitted provided that the following conditions
 13  * are met:
 14  *
 15  *      Redistributions of source code must retain the above copyright
 16  *      notice, this list of conditions and the following disclaimer.
 17  *
 18  *      Redistributions in binary form must reproduce the above
 19  *      copyright notice, this list of conditions and the following
 20  *      disclaimer in the documentation and/or other materials provided
 21  *      with the distribution.
 22  *
 23  *      Neither the name of the Network Appliance, Inc. nor the names of
 24  *      its contributors may be used to endorse or promote products
 25  *      derived from this software without specific prior written
 26  *      permission.
 27  *
 28  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 29  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 30  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 31  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 32  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 33  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 34  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 35  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 36  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 37  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 38  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 39  */
 40 
 41 /*
 42  * transport.c
 43  *
 44  * This file contains the top-level implementation of an RPC RDMA
 45  * transport.
 46  *
 47  * Naming convention: functions beginning with xprt_ are part of the
 48  * transport switch. All others are RPC RDMA internal.
 49  */
 50 
 51 #include <linux/module.h>
 52 #include <linux/slab.h>
 53 #include <linux/seq_file.h>
 54 #include <linux/sunrpc/addr.h>
 55 
 56 #include "xprt_rdma.h"
 57 
 58 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 59 # define RPCDBG_FACILITY        RPCDBG_TRANS
 60 #endif
 61 
 62 /*
 63  * tunables
 64  */
 65 
 66 static unsigned int xprt_rdma_slot_table_entries = RPCRDMA_DEF_SLOT_TABLE;
 67 unsigned int xprt_rdma_max_inline_read = RPCRDMA_DEF_INLINE;
 68 static unsigned int xprt_rdma_max_inline_write = RPCRDMA_DEF_INLINE;
 69 unsigned int xprt_rdma_memreg_strategy          = RPCRDMA_FRWR;
 70 int xprt_rdma_pad_optimize;
 71 
 72 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 73 
 74 static unsigned int min_slot_table_size = RPCRDMA_MIN_SLOT_TABLE;
 75 static unsigned int max_slot_table_size = RPCRDMA_MAX_SLOT_TABLE;
 76 static unsigned int min_inline_size = RPCRDMA_MIN_INLINE;
 77 static unsigned int max_inline_size = RPCRDMA_MAX_INLINE;
 78 static unsigned int zero;
 79 static unsigned int max_padding = PAGE_SIZE;
 80 static unsigned int min_memreg = RPCRDMA_BOUNCEBUFFERS;
 81 static unsigned int max_memreg = RPCRDMA_LAST - 1;
 82 static unsigned int dummy;
 83 
 84 static struct ctl_table_header *sunrpc_table_header;
 85 
 86 static struct ctl_table xr_tunables_table[] = {
 87         {
 88                 .procname       = "rdma_slot_table_entries",
 89                 .data           = &xprt_rdma_slot_table_entries,
 90                 .maxlen         = sizeof(unsigned int),
 91                 .mode           = 0644,
 92                 .proc_handler   = proc_dointvec_minmax,
 93                 .extra1         = &min_slot_table_size,
 94                 .extra2         = &max_slot_table_size
 95         },
 96         {
 97                 .procname       = "rdma_max_inline_read",
 98                 .data           = &xprt_rdma_max_inline_read,
 99                 .maxlen         = sizeof(unsigned int),
100                 .mode           = 0644,
101                 .proc_handler   = proc_dointvec_minmax,
102                 .extra1         = &min_inline_size,
103                 .extra2         = &max_inline_size,
104         },
105         {
106                 .procname       = "rdma_max_inline_write",
107                 .data           = &xprt_rdma_max_inline_write,
108                 .maxlen         = sizeof(unsigned int),
109                 .mode           = 0644,
110                 .proc_handler   = proc_dointvec_minmax,
111                 .extra1         = &min_inline_size,
112                 .extra2         = &max_inline_size,
113         },
114         {
115                 .procname       = "rdma_inline_write_padding",
116                 .data           = &dummy,
117                 .maxlen         = sizeof(unsigned int),
118                 .mode           = 0644,
119                 .proc_handler   = proc_dointvec_minmax,
120                 .extra1         = &zero,
121                 .extra2         = &max_padding,
122         },
123         {
124                 .procname       = "rdma_memreg_strategy",
125                 .data           = &xprt_rdma_memreg_strategy,
126                 .maxlen         = sizeof(unsigned int),
127                 .mode           = 0644,
128                 .proc_handler   = proc_dointvec_minmax,
129                 .extra1         = &min_memreg,
130                 .extra2         = &max_memreg,
131         },
132         {
133                 .procname       = "rdma_pad_optimize",
134                 .data           = &xprt_rdma_pad_optimize,
135                 .maxlen         = sizeof(unsigned int),
136                 .mode           = 0644,
137                 .proc_handler   = proc_dointvec,
138         },
139         { },
140 };
141 
142 static struct ctl_table sunrpc_table[] = {
143         {
144                 .procname       = "sunrpc",
145                 .mode           = 0555,
146                 .child          = xr_tunables_table
147         },
148         { },
149 };
150 
151 #endif
152 
153 static const struct rpc_xprt_ops xprt_rdma_procs;
154 
155 static void
156 xprt_rdma_format_addresses4(struct rpc_xprt *xprt, struct sockaddr *sap)
157 {
158         struct sockaddr_in *sin = (struct sockaddr_in *)sap;
159         char buf[20];
160 
161         snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr));
162         xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
163 
164         xprt->address_strings[RPC_DISPLAY_NETID] = RPCBIND_NETID_RDMA;
165 }
166 
167 static void
168 xprt_rdma_format_addresses6(struct rpc_xprt *xprt, struct sockaddr *sap)
169 {
170         struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)sap;
171         char buf[40];
172 
173         snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr);
174         xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL);
175 
176         xprt->address_strings[RPC_DISPLAY_NETID] = RPCBIND_NETID_RDMA6;
177 }
178 
179 void
180 xprt_rdma_format_addresses(struct rpc_xprt *xprt, struct sockaddr *sap)
181 {
182         char buf[128];
183 
184         switch (sap->sa_family) {
185         case AF_INET:
186                 xprt_rdma_format_addresses4(xprt, sap);
187                 break;
188         case AF_INET6:
189                 xprt_rdma_format_addresses6(xprt, sap);
190                 break;
191         default:
192                 pr_err("rpcrdma: Unrecognized address family\n");
193                 return;
194         }
195 
196         (void)rpc_ntop(sap, buf, sizeof(buf));
197         xprt->address_strings[RPC_DISPLAY_ADDR] = kstrdup(buf, GFP_KERNEL);
198 
199         snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap));
200         xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
201 
202         snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap));
203         xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
204 
205         xprt->address_strings[RPC_DISPLAY_PROTO] = "rdma";
206 }
207 
208 void
209 xprt_rdma_free_addresses(struct rpc_xprt *xprt)
210 {
211         unsigned int i;
212 
213         for (i = 0; i < RPC_DISPLAY_MAX; i++)
214                 switch (i) {
215                 case RPC_DISPLAY_PROTO:
216                 case RPC_DISPLAY_NETID:
217                         continue;
218                 default:
219                         kfree(xprt->address_strings[i]);
220                 }
221 }
222 
223 void
224 rpcrdma_conn_func(struct rpcrdma_ep *ep)
225 {
226         schedule_delayed_work(&ep->rep_connect_worker, 0);
227 }
228 
229 void
230 rpcrdma_connect_worker(struct work_struct *work)
231 {
232         struct rpcrdma_ep *ep =
233                 container_of(work, struct rpcrdma_ep, rep_connect_worker.work);
234         struct rpcrdma_xprt *r_xprt =
235                 container_of(ep, struct rpcrdma_xprt, rx_ep);
236         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
237 
238         spin_lock_bh(&xprt->transport_lock);
239         if (ep->rep_connected > 0) {
240                 if (!xprt_test_and_set_connected(xprt))
241                         xprt_wake_pending_tasks(xprt, 0);
242         } else {
243                 if (xprt_test_and_clear_connected(xprt))
244                         xprt_wake_pending_tasks(xprt, -ENOTCONN);
245         }
246         spin_unlock_bh(&xprt->transport_lock);
247 }
248 
249 static void
250 xprt_rdma_connect_worker(struct work_struct *work)
251 {
252         struct rpcrdma_xprt *r_xprt = container_of(work, struct rpcrdma_xprt,
253                                                    rx_connect_worker.work);
254         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
255         int rc = 0;
256 
257         xprt_clear_connected(xprt);
258 
259         rc = rpcrdma_ep_connect(&r_xprt->rx_ep, &r_xprt->rx_ia);
260         if (rc)
261                 xprt_wake_pending_tasks(xprt, rc);
262 
263         xprt_clear_connecting(xprt);
264 }
265 
266 static void
267 xprt_rdma_inject_disconnect(struct rpc_xprt *xprt)
268 {
269         struct rpcrdma_xprt *r_xprt = container_of(xprt, struct rpcrdma_xprt,
270                                                    rx_xprt);
271 
272         trace_xprtrdma_inject_dsc(r_xprt);
273         rdma_disconnect(r_xprt->rx_ia.ri_id);
274 }
275 
276 /*
277  * xprt_rdma_destroy
278  *
279  * Destroy the xprt.
280  * Free all memory associated with the object, including its own.
281  * NOTE: none of the *destroy methods free memory for their top-level
282  * objects, even though they may have allocated it (they do free
283  * private memory). It's up to the caller to handle it. In this
284  * case (RDMA transport), all structure memory is inlined with the
285  * struct rpcrdma_xprt.
286  */
287 static void
288 xprt_rdma_destroy(struct rpc_xprt *xprt)
289 {
290         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
291 
292         trace_xprtrdma_destroy(r_xprt);
293 
294         cancel_delayed_work_sync(&r_xprt->rx_connect_worker);
295 
296         xprt_clear_connected(xprt);
297 
298         rpcrdma_ep_destroy(&r_xprt->rx_ep, &r_xprt->rx_ia);
299         rpcrdma_buffer_destroy(&r_xprt->rx_buf);
300         rpcrdma_ia_close(&r_xprt->rx_ia);
301 
302         xprt_rdma_free_addresses(xprt);
303         xprt_free(xprt);
304 
305         module_put(THIS_MODULE);
306 }
307 
308 static const struct rpc_timeout xprt_rdma_default_timeout = {
309         .to_initval = 60 * HZ,
310         .to_maxval = 60 * HZ,
311 };
312 
313 /**
314  * xprt_setup_rdma - Set up transport to use RDMA
315  *
316  * @args: rpc transport arguments
317  */
318 static struct rpc_xprt *
319 xprt_setup_rdma(struct xprt_create *args)
320 {
321         struct rpcrdma_create_data_internal cdata;
322         struct rpc_xprt *xprt;
323         struct rpcrdma_xprt *new_xprt;
324         struct rpcrdma_ep *new_ep;
325         struct sockaddr *sap;
326         int rc;
327 
328         if (args->addrlen > sizeof(xprt->addr)) {
329                 dprintk("RPC:       %s: address too large\n", __func__);
330                 return ERR_PTR(-EBADF);
331         }
332 
333         xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt),
334                         xprt_rdma_slot_table_entries,
335                         xprt_rdma_slot_table_entries);
336         if (xprt == NULL) {
337                 dprintk("RPC:       %s: couldn't allocate rpcrdma_xprt\n",
338                         __func__);
339                 return ERR_PTR(-ENOMEM);
340         }
341 
342         /* 60 second timeout, no retries */
343         xprt->timeout = &xprt_rdma_default_timeout;
344         xprt->bind_timeout = RPCRDMA_BIND_TO;
345         xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO;
346         xprt->idle_timeout = RPCRDMA_IDLE_DISC_TO;
347 
348         xprt->resvport = 0;             /* privileged port not needed */
349         xprt->tsh_size = 0;             /* RPC-RDMA handles framing */
350         xprt->ops = &xprt_rdma_procs;
351 
352         /*
353          * Set up RDMA-specific connect data.
354          */
355         sap = args->dstaddr;
356 
357         /* Ensure xprt->addr holds valid server TCP (not RDMA)
358          * address, for any side protocols which peek at it */
359         xprt->prot = IPPROTO_TCP;
360         xprt->addrlen = args->addrlen;
361         memcpy(&xprt->addr, sap, xprt->addrlen);
362 
363         if (rpc_get_port(sap))
364                 xprt_set_bound(xprt);
365         xprt_rdma_format_addresses(xprt, sap);
366 
367         cdata.max_requests = xprt->max_reqs;
368 
369         cdata.rsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA write max */
370         cdata.wsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA read max */
371 
372         cdata.inline_wsize = xprt_rdma_max_inline_write;
373         if (cdata.inline_wsize > cdata.wsize)
374                 cdata.inline_wsize = cdata.wsize;
375 
376         cdata.inline_rsize = xprt_rdma_max_inline_read;
377         if (cdata.inline_rsize > cdata.rsize)
378                 cdata.inline_rsize = cdata.rsize;
379 
380         /*
381          * Create new transport instance, which includes initialized
382          *  o ia
383          *  o endpoint
384          *  o buffers
385          */
386 
387         new_xprt = rpcx_to_rdmax(xprt);
388 
389         rc = rpcrdma_ia_open(new_xprt);
390         if (rc)
391                 goto out1;
392 
393         /*
394          * initialize and create ep
395          */
396         new_xprt->rx_data = cdata;
397         new_ep = &new_xprt->rx_ep;
398 
399         rc = rpcrdma_ep_create(&new_xprt->rx_ep,
400                                 &new_xprt->rx_ia, &new_xprt->rx_data);
401         if (rc)
402                 goto out2;
403 
404         rc = rpcrdma_buffer_create(new_xprt);
405         if (rc)
406                 goto out3;
407 
408         INIT_DELAYED_WORK(&new_xprt->rx_connect_worker,
409                           xprt_rdma_connect_worker);
410 
411         xprt->max_payload = new_xprt->rx_ia.ri_ops->ro_maxpages(new_xprt);
412         if (xprt->max_payload == 0)
413                 goto out4;
414         xprt->max_payload <<= PAGE_SHIFT;
415         dprintk("RPC:       %s: transport data payload maximum: %zu bytes\n",
416                 __func__, xprt->max_payload);
417 
418         if (!try_module_get(THIS_MODULE))
419                 goto out4;
420 
421         dprintk("RPC:       %s: %s:%s\n", __func__,
422                 xprt->address_strings[RPC_DISPLAY_ADDR],
423                 xprt->address_strings[RPC_DISPLAY_PORT]);
424         trace_xprtrdma_create(new_xprt);
425         return xprt;
426 
427 out4:
428         rpcrdma_buffer_destroy(&new_xprt->rx_buf);
429         rc = -ENODEV;
430 out3:
431         rpcrdma_ep_destroy(new_ep, &new_xprt->rx_ia);
432 out2:
433         rpcrdma_ia_close(&new_xprt->rx_ia);
434 out1:
435         trace_xprtrdma_destroy(new_xprt);
436         xprt_rdma_free_addresses(xprt);
437         xprt_free(xprt);
438         return ERR_PTR(rc);
439 }
440 
441 /**
442  * xprt_rdma_close - Close down RDMA connection
443  * @xprt: generic transport to be closed
444  *
445  * Called during transport shutdown reconnect, or device
446  * removal. Caller holds the transport's write lock.
447  */
448 static void
449 xprt_rdma_close(struct rpc_xprt *xprt)
450 {
451         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
452         struct rpcrdma_ep *ep = &r_xprt->rx_ep;
453         struct rpcrdma_ia *ia = &r_xprt->rx_ia;
454 
455         dprintk("RPC:       %s: closing xprt %p\n", __func__, xprt);
456 
457         if (test_and_clear_bit(RPCRDMA_IAF_REMOVING, &ia->ri_flags)) {
458                 xprt_clear_connected(xprt);
459                 rpcrdma_ia_remove(ia);
460                 return;
461         }
462         if (ep->rep_connected == -ENODEV)
463                 return;
464         if (ep->rep_connected > 0)
465                 xprt->reestablish_timeout = 0;
466         xprt_disconnect_done(xprt);
467         rpcrdma_ep_disconnect(ep, ia);
468 }
469 
470 /**
471  * xprt_rdma_set_port - update server port with rpcbind result
472  * @xprt: controlling RPC transport
473  * @port: new port value
474  *
475  * Transport connect status is unchanged.
476  */
477 static void
478 xprt_rdma_set_port(struct rpc_xprt *xprt, u16 port)
479 {
480         struct sockaddr *sap = (struct sockaddr *)&xprt->addr;
481         char buf[8];
482 
483         dprintk("RPC:       %s: setting port for xprt %p (%s:%s) to %u\n",
484                 __func__, xprt,
485                 xprt->address_strings[RPC_DISPLAY_ADDR],
486                 xprt->address_strings[RPC_DISPLAY_PORT],
487                 port);
488 
489         rpc_set_port(sap, port);
490 
491         kfree(xprt->address_strings[RPC_DISPLAY_PORT]);
492         snprintf(buf, sizeof(buf), "%u", port);
493         xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL);
494 
495         kfree(xprt->address_strings[RPC_DISPLAY_HEX_PORT]);
496         snprintf(buf, sizeof(buf), "%4hx", port);
497         xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL);
498 }
499 
500 /**
501  * xprt_rdma_timer - invoked when an RPC times out
502  * @xprt: controlling RPC transport
503  * @task: RPC task that timed out
504  *
505  * Invoked when the transport is still connected, but an RPC
506  * retransmit timeout occurs.
507  *
508  * Since RDMA connections don't have a keep-alive, forcibly
509  * disconnect and retry to connect. This drives full
510  * detection of the network path, and retransmissions of
511  * all pending RPCs.
512  */
513 static void
514 xprt_rdma_timer(struct rpc_xprt *xprt, struct rpc_task *task)
515 {
516         xprt_force_disconnect(xprt);
517 }
518 
519 static void
520 xprt_rdma_connect(struct rpc_xprt *xprt, struct rpc_task *task)
521 {
522         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
523 
524         if (r_xprt->rx_ep.rep_connected != 0) {
525                 /* Reconnect */
526                 schedule_delayed_work(&r_xprt->rx_connect_worker,
527                                       xprt->reestablish_timeout);
528                 xprt->reestablish_timeout <<= 1;
529                 if (xprt->reestablish_timeout > RPCRDMA_MAX_REEST_TO)
530                         xprt->reestablish_timeout = RPCRDMA_MAX_REEST_TO;
531                 else if (xprt->reestablish_timeout < RPCRDMA_INIT_REEST_TO)
532                         xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO;
533         } else {
534                 schedule_delayed_work(&r_xprt->rx_connect_worker, 0);
535                 if (!RPC_IS_ASYNC(task))
536                         flush_delayed_work(&r_xprt->rx_connect_worker);
537         }
538 }
539 
540 static bool
541 rpcrdma_get_sendbuf(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req,
542                     size_t size, gfp_t flags)
543 {
544         struct rpcrdma_regbuf *rb;
545 
546         if (req->rl_sendbuf && rdmab_length(req->rl_sendbuf) >= size)
547                 return true;
548 
549         rb = rpcrdma_alloc_regbuf(size, DMA_TO_DEVICE, flags);
550         if (IS_ERR(rb))
551                 return false;
552 
553         rpcrdma_free_regbuf(req->rl_sendbuf);
554         r_xprt->rx_stats.hardway_register_count += size;
555         req->rl_sendbuf = rb;
556         return true;
557 }
558 
559 /* The rq_rcv_buf is used only if a Reply chunk is necessary.
560  * The decision to use a Reply chunk is made later in
561  * rpcrdma_marshal_req. This buffer is registered at that time.
562  *
563  * Otherwise, the associated RPC Reply arrives in a separate
564  * Receive buffer, arbitrarily chosen by the HCA. The buffer
565  * allocated here for the RPC Reply is not utilized in that
566  * case. See rpcrdma_inline_fixup.
567  *
568  * A regbuf is used here to remember the buffer size.
569  */
570 static bool
571 rpcrdma_get_recvbuf(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req,
572                     size_t size, gfp_t flags)
573 {
574         struct rpcrdma_regbuf *rb;
575 
576         if (req->rl_recvbuf && rdmab_length(req->rl_recvbuf) >= size)
577                 return true;
578 
579         rb = rpcrdma_alloc_regbuf(size, DMA_NONE, flags);
580         if (IS_ERR(rb))
581                 return false;
582 
583         rpcrdma_free_regbuf(req->rl_recvbuf);
584         r_xprt->rx_stats.hardway_register_count += size;
585         req->rl_recvbuf = rb;
586         return true;
587 }
588 
589 /**
590  * xprt_rdma_allocate - allocate transport resources for an RPC
591  * @task: RPC task
592  *
593  * Return values:
594  *        0:    Success; rq_buffer points to RPC buffer to use
595  *   ENOMEM:    Out of memory, call again later
596  *      EIO:    A permanent error occurred, do not retry
597  *
598  * The RDMA allocate/free functions need the task structure as a place
599  * to hide the struct rpcrdma_req, which is necessary for the actual
600  * send/recv sequence.
601  *
602  * xprt_rdma_allocate provides buffers that are already mapped for
603  * DMA, and a local DMA lkey is provided for each.
604  */
605 static int
606 xprt_rdma_allocate(struct rpc_task *task)
607 {
608         struct rpc_rqst *rqst = task->tk_rqstp;
609         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt);
610         struct rpcrdma_req *req;
611         gfp_t flags;
612 
613         req = rpcrdma_buffer_get(&r_xprt->rx_buf);
614         if (req == NULL)
615                 goto out_get;
616 
617         flags = RPCRDMA_DEF_GFP;
618         if (RPC_IS_SWAPPER(task))
619                 flags = __GFP_MEMALLOC | GFP_NOWAIT | __GFP_NOWARN;
620 
621         if (!rpcrdma_get_sendbuf(r_xprt, req, rqst->rq_callsize, flags))
622                 goto out_fail;
623         if (!rpcrdma_get_recvbuf(r_xprt, req, rqst->rq_rcvsize, flags))
624                 goto out_fail;
625 
626         rpcrdma_set_xprtdata(rqst, req);
627         rqst->rq_buffer = req->rl_sendbuf->rg_base;
628         rqst->rq_rbuffer = req->rl_recvbuf->rg_base;
629         trace_xprtrdma_allocate(task, req);
630         return 0;
631 
632 out_fail:
633         rpcrdma_buffer_put(req);
634 out_get:
635         trace_xprtrdma_allocate(task, NULL);
636         return -ENOMEM;
637 }
638 
639 /**
640  * xprt_rdma_free - release resources allocated by xprt_rdma_allocate
641  * @task: RPC task
642  *
643  * Caller guarantees rqst->rq_buffer is non-NULL.
644  */
645 static void
646 xprt_rdma_free(struct rpc_task *task)
647 {
648         struct rpc_rqst *rqst = task->tk_rqstp;
649         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt);
650         struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
651 
652         if (test_bit(RPCRDMA_REQ_F_PENDING, &req->rl_flags))
653                 rpcrdma_release_rqst(r_xprt, req);
654         trace_xprtrdma_rpc_done(task, req);
655         rpcrdma_buffer_put(req);
656 }
657 
658 /**
659  * xprt_rdma_send_request - marshal and send an RPC request
660  * @task: RPC task with an RPC message in rq_snd_buf
661  *
662  * Caller holds the transport's write lock.
663  *
664  * Returns:
665  *      %0 if the RPC message has been sent
666  *      %-ENOTCONN if the caller should reconnect and call again
667  *      %-EAGAIN if the caller should call again
668  *      %-ENOBUFS if the caller should call again after a delay
669  *      %-EIO if a permanent error occurred and the request was not
670  *              sent. Do not try to send this message again.
671  */
672 static int
673 xprt_rdma_send_request(struct rpc_task *task)
674 {
675         struct rpc_rqst *rqst = task->tk_rqstp;
676         struct rpc_xprt *xprt = rqst->rq_xprt;
677         struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
678         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
679         int rc = 0;
680 
681 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
682         if (unlikely(!rqst->rq_buffer))
683                 return xprt_rdma_bc_send_reply(rqst);
684 #endif  /* CONFIG_SUNRPC_BACKCHANNEL */
685 
686         if (!xprt_connected(xprt))
687                 goto drop_connection;
688 
689         rc = rpcrdma_marshal_req(r_xprt, rqst);
690         if (rc < 0)
691                 goto failed_marshal;
692 
693         if (req->rl_reply == NULL)              /* e.g. reconnection */
694                 rpcrdma_recv_buffer_get(req);
695 
696         /* Must suppress retransmit to maintain credits */
697         if (rqst->rq_connect_cookie == xprt->connect_cookie)
698                 goto drop_connection;
699         rqst->rq_xtime = ktime_get();
700 
701         __set_bit(RPCRDMA_REQ_F_PENDING, &req->rl_flags);
702         if (rpcrdma_ep_post(&r_xprt->rx_ia, &r_xprt->rx_ep, req))
703                 goto drop_connection;
704 
705         rqst->rq_xmit_bytes_sent += rqst->rq_snd_buf.len;
706         rqst->rq_bytes_sent = 0;
707 
708         /* An RPC with no reply will throw off credit accounting,
709          * so drop the connection to reset the credit grant.
710          */
711         if (!rpc_reply_expected(task))
712                 goto drop_connection;
713         return 0;
714 
715 failed_marshal:
716         if (rc != -ENOTCONN)
717                 return rc;
718 drop_connection:
719         xprt_disconnect_done(xprt);
720         return -ENOTCONN;       /* implies disconnect */
721 }
722 
723 void xprt_rdma_print_stats(struct rpc_xprt *xprt, struct seq_file *seq)
724 {
725         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
726         long idle_time = 0;
727 
728         if (xprt_connected(xprt))
729                 idle_time = (long)(jiffies - xprt->last_used) / HZ;
730 
731         seq_puts(seq, "\txprt:\trdma ");
732         seq_printf(seq, "%u %lu %lu %lu %ld %lu %lu %lu %llu %llu ",
733                    0,   /* need a local port? */
734                    xprt->stat.bind_count,
735                    xprt->stat.connect_count,
736                    xprt->stat.connect_time,
737                    idle_time,
738                    xprt->stat.sends,
739                    xprt->stat.recvs,
740                    xprt->stat.bad_xids,
741                    xprt->stat.req_u,
742                    xprt->stat.bklog_u);
743         seq_printf(seq, "%lu %lu %lu %llu %llu %llu %llu %lu %lu %lu %lu ",
744                    r_xprt->rx_stats.read_chunk_count,
745                    r_xprt->rx_stats.write_chunk_count,
746                    r_xprt->rx_stats.reply_chunk_count,
747                    r_xprt->rx_stats.total_rdma_request,
748                    r_xprt->rx_stats.total_rdma_reply,
749                    r_xprt->rx_stats.pullup_copy_count,
750                    r_xprt->rx_stats.fixup_copy_count,
751                    r_xprt->rx_stats.hardway_register_count,
752                    r_xprt->rx_stats.failed_marshal_count,
753                    r_xprt->rx_stats.bad_reply_count,
754                    r_xprt->rx_stats.nomsg_call_count);
755         seq_printf(seq, "%lu %lu %lu %lu %lu %lu\n",
756                    r_xprt->rx_stats.mrs_recovered,
757                    r_xprt->rx_stats.mrs_orphaned,
758                    r_xprt->rx_stats.mrs_allocated,
759                    r_xprt->rx_stats.local_inv_needed,
760                    r_xprt->rx_stats.empty_sendctx_q,
761                    r_xprt->rx_stats.reply_waits_for_send);
762 }
763 
764 static int
765 xprt_rdma_enable_swap(struct rpc_xprt *xprt)
766 {
767         return 0;
768 }
769 
770 static void
771 xprt_rdma_disable_swap(struct rpc_xprt *xprt)
772 {
773 }
774 
775 /*
776  * Plumbing for rpc transport switch and kernel module
777  */
778 
779 static const struct rpc_xprt_ops xprt_rdma_procs = {
780         .reserve_xprt           = xprt_reserve_xprt_cong,
781         .release_xprt           = xprt_release_xprt_cong, /* sunrpc/xprt.c */
782         .alloc_slot             = xprt_alloc_slot,
783         .release_request        = xprt_release_rqst_cong,       /* ditto */
784         .set_retrans_timeout    = xprt_set_retrans_timeout_def, /* ditto */
785         .timer                  = xprt_rdma_timer,
786         .rpcbind                = rpcb_getport_async,   /* sunrpc/rpcb_clnt.c */
787         .set_port               = xprt_rdma_set_port,
788         .connect                = xprt_rdma_connect,
789         .buf_alloc              = xprt_rdma_allocate,
790         .buf_free               = xprt_rdma_free,
791         .send_request           = xprt_rdma_send_request,
792         .close                  = xprt_rdma_close,
793         .destroy                = xprt_rdma_destroy,
794         .print_stats            = xprt_rdma_print_stats,
795         .enable_swap            = xprt_rdma_enable_swap,
796         .disable_swap           = xprt_rdma_disable_swap,
797         .inject_disconnect      = xprt_rdma_inject_disconnect,
798 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
799         .bc_setup               = xprt_rdma_bc_setup,
800         .bc_up                  = xprt_rdma_bc_up,
801         .bc_maxpayload          = xprt_rdma_bc_maxpayload,
802         .bc_free_rqst           = xprt_rdma_bc_free_rqst,
803         .bc_destroy             = xprt_rdma_bc_destroy,
804 #endif
805 };
806 
807 static struct xprt_class xprt_rdma = {
808         .list                   = LIST_HEAD_INIT(xprt_rdma.list),
809         .name                   = "rdma",
810         .owner                  = THIS_MODULE,
811         .ident                  = XPRT_TRANSPORT_RDMA,
812         .setup                  = xprt_setup_rdma,
813 };
814 
815 void xprt_rdma_cleanup(void)
816 {
817         int rc;
818 
819         dprintk("RPCRDMA Module Removed, deregister RPC RDMA transport\n");
820 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
821         if (sunrpc_table_header) {
822                 unregister_sysctl_table(sunrpc_table_header);
823                 sunrpc_table_header = NULL;
824         }
825 #endif
826         rc = xprt_unregister_transport(&xprt_rdma);
827         if (rc)
828                 dprintk("RPC:       %s: xprt_unregister returned %i\n",
829                         __func__, rc);
830 
831         rpcrdma_destroy_wq();
832 
833         rc = xprt_unregister_transport(&xprt_rdma_bc);
834         if (rc)
835                 dprintk("RPC:       %s: xprt_unregister(bc) returned %i\n",
836                         __func__, rc);
837 }
838 
839 int xprt_rdma_init(void)
840 {
841         int rc;
842 
843         rc = rpcrdma_alloc_wq();
844         if (rc)
845                 return rc;
846 
847         rc = xprt_register_transport(&xprt_rdma);
848         if (rc) {
849                 rpcrdma_destroy_wq();
850                 return rc;
851         }
852 
853         rc = xprt_register_transport(&xprt_rdma_bc);
854         if (rc) {
855                 xprt_unregister_transport(&xprt_rdma);
856                 rpcrdma_destroy_wq();
857                 return rc;
858         }
859 
860         dprintk("RPCRDMA Module Init, register RPC RDMA transport\n");
861 
862         dprintk("Defaults:\n");
863         dprintk("\tSlots %d\n"
864                 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n",
865                 xprt_rdma_slot_table_entries,
866                 xprt_rdma_max_inline_read, xprt_rdma_max_inline_write);
867         dprintk("\tPadding 0\n\tMemreg %d\n", xprt_rdma_memreg_strategy);
868 
869 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
870         if (!sunrpc_table_header)
871                 sunrpc_table_header = register_sysctl_table(sunrpc_table);
872 #endif
873         return 0;
874 }
875 

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp