[TIPC]: Multicast link failure now resets all links to "nacking" node.
[linux-2.6.git] / net / tipc / link.c
index 511872a..ba7d3f1 100644 (file)
@@ -157,13 +157,13 @@ static void link_print(struct link *l_ptr, struct print_buf *buf,
        } \
 } while (0)
 
-static inline void dbg_print_link(struct link *l_ptr, const char *str)
+static void dbg_print_link(struct link *l_ptr, const char *str)
 {
        if (DBG_OUTPUT)
                link_print(l_ptr, DBG_OUTPUT, str);
 }
 
-static inline void dbg_print_buf_chain(struct sk_buff *root_buf)
+static void dbg_print_buf_chain(struct sk_buff *root_buf)
 {
        if (DBG_OUTPUT) {
                struct sk_buff *buf = root_buf;
@@ -176,50 +176,50 @@ static inline void dbg_print_buf_chain(struct sk_buff *root_buf)
 }
 
 /*
- *  Simple inlined link routines
+ *  Simple link routines
  */
 
-static inline unsigned int align(unsigned int i)
+static unsigned int align(unsigned int i)
 {
        return (i + 3) & ~3u;
 }
 
-static inline int link_working_working(struct link *l_ptr)
+static int link_working_working(struct link *l_ptr)
 {
        return (l_ptr->state == WORKING_WORKING);
 }
 
-static inline int link_working_unknown(struct link *l_ptr)
+static int link_working_unknown(struct link *l_ptr)
 {
        return (l_ptr->state == WORKING_UNKNOWN);
 }
 
-static inline int link_reset_unknown(struct link *l_ptr)
+static int link_reset_unknown(struct link *l_ptr)
 {
        return (l_ptr->state == RESET_UNKNOWN);
 }
 
-static inline int link_reset_reset(struct link *l_ptr)
+static int link_reset_reset(struct link *l_ptr)
 {
        return (l_ptr->state == RESET_RESET);
 }
 
-static inline int link_blocked(struct link *l_ptr)
+static int link_blocked(struct link *l_ptr)
 {
        return (l_ptr->exp_msg_count || l_ptr->blocked);
 }
 
-static inline int link_congested(struct link *l_ptr)
+static int link_congested(struct link *l_ptr)
 {
        return (l_ptr->out_queue_size >= l_ptr->queue_limit[0]);
 }
 
-static inline u32 link_max_pkt(struct link *l_ptr)
+static u32 link_max_pkt(struct link *l_ptr)
 {
        return l_ptr->max_pkt;
 }
 
-static inline void link_init_max_pkt(struct link *l_ptr)
+static void link_init_max_pkt(struct link *l_ptr)
 {
        u32 max_pkt;
        
@@ -236,20 +236,20 @@ static inline void link_init_max_pkt(struct link *l_ptr)
         l_ptr->max_pkt_probes = 0;
 }
 
-static inline u32 link_next_sent(struct link *l_ptr)
+static u32 link_next_sent(struct link *l_ptr)
 {
        if (l_ptr->next_out)
                return msg_seqno(buf_msg(l_ptr->next_out));
        return mod(l_ptr->next_out_no);
 }
 
-static inline u32 link_last_sent(struct link *l_ptr)
+static u32 link_last_sent(struct link *l_ptr)
 {
        return mod(link_next_sent(l_ptr) - 1);
 }
 
 /*
- *  Simple non-inlined link routines (i.e. referenced outside this file)
+ *  Simple non-static link routines (i.e. referenced outside this file)
  */
 
 int tipc_link_is_up(struct link *l_ptr)
@@ -396,7 +396,7 @@ static void link_timeout(struct link *l_ptr)
        tipc_node_unlock(l_ptr->owner);
 }
 
-static inline void link_set_timer(struct link *l_ptr, u32 time)
+static void link_set_timer(struct link *l_ptr, u32 time)
 {
        k_start_timer(&l_ptr->timer, time);
 }
@@ -573,7 +573,7 @@ void tipc_link_wakeup_ports(struct link *l_ptr, int all)
                if (win <= 0)
                        break;
                list_del_init(&p_ptr->wait_list);
-               p_ptr->congested_link = 0;
+               p_ptr->congested_link = NULL;
                assert(p_ptr->wakeup);
                spin_lock_bh(p_ptr->publ.lock);
                p_ptr->publ.congested = 0;
@@ -1004,9 +1004,9 @@ static int link_bundle_buf(struct link *l_ptr,
        return 1;
 }
 
