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>
34 #include <linux/list.h>
35 #include <linux/slab.h>
36 #include <linux/export.h>
37 #include <net/inet_hashtables.h>
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)
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;
52 static struct hlist_head *rds_conn_bucket(__be32 laddr, __be32 faddr)
54 static u32 rds_hash_secret __read_mostly;
58 net_get_random_once(&rds_hash_secret, sizeof(rds_hash_secret));
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,
64 return &rds_conn_hash[hash & RDS_CONNECTION_HASH_MASK];
67 #define rds_conn_info_set(var, test, suffix) do { \
69 var |= RDS_INFO_CONNECTION_FLAG_##suffix; \
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)
78 struct rds_connection *conn, *ret = NULL;
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)) {
87 rdsdebug("returning conn %p for %pI4 -> %pI4\n", ret,
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.
98 static void rds_conn_path_reset(struct rds_conn_path *cp)
100 struct rds_connection *conn = cp->cp_conn;
102 rdsdebug("connection %pI4 to %pI4 reset\n",
103 &conn->c_laddr, &conn->c_faddr);
105 rds_stats_inc(s_conn_reset);
106 rds_send_path_reset(cp);
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. */
115 static void __rds_conn_path_init(struct rds_connection *conn,
116 struct rds_conn_path *cp, bool is_outgoing)
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);
125 atomic_set(&cp->cp_state, RDS_CONN_DOWN);
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);
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.
141 * For now they are not garbage collected once they're created. They
142 * are torn down as the module is removed, if ever.
144 static struct rds_connection *__rds_conn_create(struct net *net,
145 __be32 laddr, __be32 faddr,
146 struct rds_transport *trans, gfp_t gfp,
149 struct rds_connection *conn, *parent = NULL;
150 struct hlist_head *head = rds_conn_bucket(laddr, faddr);
151 struct rds_transport *loop_trans;
154 int npaths = (trans->t_mp_capable ? RDS_MPATH_WORKERS : 1);
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. */
165 conn = parent->c_passive;
171 conn = kmem_cache_zalloc(rds_conn_slab, gfp);
173 conn = ERR_PTR(-ENOMEM);
176 conn->c_path = kcalloc(npaths, sizeof(struct rds_conn_path), gfp);
178 kmem_cache_free(rds_conn_slab, conn);
179 conn = ERR_PTR(-ENOMEM);
183 INIT_HLIST_NODE(&conn->c_hash_node);
184 conn->c_laddr = laddr;
185 conn->c_faddr = faddr;
187 rds_conn_net_set(conn, net);
189 ret = rds_cong_get_maps(conn);
192 kmem_cache_free(rds_conn_slab, conn);
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.
202 loop_trans = rds_trans_get_preferred(net, faddr);
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.
211 trans = &rds_loop_transport;
215 conn->c_trans = trans;
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],
221 conn->c_path[i].cp_index = i;
223 ret = trans->conn_alloc(conn, gfp);
226 kmem_cache_free(rds_conn_slab, conn);
231 rdsdebug("allocated conn %p for %pI4 -> %pI4 over %s %s\n",
232 conn, &laddr, &faddr,
233 trans->t_name ? trans->t_name : "[unknown]",
234 is_outgoing ? "(outgoing)" : "");
237 * Since we ran without holding the conn lock, someone could
238 * have created the same conn (either normal or passive) in the
239 * interim. We check while holding the lock. If we won, we complete
240 * init and return our conn. If we lost, we rollback and return the
243 spin_lock_irqsave(&rds_conn_lock, flags);
245 /* Creating passive conn */
246 if (parent->c_passive) {
247 trans->conn_free(conn->c_path[0].cp_transport_data);
249 kmem_cache_free(rds_conn_slab, conn);
250 conn = parent->c_passive;
252 parent->c_passive = conn;
253 rds_cong_add_conn(conn);
257 /* Creating normal conn */
258 struct rds_connection *found;
260 found = rds_conn_lookup(net, head, laddr, faddr, trans);
262 struct rds_conn_path *cp;
265 for (i = 0; i < npaths; i++) {
266 cp = &conn->c_path[i];
267 /* The ->conn_alloc invocation may have
268 * allocated resource for all paths, so all
269 * of them may have to be freed here.
271 if (cp->cp_transport_data)
272 trans->conn_free(cp->cp_transport_data);
275 kmem_cache_free(rds_conn_slab, conn);
278 conn->c_my_gen_num = rds_gen_num;
279 conn->c_peer_gen_num = 0;
280 hlist_add_head_rcu(&conn->c_hash_node, head);
281 rds_cong_add_conn(conn);
285 spin_unlock_irqrestore(&rds_conn_lock, flags);
291 struct rds_connection *rds_conn_create(struct net *net,
292 __be32 laddr, __be32 faddr,
293 struct rds_transport *trans, gfp_t gfp)
295 return __rds_conn_create(net, laddr, faddr, trans, gfp, 0);
297 EXPORT_SYMBOL_GPL(rds_conn_create);
299 struct rds_connection *rds_conn_create_outgoing(struct net *net,
300 __be32 laddr, __be32 faddr,
301 struct rds_transport *trans, gfp_t gfp)
303 return __rds_conn_create(net, laddr, faddr, trans, gfp, 1);
305 EXPORT_SYMBOL_GPL(rds_conn_create_outgoing);
307 void rds_conn_shutdown(struct rds_conn_path *cp)
309 struct rds_connection *conn = cp->cp_conn;
311 /* shut it down unless it's down already */
312 if (!rds_conn_path_transition(cp, RDS_CONN_DOWN, RDS_CONN_DOWN)) {
314 * Quiesce the connection mgmt handlers before we start tearing
315 * things down. We don't hold the mutex for the entire
316 * duration of the shutdown operation, else we may be
317 * deadlocking with the CM handler. Instead, the CM event
318 * handler is supposed to check for state DISCONNECTING
320 mutex_lock(&cp->cp_cm_lock);
321 if (!rds_conn_path_transition(cp, RDS_CONN_UP,
322 RDS_CONN_DISCONNECTING) &&
323 !rds_conn_path_transition(cp, RDS_CONN_ERROR,
324 RDS_CONN_DISCONNECTING)) {
325 rds_conn_path_error(cp,
326 "shutdown called in state %d\n",
327 atomic_read(&cp->cp_state));
328 mutex_unlock(&cp->cp_cm_lock);
331 mutex_unlock(&cp->cp_cm_lock);
333 wait_event(cp->cp_waitq,
334 !test_bit(RDS_IN_XMIT, &cp->cp_flags));
335 wait_event(cp->cp_waitq,
336 !test_bit(RDS_RECV_REFILL, &cp->cp_flags));
338 conn->c_trans->conn_path_shutdown(cp);
339 rds_conn_path_reset(cp);
341 if (!rds_conn_path_transition(cp, RDS_CONN_DISCONNECTING,
343 !rds_conn_path_transition(cp, RDS_CONN_ERROR,
345 /* This can happen - eg when we're in the middle of tearing
346 * down the connection, and someone unloads the rds module.
347 * Quite reproducible with loopback connections.
350 * Note that this also happens with rds-tcp because
351 * we could have triggered rds_conn_path_drop in irq
352 * mode from rds_tcp_state change on the receipt of
353 * a FIN, thus we need to recheck for RDS_CONN_ERROR
356 rds_conn_path_error(cp, "%s: failed to transition "
357 "to state DOWN, current state "
359 atomic_read(&cp->cp_state));
364 /* Then reconnect if it's still live.
365 * The passive side of an IB loopback connection is never added
366 * to the conn hash, so we never trigger a reconnect on this
367 * conn - the reconnect is always triggered by the active peer. */
368 cancel_delayed_work_sync(&cp->cp_conn_w);
369 if (conn->c_destroy_in_prog)
372 if (!hlist_unhashed(&conn->c_hash_node)) {
374 rds_queue_reconnect(cp);
380 /* destroy a single rds_conn_path. rds_conn_destroy() iterates over
381 * all paths using rds_conn_path_destroy()
383 static void rds_conn_path_destroy(struct rds_conn_path *cp)
385 struct rds_message *rm, *rtmp;
387 if (!cp->cp_transport_data)
390 /* make sure lingering queued work won't try to ref the conn */
391 cancel_delayed_work_sync(&cp->cp_send_w);
392 cancel_delayed_work_sync(&cp->cp_recv_w);
394 rds_conn_path_drop(cp, true);
395 flush_work(&cp->cp_down_w);
397 /* tear down queued messages */
398 list_for_each_entry_safe(rm, rtmp,
401 list_del_init(&rm->m_conn_item);
402 BUG_ON(!list_empty(&rm->m_sock_item));
406 rds_message_put(cp->cp_xmit_rm);
408 cp->cp_conn->c_trans->conn_free(cp->cp_transport_data);
412 * Stop and free a connection.
414 * This can only be used in very limited circumstances. It assumes that once
415 * the conn has been shutdown that no one else is referencing the connection.
416 * We can only ensure this in the rmmod path in the current code.
418 void rds_conn_destroy(struct rds_connection *conn)
422 struct rds_conn_path *cp;
423 int npaths = (conn->c_trans->t_mp_capable ? RDS_MPATH_WORKERS : 1);
425 rdsdebug("freeing conn %p for %pI4 -> "
426 "%pI4\n", conn, &conn->c_laddr,
429 conn->c_destroy_in_prog = 1;
430 /* Ensure conn will not be scheduled for reconnect */
431 spin_lock_irq(&rds_conn_lock);
432 hlist_del_init_rcu(&conn->c_hash_node);
433 spin_unlock_irq(&rds_conn_lock);
436 /* shut the connection down */
437 for (i = 0; i < npaths; i++) {
438 cp = &conn->c_path[i];
439 rds_conn_path_destroy(cp);
440 BUG_ON(!list_empty(&cp->cp_retrans));
444 * The congestion maps aren't freed up here. They're
445 * freed by rds_cong_exit() after all the connections
448 rds_cong_remove_conn(conn);
451 kmem_cache_free(rds_conn_slab, conn);
453 spin_lock_irqsave(&rds_conn_lock, flags);
455 spin_unlock_irqrestore(&rds_conn_lock, flags);
457 EXPORT_SYMBOL_GPL(rds_conn_destroy);
459 static void rds_conn_message_info(struct socket *sock, unsigned int len,
460 struct rds_info_iterator *iter,
461 struct rds_info_lengths *lens,
464 struct hlist_head *head;
465 struct list_head *list;
466 struct rds_connection *conn;
467 struct rds_message *rm;
468 unsigned int total = 0;
473 len /= sizeof(struct rds_info_message);
477 for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash);
479 hlist_for_each_entry_rcu(conn, head, c_hash_node) {
480 struct rds_conn_path *cp;
483 npaths = (conn->c_trans->t_mp_capable ?
484 RDS_MPATH_WORKERS : 1);
486 for (j = 0; j < npaths; j++) {
487 cp = &conn->c_path[j];
489 list = &cp->cp_send_queue;
491 list = &cp->cp_retrans;
493 spin_lock_irqsave(&cp->cp_lock, flags);
495 /* XXX too lazy to maintain counts.. */
496 list_for_each_entry(rm, list, m_conn_item) {
499 rds_inc_info_copy(&rm->m_inc,
506 spin_unlock_irqrestore(&cp->cp_lock, flags);
513 lens->each = sizeof(struct rds_info_message);
516 static void rds_conn_message_info_send(struct socket *sock, unsigned int len,
517 struct rds_info_iterator *iter,
518 struct rds_info_lengths *lens)
520 rds_conn_message_info(sock, len, iter, lens, 1);
523 static void rds_conn_message_info_retrans(struct socket *sock,
525 struct rds_info_iterator *iter,
526 struct rds_info_lengths *lens)
528 rds_conn_message_info(sock, len, iter, lens, 0);
531 void rds_for_each_conn_info(struct socket *sock, unsigned int len,
532 struct rds_info_iterator *iter,
533 struct rds_info_lengths *lens,
534 int (*visitor)(struct rds_connection *, void *),
537 uint64_t buffer[(item_len + 7) / 8];
538 struct hlist_head *head;
539 struct rds_connection *conn;
545 lens->each = item_len;
547 for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash);
549 hlist_for_each_entry_rcu(conn, head, c_hash_node) {
551 /* XXX no c_lock usage.. */
552 if (!visitor(conn, buffer))
555 /* We copy as much as we can fit in the buffer,
556 * but we count all items so that the caller
557 * can resize the buffer. */
558 if (len >= item_len) {
559 rds_info_copy(iter, buffer, item_len);
567 EXPORT_SYMBOL_GPL(rds_for_each_conn_info);
569 static void rds_walk_conn_path_info(struct socket *sock, unsigned int len,
570 struct rds_info_iterator *iter,
571 struct rds_info_lengths *lens,
572 int (*visitor)(struct rds_conn_path *, void *),
575 u64 buffer[(item_len + 7) / 8];
576 struct hlist_head *head;
577 struct rds_connection *conn;
584 lens->each = item_len;
586 for (i = 0, head = rds_conn_hash; i < ARRAY_SIZE(rds_conn_hash);
588 hlist_for_each_entry_rcu(conn, head, c_hash_node) {
589 struct rds_conn_path *cp;
592 npaths = (conn->c_trans->t_mp_capable ?
593 RDS_MPATH_WORKERS : 1);
594 for (j = 0; j < npaths; j++) {
595 cp = &conn->c_path[j];
597 /* XXX no cp_lock usage.. */
598 if (!visitor(cp, buffer))
602 /* We copy as much as we can fit in the buffer,
603 * but we count all items so that the caller
604 * can resize the buffer.
606 if (len >= item_len) {
607 rds_info_copy(iter, buffer, item_len);
616 static int rds_conn_info_visitor(struct rds_conn_path *cp, void *buffer)
618 struct rds_info_connection *cinfo = buffer;
620 cinfo->next_tx_seq = cp->cp_next_tx_seq;
621 cinfo->next_rx_seq = cp->cp_next_rx_seq;
622 cinfo->laddr = cp->cp_conn->c_laddr;
623 cinfo->faddr = cp->cp_conn->c_faddr;
624 strncpy(cinfo->transport, cp->cp_conn->c_trans->t_name,
625 sizeof(cinfo->transport));
628 rds_conn_info_set(cinfo->flags, test_bit(RDS_IN_XMIT, &cp->cp_flags),
630 /* XXX Future: return the state rather than these funky bits */
631 rds_conn_info_set(cinfo->flags,
632 atomic_read(&cp->cp_state) == RDS_CONN_CONNECTING,
634 rds_conn_info_set(cinfo->flags,
635 atomic_read(&cp->cp_state) == RDS_CONN_UP,
640 static void rds_conn_info(struct socket *sock, unsigned int len,
641 struct rds_info_iterator *iter,
642 struct rds_info_lengths *lens)
644 rds_walk_conn_path_info(sock, len, iter, lens,
645 rds_conn_info_visitor,
646 sizeof(struct rds_info_connection));
649 int rds_conn_init(void)
651 rds_conn_slab = kmem_cache_create("rds_connection",
652 sizeof(struct rds_connection),
657 rds_info_register_func(RDS_INFO_CONNECTIONS, rds_conn_info);
658 rds_info_register_func(RDS_INFO_SEND_MESSAGES,
659 rds_conn_message_info_send);
660 rds_info_register_func(RDS_INFO_RETRANS_MESSAGES,
661 rds_conn_message_info_retrans);
666 void rds_conn_exit(void)
670 WARN_ON(!hlist_empty(rds_conn_hash));
672 kmem_cache_destroy(rds_conn_slab);
674 rds_info_deregister_func(RDS_INFO_CONNECTIONS, rds_conn_info);
675 rds_info_deregister_func(RDS_INFO_SEND_MESSAGES,
676 rds_conn_message_info_send);
677 rds_info_deregister_func(RDS_INFO_RETRANS_MESSAGES,
678 rds_conn_message_info_retrans);
684 void rds_conn_path_drop(struct rds_conn_path *cp, bool destroy)
686 atomic_set(&cp->cp_state, RDS_CONN_ERROR);
688 if (!destroy && cp->cp_conn->c_destroy_in_prog)
691 queue_work(rds_wq, &cp->cp_down_w);
693 EXPORT_SYMBOL_GPL(rds_conn_path_drop);
695 void rds_conn_drop(struct rds_connection *conn)
697 WARN_ON(conn->c_trans->t_mp_capable);
698 rds_conn_path_drop(&conn->c_path[0], false);
700 EXPORT_SYMBOL_GPL(rds_conn_drop);
703 * If the connection is down, trigger a connect. We may have scheduled a
704 * delayed reconnect however - in this case we should not interfere.
706 void rds_conn_path_connect_if_down(struct rds_conn_path *cp)
708 if (rds_conn_path_state(cp) == RDS_CONN_DOWN &&
709 !test_and_set_bit(RDS_RECONNECT_PENDING, &cp->cp_flags))
710 queue_delayed_work(rds_wq, &cp->cp_conn_w, 0);
712 EXPORT_SYMBOL_GPL(rds_conn_path_connect_if_down);
714 void rds_conn_connect_if_down(struct rds_connection *conn)
716 WARN_ON(conn->c_trans->t_mp_capable);
717 rds_conn_path_connect_if_down(&conn->c_path[0]);
719 EXPORT_SYMBOL_GPL(rds_conn_connect_if_down);
722 __rds_conn_path_error(struct rds_conn_path *cp, const char *fmt, ...)
730 rds_conn_path_drop(cp, false);