2 * Copyright (c) 2006 Oracle. All rights reserved.
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:
10 * Redistribution and use in source and binary forms, with or
11 * without modification, are permitted provided that the following
14 * - Redistributions of source code must retain the above
15 * copyright notice, this list of conditions and the following
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.
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
33 #include <linux/kernel.h>
34 #include <linux/slab.h>
38 static DECLARE_WAIT_QUEUE_HEAD(rds_message_flush_waitq
);
40 static unsigned int rds_exthdr_size
[__RDS_EXTHDR_MAX
] = {
41 [RDS_EXTHDR_NONE
] = 0,
42 [RDS_EXTHDR_VERSION
] = sizeof(struct rds_ext_header_version
),
43 [RDS_EXTHDR_RDMA
] = sizeof(struct rds_ext_header_rdma
),
44 [RDS_EXTHDR_RDMA_DEST
] = sizeof(struct rds_ext_header_rdma_dest
),
48 void rds_message_addref(struct rds_message
*rm
)
50 rdsdebug("addref rm %p ref %d\n", rm
, atomic_read(&rm
->m_refcount
));
51 atomic_inc(&rm
->m_refcount
);
53 EXPORT_SYMBOL_GPL(rds_message_addref
);
56 * This relies on dma_map_sg() not touching sg[].page during merging.
58 static void rds_message_purge(struct rds_message
*rm
)
62 if (unlikely(test_bit(RDS_MSG_PAGEVEC
, &rm
->m_flags
)))
65 for (i
= 0; i
< rm
->data
.m_nents
; i
++) {
66 rdsdebug("putting data page %p\n", (void *)sg_page(&rm
->data
.m_sg
[i
]));
67 /* XXX will have to put_page for page refs */
68 __free_page(sg_page(&rm
->data
.m_sg
[i
]));
72 if (rm
->rdma
.m_rdma_op
.r_active
)
73 rds_rdma_free_op(&rm
->rdma
.m_rdma_op
);
74 if (rm
->rdma
.m_rdma_mr
)
75 rds_mr_put(rm
->rdma
.m_rdma_mr
);
78 void rds_message_put(struct rds_message
*rm
)
80 rdsdebug("put rm %p ref %d\n", rm
, atomic_read(&rm
->m_refcount
));
82 if (atomic_dec_and_test(&rm
->m_refcount
)) {
83 BUG_ON(!list_empty(&rm
->m_sock_item
));
84 BUG_ON(!list_empty(&rm
->m_conn_item
));
85 rds_message_purge(rm
);
90 EXPORT_SYMBOL_GPL(rds_message_put
);
92 void rds_message_populate_header(struct rds_header
*hdr
, __be16 sport
,
93 __be16 dport
, u64 seq
)
98 hdr
->h_sequence
= cpu_to_be64(seq
);
99 hdr
->h_exthdr
[0] = RDS_EXTHDR_NONE
;
101 EXPORT_SYMBOL_GPL(rds_message_populate_header
);
103 int rds_message_add_extension(struct rds_header
*hdr
,
104 unsigned int type
, const void *data
, unsigned int len
)
106 unsigned int ext_len
= sizeof(u8
) + len
;
109 /* For now, refuse to add more than one extension header */
110 if (hdr
->h_exthdr
[0] != RDS_EXTHDR_NONE
)
113 if (type
>= __RDS_EXTHDR_MAX
|| len
!= rds_exthdr_size
[type
])
116 if (ext_len
>= RDS_HEADER_EXT_SPACE
)
121 memcpy(dst
, data
, len
);
123 dst
[len
] = RDS_EXTHDR_NONE
;
126 EXPORT_SYMBOL_GPL(rds_message_add_extension
);
129 * If a message has extension headers, retrieve them here.
132 * unsigned int pos = 0;
135 * buflen = sizeof(buffer);
136 * type = rds_message_next_extension(hdr, &pos, buffer, &buflen);
137 * if (type == RDS_EXTHDR_NONE)
142 int rds_message_next_extension(struct rds_header
*hdr
,
143 unsigned int *pos
, void *buf
, unsigned int *buflen
)
145 unsigned int offset
, ext_type
, ext_len
;
146 u8
*src
= hdr
->h_exthdr
;
149 if (offset
>= RDS_HEADER_EXT_SPACE
)
152 /* Get the extension type and length. For now, the
153 * length is implied by the extension type. */
154 ext_type
= src
[offset
++];
156 if (ext_type
== RDS_EXTHDR_NONE
|| ext_type
>= __RDS_EXTHDR_MAX
)
158 ext_len
= rds_exthdr_size
[ext_type
];
159 if (offset
+ ext_len
> RDS_HEADER_EXT_SPACE
)
162 *pos
= offset
+ ext_len
;
163 if (ext_len
< *buflen
)
165 memcpy(buf
, src
+ offset
, *buflen
);
169 *pos
= RDS_HEADER_EXT_SPACE
;
171 return RDS_EXTHDR_NONE
;
174 int rds_message_add_version_extension(struct rds_header
*hdr
, unsigned int version
)
176 struct rds_ext_header_version ext_hdr
;
178 ext_hdr
.h_version
= cpu_to_be32(version
);
179 return rds_message_add_extension(hdr
, RDS_EXTHDR_VERSION
, &ext_hdr
, sizeof(ext_hdr
));
182 int rds_message_get_version_extension(struct rds_header
*hdr
, unsigned int *version
)
184 struct rds_ext_header_version ext_hdr
;
185 unsigned int pos
= 0, len
= sizeof(ext_hdr
);
187 /* We assume the version extension is the only one present */
188 if (rds_message_next_extension(hdr
, &pos
, &ext_hdr
, &len
) != RDS_EXTHDR_VERSION
)
190 *version
= be32_to_cpu(ext_hdr
.h_version
);
194 int rds_message_add_rdma_dest_extension(struct rds_header
*hdr
, u32 r_key
, u32 offset
)
196 struct rds_ext_header_rdma_dest ext_hdr
;
198 ext_hdr
.h_rdma_rkey
= cpu_to_be32(r_key
);
199 ext_hdr
.h_rdma_offset
= cpu_to_be32(offset
);
200 return rds_message_add_extension(hdr
, RDS_EXTHDR_RDMA_DEST
, &ext_hdr
, sizeof(ext_hdr
));
202 EXPORT_SYMBOL_GPL(rds_message_add_rdma_dest_extension
);
205 * Each rds_message is allocated with extra space for the scatterlist entries
206 * rds ops will need. This is to minimize memory allocation count. Then, each rds op
207 * can grab SGs when initializing its part of the rds_message.
209 struct rds_message
*rds_message_alloc(unsigned int extra_len
, gfp_t gfp
)
211 struct rds_message
*rm
;
213 rm
= kzalloc(sizeof(struct rds_message
) + extra_len
, gfp
);
218 rm
->m_total_sgs
= extra_len
/ sizeof(struct scatterlist
);
220 atomic_set(&rm
->m_refcount
, 1);
221 INIT_LIST_HEAD(&rm
->m_sock_item
);
222 INIT_LIST_HEAD(&rm
->m_conn_item
);
223 spin_lock_init(&rm
->m_rs_lock
);
230 * RDS ops use this to grab SG entries from the rm's sg pool.
232 struct scatterlist
*rds_message_alloc_sgs(struct rds_message
*rm
, int nents
)
234 struct scatterlist
*sg_first
= (struct scatterlist
*) &rm
[1];
235 struct scatterlist
*sg_ret
;
237 WARN_ON(rm
->m_used_sgs
+ nents
> rm
->m_total_sgs
);
239 sg_ret
= &sg_first
[rm
->m_used_sgs
];
240 sg_init_table(sg_ret
, nents
);
241 rm
->m_used_sgs
+= nents
;
246 struct rds_message
*rds_message_map_pages(unsigned long *page_addrs
, unsigned int total_len
)
248 struct rds_message
*rm
;
250 int num_sgs
= ceil(total_len
, PAGE_SIZE
);
251 int extra_bytes
= num_sgs
* sizeof(struct scatterlist
);
253 rm
= rds_message_alloc(extra_bytes
, GFP_KERNEL
);
255 return ERR_PTR(-ENOMEM
);
257 set_bit(RDS_MSG_PAGEVEC
, &rm
->m_flags
);
258 rm
->m_inc
.i_hdr
.h_len
= cpu_to_be32(total_len
);
259 rm
->data
.m_nents
= ceil(total_len
, PAGE_SIZE
);
260 rm
->data
.m_sg
= rds_message_alloc_sgs(rm
, num_sgs
);
262 for (i
= 0; i
< rm
->data
.m_nents
; ++i
) {
263 sg_set_page(&rm
->data
.m_sg
[i
],
264 virt_to_page(page_addrs
[i
]),
271 int rds_message_copy_from_user(struct rds_message
*rm
, struct iovec
*first_iov
,
274 unsigned long to_copy
;
275 unsigned long iov_off
;
276 unsigned long sg_off
;
278 struct scatterlist
*sg
;
281 rm
->m_inc
.i_hdr
.h_len
= cpu_to_be32(total_len
);
284 * now allocate and copy in the data payload.
289 sg_off
= 0; /* Dear gcc, sg->page will be null from kzalloc. */
293 ret
= rds_page_remainder_alloc(sg
, total_len
,
301 while (iov_off
== iov
->iov_len
) {
306 to_copy
= min(iov
->iov_len
- iov_off
, sg
->length
- sg_off
);
307 to_copy
= min_t(size_t, to_copy
, total_len
);
309 rdsdebug("copying %lu bytes from user iov [%p, %zu] + %lu to "
310 "sg [%p, %u, %u] + %lu\n",
311 to_copy
, iov
->iov_base
, iov
->iov_len
, iov_off
,
312 (void *)sg_page(sg
), sg
->offset
, sg
->length
, sg_off
);
314 ret
= rds_page_copy_from_user(sg_page(sg
), sg
->offset
+ sg_off
,
315 iov
->iov_base
+ iov_off
,
321 total_len
-= to_copy
;
324 if (sg_off
== sg
->length
)
328 rm
->data
.op_active
= 1;
334 int rds_message_inc_copy_to_user(struct rds_incoming
*inc
,
335 struct iovec
*first_iov
, size_t size
)
337 struct rds_message
*rm
;
339 struct scatterlist
*sg
;
340 unsigned long to_copy
;
341 unsigned long iov_off
;
342 unsigned long vec_off
;
347 rm
= container_of(inc
, struct rds_message
, m_inc
);
348 len
= be32_to_cpu(rm
->m_inc
.i_hdr
.h_len
);
356 while (copied
< size
&& copied
< len
) {
357 while (iov_off
== iov
->iov_len
) {
362 to_copy
= min(iov
->iov_len
- iov_off
, sg
->length
- vec_off
);
363 to_copy
= min_t(size_t, to_copy
, size
- copied
);
364 to_copy
= min_t(unsigned long, to_copy
, len
- copied
);
366 rdsdebug("copying %lu bytes to user iov [%p, %zu] + %lu to "
367 "sg [%p, %u, %u] + %lu\n",
368 to_copy
, iov
->iov_base
, iov
->iov_len
, iov_off
,
369 sg_page(sg
), sg
->offset
, sg
->length
, vec_off
);
371 ret
= rds_page_copy_to_user(sg_page(sg
), sg
->offset
+ vec_off
,
372 iov
->iov_base
+ iov_off
,
383 if (vec_off
== sg
->length
) {
393 * If the message is still on the send queue, wait until the transport
394 * is done with it. This is particularly important for RDMA operations.
396 void rds_message_wait(struct rds_message
*rm
)
398 wait_event(rds_message_flush_waitq
,
399 !test_bit(RDS_MSG_MAPPED
, &rm
->m_flags
));
402 void rds_message_unmapped(struct rds_message
*rm
)
404 clear_bit(RDS_MSG_MAPPED
, &rm
->m_flags
);
405 if (waitqueue_active(&rds_message_flush_waitq
))
406 wake_up(&rds_message_flush_waitq
);
408 EXPORT_SYMBOL_GPL(rds_message_unmapped
);