Commit 9aa88c2a authored by Allan Stephens's avatar Allan Stephens Committed by Paul Gortmaker

tipc: Enhance sending of bulk name table messages

Modifies the initial transfer of name table entries to a new neighboring
node so that the messages are enqueued as a unit, rather than individually.

The revised algorithm now locates the link carrying the message only once,
and eliminates unnecessary checks for link congestion, message fragmentation,
and message bundling that are not required when sending these messages.
Signed-off-by: default avatarAllan Stephens <allan.stephens@windriver.com>
Signed-off-by: default avatarPaul Gortmaker <paul.gortmaker@windriver.com>
parent 1c553bb5
...@@ -985,6 +985,51 @@ int tipc_link_send(struct sk_buff *buf, u32 dest, u32 selector) ...@@ -985,6 +985,51 @@ int tipc_link_send(struct sk_buff *buf, u32 dest, u32 selector)
return res; return res;
} }
/*
* tipc_link_send_names - send name table entries to new neighbor
*
* Send routine for bulk delivery of name table messages when contact
* with a new neighbor occurs. No link congestion checking is performed
* because name table messages *must* be delivered. The messages must be
* small enough not to require fragmentation.
* Called without any locks held.
*/
void tipc_link_send_names(struct list_head *message_list, u32 dest)
{
struct tipc_node *n_ptr;
struct link *l_ptr;
struct sk_buff *buf;
struct sk_buff *temp_buf;
if (list_empty(message_list))
return;
read_lock_bh(&tipc_net_lock);
n_ptr = tipc_node_find(dest);
if (n_ptr) {
tipc_node_lock(n_ptr);
l_ptr = n_ptr->active_links[0];
if (l_ptr) {
/* convert circular list to linear list */
((struct sk_buff *)message_list->prev)->next = NULL;
link_add_chain_to_outqueue(l_ptr,
(struct sk_buff *)message_list->next, 0);
tipc_link_push_queue(l_ptr);
INIT_LIST_HEAD(message_list);
}
tipc_node_unlock(n_ptr);
}
read_unlock_bh(&tipc_net_lock);
/* discard the messages if they couldn't be sent */
list_for_each_safe(buf, temp_buf, ((struct sk_buff *)message_list)) {
list_del((struct list_head *)buf);
buf_discard(buf);
}
}
/* /*
* link_send_buf_fast: Entry for data messages where the * link_send_buf_fast: Entry for data messages where the
* destination link is known and the header is complete, * destination link is known and the header is complete,
......
...@@ -223,6 +223,7 @@ struct sk_buff *tipc_link_cmd_show_stats(const void *req_tlv_area, int req_tlv_s ...@@ -223,6 +223,7 @@ struct sk_buff *tipc_link_cmd_show_stats(const void *req_tlv_area, int req_tlv_s
struct sk_buff *tipc_link_cmd_reset_stats(const void *req_tlv_area, int req_tlv_space); struct sk_buff *tipc_link_cmd_reset_stats(const void *req_tlv_area, int req_tlv_space);
void tipc_link_reset(struct link *l_ptr); void tipc_link_reset(struct link *l_ptr);
int tipc_link_send(struct sk_buff *buf, u32 dest, u32 selector); int tipc_link_send(struct sk_buff *buf, u32 dest, u32 selector);
void tipc_link_send_names(struct list_head *message_list, u32 dest);
int tipc_link_send_buf(struct link *l_ptr, struct sk_buff *buf); int tipc_link_send_buf(struct link *l_ptr, struct sk_buff *buf);
u32 tipc_link_get_max_pkt(u32 dest, u32 selector); u32 tipc_link_get_max_pkt(u32 dest, u32 selector);
int tipc_link_send_sections_fast(struct tipc_port *sender, int tipc_link_send_sections_fast(struct tipc_port *sender,
......
...@@ -180,6 +180,7 @@ void tipc_named_node_up(unsigned long nodearg) ...@@ -180,6 +180,7 @@ void tipc_named_node_up(unsigned long nodearg)
struct publication *publ; struct publication *publ;
struct distr_item *item = NULL; struct distr_item *item = NULL;
struct sk_buff *buf = NULL; struct sk_buff *buf = NULL;
struct list_head message_list;
u32 node = (u32)nodearg; u32 node = (u32)nodearg;
u32 left = 0; u32 left = 0;
u32 rest; u32 rest;
...@@ -201,6 +202,10 @@ void tipc_named_node_up(unsigned long nodearg) ...@@ -201,6 +202,10 @@ void tipc_named_node_up(unsigned long nodearg)
if (!max_item_buf) if (!max_item_buf)
return; return;
/* create list of publication messages, then send them as a unit */
INIT_LIST_HEAD(&message_list);
read_lock_bh(&tipc_nametbl_lock); read_lock_bh(&tipc_nametbl_lock);
rest = publ_cnt * ITEM_SIZE; rest = publ_cnt * ITEM_SIZE;
...@@ -219,13 +224,14 @@ void tipc_named_node_up(unsigned long nodearg) ...@@ -219,13 +224,14 @@ void tipc_named_node_up(unsigned long nodearg)
item++; item++;
left -= ITEM_SIZE; left -= ITEM_SIZE;
if (!left) { if (!left) {
msg_set_link_selector(buf_msg(buf), node); list_add_tail((struct list_head *)buf, &message_list);
tipc_link_send(buf, node, node);
buf = NULL; buf = NULL;
} }
} }
exit: exit:
read_unlock_bh(&tipc_nametbl_lock); read_unlock_bh(&tipc_nametbl_lock);
tipc_link_send_names(&message_list, (u32)node);
} }
/** /**
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment