X-Git-Url: http://ftp.safe.ca/?a=blobdiff_plain;f=net%2Ftipc%2Fbcast.c;h=a008c6689305bd7b037a5226f1107a878c943060;hb=b2e75eff5e859d0c294e7405958362b26a423c6e;hp=a7b04f397c12c322adabc8f4b5be30efebff041c;hpb=4323add67792ced172d0d93b8b2e6187023115f1;p=safe%2Fjmp%2Flinux-2.6 diff --git a/net/tipc/bcast.c b/net/tipc/bcast.c index a7b04f3..a008c66 100644 --- a/net/tipc/bcast.c +++ b/net/tipc/bcast.c @@ -1,6 +1,6 @@ /* * net/tipc/bcast.c: TIPC broadcast code - * + * * Copyright (c) 2004-2006, Ericsson AB * Copyright (c) 2004, Intel Corporation. * Copyright (c) 2005, Wind River Systems @@ -49,19 +49,25 @@ #include "name_table.h" #include "bcast.h" - #define MAX_PKT_DEFAULT_MCAST 1500 /* bcast link max packet size (fixed) */ #define BCLINK_WIN_DEFAULT 20 /* bcast link window size (default) */ #define BCLINK_LOG_BUF_SIZE 0 +/* + * Loss rate for incoming broadcast frames; used to test retransmission code. + * Set to N to cause every N'th frame to be discarded; 0 => don't discard any. + */ + +#define TIPC_BCAST_LOSS_RATE 0 + /** * struct bcbearer_pair - a pair of bearers used by broadcast link * @primary: pointer to primary bearer * @secondary: pointer to secondary bearer - * - * Bearers must have same priority and same set of reachable destinations + * + * Bearers must have same priority and same set of reachable destinations * to be paired. */ @@ -75,7 +81,14 @@ struct bcbearer_pair { * @bearer: (non-standard) broadcast bearer structure * @media: (non-standard) broadcast media structure * @bpairs: array of bearer pairs - * @bpairs_temp: array of bearer pairs used during creation of "bpairs" + * @bpairs_temp: temporary array of bearer pairs used by tipc_bcbearer_sort() + * @remains: temporary node map used by tipc_bcbearer_send() + * @remains_new: temporary node map used tipc_bcbearer_send() + * + * Note: The fields labelled "temporary" are incorporated into the bearer + * to avoid consuming potentially limited stack space through the use of + * large local variables within multicast routines. Concurrent access is + * prevented through use of the spinlock "bc_lock". */ struct bcbearer { @@ -83,58 +96,60 @@ struct bcbearer { struct media media; struct bcbearer_pair bpairs[MAX_BEARERS]; struct bcbearer_pair bpairs_temp[TIPC_MAX_LINK_PRI + 1]; + struct tipc_node_map remains; + struct tipc_node_map remains_new; }; /** * struct bclink - link used for broadcast messages * @link: (non-standard) broadcast link structure * @node: (non-standard) node structure representing b'cast link's peer node - * + * * Handles sequence numbering, fragmentation, bundling, etc. */ struct bclink { struct link link; - struct node node; + struct tipc_node node; }; static struct bcbearer *bcbearer = NULL; static struct bclink *bclink = NULL; static struct link *bcl = NULL; -static spinlock_t bc_lock = SPIN_LOCK_UNLOCKED; +static DEFINE_SPINLOCK(bc_lock); -char tipc_bclink_name[] = "multicast-link"; +const char tipc_bclink_name[] = "broadcast-link"; -static inline u32 buf_seqno(struct sk_buff *buf) +static u32 buf_seqno(struct sk_buff *buf) { return msg_seqno(buf_msg(buf)); -} +} -static inline u32 bcbuf_acks(struct sk_buff *buf) +static u32 bcbuf_acks(struct sk_buff *buf) { return (u32)(unsigned long)TIPC_SKB_CB(buf)->handle; } -static inline void bcbuf_set_acks(struct sk_buff *buf, u32 acks) +static void bcbuf_set_acks(struct sk_buff *buf, u32 acks) { TIPC_SKB_CB(buf)->handle = (void *)(unsigned long)acks; } -static inline void bcbuf_decr_acks(struct sk_buff *buf) +static void bcbuf_decr_acks(struct sk_buff *buf) { bcbuf_set_acks(buf, bcbuf_acks(buf) - 1); } -/** +/** * bclink_set_gap - set gap according to contents of current deferred pkt queue - * + * * Called with 'node' locked, bc_lock unlocked */ -static inline void bclink_set_gap(struct node *n_ptr) +static void bclink_set_gap(struct tipc_node *n_ptr) { struct sk_buff *buf = n_ptr->bclink.deferred_head; @@ -144,53 +159,50 @@ static inline void bclink_set_gap(struct node *n_ptr) n_ptr->bclink.gap_to = mod(buf_seqno(buf) - 1); } -/** +/** * bclink_ack_allowed - test if ACK or NACK message can be sent at this moment - * + * * This mechanism endeavours to prevent all nodes in network from trying * to ACK or NACK at the same time. - * + * * Note: TIPC uses a different trigger to distribute ACKs than it does to - * distribute NACKs, but tries to use the same spacing (divide by 16). + * distribute NACKs, but tries to use the same spacing (divide by 16). */ -static inline int bclink_ack_allowed(u32 n) +static int bclink_ack_allowed(u32 n) { return((n % TIPC_MIN_LINK_WIN) == tipc_own_tag); } -/** +/** * bclink_retransmit_pkt - retransmit broadcast packets * @after: sequence number of last packet to *not* retransmit * @to: sequence number of last packet to retransmit - * - * Called with 'node' locked, bc_lock unlocked + * + * Called with bc_lock locked */ static void bclink_retransmit_pkt(u32 after, u32 to) { struct sk_buff *buf; - spin_lock_bh(&bc_lock); buf = bcl->first_out; while (buf && less_eq(buf_seqno(buf), after)) { - buf = buf->next; + buf = buf->next; } - if (buf != NULL) - tipc_link_retransmit(bcl, buf, mod(to - after)); - spin_unlock_bh(&bc_lock); + tipc_link_retransmit(bcl, buf, mod(to - after)); } -/** +/** * tipc_bclink_acknowledge - handle acknowledgement of broadcast packets * @n_ptr: node that sent acknowledgement info * @acked: broadcast sequence # that has been acknowledged - * + * * Node is locked, bc_lock unlocked. */ -void tipc_bclink_acknowledge(struct node *n_ptr, u32 acked) +void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked) { struct sk_buff *crs; struct sk_buff *next; @@ -232,13 +244,13 @@ void tipc_bclink_acknowledge(struct node *n_ptr, u32 acked) spin_unlock_bh(&bc_lock); } -/** +/** * bclink_send_ack - unicast an ACK msg - * + * * tipc_net_lock and node lock set */ -static void bclink_send_ack(struct node *n_ptr) +static void bclink_send_ack(struct tipc_node *n_ptr) { struct link *l_ptr = n_ptr->active_links[n_ptr->addr & 1]; @@ -246,13 +258,13 @@ static void bclink_send_ack(struct node *n_ptr) tipc_link_send_proto_msg(l_ptr, STATE_MSG, 0, 0, 0, 0, 0); } -/** +/** * bclink_send_nack- broadcast a NACK msg - * + * * tipc_net_lock and node lock set */ -static void bclink_send_nack(struct node *n_ptr) +static void bclink_send_nack(struct tipc_node *n_ptr) { struct sk_buff *buf; struct tipc_msg *msg; @@ -263,15 +275,15 @@ static void bclink_send_nack(struct node *n_ptr) buf = buf_acquire(INT_H_SIZE); if (buf) { msg = buf_msg(buf); - msg_init(msg, BCAST_PROTOCOL, STATE_MSG, - TIPC_OK, INT_H_SIZE, n_ptr->addr); + tipc_msg_init(msg, BCAST_PROTOCOL, STATE_MSG, + INT_H_SIZE, n_ptr->addr); msg_set_mc_netid(msg, tipc_net_id); - msg_set_bcast_ack(msg, mod(n_ptr->bclink.last_in)); + msg_set_bcast_ack(msg, mod(n_ptr->bclink.last_in)); msg_set_bcgap_after(msg, n_ptr->bclink.gap_after); msg_set_bcgap_to(msg, n_ptr->bclink.gap_to); msg_set_bcast_tag(msg, tipc_own_tag); - if (tipc_bearer_send(&bcbearer->bearer, buf, 0)) { + if (tipc_bearer_send(&bcbearer->bearer, buf, NULL)) { bcl->stats.sent_nacks++; buf_discard(buf); } else { @@ -280,23 +292,23 @@ static void bclink_send_nack(struct node *n_ptr) bcl->stats.bearer_congs++; } - /* + /* * Ensure we doesn't send another NACK msg to the node * until 16 more deferred messages arrive from it * (i.e. helps prevent all nodes from NACK'ing at same time) */ - + n_ptr->bclink.nack_sync = tipc_own_tag; } } -/** +/** * tipc_bclink_check_gap - send a NACK if a sequence gap exists * * tipc_net_lock and node lock set */ -void tipc_bclink_check_gap(struct node *n_ptr, u32 last_sent) +void tipc_bclink_check_gap(struct tipc_node *n_ptr, u32 last_sent) { if (!n_ptr->bclink.supported || less_eq(last_sent, mod(n_ptr->bclink.last_in))) @@ -308,15 +320,15 @@ void tipc_bclink_check_gap(struct node *n_ptr, u32 last_sent) bclink_send_nack(n_ptr); } -/** +/** * tipc_bclink_peek_nack - process a NACK msg meant for another node - * + * * Only tipc_net_lock set. */ -void tipc_bclink_peek_nack(u32 dest, u32 sender_tag, u32 gap_after, u32 gap_to) +static void tipc_bclink_peek_nack(u32 dest, u32 sender_tag, u32 gap_after, u32 gap_to) { - struct node *n_ptr = tipc_node_find(dest); + struct tipc_node *n_ptr = tipc_node_find(dest); u32 my_after, my_to; if (unlikely(!n_ptr || !tipc_node_is_up(n_ptr))) @@ -337,7 +349,7 @@ void tipc_bclink_peek_nack(u32 dest, u32 sender_tag, u32 gap_after, u32 gap_to) if (less_eq(my_to, gap_to)) n_ptr->bclink.gap_to = gap_after; } else { - /* + /* * Expand gap if missing bufs not in deferred queue: */ struct sk_buff *buf = n_ptr->bclink.deferred_head; @@ -346,8 +358,10 @@ void tipc_bclink_peek_nack(u32 dest, u32 sender_tag, u32 gap_after, u32 gap_to) for (; buf; buf = buf->next) { u32 seqno = buf_seqno(buf); - if (mod(seqno - prev) != 1) + if (mod(seqno - prev) != 1) { buf = NULL; + break; + } if (seqno == gap_after) break; prev = seqno; @@ -357,7 +371,7 @@ void tipc_bclink_peek_nack(u32 dest, u32 sender_tag, u32 gap_after, u32 gap_to) } /* * Some nodes may send a complementary NACK now: - */ + */ if (bclink_ack_allowed(sender_tag + 1)) { if (n_ptr->bclink.gap_to != n_ptr->bclink.gap_after) { bclink_send_nack(n_ptr); @@ -394,21 +408,24 @@ int tipc_bclink_send_msg(struct sk_buff *buf) /** * tipc_bclink_recv_pkt - receive a broadcast packet, and deliver upwards - * + * * tipc_net_lock is read_locked, no other locks set */ void tipc_bclink_recv_pkt(struct sk_buff *buf) -{ +{ +#if (TIPC_BCAST_LOSS_RATE) + static int rx_count = 0; +#endif struct tipc_msg *msg = buf_msg(buf); - struct node* node = tipc_node_find(msg_prevnode(msg)); + struct tipc_node* node = tipc_node_find(msg_prevnode(msg)); u32 next_in; u32 seqno; struct sk_buff *deferred; msg_dbg(msg, "bclink.supported || + if (unlikely(!node || !tipc_node_is_up(node) || !node->bclink.supported || (msg_mc_netid(msg) != tipc_net_id))) { buf_discard(buf); return; @@ -420,19 +437,31 @@ void tipc_bclink_recv_pkt(struct sk_buff *buf) tipc_node_lock(node); tipc_bclink_acknowledge(node, msg_bcast_ack(msg)); tipc_node_unlock(node); + spin_lock_bh(&bc_lock); bcl->stats.recv_nacks++; + bcl->owner->next = node; /* remember requestor */ bclink_retransmit_pkt(msg_bcgap_after(msg), msg_bcgap_to(msg)); + bcl->owner->next = NULL; + spin_unlock_bh(&bc_lock); } else { tipc_bclink_peek_nack(msg_destnode(msg), - msg_bcast_tag(msg), - msg_bcgap_after(msg), - msg_bcgap_to(msg)); + msg_bcast_tag(msg), + msg_bcgap_after(msg), + msg_bcgap_to(msg)); } buf_discard(buf); return; } +#if (TIPC_BCAST_LOSS_RATE) + if (++rx_count == TIPC_BCAST_LOSS_RATE) { + rx_count = 0; + buf_discard(buf); + return; + } +#endif + tipc_node_lock(node); receive: deferred = node->bclink.deferred_head; @@ -509,7 +538,7 @@ u32 tipc_bclink_get_last_sent(void) return last_sent; } -u32 tipc_bclink_acks_missing(struct node *n_ptr) +u32 tipc_bclink_acks_missing(struct tipc_node *n_ptr) { return (n_ptr->bclink.supported && (tipc_bclink_get_last_sent() != n_ptr->bclink.acked)); @@ -518,23 +547,18 @@ u32 tipc_bclink_acks_missing(struct node *n_ptr) /** * tipc_bcbearer_send - send a packet through the broadcast pseudo-bearer - * + * * Send through as many bearers as necessary to reach all nodes * that support TIPC multicasting. - * + * * Returns 0 if packet sent successfully, non-zero if not */ -int tipc_bcbearer_send(struct sk_buff *buf, - struct tipc_bearer *unused1, - struct tipc_media_addr *unused2) +static int tipc_bcbearer_send(struct sk_buff *buf, + struct tipc_bearer *unused1, + struct tipc_media_addr *unused2) { - static int send_count = 0; - - struct node_map remains; - struct node_map remains_new; int bp_index; - int swap_time; /* Prepare buffer for broadcasting (if first time trying to send it) */ @@ -544,18 +568,13 @@ int tipc_bcbearer_send(struct sk_buff *buf, assert(tipc_cltr_bcast_nodes.count != 0); bcbuf_set_acks(buf, tipc_cltr_bcast_nodes.count); msg = buf_msg(buf); - msg_set_non_seq(msg); + msg_set_non_seq(msg, 1); msg_set_mc_netid(msg, tipc_net_id); } - /* Determine if bearer pairs should be swapped following this attempt */ - - if ((swap_time = (++send_count >= 10))) - send_count = 0; - /* Send buffer over bearers until all targets reached */ - - remains = tipc_cltr_bcast_nodes; + + bcbearer->remains = tipc_cltr_bcast_nodes; for (bp_index = 0; bp_index < MAX_BEARERS; bp_index++) { struct bearer *p = bcbearer->bpairs[bp_index].primary; @@ -564,36 +583,37 @@ int tipc_bcbearer_send(struct sk_buff *buf, if (!p) break; /* no more bearers to try */ - tipc_nmap_diff(&remains, &p->nodes, &remains_new); - if (remains_new.count == remains.count) + tipc_nmap_diff(&bcbearer->remains, &p->nodes, &bcbearer->remains_new); + if (bcbearer->remains_new.count == bcbearer->remains.count) continue; /* bearer pair doesn't add anything */ - if (!p->publ.blocked && - !p->media->send_msg(buf, &p->publ, &p->media->bcast_addr)) { - if (swap_time && s && !s->publ.blocked) - goto swap; - else - goto update; + if (p->publ.blocked || + p->media->send_msg(buf, &p->publ, &p->media->bcast_addr)) { + /* unable to send on primary bearer */ + if (!s || s->publ.blocked || + s->media->send_msg(buf, &s->publ, + &s->media->bcast_addr)) { + /* unable to send on either bearer */ + continue; + } + } + + if (s) { + bcbearer->bpairs[bp_index].primary = s; + bcbearer->bpairs[bp_index].secondary = p; } - if (!s || s->publ.blocked || - s->media->send_msg(buf, &s->publ, &s->media->bcast_addr)) - continue; /* unable to send using bearer pair */ -swap: - bcbearer->bpairs[bp_index].primary = s; - bcbearer->bpairs[bp_index].secondary = p; -update: - if (remains_new.count == 0) - return TIPC_OK; - - remains = remains_new; + if (bcbearer->remains_new.count == 0) + return 0; + + bcbearer->remains = bcbearer->remains_new; } - + /* Unable to reach all targets */ bcbearer->bearer.publ.blocked = 1; bcl->stats.bearer_congs++; - return ~TIPC_OK; + return 1; } /** @@ -655,7 +675,7 @@ void tipc_bcbearer_sort(void) /** * tipc_bcbearer_push - resolve bearer congestion - * + * * Forces bclink to push out any unsent packets, until all packets are gone * or congestion reoccurs. * No locks set when function called @@ -687,27 +707,27 @@ int tipc_bclink_stats(char *buf, const u32 buf_size) spin_lock_bh(&bc_lock); tipc_printf(&pb, "Link <%s>\n" - " Window:%u packets\n", + " Window:%u packets\n", bcl->name, bcl->queue_limit[0]); - tipc_printf(&pb, " RX packets:%u fragments:%u/%u bundles:%u/%u\n", + tipc_printf(&pb, " RX packets:%u fragments:%u/%u bundles:%u/%u\n", bcl->stats.recv_info, bcl->stats.recv_fragments, bcl->stats.recv_fragmented, bcl->stats.recv_bundles, bcl->stats.recv_bundled); - tipc_printf(&pb, " TX packets:%u fragments:%u/%u bundles:%u/%u\n", + tipc_printf(&pb, " TX packets:%u fragments:%u/%u bundles:%u/%u\n", bcl->stats.sent_info, bcl->stats.sent_fragments, - bcl->stats.sent_fragmented, + bcl->stats.sent_fragmented, bcl->stats.sent_bundles, bcl->stats.sent_bundled); - tipc_printf(&pb, " RX naks:%u defs:%u dups:%u\n", + tipc_printf(&pb, " RX naks:%u defs:%u dups:%u\n", bcl->stats.recv_nacks, - bcl->stats.deferred_recv, + bcl->stats.deferred_recv, bcl->stats.duplicates); - tipc_printf(&pb, " TX naks:%u acks:%u dups:%u\n", - bcl->stats.sent_nacks, - bcl->stats.sent_acks, + tipc_printf(&pb, " TX naks:%u acks:%u dups:%u\n", + bcl->stats.sent_nacks, + bcl->stats.sent_acks, bcl->stats.retransmitted); tipc_printf(&pb, " Congestion bearer:%u link:%u Send queue max:%u avg:%u\n", bcl->stats.bearer_congs, @@ -729,7 +749,7 @@ int tipc_bclink_reset_stats(void) spin_lock_bh(&bc_lock); memset(&bcl->stats, 0, sizeof(bcl->stats)); spin_unlock_bh(&bc_lock); - return TIPC_OK; + return 0; } int tipc_bclink_set_queue_limits(u32 limit) @@ -742,16 +762,16 @@ int tipc_bclink_set_queue_limits(u32 limit) spin_lock_bh(&bc_lock); tipc_link_set_queue_limits(bcl, limit); spin_unlock_bh(&bc_lock); - return TIPC_OK; + return 0; } int tipc_bclink_init(void) { - bcbearer = kmalloc(sizeof(*bcbearer), GFP_ATOMIC); - bclink = kmalloc(sizeof(*bclink), GFP_ATOMIC); + bcbearer = kzalloc(sizeof(*bcbearer), GFP_ATOMIC); + bclink = kzalloc(sizeof(*bclink), GFP_ATOMIC); if (!bcbearer || !bclink) { nomem: - warn("Memory squeeze; Failed to create multicast link\n"); + warn("Multicast link creation failed, no memory\n"); kfree(bcbearer); bcbearer = NULL; kfree(bclink); @@ -759,23 +779,21 @@ int tipc_bclink_init(void) return -ENOMEM; } - memset(bcbearer, 0, sizeof(struct bcbearer)); INIT_LIST_HEAD(&bcbearer->bearer.cong_links); bcbearer->bearer.media = &bcbearer->media; bcbearer->media.send_msg = tipc_bcbearer_send; sprintf(bcbearer->media.name, "tipc-multicast"); bcl = &bclink->link; - memset(bclink, 0, sizeof(struct bclink)); INIT_LIST_HEAD(&bcl->waiting_ports); bcl->next_out_no = 1; - bclink->node.lock = SPIN_LOCK_UNLOCKED; + spin_lock_init(&bclink->node.lock); bcl->owner = &bclink->node; - bcl->max_pkt = MAX_PKT_DEFAULT_MCAST; + bcl->max_pkt = MAX_PKT_DEFAULT_MCAST; tipc_link_set_queue_limits(bcl, BCLINK_WIN_DEFAULT); bcl->b_ptr = &bcbearer->bearer; bcl->state = WORKING_WORKING; - sprintf(bcl->name, tipc_bclink_name); + strlcpy(bcl->name, tipc_bclink_name, TIPC_MAX_LINK_NAME); if (BCLINK_LOG_BUF_SIZE) { char *pb = kmalloc(BCLINK_LOG_BUF_SIZE, GFP_ATOMIC); @@ -785,7 +803,7 @@ int tipc_bclink_init(void) tipc_printbuf_init(&bcl->print_buf, pb, BCLINK_LOG_BUF_SIZE); } - return TIPC_OK; + return 0; } void tipc_bclink_stop(void) @@ -804,3 +822,113 @@ void tipc_bclink_stop(void) spin_unlock_bh(&bc_lock); } + +/** + * tipc_nmap_add - add a node to a node map + */ + +void tipc_nmap_add(struct tipc_node_map *nm_ptr, u32 node) +{ + int n = tipc_node(node); + int w = n / WSIZE; + u32 mask = (1 << (n % WSIZE)); + + if ((nm_ptr->map[w] & mask) == 0) { + nm_ptr->count++; + nm_ptr->map[w] |= mask; + } +} + +/** + * tipc_nmap_remove - remove a node from a node map + */ + +void tipc_nmap_remove(struct tipc_node_map *nm_ptr, u32 node) +{ + int n = tipc_node(node); + int w = n / WSIZE; + u32 mask = (1 << (n % WSIZE)); + + if ((nm_ptr->map[w] & mask) != 0) { + nm_ptr->map[w] &= ~mask; + nm_ptr->count--; + } +} + +/** + * tipc_nmap_diff - find differences between node maps + * @nm_a: input node map A + * @nm_b: input node map B + * @nm_diff: output node map A-B (i.e. nodes of A that are not in B) + */ + +void tipc_nmap_diff(struct tipc_node_map *nm_a, struct tipc_node_map *nm_b, + struct tipc_node_map *nm_diff) +{ + int stop = ARRAY_SIZE(nm_a->map); + int w; + int b; + u32 map; + + memset(nm_diff, 0, sizeof(*nm_diff)); + for (w = 0; w < stop; w++) { + map = nm_a->map[w] ^ (nm_a->map[w] & nm_b->map[w]); + nm_diff->map[w] = map; + if (map != 0) { + for (b = 0 ; b < WSIZE; b++) { + if (map & (1 << b)) + nm_diff->count++; + } + } + } +} + +/** + * tipc_port_list_add - add a port to a port list, ensuring no duplicates + */ + +void tipc_port_list_add(struct port_list *pl_ptr, u32 port) +{ + struct port_list *item = pl_ptr; + int i; + int item_sz = PLSIZE; + int cnt = pl_ptr->count; + + for (; ; cnt -= item_sz, item = item->next) { + if (cnt < PLSIZE) + item_sz = cnt; + for (i = 0; i < item_sz; i++) + if (item->ports[i] == port) + return; + if (i < PLSIZE) { + item->ports[i] = port; + pl_ptr->count++; + return; + } + if (!item->next) { + item->next = kmalloc(sizeof(*item), GFP_ATOMIC); + if (!item->next) { + warn("Incomplete multicast delivery, no memory\n"); + return; + } + item->next->next = NULL; + } + } +} + +/** + * tipc_port_list_free - free dynamically created entries in port_list chain + * + */ + +void tipc_port_list_free(struct port_list *pl_ptr) +{ + struct port_list *item; + struct port_list *next; + + for (item = pl_ptr->next; item; item = next) { + next = item->next; + kfree(item); + } +} +