-static inline void link_add_to_outqueue(struct link *l_ptr, 
-                                       struct sk_buff *buf, 
-                                       struct tipc_msg *msg)
+static void link_add_to_outqueue(struct link *l_ptr,
+                                struct sk_buff *buf,
+                                struct tipc_msg *msg)
 {
        u32 ack = mod(l_ptr->next_in_no - 1);
        u32 seqno = mod(l_ptr->next_out_no++);
@@ -1156,8 +1156,8 @@ int tipc_link_send(struct sk_buff *buf, u32 dest, u32 selector)
  * Link is locked. Returns user data length.
  */
 
-static inline int link_send_buf_fast(struct link *l_ptr, struct sk_buff *buf,
-                                    u32 *used_max_pkt)
+static int link_send_buf_fast(struct link *l_ptr, struct sk_buff *buf,
+                             u32 *used_max_pkt)
 {
        struct tipc_msg *msg = buf_msg(buf);
        int res = msg_data_sz(msg);
@@ -1355,7 +1355,7 @@ again:
        fragm_crs = 0;
        fragm_rest = 0;
        sect_rest = 0;
-       sect_crs = 0;
+       sect_crs = NULL;
        curr_sect = -1;
 
        /* Prepare reusable fragment header: */
@@ -1549,7 +1549,7 @@ u32 tipc_link_push_packet(struct link *l_ptr)
                        msg_dbg(buf_msg(buf), ">DEF-PROT>");
                        l_ptr->unacked_window = 0;
                        buf_discard(buf);
-                       l_ptr->proto_msg_queue = 0;
+                       l_ptr->proto_msg_queue = NULL;
                        return TIPC_OK;
                } else {
                        msg_dbg(buf_msg(buf), "|>DEF-PROT>");
@@ -1604,40 +1604,121 @@ void tipc_link_push_queue(struct link *l_ptr)
                tipc_bearer_schedule(l_ptr->b_ptr, l_ptr);
 }
 
+static void link_reset_all(unsigned long addr)
+{
+       struct node *n_ptr;
+       char addr_string[16];
+       u32 i;
+
+       read_lock_bh(&tipc_net_lock);
+       n_ptr = tipc_node_find((u32)addr);
+       if (!n_ptr) {
+               read_unlock_bh(&tipc_net_lock);
+               return; /* node no longer exists */
+       }
+
+       tipc_node_lock(n_ptr);
+
+       warn("Resetting all links to %s\n", 
+            addr_string_fill(addr_string, n_ptr->addr));
+
+       for (i = 0; i < MAX_BEARERS; i++) {
+               if (n_ptr->links[i]) {
+                       link_print(n_ptr->links[i], TIPC_OUTPUT, 
+                                  "Resetting link\n");
+                       tipc_link_reset(n_ptr->links[i]);
+               }
+       }
+
+       tipc_node_unlock(n_ptr);
+       read_unlock_bh(&tipc_net_lock);
+}
+
+static void link_retransmit_failure(struct link *l_ptr, struct sk_buff *buf)
+{
+       struct tipc_msg *msg = buf_msg(buf);
+
+       warn("Retransmission failure on link <%s>\n", l_ptr->name);
+       tipc_msg_print(TIPC_OUTPUT, msg, ">RETR-FAIL>");
+
+       if (l_ptr->addr) {
+
+               /* Handle failure on standard link */
+
+               link_print(l_ptr, TIPC_OUTPUT, "Resetting link\n");
+               tipc_link_reset(l_ptr);
+
+       } else {
+
+               /* Handle failure on broadcast link */
+
+               struct node *n_ptr;
+               char addr_string[16];
+
+               tipc_printf(TIPC_OUTPUT, "Msg seq number: %u,  ", msg_seqno(msg));
+               tipc_printf(TIPC_OUTPUT, "Outstanding acks: %u\n", (u32)TIPC_SKB_CB(buf)->handle);
+               
+               n_ptr = l_ptr->owner->next;
+               tipc_node_lock(n_ptr);
+
+               addr_string_fill(addr_string, n_ptr->addr);
+               tipc_printf(TIPC_OUTPUT, "Multicast link info for %s\n", addr_string);
+               tipc_printf(TIPC_OUTPUT, "Supported: %d,  ", n_ptr->bclink.supported);
+               tipc_printf(TIPC_OUTPUT, "Acked: %u\n", n_ptr->bclink.acked);
+               tipc_printf(TIPC_OUTPUT, "Last in: %u,  ", n_ptr->bclink.last_in);
+               tipc_printf(TIPC_OUTPUT, "Gap after: %u,  ", n_ptr->bclink.gap_after);
+               tipc_printf(TIPC_OUTPUT, "Gap to: %u\n", n_ptr->bclink.gap_to);
+               tipc_printf(TIPC_OUTPUT, "Nack sync: %u\n\n", n_ptr->bclink.nack_sync);
+
+               tipc_k_signal((Handler)link_reset_all, (unsigned long)n_ptr->addr);
+
+               tipc_node_unlock(n_ptr);
+
+               l_ptr->stale_count = 0;
+       }
+}
+
 void tipc_link_retransmit(struct link *l_ptr, struct sk_buff *buf, 
                          u32 retransmits)
 {
        struct tipc_msg *msg;
 
+       if (!buf)
+               return;
+
+       msg = buf_msg(buf);
+       
        dbg("Retransmitting %u in link %x\n", retransmits, l_ptr);
 
-       if (tipc_bearer_congested(l_ptr->b_ptr, l_ptr) && buf && !skb_cloned(buf)) {
-               msg_dbg(buf_msg(buf), ">NO_RETR->BCONG>");
-               dbg_print_link(l_ptr, "   ");
-               l_ptr->retransm_queue_head = msg_seqno(buf_msg(buf));
-               l_ptr->retransm_queue_size = retransmits;
-               return;
+       if (tipc_bearer_congested(l_ptr->b_ptr, l_ptr)) {
+               if (!skb_cloned(buf)) {
+                       msg_dbg(msg, ">NO_RETR->BCONG>");
+                       dbg_print_link(l_ptr, "   ");
+                       l_ptr->retransm_queue_head = msg_seqno(msg);
+                       l_ptr->retransm_queue_size = retransmits;
+                       return;
+               } else {
+                       /* Don't retransmit if driver already has the buffer */
+               }
+       } else {
+               /* Detect repeated retransmit failures on uncongested bearer */
+
+               if (l_ptr->last_retransmitted == msg_seqno(msg)) {
+                       if (++l_ptr->stale_count > 100) {
+                               link_retransmit_failure(l_ptr, buf);
+                               return;
+                       }
+               } else {
+                       l_ptr->last_retransmitted = msg_seqno(msg);
+                       l_ptr->stale_count = 1;
+               }
        }
+
        while (retransmits && (buf != l_ptr->next_out) && buf && !skb_cloned(buf)) {
                msg = buf_msg(buf);
                msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
                msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in); 
                if (tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr)) {
-                        /* Catch if retransmissions fail repeatedly: */
-                        if (l_ptr->last_retransmitted == msg_seqno(msg)) {
-                                if (++l_ptr->stale_count > 100) {
-                                        tipc_msg_print(TIPC_CONS, buf_msg(buf), ">RETR>");
-                                        info("...Retransmitted %u times\n",
-                                            l_ptr->stale_count);
-                                        link_print(l_ptr, TIPC_CONS, "Resetting Link\n");;
-                                        tipc_link_reset(l_ptr);
-                                        break;
-                                }
-                        } else {
-                                l_ptr->stale_count = 0;
-                        }
-                        l_ptr->last_retransmitted = msg_seqno(msg);
-
                        msg_dbg(buf_msg(buf), ">RETR>");
                        buf = buf->next;
                        retransmits--;
@@ -1650,6 +1731,7 @@ void tipc_link_retransmit(struct link *l_ptr, struct sk_buff *buf,
                        return;
                }
        }
+
        l_ptr->retransm_queue_head = l_ptr->retransm_queue_size = 0;
 }
 
@@ -1720,6 +1802,11 @@ void tipc_recv_msg(struct sk_buff *head, struct tipc_bearer *tb_ptr)
                        link_recv_non_seq(buf);
                        continue;
                }
