1 /* 2 * Copyright (c) 2006 Oracle. All rights reserved. 3 * 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: 9 * 10 * Redistribution and use in source and binary forms, with or 11 * without modification, are permitted provided that the following 12 * conditions are met: 13 * 14 * - Redistributions of source code must retain the above 15 * copyright notice, this list of conditions and the following 16 * disclaimer. 17 * 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. 22 * 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 30 * SOFTWARE. 31 * 32 */ 33 #include <linux/kernel.h> 34 #include <linux/in.h> 35 #include <linux/slab.h> 36 #include <linux/vmalloc.h> 37 38 #include "rds.h" 39 #include "ib.h" 40 41 static char *rds_ib_event_type_strings[] = { 42 #define RDS_IB_EVENT_STRING(foo) \ 43 [IB_EVENT_##foo] = __stringify(IB_EVENT_##foo) 44 RDS_IB_EVENT_STRING(CQ_ERR), 45 RDS_IB_EVENT_STRING(QP_FATAL), 46 RDS_IB_EVENT_STRING(QP_REQ_ERR), 47 RDS_IB_EVENT_STRING(QP_ACCESS_ERR), 48 RDS_IB_EVENT_STRING(COMM_EST), 49 RDS_IB_EVENT_STRING(SQ_DRAINED), 50 RDS_IB_EVENT_STRING(PATH_MIG), 51 RDS_IB_EVENT_STRING(PATH_MIG_ERR), 52 RDS_IB_EVENT_STRING(DEVICE_FATAL), 53 RDS_IB_EVENT_STRING(PORT_ACTIVE), 54 RDS_IB_EVENT_STRING(PORT_ERR), 55 RDS_IB_EVENT_STRING(LID_CHANGE), 56 RDS_IB_EVENT_STRING(PKEY_CHANGE), 57 RDS_IB_EVENT_STRING(SM_CHANGE), 58 RDS_IB_EVENT_STRING(SRQ_ERR), 59 RDS_IB_EVENT_STRING(SRQ_LIMIT_REACHED), 60 RDS_IB_EVENT_STRING(QP_LAST_WQE_REACHED), 61 RDS_IB_EVENT_STRING(CLIENT_REREGISTER), 62 #undef RDS_IB_EVENT_STRING 63 }; 64 65 static char *rds_ib_event_str(enum ib_event_type type) 66 { 67 return rds_str_array(rds_ib_event_type_strings, 68 ARRAY_SIZE(rds_ib_event_type_strings), type); 69 }; 70 71 /* 72 * Set the selected protocol version 73 */ 74 static void rds_ib_set_protocol(struct rds_connection *conn, unsigned int version) 75 { 76 conn->c_version = version; 77 } 78 79 /* 80 * Set up flow control 81 */ 82 static void rds_ib_set_flow_control(struct rds_connection *conn, u32 credits) 83 { 84 struct rds_ib_connection *ic = conn->c_transport_data; 85 86 if (rds_ib_sysctl_flow_control && credits != 0) { 87 /* We're doing flow control */ 88 ic->i_flowctl = 1; 89 rds_ib_send_add_credits(conn, credits); 90 } else { 91 ic->i_flowctl = 0; 92 } 93 } 94 95 /* 96 * Tune RNR behavior. Without flow control, we use a rather 97 * low timeout, but not the absolute minimum - this should 98 * be tunable. 99 * 100 * We already set the RNR retry count to 7 (which is the 101 * smallest infinite number :-) above. 102 * If flow control is off, we want to change this back to 0 103 * so that we learn quickly when our credit accounting is 104 * buggy. 105 * 106 * Caller passes in a qp_attr pointer - don't waste stack spacv 107 * by allocation this twice. 108 */ 109 static void 110 rds_ib_tune_rnr(struct rds_ib_connection *ic, struct ib_qp_attr *attr) 111 { 112 int ret; 113 114 attr->min_rnr_timer = IB_RNR_TIMER_000_32; 115 ret = ib_modify_qp(ic->i_cm_id->qp, attr, IB_QP_MIN_RNR_TIMER); 116 if (ret) 117 printk(KERN_NOTICE "ib_modify_qp(IB_QP_MIN_RNR_TIMER): err=%d\n", -ret); 118 } 119 120 /* 121 * Connection established. 122 * We get here for both outgoing and incoming connection. 123 */ 124 void rds_ib_cm_connect_complete(struct rds_connection *conn, struct rdma_cm_event *event) 125 { 126 const struct rds_ib_connect_private *dp = NULL; 127 struct rds_ib_connection *ic = conn->c_transport_data; 128 struct ib_qp_attr qp_attr; 129 int err; 130 131 if (event->param.conn.private_data_len >= sizeof(*dp)) { 132 dp = event->param.conn.private_data; 133 134 /* make sure it isn't empty data */ 135 if (dp->dp_protocol_major) { 136 rds_ib_set_protocol(conn, 137 RDS_PROTOCOL(dp->dp_protocol_major, 138 dp->dp_protocol_minor)); 139 rds_ib_set_flow_control(conn, be32_to_cpu(dp->dp_credit)); 140 } 141 } 142 143 if (conn->c_version < RDS_PROTOCOL(3,1)) { 144 printk(KERN_NOTICE "RDS/IB: Connection to %pI4 version %u.%u failed," 145 " no longer supported\n", 146 &conn->c_faddr, 147 RDS_PROTOCOL_MAJOR(conn->c_version), 148 RDS_PROTOCOL_MINOR(conn->c_version)); 149 rds_conn_destroy(conn); 150 return; 151 } else { 152 printk(KERN_NOTICE "RDS/IB: connected to %pI4 version %u.%u%s\n", 153 &conn->c_faddr, 154 RDS_PROTOCOL_MAJOR(conn->c_version), 155 RDS_PROTOCOL_MINOR(conn->c_version), 156 ic->i_flowctl ? ", flow control" : ""); 157 } 158 159 /* 160 * Init rings and fill recv. this needs to wait until protocol negotiation 161 * is complete, since ring layout is different from 3.0 to 3.1. 162 */ 163 rds_ib_send_init_ring(ic); 164 rds_ib_recv_init_ring(ic); 165 /* Post receive buffers - as a side effect, this will update 166 * the posted credit count. */ 167 rds_ib_recv_refill(conn, 1); 168 169 /* Tune RNR behavior */ 170 rds_ib_tune_rnr(ic, &qp_attr); 171 172 qp_attr.qp_state = IB_QPS_RTS; 173 err = ib_modify_qp(ic->i_cm_id->qp, &qp_attr, IB_QP_STATE); 174 if (err) 175 printk(KERN_NOTICE "ib_modify_qp(IB_QP_STATE, RTS): err=%d\n", err); 176 177 /* update ib_device with this local ipaddr */ 178 err = rds_ib_update_ipaddr(ic->rds_ibdev, conn->c_laddr); 179 if (err) 180 printk(KERN_ERR "rds_ib_update_ipaddr failed (%d)\n", 181 err); 182 183 /* If the peer gave us the last packet it saw, process this as if 184 * we had received a regular ACK. */ 185 if (dp && dp->dp_ack_seq) 186 rds_send_drop_acked(conn, be64_to_cpu(dp->dp_ack_seq), NULL); 187 188 rds_connect_complete(conn); 189 } 190 191 static void rds_ib_cm_fill_conn_param(struct rds_connection *conn, 192 struct rdma_conn_param *conn_param, 193 struct rds_ib_connect_private *dp, 194 u32 protocol_version, 195 u32 max_responder_resources, 196 u32 max_initiator_depth) 197 { 198 struct rds_ib_connection *ic = conn->c_transport_data; 199 struct rds_ib_device *rds_ibdev = ic->rds_ibdev; 200 201 memset(conn_param, 0, sizeof(struct rdma_conn_param)); 202 203 conn_param->responder_resources = 204 min_t(u32, rds_ibdev->max_responder_resources, max_responder_resources); 205 conn_param->initiator_depth = 206 min_t(u32, rds_ibdev->max_initiator_depth, max_initiator_depth); 207 conn_param->retry_count = min_t(unsigned int, rds_ib_retry_count, 7); 208 conn_param->rnr_retry_count = 7; 209 210 if (dp) { 211 memset(dp, 0, sizeof(*dp)); 212 dp->dp_saddr = conn->c_laddr; 213 dp->dp_daddr = conn->c_faddr; 214 dp->dp_protocol_major = RDS_PROTOCOL_MAJOR(protocol_version); 215 dp->dp_protocol_minor = RDS_PROTOCOL_MINOR(protocol_version); 216 dp->dp_protocol_minor_mask = cpu_to_be16(RDS_IB_SUPPORTED_PROTOCOLS); 217 dp->dp_ack_seq = rds_ib_piggyb_ack(ic); 218 219 /* Advertise flow control */ 220 if (ic->i_flowctl) { 221 unsigned int credits; 222 223 credits = IB_GET_POST_CREDITS(atomic_read(&ic->i_credits)); 224 dp->dp_credit = cpu_to_be32(credits); 225 atomic_sub(IB_SET_POST_CREDITS(credits), &ic->i_credits); 226 } 227 228 conn_param->private_data = dp; 229 conn_param->private_data_len = sizeof(*dp); 230 } 231 } 232 233 static void rds_ib_cq_event_handler(struct ib_event *event, void *data) 234 { 235 rdsdebug("event %u (%s) data %p\n", 236 event->event, rds_ib_event_str(event->event), data); 237 } 238 239 static void rds_ib_qp_event_handler(struct ib_event *event, void *data) 240 { 241 struct rds_connection *conn = data; 242 struct rds_ib_connection *ic = conn->c_transport_data; 243 244 rdsdebug("conn %p ic %p event %u (%s)\n", conn, ic, event->event, 245 rds_ib_event_str(event->event)); 246 247 switch (event->event) { 248 case IB_EVENT_COMM_EST: 249 rdma_notify(ic->i_cm_id, IB_EVENT_COMM_EST); 250 break; 251 default: 252 rdsdebug("Fatal QP Event %u (%s) " 253 "- connection %pI4->%pI4, reconnecting\n", 254 event->event, rds_ib_event_str(event->event), 255 &conn->c_laddr, &conn->c_faddr); 256 rds_conn_drop(conn); 257 break; 258 } 259 } 260 261 /* 262 * This needs to be very careful to not leave IS_ERR pointers around for 263 * cleanup to trip over. 264 */ 265 static int rds_ib_setup_qp(struct rds_connection *conn) 266 { 267 struct rds_ib_connection *ic = conn->c_transport_data; 268 struct ib_device *dev = ic->i_cm_id->device; 269 struct ib_qp_init_attr attr; 270 struct rds_ib_device *rds_ibdev; 271 int ret; 272 273 /* 274 * It's normal to see a null device if an incoming connection races 275 * with device removal, so we don't print a warning. 276 */ 277 rds_ibdev = rds_ib_get_client_data(dev); 278 if (!rds_ibdev) 279 return -EOPNOTSUPP; 280 281 /* add the conn now so that connection establishment has the dev */ 282 rds_ib_add_conn(rds_ibdev, conn); 283 284 if (rds_ibdev->max_wrs < ic->i_send_ring.w_nr + 1) 285 rds_ib_ring_resize(&ic->i_send_ring, rds_ibdev->max_wrs - 1); 286 if (rds_ibdev->max_wrs < ic->i_recv_ring.w_nr + 1) 287 rds_ib_ring_resize(&ic->i_recv_ring, rds_ibdev->max_wrs - 1); 288 289 /* Protection domain and memory range */ 290 ic->i_pd = rds_ibdev->pd; 291 ic->i_mr = rds_ibdev->mr; 292 293 ic->i_send_cq = ib_create_cq(dev, rds_ib_send_cq_comp_handler, 294 rds_ib_cq_event_handler, conn, 295 ic->i_send_ring.w_nr + 1, 0); 296 if (IS_ERR(ic->i_send_cq)) { 297 ret = PTR_ERR(ic->i_send_cq); 298 ic->i_send_cq = NULL; 299 rdsdebug("ib_create_cq send failed: %d\n", ret); 300 goto out; 301 } 302 303 ic->i_recv_cq = ib_create_cq(dev, rds_ib_recv_cq_comp_handler, 304 rds_ib_cq_event_handler, conn, 305 ic->i_recv_ring.w_nr, 0); 306 if (IS_ERR(ic->i_recv_cq)) { 307 ret = PTR_ERR(ic->i_recv_cq); 308 ic->i_recv_cq = NULL; 309 rdsdebug("ib_create_cq recv failed: %d\n", ret); 310 goto out; 311 } 312 313 ret = ib_req_notify_cq(ic->i_send_cq, IB_CQ_NEXT_COMP); 314 if (ret) { 315 rdsdebug("ib_req_notify_cq send failed: %d\n", ret); 316 goto out; 317 } 318 319 ret = ib_req_notify_cq(ic->i_recv_cq, IB_CQ_SOLICITED); 320 if (ret) { 321 rdsdebug("ib_req_notify_cq recv failed: %d\n", ret); 322 goto out; 323 } 324 325 /* XXX negotiate max send/recv with remote? */ 326 memset(&attr, 0, sizeof(attr)); 327 attr.event_handler = rds_ib_qp_event_handler; 328 attr.qp_context = conn; 329 /* + 1 to allow for the single ack message */ 330 attr.cap.max_send_wr = ic->i_send_ring.w_nr + 1; 331 attr.cap.max_recv_wr = ic->i_recv_ring.w_nr + 1; 332 attr.cap.max_send_sge = rds_ibdev->max_sge; 333 attr.cap.max_recv_sge = RDS_IB_RECV_SGE; 334 attr.sq_sig_type = IB_SIGNAL_REQ_WR; 335 attr.qp_type = IB_QPT_RC; 336 attr.send_cq = ic->i_send_cq; 337 attr.recv_cq = ic->i_recv_cq; 338 339 /* 340 * XXX this can fail if max_*_wr is too large? Are we supposed 341 * to back off until we get a value that the hardware can support? 342 */ 343 ret = rdma_create_qp(ic->i_cm_id, ic->i_pd, &attr); 344 if (ret) { 345 rdsdebug("rdma_create_qp failed: %d\n", ret); 346 goto out; 347 } 348 349 ic->i_send_hdrs = ib_dma_alloc_coherent(dev, 350 ic->i_send_ring.w_nr * 351 sizeof(struct rds_header), 352 &ic->i_send_hdrs_dma, GFP_KERNEL); 353 if (!ic->i_send_hdrs) { 354 ret = -ENOMEM; 355 rdsdebug("ib_dma_alloc_coherent send failed\n"); 356 goto out; 357 } 358 359 ic->i_recv_hdrs = ib_dma_alloc_coherent(dev, 360 ic->i_recv_ring.w_nr * 361 sizeof(struct rds_header), 362 &ic->i_recv_hdrs_dma, GFP_KERNEL); 363 if (!ic->i_recv_hdrs) { 364 ret = -ENOMEM; 365 rdsdebug("ib_dma_alloc_coherent recv failed\n"); 366 goto out; 367 } 368 369 ic->i_ack = ib_dma_alloc_coherent(dev, sizeof(struct rds_header), 370 &ic->i_ack_dma, GFP_KERNEL); 371 if (!ic->i_ack) { 372 ret = -ENOMEM; 373 rdsdebug("ib_dma_alloc_coherent ack failed\n"); 374 goto out; 375 } 376 377 ic->i_sends = vmalloc_node(ic->i_send_ring.w_nr * sizeof(struct rds_ib_send_work), 378 ibdev_to_node(dev)); 379 if (!ic->i_sends) { 380 ret = -ENOMEM; 381 rdsdebug("send allocation failed\n"); 382 goto out; 383 } 384 memset(ic->i_sends, 0, ic->i_send_ring.w_nr * sizeof(struct rds_ib_send_work)); 385 386 ic->i_recvs = vmalloc_node(ic->i_recv_ring.w_nr * sizeof(struct rds_ib_recv_work), 387 ibdev_to_node(dev)); 388 if (!ic->i_recvs) { 389 ret = -ENOMEM; 390 rdsdebug("recv allocation failed\n"); 391 goto out; 392 } 393 memset(ic->i_recvs, 0, ic->i_recv_ring.w_nr * sizeof(struct rds_ib_recv_work)); 394 395 rds_ib_recv_init_ack(ic); 396 397 rdsdebug("conn %p pd %p mr %p cq %p %p\n", conn, ic->i_pd, ic->i_mr, 398 ic->i_send_cq, ic->i_recv_cq); 399 400 out: 401 rds_ib_dev_put(rds_ibdev); 402 return ret; 403 } 404 405 static u32 rds_ib_protocol_compatible(struct rdma_cm_event *event) 406 { 407 const struct rds_ib_connect_private *dp = event->param.conn.private_data; 408 u16 common; 409 u32 version = 0; 410 411 /* 412 * rdma_cm private data is odd - when there is any private data in the 413 * request, we will be given a pretty large buffer without telling us the 414 * original size. The only way to tell the difference is by looking at 415 * the contents, which are initialized to zero. 416 * If the protocol version fields aren't set, this is a connection attempt 417 * from an older version. This could could be 3.0 or 2.0 - we can't tell. 418 * We really should have changed this for OFED 1.3 :-( 419 */ 420 421 /* Be paranoid. RDS always has privdata */ 422 if (!event->param.conn.private_data_len) { 423 printk(KERN_NOTICE "RDS incoming connection has no private data, " 424 "rejecting\n"); 425 return 0; 426 } 427 428 /* Even if len is crap *now* I still want to check it. -ASG */ 429 if (event->param.conn.private_data_len < sizeof (*dp) || 430 dp->dp_protocol_major == 0) 431 return RDS_PROTOCOL_3_0; 432 433 common = be16_to_cpu(dp->dp_protocol_minor_mask) & RDS_IB_SUPPORTED_PROTOCOLS; 434 if (dp->dp_protocol_major == 3 && common) { 435 version = RDS_PROTOCOL_3_0; 436 while ((common >>= 1) != 0) 437 version++; 438 } else if (printk_ratelimit()) { 439 printk(KERN_NOTICE "RDS: Connection from %pI4 using " 440 "incompatible protocol version %u.%u\n", 441 &dp->dp_saddr, 442 dp->dp_protocol_major, 443 dp->dp_protocol_minor); 444 } 445 return version; 446 } 447 448 int rds_ib_cm_handle_connect(struct rdma_cm_id *cm_id, 449 struct rdma_cm_event *event) 450 { 451 __be64 lguid = cm_id->route.path_rec->sgid.global.interface_id; 452 __be64 fguid = cm_id->route.path_rec->dgid.global.interface_id; 453 const struct rds_ib_connect_private *dp = event->param.conn.private_data; 454 struct rds_ib_connect_private dp_rep; 455 struct rds_connection *conn = NULL; 456 struct rds_ib_connection *ic = NULL; 457 struct rdma_conn_param conn_param; 458 u32 version; 459 int err = 1, destroy = 1; 460 461 /* Check whether the remote protocol version matches ours. */ 462 version = rds_ib_protocol_compatible(event); 463 if (!version) 464 goto out; 465 466 rdsdebug("saddr %pI4 daddr %pI4 RDSv%u.%u lguid 0x%llx fguid " 467 "0x%llx\n", &dp->dp_saddr, &dp->dp_daddr, 468 RDS_PROTOCOL_MAJOR(version), RDS_PROTOCOL_MINOR(version), 469 (unsigned long long)be64_to_cpu(lguid), 470 (unsigned long long)be64_to_cpu(fguid)); 471 472 conn = rds_conn_create(dp->dp_daddr, dp->dp_saddr, &rds_ib_transport, 473 GFP_KERNEL); 474 if (IS_ERR(conn)) { 475 rdsdebug("rds_conn_create failed (%ld)\n", PTR_ERR(conn)); 476 conn = NULL; 477 goto out; 478 } 479 480 /* 481 * The connection request may occur while the 482 * previous connection exist, e.g. in case of failover. 483 * But as connections may be initiated simultaneously 484 * by both hosts, we have a random backoff mechanism - 485 * see the comment above rds_queue_reconnect() 486 */ 487 mutex_lock(&conn->c_cm_lock); 488 if (!rds_conn_transition(conn, RDS_CONN_DOWN, RDS_CONN_CONNECTING)) { 489 if (rds_conn_state(conn) == RDS_CONN_UP) { 490 rdsdebug("incoming connect while connecting\n"); 491 rds_conn_drop(conn); 492 rds_ib_stats_inc(s_ib_listen_closed_stale); 493 } else 494 if (rds_conn_state(conn) == RDS_CONN_CONNECTING) { 495 /* Wait and see - our connect may still be succeeding */ 496 rds_ib_stats_inc(s_ib_connect_raced); 497 } 498 goto out; 499 } 500 501 ic = conn->c_transport_data; 502 503 rds_ib_set_protocol(conn, version); 504 rds_ib_set_flow_control(conn, be32_to_cpu(dp->dp_credit)); 505 506 /* If the peer gave us the last packet it saw, process this as if 507 * we had received a regular ACK. */ 508 if (dp->dp_ack_seq) 509 rds_send_drop_acked(conn, be64_to_cpu(dp->dp_ack_seq), NULL); 510 511 BUG_ON(cm_id->context); 512 BUG_ON(ic->i_cm_id); 513 514 ic->i_cm_id = cm_id; 515 cm_id->context = conn; 516 517 /* We got halfway through setting up the ib_connection, if we 518 * fail now, we have to take the long route out of this mess. */ 519 destroy = 0; 520 521 err = rds_ib_setup_qp(conn); 522 if (err) { 523 rds_ib_conn_error(conn, "rds_ib_setup_qp failed (%d)\n", err); 524 goto out; 525 } 526 527 rds_ib_cm_fill_conn_param(conn, &conn_param, &dp_rep, version, 528 event->param.conn.responder_resources, 529 event->param.conn.initiator_depth); 530 531 /* rdma_accept() calls rdma_reject() internally if it fails */ 532 err = rdma_accept(cm_id, &conn_param); 533 if (err) 534 rds_ib_conn_error(conn, "rdma_accept failed (%d)\n", err); 535 536 out: 537 if (conn) 538 mutex_unlock(&conn->c_cm_lock); 539 if (err) 540 rdma_reject(cm_id, NULL, 0); 541 return destroy; 542 } 543 544 545 int rds_ib_cm_initiate_connect(struct rdma_cm_id *cm_id) 546 { 547 struct rds_connection *conn = cm_id->context; 548 struct rds_ib_connection *ic = conn->c_transport_data; 549 struct rdma_conn_param conn_param; 550 struct rds_ib_connect_private dp; 551 int ret; 552 553 /* If the peer doesn't do protocol negotiation, we must 554 * default to RDSv3.0 */ 555 rds_ib_set_protocol(conn, RDS_PROTOCOL_3_0); 556 ic->i_flowctl = rds_ib_sysctl_flow_control; /* advertise flow control */ 557 558 ret = rds_ib_setup_qp(conn); 559 if (ret) { 560 rds_ib_conn_error(conn, "rds_ib_setup_qp failed (%d)\n", ret); 561 goto out; 562 } 563 564 rds_ib_cm_fill_conn_param(conn, &conn_param, &dp, RDS_PROTOCOL_VERSION, 565 UINT_MAX, UINT_MAX); 566 ret = rdma_connect(cm_id, &conn_param); 567 if (ret) 568 rds_ib_conn_error(conn, "rdma_connect failed (%d)\n", ret); 569 570 out: 571 /* Beware - returning non-zero tells the rdma_cm to destroy 572 * the cm_id. We should certainly not do it as long as we still 573 * "own" the cm_id. */ 574 if (ret) { 575 if (ic->i_cm_id == cm_id) 576 ret = 0; 577 } 578 return ret; 579 } 580 581 int rds_ib_conn_connect(struct rds_connection *conn) 582 { 583 struct rds_ib_connection *ic = conn->c_transport_data; 584 struct sockaddr_in src, dest; 585 int ret; 586 587 /* XXX I wonder what affect the port space has */ 588 /* delegate cm event handler to rdma_transport */ 589 ic->i_cm_id = rdma_create_id(rds_rdma_cm_event_handler, conn, 590 RDMA_PS_TCP); 591 if (IS_ERR(ic->i_cm_id)) { 592 ret = PTR_ERR(ic->i_cm_id); 593 ic->i_cm_id = NULL; 594 rdsdebug("rdma_create_id() failed: %d\n", ret); 595 goto out; 596 } 597 598 rdsdebug("created cm id %p for conn %p\n", ic->i_cm_id, conn); 599 600 src.sin_family = AF_INET; 601 src.sin_addr.s_addr = (__force u32)conn->c_laddr; 602 src.sin_port = (__force u16)htons(0); 603 604 dest.sin_family = AF_INET; 605 dest.sin_addr.s_addr = (__force u32)conn->c_faddr; 606 dest.sin_port = (__force u16)htons(RDS_PORT); 607 608 ret = rdma_resolve_addr(ic->i_cm_id, (struct sockaddr *)&src, 609 (struct sockaddr *)&dest, 610 RDS_RDMA_RESOLVE_TIMEOUT_MS); 611 if (ret) { 612 rdsdebug("addr resolve failed for cm id %p: %d\n", ic->i_cm_id, 613 ret); 614 rdma_destroy_id(ic->i_cm_id); 615 ic->i_cm_id = NULL; 616 } 617 618 out: 619 return ret; 620 } 621 622 /* 623 * This is so careful about only cleaning up resources that were built up 624 * so that it can be called at any point during startup. In fact it 625 * can be called multiple times for a given connection. 626 */ 627 void rds_ib_conn_shutdown(struct rds_connection *conn) 628 { 629 struct rds_ib_connection *ic = conn->c_transport_data; 630 int err = 0; 631 632 rdsdebug("cm %p pd %p cq %p %p qp %p\n", ic->i_cm_id, 633 ic->i_pd, ic->i_send_cq, ic->i_recv_cq, 634 ic->i_cm_id ? ic->i_cm_id->qp : NULL); 635 636 if (ic->i_cm_id) { 637 struct ib_device *dev = ic->i_cm_id->device; 638 639 rdsdebug("disconnecting cm %p\n", ic->i_cm_id); 640 err = rdma_disconnect(ic->i_cm_id); 641 if (err) { 642 /* Actually this may happen quite frequently, when 643 * an outgoing connect raced with an incoming connect. 644 */ 645 rdsdebug("failed to disconnect, cm: %p err %d\n", 646 ic->i_cm_id, err); 647 } 648 649 /* 650 * We want to wait for tx and rx completion to finish 651 * before we tear down the connection, but we have to be 652 * careful not to get stuck waiting on a send ring that 653 * only has unsignaled sends in it. We've shutdown new 654 * sends before getting here so by waiting for signaled 655 * sends to complete we're ensured that there will be no 656 * more tx processing. 657 */ 658 wait_event(rds_ib_ring_empty_wait, 659 rds_ib_ring_empty(&ic->i_recv_ring) && 660 (atomic_read(&ic->i_signaled_sends) == 0)); 661 tasklet_kill(&ic->i_recv_tasklet); 662 663 if (ic->i_send_hdrs) 664 ib_dma_free_coherent(dev, 665 ic->i_send_ring.w_nr * 666 sizeof(struct rds_header), 667 ic->i_send_hdrs, 668 ic->i_send_hdrs_dma); 669 670 if (ic->i_recv_hdrs) 671 ib_dma_free_coherent(dev, 672 ic->i_recv_ring.w_nr * 673 sizeof(struct rds_header), 674 ic->i_recv_hdrs, 675 ic->i_recv_hdrs_dma); 676 677 if (ic->i_ack) 678 ib_dma_free_coherent(dev, sizeof(struct rds_header), 679 ic->i_ack, ic->i_ack_dma); 680 681 if (ic->i_sends) 682 rds_ib_send_clear_ring(ic); 683 if (ic->i_recvs) 684 rds_ib_recv_clear_ring(ic); 685 686 if (ic->i_cm_id->qp) 687 rdma_destroy_qp(ic->i_cm_id); 688 if (ic->i_send_cq) 689 ib_destroy_cq(ic->i_send_cq); 690 if (ic->i_recv_cq) 691 ib_destroy_cq(ic->i_recv_cq); 692 rdma_destroy_id(ic->i_cm_id); 693 694 /* 695 * Move connection back to the nodev list. 696 */ 697 if (ic->rds_ibdev) 698 rds_ib_remove_conn(ic->rds_ibdev, conn); 699 700 ic->i_cm_id = NULL; 701 ic->i_pd = NULL; 702 ic->i_mr = NULL; 703 ic->i_send_cq = NULL; 704 ic->i_recv_cq = NULL; 705 ic->i_send_hdrs = NULL; 706 ic->i_recv_hdrs = NULL; 707 ic->i_ack = NULL; 708 } 709 BUG_ON(ic->rds_ibdev); 710 711 /* Clear pending transmit */ 712 if (ic->i_data_op) { 713 struct rds_message *rm; 714 715 rm = container_of(ic->i_data_op, struct rds_message, data); 716 rds_message_put(rm); 717 ic->i_data_op = NULL; 718 } 719 720 /* Clear the ACK state */ 721 clear_bit(IB_ACK_IN_FLIGHT, &ic->i_ack_flags); 722 #ifdef KERNEL_HAS_ATOMIC64 723 atomic64_set(&ic->i_ack_next, 0); 724 #else 725 ic->i_ack_next = 0; 726 #endif 727 ic->i_ack_recv = 0; 728 729 /* Clear flow control state */ 730 ic->i_flowctl = 0; 731 atomic_set(&ic->i_credits, 0); 732 733 rds_ib_ring_init(&ic->i_send_ring, rds_ib_sysctl_max_send_wr); 734 rds_ib_ring_init(&ic->i_recv_ring, rds_ib_sysctl_max_recv_wr); 735 736 if (ic->i_ibinc) { 737 rds_inc_put(&ic->i_ibinc->ii_inc); 738 ic->i_ibinc = NULL; 739 } 740 741 vfree(ic->i_sends); 742 ic->i_sends = NULL; 743 vfree(ic->i_recvs); 744 ic->i_recvs = NULL; 745 } 746 747 int rds_ib_conn_alloc(struct rds_connection *conn, gfp_t gfp) 748 { 749 struct rds_ib_connection *ic; 750 unsigned long flags; 751 int ret; 752 753 /* XXX too lazy? */ 754 ic = kzalloc(sizeof(struct rds_ib_connection), GFP_KERNEL); 755 if (!ic) 756 return -ENOMEM; 757 758 ret = rds_ib_recv_alloc_caches(ic); 759 if (ret) { 760 kfree(ic); 761 return ret; 762 } 763 764 INIT_LIST_HEAD(&ic->ib_node); 765 tasklet_init(&ic->i_recv_tasklet, rds_ib_recv_tasklet_fn, 766 (unsigned long) ic); 767 mutex_init(&ic->i_recv_mutex); 768 #ifndef KERNEL_HAS_ATOMIC64 769 spin_lock_init(&ic->i_ack_lock); 770 #endif 771 atomic_set(&ic->i_signaled_sends, 0); 772 773 /* 774 * rds_ib_conn_shutdown() waits for these to be emptied so they 775 * must be initialized before it can be called. 776 */ 777 rds_ib_ring_init(&ic->i_send_ring, rds_ib_sysctl_max_send_wr); 778 rds_ib_ring_init(&ic->i_recv_ring, rds_ib_sysctl_max_recv_wr); 779 780 ic->conn = conn; 781 conn->c_transport_data = ic; 782 783 spin_lock_irqsave(&ib_nodev_conns_lock, flags); 784 list_add_tail(&ic->ib_node, &ib_nodev_conns); 785 spin_unlock_irqrestore(&ib_nodev_conns_lock, flags); 786 787 788 rdsdebug("conn %p conn ic %p\n", conn, conn->c_transport_data); 789 return 0; 790 } 791 792 /* 793 * Free a connection. Connection must be shut down and not set for reconnect. 794 */ 795 void rds_ib_conn_free(void *arg) 796 { 797 struct rds_ib_connection *ic = arg; 798 spinlock_t *lock_ptr; 799 800 rdsdebug("ic %p\n", ic); 801 802 /* 803 * Conn is either on a dev's list or on the nodev list. 804 * A race with shutdown() or connect() would cause problems 805 * (since rds_ibdev would change) but that should never happen. 806 */ 807 lock_ptr = ic->rds_ibdev ? &ic->rds_ibdev->spinlock : &ib_nodev_conns_lock; 808 809 spin_lock_irq(lock_ptr); 810 list_del(&ic->ib_node); 811 spin_unlock_irq(lock_ptr); 812 813 rds_ib_recv_free_caches(ic); 814 815 kfree(ic); 816 } 817 818 819 /* 820 * An error occurred on the connection 821 */ 822 void 823 __rds_ib_conn_error(struct rds_connection *conn, const char *fmt, ...) 824 { 825 va_list ap; 826 827 rds_conn_drop(conn); 828 829 va_start(ap, fmt); 830 vprintk(fmt, ap); 831 va_end(ap); 832 } 833