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 unsigned int rds_exthdr_size
[__RDS_EXTHDR_MAX
] = {
39 [RDS_EXTHDR_NONE
] = 0,
40 [RDS_EXTHDR_VERSION
] = sizeof(struct rds_ext_header_version
),
41 [RDS_EXTHDR_RDMA
] = sizeof(struct rds_ext_header_rdma
),
42 [RDS_EXTHDR_RDMA_DEST
] = sizeof(struct rds_ext_header_rdma_dest
),
46 void rds_message_addref(struct rds_message
*rm
)
48 rdsdebug("addref rm %p ref %d\n", rm
, atomic_read(&rm
->m_refcount
));
49 atomic_inc(&rm
->m_refcount
);
51 EXPORT_SYMBOL_GPL(rds_message_addref
);
54 * This relies on dma_map_sg() not touching sg[].page during merging.
56 static void rds_message_purge(struct rds_message
*rm
)
60 if (unlikely(test_bit(RDS_MSG_PAGEVEC
, &rm
->m_flags
)))
63 for (i
= 0; i
< rm
->data
.op_nents
; i
++) {
64 rdsdebug("putting data page %p\n", (void *)sg_page(&rm
->data
.op_sg
[i
]));
65 /* XXX will have to put_page for page refs */
66 __free_page(sg_page(&rm
->data
.op_sg
[i
]));
68 rm
->data
.op_nents
= 0;
70 if (rm
->rdma
.op_active
)
71 rds_rdma_free_op(&rm
->rdma
);
72 if (rm
->rdma
.op_rdma_mr
)
73 rds_mr_put(rm
->rdma
.op_rdma_mr
);
75 if (rm
->atomic
.op_active
)
76 rds_atomic_free_op(&rm
->atomic
);
77 if (rm
->atomic
.op_rdma_mr
)
78 rds_mr_put(rm
->atomic
.op_rdma_mr
);
81 void rds_message_put(struct rds_message
*rm
)
83 rdsdebug("put rm %p ref %d\n", rm
, atomic_read(&rm
->m_refcount
));
84 if (atomic_read(&rm
->m_refcount
) == 0) {
85 printk(KERN_CRIT
"danger refcount zero on %p\n", rm
);
88 if (atomic_dec_and_test(&rm
->m_refcount
)) {
89 BUG_ON(!list_empty(&rm
->m_sock_item
));
90 BUG_ON(!list_empty(&rm
->m_conn_item
));
91 rds_message_purge(rm
);
96 EXPORT_SYMBOL_GPL(rds_message_put
);
98 void rds_message_populate_header(struct rds_header
*hdr
, __be16 sport
,
99 __be16 dport
, u64 seq
)
102 hdr
->h_sport
= sport
;
103 hdr
->h_dport
= dport
;
104 hdr
->h_sequence
= cpu_to_be64(seq
);
105 hdr
->h_exthdr
[0] = RDS_EXTHDR_NONE
;
107 EXPORT_SYMBOL_GPL(rds_message_populate_header
);
109 int rds_message_add_extension(struct rds_header
*hdr
,
110 unsigned int type
, const void *data
, unsigned int len
)
112 unsigned int ext_len
= sizeof(u8
) + len
;
115 /* For now, refuse to add more than one extension header */
116 if (hdr
->h_exthdr
[0] != RDS_EXTHDR_NONE
)
119 if (type
>= __RDS_EXTHDR_MAX
|| len
!= rds_exthdr_size
[type
])
122 if (ext_len
>= RDS_HEADER_EXT_SPACE
)
127 memcpy(dst
, data
, len
);
129 dst
[len
] = RDS_EXTHDR_NONE
;
132 EXPORT_SYMBOL_GPL(rds_message_add_extension
);
135 * If a message has extension headers, retrieve them here.
138 * unsigned int pos = 0;
141 * buflen = sizeof(buffer);
142 * type = rds_message_next_extension(hdr, &pos, buffer, &buflen);
143 * if (type == RDS_EXTHDR_NONE)
148 int rds_message_next_extension(struct rds_header
*hdr
,
149 unsigned int *pos
, void *buf
, unsigned int *buflen
)
151 unsigned int offset
, ext_type
, ext_len
;
152 u8
*src
= hdr
->h_exthdr
;
155 if (offset
>= RDS_HEADER_EXT_SPACE
)
158 /* Get the extension type and length. For now, the
159 * length is implied by the extension type. */
160 ext_type
= src
[offset
++];
162 if (ext_type
== RDS_EXTHDR_NONE
|| ext_type
>= __RDS_EXTHDR_MAX
)
164 ext_len
= rds_exthdr_size
[ext_type
];
165 if (offset
+ ext_len
> RDS_HEADER_EXT_SPACE
)
168 *pos
= offset
+ ext_len
;
169 if (ext_len
< *buflen
)
171 memcpy(buf
, src
+ offset
, *buflen
);
175 *pos
= RDS_HEADER_EXT_SPACE
;
177 return RDS_EXTHDR_NONE
;
180 int rds_message_add_version_extension(struct rds_header
*hdr
, unsigned int version
)
182 struct rds_ext_header_version ext_hdr
;
184 ext_hdr
.h_version
= cpu_to_be32(version
);
185 return rds_message_add_extension(hdr
, RDS_EXTHDR_VERSION
, &ext_hdr
, sizeof(ext_hdr
));
188 int rds_message_get_version_extension(struct rds_header
*hdr
, unsigned int *version
)
190 struct rds_ext_header_version ext_hdr
;
191 unsigned int pos
= 0, len
= sizeof(ext_hdr
);
193 /* We assume the version extension is the only one present */
194 if (rds_message_next_extension(hdr
, &pos
, &ext_hdr
, &len
) != RDS_EXTHDR_VERSION
)
196 *version
= be32_to_cpu(ext_hdr
.h_version
);
200 int rds_message_add_rdma_dest_extension(struct rds_header
*hdr
, u32 r_key
, u32 offset
)
202 struct rds_ext_header_rdma_dest ext_hdr
;
204 ext_hdr
.h_rdma_rkey
= cpu_to_be32(r_key
);
205 ext_hdr
.h_rdma_offset
= cpu_to_be32(offset
);
206 return rds_message_add_extension(hdr
, RDS_EXTHDR_RDMA_DEST
, &ext_hdr
, sizeof(ext_hdr
));
208 EXPORT_SYMBOL_GPL(rds_message_add_rdma_dest_extension
);
211 * Each rds_message is allocated with extra space for the scatterlist entries
212 * rds ops will need. This is to minimize memory allocation count. Then, each rds op
213 * can grab SGs when initializing its part of the rds_message.
215 struct rds_message
*rds_message_alloc(unsigned int extra_len
, gfp_t gfp
)
217 struct rds_message
*rm
;
219 rm
= kzalloc(sizeof(struct rds_message
) + extra_len
, gfp
);
224 rm
->m_total_sgs
= extra_len
/ sizeof(struct scatterlist
);
226 atomic_set(&rm
->m_refcount
, 1);
227 INIT_LIST_HEAD(&rm
->m_sock_item
);
228 INIT_LIST_HEAD(&rm
->m_conn_item
);
229 spin_lock_init(&rm
->m_rs_lock
);
230 init_waitqueue_head(&rm
->m_flush_wait
);
237 * RDS ops use this to grab SG entries from the rm's sg pool.
239 struct scatterlist
*rds_message_alloc_sgs(struct rds_message
*rm
, int nents
)
241 struct scatterlist
*sg_first
= (struct scatterlist
*) &rm
[1];
242 struct scatterlist
*sg_ret
;
244 WARN_ON(rm
->m_used_sgs
+ nents
> rm
->m_total_sgs
);
247 sg_ret
= &sg_first
[rm
->m_used_sgs
];
248 sg_init_table(sg_ret
, nents
);
249 rm
->m_used_sgs
+= nents
;
254 struct rds_message
*rds_message_map_pages(unsigned long *page_addrs
, unsigned int total_len
)
256 struct rds_message
*rm
;
258 int num_sgs
= ceil(total_len
, PAGE_SIZE
);
259 int extra_bytes
= num_sgs
* sizeof(struct scatterlist
);
261 rm
= rds_message_alloc(extra_bytes
, GFP_NOWAIT
);
263 return ERR_PTR(-ENOMEM
);
265 set_bit(RDS_MSG_PAGEVEC
, &rm
->m_flags
);
266 rm
->m_inc
.i_hdr
.h_len
= cpu_to_be32(total_len
);
267 rm
->data
.op_nents
= ceil(total_len
, PAGE_SIZE
);
268 rm
->data
.op_sg
= rds_message_alloc_sgs(rm
, num_sgs
);
270 for (i
= 0; i
< rm
->data
.op_nents
; ++i
) {
271 sg_set_page(&rm
->data
.op_sg
[i
],
272 virt_to_page(page_addrs
[i
]),
279 int rds_message_copy_from_user(struct rds_message
*rm
, struct iovec
*first_iov
,
282 unsigned long to_copy
;
283 unsigned long iov_off
;
284 unsigned long sg_off
;
286 struct scatterlist
*sg
;
289 rm
->m_inc
.i_hdr
.h_len
= cpu_to_be32(total_len
);
292 * now allocate and copy in the data payload.
297 sg_off
= 0; /* Dear gcc, sg->page will be null from kzalloc. */
301 ret
= rds_page_remainder_alloc(sg
, total_len
,
309 while (iov_off
== iov
->iov_len
) {
314 to_copy
= min(iov
->iov_len
- iov_off
, sg
->length
- sg_off
);
315 to_copy
= min_t(size_t, to_copy
, total_len
);
317 rdsdebug("copying %lu bytes from user iov [%p, %zu] + %lu to "
318 "sg [%p, %u, %u] + %lu\n",
319 to_copy
, iov
->iov_base
, iov
->iov_len
, iov_off
,
320 (void *)sg_page(sg
), sg
->offset
, sg
->length
, sg_off
);
322 ret
= rds_page_copy_from_user(sg_page(sg
), sg
->offset
+ sg_off
,
323 iov
->iov_base
+ iov_off
,
329 total_len
-= to_copy
;
332 if (sg_off
== sg
->length
)
340 int rds_message_inc_copy_to_user(struct rds_incoming
*inc
,
341 struct iovec
*first_iov
, size_t size
)
343 struct rds_message
*rm
;
345 struct scatterlist
*sg
;
346 unsigned long to_copy
;
347 unsigned long iov_off
;
348 unsigned long vec_off
;
353 rm
= container_of(inc
, struct rds_message
, m_inc
);
354 len
= be32_to_cpu(rm
->m_inc
.i_hdr
.h_len
);
362 while (copied
< size
&& copied
< len
) {
363 while (iov_off
== iov
->iov_len
) {
368 to_copy
= min(iov
->iov_len
- iov_off
, sg
->length
- vec_off
);
369 to_copy
= min_t(size_t, to_copy
, size
- copied
);
370 to_copy
= min_t(unsigned long, to_copy
, len
- copied
);
372 rdsdebug("copying %lu bytes to user iov [%p, %zu] + %lu to "
373 "sg [%p, %u, %u] + %lu\n",
374 to_copy
, iov
->iov_base
, iov
->iov_len
, iov_off
,
375 sg_page(sg
), sg
->offset
, sg
->length
, vec_off
);
377 ret
= rds_page_copy_to_user(sg_page(sg
), sg
->offset
+ vec_off
,
378 iov
->iov_base
+ iov_off
,
389 if (vec_off
== sg
->length
) {
399 * If the message is still on the send queue, wait until the transport
400 * is done with it. This is particularly important for RDMA operations.
402 void rds_message_wait(struct rds_message
*rm
)
404 wait_event_interruptible(rm
->m_flush_wait
,
405 !test_bit(RDS_MSG_MAPPED
, &rm
->m_flags
));
408 void rds_message_unmapped(struct rds_message
*rm
)
410 clear_bit(RDS_MSG_MAPPED
, &rm
->m_flags
);
411 wake_up_interruptible(&rm
->m_flush_wait
);
413 EXPORT_SYMBOL_GPL(rds_message_unmapped
);