Merge git://git.kernel.org/pub/scm/linux/kernel/git/davem/sparc-2.6
[linux-2.6.git] / net / rds / rdma.c
1 /*
2  * Copyright (c) 2007 Oracle.  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
8  * OpenIB.org BSD license below:
9  *
10  *     Redistribution and use in source and binary forms, with or
11  *     without modification, are permitted provided that the following
12  *     conditions are met:
13  *
14  *      - Redistributions of source code must retain the above
15  *        copyright notice, this list of conditions and the following
16  *        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
21  *        provided with the distribution.
22  *
23  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
24  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
25  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
26  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
27  * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
28  * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
29  * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
30  * SOFTWARE.
31  *
32  */
33 #include <linux/pagemap.h>
34 #include <linux/rbtree.h>
35 #include <linux/dma-mapping.h> /* for DMA_*_DEVICE */
36
37 #include "rdma.h"
38
39 /*
40  * XXX
41  *  - build with sparse
42  *  - should we limit the size of a mr region?  let transport return failure?
43  *  - should we detect duplicate keys on a socket?  hmm.
44  *  - an rdma is an mlock, apply rlimit?
45  */
46
47 /*
48  * get the number of pages by looking at the page indices that the start and
49  * end addresses fall in.
50  *
51  * Returns 0 if the vec is invalid.  It is invalid if the number of bytes
52  * causes the address to wrap or overflows an unsigned int.  This comes
53  * from being stored in the 'length' member of 'struct scatterlist'.
54  */
55 static unsigned int rds_pages_in_vec(struct rds_iovec *vec)
56 {
57         if ((vec->addr + vec->bytes <= vec->addr) ||
58             (vec->bytes > (u64)UINT_MAX))
59                 return 0;
60
61         return ((vec->addr + vec->bytes + PAGE_SIZE - 1) >> PAGE_SHIFT) -
62                 (vec->addr >> PAGE_SHIFT);
63 }
64
65 static struct rds_mr *rds_mr_tree_walk(struct rb_root *root, u64 key,
66                                        struct rds_mr *insert)
67 {
68         struct rb_node **p = &root->rb_node;
69         struct rb_node *parent = NULL;
70         struct rds_mr *mr;
71
72         while (*p) {
73                 parent = *p;
74                 mr = rb_entry(parent, struct rds_mr, r_rb_node);
75
76                 if (key < mr->r_key)
77                         p = &(*p)->rb_left;
78                 else if (key > mr->r_key)
79                         p = &(*p)->rb_right;
80                 else
81                         return mr;
82         }
83
84         if (insert) {
85                 rb_link_node(&insert->r_rb_node, parent, p);
86                 rb_insert_color(&insert->r_rb_node, root);
87                 atomic_inc(&insert->r_refcount);
88         }
89         return NULL;
90 }
91
92 /*
93  * Destroy the transport-specific part of a MR.
94  */
95 static void rds_destroy_mr(struct rds_mr *mr)
96 {
97         struct rds_sock *rs = mr->r_sock;
98         void *trans_private = NULL;
99         unsigned long flags;
100
101         rdsdebug("RDS: destroy mr key is %x refcnt %u\n",
102                         mr->r_key, atomic_read(&mr->r_refcount));
103
104         if (test_and_set_bit(RDS_MR_DEAD, &mr->r_state))
105                 return;
106
107         spin_lock_irqsave(&rs->rs_rdma_lock, flags);
108         if (!RB_EMPTY_NODE(&mr->r_rb_node))
109                 rb_erase(&mr->r_rb_node, &rs->rs_rdma_keys);
110         trans_private = mr->r_trans_private;
111         mr->r_trans_private = NULL;
112         spin_unlock_irqrestore(&rs->rs_rdma_lock, flags);
113
114         if (trans_private)
115                 mr->r_trans->free_mr(trans_private, mr->r_invalidate);
116 }
117
118 void __rds_put_mr_final(struct rds_mr *mr)
119 {
120         rds_destroy_mr(mr);
121         kfree(mr);
122 }
123
124 /*
125  * By the time this is called we can't have any more ioctls called on
126  * the socket so we don't need to worry about racing with others.
127  */
128 void rds_rdma_drop_keys(struct rds_sock *rs)
129 {
130         struct rds_mr *mr;
131         struct rb_node *node;
132
133         /* Release any MRs associated with this socket */
134         while ((node = rb_first(&rs->rs_rdma_keys))) {
135                 mr = container_of(node, struct rds_mr, r_rb_node);
136                 if (mr->r_trans == rs->rs_transport)
137                         mr->r_invalidate = 0;
138                 rds_mr_put(mr);
139         }
140
141         if (rs->rs_transport && rs->rs_transport->flush_mrs)
142                 rs->rs_transport->flush_mrs();
143 }
144
145 /*
146  * Helper function to pin user pages.
147  */
148 static int rds_pin_pages(unsigned long user_addr, unsigned int nr_pages,
149                         struct page **pages, int write)
150 {
151         int ret;
152
153         ret = get_user_pages_fast(user_addr, nr_pages, write, pages);
154
155         if (ret >= 0 && ret < nr_pages) {
156                 while (ret--)
157                         put_page(pages[ret]);
158                 ret = -EFAULT;
159         }
160
161         return ret;
162 }
163
164 static int __rds_rdma_map(struct rds_sock *rs, struct rds_get_mr_args *args,
165                                 u64 *cookie_ret, struct rds_mr **mr_ret)
166 {
167         struct rds_mr *mr = NULL, *found;
168         unsigned int nr_pages;
169         struct page **pages = NULL;
170         struct scatterlist *sg;
171         void *trans_private;
172         unsigned long flags;
173         rds_rdma_cookie_t cookie;
174         unsigned int nents;
175         long i;
176         int ret;
177
178         if (rs->rs_bound_addr == 0) {
179                 ret = -ENOTCONN; /* XXX not a great errno */
180                 goto out;
181         }
182
183         if (rs->rs_transport->get_mr == NULL) {
184                 ret = -EOPNOTSUPP;
185                 goto out;
186         }
187
188         nr_pages = rds_pages_in_vec(&args->vec);
189         if (nr_pages == 0) {
190                 ret = -EINVAL;
191                 goto out;
192         }
193
194         rdsdebug("RDS: get_mr addr %llx len %llu nr_pages %u\n",
195                 args->vec.addr, args->vec.bytes, nr_pages);
196
197         /* XXX clamp nr_pages to limit the size of this alloc? */
198         pages = kcalloc(nr_pages, sizeof(struct page *), GFP_KERNEL);
199         if (pages == NULL) {
200                 ret = -ENOMEM;
201                 goto out;
202         }
203
204         mr = kzalloc(sizeof(struct rds_mr), GFP_KERNEL);
205         if (mr == NULL) {
206                 ret = -ENOMEM;
207                 goto out;
208         }
209
210         atomic_set(&mr->r_refcount, 1);
211         RB_CLEAR_NODE(&mr->r_rb_node);
212         mr->r_trans = rs->rs_transport;
213         mr->r_sock = rs;
214
215         if (args->flags & RDS_RDMA_USE_ONCE)
216                 mr->r_use_once = 1;
217         if (args->flags & RDS_RDMA_INVALIDATE)
218                 mr->r_invalidate = 1;
219         if (args->flags & RDS_RDMA_READWRITE)
220                 mr->r_write = 1;
221
222         /*
223          * Pin the pages that make up the user buffer and transfer the page
224          * pointers to the mr's sg array.  We check to see if we've mapped
225          * the whole region after transferring the partial page references
226          * to the sg array so that we can have one page ref cleanup path.
227          *
228          * For now we have no flag that tells us whether the mapping is
229          * r/o or r/w. We need to assume r/w, or we'll do a lot of RDMA to
230          * the zero page.
231          */
232         ret = rds_pin_pages(args->vec.addr & PAGE_MASK, nr_pages, pages, 1);
233         if (ret < 0)
234                 goto out;
235
236         nents = ret;
237         sg = kcalloc(nents, sizeof(*sg), GFP_KERNEL);
238         if (sg == NULL) {
239                 ret = -ENOMEM;
240                 goto out;
241         }
242         WARN_ON(!nents);
243         sg_init_table(sg, nents);
244
245         /* Stick all pages into the scatterlist */
246         for (i = 0 ; i < nents; i++)
247                 sg_set_page(&sg[i], pages[i], PAGE_SIZE, 0);
248
249         rdsdebug("RDS: trans_private nents is %u\n", nents);
250
251         /* Obtain a transport specific MR. If this succeeds, the
252          * s/g list is now owned by the MR.
253          * Note that dma_map() implies that pending writes are
254          * flushed to RAM, so no dma_sync is needed here. */
255         trans_private = rs->rs_transport->get_mr(sg, nents, rs,
256                                                  &mr->r_key);
257
258         if (IS_ERR(trans_private)) {
259                 for (i = 0 ; i < nents; i++)
260                         put_page(sg_page(&sg[i]));
261                 kfree(sg);
262                 ret = PTR_ERR(trans_private);
263                 goto out;
264         }
265
266         mr->r_trans_private = trans_private;
267
268         rdsdebug("RDS: get_mr put_user key is %x cookie_addr %p\n",
269                mr->r_key, (void *)(unsigned long) args->cookie_addr);
270
271         /* The user may pass us an unaligned address, but we can only
272          * map page aligned regions. So we keep the offset, and build
273          * a 64bit cookie containing <R_Key, offset> and pass that
274          * around. */
275         cookie = rds_rdma_make_cookie(mr->r_key, args->vec.addr & ~PAGE_MASK);
276         if (cookie_ret)
277                 *cookie_ret = cookie;
278
279         if (args->cookie_addr && put_user(cookie, (u64 __user *)(unsigned long) args->cookie_addr)) {
280                 ret = -EFAULT;
281                 goto out;
282         }
283
284         /* Inserting the new MR into the rbtree bumps its
285          * reference count. */
286         spin_lock_irqsave(&rs->rs_rdma_lock, flags);
287         found = rds_mr_tree_walk(&rs->rs_rdma_keys, mr->r_key, mr);
288         spin_unlock_irqrestore(&rs->rs_rdma_lock, flags);
289
290         BUG_ON(found && found != mr);
291
292         rdsdebug("RDS: get_mr key is %x\n", mr->r_key);
293         if (mr_ret) {
294                 atomic_inc(&mr->r_refcount);
295                 *mr_ret = mr;
296         }
297
298         ret = 0;
299 out:
300         kfree(pages);
301         if (mr)
302                 rds_mr_put(mr);
303         return ret;
304 }
305
306 int rds_get_mr(struct rds_sock *rs, char __user *optval, int optlen)
307 {
308         struct rds_get_mr_args args;
309
310         if (optlen != sizeof(struct rds_get_mr_args))
311                 return -EINVAL;
312
313         if (copy_from_user(&args, (struct rds_get_mr_args __user *)optval,
314                            sizeof(struct rds_get_mr_args)))
315                 return -EFAULT;
316
317         return __rds_rdma_map(rs, &args, NULL, NULL);
318 }
319
320 int rds_get_mr_for_dest(struct rds_sock *rs, char __user *optval, int optlen)
321 {
322         struct rds_get_mr_for_dest_args args;
323         struct rds_get_mr_args new_args;
324
325         if (optlen != sizeof(struct rds_get_mr_for_dest_args))
326                 return -EINVAL;
327
328         if (copy_from_user(&args, (struct rds_get_mr_for_dest_args __user *)optval,
329                            sizeof(struct rds_get_mr_for_dest_args)))
330                 return -EFAULT;
331
332         /*
333          * Initially, just behave like get_mr().
334          * TODO: Implement get_mr as wrapper around this
335          *       and deprecate it.
336          */
337         new_args.vec = args.vec;
338         new_args.cookie_addr = args.cookie_addr;
339         new_args.flags = args.flags;
340
341         return __rds_rdma_map(rs, &new_args, NULL, NULL);
342 }
343
344 /*
345  * Free the MR indicated by the given R_Key
346  */
347 int rds_free_mr(struct rds_sock *rs, char __user *optval, int optlen)
348 {
349         struct rds_free_mr_args args;
350         struct rds_mr *mr;
351         unsigned long flags;
352
353         if (optlen != sizeof(struct rds_free_mr_args))
354                 return -EINVAL;
355
356         if (copy_from_user(&args, (struct rds_free_mr_args __user *)optval,
357                            sizeof(struct rds_free_mr_args)))
358                 return -EFAULT;
359
360         /* Special case - a null cookie means flush all unused MRs */
361         if (args.cookie == 0) {
362                 if (!rs->rs_transport || !rs->rs_transport->flush_mrs)
363                         return -EINVAL;
364                 rs->rs_transport->flush_mrs();
365                 return 0;
366         }
367
368         /* Look up the MR given its R_key and remove it from the rbtree
369          * so nobody else finds it.
370          * This should also prevent races with rds_rdma_unuse.
371          */
372         spin_lock_irqsave(&rs->rs_rdma_lock, flags);
373         mr = rds_mr_tree_walk(&rs->rs_rdma_keys, rds_rdma_cookie_key(args.cookie), NULL);
374         if (mr) {
375                 rb_erase(&mr->r_rb_node, &rs->rs_rdma_keys);
376                 RB_CLEAR_NODE(&mr->r_rb_node);
377                 if (args.flags & RDS_RDMA_INVALIDATE)
378                         mr->r_invalidate = 1;
379         }
380         spin_unlock_irqrestore(&rs->rs_rdma_lock, flags);
381
382         if (!mr)
383                 return -EINVAL;
384
385         /*
386          * call rds_destroy_mr() ourselves so that we're sure it's done by the time
387          * we return.  If we let rds_mr_put() do it it might not happen until
388          * someone else drops their ref.
389          */
390         rds_destroy_mr(mr);
391         rds_mr_put(mr);
392         return 0;
393 }
394
395 /*
396  * This is called when we receive an extension header that
397  * tells us this MR was used. It allows us to implement
398  * use_once semantics
399  */
400 void rds_rdma_unuse(struct rds_sock *rs, u32 r_key, int force)
401 {
402         struct rds_mr *mr;
403         unsigned long flags;
404         int zot_me = 0;
405
406         spin_lock_irqsave(&rs->rs_rdma_lock, flags);
407         mr = rds_mr_tree_walk(&rs->rs_rdma_keys, r_key, NULL);
408         if (mr && (mr->r_use_once || force)) {
409                 rb_erase(&mr->r_rb_node, &rs->rs_rdma_keys);
410                 RB_CLEAR_NODE(&mr->r_rb_node);
411                 zot_me = 1;
412         } else if (mr)
413                 atomic_inc(&mr->r_refcount);
414         spin_unlock_irqrestore(&rs->rs_rdma_lock, flags);
415
416         /* May have to issue a dma_sync on this memory region.
417          * Note we could avoid this if the operation was a RDMA READ,
418          * but at this point we can't tell. */
419         if (mr != NULL) {
420                 if (mr->r_trans->sync_mr)
421                         mr->r_trans->sync_mr(mr->r_trans_private, DMA_FROM_DEVICE);
422
423                 /* If the MR was marked as invalidate, this will
424                  * trigger an async flush. */
425                 if (zot_me)
426                         rds_destroy_mr(mr);
427                 rds_mr_put(mr);
428         }
429 }
430
431 void rds_rdma_free_op(struct rds_rdma_op *ro)
432 {
433         unsigned int i;
434
435         for (i = 0; i < ro->r_nents; i++) {
436                 struct page *page = sg_page(&ro->r_sg[i]);
437
438                 /* Mark page dirty if it was possibly modified, which
439                  * is the case for a RDMA_READ which copies from remote
440                  * to local memory */
441                 if (!ro->r_write)
442                         set_page_dirty(page);
443                 put_page(page);
444         }
445
446         kfree(ro->r_notifier);
447         kfree(ro);
448 }
449
450 /*
451  * args is a pointer to an in-kernel copy in the sendmsg cmsg.
452  */
453 static struct rds_rdma_op *rds_rdma_prepare(struct rds_sock *rs,
454                                             struct rds_rdma_args *args)
455 {
456         struct rds_iovec vec;
457         struct rds_rdma_op *op = NULL;
458         unsigned int nr_pages;
459         unsigned int max_pages;
460         unsigned int nr_bytes;
461         struct page **pages = NULL;
462         struct rds_iovec __user *local_vec;
463         struct scatterlist *sg;
464         unsigned int nr;
465         unsigned int i, j;
466         int ret;
467
468
469         if (rs->rs_bound_addr == 0) {
470                 ret = -ENOTCONN; /* XXX not a great errno */
471                 goto out;
472         }
473
474         if (args->nr_local > (u64)UINT_MAX) {
475                 ret = -EMSGSIZE;
476                 goto out;
477         }
478
479         nr_pages = 0;
480         max_pages = 0;
481
482         local_vec = (struct rds_iovec __user *)(unsigned long) args->local_vec_addr;
483
484         /* figure out the number of pages in the vector */
485         for (i = 0; i < args->nr_local; i++) {
486                 if (copy_from_user(&vec, &local_vec[i],
487                                    sizeof(struct rds_iovec))) {
488                         ret = -EFAULT;
489                         goto out;
490                 }
491
492                 nr = rds_pages_in_vec(&vec);
493                 if (nr == 0) {
494                         ret = -EINVAL;
495                         goto out;
496                 }
497
498                 max_pages = max(nr, max_pages);
499                 nr_pages += nr;
500         }
501
502         pages = kcalloc(max_pages, sizeof(struct page *), GFP_KERNEL);
503         if (pages == NULL) {
504                 ret = -ENOMEM;
505                 goto out;
506         }
507
508         op = kzalloc(offsetof(struct rds_rdma_op, r_sg[nr_pages]), GFP_KERNEL);
509         if (op == NULL) {
510                 ret = -ENOMEM;
511                 goto out;
512         }
513
514         op->r_write = !!(args->flags & RDS_RDMA_READWRITE);
515         op->r_fence = !!(args->flags & RDS_RDMA_FENCE);
516         op->r_notify = !!(args->flags & RDS_RDMA_NOTIFY_ME);
517         op->r_recverr = rs->rs_recverr;
518         WARN_ON(!nr_pages);
519         sg_init_table(op->r_sg, nr_pages);
520
521         if (op->r_notify || op->r_recverr) {
522                 /* We allocate an uninitialized notifier here, because
523                  * we don't want to do that in the completion handler. We
524                  * would have to use GFP_ATOMIC there, and don't want to deal
525                  * with failed allocations.
526                  */
527                 op->r_notifier = kmalloc(sizeof(struct rds_notifier), GFP_KERNEL);
528                 if (!op->r_notifier) {
529                         ret = -ENOMEM;
530                         goto out;
531                 }
532                 op->r_notifier->n_user_token = args->user_token;
533                 op->r_notifier->n_status = RDS_RDMA_SUCCESS;
534         }
535
536         /* The cookie contains the R_Key of the remote memory region, and
537          * optionally an offset into it. This is how we implement RDMA into
538          * unaligned memory.
539          * When setting up the RDMA, we need to add that offset to the
540          * destination address (which is really an offset into the MR)
541          * FIXME: We may want to move this into ib_rdma.c
542          */
543         op->r_key = rds_rdma_cookie_key(args->cookie);
544         op->r_remote_addr = args->remote_vec.addr + rds_rdma_cookie_offset(args->cookie);
545
546         nr_bytes = 0;
547
548         rdsdebug("RDS: rdma prepare nr_local %llu rva %llx rkey %x\n",
549                (unsigned long long)args->nr_local,
550                (unsigned long long)args->remote_vec.addr,
551                op->r_key);
552
553         for (i = 0; i < args->nr_local; i++) {
554                 if (copy_from_user(&vec, &local_vec[i],
555                                    sizeof(struct rds_iovec))) {
556                         ret = -EFAULT;
557                         goto out;
558                 }
559
560                 nr = rds_pages_in_vec(&vec);
561                 if (nr == 0) {
562                         ret = -EINVAL;
563                         goto out;
564                 }
565
566                 rs->rs_user_addr = vec.addr;
567                 rs->rs_user_bytes = vec.bytes;
568
569                 /* did the user change the vec under us? */
570                 if (nr > max_pages || op->r_nents + nr > nr_pages) {
571                         ret = -EINVAL;
572                         goto out;
573                 }
574                 /* If it's a WRITE operation, we want to pin the pages for reading.
575                  * If it's a READ operation, we need to pin the pages for writing.
576                  */
577                 ret = rds_pin_pages(vec.addr & PAGE_MASK, nr, pages, !op->r_write);
578                 if (ret < 0)
579                         goto out;
580
581                 rdsdebug("RDS: nr_bytes %u nr %u vec.bytes %llu vec.addr %llx\n",
582                        nr_bytes, nr, vec.bytes, vec.addr);
583
584                 nr_bytes += vec.bytes;
585
586                 for (j = 0; j < nr; j++) {
587                         unsigned int offset = vec.addr & ~PAGE_MASK;
588
589                         sg = &op->r_sg[op->r_nents + j];
590                         sg_set_page(sg, pages[j],
591                                         min_t(unsigned int, vec.bytes, PAGE_SIZE - offset),
592                                         offset);
593
594                         rdsdebug("RDS: sg->offset %x sg->len %x vec.addr %llx vec.bytes %llu\n",
595                                sg->offset, sg->length, vec.addr, vec.bytes);
596
597                         vec.addr += sg->length;
598                         vec.bytes -= sg->length;
599                 }
600
601                 op->r_nents += nr;
602         }
603
604
605         if (nr_bytes > args->remote_vec.bytes) {
606                 rdsdebug("RDS nr_bytes %u remote_bytes %u do not match\n",
607                                 nr_bytes,
608                                 (unsigned int) args->remote_vec.bytes);
609                 ret = -EINVAL;
610                 goto out;
611         }
612         op->r_bytes = nr_bytes;
613
614         ret = 0;
615 out:
616         kfree(pages);
617         if (ret) {
618                 if (op)
619                         rds_rdma_free_op(op);
620                 op = ERR_PTR(ret);
621         }
622         return op;
623 }
624
625 /*
626  * The application asks for a RDMA transfer.
627  * Extract all arguments and set up the rdma_op
628  */
629 int rds_cmsg_rdma_args(struct rds_sock *rs, struct rds_message *rm,
630                           struct cmsghdr *cmsg)
631 {
632         struct rds_rdma_op *op;
633
634         if (cmsg->cmsg_len < CMSG_LEN(sizeof(struct rds_rdma_args)) ||
635             rm->m_rdma_op != NULL)
636                 return -EINVAL;
637
638         op = rds_rdma_prepare(rs, CMSG_DATA(cmsg));
639         if (IS_ERR(op))
640                 return PTR_ERR(op);
641         rds_stats_inc(s_send_rdma);
642         rm->m_rdma_op = op;
643         return 0;
644 }
645
646 /*
647  * The application wants us to pass an RDMA destination (aka MR)
648  * to the remote
649  */
650 int rds_cmsg_rdma_dest(struct rds_sock *rs, struct rds_message *rm,
651                           struct cmsghdr *cmsg)
652 {
653         unsigned long flags;
654         struct rds_mr *mr;
655         u32 r_key;
656         int err = 0;
657
658         if (cmsg->cmsg_len < CMSG_LEN(sizeof(rds_rdma_cookie_t)) ||
659             rm->m_rdma_cookie != 0)
660                 return -EINVAL;
661
662         memcpy(&rm->m_rdma_cookie, CMSG_DATA(cmsg), sizeof(rm->m_rdma_cookie));
663
664         /* We are reusing a previously mapped MR here. Most likely, the
665          * application has written to the buffer, so we need to explicitly
666          * flush those writes to RAM. Otherwise the HCA may not see them
667          * when doing a DMA from that buffer.
668          */
669         r_key = rds_rdma_cookie_key(rm->m_rdma_cookie);
670
671         spin_lock_irqsave(&rs->rs_rdma_lock, flags);
672         mr = rds_mr_tree_walk(&rs->rs_rdma_keys, r_key, NULL);
673         if (mr == NULL)
674                 err = -EINVAL;  /* invalid r_key */
675         else
676                 atomic_inc(&mr->r_refcount);
677         spin_unlock_irqrestore(&rs->rs_rdma_lock, flags);
678
679         if (mr) {
680                 mr->r_trans->sync_mr(mr->r_trans_private, DMA_TO_DEVICE);
681                 rm->m_rdma_mr = mr;
682         }
683         return err;
684 }
685
686 /*
687  * The application passes us an address range it wants to enable RDMA
688  * to/from. We map the area, and save the <R_Key,offset> pair
689  * in rm->m_rdma_cookie. This causes it to be sent along to the peer
690  * in an extension header.
691  */
692 int rds_cmsg_rdma_map(struct rds_sock *rs, struct rds_message *rm,
693                           struct cmsghdr *cmsg)
694 {
695         if (cmsg->cmsg_len < CMSG_LEN(sizeof(struct rds_get_mr_args)) ||
696             rm->m_rdma_cookie != 0)
697                 return -EINVAL;
698
699         return __rds_rdma_map(rs, CMSG_DATA(cmsg), &rm->m_rdma_cookie, &rm->m_rdma_mr);
700 }