fib_hash: RCU conversion phase 2

[mv-sheeva.git] / net / rds / connection.c

diff --git a/net/rds/connection.c b/net/rds/connection.c
index 180b83ab260754191296c6eeb29748c3ee3e4f8c..870992e08cae2e86681d569b3ac27d83b6dffb89 100644 (file)
--- a/net/rds/connection.c
+++ b/net/rds/connection.c
@@ -117,10 +117,10 @@ static struct rds_connection *__rds_conn_create(__be32 laddr, __be32 faddr,
 {
        struct rds_connection *conn, *parent = NULL;
        struct hlist_head *head = rds_conn_bucket(laddr, faddr);
+       struct rds_transport *loop_trans;
        unsigned long flags;
        int ret;
 
-
        rcu_read_lock();
        conn = rds_conn_lookup(head, laddr, faddr, trans);
        if (conn && conn->c_loopback && conn->c_trans != &rds_loop_transport &&
@@ -148,9 +148,7 @@ static struct rds_connection *__rds_conn_create(__be32 laddr, __be32 faddr,
        spin_lock_init(&conn->c_lock);
        conn->c_next_tx_seq = 1;
 
-       spin_lock_init(&conn->c_send_lock);
-       atomic_set(&conn->c_send_generation, 1);
-       atomic_set(&conn->c_senders, 0);
+       init_waitqueue_head(&conn->c_waitq);
        INIT_LIST_HEAD(&conn->c_send_queue);
        INIT_LIST_HEAD(&conn->c_retrans);
 
@@ -166,7 +164,9 @@ static struct rds_connection *__rds_conn_create(__be32 laddr, __be32 faddr,
         * can bind to the destination address then we'd rather the messages
         * flow through loopback rather than either transport.
         */
-       if (rds_trans_get_preferred(faddr)) {
+       loop_trans = rds_trans_get_preferred(faddr);
+       if (loop_trans) {
+               rds_trans_put(loop_trans);
                conn->c_loopback = 1;
                if (is_outgoing && trans->t_prefer_loopback) {
                        /* "outgoing" connection - and the transport
@@ -275,15 +275,8 @@ void rds_conn_shutdown(struct rds_connection *conn)
                }
                mutex_unlock(&conn->c_cm_lock);
 
-               /* verify everybody's out of rds_send_xmit() */
-               spin_lock_irq(&conn->c_send_lock);
-               spin_unlock_irq(&conn->c_send_lock);
-
-               while(atomic_read(&conn->c_senders)) {
-                       schedule_timeout(1);
-                       spin_lock_irq(&conn->c_send_lock);
-                       spin_unlock_irq(&conn->c_send_lock);
-               }
+               wait_event(conn->c_waitq,
+                          !test_bit(RDS_IN_XMIT, &conn->c_flags));
 
                conn->c_trans->conn_shutdown(conn);
                rds_conn_reset(conn);
@@ -319,10 +312,15 @@ void rds_conn_shutdown(struct rds_connection *conn)
 
 /*
  * Stop and free a connection.
+ *
+ * This can only be used in very limited circumstances.  It assumes that once
+ * the conn has been shutdown that no one else is referencing the connection.
+ * We can only ensure this in the rmmod path in the current code.
  */
 void rds_conn_destroy(struct rds_connection *conn)
 {
        struct rds_message *rm, *rtmp;
+       unsigned long flags;
 
        rdsdebug("freeing conn %p for %pI4 -> "
                 "%pI4\n", conn, &conn->c_laddr,
@@ -332,10 +330,15 @@ void rds_conn_destroy(struct rds_connection *conn)
        spin_lock_irq(&rds_conn_lock);
        hlist_del_init_rcu(&conn->c_hash_node);
        spin_unlock_irq(&rds_conn_lock);
-
        synchronize_rcu();
 
-       rds_conn_shutdown(conn);
+       /* shut the connection down */
+       rds_conn_drop(conn);
+       flush_work(&conn->c_down_w);
+
+       /* make sure lingering queued work won't try to ref the conn */
+       cancel_delayed_work_sync(&conn->c_send_w);
+       cancel_delayed_work_sync(&conn->c_recv_w);
 
        /* tear down queued messages */
        list_for_each_entry_safe(rm, rtmp,
@@ -360,7 +363,9 @@ void rds_conn_destroy(struct rds_connection *conn)
        BUG_ON(!list_empty(&conn->c_retrans));
        kmem_cache_free(rds_conn_slab, conn);
 
+       spin_lock_irqsave(&rds_conn_lock, flags);
        rds_conn_count--;
+       spin_unlock_irqrestore(&rds_conn_lock, flags);
 }
 EXPORT_SYMBOL_GPL(rds_conn_destroy);
 
@@ -375,6 +380,7 @@ static void rds_conn_message_info(struct socket *sock, unsigned int len,
        struct rds_connection *conn;
        struct rds_message *rm;
        unsigned int total = 0;
+       unsigned long flags;
        size_t i;
 
        len /= sizeof(struct rds_info_message);
@@ -389,7 +395,7 @@ static void rds_conn_message_info(struct socket *sock, unsigned int len,
                        else
                                list = &conn->c_retrans;
 
-                       spin_lock(&conn->c_lock);
+                       spin_lock_irqsave(&conn->c_lock, flags);
 
                        /* XXX too lazy to maintain counts.. */
                        list_for_each_entry(rm, list, m_conn_item) {
@@ -400,7 +406,7 @@ static void rds_conn_message_info(struct socket *sock, unsigned int len,
                                                          conn->c_faddr, 0);
                        }
 
-                       spin_unlock(&conn->c_lock);
+                       spin_unlock_irqrestore(&conn->c_lock, flags);
                }
        }
        rcu_read_unlock();
@@ -476,8 +482,8 @@ static int rds_conn_info_visitor(struct rds_connection *conn,
                sizeof(cinfo->transport));
        cinfo->flags = 0;
 
-       rds_conn_info_set(cinfo->flags,
-                         spin_is_locked(&conn->c_send_lock), SENDING);
+       rds_conn_info_set(cinfo->flags, test_bit(RDS_IN_XMIT, &conn->c_flags),
+                         SENDING);
        /* XXX Future: return the state rather than these funky bits */
        rds_conn_info_set(cinfo->flags,
                          atomic_read(&conn->c_state) == RDS_CONN_CONNECTING,
@@ -497,7 +503,7 @@ static void rds_conn_info(struct socket *sock, unsigned int len,
                                sizeof(struct rds_info_connection));
 }
 
-int __init rds_conn_init(void)
+int rds_conn_init(void)
 {
        rds_conn_slab = kmem_cache_create("rds_connection",
                                          sizeof(struct rds_connection),
@@ -539,6 +545,18 @@ void rds_conn_drop(struct rds_connection *conn)
 }
 EXPORT_SYMBOL_GPL(rds_conn_drop);
 
+/*
+ * If the connection is down, trigger a connect. We may have scheduled a
+ * delayed reconnect however - in this case we should not interfere.
+ */
+void rds_conn_connect_if_down(struct rds_connection *conn)
+{
+       if (rds_conn_state(conn) == RDS_CONN_DOWN &&
+           !test_and_set_bit(RDS_RECONNECT_PENDING, &conn->c_flags))
+               queue_delayed_work(rds_wq, &conn->c_conn_w, 0);
+}
+EXPORT_SYMBOL_GPL(rds_conn_connect_if_down);
+
 /*
  * An error occurred on the connection
  */