From f2f9800d4955a96d92896841d8ba9b04201deaa1 Mon Sep 17 00:00:00 2001 From: Ying Xue Date: Fri, 9 Jan 2015 15:27:05 +0800 Subject: tipc: make tipc node table aware of net namespace Global variables associated with node table are below: - node table list (node_htable) - node hash table list (tipc_node_list) - node table lock (node_list_lock) - node number counter (tipc_num_nodes) - node link number counter (tipc_num_links) To make node table support namespace, above global variables must be moved to tipc_net structure in order to keep secret for different namespaces. As a consequence, these variables are allocated and initialized when namespace is created, and deallocated when namespace is destroyed. After the change, functions associated with these variables have to utilize a namespace pointer to access them. So adding namespace pointer as a parameter of these functions is the major change made in the commit. Signed-off-by: Ying Xue Tested-by: Tero Aho Reviewed-by: Jon Maloy Signed-off-by: David S. Miller --- net/tipc/node.h | 35 +++++++++++++++++++---------------- 1 file changed, 19 insertions(+), 16 deletions(-) (limited to 'net/tipc/node.h') diff --git a/net/tipc/node.h b/net/tipc/node.h index cbe0e950f1cc..43ef88ef3035 100644 --- a/net/tipc/node.h +++ b/net/tipc/node.h @@ -42,10 +42,10 @@ #include "bearer.h" #include "msg.h" -/* - * Out-of-range value for node signature - */ -#define INVALID_NODE_SIG 0x10000 +/* Out-of-range value for node signature */ +#define INVALID_NODE_SIG 0x10000 + +#define NODE_HTABLE_SIZE 512 /* Flags used to take different actions according to flag type * TIPC_WAIT_PEER_LINKS_DOWN: wait to see that peer's links are down @@ -90,6 +90,7 @@ struct tipc_node_bclink { * struct tipc_node - TIPC node structure * @addr: network address of node * @lock: spinlock governing access to structure + * @net: the applicable net namespace * @hash: links to adjacent nodes in unsorted hash chain * @active_links: pointers to active links to node * @links: pointers to all links to node @@ -106,6 +107,7 @@ struct tipc_node_bclink { struct tipc_node { u32 addr; spinlock_t lock; + struct net *net; struct hlist_node hash; struct tipc_link *active_links[2]; u32 act_mtus[2]; @@ -123,23 +125,24 @@ struct tipc_node { struct rcu_head rcu; }; -extern struct list_head tipc_node_list; - -struct tipc_node *tipc_node_find(u32 addr); -struct tipc_node *tipc_node_create(u32 addr); -void tipc_node_stop(void); +struct tipc_node *tipc_node_find(struct net *net, u32 addr); +struct tipc_node *tipc_node_create(struct net *net, u32 addr); +void tipc_node_stop(struct net *net); void tipc_node_attach_link(struct tipc_node *n_ptr, struct tipc_link *l_ptr); void tipc_node_detach_link(struct tipc_node *n_ptr, struct tipc_link *l_ptr); void tipc_node_link_down(struct tipc_node *n_ptr, struct tipc_link *l_ptr); void tipc_node_link_up(struct tipc_node *n_ptr, struct tipc_link *l_ptr); int tipc_node_active_links(struct tipc_node *n_ptr); int tipc_node_is_up(struct tipc_node *n_ptr); -struct sk_buff *tipc_node_get_links(const void *req_tlv_area, int req_tlv_space); -struct sk_buff *tipc_node_get_nodes(const void *req_tlv_area, int req_tlv_space); -int tipc_node_get_linkname(u32 bearer_id, u32 node, char *linkname, size_t len); +struct sk_buff *tipc_node_get_links(struct net *net, const void *req_tlv_area, + int req_tlv_space); +struct sk_buff *tipc_node_get_nodes(struct net *net, const void *req_tlv_area, + int req_tlv_space); +int tipc_node_get_linkname(struct net *net, u32 bearer_id, u32 node, + char *linkname, size_t len); void tipc_node_unlock(struct tipc_node *node); -int tipc_node_add_conn(u32 dnode, u32 port, u32 peer_port); -void tipc_node_remove_conn(u32 dnode, u32 port); +int tipc_node_add_conn(struct net *net, u32 dnode, u32 port, u32 peer_port); +void tipc_node_remove_conn(struct net *net, u32 dnode, u32 port); int tipc_nl_node_dump(struct sk_buff *skb, struct netlink_callback *cb); @@ -154,12 +157,12 @@ static inline bool tipc_node_blocked(struct tipc_node *node) TIPC_NOTIFY_NODE_DOWN | TIPC_WAIT_OWN_LINKS_DOWN)); } -static inline uint tipc_node_get_mtu(u32 addr, u32 selector) +static inline uint tipc_node_get_mtu(struct net *net, u32 addr, u32 selector) { struct tipc_node *node; u32 mtu; - node = tipc_node_find(addr); + node = tipc_node_find(net, addr); if (likely(node)) mtu = node->act_mtus[selector & 1]; -- cgit v1.2.3-70-g09d2 From c637c1035534867b85b78b453c38c495b58e2c5a Mon Sep 17 00:00:00 2001 From: Jon Paul Maloy Date: Thu, 5 Feb 2015 08:36:41 -0500 Subject: tipc: resolve race problem at unicast message reception TIPC handles message cardinality and sequencing at the link layer, before passing messages upwards to the destination sockets. During the upcall from link to socket no locks are held. It is therefore possible, and we see it happen occasionally, that messages arriving in different threads and delivered in sequence still bypass each other before they reach the destination socket. This must not happen, since it violates the sequentiality guarantee. We solve this by adding a new input buffer queue to the link structure. Arriving messages are added safely to the tail of that queue by the link, while the head of the queue is consumed, also safely, by the receiving socket. Sequentiality is secured per socket by only allowing buffers to be dequeued inside the socket lock. Since there may be multiple simultaneous readers of the queue, we use a 'filter' parameter to reduce the risk that they peek the same buffer from the queue, hence also reducing the risk of contention on the receiving socket locks. This solves the sequentiality problem, and seems to cause no measurable performance degradation. A nice side effect of this change is that lock handling in the functions tipc_rcv() and tipc_bcast_rcv() now becomes uniform, something that will enable future simplifications of those functions. Reviewed-by: Ying Xue Signed-off-by: Jon Maloy Signed-off-by: David S. Miller --- net/tipc/bcast.c | 20 ++-- net/tipc/link.c | 247 +++++++++++++++++++++----------------------------- net/tipc/link.h | 10 +- net/tipc/msg.c | 34 +++++++ net/tipc/msg.h | 73 +++++++++++++++ net/tipc/name_distr.c | 33 ++++--- net/tipc/name_distr.h | 2 +- net/tipc/node.c | 43 +++++---- net/tipc/node.h | 17 ++-- net/tipc/socket.c | 132 +++++++++++++++++---------- net/tipc/socket.h | 2 +- 11 files changed, 372 insertions(+), 241 deletions(-) (limited to 'net/tipc/node.h') diff --git a/net/tipc/bcast.c b/net/tipc/bcast.c index 3b886eb35c87..2dfaf272928a 100644 --- a/net/tipc/bcast.c +++ b/net/tipc/bcast.c @@ -189,10 +189,8 @@ static void bclink_retransmit_pkt(struct tipc_net *tn, u32 after, u32 to) void tipc_bclink_wakeup_users(struct net *net) { struct tipc_net *tn = net_generic(net, tipc_net_id); - struct sk_buff *skb; - while ((skb = skb_dequeue(&tn->bclink->link.waiting_sks))) - tipc_sk_rcv(net, skb); + tipc_sk_rcv(net, &tn->bclink->link.wakeupq); } /** @@ -271,9 +269,8 @@ void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked) tipc_link_push_packets(tn->bcl); bclink_set_last_sent(net); } - if (unlikely(released && !skb_queue_empty(&tn->bcl->waiting_sks))) + if (unlikely(released && !skb_queue_empty(&tn->bcl->wakeupq))) n_ptr->action_flags |= TIPC_WAKEUP_BCAST_USERS; - exit: tipc_bclink_unlock(net); } @@ -450,6 +447,9 @@ void tipc_bclink_rcv(struct net *net, struct sk_buff *buf) u32 next_in; u32 seqno; int deferred = 0; + int pos = 0; + struct sk_buff *iskb; + struct sk_buff_head msgs; /* Screen out unwanted broadcast messages */ if (msg_mc_netid(msg) != tn->net_id) @@ -506,7 +506,8 @@ receive: bcl->stats.recv_bundled += msg_msgcnt(msg); tipc_bclink_unlock(net); tipc_node_unlock(node); - tipc_link_bundle_rcv(net, buf); + while (tipc_msg_extract(buf, &iskb, &pos)) + tipc_sk_mcast_rcv(net, iskb); } else if (msg_user(msg) == MSG_FRAGMENTER) { tipc_buf_append(&node->bclink.reasm_buf, &buf); if (unlikely(!buf && !node->bclink.reasm_buf)) @@ -527,7 +528,9 @@ receive: bclink_accept_pkt(node, seqno); tipc_bclink_unlock(net); tipc_node_unlock(node); - tipc_named_rcv(net, buf); + skb_queue_head_init(&msgs); + skb_queue_tail(&msgs, buf); + tipc_named_rcv(net, &msgs); } else { tipc_bclink_lock(net); bclink_accept_pkt(node, seqno); @@ -944,10 +947,9 @@ int tipc_bclink_init(struct net *net) spin_lock_init(&bclink->lock); __skb_queue_head_init(&bcl->outqueue); __skb_queue_head_init(&bcl->deferred_queue); - skb_queue_head_init(&bcl->waiting_sks); + skb_queue_head_init(&bcl->wakeupq); bcl->next_out_no = 1; spin_lock_init(&bclink->node.lock); - __skb_queue_head_init(&bclink->node.waiting_sks); bcl->owner = &bclink->node; bcl->owner->net = net; bcl->max_pkt = MAX_PKT_DEFAULT_MCAST; diff --git a/net/tipc/link.c b/net/tipc/link.c index 41cb09aa41de..942491234099 100644 --- a/net/tipc/link.c +++ b/net/tipc/link.c @@ -113,10 +113,8 @@ static void link_reset_statistics(struct tipc_link *l_ptr); static void link_print(struct tipc_link *l_ptr, const char *str); static void tipc_link_sync_xmit(struct tipc_link *l); static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf); -static int tipc_link_input(struct net *net, struct tipc_link *l, - struct sk_buff *buf); -static int tipc_link_prepare_input(struct net *net, struct tipc_link *l, - struct sk_buff **buf); +static void tipc_link_input(struct tipc_link *l, struct sk_buff *skb); +static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb); /* * Simple link routines @@ -318,8 +316,9 @@ struct tipc_link *tipc_link_create(struct tipc_node *n_ptr, l_ptr->next_out_no = 1; __skb_queue_head_init(&l_ptr->outqueue); __skb_queue_head_init(&l_ptr->deferred_queue); - skb_queue_head_init(&l_ptr->waiting_sks); - + skb_queue_head_init(&l_ptr->wakeupq); + skb_queue_head_init(&l_ptr->inputq); + skb_queue_head_init(&l_ptr->namedq); link_reset_statistics(l_ptr); tipc_node_attach_link(n_ptr, l_ptr); setup_timer(&l_ptr->timer, link_timeout, (unsigned long)l_ptr); @@ -387,7 +386,7 @@ static bool link_schedule_user(struct tipc_link *link, u32 oport, return false; TIPC_SKB_CB(buf)->chain_sz = chain_sz; TIPC_SKB_CB(buf)->chain_imp = imp; - skb_queue_tail(&link->waiting_sks, buf); + skb_queue_tail(&link->wakeupq, buf); link->stats.link_congs++; return true; } @@ -398,17 +397,19 @@ static bool link_schedule_user(struct tipc_link *link, u32 oport, * Move a number of waiting users, as permitted by available space in * the send queue, from link wait queue to node wait queue for wakeup */ -static void link_prepare_wakeup(struct tipc_link *link) +void link_prepare_wakeup(struct tipc_link *link) { uint pend_qsz = skb_queue_len(&link->outqueue); struct sk_buff *skb, *tmp; - skb_queue_walk_safe(&link->waiting_sks, skb, tmp) { + skb_queue_walk_safe(&link->wakeupq, skb, tmp) { if (pend_qsz >= link->queue_limit[TIPC_SKB_CB(skb)->chain_imp]) break; pend_qsz += TIPC_SKB_CB(skb)->chain_sz; - skb_unlink(skb, &link->waiting_sks); - skb_queue_tail(&link->owner->waiting_sks, skb); + skb_unlink(skb, &link->wakeupq); + skb_queue_tail(&link->inputq, skb); + link->owner->inputq = &link->inputq; + link->owner->action_flags |= TIPC_MSG_EVT; } } @@ -461,13 +462,13 @@ void tipc_link_reset(struct tipc_link *l_ptr) l_ptr->exp_msg_count = START_CHANGEOVER; } - /* Clean up all queues: */ + /* Clean up all queues, except inputq: */ __skb_queue_purge(&l_ptr->outqueue); __skb_queue_purge(&l_ptr->deferred_queue); - if (!skb_queue_empty(&l_ptr->waiting_sks)) { - skb_queue_splice_init(&l_ptr->waiting_sks, &owner->waiting_sks); - owner->action_flags |= TIPC_WAKEUP_USERS; - } + skb_queue_splice_init(&l_ptr->wakeupq, &l_ptr->inputq); + if (!skb_queue_empty(&l_ptr->inputq)) + owner->action_flags |= TIPC_MSG_EVT; + owner->inputq = &l_ptr->inputq; l_ptr->next_out = NULL; l_ptr->unacked_window = 0; l_ptr->checkpoint = 1; @@ -795,7 +796,7 @@ int __tipc_link_xmit(struct net *net, struct tipc_link *link, static void skb2list(struct sk_buff *skb, struct sk_buff_head *list) { - __skb_queue_head_init(list); + skb_queue_head_init(list); __skb_queue_tail(list, skb); } @@ -841,19 +842,13 @@ int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dnode, rc = __tipc_link_xmit(net, link, list); tipc_node_unlock(node); } - if (link) return rc; - if (likely(in_own_node(net, dnode))) { - /* As a node local message chain never contains more than one - * buffer, we just need to dequeue one SKB buffer from the - * head list. - */ - return tipc_sk_rcv(net, __skb_dequeue(list)); - } - __skb_queue_purge(list); + if (likely(in_own_node(net, dnode))) + return tipc_sk_rcv(net, list); + __skb_queue_purge(list); return rc; } @@ -1162,7 +1157,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) /* Locate unicast link endpoint that should handle message */ l_ptr = n_ptr->links[b_ptr->identity]; if (unlikely(!l_ptr)) - goto unlock_discard; + goto unlock; /* Verify that communication with node is currently allowed */ if ((n_ptr->action_flags & TIPC_WAIT_PEER_LINKS_DOWN) && @@ -1173,7 +1168,7 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) n_ptr->action_flags &= ~TIPC_WAIT_PEER_LINKS_DOWN; if (tipc_node_blocked(n_ptr)) - goto unlock_discard; + goto unlock; /* Validate message sequence number info */ seq_no = msg_seqno(msg); @@ -1197,18 +1192,16 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) if (unlikely(l_ptr->next_out)) tipc_link_push_packets(l_ptr); - if (released && !skb_queue_empty(&l_ptr->waiting_sks)) { + if (released && !skb_queue_empty(&l_ptr->wakeupq)) link_prepare_wakeup(l_ptr); - l_ptr->owner->action_flags |= TIPC_WAKEUP_USERS; - } /* Process the incoming packet */ if (unlikely(!link_working_working(l_ptr))) { if (msg_user(msg) == LINK_PROTOCOL) { tipc_link_proto_rcv(l_ptr, skb); link_retrieve_defq(l_ptr, &head); - tipc_node_unlock(n_ptr); - continue; + skb = NULL; + goto unlock; } /* Traffic message. Conditionally activate link */ @@ -1217,18 +1210,18 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) if (link_working_working(l_ptr)) { /* Re-insert buffer in front of queue */ __skb_queue_head(&head, skb); - tipc_node_unlock(n_ptr); - continue; + skb = NULL; + goto unlock; } - goto unlock_discard; + goto unlock; } /* Link is now in state WORKING_WORKING */ if (unlikely(seq_no != mod(l_ptr->next_in_no))) { link_handle_out_of_seq_msg(l_ptr, skb); link_retrieve_defq(l_ptr, &head); - tipc_node_unlock(n_ptr); - continue; + skb = NULL; + goto unlock; } l_ptr->next_in_no++; if (unlikely(!skb_queue_empty(&l_ptr->deferred_queue))) @@ -1238,97 +1231,102 @@ void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr) l_ptr->stats.sent_acks++; tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0, 0); } - - if (tipc_link_prepare_input(net, l_ptr, &skb)) { - tipc_node_unlock(n_ptr); - continue; - } - tipc_node_unlock(n_ptr); - - if (tipc_link_input(net, l_ptr, skb) != 0) - goto discard; - continue; -unlock_discard: + tipc_link_input(l_ptr, skb); + skb = NULL; +unlock: tipc_node_unlock(n_ptr); discard: - kfree_skb(skb); + if (unlikely(skb)) + kfree_skb(skb); } } -/** - * tipc_link_prepare_input - process TIPC link messages - * - * returns nonzero if the message was consumed +/* tipc_data_input - deliver data and name distr msgs to upper layer * + * Consumes buffer if message is of right type * Node lock must be held */ -static int tipc_link_prepare_input(struct net *net, struct tipc_link *l, - struct sk_buff **buf) +static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb) { - struct tipc_node *n; - struct tipc_msg *msg; - int res = -EINVAL; + struct tipc_node *node = link->owner; + struct tipc_msg *msg = buf_msg(skb); + u32 dport = msg_destport(msg); - n = l->owner; - msg = buf_msg(*buf); switch (msg_user(msg)) { - case CHANGEOVER_PROTOCOL: - if (tipc_link_tunnel_rcv(n, buf)) - res = 0; - break; - case MSG_FRAGMENTER: - l->stats.recv_fragments++; - if (tipc_buf_append(&l->reasm_buf, buf)) { - l->stats.recv_fragmented++; - res = 0; - } else if (!l->reasm_buf) { - tipc_link_reset(l); + case TIPC_LOW_IMPORTANCE: + case TIPC_MEDIUM_IMPORTANCE: + case TIPC_HIGH_IMPORTANCE: + case TIPC_CRITICAL_IMPORTANCE: + case CONN_MANAGER: + if (tipc_skb_queue_tail(&link->inputq, skb, dport)) { + node->inputq = &link->inputq; + node->action_flags |= TIPC_MSG_EVT; } - break; - case MSG_BUNDLER: - l->stats.recv_bundles++; - l->stats.recv_bundled += msg_msgcnt(msg); - res = 0; - break; + return true; case NAME_DISTRIBUTOR: - n->bclink.recv_permitted = true; - res = 0; - break; + node->bclink.recv_permitted = true; + node->namedq = &link->namedq; + skb_queue_tail(&link->namedq, skb); + if (skb_queue_len(&link->namedq) == 1) + node->action_flags |= TIPC_NAMED_MSG_EVT; + return true; + case MSG_BUNDLER: + case CHANGEOVER_PROTOCOL: + case MSG_FRAGMENTER: case BCAST_PROTOCOL: - tipc_link_sync_rcv(n, *buf); - break; + return false; default: - res = 0; - } - return res; + pr_warn("Dropping received illegal msg type\n"); + kfree_skb(skb); + return false; + }; } -/** - * tipc_link_input - Deliver message too higher layers + +/* tipc_link_input - process packet that has passed link protocol check + * + * Consumes buffer + * Node lock must be held */ -static int tipc_link_input(struct net *net, struct tipc_link *l, - struct sk_buff *buf) +static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb) { - struct tipc_msg *msg = buf_msg(buf); - int res = 0; + struct tipc_node *node = link->owner; + struct tipc_msg *msg = buf_msg(skb); + struct sk_buff *iskb; + int pos = 0; + + if (likely(tipc_data_input(link, skb))) + return; switch (msg_user(msg)) { - case TIPC_LOW_IMPORTANCE: - case TIPC_MEDIUM_IMPORTANCE: - case TIPC_HIGH_IMPORTANCE: - case TIPC_CRITICAL_IMPORTANCE: - case CONN_MANAGER: - tipc_sk_rcv(net, buf); + case CHANGEOVER_PROTOCOL: + if (!tipc_link_tunnel_rcv(node, &skb)) + break; + if (msg_user(buf_msg(skb)) != MSG_BUNDLER) { + tipc_data_input(link, skb); + break; + } + case MSG_BUNDLER: + link->stats.recv_bundles++; + link->stats.recv_bundled += msg_msgcnt(msg); + + while (tipc_msg_extract(skb, &iskb, &pos)) + tipc_data_input(link, iskb); break; - case NAME_DISTRIBUTOR: - tipc_named_rcv(net, buf); + case MSG_FRAGMENTER: + link->stats.recv_fragments++; + if (tipc_buf_append(&link->reasm_buf, &skb)) { + link->stats.recv_fragmented++; + tipc_data_input(link, skb); + } else if (!link->reasm_buf) { + tipc_link_reset(link); + } break; - case MSG_BUNDLER: - tipc_link_bundle_rcv(net, buf); + case BCAST_PROTOCOL: + tipc_link_sync_rcv(node, skb); break; default: - res = -EINVAL; - } - return res; + break; + }; } /** @@ -1779,7 +1777,7 @@ void tipc_link_dup_queue_xmit(struct tipc_link *l_ptr, * @from_pos: offset to extract from * * Returns a new message buffer containing an embedded message. The - * encapsulating message itself is left unchanged. + * encapsulating buffer is left unchanged. */ static struct sk_buff *buf_extract(struct sk_buff *skb, u32 from_pos) { @@ -1793,8 +1791,6 @@ static struct sk_buff *buf_extract(struct sk_buff *skb, u32 from_pos) return eb; } - - /* tipc_link_dup_rcv(): Receive a tunnelled DUPLICATE_MSG packet. * Owner node is locked. */ @@ -1893,41 +1889,6 @@ exit: return *buf != NULL; } -/* - * Bundler functionality: - */ -void tipc_link_bundle_rcv(struct net *net, struct sk_buff *buf) -{ - u32 msgcount = msg_msgcnt(buf_msg(buf)); - u32 pos = INT_H_SIZE; - struct sk_buff *obuf; - struct tipc_msg *omsg; - - while (msgcount--) { - obuf = buf_extract(buf, pos); - if (obuf == NULL) { - pr_warn("Link unable to unbundle message(s)\n"); - break; - } - omsg = buf_msg(obuf); - pos += align(msg_size(omsg)); - if (msg_isdata(omsg)) { - if (unlikely(msg_type(omsg) == TIPC_MCAST_MSG)) - tipc_sk_mcast_rcv(net, obuf); - else - tipc_sk_rcv(net, obuf); - } else if (msg_user(omsg) == CONN_MANAGER) { - tipc_sk_rcv(net, obuf); - } else if (msg_user(omsg) == NAME_DISTRIBUTOR) { - tipc_named_rcv(net, obuf); - } else { - pr_warn("Illegal bundled msg: %u\n", msg_user(omsg)); - kfree_skb(obuf); - } - } - kfree_skb(buf); -} - static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol) { unsigned long intv = ((tol / 4) > 500) ? 500 : tol / 4; diff --git a/net/tipc/link.h b/net/tipc/link.h index 5b9a17f26280..34d3f55c4cea 100644 --- a/net/tipc/link.h +++ b/net/tipc/link.h @@ -131,8 +131,10 @@ struct tipc_stats { * @next_in_no: next sequence number to expect for inbound messages * @deferred_queue: deferred queue saved OOS b'cast message received from node * @unacked_window: # of inbound messages rx'd without ack'ing back to peer + * @inputq: buffer queue for messages to be delivered upwards + * @namedq: buffer queue for name table messages to be delivered upwards * @next_out: ptr to first unsent outbound message in queue - * @waiting_sks: linked list of sockets waiting for link congestion to abate + * @wakeupq: linked list of wakeup msgs waiting for link congestion to abate * @long_msg_seq_no: next identifier to use for outbound fragmented messages * @reasm_buf: head of partially reassembled inbound message fragments * @stats: collects statistics regarding link activity @@ -184,10 +186,12 @@ struct tipc_link { u32 next_in_no; struct sk_buff_head deferred_queue; u32 unacked_window; + struct sk_buff_head inputq; + struct sk_buff_head namedq; /* Congestion handling */ struct sk_buff *next_out; - struct sk_buff_head waiting_sks; + struct sk_buff_head wakeupq; /* Fragmentation/reassembly */ u32 long_msg_seq_no; @@ -228,7 +232,6 @@ int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dest, u32 selector); int __tipc_link_xmit(struct net *net, struct tipc_link *link, struct sk_buff_head *list); -void tipc_link_bundle_rcv(struct net *net, struct sk_buff *buf); void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int prob, u32 gap, u32 tolerance, u32 priority, u32 acked_mtu); void tipc_link_push_packets(struct tipc_link *l_ptr); @@ -244,6 +247,7 @@ int tipc_nl_link_get(struct sk_buff *skb, struct genl_info *info); int tipc_nl_link_set(struct sk_buff *skb, struct genl_info *info); int tipc_nl_link_reset_stats(struct sk_buff *skb, struct genl_info *info); int tipc_nl_parse_link_prop(struct nlattr *prop, struct nlattr *props[]); +void link_prepare_wakeup(struct tipc_link *l); /* * Link sequence number manipulation routines (uses modulo 2**16 arithmetic) diff --git a/net/tipc/msg.c b/net/tipc/msg.c index 697223a21240..b6eb90cd3ef7 100644 --- a/net/tipc/msg.c +++ b/net/tipc/msg.c @@ -326,6 +326,40 @@ bool tipc_msg_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu) return true; } +/** + * tipc_msg_extract(): extract bundled inner packet from buffer + * @skb: linear outer buffer, to be extracted from. + * @iskb: extracted inner buffer, to be returned + * @pos: position of msg to be extracted. Returns with pointer of next msg + * Consumes outer buffer when last packet extracted + * Returns true when when there is an extracted buffer, otherwise false + */ +bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos) +{ + struct tipc_msg *msg = buf_msg(skb); + int imsz; + struct tipc_msg *imsg = (struct tipc_msg *)(msg_data(msg) + *pos); + + /* Is there space left for shortest possible message? */ + if (*pos > (msg_data_sz(msg) - SHORT_H_SIZE)) + goto none; + imsz = msg_size(imsg); + + /* Is there space left for current message ? */ + if ((*pos + imsz) > msg_data_sz(msg)) + goto none; + *iskb = tipc_buf_acquire(imsz); + if (!*iskb) + goto none; + skb_copy_to_linear_data(*iskb, imsg, imsz); + *pos += align(imsz); + return true; +none: + kfree_skb(skb); + *iskb = NULL; + return false; +} + /** * tipc_msg_make_bundle(): Create bundle buf and append message to its tail * @list: the buffer chain diff --git a/net/tipc/msg.h b/net/tipc/msg.h index 60702992933d..ab467261bd9d 100644 --- a/net/tipc/msg.h +++ b/net/tipc/msg.h @@ -45,6 +45,7 @@ * Note: Some items are also used with TIPC internal message headers */ #define TIPC_VERSION 2 +struct plist; /* * Payload message users are defined in TIPC's public API: @@ -759,10 +760,82 @@ int tipc_buf_append(struct sk_buff **headbuf, struct sk_buff **buf); bool tipc_msg_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu); bool tipc_msg_make_bundle(struct sk_buff_head *list, struct sk_buff *skb, u32 mtu, u32 dnode); +bool tipc_msg_extract(struct sk_buff *skb, struct sk_buff **iskb, int *pos); int tipc_msg_build(struct tipc_msg *mhdr, struct msghdr *m, int offset, int dsz, int mtu, struct sk_buff_head *list); bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, u32 *dnode, int *err); struct sk_buff *tipc_msg_reassemble(struct sk_buff_head *list); +/* tipc_skb_peek_port(): find a destination port, ignoring all destinations + * up to and including 'filter'. + * Note: ignoring previously tried destinations minimizes the risk of + * contention on the socket lock + * @list: list to be peeked in + * @filter: last destination to be ignored from search + * Returns a destination port number, of applicable. + */ +static inline u32 tipc_skb_peek_port(struct sk_buff_head *list, u32 filter) +{ + struct sk_buff *skb; + u32 dport = 0; + bool ignore = true; + + spin_lock_bh(&list->lock); + skb_queue_walk(list, skb) { + dport = msg_destport(buf_msg(skb)); + if (!filter || skb_queue_is_last(list, skb)) + break; + if (dport == filter) + ignore = false; + else if (!ignore) + break; + } + spin_unlock_bh(&list->lock); + return dport; +} + +/* tipc_skb_dequeue(): unlink first buffer with dest 'dport' from list + * @list: list to be unlinked from + * @dport: selection criteria for buffer to unlink + */ +static inline struct sk_buff *tipc_skb_dequeue(struct sk_buff_head *list, + u32 dport) +{ + struct sk_buff *_skb, *tmp, *skb = NULL; + + spin_lock_bh(&list->lock); + skb_queue_walk_safe(list, _skb, tmp) { + if (msg_destport(buf_msg(_skb)) == dport) { + __skb_unlink(_skb, list); + skb = _skb; + break; + } + } + spin_unlock_bh(&list->lock); + return skb; +} + +/* tipc_skb_queue_tail(): add buffer to tail of list; + * @list: list to be appended to + * @skb: buffer to append. Always appended + * @dport: the destination port of the buffer + * returns true if dport differs from previous destination + */ +static inline bool tipc_skb_queue_tail(struct sk_buff_head *list, + struct sk_buff *skb, u32 dport) +{ + struct sk_buff *_skb = NULL; + bool rv = false; + + spin_lock_bh(&list->lock); + _skb = skb_peek_tail(list); + if (!_skb || (msg_destport(buf_msg(_skb)) != dport) || + (skb_queue_len(list) > 32)) + rv = true; + __skb_queue_tail(list, skb); + spin_unlock_bh(&list->lock); + return rv; +} + #endif diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c index dd8564cd9dbb..fcb07915aaac 100644 --- a/net/tipc/name_distr.c +++ b/net/tipc/name_distr.c @@ -381,25 +381,34 @@ void tipc_named_process_backlog(struct net *net) } /** - * tipc_named_rcv - process name table update message sent by another node + * tipc_named_rcv - process name table update messages sent by another node */ -void tipc_named_rcv(struct net *net, struct sk_buff *buf) +void tipc_named_rcv(struct net *net, struct sk_buff_head *inputq) { struct tipc_net *tn = net_generic(net, tipc_net_id); - struct tipc_msg *msg = buf_msg(buf); - struct distr_item *item = (struct distr_item *)msg_data(msg); - u32 count = msg_data_sz(msg) / ITEM_SIZE; - u32 node = msg_orignode(msg); + struct tipc_msg *msg; + struct distr_item *item; + uint count; + u32 node; + struct sk_buff *skb; + int mtype; spin_lock_bh(&tn->nametbl_lock); - while (count--) { - if (!tipc_update_nametbl(net, item, node, msg_type(msg))) - tipc_named_add_backlog(item, msg_type(msg), node); - item++; + for (skb = skb_dequeue(inputq); skb; skb = skb_dequeue(inputq)) { + msg = buf_msg(skb); + mtype = msg_type(msg); + item = (struct distr_item *)msg_data(msg); + count = msg_data_sz(msg) / ITEM_SIZE; + node = msg_orignode(msg); + while (count--) { + if (!tipc_update_nametbl(net, item, node, mtype)) + tipc_named_add_backlog(item, mtype, node); + item++; + } + kfree_skb(skb); + tipc_named_process_backlog(net); } - tipc_named_process_backlog(net); spin_unlock_bh(&tn->nametbl_lock); - kfree_skb(buf); } /** diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h index 5ec10b59527b..dd2d9fd80da2 100644 --- a/net/tipc/name_distr.h +++ b/net/tipc/name_distr.h @@ -71,7 +71,7 @@ struct sk_buff *tipc_named_publish(struct net *net, struct publication *publ); struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *publ); void named_cluster_distribute(struct net *net, struct sk_buff *buf); void tipc_named_node_up(struct net *net, u32 dnode); -void tipc_named_rcv(struct net *net, struct sk_buff *buf); +void tipc_named_rcv(struct net *net, struct sk_buff_head *msg_queue); void tipc_named_reinit(struct net *net); void tipc_named_process_backlog(struct net *net); void tipc_publ_notify(struct net *net, struct list_head *nsub_list, u32 addr); diff --git a/net/tipc/node.c b/net/tipc/node.c index 1c409c45f0fe..dcb83d9b2193 100644 --- a/net/tipc/node.c +++ b/net/tipc/node.c @@ -111,11 +111,8 @@ struct tipc_node *tipc_node_create(struct net *net, u32 addr) INIT_LIST_HEAD(&n_ptr->list); INIT_LIST_HEAD(&n_ptr->publ_list); INIT_LIST_HEAD(&n_ptr->conn_sks); - skb_queue_head_init(&n_ptr->waiting_sks); __skb_queue_head_init(&n_ptr->bclink.deferred_queue); - hlist_add_head_rcu(&n_ptr->hash, &tn->node_htable[tipc_hashfn(addr)]); - list_for_each_entry_rcu(temp_node, &tn->node_list, list) { if (n_ptr->addr < temp_node->addr) break; @@ -201,19 +198,22 @@ void tipc_node_abort_sock_conns(struct net *net, struct list_head *conns) { struct tipc_net *tn = net_generic(net, tipc_net_id); struct tipc_sock_conn *conn, *safe; - struct sk_buff *buf; + struct sk_buff *skb; + struct sk_buff_head skbs; + skb_queue_head_init(&skbs); list_for_each_entry_safe(conn, safe, conns, list) { - buf = tipc_msg_create(TIPC_CRITICAL_IMPORTANCE, + skb = tipc_msg_create(TIPC_CRITICAL_IMPORTANCE, TIPC_CONN_MSG, SHORT_H_SIZE, 0, tn->own_addr, conn->peer_node, conn->port, conn->peer_port, TIPC_ERR_NO_NODE); - if (likely(buf)) - tipc_sk_rcv(net, buf); + if (likely(skb)) + skb_queue_tail(&skbs, skb); list_del(&conn->list); kfree(conn); } + tipc_sk_rcv(net, &skbs); } /** @@ -568,37 +568,36 @@ void tipc_node_unlock(struct tipc_node *node) struct net *net = node->net; LIST_HEAD(nsub_list); LIST_HEAD(conn_sks); - struct sk_buff_head waiting_sks; u32 addr = 0; - int flags = node->action_flags; + u32 flags = node->action_flags; u32 link_id = 0; + struct sk_buff_head *inputq = node->inputq; + struct sk_buff_head *namedq = node->inputq; - if (likely(!flags)) { + if (likely(!flags || (flags == TIPC_MSG_EVT))) { + node->action_flags = 0; spin_unlock_bh(&node->lock); + if (flags == TIPC_MSG_EVT) + tipc_sk_rcv(net, inputq); return; } addr = node->addr; link_id = node->link_id; - __skb_queue_head_init(&waiting_sks); - - if (flags & TIPC_WAKEUP_USERS) - skb_queue_splice_init(&node->waiting_sks, &waiting_sks); + namedq = node->namedq; if (flags & TIPC_NOTIFY_NODE_DOWN) { list_replace_init(&node->publ_list, &nsub_list); list_replace_init(&node->conn_sks, &conn_sks); } - node->action_flags &= ~(TIPC_WAKEUP_USERS | TIPC_NOTIFY_NODE_DOWN | + node->action_flags &= ~(TIPC_MSG_EVT | TIPC_NOTIFY_NODE_DOWN | TIPC_NOTIFY_NODE_UP | TIPC_NOTIFY_LINK_UP | TIPC_NOTIFY_LINK_DOWN | - TIPC_WAKEUP_BCAST_USERS); + TIPC_WAKEUP_BCAST_USERS | + TIPC_NAMED_MSG_EVT); spin_unlock_bh(&node->lock); - while (!skb_queue_empty(&waiting_sks)) - tipc_sk_rcv(net, __skb_dequeue(&waiting_sks)); - if (!list_empty(&conn_sks)) tipc_node_abort_sock_conns(net, &conn_sks); @@ -618,6 +617,12 @@ void tipc_node_unlock(struct tipc_node *node) if (flags & TIPC_NOTIFY_LINK_DOWN) tipc_nametbl_withdraw(net, TIPC_LINK_STATE, addr, link_id, addr); + + if (flags & TIPC_MSG_EVT) + tipc_sk_rcv(net, inputq); + + if (flags & TIPC_NAMED_MSG_EVT) + tipc_named_rcv(net, namedq); } /* Caller should hold node lock for the passed node */ diff --git a/net/tipc/node.h b/net/tipc/node.h index 43ef88ef3035..c2b0fcf4042b 100644 --- a/net/tipc/node.h +++ b/net/tipc/node.h @@ -55,14 +55,15 @@ * TIPC_DISTRIBUTE_NAME: publish or withdraw link state name type */ enum { + TIPC_MSG_EVT = 1, TIPC_WAIT_PEER_LINKS_DOWN = (1 << 1), TIPC_WAIT_OWN_LINKS_DOWN = (1 << 2), TIPC_NOTIFY_NODE_DOWN = (1 << 3), TIPC_NOTIFY_NODE_UP = (1 << 4), - TIPC_WAKEUP_USERS = (1 << 5), - TIPC_WAKEUP_BCAST_USERS = (1 << 6), - TIPC_NOTIFY_LINK_UP = (1 << 7), - TIPC_NOTIFY_LINK_DOWN = (1 << 8) + TIPC_WAKEUP_BCAST_USERS = (1 << 5), + TIPC_NOTIFY_LINK_UP = (1 << 6), + TIPC_NOTIFY_LINK_DOWN = (1 << 7), + TIPC_NAMED_MSG_EVT = (1 << 8) }; /** @@ -92,6 +93,9 @@ struct tipc_node_bclink { * @lock: spinlock governing access to structure * @net: the applicable net namespace * @hash: links to adjacent nodes in unsorted hash chain + * @inputq: pointer to input queue containing messages for msg event + * @namedq: pointer to name table input queue with name table messages + * @curr_link: the link holding the node lock, if any * @active_links: pointers to active links to node * @links: pointers to all links to node * @action_flags: bit mask of different types of node actions @@ -109,10 +113,12 @@ struct tipc_node { spinlock_t lock; struct net *net; struct hlist_node hash; + struct sk_buff_head *inputq; + struct sk_buff_head *namedq; struct tipc_link *active_links[2]; u32 act_mtus[2]; struct tipc_link *links[MAX_BEARERS]; - unsigned int action_flags; + int action_flags; struct tipc_node_bclink bclink; struct list_head list; int link_cnt; @@ -120,7 +126,6 @@ struct tipc_node { u32 signature; u32 link_id; struct list_head publ_list; - struct sk_buff_head waiting_sks; struct list_head conn_sks; struct rcu_head rcu; }; diff --git a/net/tipc/socket.c b/net/tipc/socket.c index 611a04fb0ddc..c1a4611649ab 100644 --- a/net/tipc/socket.c +++ b/net/tipc/socket.c @@ -41,6 +41,7 @@ #include "node.h" #include "link.h" #include "config.h" +#include "name_distr.h" #include "socket.h" #define SS_LISTENING -1 /* socket is listening */ @@ -785,10 +786,16 @@ void tipc_sk_mcast_rcv(struct net *net, struct sk_buff *buf) struct sk_buff *b; uint i, last, dst = 0; u32 scope = TIPC_CLUSTER_SCOPE; + struct sk_buff_head msgs; if (in_own_node(net, msg_orignode(msg))) scope = TIPC_NODE_SCOPE; + if (unlikely(!msg_mcast(msg))) { + pr_warn("Received non-multicast msg in multicast\n"); + kfree_skb(buf); + goto exit; + } /* Create destination port list: */ tipc_nametbl_mc_translate(net, msg_nametype(msg), msg_namelower(msg), msg_nameupper(msg), scope, &dports); @@ -806,9 +813,12 @@ void tipc_sk_mcast_rcv(struct net *net, struct sk_buff *buf) continue; } msg_set_destport(msg, item->ports[i]); - tipc_sk_rcv(net, b); + skb_queue_head_init(&msgs); + skb_queue_tail(&msgs, b); + tipc_sk_rcv(net, &msgs); } } +exit: tipc_port_list_free(&dports); } @@ -1760,71 +1770,99 @@ static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb) } /** - * tipc_sk_enqueue_skb - enqueue buffer to socket or backlog queue - * @sk: socket - * @skb: pointer to message. Set to NULL if buffer is consumed. - * @dnode: if buffer should be forwarded/returned, send to this node + * tipc_sk_enqueue - extract all buffers with destination 'dport' from + * inputq and try adding them to socket or backlog queue + * @inputq: list of incoming buffers with potentially different destinations + * @sk: socket where the buffers should be enqueued + * @dport: port number for the socket + * @_skb: returned buffer to be forwarded or rejected, if applicable * * Caller must hold socket lock * - * Returns TIPC_OK (0) or -tipc error code + * Returns TIPC_OK if all buffers enqueued, otherwise -TIPC_ERR_OVERLOAD + * or -TIPC_ERR_NO_PORT */ -static int tipc_sk_enqueue_skb(struct sock *sk, struct sk_buff **skb) +static int tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk, + u32 dport, struct sk_buff **_skb) { unsigned int lim; atomic_t *dcnt; - - if (unlikely(!*skb)) - return TIPC_OK; - if (!sock_owned_by_user(sk)) - return filter_rcv(sk, skb); - dcnt = &tipc_sk(sk)->dupl_rcvcnt; - if (sk->sk_backlog.len) - atomic_set(dcnt, 0); - lim = rcvbuf_limit(sk, *skb) + atomic_read(dcnt); - if (unlikely(sk_add_backlog(sk, *skb, lim))) + int err; + struct sk_buff *skb; + unsigned long time_limit = jiffies + 2; + + while (skb_queue_len(inputq)) { + skb = tipc_skb_dequeue(inputq, dport); + if (unlikely(!skb)) + return TIPC_OK; + /* Return if softirq window exhausted */ + if (unlikely(time_after_eq(jiffies, time_limit))) + return TIPC_OK; + if (!sock_owned_by_user(sk)) { + err = filter_rcv(sk, &skb); + if (likely(!skb)) + continue; + *_skb = skb; + return err; + } + dcnt = &tipc_sk(sk)->dupl_rcvcnt; + if (sk->sk_backlog.len) + atomic_set(dcnt, 0); + lim = rcvbuf_limit(sk, skb) + atomic_read(dcnt); + if (likely(!sk_add_backlog(sk, skb, lim))) + continue; + *_skb = skb; return -TIPC_ERR_OVERLOAD; - *skb = NULL; + } return TIPC_OK; } /** - * tipc_sk_rcv - handle incoming message - * @skb: buffer containing arriving message - * Consumes buffer - * Returns 0 if success, or errno: -EHOSTUNREACH + * tipc_sk_rcv - handle a chain of incoming buffers + * @inputq: buffer list containing the buffers + * Consumes all buffers in list until inputq is empty + * Note: may be called in multiple threads referring to the same queue + * Returns 0 if last buffer was accepted, otherwise -EHOSTUNREACH + * Only node local calls check the return value, sending single-buffer queues */ -int tipc_sk_rcv(struct net *net, struct sk_buff *skb) +int tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq) { + u32 dnode, dport = 0; + int err = -TIPC_ERR_NO_PORT; + struct sk_buff *skb; struct tipc_sock *tsk; struct tipc_net *tn; struct sock *sk; - u32 dport = msg_destport(buf_msg(skb)); - int err = -TIPC_ERR_NO_PORT; - u32 dnode; - /* Find destination */ - tsk = tipc_sk_lookup(net, dport); - if (likely(tsk)) { - sk = &tsk->sk; - spin_lock_bh(&sk->sk_lock.slock); - err = tipc_sk_enqueue_skb(sk, &skb); - spin_unlock_bh(&sk->sk_lock.slock); - sock_put(sk); - } - if (likely(!skb)) - return 0; - if (tipc_msg_lookup_dest(net, skb, &dnode, &err)) - goto xmit; - if (!err) { - dnode = msg_destnode(buf_msg(skb)); - goto xmit; - } - tn = net_generic(net, tipc_net_id); - if (!tipc_msg_reverse(tn->own_addr, skb, &dnode, -err)) - return -EHOSTUNREACH; + while (skb_queue_len(inputq)) { + skb = NULL; + dport = tipc_skb_peek_port(inputq, dport); + tsk = tipc_sk_lookup(net, dport); + if (likely(tsk)) { + sk = &tsk->sk; + if (likely(spin_trylock_bh(&sk->sk_lock.slock))) { + err = tipc_sk_enqueue(inputq, sk, dport, &skb); + spin_unlock_bh(&sk->sk_lock.slock); + dport = 0; + } + sock_put(sk); + } else { + skb = tipc_skb_dequeue(inputq, dport); + } + if (likely(!skb)) + continue; + if (tipc_msg_lookup_dest(net, skb, &dnode, &err)) + goto xmit; + if (!err) { + dnode = msg_destnode(buf_msg(skb)); + goto xmit; + } + tn = net_generic(net, tipc_net_id); + if (!tipc_msg_reverse(tn->own_addr, skb, &dnode, -err)) + continue; xmit: - tipc_link_xmit_skb(net, skb, dnode, dport); + tipc_link_xmit_skb(net, skb, dnode, dport); + } return err ? -EHOSTUNREACH : 0; } diff --git a/net/tipc/socket.h b/net/tipc/socket.h index f56c3fded51f..e3dbdc0e1be7 100644 --- a/net/tipc/socket.h +++ b/net/tipc/socket.h @@ -49,7 +49,7 @@ int tipc_sock_create_local(struct net *net, int type, struct socket **res); void tipc_sock_release_local(struct socket *sock); int tipc_sock_accept_local(struct socket *sock, struct socket **newsock, int flags); -int tipc_sk_rcv(struct net *net, struct sk_buff *buf); +int tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq); struct sk_buff *tipc_sk_socks_show(struct net *net); void tipc_sk_mcast_rcv(struct net *net, struct sk_buff *buf); void tipc_sk_reinit(struct net *net); -- cgit v1.2.3-70-g09d2 From cb1b728096f54e7408d60fb571944bed00c5b771 Mon Sep 17 00:00:00 2001 From: Jon Paul Maloy Date: Thu, 5 Feb 2015 08:36:44 -0500 Subject: tipc: eliminate race condition at multicast reception In a previous commit in this series we resolved a race problem during unicast message reception. Here, we resolve the same problem at multicast reception. We apply the same technique: an input queue serializing the delivery of arriving buffers. The main difference is that here we do it in two steps. First, the broadcast link feeds arriving buffers into the tail of an arrival queue, which head is consumed at the socket level, and where destination lookup is performed. Second, if the lookup is successful, the resulting buffer clones are fed into a second queue, the input queue. This queue is consumed at reception in the socket just like in the unicast case. Both queues are protected by the same lock, -the one of the input queue. Reviewed-by: Ying Xue Signed-off-by: Jon Maloy Signed-off-by: David S. Miller --- net/tipc/bcast.c | 57 ++++++++++++++++++++++++---------------- net/tipc/bcast.h | 3 +++ net/tipc/msg.h | 17 ++++++++++++ net/tipc/name_table.h | 2 +- net/tipc/node.c | 11 +++++--- net/tipc/node.h | 7 +++-- net/tipc/socket.c | 72 +++++++++++++++++++++++++++++++-------------------- net/tipc/socket.h | 4 +-- 8 files changed, 114 insertions(+), 59 deletions(-) (limited to 'net/tipc/node.h') diff --git a/net/tipc/bcast.c b/net/tipc/bcast.c index 3eaa931e2e8c..81b1fef1f5e0 100644 --- a/net/tipc/bcast.c +++ b/net/tipc/bcast.c @@ -79,6 +79,13 @@ static void tipc_bclink_unlock(struct net *net) tipc_link_reset_all(node); } +void tipc_bclink_input(struct net *net) +{ + struct tipc_net *tn = net_generic(net, tipc_net_id); + + tipc_sk_mcast_rcv(net, &tn->bclink->arrvq, &tn->bclink->inputq); +} + uint tipc_bclink_get_mtu(void) { return MAX_PKT_DEFAULT_MCAST; @@ -356,7 +363,7 @@ static void bclink_peek_nack(struct net *net, struct tipc_msg *msg) tipc_node_unlock(n_ptr); } -/* tipc_bclink_xmit - broadcast buffer chain to all nodes in cluster +/* tipc_bclink_xmit - deliver buffer chain to all nodes in cluster * and to identified node local sockets * @net: the applicable net namespace * @list: chain of buffers containing message @@ -371,6 +378,8 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list) int rc = 0; int bc = 0; struct sk_buff *skb; + struct sk_buff_head arrvq; + struct sk_buff_head inputq; /* Prepare clone of message for local node */ skb = tipc_msg_reassemble(list); @@ -379,7 +388,7 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list) return -EHOSTUNREACH; } - /* Broadcast to all other nodes */ + /* Broadcast to all nodes */ if (likely(bclink)) { tipc_bclink_lock(net); if (likely(bclink->bcast_nodes.count)) { @@ -399,12 +408,15 @@ int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list) if (unlikely(!bc)) __skb_queue_purge(list); - /* Deliver message clone */ - if (likely(!rc)) - tipc_sk_mcast_rcv(net, skb); - else + if (unlikely(rc)) { kfree_skb(skb); - + return rc; + } + /* Deliver message clone */ + __skb_queue_head_init(&arrvq); + skb_queue_head_init(&inputq); + __skb_queue_tail(&arrvq, skb); + tipc_sk_mcast_rcv(net, &arrvq, &inputq); return rc; } @@ -449,7 +461,7 @@ void tipc_bclink_rcv(struct net *net, struct sk_buff *buf) int deferred = 0; int pos = 0; struct sk_buff *iskb; - struct sk_buff_head msgs; + struct sk_buff_head *arrvq, *inputq; /* Screen out unwanted broadcast messages */ if (msg_mc_netid(msg) != tn->net_id) @@ -486,6 +498,8 @@ void tipc_bclink_rcv(struct net *net, struct sk_buff *buf) /* Handle in-sequence broadcast message */ seqno = msg_seqno(msg); next_in = mod(node->bclink.last_in + 1); + arrvq = &tn->bclink->arrvq; + inputq = &tn->bclink->inputq; if (likely(seqno == next_in)) { receive: @@ -493,21 +507,26 @@ receive: if (likely(msg_isdata(msg))) { tipc_bclink_lock(net); bclink_accept_pkt(node, seqno); + spin_lock_bh(&inputq->lock); + __skb_queue_tail(arrvq, buf); + spin_unlock_bh(&inputq->lock); + node->action_flags |= TIPC_BCAST_MSG_EVT; tipc_bclink_unlock(net); tipc_node_unlock(node); - if (likely(msg_mcast(msg))) - tipc_sk_mcast_rcv(net, buf); - else - kfree_skb(buf); } else if (msg_user(msg) == MSG_BUNDLER) { tipc_bclink_lock(net); bclink_accept_pkt(node, seqno); bcl->stats.recv_bundles++; bcl->stats.recv_bundled += msg_msgcnt(msg); + pos = 0; + while (tipc_msg_extract(buf, &iskb, &pos)) { + spin_lock_bh(&inputq->lock); + __skb_queue_tail(arrvq, iskb); + spin_unlock_bh(&inputq->lock); + } + node->action_flags |= TIPC_BCAST_MSG_EVT; tipc_bclink_unlock(net); tipc_node_unlock(node); - while (tipc_msg_extract(buf, &iskb, &pos)) - tipc_sk_mcast_rcv(net, iskb); } else if (msg_user(msg) == MSG_FRAGMENTER) { tipc_buf_append(&node->bclink.reasm_buf, &buf); if (unlikely(!buf && !node->bclink.reasm_buf)) @@ -523,14 +542,6 @@ receive: } tipc_bclink_unlock(net); tipc_node_unlock(node); - } else if (msg_user(msg) == NAME_DISTRIBUTOR) { - tipc_bclink_lock(net); - bclink_accept_pkt(node, seqno); - tipc_bclink_unlock(net); - tipc_node_unlock(node); - skb_queue_head_init(&msgs); - skb_queue_tail(&msgs, buf); - tipc_named_rcv(net, &msgs); } else { tipc_bclink_lock(net); bclink_accept_pkt(node, seqno); @@ -950,6 +961,8 @@ int tipc_bclink_init(struct net *net) skb_queue_head_init(&bcl->wakeupq); bcl->next_out_no = 1; spin_lock_init(&bclink->node.lock); + __skb_queue_head_init(&bclink->arrvq); + skb_queue_head_init(&bclink->inputq); bcl->owner = &bclink->node; bcl->owner->net = net; bcl->max_pkt = MAX_PKT_DEFAULT_MCAST; diff --git a/net/tipc/bcast.h b/net/tipc/bcast.h index 8f4d4dc38e11..a910c0b9f249 100644 --- a/net/tipc/bcast.h +++ b/net/tipc/bcast.h @@ -97,6 +97,8 @@ struct tipc_bclink { struct tipc_link link; struct tipc_node node; unsigned int flags; + struct sk_buff_head arrvq; + struct sk_buff_head inputq; struct tipc_node_map bcast_nodes; struct tipc_node *retransmit_to; }; @@ -134,5 +136,6 @@ uint tipc_bclink_get_mtu(void); int tipc_bclink_xmit(struct net *net, struct sk_buff_head *list); void tipc_bclink_wakeup_users(struct net *net); int tipc_nl_add_bc_link(struct net *net, struct tipc_nl_msg *msg); +void tipc_bclink_input(struct net *net); #endif diff --git a/net/tipc/msg.h b/net/tipc/msg.h index ab467261bd9d..9ace47f44a69 100644 --- a/net/tipc/msg.h +++ b/net/tipc/msg.h @@ -767,6 +767,23 @@ bool tipc_msg_lookup_dest(struct net *net, struct sk_buff *skb, u32 *dnode, int *err); struct sk_buff *tipc_msg_reassemble(struct sk_buff_head *list); +/* tipc_skb_peek(): peek and reserve first buffer in list + * @list: list to be peeked in + * Returns pointer to first buffer in list, if any + */ +static inline struct sk_buff *tipc_skb_peek(struct sk_buff_head *list, + spinlock_t *lock) +{ + struct sk_buff *skb; + + spin_lock_bh(lock); + skb = skb_peek(list); + if (skb) + skb_get(skb); + spin_unlock_bh(lock); + return skb; +} + /* tipc_skb_peek_port(): find a destination port, ignoring all destinations * up to and including 'filter'. * Note: ignoring previously tried destinations minimizes the risk of diff --git a/net/tipc/name_table.h b/net/tipc/name_table.h index 52501fdaafa5..0304ddc6b101 100644 --- a/net/tipc/name_table.h +++ b/net/tipc/name_table.h @@ -1,7 +1,7 @@ /* * net/tipc/name_table.h: Include file for TIPC name table code * - * Copyright (c) 2000-2006, 2014, Ericsson AB + * Copyright (c) 2000-2006, 2014-2015, Ericsson AB * Copyright (c) 2004-2005, 2010-2011, Wind River Systems * All rights reserved. * diff --git a/net/tipc/node.c b/net/tipc/node.c index c7fdf3dec92c..52308498f208 100644 --- a/net/tipc/node.c +++ b/net/tipc/node.c @@ -582,10 +582,10 @@ void tipc_node_unlock(struct tipc_node *node) namedq = node->namedq; publ_list = &node->publ_list; - node->action_flags &= ~(TIPC_MSG_EVT | TIPC_NOTIFY_NODE_DOWN | - TIPC_NOTIFY_NODE_UP | TIPC_NOTIFY_LINK_UP | - TIPC_NOTIFY_LINK_DOWN | - TIPC_WAKEUP_BCAST_USERS | + node->action_flags &= ~(TIPC_MSG_EVT | + TIPC_NOTIFY_NODE_DOWN | TIPC_NOTIFY_NODE_UP | + TIPC_NOTIFY_LINK_DOWN | TIPC_NOTIFY_LINK_UP | + TIPC_WAKEUP_BCAST_USERS | TIPC_BCAST_MSG_EVT | TIPC_NAMED_MSG_EVT); spin_unlock_bh(&node->lock); @@ -612,6 +612,9 @@ void tipc_node_unlock(struct tipc_node *node) if (flags & TIPC_NAMED_MSG_EVT) tipc_named_rcv(net, namedq); + + if (flags & TIPC_BCAST_MSG_EVT) + tipc_bclink_input(net); } /* Caller should hold node lock for the passed node */ diff --git a/net/tipc/node.h b/net/tipc/node.h index c2b0fcf4042b..20ec13f9bede 100644 --- a/net/tipc/node.h +++ b/net/tipc/node.h @@ -1,7 +1,7 @@ /* * net/tipc/node.h: Include file for TIPC node management routines * - * Copyright (c) 2000-2006, 2014, Ericsson AB + * Copyright (c) 2000-2006, 2014-2015, Ericsson AB * Copyright (c) 2005, 2010-2014, Wind River Systems * All rights reserved. * @@ -63,7 +63,8 @@ enum { TIPC_WAKEUP_BCAST_USERS = (1 << 5), TIPC_NOTIFY_LINK_UP = (1 << 6), TIPC_NOTIFY_LINK_DOWN = (1 << 7), - TIPC_NAMED_MSG_EVT = (1 << 8) + TIPC_NAMED_MSG_EVT = (1 << 8), + TIPC_BCAST_MSG_EVT = (1 << 9) }; /** @@ -74,6 +75,7 @@ enum { * @oos_state: state tracker for handling OOS b'cast messages * @deferred_queue: deferred queue saved OOS b'cast message received from node * @reasm_buf: broadcast reassembly queue head from node + * @inputq_map: bitmap indicating which inqueues should be kicked * @recv_permitted: true if node is allowed to receive b'cast messages */ struct tipc_node_bclink { @@ -84,6 +86,7 @@ struct tipc_node_bclink { u32 deferred_size; struct sk_buff_head deferred_queue; struct sk_buff *reasm_buf; + int inputq_map; bool recv_permitted; }; diff --git a/net/tipc/socket.c b/net/tipc/socket.c index 26aec8414ac1..66666805b53c 100644 --- a/net/tipc/socket.c +++ b/net/tipc/socket.c @@ -776,44 +776,60 @@ new_mtu: return rc; } -/* tipc_sk_mcast_rcv - Deliver multicast message to all destination sockets +/** + * tipc_sk_mcast_rcv - Deliver multicast messages to all destination sockets + * @arrvq: queue with arriving messages, to be cloned after destination lookup + * @inputq: queue with cloned messages, delivered to socket after dest lookup + * + * Multi-threaded: parallel calls with reference to same queues may occur */ -void tipc_sk_mcast_rcv(struct net *net, struct sk_buff *skb) +void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq, + struct sk_buff_head *inputq) { - struct tipc_msg *msg = buf_msg(skb); + struct tipc_msg *msg; struct tipc_plist dports; - struct sk_buff *cskb; u32 portid; u32 scope = TIPC_CLUSTER_SCOPE; - struct sk_buff_head msgq; - uint hsz = skb_headroom(skb) + msg_hdr_sz(msg); + struct sk_buff_head tmpq; + uint hsz; + struct sk_buff *skb, *_skb; - skb_queue_head_init(&msgq); + __skb_queue_head_init(&tmpq); tipc_plist_init(&dports); - if (in_own_node(net, msg_orignode(msg))) - scope = TIPC_NODE_SCOPE; - - if (unlikely(!msg_mcast(msg))) { - pr_warn("Received non-multicast msg in multicast\n"); - goto exit; - } - /* Create destination port list: */ - tipc_nametbl_mc_translate(net, msg_nametype(msg), msg_namelower(msg), - msg_nameupper(msg), scope, &dports); - portid = tipc_plist_pop(&dports); - for (; portid; portid = tipc_plist_pop(&dports)) { - cskb = __pskb_copy(skb, hsz, GFP_ATOMIC); - if (!cskb) { - pr_warn("Failed do clone mcast rcv buffer\n"); - continue; + skb = tipc_skb_peek(arrvq, &inputq->lock); + for (; skb; skb = tipc_skb_peek(arrvq, &inputq->lock)) { + msg = buf_msg(skb); + hsz = skb_headroom(skb) + msg_hdr_sz(msg); + + if (in_own_node(net, msg_orignode(msg))) + scope = TIPC_NODE_SCOPE; + + /* Create destination port list and message clones: */ + tipc_nametbl_mc_translate(net, + msg_nametype(msg), msg_namelower(msg), + msg_nameupper(msg), scope, &dports); + portid = tipc_plist_pop(&dports); + for (; portid; portid = tipc_plist_pop(&dports)) { + _skb = __pskb_copy(skb, hsz, GFP_ATOMIC); + if (_skb) { + msg_set_destport(buf_msg(_skb), portid); + __skb_queue_tail(&tmpq, _skb); + continue; + } + pr_warn("Failed to clone mcast rcv buffer\n"); } - msg_set_destport(buf_msg(cskb), portid); - skb_queue_tail(&msgq, cskb); + /* Append to inputq if not already done by other thread */ + spin_lock_bh(&inputq->lock); + if (skb_peek(arrvq) == skb) { + skb_queue_splice_tail_init(&tmpq, inputq); + kfree_skb(__skb_dequeue(arrvq)); + } + spin_unlock_bh(&inputq->lock); + __skb_queue_purge(&tmpq); + kfree_skb(skb); } - tipc_sk_rcv(net, &msgq); -exit: - kfree_skb(skb); + tipc_sk_rcv(net, inputq); } /** diff --git a/net/tipc/socket.h b/net/tipc/socket.h index 95b015909ac1..8be0da7df8fc 100644 --- a/net/tipc/socket.h +++ b/net/tipc/socket.h @@ -42,7 +42,6 @@ #define TIPC_FLOWCTRL_WIN (TIPC_CONNACK_INTV * 2) #define TIPC_CONN_OVERLOAD_LIMIT ((TIPC_FLOWCTRL_WIN * 2 + 1) * \ SKB_TRUESIZE(TIPC_MAX_USER_MSG_SIZE)) - int tipc_socket_init(void); void tipc_socket_stop(void); int tipc_sock_create_local(struct net *net, int type, struct socket **res); @@ -51,7 +50,8 @@ int tipc_sock_accept_local(struct socket *sock, struct socket **newsock, int flags); int tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq); struct sk_buff *tipc_sk_socks_show(struct net *net); -void tipc_sk_mcast_rcv(struct net *net, struct sk_buff *buf); +void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq, + struct sk_buff_head *inputq); void tipc_sk_reinit(struct net *net); int tipc_sk_rht_init(struct net *net); void tipc_sk_rht_destroy(struct net *net); -- cgit v1.2.3-70-g09d2 From 357ebdbfca0baa9a8d8d85307393e9ec3406affc Mon Sep 17 00:00:00 2001 From: Richard Alpe Date: Mon, 9 Feb 2015 09:50:07 +0100 Subject: tipc: convert legacy nl link dump to nl compat Convert TIPC_CMD_GET_LINKS to compat dumpit and remove global link counter solely used by the legacy API. Signed-off-by: Richard Alpe Reviewed-by: Erik Hugne Reviewed-by: Ying Xue Reviewed-by: Jon Maloy Signed-off-by: David S. Miller --- net/tipc/config.c | 4 --- net/tipc/netlink_compat.c | 23 +++++++++++++++ net/tipc/node.c | 73 ----------------------------------------------- net/tipc/node.h | 2 -- 4 files changed, 23 insertions(+), 79 deletions(-) (limited to 'net/tipc/node.h') diff --git a/net/tipc/config.c b/net/tipc/config.c index 67b76ee847f3..1b17f5846e86 100644 --- a/net/tipc/config.c +++ b/net/tipc/config.c @@ -209,10 +209,6 @@ struct sk_buff *tipc_cfg_do_cmd(struct net *net, u32 orig_node, u16 cmd, rep_tlv_buf = tipc_node_get_nodes(net, req_tlv_area, req_tlv_space); break; - case TIPC_CMD_GET_LINKS: - rep_tlv_buf = tipc_node_get_links(net, req_tlv_area, - req_tlv_space); - break; case TIPC_CMD_RESET_LINK_STATS: rep_tlv_buf = tipc_link_cmd_reset_stats(net, req_tlv_area, req_tlv_space); diff --git a/net/tipc/netlink_compat.c b/net/tipc/netlink_compat.c index 899bd94da467..bff9403899ed 100644 --- a/net/tipc/netlink_compat.c +++ b/net/tipc/netlink_compat.c @@ -553,6 +553,22 @@ static int tipc_nl_compat_link_stat_dump(struct tipc_nl_compat_msg *msg, return 0; } +static int tipc_nl_compat_link_dump(struct tipc_nl_compat_msg *msg, + struct nlattr **attrs) +{ + struct nlattr *link[TIPC_NLA_LINK_MAX + 1]; + struct tipc_link_info link_info; + + nla_parse_nested(link, TIPC_NLA_LINK_MAX, attrs[TIPC_NLA_LINK], NULL); + + link_info.dest = nla_get_flag(link[TIPC_NLA_LINK_DEST]); + link_info.up = htonl(nla_get_flag(link[TIPC_NLA_LINK_UP])); + strcpy(link_info.str, nla_data(link[TIPC_NLA_LINK_NAME])); + + return tipc_add_tlv(msg->rep, TIPC_TLV_LINK_INFO, + &link_info, sizeof(link_info)); +} + static int tipc_nl_compat_handle(struct tipc_nl_compat_msg *msg) { struct tipc_nl_compat_cmd_dump dump; @@ -584,6 +600,12 @@ static int tipc_nl_compat_handle(struct tipc_nl_compat_msg *msg) dump.dumpit = tipc_nl_link_dump; dump.format = tipc_nl_compat_link_stat_dump; return tipc_nl_compat_dumpit(&dump, msg); + case TIPC_CMD_GET_LINKS: + msg->req_type = TIPC_TLV_NET_ADDR; + msg->rep_size = ULTRA_STRING_MAX_LEN; + dump.dumpit = tipc_nl_link_dump; + dump.format = tipc_nl_compat_link_dump; + return tipc_nl_compat_dumpit(&dump, msg); } return -EOPNOTSUPP; @@ -684,6 +706,7 @@ static int tipc_nl_compat_tmp_wrap(struct sk_buff *skb, struct genl_info *info) case TIPC_CMD_ENABLE_BEARER: case TIPC_CMD_DISABLE_BEARER: case TIPC_CMD_SHOW_LINK_STATS: + case TIPC_CMD_GET_LINKS: return tipc_nl_compat_recv(skb, info); } diff --git a/net/tipc/node.c b/net/tipc/node.c index 995618d6da9d..46b87f77d342 100644 --- a/net/tipc/node.c +++ b/net/tipc/node.c @@ -319,27 +319,18 @@ int tipc_node_is_up(struct tipc_node *n_ptr) void tipc_node_attach_link(struct tipc_node *n_ptr, struct tipc_link *l_ptr) { - struct tipc_net *tn = net_generic(n_ptr->net, tipc_net_id); - n_ptr->links[l_ptr->bearer_id] = l_ptr; - spin_lock_bh(&tn->node_list_lock); - tn->num_links++; - spin_unlock_bh(&tn->node_list_lock); n_ptr->link_cnt++; } void tipc_node_detach_link(struct tipc_node *n_ptr, struct tipc_link *l_ptr) { - struct tipc_net *tn = net_generic(n_ptr->net, tipc_net_id); int i; for (i = 0; i < MAX_BEARERS; i++) { if (l_ptr != n_ptr->links[i]) continue; n_ptr->links[i] = NULL; - spin_lock_bh(&tn->node_list_lock); - tn->num_links--; - spin_unlock_bh(&tn->node_list_lock); n_ptr->link_cnt--; } } @@ -467,70 +458,6 @@ struct sk_buff *tipc_node_get_nodes(struct net *net, const void *req_tlv_area, return buf; } -struct sk_buff *tipc_node_get_links(struct net *net, const void *req_tlv_area, - int req_tlv_space) -{ - struct tipc_net *tn = net_generic(net, tipc_net_id); - u32 domain; - struct sk_buff *buf; - struct tipc_node *n_ptr; - struct tipc_link_info link_info; - u32 payload_size; - - if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_NET_ADDR)) - return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR); - - domain = ntohl(*(__be32 *)TLV_DATA(req_tlv_area)); - if (!tipc_addr_domain_valid(domain)) - return tipc_cfg_reply_error_string(TIPC_CFG_INVALID_VALUE - " (network address)"); - - if (!tn->own_addr) - return tipc_cfg_reply_none(); - - spin_lock_bh(&tn->node_list_lock); - /* Get space for all unicast links + broadcast link */ - payload_size = TLV_SPACE((sizeof(link_info)) * (tn->num_links + 1)); - if (payload_size > 32768u) { - spin_unlock_bh(&tn->node_list_lock); - return tipc_cfg_reply_error_string(TIPC_CFG_NOT_SUPPORTED - " (too many links)"); - } - spin_unlock_bh(&tn->node_list_lock); - - buf = tipc_cfg_reply_alloc(payload_size); - if (!buf) - return NULL; - - /* Add TLV for broadcast link */ - link_info.dest = htonl(tipc_cluster_mask(tn->own_addr)); - link_info.up = htonl(1); - strlcpy(link_info.str, tipc_bclink_name, TIPC_MAX_LINK_NAME); - tipc_cfg_append_tlv(buf, TIPC_TLV_LINK_INFO, &link_info, sizeof(link_info)); - - /* Add TLVs for any other links in scope */ - rcu_read_lock(); - list_for_each_entry_rcu(n_ptr, &tn->node_list, list) { - u32 i; - - if (!tipc_in_scope(domain, n_ptr->addr)) - continue; - tipc_node_lock(n_ptr); - for (i = 0; i < MAX_BEARERS; i++) { - if (!n_ptr->links[i]) - continue; - link_info.dest = htonl(n_ptr->addr); - link_info.up = htonl(tipc_link_is_up(n_ptr->links[i])); - strcpy(link_info.str, n_ptr->links[i]->name); - tipc_cfg_append_tlv(buf, TIPC_TLV_LINK_INFO, - &link_info, sizeof(link_info)); - } - tipc_node_unlock(n_ptr); - } - rcu_read_unlock(); - return buf; -} - /** * tipc_node_get_linkname - get the name of a link * diff --git a/net/tipc/node.h b/net/tipc/node.h index 20ec13f9bede..59dafee61aa0 100644 --- a/net/tipc/node.h +++ b/net/tipc/node.h @@ -142,8 +142,6 @@ void tipc_node_link_down(struct tipc_node *n_ptr, struct tipc_link *l_ptr); void tipc_node_link_up(struct tipc_node *n_ptr, struct tipc_link *l_ptr); int tipc_node_active_links(struct tipc_node *n_ptr); int tipc_node_is_up(struct tipc_node *n_ptr); -struct sk_buff *tipc_node_get_links(struct net *net, const void *req_tlv_area, - int req_tlv_space); struct sk_buff *tipc_node_get_nodes(struct net *net, const void *req_tlv_area, int req_tlv_space); int tipc_node_get_linkname(struct net *net, u32 bearer_id, u32 node, -- cgit v1.2.3-70-g09d2 From 4b28cb581dd0df8d8ff19151f39683f641e576ba Mon Sep 17 00:00:00 2001 From: Richard Alpe Date: Mon, 9 Feb 2015 09:50:13 +0100 Subject: tipc: convert legacy nl node dump to nl compat Convert TIPC_CMD_GET_NODES to compat dumpit and remove global node counter solely used by the legacy API. Signed-off-by: Richard Alpe Reviewed-by: Erik Hugne Reviewed-by: Ying Xue Reviewed-by: Jon Maloy Signed-off-by: David S. Miller --- net/tipc/config.c | 4 ---- net/tipc/netlink_compat.c | 22 +++++++++++++++++++ net/tipc/node.c | 54 ----------------------------------------------- net/tipc/node.h | 2 -- 4 files changed, 22 insertions(+), 60 deletions(-) (limited to 'net/tipc/node.h') diff --git a/net/tipc/config.c b/net/tipc/config.c index df410823e889..74babd75c07d 100644 --- a/net/tipc/config.c +++ b/net/tipc/config.c @@ -205,10 +205,6 @@ struct sk_buff *tipc_cfg_do_cmd(struct net *net, u32 orig_node, u16 cmd, case TIPC_CMD_NOOP: rep_tlv_buf = tipc_cfg_reply_none(); break; - case TIPC_CMD_GET_NODES: - rep_tlv_buf = tipc_node_get_nodes(net, req_tlv_area, - req_tlv_space); - break; case TIPC_CMD_SHOW_STATS: rep_tlv_buf = tipc_show_stats(); break; diff --git a/net/tipc/netlink_compat.c b/net/tipc/netlink_compat.c index d23075efaa76..0dc448e77e13 100644 --- a/net/tipc/netlink_compat.c +++ b/net/tipc/netlink_compat.c @@ -37,6 +37,7 @@ #include "link.h" #include "name_table.h" #include "socket.h" +#include "node.h" #include #include @@ -835,6 +836,21 @@ static int tipc_nl_compat_media_dump(struct tipc_nl_compat_msg *msg, nla_len(media[TIPC_NLA_MEDIA_NAME])); } +static int tipc_nl_compat_node_dump(struct tipc_nl_compat_msg *msg, + struct nlattr **attrs) +{ + struct tipc_node_info node_info; + struct nlattr *node[TIPC_NLA_NODE_MAX + 1]; + + nla_parse_nested(node, TIPC_NLA_NODE_MAX, attrs[TIPC_NLA_NODE], NULL); + + node_info.addr = htonl(nla_get_u32(node[TIPC_NLA_NODE_ADDR])); + node_info.up = htonl(nla_get_flag(node[TIPC_NLA_NODE_UP])); + + return tipc_add_tlv(msg->rep, TIPC_TLV_NODE_INFO, &node_info, + sizeof(node_info)); +} + static int tipc_nl_compat_handle(struct tipc_nl_compat_msg *msg) { struct tipc_nl_compat_cmd_dump dump; @@ -903,6 +919,11 @@ static int tipc_nl_compat_handle(struct tipc_nl_compat_msg *msg) dump.dumpit = tipc_nl_media_dump; dump.format = tipc_nl_compat_media_dump; return tipc_nl_compat_dumpit(&dump, msg); + case TIPC_CMD_GET_NODES: + msg->rep_size = ULTRA_STRING_MAX_LEN; + dump.dumpit = tipc_nl_node_dump; + dump.format = tipc_nl_compat_node_dump; + return tipc_nl_compat_dumpit(&dump, msg); } return -EOPNOTSUPP; @@ -1011,6 +1032,7 @@ static int tipc_nl_compat_tmp_wrap(struct sk_buff *skb, struct genl_info *info) case TIPC_CMD_SHOW_NAME_TABLE: case TIPC_CMD_SHOW_PORTS: case TIPC_CMD_GET_MEDIA_NAMES: + case TIPC_CMD_GET_NODES: return tipc_nl_compat_recv(skb, info); } diff --git a/net/tipc/node.c b/net/tipc/node.c index 46b87f77d342..ddaa2bbaa35d 100644 --- a/net/tipc/node.c +++ b/net/tipc/node.c @@ -120,7 +120,6 @@ struct tipc_node *tipc_node_create(struct net *net, u32 addr) list_add_tail_rcu(&n_ptr->list, &temp_node->list); n_ptr->action_flags = TIPC_WAIT_PEER_LINKS_DOWN; n_ptr->signature = INVALID_NODE_SIG; - tn->num_nodes++; exit: spin_unlock_bh(&tn->node_list_lock); return n_ptr; @@ -131,8 +130,6 @@ static void tipc_node_delete(struct tipc_net *tn, struct tipc_node *n_ptr) list_del_rcu(&n_ptr->list); hlist_del_rcu(&n_ptr->hash); kfree_rcu(n_ptr, rcu); - - tn->num_nodes--; } void tipc_node_stop(struct net *net) @@ -407,57 +404,6 @@ static void node_lost_contact(struct tipc_node *n_ptr) } } -struct sk_buff *tipc_node_get_nodes(struct net *net, const void *req_tlv_area, - int req_tlv_space) -{ - struct tipc_net *tn = net_generic(net, tipc_net_id); - u32 domain; - struct sk_buff *buf; - struct tipc_node *n_ptr; - struct tipc_node_info node_info; - u32 payload_size; - - if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_NET_ADDR)) - return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR); - - domain = ntohl(*(__be32 *)TLV_DATA(req_tlv_area)); - if (!tipc_addr_domain_valid(domain)) - return tipc_cfg_reply_error_string(TIPC_CFG_INVALID_VALUE - " (network address)"); - - spin_lock_bh(&tn->node_list_lock); - if (!tn->num_nodes) { - spin_unlock_bh(&tn->node_list_lock); - return tipc_cfg_reply_none(); - } - - /* For now, get space for all other nodes */ - payload_size = TLV_SPACE(sizeof(node_info)) * tn->num_nodes; - if (payload_size > 32768u) { - spin_unlock_bh(&tn->node_list_lock); - return tipc_cfg_reply_error_string(TIPC_CFG_NOT_SUPPORTED - " (too many nodes)"); - } - spin_unlock_bh(&tn->node_list_lock); - - buf = tipc_cfg_reply_alloc(payload_size); - if (!buf) - return NULL; - - /* Add TLVs for all nodes in scope */ - rcu_read_lock(); - list_for_each_entry_rcu(n_ptr, &tn->node_list, list) { - if (!tipc_in_scope(domain, n_ptr->addr)) - continue; - node_info.addr = htonl(n_ptr->addr); - node_info.up = htonl(tipc_node_is_up(n_ptr)); - tipc_cfg_append_tlv(buf, TIPC_TLV_NODE_INFO, - &node_info, sizeof(node_info)); - } - rcu_read_unlock(); - return buf; -} - /** * tipc_node_get_linkname - get the name of a link * diff --git a/net/tipc/node.h b/net/tipc/node.h index 59dafee61aa0..3d18c66b7f78 100644 --- a/net/tipc/node.h +++ b/net/tipc/node.h @@ -142,8 +142,6 @@ void tipc_node_link_down(struct tipc_node *n_ptr, struct tipc_link *l_ptr); void tipc_node_link_up(struct tipc_node *n_ptr, struct tipc_link *l_ptr); int tipc_node_active_links(struct tipc_node *n_ptr); int tipc_node_is_up(struct tipc_node *n_ptr); -struct sk_buff *tipc_node_get_nodes(struct net *net, const void *req_tlv_area, - int req_tlv_space); int tipc_node_get_linkname(struct net *net, u32 bearer_id, u32 node, char *linkname, size_t len); void tipc_node_unlock(struct tipc_node *node); -- cgit v1.2.3-70-g09d2