+               
+               if (unlikely(!msg_short(msg) &&
+                            (msg_destnode(msg) != tipc_own_addr)))
+                       goto cont;
+               
                n_ptr = tipc_node_find(msg_prevnode(msg));
                if (unlikely(!n_ptr))
                        goto cont;
@@ -1860,7 +1947,7 @@ u32 tipc_link_defer_pkt(struct sk_buff **head,
                        struct sk_buff **tail,
                        struct sk_buff *buf)
 {
-       struct sk_buff *prev = 0;
+       struct sk_buff *prev = NULL;
        struct sk_buff *crs = *head;
        u32 seq_no = msg_seqno(buf_msg(buf));
 
@@ -1953,7 +2040,7 @@ static void link_handle_out_of_seq_msg(struct link *l_ptr,
 void tipc_link_send_proto_msg(struct link *l_ptr, u32 msg_typ, int probe_msg,
                              u32 gap, u32 tolerance, u32 priority, u32 ack_mtu)
 {
-       struct sk_buff *buf = 0;
+       struct sk_buff *buf = NULL;
        struct tipc_msg *msg = l_ptr->pmsg;
         u32 msg_size = sizeof(l_ptr->proto_msg);
 
@@ -2426,7 +2513,7 @@ static int link_recv_changeover_msg(struct link **l_ptr,
                }
        }
 exit:
-       *buf = 0;
+       *buf = NULL;
        buf_discard(tunnel_buf);
        return 0;
 }
@@ -2539,42 +2626,37 @@ exit:
  * pending message. This makes dynamic memory allocation unecessary.
  */
 
-static inline u32 get_long_msg_seqno(struct sk_buff *buf)
-{
-       return msg_seqno(buf_msg(buf));
-}
-
-static inline void set_long_msg_seqno(struct sk_buff *buf, u32 seqno)
+static void set_long_msg_seqno(struct sk_buff *buf, u32 seqno)
 {
        msg_set_seqno(buf_msg(buf), seqno);
 }
 
-static inline u32 get_fragm_size(struct sk_buff *buf)
+static u32 get_fragm_size(struct sk_buff *buf)
 {
        return msg_ack(buf_msg(buf));
 }
 
-static inline void set_fragm_size(struct sk_buff *buf, u32 sz)
+static void set_fragm_size(struct sk_buff *buf, u32 sz)
 {
        msg_set_ack(buf_msg(buf), sz);
 }
 
-static inline u32 get_expected_frags(struct sk_buff *buf)
+static u32 get_expected_frags(struct sk_buff *buf)
 {
        return msg_bcast_ack(buf_msg(buf));
 }
 
-static inline void set_expected_frags(struct sk_buff *buf, u32 exp)
+static void set_expected_frags(struct sk_buff *buf, u32 exp)
 {
        msg_set_bcast_ack(buf_msg(buf), exp);
 }
 
-static inline u32 get_timer_cnt(struct sk_buff *buf)
+static u32 get_timer_cnt(struct sk_buff *buf)
 {
        return msg_reroute_cnt(buf_msg(buf));
 }
 
-static inline void incr_timer_cnt(struct sk_buff *buf)
+static void incr_timer_cnt(struct sk_buff *buf)
 {
        msg_incr_reroute_cnt(buf_msg(buf));
 }
@@ -2586,13 +2668,13 @@ static inline void incr_timer_cnt(struct sk_buff *buf)
 int tipc_link_recv_fragment(struct sk_buff **pending, struct sk_buff **fb, 
                            struct tipc_msg **m)
 {
-       struct sk_buff *prev = 0;
+       struct sk_buff *prev = NULL;
        struct sk_buff *fbuf = *fb;
        struct tipc_msg *fragm = buf_msg(fbuf);
        struct sk_buff *pbuf = *pending;
        u32 long_msg_seq_no = msg_long_msgno(fragm);
 
-       *fb = 0;
+       *fb = NULL;
        msg_dbg(fragm,"FRG<REC<");
 
        /* Is there an incomplete message waiting for this fragment? */
@@ -2670,8 +2752,8 @@ int tipc_link_recv_fragment(struct sk_buff **pending, struct sk_buff **fb,
 
 static void link_check_defragm_bufs(struct link *l_ptr)
 {
-       struct sk_buff *prev = 0;
-       struct sk_buff *next = 0;
+       struct sk_buff *prev = NULL;
+       struct sk_buff *next = NULL;
        struct sk_buff *buf = l_ptr->defragm_buf;
 
        if (!buf)
@@ -2750,19 +2832,19 @@ static struct link *link_find_link(const char *name, struct node **node)
        struct link *l_ptr; 
 
        if (!link_name_validate(name, &link_name_parts))
-               return 0;
+               return NULL;
 
        b_ptr = tipc_bearer_find_interface(link_name_parts.if_local);
        if (!b_ptr)
-               return 0;
+               return NULL;
 
        *node = tipc_node_find(link_name_parts.addr_peer); 
        if (!*node)
-               return 0;
+               return NULL;
 
        l_ptr = (*node)->links[b_ptr->identity];
        if (!l_ptr || strcmp(l_ptr->name, name))
-               return 0;
+               return NULL;
 
        return l_ptr;
 }