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