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>
35 #include <linux/vmalloc.h>
41 * Set the selected protocol version
43 static void rds_ib_set_protocol(struct rds_connection
*conn
, unsigned int version
)
45 conn
->c_version
= version
;
51 static void rds_ib_set_flow_control(struct rds_connection
*conn
, u32 credits
)
53 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
55 if (rds_ib_sysctl_flow_control
&& credits
!= 0) {
56 /* We're doing flow control */
58 rds_ib_send_add_credits(conn
, credits
);
65 * Tune RNR behavior. Without flow control, we use a rather
66 * low timeout, but not the absolute minimum - this should
69 * We already set the RNR retry count to 7 (which is the
70 * smallest infinite number :-) above.
71 * If flow control is off, we want to change this back to 0
72 * so that we learn quickly when our credit accounting is
75 * Caller passes in a qp_attr pointer - don't waste stack spacv
76 * by allocation this twice.
79 rds_ib_tune_rnr(struct rds_ib_connection
*ic
, struct ib_qp_attr
*attr
)
83 attr
->min_rnr_timer
= IB_RNR_TIMER_000_32
;
84 ret
= ib_modify_qp(ic
->i_cm_id
->qp
, attr
, IB_QP_MIN_RNR_TIMER
);
86 printk(KERN_NOTICE
"ib_modify_qp(IB_QP_MIN_RNR_TIMER): err=%d\n", -ret
);
90 * Connection established.
91 * We get here for both outgoing and incoming connection.
93 void rds_ib_cm_connect_complete(struct rds_connection
*conn
, struct rdma_cm_event
*event
)
95 const struct rds_ib_connect_private
*dp
= NULL
;
96 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
97 struct rds_ib_device
*rds_ibdev
;
98 struct ib_qp_attr qp_attr
;
101 if (event
->param
.conn
.private_data_len
>= sizeof(*dp
)) {
102 dp
= event
->param
.conn
.private_data
;
104 /* make sure it isn't empty data */
105 if (dp
->dp_protocol_major
) {
106 rds_ib_set_protocol(conn
,
107 RDS_PROTOCOL(dp
->dp_protocol_major
,
108 dp
->dp_protocol_minor
));
109 rds_ib_set_flow_control(conn
, be32_to_cpu(dp
->dp_credit
));
113 printk(KERN_NOTICE
"RDS/IB: connected to %pI4 version %u.%u%s\n",
115 RDS_PROTOCOL_MAJOR(conn
->c_version
),
116 RDS_PROTOCOL_MINOR(conn
->c_version
),
117 ic
->i_flowctl
? ", flow control" : "");
119 /* Tune RNR behavior */
120 rds_ib_tune_rnr(ic
, &qp_attr
);
122 qp_attr
.qp_state
= IB_QPS_RTS
;
123 err
= ib_modify_qp(ic
->i_cm_id
->qp
, &qp_attr
, IB_QP_STATE
);
125 printk(KERN_NOTICE
"ib_modify_qp(IB_QP_STATE, RTS): err=%d\n", err
);
127 /* update ib_device with this local ipaddr & conn */
128 rds_ibdev
= ib_get_client_data(ic
->i_cm_id
->device
, &rds_ib_client
);
129 err
= rds_ib_update_ipaddr(rds_ibdev
, conn
->c_laddr
);
131 printk(KERN_ERR
"rds_ib_update_ipaddr failed (%d)\n", err
);
132 rds_ib_add_conn(rds_ibdev
, conn
);
134 /* If the peer gave us the last packet it saw, process this as if
135 * we had received a regular ACK. */
136 if (dp
&& dp
->dp_ack_seq
)
137 rds_send_drop_acked(conn
, be64_to_cpu(dp
->dp_ack_seq
), NULL
);
139 rds_connect_complete(conn
);
142 static void rds_ib_cm_fill_conn_param(struct rds_connection
*conn
,
143 struct rdma_conn_param
*conn_param
,
144 struct rds_ib_connect_private
*dp
,
145 u32 protocol_version
)
147 memset(conn_param
, 0, sizeof(struct rdma_conn_param
));
148 /* XXX tune these? */
149 conn_param
->responder_resources
= 1;
150 conn_param
->initiator_depth
= 1;
151 conn_param
->retry_count
= min_t(unsigned int, rds_ib_retry_count
, 7);
152 conn_param
->rnr_retry_count
= 7;
155 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
157 memset(dp
, 0, sizeof(*dp
));
158 dp
->dp_saddr
= conn
->c_laddr
;
159 dp
->dp_daddr
= conn
->c_faddr
;
160 dp
->dp_protocol_major
= RDS_PROTOCOL_MAJOR(protocol_version
);
161 dp
->dp_protocol_minor
= RDS_PROTOCOL_MINOR(protocol_version
);
162 dp
->dp_protocol_minor_mask
= cpu_to_be16(RDS_IB_SUPPORTED_PROTOCOLS
);
163 dp
->dp_ack_seq
= rds_ib_piggyb_ack(ic
);
165 /* Advertise flow control */
167 unsigned int credits
;
169 credits
= IB_GET_POST_CREDITS(atomic_read(&ic
->i_credits
));
170 dp
->dp_credit
= cpu_to_be32(credits
);
171 atomic_sub(IB_SET_POST_CREDITS(credits
), &ic
->i_credits
);
174 conn_param
->private_data
= dp
;
175 conn_param
->private_data_len
= sizeof(*dp
);
179 static void rds_ib_cq_event_handler(struct ib_event
*event
, void *data
)
181 rdsdebug("event %u data %p\n", event
->event
, data
);
184 static void rds_ib_qp_event_handler(struct ib_event
*event
, void *data
)
186 struct rds_connection
*conn
= data
;
187 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
189 rdsdebug("conn %p ic %p event %u\n", conn
, ic
, event
->event
);
191 switch (event
->event
) {
192 case IB_EVENT_COMM_EST
:
193 rdma_notify(ic
->i_cm_id
, IB_EVENT_COMM_EST
);
196 printk(KERN_WARNING
"RDS/ib: unhandled QP event %u "
197 "on connection to %pI4\n", event
->event
,
204 * This needs to be very careful to not leave IS_ERR pointers around for
205 * cleanup to trip over.
207 static int rds_ib_setup_qp(struct rds_connection
*conn
)
209 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
210 struct ib_device
*dev
= ic
->i_cm_id
->device
;
211 struct ib_qp_init_attr attr
;
212 struct rds_ib_device
*rds_ibdev
;
215 /* rds_ib_add_one creates a rds_ib_device object per IB device,
216 * and allocates a protection domain, memory range and FMR pool
217 * for each. If that fails for any reason, it will not register
218 * the rds_ibdev at all.
220 rds_ibdev
= ib_get_client_data(dev
, &rds_ib_client
);
221 if (rds_ibdev
== NULL
) {
222 if (printk_ratelimit())
223 printk(KERN_NOTICE
"RDS/IB: No client_data for device %s\n",
228 if (rds_ibdev
->max_wrs
< ic
->i_send_ring
.w_nr
+ 1)
229 rds_ib_ring_resize(&ic
->i_send_ring
, rds_ibdev
->max_wrs
- 1);
230 if (rds_ibdev
->max_wrs
< ic
->i_recv_ring
.w_nr
+ 1)
231 rds_ib_ring_resize(&ic
->i_recv_ring
, rds_ibdev
->max_wrs
- 1);
233 /* Protection domain and memory range */
234 ic
->i_pd
= rds_ibdev
->pd
;
235 ic
->i_mr
= rds_ibdev
->mr
;
237 ic
->i_send_cq
= ib_create_cq(dev
, rds_ib_send_cq_comp_handler
,
238 rds_ib_cq_event_handler
, conn
,
239 ic
->i_send_ring
.w_nr
+ 1, 0);
240 if (IS_ERR(ic
->i_send_cq
)) {
241 ret
= PTR_ERR(ic
->i_send_cq
);
242 ic
->i_send_cq
= NULL
;
243 rdsdebug("ib_create_cq send failed: %d\n", ret
);
247 ic
->i_recv_cq
= ib_create_cq(dev
, rds_ib_recv_cq_comp_handler
,
248 rds_ib_cq_event_handler
, conn
,
249 ic
->i_recv_ring
.w_nr
, 0);
250 if (IS_ERR(ic
->i_recv_cq
)) {
251 ret
= PTR_ERR(ic
->i_recv_cq
);
252 ic
->i_recv_cq
= NULL
;
253 rdsdebug("ib_create_cq recv failed: %d\n", ret
);
257 ret
= ib_req_notify_cq(ic
->i_send_cq
, IB_CQ_NEXT_COMP
);
259 rdsdebug("ib_req_notify_cq send failed: %d\n", ret
);
263 ret
= ib_req_notify_cq(ic
->i_recv_cq
, IB_CQ_SOLICITED
);
265 rdsdebug("ib_req_notify_cq recv failed: %d\n", ret
);
269 /* XXX negotiate max send/recv with remote? */
270 memset(&attr
, 0, sizeof(attr
));
271 attr
.event_handler
= rds_ib_qp_event_handler
;
272 attr
.qp_context
= conn
;
273 /* + 1 to allow for the single ack message */
274 attr
.cap
.max_send_wr
= ic
->i_send_ring
.w_nr
+ 1;
275 attr
.cap
.max_recv_wr
= ic
->i_recv_ring
.w_nr
+ 1;
276 attr
.cap
.max_send_sge
= rds_ibdev
->max_sge
;
277 attr
.cap
.max_recv_sge
= RDS_IB_RECV_SGE
;
278 attr
.sq_sig_type
= IB_SIGNAL_REQ_WR
;
279 attr
.qp_type
= IB_QPT_RC
;
280 attr
.send_cq
= ic
->i_send_cq
;
281 attr
.recv_cq
= ic
->i_recv_cq
;
284 * XXX this can fail if max_*_wr is too large? Are we supposed
285 * to back off until we get a value that the hardware can support?
287 ret
= rdma_create_qp(ic
->i_cm_id
, ic
->i_pd
, &attr
);
289 rdsdebug("rdma_create_qp failed: %d\n", ret
);
293 ic
->i_send_hdrs
= ib_dma_alloc_coherent(dev
,
294 ic
->i_send_ring
.w_nr
*
295 sizeof(struct rds_header
),
296 &ic
->i_send_hdrs_dma
, GFP_KERNEL
);
297 if (ic
->i_send_hdrs
== NULL
) {
299 rdsdebug("ib_dma_alloc_coherent send failed\n");
303 ic
->i_recv_hdrs
= ib_dma_alloc_coherent(dev
,
304 ic
->i_recv_ring
.w_nr
*
305 sizeof(struct rds_header
),
306 &ic
->i_recv_hdrs_dma
, GFP_KERNEL
);
307 if (ic
->i_recv_hdrs
== NULL
) {
309 rdsdebug("ib_dma_alloc_coherent recv failed\n");
313 ic
->i_ack
= ib_dma_alloc_coherent(dev
, sizeof(struct rds_header
),
314 &ic
->i_ack_dma
, GFP_KERNEL
);
315 if (ic
->i_ack
== NULL
) {
317 rdsdebug("ib_dma_alloc_coherent ack failed\n");
321 ic
->i_sends
= vmalloc(ic
->i_send_ring
.w_nr
* sizeof(struct rds_ib_send_work
));
322 if (ic
->i_sends
== NULL
) {
324 rdsdebug("send allocation failed\n");
327 rds_ib_send_init_ring(ic
);
329 ic
->i_recvs
= vmalloc(ic
->i_recv_ring
.w_nr
* sizeof(struct rds_ib_recv_work
));
330 if (ic
->i_recvs
== NULL
) {
332 rdsdebug("recv allocation failed\n");
336 rds_ib_recv_init_ring(ic
);
337 rds_ib_recv_init_ack(ic
);
339 /* Post receive buffers - as a side effect, this will update
340 * the posted credit count. */
341 rds_ib_recv_refill(conn
, GFP_KERNEL
, GFP_HIGHUSER
, 1);
343 rdsdebug("conn %p pd %p mr %p cq %p %p\n", conn
, ic
->i_pd
, ic
->i_mr
,
344 ic
->i_send_cq
, ic
->i_recv_cq
);
350 static u32
rds_ib_protocol_compatible(struct rdma_cm_event
*event
)
352 const struct rds_ib_connect_private
*dp
= event
->param
.conn
.private_data
;
357 * rdma_cm private data is odd - when there is any private data in the
358 * request, we will be given a pretty large buffer without telling us the
359 * original size. The only way to tell the difference is by looking at
360 * the contents, which are initialized to zero.
361 * If the protocol version fields aren't set, this is a connection attempt
362 * from an older version. This could could be 3.0 or 2.0 - we can't tell.
363 * We really should have changed this for OFED 1.3 :-(
366 /* Be paranoid. RDS always has privdata */
367 if (!event
->param
.conn
.private_data_len
) {
368 printk(KERN_NOTICE
"RDS incoming connection has no private data, "
373 /* Even if len is crap *now* I still want to check it. -ASG */
374 if (event
->param
.conn
.private_data_len
< sizeof (*dp
)
375 || dp
->dp_protocol_major
== 0)
376 return RDS_PROTOCOL_3_0
;
378 common
= be16_to_cpu(dp
->dp_protocol_minor_mask
) & RDS_IB_SUPPORTED_PROTOCOLS
;
379 if (dp
->dp_protocol_major
== 3 && common
) {
380 version
= RDS_PROTOCOL_3_0
;
381 while ((common
>>= 1) != 0)
383 } else if (printk_ratelimit()) {
384 printk(KERN_NOTICE
"RDS: Connection from %pI4 using "
385 "incompatible protocol version %u.%u\n",
387 dp
->dp_protocol_major
,
388 dp
->dp_protocol_minor
);
393 int rds_ib_cm_handle_connect(struct rdma_cm_id
*cm_id
,
394 struct rdma_cm_event
*event
)
396 __be64 lguid
= cm_id
->route
.path_rec
->sgid
.global
.interface_id
;
397 __be64 fguid
= cm_id
->route
.path_rec
->dgid
.global
.interface_id
;
398 const struct rds_ib_connect_private
*dp
= event
->param
.conn
.private_data
;
399 struct rds_ib_connect_private dp_rep
;
400 struct rds_connection
*conn
= NULL
;
401 struct rds_ib_connection
*ic
= NULL
;
402 struct rdma_conn_param conn_param
;
404 int err
, destroy
= 1;
406 /* Check whether the remote protocol version matches ours. */
407 version
= rds_ib_protocol_compatible(event
);
411 rdsdebug("saddr %pI4 daddr %pI4 RDSv%u.%u lguid 0x%llx fguid "
412 "0x%llx\n", &dp
->dp_saddr
, &dp
->dp_daddr
,
413 RDS_PROTOCOL_MAJOR(version
), RDS_PROTOCOL_MINOR(version
),
414 (unsigned long long)be64_to_cpu(lguid
),
415 (unsigned long long)be64_to_cpu(fguid
));
417 conn
= rds_conn_create(dp
->dp_daddr
, dp
->dp_saddr
, &rds_ib_transport
,
420 rdsdebug("rds_conn_create failed (%ld)\n", PTR_ERR(conn
));
426 * The connection request may occur while the
427 * previous connection exist, e.g. in case of failover.
428 * But as connections may be initiated simultaneously
429 * by both hosts, we have a random backoff mechanism -
430 * see the comment above rds_queue_reconnect()
432 mutex_lock(&conn
->c_cm_lock
);
433 if (!rds_conn_transition(conn
, RDS_CONN_DOWN
, RDS_CONN_CONNECTING
)) {
434 if (rds_conn_state(conn
) == RDS_CONN_UP
) {
435 rdsdebug("incoming connect while connecting\n");
437 rds_ib_stats_inc(s_ib_listen_closed_stale
);
439 if (rds_conn_state(conn
) == RDS_CONN_CONNECTING
) {
440 /* Wait and see - our connect may still be succeeding */
441 rds_ib_stats_inc(s_ib_connect_raced
);
443 mutex_unlock(&conn
->c_cm_lock
);
447 ic
= conn
->c_transport_data
;
449 rds_ib_set_protocol(conn
, version
);
450 rds_ib_set_flow_control(conn
, be32_to_cpu(dp
->dp_credit
));
452 /* If the peer gave us the last packet it saw, process this as if
453 * we had received a regular ACK. */
455 rds_send_drop_acked(conn
, be64_to_cpu(dp
->dp_ack_seq
), NULL
);
457 BUG_ON(cm_id
->context
);
461 cm_id
->context
= conn
;
463 /* We got halfway through setting up the ib_connection, if we
464 * fail now, we have to take the long route out of this mess. */
467 err
= rds_ib_setup_qp(conn
);
469 rds_ib_conn_error(conn
, "rds_ib_setup_qp failed (%d)\n", err
);
473 rds_ib_cm_fill_conn_param(conn
, &conn_param
, &dp_rep
, version
);
475 /* rdma_accept() calls rdma_reject() internally if it fails */
476 err
= rdma_accept(cm_id
, &conn_param
);
477 mutex_unlock(&conn
->c_cm_lock
);
479 rds_ib_conn_error(conn
, "rdma_accept failed (%d)\n", err
);
486 rdma_reject(cm_id
, NULL
, 0);
491 int rds_ib_cm_initiate_connect(struct rdma_cm_id
*cm_id
)
493 struct rds_connection
*conn
= cm_id
->context
;
494 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
495 struct rdma_conn_param conn_param
;
496 struct rds_ib_connect_private dp
;
499 /* If the peer doesn't do protocol negotiation, we must
500 * default to RDSv3.0 */
501 rds_ib_set_protocol(conn
, RDS_PROTOCOL_3_0
);
502 ic
->i_flowctl
= rds_ib_sysctl_flow_control
; /* advertise flow control */
504 ret
= rds_ib_setup_qp(conn
);
506 rds_ib_conn_error(conn
, "rds_ib_setup_qp failed (%d)\n", ret
);
510 rds_ib_cm_fill_conn_param(conn
, &conn_param
, &dp
, RDS_PROTOCOL_VERSION
);
512 ret
= rdma_connect(cm_id
, &conn_param
);
514 rds_ib_conn_error(conn
, "rdma_connect failed (%d)\n", ret
);
517 /* Beware - returning non-zero tells the rdma_cm to destroy
518 * the cm_id. We should certainly not do it as long as we still
519 * "own" the cm_id. */
521 if (ic
->i_cm_id
== cm_id
)
527 int rds_ib_conn_connect(struct rds_connection
*conn
)
529 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
530 struct sockaddr_in src
, dest
;
533 /* XXX I wonder what affect the port space has */
534 /* delegate cm event handler to rdma_transport */
535 ic
->i_cm_id
= rdma_create_id(rds_rdma_cm_event_handler
, conn
,
537 if (IS_ERR(ic
->i_cm_id
)) {
538 ret
= PTR_ERR(ic
->i_cm_id
);
540 rdsdebug("rdma_create_id() failed: %d\n", ret
);
544 rdsdebug("created cm id %p for conn %p\n", ic
->i_cm_id
, conn
);
546 src
.sin_family
= AF_INET
;
547 src
.sin_addr
.s_addr
= (__force u32
)conn
->c_laddr
;
548 src
.sin_port
= (__force u16
)htons(0);
550 dest
.sin_family
= AF_INET
;
551 dest
.sin_addr
.s_addr
= (__force u32
)conn
->c_faddr
;
552 dest
.sin_port
= (__force u16
)htons(RDS_PORT
);
554 ret
= rdma_resolve_addr(ic
->i_cm_id
, (struct sockaddr
*)&src
,
555 (struct sockaddr
*)&dest
,
556 RDS_RDMA_RESOLVE_TIMEOUT_MS
);
558 rdsdebug("addr resolve failed for cm id %p: %d\n", ic
->i_cm_id
,
560 rdma_destroy_id(ic
->i_cm_id
);
569 * This is so careful about only cleaning up resources that were built up
570 * so that it can be called at any point during startup. In fact it
571 * can be called multiple times for a given connection.
573 void rds_ib_conn_shutdown(struct rds_connection
*conn
)
575 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
578 rdsdebug("cm %p pd %p cq %p %p qp %p\n", ic
->i_cm_id
,
579 ic
->i_pd
, ic
->i_send_cq
, ic
->i_recv_cq
,
580 ic
->i_cm_id
? ic
->i_cm_id
->qp
: NULL
);
583 struct ib_device
*dev
= ic
->i_cm_id
->device
;
585 rdsdebug("disconnecting cm %p\n", ic
->i_cm_id
);
586 err
= rdma_disconnect(ic
->i_cm_id
);
588 /* Actually this may happen quite frequently, when
589 * an outgoing connect raced with an incoming connect.
591 rdsdebug("failed to disconnect, cm: %p err %d\n",
595 wait_event(rds_ib_ring_empty_wait
,
596 rds_ib_ring_empty(&ic
->i_send_ring
) &&
597 rds_ib_ring_empty(&ic
->i_recv_ring
));
600 ib_dma_free_coherent(dev
,
601 ic
->i_send_ring
.w_nr
*
602 sizeof(struct rds_header
),
604 ic
->i_send_hdrs_dma
);
607 ib_dma_free_coherent(dev
,
608 ic
->i_recv_ring
.w_nr
*
609 sizeof(struct rds_header
),
611 ic
->i_recv_hdrs_dma
);
614 ib_dma_free_coherent(dev
, sizeof(struct rds_header
),
615 ic
->i_ack
, ic
->i_ack_dma
);
618 rds_ib_send_clear_ring(ic
);
620 rds_ib_recv_clear_ring(ic
);
623 rdma_destroy_qp(ic
->i_cm_id
);
625 ib_destroy_cq(ic
->i_send_cq
);
627 ib_destroy_cq(ic
->i_recv_cq
);
628 rdma_destroy_id(ic
->i_cm_id
);
631 * Move connection back to the nodev list.
634 rds_ib_remove_conn(ic
->rds_ibdev
, conn
);
639 ic
->i_send_cq
= NULL
;
640 ic
->i_recv_cq
= NULL
;
641 ic
->i_send_hdrs
= NULL
;
642 ic
->i_recv_hdrs
= NULL
;
645 BUG_ON(ic
->rds_ibdev
);
647 /* Clear pending transmit */
649 rds_message_put(ic
->i_rm
);
653 /* Clear the ACK state */
654 clear_bit(IB_ACK_IN_FLIGHT
, &ic
->i_ack_flags
);
655 #ifdef KERNEL_HAS_ATOMIC64
656 atomic64_set(&ic
->i_ack_next
, 0);
662 /* Clear flow control state */
664 atomic_set(&ic
->i_credits
, 0);
666 rds_ib_ring_init(&ic
->i_send_ring
, rds_ib_sysctl_max_send_wr
);
667 rds_ib_ring_init(&ic
->i_recv_ring
, rds_ib_sysctl_max_recv_wr
);
670 rds_inc_put(&ic
->i_ibinc
->ii_inc
);
680 int rds_ib_conn_alloc(struct rds_connection
*conn
, gfp_t gfp
)
682 struct rds_ib_connection
*ic
;
686 ic
= kzalloc(sizeof(struct rds_ib_connection
), GFP_KERNEL
);
690 INIT_LIST_HEAD(&ic
->ib_node
);
691 mutex_init(&ic
->i_recv_mutex
);
692 #ifndef KERNEL_HAS_ATOMIC64
693 spin_lock_init(&ic
->i_ack_lock
);
697 * rds_ib_conn_shutdown() waits for these to be emptied so they
698 * must be initialized before it can be called.
700 rds_ib_ring_init(&ic
->i_send_ring
, rds_ib_sysctl_max_send_wr
);
701 rds_ib_ring_init(&ic
->i_recv_ring
, rds_ib_sysctl_max_recv_wr
);
704 conn
->c_transport_data
= ic
;
706 spin_lock_irqsave(&ib_nodev_conns_lock
, flags
);
707 list_add_tail(&ic
->ib_node
, &ib_nodev_conns
);
708 spin_unlock_irqrestore(&ib_nodev_conns_lock
, flags
);
711 rdsdebug("conn %p conn ic %p\n", conn
, conn
->c_transport_data
);
716 * Free a connection. Connection must be shut down and not set for reconnect.
718 void rds_ib_conn_free(void *arg
)
720 struct rds_ib_connection
*ic
= arg
;
721 spinlock_t
*lock_ptr
;
723 rdsdebug("ic %p\n", ic
);
726 * Conn is either on a dev's list or on the nodev list.
727 * A race with shutdown() or connect() would cause problems
728 * (since rds_ibdev would change) but that should never happen.
730 lock_ptr
= ic
->rds_ibdev
? &ic
->rds_ibdev
->spinlock
: &ib_nodev_conns_lock
;
732 spin_lock_irq(lock_ptr
);
733 list_del(&ic
->ib_node
);
734 spin_unlock_irq(lock_ptr
);
741 * An error occurred on the connection
744 __rds_ib_conn_error(struct rds_connection
*conn
, const char *fmt
, ...)