net/rds/connection.c

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