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/slab.h>
36 #include <linux/vmalloc.h>
37 #include <linux/ratelimit.h>
42 static char *rds_ib_event_type_strings
[] = {
43 #define RDS_IB_EVENT_STRING(foo) \
44 [IB_EVENT_##foo] = __stringify(IB_EVENT_##foo)
45 RDS_IB_EVENT_STRING(CQ_ERR
),
46 RDS_IB_EVENT_STRING(QP_FATAL
),
47 RDS_IB_EVENT_STRING(QP_REQ_ERR
),
48 RDS_IB_EVENT_STRING(QP_ACCESS_ERR
),
49 RDS_IB_EVENT_STRING(COMM_EST
),
50 RDS_IB_EVENT_STRING(SQ_DRAINED
),
51 RDS_IB_EVENT_STRING(PATH_MIG
),
52 RDS_IB_EVENT_STRING(PATH_MIG_ERR
),
53 RDS_IB_EVENT_STRING(DEVICE_FATAL
),
54 RDS_IB_EVENT_STRING(PORT_ACTIVE
),
55 RDS_IB_EVENT_STRING(PORT_ERR
),
56 RDS_IB_EVENT_STRING(LID_CHANGE
),
57 RDS_IB_EVENT_STRING(PKEY_CHANGE
),
58 RDS_IB_EVENT_STRING(SM_CHANGE
),
59 RDS_IB_EVENT_STRING(SRQ_ERR
),
60 RDS_IB_EVENT_STRING(SRQ_LIMIT_REACHED
),
61 RDS_IB_EVENT_STRING(QP_LAST_WQE_REACHED
),
62 RDS_IB_EVENT_STRING(CLIENT_REREGISTER
),
63 #undef RDS_IB_EVENT_STRING
66 static char *rds_ib_event_str(enum ib_event_type type
)
68 return rds_str_array(rds_ib_event_type_strings
,
69 ARRAY_SIZE(rds_ib_event_type_strings
), type
);
73 * Set the selected protocol version
75 static void rds_ib_set_protocol(struct rds_connection
*conn
, unsigned int version
)
77 conn
->c_version
= version
;
83 static void rds_ib_set_flow_control(struct rds_connection
*conn
, u32 credits
)
85 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
87 if (rds_ib_sysctl_flow_control
&& credits
!= 0) {
88 /* We're doing flow control */
90 rds_ib_send_add_credits(conn
, credits
);
97 * Tune RNR behavior. Without flow control, we use a rather
98 * low timeout, but not the absolute minimum - this should
101 * We already set the RNR retry count to 7 (which is the
102 * smallest infinite number :-) above.
103 * If flow control is off, we want to change this back to 0
104 * so that we learn quickly when our credit accounting is
107 * Caller passes in a qp_attr pointer - don't waste stack spacv
108 * by allocation this twice.
111 rds_ib_tune_rnr(struct rds_ib_connection
*ic
, struct ib_qp_attr
*attr
)
115 attr
->min_rnr_timer
= IB_RNR_TIMER_000_32
;
116 ret
= ib_modify_qp(ic
->i_cm_id
->qp
, attr
, IB_QP_MIN_RNR_TIMER
);
118 printk(KERN_NOTICE
"ib_modify_qp(IB_QP_MIN_RNR_TIMER): err=%d\n", -ret
);
122 * Connection established.
123 * We get here for both outgoing and incoming connection.
125 void rds_ib_cm_connect_complete(struct rds_connection
*conn
, struct rdma_cm_event
*event
)
127 const struct rds_ib_connect_private
*dp
= NULL
;
128 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
129 struct ib_qp_attr qp_attr
;
132 if (event
->param
.conn
.private_data_len
>= sizeof(*dp
)) {
133 dp
= event
->param
.conn
.private_data
;
135 /* make sure it isn't empty data */
136 if (dp
->dp_protocol_major
) {
137 rds_ib_set_protocol(conn
,
138 RDS_PROTOCOL(dp
->dp_protocol_major
,
139 dp
->dp_protocol_minor
));
140 rds_ib_set_flow_control(conn
, be32_to_cpu(dp
->dp_credit
));
144 if (conn
->c_version
< RDS_PROTOCOL(3,1)) {
145 printk(KERN_NOTICE
"RDS/IB: Connection to %pI4 version %u.%u failed,"
146 " no longer supported\n",
148 RDS_PROTOCOL_MAJOR(conn
->c_version
),
149 RDS_PROTOCOL_MINOR(conn
->c_version
));
150 rds_conn_destroy(conn
);
153 printk(KERN_NOTICE
"RDS/IB: connected to %pI4 version %u.%u%s\n",
155 RDS_PROTOCOL_MAJOR(conn
->c_version
),
156 RDS_PROTOCOL_MINOR(conn
->c_version
),
157 ic
->i_flowctl
? ", flow control" : "");
161 * Init rings and fill recv. this needs to wait until protocol negotiation
162 * is complete, since ring layout is different from 3.0 to 3.1.
164 rds_ib_send_init_ring(ic
);
165 rds_ib_recv_init_ring(ic
);
166 /* Post receive buffers - as a side effect, this will update
167 * the posted credit count. */
168 rds_ib_recv_refill(conn
, 1);
170 /* Tune RNR behavior */
171 rds_ib_tune_rnr(ic
, &qp_attr
);
173 qp_attr
.qp_state
= IB_QPS_RTS
;
174 err
= ib_modify_qp(ic
->i_cm_id
->qp
, &qp_attr
, IB_QP_STATE
);
176 printk(KERN_NOTICE
"ib_modify_qp(IB_QP_STATE, RTS): err=%d\n", err
);
178 /* update ib_device with this local ipaddr */
179 err
= rds_ib_update_ipaddr(ic
->rds_ibdev
, conn
->c_laddr
);
181 printk(KERN_ERR
"rds_ib_update_ipaddr failed (%d)\n",
184 /* If the peer gave us the last packet it saw, process this as if
185 * we had received a regular ACK. */
186 if (dp
&& dp
->dp_ack_seq
)
187 rds_send_drop_acked(conn
, be64_to_cpu(dp
->dp_ack_seq
), NULL
);
189 rds_connect_complete(conn
);
192 static void rds_ib_cm_fill_conn_param(struct rds_connection
*conn
,
193 struct rdma_conn_param
*conn_param
,
194 struct rds_ib_connect_private
*dp
,
195 u32 protocol_version
,
196 u32 max_responder_resources
,
197 u32 max_initiator_depth
)
199 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
200 struct rds_ib_device
*rds_ibdev
= ic
->rds_ibdev
;
202 memset(conn_param
, 0, sizeof(struct rdma_conn_param
));
204 conn_param
->responder_resources
=
205 min_t(u32
, rds_ibdev
->max_responder_resources
, max_responder_resources
);
206 conn_param
->initiator_depth
=
207 min_t(u32
, rds_ibdev
->max_initiator_depth
, max_initiator_depth
);
208 conn_param
->retry_count
= min_t(unsigned int, rds_ib_retry_count
, 7);
209 conn_param
->rnr_retry_count
= 7;
212 memset(dp
, 0, sizeof(*dp
));
213 dp
->dp_saddr
= conn
->c_laddr
;
214 dp
->dp_daddr
= conn
->c_faddr
;
215 dp
->dp_protocol_major
= RDS_PROTOCOL_MAJOR(protocol_version
);
216 dp
->dp_protocol_minor
= RDS_PROTOCOL_MINOR(protocol_version
);
217 dp
->dp_protocol_minor_mask
= cpu_to_be16(RDS_IB_SUPPORTED_PROTOCOLS
);
218 dp
->dp_ack_seq
= rds_ib_piggyb_ack(ic
);
220 /* Advertise flow control */
222 unsigned int credits
;
224 credits
= IB_GET_POST_CREDITS(atomic_read(&ic
->i_credits
));
225 dp
->dp_credit
= cpu_to_be32(credits
);
226 atomic_sub(IB_SET_POST_CREDITS(credits
), &ic
->i_credits
);
229 conn_param
->private_data
= dp
;
230 conn_param
->private_data_len
= sizeof(*dp
);
234 static void rds_ib_cq_event_handler(struct ib_event
*event
, void *data
)
236 rdsdebug("event %u (%s) data %p\n",
237 event
->event
, rds_ib_event_str(event
->event
), data
);
240 static void rds_ib_qp_event_handler(struct ib_event
*event
, void *data
)
242 struct rds_connection
*conn
= data
;
243 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
245 rdsdebug("conn %p ic %p event %u (%s)\n", conn
, ic
, event
->event
,
246 rds_ib_event_str(event
->event
));
248 switch (event
->event
) {
249 case IB_EVENT_COMM_EST
:
250 rdma_notify(ic
->i_cm_id
, IB_EVENT_COMM_EST
);
253 rdsdebug("Fatal QP Event %u (%s) "
254 "- connection %pI4->%pI4, reconnecting\n",
255 event
->event
, rds_ib_event_str(event
->event
),
256 &conn
->c_laddr
, &conn
->c_faddr
);
263 * This needs to be very careful to not leave IS_ERR pointers around for
264 * cleanup to trip over.
266 static int rds_ib_setup_qp(struct rds_connection
*conn
)
268 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
269 struct ib_device
*dev
= ic
->i_cm_id
->device
;
270 struct ib_qp_init_attr attr
;
271 struct rds_ib_device
*rds_ibdev
;
275 * It's normal to see a null device if an incoming connection races
276 * with device removal, so we don't print a warning.
278 rds_ibdev
= rds_ib_get_client_data(dev
);
282 /* add the conn now so that connection establishment has the dev */
283 rds_ib_add_conn(rds_ibdev
, conn
);
285 if (rds_ibdev
->max_wrs
< ic
->i_send_ring
.w_nr
+ 1)
286 rds_ib_ring_resize(&ic
->i_send_ring
, rds_ibdev
->max_wrs
- 1);
287 if (rds_ibdev
->max_wrs
< ic
->i_recv_ring
.w_nr
+ 1)
288 rds_ib_ring_resize(&ic
->i_recv_ring
, rds_ibdev
->max_wrs
- 1);
290 /* Protection domain and memory range */
291 ic
->i_pd
= rds_ibdev
->pd
;
292 ic
->i_mr
= rds_ibdev
->mr
;
294 ic
->i_send_cq
= ib_create_cq(dev
, rds_ib_send_cq_comp_handler
,
295 rds_ib_cq_event_handler
, conn
,
296 ic
->i_send_ring
.w_nr
+ 1, 0);
297 if (IS_ERR(ic
->i_send_cq
)) {
298 ret
= PTR_ERR(ic
->i_send_cq
);
299 ic
->i_send_cq
= NULL
;
300 rdsdebug("ib_create_cq send failed: %d\n", ret
);
304 ic
->i_recv_cq
= ib_create_cq(dev
, rds_ib_recv_cq_comp_handler
,
305 rds_ib_cq_event_handler
, conn
,
306 ic
->i_recv_ring
.w_nr
, 0);
307 if (IS_ERR(ic
->i_recv_cq
)) {
308 ret
= PTR_ERR(ic
->i_recv_cq
);
309 ic
->i_recv_cq
= NULL
;
310 rdsdebug("ib_create_cq recv failed: %d\n", ret
);
314 ret
= ib_req_notify_cq(ic
->i_send_cq
, IB_CQ_NEXT_COMP
);
316 rdsdebug("ib_req_notify_cq send failed: %d\n", ret
);
320 ret
= ib_req_notify_cq(ic
->i_recv_cq
, IB_CQ_SOLICITED
);
322 rdsdebug("ib_req_notify_cq recv failed: %d\n", ret
);
326 /* XXX negotiate max send/recv with remote? */
327 memset(&attr
, 0, sizeof(attr
));
328 attr
.event_handler
= rds_ib_qp_event_handler
;
329 attr
.qp_context
= conn
;
330 /* + 1 to allow for the single ack message */
331 attr
.cap
.max_send_wr
= ic
->i_send_ring
.w_nr
+ 1;
332 attr
.cap
.max_recv_wr
= ic
->i_recv_ring
.w_nr
+ 1;
333 attr
.cap
.max_send_sge
= rds_ibdev
->max_sge
;
334 attr
.cap
.max_recv_sge
= RDS_IB_RECV_SGE
;
335 attr
.sq_sig_type
= IB_SIGNAL_REQ_WR
;
336 attr
.qp_type
= IB_QPT_RC
;
337 attr
.send_cq
= ic
->i_send_cq
;
338 attr
.recv_cq
= ic
->i_recv_cq
;
341 * XXX this can fail if max_*_wr is too large? Are we supposed
342 * to back off until we get a value that the hardware can support?
344 ret
= rdma_create_qp(ic
->i_cm_id
, ic
->i_pd
, &attr
);
346 rdsdebug("rdma_create_qp failed: %d\n", ret
);
350 ic
->i_send_hdrs
= ib_dma_alloc_coherent(dev
,
351 ic
->i_send_ring
.w_nr
*
352 sizeof(struct rds_header
),
353 &ic
->i_send_hdrs_dma
, GFP_KERNEL
);
354 if (!ic
->i_send_hdrs
) {
356 rdsdebug("ib_dma_alloc_coherent send failed\n");
360 ic
->i_recv_hdrs
= ib_dma_alloc_coherent(dev
,
361 ic
->i_recv_ring
.w_nr
*
362 sizeof(struct rds_header
),
363 &ic
->i_recv_hdrs_dma
, GFP_KERNEL
);
364 if (!ic
->i_recv_hdrs
) {
366 rdsdebug("ib_dma_alloc_coherent recv failed\n");
370 ic
->i_ack
= ib_dma_alloc_coherent(dev
, sizeof(struct rds_header
),
371 &ic
->i_ack_dma
, GFP_KERNEL
);
374 rdsdebug("ib_dma_alloc_coherent ack failed\n");
378 ic
->i_sends
= vzalloc_node(ic
->i_send_ring
.w_nr
* sizeof(struct rds_ib_send_work
),
382 rdsdebug("send allocation failed\n");
386 ic
->i_recvs
= vzalloc_node(ic
->i_recv_ring
.w_nr
* sizeof(struct rds_ib_recv_work
),
390 rdsdebug("recv allocation failed\n");
394 rds_ib_recv_init_ack(ic
);
396 rdsdebug("conn %p pd %p mr %p cq %p %p\n", conn
, ic
->i_pd
, ic
->i_mr
,
397 ic
->i_send_cq
, ic
->i_recv_cq
);
400 rds_ib_dev_put(rds_ibdev
);
404 static u32
rds_ib_protocol_compatible(struct rdma_cm_event
*event
)
406 const struct rds_ib_connect_private
*dp
= event
->param
.conn
.private_data
;
411 * rdma_cm private data is odd - when there is any private data in the
412 * request, we will be given a pretty large buffer without telling us the
413 * original size. The only way to tell the difference is by looking at
414 * the contents, which are initialized to zero.
415 * If the protocol version fields aren't set, this is a connection attempt
416 * from an older version. This could could be 3.0 or 2.0 - we can't tell.
417 * We really should have changed this for OFED 1.3 :-(
420 /* Be paranoid. RDS always has privdata */
421 if (!event
->param
.conn
.private_data_len
) {
422 printk(KERN_NOTICE
"RDS incoming connection has no private data, "
427 /* Even if len is crap *now* I still want to check it. -ASG */
428 if (event
->param
.conn
.private_data_len
< sizeof (*dp
) ||
429 dp
->dp_protocol_major
== 0)
430 return RDS_PROTOCOL_3_0
;
432 common
= be16_to_cpu(dp
->dp_protocol_minor_mask
) & RDS_IB_SUPPORTED_PROTOCOLS
;
433 if (dp
->dp_protocol_major
== 3 && common
) {
434 version
= RDS_PROTOCOL_3_0
;
435 while ((common
>>= 1) != 0)
438 printk_ratelimited(KERN_NOTICE
"RDS: Connection from %pI4 using "
439 "incompatible protocol version %u.%u\n",
441 dp
->dp_protocol_major
,
442 dp
->dp_protocol_minor
);
446 int rds_ib_cm_handle_connect(struct rdma_cm_id
*cm_id
,
447 struct rdma_cm_event
*event
)
449 __be64 lguid
= cm_id
->route
.path_rec
->sgid
.global
.interface_id
;
450 __be64 fguid
= cm_id
->route
.path_rec
->dgid
.global
.interface_id
;
451 const struct rds_ib_connect_private
*dp
= event
->param
.conn
.private_data
;
452 struct rds_ib_connect_private dp_rep
;
453 struct rds_connection
*conn
= NULL
;
454 struct rds_ib_connection
*ic
= NULL
;
455 struct rdma_conn_param conn_param
;
457 int err
= 1, destroy
= 1;
459 /* Check whether the remote protocol version matches ours. */
460 version
= rds_ib_protocol_compatible(event
);
464 rdsdebug("saddr %pI4 daddr %pI4 RDSv%u.%u lguid 0x%llx fguid "
465 "0x%llx\n", &dp
->dp_saddr
, &dp
->dp_daddr
,
466 RDS_PROTOCOL_MAJOR(version
), RDS_PROTOCOL_MINOR(version
),
467 (unsigned long long)be64_to_cpu(lguid
),
468 (unsigned long long)be64_to_cpu(fguid
));
470 conn
= rds_conn_create(dp
->dp_daddr
, dp
->dp_saddr
, &rds_ib_transport
,
473 rdsdebug("rds_conn_create failed (%ld)\n", PTR_ERR(conn
));
479 * The connection request may occur while the
480 * previous connection exist, e.g. in case of failover.
481 * But as connections may be initiated simultaneously
482 * by both hosts, we have a random backoff mechanism -
483 * see the comment above rds_queue_reconnect()
485 mutex_lock(&conn
->c_cm_lock
);
486 if (!rds_conn_transition(conn
, RDS_CONN_DOWN
, RDS_CONN_CONNECTING
)) {
487 if (rds_conn_state(conn
) == RDS_CONN_UP
) {
488 rdsdebug("incoming connect while connecting\n");
490 rds_ib_stats_inc(s_ib_listen_closed_stale
);
492 if (rds_conn_state(conn
) == RDS_CONN_CONNECTING
) {
493 /* Wait and see - our connect may still be succeeding */
494 rds_ib_stats_inc(s_ib_connect_raced
);
499 ic
= conn
->c_transport_data
;
501 rds_ib_set_protocol(conn
, version
);
502 rds_ib_set_flow_control(conn
, be32_to_cpu(dp
->dp_credit
));
504 /* If the peer gave us the last packet it saw, process this as if
505 * we had received a regular ACK. */
507 rds_send_drop_acked(conn
, be64_to_cpu(dp
->dp_ack_seq
), NULL
);
509 BUG_ON(cm_id
->context
);
513 cm_id
->context
= conn
;
515 /* We got halfway through setting up the ib_connection, if we
516 * fail now, we have to take the long route out of this mess. */
519 err
= rds_ib_setup_qp(conn
);
521 rds_ib_conn_error(conn
, "rds_ib_setup_qp failed (%d)\n", err
);
525 rds_ib_cm_fill_conn_param(conn
, &conn_param
, &dp_rep
, version
,
526 event
->param
.conn
.responder_resources
,
527 event
->param
.conn
.initiator_depth
);
529 /* rdma_accept() calls rdma_reject() internally if it fails */
530 err
= rdma_accept(cm_id
, &conn_param
);
532 rds_ib_conn_error(conn
, "rdma_accept failed (%d)\n", err
);
536 mutex_unlock(&conn
->c_cm_lock
);
538 rdma_reject(cm_id
, NULL
, 0);
543 int rds_ib_cm_initiate_connect(struct rdma_cm_id
*cm_id
)
545 struct rds_connection
*conn
= cm_id
->context
;
546 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
547 struct rdma_conn_param conn_param
;
548 struct rds_ib_connect_private dp
;
551 /* If the peer doesn't do protocol negotiation, we must
552 * default to RDSv3.0 */
553 rds_ib_set_protocol(conn
, RDS_PROTOCOL_3_0
);
554 ic
->i_flowctl
= rds_ib_sysctl_flow_control
; /* advertise flow control */
556 ret
= rds_ib_setup_qp(conn
);
558 rds_ib_conn_error(conn
, "rds_ib_setup_qp failed (%d)\n", ret
);
562 rds_ib_cm_fill_conn_param(conn
, &conn_param
, &dp
, RDS_PROTOCOL_VERSION
,
564 ret
= rdma_connect(cm_id
, &conn_param
);
566 rds_ib_conn_error(conn
, "rdma_connect failed (%d)\n", ret
);
569 /* Beware - returning non-zero tells the rdma_cm to destroy
570 * the cm_id. We should certainly not do it as long as we still
571 * "own" the cm_id. */
573 if (ic
->i_cm_id
== cm_id
)
579 int rds_ib_conn_connect(struct rds_connection
*conn
)
581 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
582 struct sockaddr_in src
, dest
;
585 /* XXX I wonder what affect the port space has */
586 /* delegate cm event handler to rdma_transport */
587 ic
->i_cm_id
= rdma_create_id(rds_rdma_cm_event_handler
, conn
,
588 RDMA_PS_TCP
, IB_QPT_RC
);
589 if (IS_ERR(ic
->i_cm_id
)) {
590 ret
= PTR_ERR(ic
->i_cm_id
);
592 rdsdebug("rdma_create_id() failed: %d\n", ret
);
596 rdsdebug("created cm id %p for conn %p\n", ic
->i_cm_id
, conn
);
598 src
.sin_family
= AF_INET
;
599 src
.sin_addr
.s_addr
= (__force u32
)conn
->c_laddr
;
600 src
.sin_port
= (__force u16
)htons(0);
602 dest
.sin_family
= AF_INET
;
603 dest
.sin_addr
.s_addr
= (__force u32
)conn
->c_faddr
;
604 dest
.sin_port
= (__force u16
)htons(RDS_PORT
);
606 ret
= rdma_resolve_addr(ic
->i_cm_id
, (struct sockaddr
*)&src
,
607 (struct sockaddr
*)&dest
,
608 RDS_RDMA_RESOLVE_TIMEOUT_MS
);
610 rdsdebug("addr resolve failed for cm id %p: %d\n", ic
->i_cm_id
,
612 rdma_destroy_id(ic
->i_cm_id
);
621 * This is so careful about only cleaning up resources that were built up
622 * so that it can be called at any point during startup. In fact it
623 * can be called multiple times for a given connection.
625 void rds_ib_conn_shutdown(struct rds_connection
*conn
)
627 struct rds_ib_connection
*ic
= conn
->c_transport_data
;
630 rdsdebug("cm %p pd %p cq %p %p qp %p\n", ic
->i_cm_id
,
631 ic
->i_pd
, ic
->i_send_cq
, ic
->i_recv_cq
,
632 ic
->i_cm_id
? ic
->i_cm_id
->qp
: NULL
);
635 struct ib_device
*dev
= ic
->i_cm_id
->device
;
637 rdsdebug("disconnecting cm %p\n", ic
->i_cm_id
);
638 err
= rdma_disconnect(ic
->i_cm_id
);
640 /* Actually this may happen quite frequently, when
641 * an outgoing connect raced with an incoming connect.
643 rdsdebug("failed to disconnect, cm: %p err %d\n",
648 * We want to wait for tx and rx completion to finish
649 * before we tear down the connection, but we have to be
650 * careful not to get stuck waiting on a send ring that
651 * only has unsignaled sends in it. We've shutdown new
652 * sends before getting here so by waiting for signaled
653 * sends to complete we're ensured that there will be no
654 * more tx processing.
656 wait_event(rds_ib_ring_empty_wait
,
657 rds_ib_ring_empty(&ic
->i_recv_ring
) &&
658 (atomic_read(&ic
->i_signaled_sends
) == 0));
659 tasklet_kill(&ic
->i_recv_tasklet
);
662 ib_dma_free_coherent(dev
,
663 ic
->i_send_ring
.w_nr
*
664 sizeof(struct rds_header
),
666 ic
->i_send_hdrs_dma
);
669 ib_dma_free_coherent(dev
,
670 ic
->i_recv_ring
.w_nr
*
671 sizeof(struct rds_header
),
673 ic
->i_recv_hdrs_dma
);
676 ib_dma_free_coherent(dev
, sizeof(struct rds_header
),
677 ic
->i_ack
, ic
->i_ack_dma
);
680 rds_ib_send_clear_ring(ic
);
682 rds_ib_recv_clear_ring(ic
);
685 rdma_destroy_qp(ic
->i_cm_id
);
687 ib_destroy_cq(ic
->i_send_cq
);
689 ib_destroy_cq(ic
->i_recv_cq
);
690 rdma_destroy_id(ic
->i_cm_id
);
693 * Move connection back to the nodev list.
696 rds_ib_remove_conn(ic
->rds_ibdev
, conn
);
701 ic
->i_send_cq
= NULL
;
702 ic
->i_recv_cq
= NULL
;
703 ic
->i_send_hdrs
= NULL
;
704 ic
->i_recv_hdrs
= NULL
;
707 BUG_ON(ic
->rds_ibdev
);
709 /* Clear pending transmit */
711 struct rds_message
*rm
;
713 rm
= container_of(ic
->i_data_op
, struct rds_message
, data
);
715 ic
->i_data_op
= NULL
;
718 /* Clear the ACK state */
719 clear_bit(IB_ACK_IN_FLIGHT
, &ic
->i_ack_flags
);
720 #ifdef KERNEL_HAS_ATOMIC64
721 atomic64_set(&ic
->i_ack_next
, 0);
727 /* Clear flow control state */
729 atomic_set(&ic
->i_credits
, 0);
731 rds_ib_ring_init(&ic
->i_send_ring
, rds_ib_sysctl_max_send_wr
);
732 rds_ib_ring_init(&ic
->i_recv_ring
, rds_ib_sysctl_max_recv_wr
);
735 rds_inc_put(&ic
->i_ibinc
->ii_inc
);
745 int rds_ib_conn_alloc(struct rds_connection
*conn
, gfp_t gfp
)
747 struct rds_ib_connection
*ic
;
752 ic
= kzalloc(sizeof(struct rds_ib_connection
), GFP_KERNEL
);
756 ret
= rds_ib_recv_alloc_caches(ic
);
762 INIT_LIST_HEAD(&ic
->ib_node
);
763 tasklet_init(&ic
->i_recv_tasklet
, rds_ib_recv_tasklet_fn
,
765 mutex_init(&ic
->i_recv_mutex
);
766 #ifndef KERNEL_HAS_ATOMIC64
767 spin_lock_init(&ic
->i_ack_lock
);
769 atomic_set(&ic
->i_signaled_sends
, 0);
772 * rds_ib_conn_shutdown() waits for these to be emptied so they
773 * must be initialized before it can be called.
775 rds_ib_ring_init(&ic
->i_send_ring
, rds_ib_sysctl_max_send_wr
);
776 rds_ib_ring_init(&ic
->i_recv_ring
, rds_ib_sysctl_max_recv_wr
);
779 conn
->c_transport_data
= ic
;
781 spin_lock_irqsave(&ib_nodev_conns_lock
, flags
);
782 list_add_tail(&ic
->ib_node
, &ib_nodev_conns
);
783 spin_unlock_irqrestore(&ib_nodev_conns_lock
, flags
);
786 rdsdebug("conn %p conn ic %p\n", conn
, conn
->c_transport_data
);
791 * Free a connection. Connection must be shut down and not set for reconnect.
793 void rds_ib_conn_free(void *arg
)
795 struct rds_ib_connection
*ic
= arg
;
796 spinlock_t
*lock_ptr
;
798 rdsdebug("ic %p\n", ic
);
801 * Conn is either on a dev's list or on the nodev list.
802 * A race with shutdown() or connect() would cause problems
803 * (since rds_ibdev would change) but that should never happen.
805 lock_ptr
= ic
->rds_ibdev
? &ic
->rds_ibdev
->spinlock
: &ib_nodev_conns_lock
;
807 spin_lock_irq(lock_ptr
);
808 list_del(&ic
->ib_node
);
809 spin_unlock_irq(lock_ptr
);
811 rds_ib_recv_free_caches(ic
);
818 * An error occurred on the connection
821 __rds_ib_conn_error(struct rds_connection
*conn
, const char *fmt
, ...)