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/list.h> 35 #include <linux/slab.h> 36 #include <linux/export.h> 37 #include <net/inet_hashtables.h> 38 39 #include "rds.h" 40 #include "loop.h" 41 42 #define RDS_CONNECTION_HASH_BITS 12 43 #define RDS_CONNECTION_HASH_ENTRIES (1 << RDS_CONNECTION_HASH_BITS) 44 #define RDS_CONNECTION_HASH_MASK (RDS_CONNECTION_HASH_ENTRIES - 1) 45 46 /* converting this to RCU is a chore for another day.. */ 47 static DEFINE_SPINLOCK(rds_conn_lock); 48 static unsigned long rds_conn_count; 49 static struct hlist_head rds_conn_hash[RDS_CONNECTION_HASH_ENTRIES]; 50 static struct kmem_cache *rds_conn_slab; 51 52 static struct hlist_head *rds_conn_bucket(__be32 laddr, __be32 faddr) 53 { 54 static u32 rds_hash_secret __read_mostly; 55 56 unsigned long hash; 57 58 net_get_random_once(&rds_hash_secret, sizeof(rds_hash_secret)); 59 60 /* Pass NULL, don't need struct net for hash */ 61 hash = __inet_ehashfn(be32_to_cpu(laddr), 0, 62 be32_to_cpu(faddr), 0, 63 rds_hash_secret); 64 return &rds_conn_hash[hash & RDS_CONNECTION_HASH_MASK]; 65 } 66 67 #define rds_conn_info_set(var, test, suffix) do { \ 68 if (test) \ 69 var |= RDS_INFO_CONNECTION_FLAG_##suffix; \ 70 } while (0) 71 72 /* rcu read lock must be held or the connection spinlock */ 73 static struct rds_connection *rds_conn_lookup(struct net *net, 74 struct hlist_head *head, 75 __be32 laddr, __be32 faddr, 76 struct rds_transport *trans) 77 { 78 struct rds_connection *conn, *ret = NULL; 79 80 hlist_for_each_entry_rcu(conn, head, c_hash_node) { 81 if (conn->c_faddr == faddr && conn->c_laddr == laddr && 82 conn->c_trans == trans && net == rds_conn_net(conn)) { 83 ret = conn; 84 break; 85 } 86 } 87 rdsdebug("returning conn %p for %pI4 -> %pI4\n", ret, 88 &laddr, &faddr); 89 return ret; 90 } 91 92 /* 93 * This is called by transports as they're bringing down a connection. 94 * It clears partial message state so that the transport can start sending 95 * and receiving over this connection again in the future. It is up to 96 * the transport to have serialized this call with its send and recv. 97 */ 98 static void rds_conn_path_reset(struct rds_conn_path *cp) 99 { 100 struct rds_connection *conn = cp->cp_conn; 101 102 rdsdebug("connection %pI4 to %pI4 reset\n", 103 &conn->c_laddr, &conn->c_faddr); 104 105 rds_stats_inc(s_conn_reset); 106 rds_send_path_reset(cp); 107 cp->cp_flags = 0; 108 109 /* Do not clear next_rx_seq here, else we cannot distinguish 110 * retransmitted packets from new packets, and will hand all 111 * of them to the application. That is not consistent with the 112 * reliability guarantees of RDS. */ 113 } 114 115 static void __rds_conn_path_init(struct rds_connection *conn, 116 struct rds_conn_path *cp, bool is_outgoing) 117 { 118 spin_lock_init(&cp->cp_lock); 119 cp->cp_next_tx_seq = 1; 120 init_waitqueue_head(&cp->cp_waitq); 121 INIT_LIST_HEAD(&cp->cp_send_queue); 122 INIT_LIST_HEAD(&cp->cp_retrans); 123 124 cp->cp_conn = conn; 125 atomic_set(&cp->cp_state, RDS_CONN_DOWN); 126 cp->cp_send_gen = 0; 127 /* cp_outgoing is per-path. So we can only set it here 128 * for the single-path transports. 129 */ 130 if (!conn->c_trans->t_mp_capable) 131 cp->cp_outgoing = (is_outgoing ? 1 : 0); 132 cp->cp_reconnect_jiffies = 0; 133 INIT_DELAYED_WORK(&cp->cp_send_w, rds_send_worker); 134 INIT_DELAYED_WORK(&cp->cp_recv_w, rds_recv_worker); 135 INIT_DELAYED_WORK(&cp->cp_conn_w, rds_connect_worker); 136 INIT_WORK(&cp->cp_down_w, rds_shutdown_worker); 137 mutex_init(&cp->cp_cm_lock); 138 cp->cp_flags = 0; 139 } 140 141 /* 142 * There is only every one 'conn' for a given pair of addresses in the 143 * system at a time. They contain messages to be retransmitted and so 144 * span the lifetime of the actual underlying transport connections. 145 * 146 * For now they are not garbage collected once they're created. They 147 * are torn down as the module is removed, if ever. 148 */ 149 static struct rds_connection *__rds_conn_create(struct net *net, 150 __be32 laddr, __be32 faddr, 151 struct rds_transport *trans, gfp_t gfp, 152 int is_outgoing) 153 { 154 struct rds_connection *conn, *parent = NULL; 155 struct hlist_head *head = rds_conn_bucket(laddr, faddr); 156 struct rds_transport *loop_trans; 157 unsigned long flags; 158 int ret, i; 159 160 rcu_read_lock(); 161 conn = rds_conn_lookup(net, head, laddr, faddr, trans); 162 if (conn && conn->c_loopback && conn->c_trans != &rds_loop_transport && 163 laddr == faddr && !is_outgoing) { 164 /* This is a looped back IB connection, and we're 165 * called by the code handling the incoming connect. 166 * We need a second connection object into which we 167 * can stick the other QP. */ 168 parent = conn; 169 conn = parent->c_passive; 170 } 171 rcu_read_unlock(); 172 if (conn) 173 goto out; 174 175 conn = kmem_cache_zalloc(rds_conn_slab, gfp); 176 if (!conn) { 177 conn = ERR_PTR(-ENOMEM); 178 goto out; 179 } 180 181 INIT_HLIST_NODE(&conn->c_hash_node); 182 conn->c_laddr = laddr; 183 conn->c_faddr = faddr; 184 185 rds_conn_net_set(conn, net); 186 187 ret = rds_cong_get_maps(conn); 188 if (ret) { 189 kmem_cache_free(rds_conn_slab, conn); 190 conn = ERR_PTR(ret); 191 goto out; 192 } 193 194 /* 195 * This is where a connection becomes loopback. If *any* RDS sockets 196 * can bind to the destination address then we'd rather the messages 197 * flow through loopback rather than either transport. 198 */ 199 loop_trans = rds_trans_get_preferred(net, faddr); 200 if (loop_trans) { 201 rds_trans_put(loop_trans); 202 conn->c_loopback = 1; 203 if (is_outgoing && trans->t_prefer_loopback) { 204 /* "outgoing" connection - and the transport 205 * says it wants the connection handled by the 206 * loopback transport. This is what TCP does. 207 */ 208 trans = &rds_loop_transport; 209 } 210 } 211 212 conn->c_trans = trans; 213 214 init_waitqueue_head(&conn->c_hs_waitq); 215 for (i = 0; i < RDS_MPATH_WORKERS; i++) { 216 __rds_conn_path_init(conn, &conn->c_path[i], 217 is_outgoing); 218 conn->c_path[i].cp_index = i; 219 } 220 ret = trans->conn_alloc(conn, gfp); 221 if (ret) { 222 kmem_cache_free(rds_conn_slab, conn); 223 conn = ERR_PTR(ret); 224 goto out; 225 } 226 227 rdsdebug("allocated conn %p for %pI4 -> %pI4 over %s %s\n", 228 conn, &laddr, &faddr, 229 trans->t_name ? trans->t_name : "[unknown]", 230 is_outgoing ? "(outgoing)" : ""); 231 232 /* 233 * Since we ran without holding the conn lock, someone could 234 * have created the same conn (either normal or passive) in the 235 * interim. We check while holding the lock. If we won, we complete 236 * init and return our conn. If we lost, we rollback and return the 237 * other one. 238 */ 239 spin_lock_irqsave(&rds_conn_lock, flags); 240 if (parent) { 241 /* Creating passive conn */ 242 if (parent->c_passive) { 243 trans->conn_free(conn->c_path[0].cp_transport_data); 244 kmem_cache_free(rds_conn_slab, conn); 245 conn = parent->c_passive; 246 } else { 247 parent->c_passive = conn; 248 rds_cong_add_conn(conn); 249 rds_conn_count++; 250 } 251 } else { 252 /* Creating normal conn */ 253 struct rds_connection *found; 254 255 found = rds_conn_lookup(net, head, laddr, faddr, trans); 256 if (found) { 257 struct rds_conn_path *cp; 258 int i; 259 260 for (i = 0; i < RDS_MPATH_WORKERS; i++) { 261 cp = &conn->c_path[i]; 262 /* The ->conn_alloc invocation may have 263 * allocated resource for all paths, so all 264 * of them may have to be freed here. 265 */ 266 if (cp->cp_transport_data) 267 trans->conn_free(cp->cp_transport_data); 268 } 269 kmem_cache_free(rds_conn_slab, conn); 270 conn = found; 271 } else { 272 conn->c_my_gen_num = rds_gen_num; 273 conn->c_peer_gen_num = 0; 274 hlist_add_head_rcu(&conn->c_hash_node, head); 275 rds_cong_add_conn(conn); 276 rds_conn_count++; 277 } 278 } 279 spin_unlock_irqrestore(&rds_conn_lock, flags); 280 281 out: 282 return conn; 283 } 284 285 struct rds_connection *rds_conn_create(struct net *net, 286 __be32 laddr, __be32 faddr, 287 struct rds_transport *trans, gfp_t gfp) 288 { 289 return __rds_conn_create(net, laddr, faddr, trans, gfp, 0); 290 } 291 EXPORT_SYMBOL_GPL(rds_conn_create); 292 293 struct rds_connection *rds_conn_create_outgoing(struct net *net, 294 __be32 laddr, __be32 faddr, 295 struct rds_transport *trans, gfp_t gfp) 296 { 297 return __rds_conn_create(net, laddr, faddr, trans, gfp, 1); 298 } 299 EXPORT_SYMBOL_GPL(rds_conn_create_outgoing); 300 301 void rds_conn_shutdown(struct rds_conn_path *cp) 302 { 303 struct rds_connection *conn = cp->cp_conn; 304 305 /* shut it down unless it's down already */ 306 if (!rds_conn_path_transition(cp, RDS_CONN_DOWN, RDS_CONN_DOWN)) { 307 /* 308 * Quiesce the connection mgmt handlers before we start tearing 309 * things down. We don't hold the mutex for the entire 310 * duration of the shutdown operation, else we may be 311 * deadlocking with the CM handler. Instead, the CM event 312 * handler is supposed to check for state DISCONNECTING 313 */ 314 mutex_lock(&cp->cp_cm_lock); 315 if (!rds_conn_path_transition(cp, RDS_CONN_UP, 316 RDS_CONN_DISCONNECTING) && 317 !rds_conn_path_transition(cp, RDS_CONN_ERROR, 318 RDS_CONN_DISCONNECTING)) { 319 rds_conn_path_error(cp, 320 "shutdown called in state %d\n", 321 atomic_read(&cp->cp_state)); 322 mutex_unlock(&cp->cp_cm_lock); 323 return; 324 } 325 mutex_unlock(&cp->cp_cm_lock); 326 327 wait_event(cp->cp_waitq, 328 !test_bit(RDS_IN_XMIT, &cp->cp_flags)); 329 wait_event(cp->cp_waitq, 330 !test_bit(RDS_RECV_REFILL, &cp->cp_flags)); 331 332 conn->c_trans->conn_path_shutdown(cp); 333 rds_conn_path_reset(cp); 334 335 if (!rds_conn_path_transition(cp, RDS_CONN_DISCONNECTING, 336 RDS_CONN_DOWN) && 337 !rds_conn_path_transition(cp, RDS_CONN_ERROR, 338 RDS_CONN_DOWN)) { 339 /* This can happen - eg when we're in the middle of tearing 340 * down the connection, and someone unloads the rds module. 341 * Quite reproducible with loopback connections. 342 * Mostly harmless. 343 * 344 * Note that this also happens with rds-tcp because 345 * we could have triggered rds_conn_path_drop in irq 346 * mode from rds_tcp_state change on the receipt of 347 * a FIN, thus we need to recheck for RDS_CONN_ERROR 348 * here. 349 */ 350 rds_conn_path_error(cp, "%s: failed to transition " 351 "to state DOWN, current state " 352 "is %d\n", __func__, 353 atomic_read(&cp->cp_state)); 354 return; 355 } 356 } 357 358 /* Then reconnect if it's still live. 359 * The passive side of an IB loopback connection is never added 360 * to the conn hash, so we never trigger a reconnect on this 361 * conn - the reconnect is always triggered by the active peer. */ 362 cancel_delayed_work_sync(&cp->cp_conn_w); 363 rcu_read_lock(); 364 if (!hlist_unhashed(&conn->c_hash_node)) { 365 rcu_read_unlock(); 366 rds_queue_reconnect(cp); 367 } else { 368 rcu_read_unlock(); 369 } 370 } 371 372 /* destroy a single rds_conn_path. rds_conn_destroy() iterates over 373 * all paths using rds_conn_path_destroy() 374 */ 375 static void rds_conn_path_destroy(struct rds_conn_path *cp) 376 { 377 struct rds_message *rm, *rtmp; 378 379 if (!cp->cp_transport_data) 380 return; 381 382 rds_conn_path_drop(cp); 383 flush_work(&cp->cp_down_w); 384 385 /* make sure lingering queued work won't try to ref the conn */ 386 cancel_delayed_work_sync(&cp->cp_send_w); 387 cancel_delayed_work_sync(&cp->cp_recv_w); 388 389 /* tear down queued messages */ 390 list_for_each_entry_safe(rm, rtmp, 391 &cp->cp_send_queue, 392 m_conn_item) { 393 list_del_init(&rm->m_conn_item); 394 BUG_ON(!list_empty(&rm->m_sock_item)); 395 rds_message_put(rm); 396 } 397 if (cp->cp_xmit_rm) 398 rds_message_put(cp->cp_xmit_rm); 399 400 cp->cp_conn->c_trans->conn_free(cp->cp_transport_data); 401 } 402 403 /* 404 * Stop and free a connection. 405 * 406 * This can only be used in very limited circumstances. It assumes that once 407 * the conn has been shutdown that no one else is referencing the connection. 408 * We can only ensure this in the rmmod path in the current code. 409 */ 410 void rds_conn_destroy(struct rds_connection *conn) 411 { 412 unsigned long flags; 413 int i; 414 struct rds_conn_path *cp; 415 416 rdsdebug("freeing conn %p for %pI4 -> " 417 "%pI4\n", conn, &conn->c_laddr, 418 &conn->c_faddr); 419 420 /* Ensure conn will not be scheduled for reconnect */ 421 spin_lock_irq(&rds_conn_lock); 422 hlist_del_init_rcu(&conn->c_hash_node); 423 spin_unlock_irq(&rds_conn_lock); 424 synchronize_rcu(); 425 426 /* shut the connection down */ 427 for (i = 0; i < RDS_MPATH_WORKERS; i++) { 428 cp = &conn->c_path[i]; 429 rds_conn_path_destroy(cp); 430 BUG_ON(!list_empty(&cp->cp_retrans)); 431 } 432 433 /* 434 * The congestion maps aren't freed up here. They're 435 * freed by rds_cong_exit() after all the connections 436 * have been freed. 437 */ 438 rds_cong_remove_conn(conn); 439 440 put_net(conn->c_net); 441 kmem_cache_free(rds_conn_slab, conn); 442 443 spin_lock_irqsave(&rds_conn_lock, flags); 444 rds_conn_count--; 445 spin_unlock_irqrestore(&rds_conn_lock, flags); 446 } 447 EXPORT_SYMBOL_GPL(rds_conn_destroy); 448 449 static void rds_conn_message_info(struct socket *sock, unsigned int len, 450 struct rds_info_iterator *iter, 451 struct rds_info_lengths *lens, 452 int want_send) 453 { 454 struct hlist_head *head; 455 struct list_head *list; 456 struct rds_connection *conn; 457 struct rds_message *rm; 458 unsigned int total = 0; 459 unsigned long flags; 460 size_t i; 461 int j; 462 463 len /= sizeof(struct rds_info_message); 464 465 rcu_read_lock(); 466 467 for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash); 468 i++, head++) { 469 hlist_for_each_entry_rcu(conn, head, c_hash_node) { 470 struct rds_conn_path *cp; 471 472 for (j = 0; j < RDS_MPATH_WORKERS; j++) { 473 cp = &conn->c_path[j]; 474 if (want_send) 475 list = &cp->cp_send_queue; 476 else 477 list = &cp->cp_retrans; 478 479 spin_lock_irqsave(&cp->cp_lock, flags); 480 481 /* XXX too lazy to maintain counts.. */ 482 list_for_each_entry(rm, list, m_conn_item) { 483 total++; 484 if (total <= len) 485 rds_inc_info_copy(&rm->m_inc, 486 iter, 487 conn->c_laddr, 488 conn->c_faddr, 489 0); 490 } 491 492 spin_unlock_irqrestore(&cp->cp_lock, flags); 493 if (!conn->c_trans->t_mp_capable) 494 break; 495 } 496 } 497 } 498 rcu_read_unlock(); 499 500 lens->nr = total; 501 lens->each = sizeof(struct rds_info_message); 502 } 503 504 static void rds_conn_message_info_send(struct socket *sock, unsigned int len, 505 struct rds_info_iterator *iter, 506 struct rds_info_lengths *lens) 507 { 508 rds_conn_message_info(sock, len, iter, lens, 1); 509 } 510 511 static void rds_conn_message_info_retrans(struct socket *sock, 512 unsigned int len, 513 struct rds_info_iterator *iter, 514 struct rds_info_lengths *lens) 515 { 516 rds_conn_message_info(sock, len, iter, lens, 0); 517 } 518 519 void rds_for_each_conn_info(struct socket *sock, unsigned int len, 520 struct rds_info_iterator *iter, 521 struct rds_info_lengths *lens, 522 int (*visitor)(struct rds_connection *, void *), 523 size_t item_len) 524 { 525 uint64_t buffer[(item_len + 7) / 8]; 526 struct hlist_head *head; 527 struct rds_connection *conn; 528 size_t i; 529 530 rcu_read_lock(); 531 532 lens->nr = 0; 533 lens->each = item_len; 534 535 for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash); 536 i++, head++) { 537 hlist_for_each_entry_rcu(conn, head, c_hash_node) { 538 539 /* XXX no c_lock usage.. */ 540 if (!visitor(conn, buffer)) 541 continue; 542 543 /* We copy as much as we can fit in the buffer, 544 * but we count all items so that the caller 545 * can resize the buffer. */ 546 if (len >= item_len) { 547 rds_info_copy(iter, buffer, item_len); 548 len -= item_len; 549 } 550 lens->nr++; 551 } 552 } 553 rcu_read_unlock(); 554 } 555 EXPORT_SYMBOL_GPL(rds_for_each_conn_info); 556 557 static void rds_walk_conn_path_info(struct socket *sock, unsigned int len, 558 struct rds_info_iterator *iter, 559 struct rds_info_lengths *lens, 560 int (*visitor)(struct rds_conn_path *, void *), 561 size_t item_len) 562 { 563 u64 buffer[(item_len + 7) / 8]; 564 struct hlist_head *head; 565 struct rds_connection *conn; 566 size_t i; 567 int j; 568 569 rcu_read_lock(); 570 571 lens->nr = 0; 572 lens->each = item_len; 573 574 for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash); 575 i++, head++) { 576 hlist_for_each_entry_rcu(conn, head, c_hash_node) { 577 struct rds_conn_path *cp; 578 579 for (j = 0; j < RDS_MPATH_WORKERS; j++) { 580 cp = &conn->c_path[j]; 581 582 /* XXX no cp_lock usage.. */ 583 if (!visitor(cp, buffer)) 584 continue; 585 if (!conn->c_trans->t_mp_capable) 586 break; 587 } 588 589 /* We copy as much as we can fit in the buffer, 590 * but we count all items so that the caller 591 * can resize the buffer. 592 */ 593 if (len >= item_len) { 594 rds_info_copy(iter, buffer, item_len); 595 len -= item_len; 596 } 597 lens->nr++; 598 } 599 } 600 rcu_read_unlock(); 601 } 602 603 static int rds_conn_info_visitor(struct rds_conn_path *cp, void *buffer) 604 { 605 struct rds_info_connection *cinfo = buffer; 606 607 cinfo->next_tx_seq = cp->cp_next_tx_seq; 608 cinfo->next_rx_seq = cp->cp_next_rx_seq; 609 cinfo->laddr = cp->cp_conn->c_laddr; 610 cinfo->faddr = cp->cp_conn->c_faddr; 611 strncpy(cinfo->transport, cp->cp_conn->c_trans->t_name, 612 sizeof(cinfo->transport)); 613 cinfo->flags = 0; 614 615 rds_conn_info_set(cinfo->flags, test_bit(RDS_IN_XMIT, &cp->cp_flags), 616 SENDING); 617 /* XXX Future: return the state rather than these funky bits */ 618 rds_conn_info_set(cinfo->flags, 619 atomic_read(&cp->cp_state) == RDS_CONN_CONNECTING, 620 CONNECTING); 621 rds_conn_info_set(cinfo->flags, 622 atomic_read(&cp->cp_state) == RDS_CONN_UP, 623 CONNECTED); 624 return 1; 625 } 626 627 static void rds_conn_info(struct socket *sock, unsigned int len, 628 struct rds_info_iterator *iter, 629 struct rds_info_lengths *lens) 630 { 631 rds_walk_conn_path_info(sock, len, iter, lens, 632 rds_conn_info_visitor, 633 sizeof(struct rds_info_connection)); 634 } 635 636 int rds_conn_init(void) 637 { 638 rds_conn_slab = kmem_cache_create("rds_connection", 639 sizeof(struct rds_connection), 640 0, 0, NULL); 641 if (!rds_conn_slab) 642 return -ENOMEM; 643 644 rds_info_register_func(RDS_INFO_CONNECTIONS, rds_conn_info); 645 rds_info_register_func(RDS_INFO_SEND_MESSAGES, 646 rds_conn_message_info_send); 647 rds_info_register_func(RDS_INFO_RETRANS_MESSAGES, 648 rds_conn_message_info_retrans); 649 650 return 0; 651 } 652 653 void rds_conn_exit(void) 654 { 655 rds_loop_exit(); 656 657 WARN_ON(!hlist_empty(rds_conn_hash)); 658 659 kmem_cache_destroy(rds_conn_slab); 660 661 rds_info_deregister_func(RDS_INFO_CONNECTIONS, rds_conn_info); 662 rds_info_deregister_func(RDS_INFO_SEND_MESSAGES, 663 rds_conn_message_info_send); 664 rds_info_deregister_func(RDS_INFO_RETRANS_MESSAGES, 665 rds_conn_message_info_retrans); 666 } 667 668 /* 669 * Force a disconnect 670 */ 671 void rds_conn_path_drop(struct rds_conn_path *cp) 672 { 673 atomic_set(&cp->cp_state, RDS_CONN_ERROR); 674 queue_work(rds_wq, &cp->cp_down_w); 675 } 676 EXPORT_SYMBOL_GPL(rds_conn_path_drop); 677 678 void rds_conn_drop(struct rds_connection *conn) 679 { 680 WARN_ON(conn->c_trans->t_mp_capable); 681 rds_conn_path_drop(&conn->c_path[0]); 682 } 683 EXPORT_SYMBOL_GPL(rds_conn_drop); 684 685 /* 686 * If the connection is down, trigger a connect. We may have scheduled a 687 * delayed reconnect however - in this case we should not interfere. 688 */ 689 void rds_conn_path_connect_if_down(struct rds_conn_path *cp) 690 { 691 if (rds_conn_path_state(cp) == RDS_CONN_DOWN && 692 !test_and_set_bit(RDS_RECONNECT_PENDING, &cp->cp_flags)) 693 queue_delayed_work(rds_wq, &cp->cp_conn_w, 0); 694 } 695 EXPORT_SYMBOL_GPL(rds_conn_path_connect_if_down); 696 697 void rds_conn_connect_if_down(struct rds_connection *conn) 698 { 699 WARN_ON(conn->c_trans->t_mp_capable); 700 rds_conn_path_connect_if_down(&conn->c_path[0]); 701 } 702 EXPORT_SYMBOL_GPL(rds_conn_connect_if_down); 703 704 void 705 __rds_conn_path_error(struct rds_conn_path *cp, const char *fmt, ...) 706 { 707 va_list ap; 708 709 va_start(ap, fmt); 710 vprintk(fmt, ap); 711 va_end(ap); 712 713 rds_conn_path_drop(cp); 714 } 715