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
.op_nents
; i
++) {
66 rdsdebug("putting data page %p\n", (void *)sg_page(&rm
->data
.op_sg
[i
]));
67 /* XXX will have to put_page for page refs */
68 __free_page(sg_page(&rm
->data
.op_sg
[i
]));
70 rm
->data
.op_nents
= 0;
72 if (rm
->rdma
.op_active
)
73 rds_rdma_free_op(&rm
->rdma
);
74 if (rm
->rdma
.op_rdma_mr
)
75 rds_mr_put(rm
->rdma
.op_rdma_mr
);
77 if (rm
->atomic
.op_active
)
78 rds_atomic_free_op(&rm
->atomic
);
79 if (rm
->atomic
.op_rdma_mr
)
80 rds_mr_put(rm
->atomic
.op_rdma_mr
);
83 void rds_message_put(struct rds_message
*rm
)
85 rdsdebug("put rm %p ref %d\n", rm
, atomic_read(&rm
->m_refcount
));
87 if (atomic_dec_and_test(&rm
->m_refcount
)) {
88 BUG_ON(!list_empty(&rm
->m_sock_item
));
89 BUG_ON(!list_empty(&rm
->m_conn_item
));
90 rds_message_purge(rm
);
95 EXPORT_SYMBOL_GPL(rds_message_put
);
97 void rds_message_populate_header(struct rds_header
*hdr
, __be16 sport
,
98 __be16 dport
, u64 seq
)
101 hdr
->h_sport
= sport
;
102 hdr
->h_dport
= dport
;
103 hdr
->h_sequence
= cpu_to_be64(seq
);
104 hdr
->h_exthdr
[0] = RDS_EXTHDR_NONE
;
106 EXPORT_SYMBOL_GPL(rds_message_populate_header
);
108 int rds_message_add_extension(struct rds_header
*hdr
,
109 unsigned int type
, const void *data
, unsigned int len
)
111 unsigned int ext_len
= sizeof(u8
) + len
;
114 /* For now, refuse to add more than one extension header */
115 if (hdr
->h_exthdr
[0] != RDS_EXTHDR_NONE
)
118 if (type
>= __RDS_EXTHDR_MAX
|| len
!= rds_exthdr_size
[type
])
121 if (ext_len
>= RDS_HEADER_EXT_SPACE
)
126 memcpy(dst
, data
, len
);
128 dst
[len
] = RDS_EXTHDR_NONE
;
131 EXPORT_SYMBOL_GPL(rds_message_add_extension
);
134 * If a message has extension headers, retrieve them here.
137 * unsigned int pos = 0;
140 * buflen = sizeof(buffer);
141 * type = rds_message_next_extension(hdr, &pos, buffer, &buflen);
142 * if (type == RDS_EXTHDR_NONE)
147 int rds_message_next_extension(struct rds_header
*hdr
,
148 unsigned int *pos
, void *buf
, unsigned int *buflen
)
150 unsigned int offset
, ext_type
, ext_len
;
151 u8
*src
= hdr
->h_exthdr
;
154 if (offset
>= RDS_HEADER_EXT_SPACE
)
157 /* Get the extension type and length. For now, the
158 * length is implied by the extension type. */
159 ext_type
= src
[offset
++];
161 if (ext_type
== RDS_EXTHDR_NONE
|| ext_type
>= __RDS_EXTHDR_MAX
)
163 ext_len
= rds_exthdr_size
[ext_type
];
164 if (offset
+ ext_len
> RDS_HEADER_EXT_SPACE
)
167 *pos
= offset
+ ext_len
;
168 if (ext_len
< *buflen
)
170 memcpy(buf
, src
+ offset
, *buflen
);
174 *pos
= RDS_HEADER_EXT_SPACE
;
176 return RDS_EXTHDR_NONE
;
179 int rds_message_add_version_extension(struct rds_header
*hdr
, unsigned int version
)
181 struct rds_ext_header_version ext_hdr
;
183 ext_hdr
.h_version
= cpu_to_be32(version
);
184 return rds_message_add_extension(hdr
, RDS_EXTHDR_VERSION
, &ext_hdr
, sizeof(ext_hdr
));
187 int rds_message_get_version_extension(struct rds_header
*hdr
, unsigned int *version
)
189 struct rds_ext_header_version ext_hdr
;
190 unsigned int pos
= 0, len
= sizeof(ext_hdr
);
192 /* We assume the version extension is the only one present */
193 if (rds_message_next_extension(hdr
, &pos
, &ext_hdr
, &len
) != RDS_EXTHDR_VERSION
)
195 *version
= be32_to_cpu(ext_hdr
.h_version
);
199 int rds_message_add_rdma_dest_extension(struct rds_header
*hdr
, u32 r_key
, u32 offset
)
201 struct rds_ext_header_rdma_dest ext_hdr
;
203 ext_hdr
.h_rdma_rkey
= cpu_to_be32(r_key
);
204 ext_hdr
.h_rdma_offset
= cpu_to_be32(offset
);
205 return rds_message_add_extension(hdr
, RDS_EXTHDR_RDMA_DEST
, &ext_hdr
, sizeof(ext_hdr
));
207 EXPORT_SYMBOL_GPL(rds_message_add_rdma_dest_extension
);
210 * Each rds_message is allocated with extra space for the scatterlist entries
211 * rds ops will need. This is to minimize memory allocation count. Then, each rds op
212 * can grab SGs when initializing its part of the rds_message.
214 struct rds_message
*rds_message_alloc(unsigned int extra_len
, gfp_t gfp
)
216 struct rds_message
*rm
;
218 rm
= kzalloc(sizeof(struct rds_message
) + extra_len
, gfp
);
223 rm
->m_total_sgs
= extra_len
/ sizeof(struct scatterlist
);
225 atomic_set(&rm
->m_refcount
, 1);
226 INIT_LIST_HEAD(&rm
->m_sock_item
);
227 INIT_LIST_HEAD(&rm
->m_conn_item
);
228 spin_lock_init(&rm
->m_rs_lock
);
235 * RDS ops use this to grab SG entries from the rm's sg pool.
237 struct scatterlist
*rds_message_alloc_sgs(struct rds_message
*rm
, int nents
)
239 struct scatterlist
*sg_first
= (struct scatterlist
*) &rm
[1];
240 struct scatterlist
*sg_ret
;
242 WARN_ON(rm
->m_used_sgs
+ nents
> rm
->m_total_sgs
);
244 sg_ret
= &sg_first
[rm
->m_used_sgs
];
245 sg_init_table(sg_ret
, nents
);
246 rm
->m_used_sgs
+= nents
;
251 struct rds_message
*rds_message_map_pages(unsigned long *page_addrs
, unsigned int total_len
)
253 struct rds_message
*rm
;
255 int num_sgs
= ceil(total_len
, PAGE_SIZE
);
256 int extra_bytes
= num_sgs
* sizeof(struct scatterlist
);
258 rm
= rds_message_alloc(extra_bytes
, GFP_KERNEL
);
260 return ERR_PTR(-ENOMEM
);
262 set_bit(RDS_MSG_PAGEVEC
, &rm
->m_flags
);
263 rm
->m_inc
.i_hdr
.h_len
= cpu_to_be32(total_len
);
264 rm
->data
.op_nents
= ceil(total_len
, PAGE_SIZE
);
265 rm
->data
.op_sg
= rds_message_alloc_sgs(rm
, num_sgs
);
267 for (i
= 0; i
< rm
->data
.op_nents
; ++i
) {
268 sg_set_page(&rm
->data
.op_sg
[i
],
269 virt_to_page(page_addrs
[i
]),
276 int rds_message_copy_from_user(struct rds_message
*rm
, struct iovec
*first_iov
,
279 unsigned long to_copy
;
280 unsigned long iov_off
;
281 unsigned long sg_off
;
283 struct scatterlist
*sg
;
286 rm
->m_inc
.i_hdr
.h_len
= cpu_to_be32(total_len
);
289 * now allocate and copy in the data payload.
294 sg_off
= 0; /* Dear gcc, sg->page will be null from kzalloc. */
298 ret
= rds_page_remainder_alloc(sg
, total_len
,
306 while (iov_off
== iov
->iov_len
) {
311 to_copy
= min(iov
->iov_len
- iov_off
, sg
->length
- sg_off
);
312 to_copy
= min_t(size_t, to_copy
, total_len
);
314 rdsdebug("copying %lu bytes from user iov [%p, %zu] + %lu to "
315 "sg [%p, %u, %u] + %lu\n",
316 to_copy
, iov
->iov_base
, iov
->iov_len
, iov_off
,
317 (void *)sg_page(sg
), sg
->offset
, sg
->length
, sg_off
);
319 ret
= rds_page_copy_from_user(sg_page(sg
), sg
->offset
+ sg_off
,
320 iov
->iov_base
+ iov_off
,
326 total_len
-= to_copy
;
329 if (sg_off
== sg
->length
)
333 rm
->data
.op_active
= 1;
339 int rds_message_inc_copy_to_user(struct rds_incoming
*inc
,
340 struct iovec
*first_iov
, size_t size
)
342 struct rds_message
*rm
;
344 struct scatterlist
*sg
;
345 unsigned long to_copy
;
346 unsigned long iov_off
;
347 unsigned long vec_off
;
352 rm
= container_of(inc
, struct rds_message
, m_inc
);
353 len
= be32_to_cpu(rm
->m_inc
.i_hdr
.h_len
);
361 while (copied
< size
&& copied
< len
) {
362 while (iov_off
== iov
->iov_len
) {
367 to_copy
= min(iov
->iov_len
- iov_off
, sg
->length
- vec_off
);
368 to_copy
= min_t(size_t, to_copy
, size
- copied
);
369 to_copy
= min_t(unsigned long, to_copy
, len
- copied
);
371 rdsdebug("copying %lu bytes to user iov [%p, %zu] + %lu to "
372 "sg [%p, %u, %u] + %lu\n",
373 to_copy
, iov
->iov_base
, iov
->iov_len
, iov_off
,
374 sg_page(sg
), sg
->offset
, sg
->length
, vec_off
);
376 ret
= rds_page_copy_to_user(sg_page(sg
), sg
->offset
+ vec_off
,
377 iov
->iov_base
+ iov_off
,
388 if (vec_off
== sg
->length
) {
398 * If the message is still on the send queue, wait until the transport
399 * is done with it. This is particularly important for RDMA operations.
401 void rds_message_wait(struct rds_message
*rm
)
403 wait_event(rds_message_flush_waitq
,
404 !test_bit(RDS_MSG_MAPPED
, &rm
->m_flags
));
407 void rds_message_unmapped(struct rds_message
*rm
)
409 clear_bit(RDS_MSG_MAPPED
, &rm
->m_flags
);
410 if (waitqueue_active(&rds_message_flush_waitq
))
411 wake_up(&rds_message_flush_waitq
);
413 EXPORT_SYMBOL_GPL(rds_message_unmapped
);