[deliverable/linux.git] / net / tipc / bcast.c

/*
 * net/tipc/bcast.c: TIPC broadcast code
 *
 * Copyright (c) 2004-2006, 2014-2015, Ericsson AB
 * Copyright (c) 2004, Intel Corporation.
 * Copyright (c) 2005, 2010-2011, Wind River Systems
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. Neither the names of the copyright holders nor the names of its
 *    contributors may be used to endorse or promote products derived from
 *    this software without specific prior written permission.
 *
 * Alternatively, this software may be distributed under the terms of the
 * GNU General Public License ("GPL") version 2 as published by the Free
 * Software Foundation.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include <linux/tipc_config.h>
#include "socket.h"
#include "msg.h"
#include "bcast.h"
#include "name_distr.h"
#include "link.h"
#include "node.h"

#define	BCLINK_WIN_DEFAULT	50	/* bcast link window size (default) */
#define	BCLINK_WIN_MIN	        32	/* bcast minimum link window size */

const char tipc_bclink_name[] = "broadcast-link";

/**
 * struct tipc_bc_base - base structure for keeping broadcast send state
 * @link: broadcast send link structure
 * @inputq: data input queue; will only carry SOCK_WAKEUP messages
 * @dest: array keeping number of reachable destinations per bearer
 * @primary_bearer: a bearer having links to all broadcast destinations, if any
 */
struct tipc_bc_base {
	struct tipc_link *link;
	struct sk_buff_head inputq;
	int dests[MAX_BEARERS];
	int primary_bearer;
};

static struct tipc_bc_base *tipc_bc_base(struct net *net)
{
	return tipc_net(net)->bcbase;
}

int tipc_bcast_get_mtu(struct net *net)
{
	return tipc_link_mtu(tipc_bc_sndlink(net));
}

/* tipc_bcbase_select_primary(): find a bearer with links to all destinations,
 *                               if any, and make it primary bearer
 */
static void tipc_bcbase_select_primary(struct net *net)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);
	int all_dests =  tipc_link_bc_peers(bb->link);
	int i, mtu;

	bb->primary_bearer = INVALID_BEARER_ID;

	if (!all_dests)
		return;

	for (i = 0; i < MAX_BEARERS; i++) {
		if (!bb->dests[i])
			continue;

		mtu = tipc_bearer_mtu(net, i);
		if (mtu < tipc_link_mtu(bb->link))
			tipc_link_set_mtu(bb->link, mtu);

		if (bb->dests[i] < all_dests)
			continue;

		bb->primary_bearer = i;

		/* Reduce risk that all nodes select same primary */
		if ((i ^ tipc_own_addr(net)) & 1)
			break;
	}
}

void tipc_bcast_inc_bearer_dst_cnt(struct net *net, int bearer_id)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);

	tipc_bcast_lock(net);
	bb->dests[bearer_id]++;
	tipc_bcbase_select_primary(net);
	tipc_bcast_unlock(net);
}

void tipc_bcast_dec_bearer_dst_cnt(struct net *net, int bearer_id)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);

	tipc_bcast_lock(net);
	bb->dests[bearer_id]--;
	tipc_bcbase_select_primary(net);
	tipc_bcast_unlock(net);
}

/* tipc_bcbase_xmit - broadcast a packet queue across one or more bearers
 *
 * Note that number of reachable destinations, as indicated in the dests[]
 * array, may transitionally differ from the number of destinations indicated
 * in each sent buffer. We can sustain this. Excess destination nodes will
 * drop and never acknowledge the unexpected packets, and missing destinations
 * will either require retransmission (if they are just about to be added to
 * the bearer), or be removed from the buffer's 'ackers' counter (if they
 * just went down)
 */
static void tipc_bcbase_xmit(struct net *net, struct sk_buff_head *xmitq)
{
	int bearer_id;
	struct tipc_bc_base *bb = tipc_bc_base(net);
	struct sk_buff *skb, *_skb;
	struct sk_buff_head _xmitq;

	if (skb_queue_empty(xmitq))
		return;

	/* The typical case: at least one bearer has links to all nodes */
	bearer_id = bb->primary_bearer;
	if (bearer_id >= 0) {
		tipc_bearer_bc_xmit(net, bearer_id, xmitq);
		return;
	}

	/* We have to transmit across all bearers */
	skb_queue_head_init(&_xmitq);
	for (bearer_id = 0; bearer_id < MAX_BEARERS; bearer_id++) {
		if (!bb->dests[bearer_id])
			continue;

		skb_queue_walk(xmitq, skb) {
			_skb = pskb_copy_for_clone(skb, GFP_ATOMIC);
			if (!_skb)
				break;
			__skb_queue_tail(&_xmitq, _skb);
		}
		tipc_bearer_bc_xmit(net, bearer_id, &_xmitq);
	}
	__skb_queue_purge(xmitq);
	__skb_queue_purge(&_xmitq);
}

/* tipc_bcast_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
 * Consumes the buffer chain, except when returning -ELINKCONG
 * Returns 0 if success, otherwise errno: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE
 */
int tipc_bcast_xmit(struct net *net, struct sk_buff_head *list)
{
	struct tipc_link *l = tipc_bc_sndlink(net);
	struct sk_buff_head xmitq, inputq, rcvq;
	int rc = 0;

	__skb_queue_head_init(&rcvq);
	__skb_queue_head_init(&xmitq);
	skb_queue_head_init(&inputq);

	/* Prepare message clone for local node */
	if (unlikely(!tipc_msg_reassemble(list, &rcvq)))
		return -EHOSTUNREACH;

	tipc_bcast_lock(net);
	if (tipc_link_bc_peers(l))
		rc = tipc_link_xmit(l, list, &xmitq);
	tipc_bcast_unlock(net);

	/* Don't send to local node if adding to link failed */
	if (unlikely(rc)) {
		__skb_queue_purge(&rcvq);
		return rc;
	}

	/* Broadcast to all nodes, inluding local node */
	tipc_bcbase_xmit(net, &xmitq);
	tipc_sk_mcast_rcv(net, &rcvq, &inputq);
	__skb_queue_purge(list);
	return 0;
}

/* tipc_bcast_rcv - receive a broadcast packet, and deliver to rcv link
 *
 * RCU is locked, no other locks set
 */
int tipc_bcast_rcv(struct net *net, struct tipc_link *l, struct sk_buff *skb)
{
	struct tipc_msg *hdr = buf_msg(skb);
	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	struct sk_buff_head xmitq;
	int rc;

	__skb_queue_head_init(&xmitq);

	if (msg_mc_netid(hdr) != tipc_netid(net) || !tipc_link_is_up(l)) {
		kfree_skb(skb);
		return 0;
	}

	tipc_bcast_lock(net);
	if (msg_user(hdr) == BCAST_PROTOCOL)
		rc = tipc_link_bc_nack_rcv(l, skb, &xmitq);
	else
		rc = tipc_link_rcv(l, skb, NULL);
	tipc_bcast_unlock(net);

	tipc_bcbase_xmit(net, &xmitq);

	/* Any socket wakeup messages ? */
	if (!skb_queue_empty(inputq))
		tipc_sk_rcv(net, inputq);

	return rc;
}

/* tipc_bcast_ack_rcv - receive and handle a broadcast acknowledge
 *
 * RCU is locked, no other locks set
 */
void tipc_bcast_ack_rcv(struct net *net, struct tipc_link *l, u32 acked)
{
	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	struct sk_buff_head xmitq;

	__skb_queue_head_init(&xmitq);

	tipc_bcast_lock(net);
	tipc_link_bc_ack_rcv(l, acked, &xmitq);
	tipc_bcast_unlock(net);

	tipc_bcbase_xmit(net, &xmitq);

	/* Any socket wakeup messages ? */
	if (!skb_queue_empty(inputq))
		tipc_sk_rcv(net, inputq);
}

/* tipc_bcast_synch_rcv -  check and update rcv link with peer's send state
 *
 * RCU is locked, no other locks set
 */
void tipc_bcast_sync_rcv(struct net *net, struct tipc_link *l,
			 struct tipc_msg *hdr)
{
	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	struct sk_buff_head xmitq;

	__skb_queue_head_init(&xmitq);

	tipc_bcast_lock(net);
	if (msg_type(hdr) == STATE_MSG) {
		tipc_link_bc_ack_rcv(l, msg_bcast_ack(hdr), &xmitq);
		tipc_link_bc_sync_rcv(l, hdr, &xmitq);
	} else {
		tipc_link_bc_init_rcv(l, hdr);
	}
	tipc_bcast_unlock(net);

	tipc_bcbase_xmit(net, &xmitq);

	/* Any socket wakeup messages ? */
	if (!skb_queue_empty(inputq))
		tipc_sk_rcv(net, inputq);
}

/* tipc_bcast_add_peer - add a peer node to broadcast link and bearer
 *
 * RCU is locked, node lock is set
 */
void tipc_bcast_add_peer(struct net *net, struct tipc_link *uc_l,
			 struct sk_buff_head *xmitq)
{
	struct tipc_link *snd_l = tipc_bc_sndlink(net);

	tipc_bcast_lock(net);
	tipc_link_add_bc_peer(snd_l, uc_l, xmitq);
	tipc_bcbase_select_primary(net);
	tipc_bcast_unlock(net);
}

/* tipc_bcast_remove_peer - remove a peer node from broadcast link and bearer
 *
 * RCU is locked, node lock is set
 */
void tipc_bcast_remove_peer(struct net *net, struct tipc_link *rcv_l)
{
	struct tipc_link *snd_l = tipc_bc_sndlink(net);
	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	struct sk_buff_head xmitq;

	__skb_queue_head_init(&xmitq);

	tipc_bcast_lock(net);
	tipc_link_remove_bc_peer(snd_l, rcv_l, &xmitq);
	tipc_bcbase_select_primary(net);
	tipc_bcast_unlock(net);

	tipc_bcbase_xmit(net, &xmitq);

	/* Any socket wakeup messages ? */
	if (!skb_queue_empty(inputq))
		tipc_sk_rcv(net, inputq);
}

int tipc_bclink_reset_stats(struct net *net)
{
	struct tipc_link *l = tipc_bc_sndlink(net);

	if (!l)
		return -ENOPROTOOPT;

	tipc_bcast_lock(net);
	tipc_link_reset_stats(l);
	tipc_bcast_unlock(net);
	return 0;
}

static int tipc_bc_link_set_queue_limits(struct net *net, u32 limit)
{
	struct tipc_link *l = tipc_bc_sndlink(net);

	if (!l)
		return -ENOPROTOOPT;
	if (limit < BCLINK_WIN_MIN)
		limit = BCLINK_WIN_MIN;
	if (limit > TIPC_MAX_LINK_WIN)
		return -EINVAL;
	tipc_bcast_lock(net);
	tipc_link_set_queue_limits(l, limit);
	tipc_bcast_unlock(net);
	return 0;
}

int tipc_nl_bc_link_set(struct net *net, struct nlattr *attrs[])
{
	int err;
	u32 win;
	struct nlattr *props[TIPC_NLA_PROP_MAX + 1];

	if (!attrs[TIPC_NLA_LINK_PROP])
		return -EINVAL;

	err = tipc_nl_parse_link_prop(attrs[TIPC_NLA_LINK_PROP], props);
	if (err)
		return err;

	if (!props[TIPC_NLA_PROP_WIN])
		return -EOPNOTSUPP;

	win = nla_get_u32(props[TIPC_NLA_PROP_WIN]);

	return tipc_bc_link_set_queue_limits(net, win);
}

int tipc_bcast_init(struct net *net)
{
	struct tipc_net *tn = tipc_net(net);
	struct tipc_bc_base *bb = NULL;
	struct tipc_link *l = NULL;

	bb = kzalloc(sizeof(*bb), GFP_ATOMIC);
	if (!bb)
		goto enomem;
	tn->bcbase = bb;
	spin_lock_init(&tipc_net(net)->bclock);

	if (!tipc_link_bc_create(net, 0, 0,
				 U16_MAX,
				 BCLINK_WIN_DEFAULT,
				 0,
				 &bb->inputq,
				 NULL,
				 NULL,
				 &l))
		goto enomem;
	bb->link = l;
	tn->bcl = l;
	return 0;
enomem:
	kfree(bb);
	kfree(l);
	return -ENOMEM;
}

void tipc_bcast_stop(struct net *net)
{
	struct tipc_net *tn = net_generic(net, tipc_net_id);

	synchronize_net();
	kfree(tn->bcbase);
	kfree(tn->bcl);
}
Commit	Line	Data
b97bf3fd PL	1	/*
b97bf3fd PL	2	* net/tipc/bcast.c: TIPC broadcast code
c4307285	3	*
3c724acd	4	* Copyright (c) 2004-2006, 2014-2015, Ericsson AB
b97bf3fd	5	* Copyright (c) 2004, Intel Corporation.
2d627b92	6	* Copyright (c) 2005, 2010-2011, Wind River Systems
b97bf3fd PL	7	* All rights reserved.
b97bf3fd PL	8	*
9ea1fd3c	9	* Redistribution and use in source and binary forms, with or without
b97bf3fd PL	10	* modification, are permitted provided that the following conditions are met:
b97bf3fd PL	11	*
9ea1fd3c PL	12	* 1. Redistributions of source code must retain the above copyright
	13	* notice, this list of conditions and the following disclaimer.
	14	* 2. Redistributions in binary form must reproduce the above copyright
	15	* notice, this list of conditions and the following disclaimer in the
	16	* documentation and/or other materials provided with the distribution.
	17	* 3. Neither the names of the copyright holders nor the names of its
	18	* contributors may be used to endorse or promote products derived from
	19	* this software without specific prior written permission.
b97bf3fd	20	*
9ea1fd3c PL	21	* Alternatively, this software may be distributed under the terms of the
	22	* GNU General Public License ("GPL") version 2 as published by the Free
	23	* Software Foundation.
	24	*
	25	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
	26	* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
	27	* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
	28	* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
	29	* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
	30	* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
	31	* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
	32	* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
	33	* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
	34	* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
b97bf3fd PL	35	* POSSIBILITY OF SUCH DAMAGE.
	36	*/
	37
6beb19a6	38	#include <linux/tipc_config.h>
078bec82 JPM	39	#include "socket.h"
078bec82 JPM	40	#include "msg.h"
b97bf3fd	41	#include "bcast.h"
9f6bdcd4	42	#include "name_distr.h"
6beb19a6 JPM	43	#include "link.h"
6beb19a6 JPM	44	#include "node.h"
b97bf3fd	45
53387c4e JPM	46	#define BCLINK_WIN_DEFAULT 50 /* bcast link window size (default) */
53387c4e JPM	47	#define BCLINK_WIN_MIN 32 /* bcast minimum link window size */
b97bf3fd	48
3aec9cc9	49	const char tipc_bclink_name[] = "broadcast-link";
b97bf3fd	50
6beb19a6	51	/**
2af5ae37	52	* struct tipc_bc_base - base structure for keeping broadcast send state
b06b281e	53	* @link: broadcast send link structure
2af5ae37 JPM	54	* @inputq: data input queue; will only carry SOCK_WAKEUP messages
	55	* @dest: array keeping number of reachable destinations per bearer
	56	* @primary_bearer: a bearer having links to all broadcast destinations, if any
6beb19a6 JPM	57	*/
6beb19a6 JPM	58	struct tipc_bc_base {
32301906	59	struct tipc_link *link;
6beb19a6	60	struct sk_buff_head inputq;
b06b281e JPM	61	int dests[MAX_BEARERS];
b06b281e JPM	62	int primary_bearer;
6beb19a6 JPM	63	};
6beb19a6 JPM	64
5fd9fd63 JPM	65	static struct tipc_bc_base tipc_bc_base(struct net net)
	66	{
	67	return tipc_net(net)->bcbase;
	68	}
	69
959e1781	70	int tipc_bcast_get_mtu(struct net *net)
078bec82	71	{
959e1781	72	return tipc_link_mtu(tipc_bc_sndlink(net));
078bec82 JPM	73	}
078bec82 JPM	74
b06b281e JPM	75	/* tipc_bcbase_select_primary(): find a bearer with links to all destinations,
	76	* if any, and make it primary bearer
	77	*/
	78	static void tipc_bcbase_select_primary(struct net *net)
	79	{
	80	struct tipc_bc_base *bb = tipc_bc_base(net);
	81	int all_dests = tipc_link_bc_peers(bb->link);
959e1781	82	int i, mtu;
b06b281e JPM	83
	84	bb->primary_bearer = INVALID_BEARER_ID;
	85
	86	if (!all_dests)
	87	return;
	88
	89	for (i = 0; i < MAX_BEARERS; i++) {
959e1781 JPM	90	if (!bb->dests[i])
	91	continue;
	92
	93	mtu = tipc_bearer_mtu(net, i);
	94	if (mtu < tipc_link_mtu(bb->link))
	95	tipc_link_set_mtu(bb->link, mtu);
	96
b06b281e JPM	97	if (bb->dests[i] < all_dests)
	98	continue;
	99
	100	bb->primary_bearer = i;
	101
	102	/* Reduce risk that all nodes select same primary */
	103	if ((i ^ tipc_own_addr(net)) & 1)
	104	break;
	105	}
	106	}
	107
	108	void tipc_bcast_inc_bearer_dst_cnt(struct net *net, int bearer_id)
	109	{
	110	struct tipc_bc_base *bb = tipc_bc_base(net);
	111
	112	tipc_bcast_lock(net);
	113	bb->dests[bearer_id]++;
	114	tipc_bcbase_select_primary(net);
	115	tipc_bcast_unlock(net);
	116	}
	117
	118	void tipc_bcast_dec_bearer_dst_cnt(struct net *net, int bearer_id)
	119	{
	120	struct tipc_bc_base *bb = tipc_bc_base(net);
	121
	122	tipc_bcast_lock(net);
	123	bb->dests[bearer_id]--;
	124	tipc_bcbase_select_primary(net);
	125	tipc_bcast_unlock(net);
	126	}
	127
b06b281e JPM	128	/* tipc_bcbase_xmit - broadcast a packet queue across one or more bearers
	129	*
	130	* Note that number of reachable destinations, as indicated in the dests[]
	131	* array, may transitionally differ from the number of destinations indicated
	132	* in each sent buffer. We can sustain this. Excess destination nodes will
	133	* drop and never acknowledge the unexpected packets, and missing destinations
	134	* will either require retransmission (if they are just about to be added to
	135	* the bearer), or be removed from the buffer's 'ackers' counter (if they
	136	* just went down)
	137	*/
	138	static void tipc_bcbase_xmit(struct net net, struct sk_buff_head xmitq)
	139	{
	140	int bearer_id;
	141	struct tipc_bc_base *bb = tipc_bc_base(net);
	142	struct sk_buff skb, _skb;
	143	struct sk_buff_head _xmitq;
	144
	145	if (skb_queue_empty(xmitq))
	146	return;
	147
	148	/* The typical case: at least one bearer has links to all nodes */
	149	bearer_id = bb->primary_bearer;
	150	if (bearer_id >= 0) {
	151	tipc_bearer_bc_xmit(net, bearer_id, xmitq);
	152	return;
	153	}
	154
	155	/* We have to transmit across all bearers */
	156	skb_queue_head_init(&_xmitq);
	157	for (bearer_id = 0; bearer_id < MAX_BEARERS; bearer_id++) {
	158	if (!bb->dests[bearer_id])
	159	continue;
	160
	161	skb_queue_walk(xmitq, skb) {
	162	_skb = pskb_copy_for_clone(skb, GFP_ATOMIC);
	163	if (!_skb)
	164	break;
	165	__skb_queue_tail(&_xmitq, _skb);
	166	}
	167	tipc_bearer_bc_xmit(net, bearer_id, &_xmitq);
	168	}
	169	__skb_queue_purge(xmitq);
	170	__skb_queue_purge(&_xmitq);
	171	}
	172
6beb19a6	173	/* tipc_bcast_xmit - deliver buffer chain to all nodes in cluster
9fbfb8b1	174	* and to identified node local sockets
f2f9800d	175	* @net: the applicable net namespace
a6ca1094	176	* @list: chain of buffers containing message
078bec82 JPM	177	* Consumes the buffer chain, except when returning -ELINKCONG
	178	* Returns 0 if success, otherwise errno: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE
	179	*/
6beb19a6	180	int tipc_bcast_xmit(struct net net, struct sk_buff_head list)
078bec82	181	{
2f566124 JPM	182	struct tipc_link *l = tipc_bc_sndlink(net);
2f566124 JPM	183	struct sk_buff_head xmitq, inputq, rcvq;
078bec82	184	int rc = 0;
078bec82	185
2f566124 JPM	186	__skb_queue_head_init(&rcvq);
	187	__skb_queue_head_init(&xmitq);
	188	skb_queue_head_init(&inputq);
22d85c79	189
2f566124 JPM	190	/* Prepare message clone for local node */
	191	if (unlikely(!tipc_msg_reassemble(list, &rcvq)))
	192	return -EHOSTUNREACH;
078bec82	193
2f566124 JPM	194	tipc_bcast_lock(net);
	195	if (tipc_link_bc_peers(l))
	196	rc = tipc_link_xmit(l, list, &xmitq);
2f566124	197	tipc_bcast_unlock(net);
078bec82	198
2f566124	199	/* Don't send to local node if adding to link failed */
cb1b7280	200	if (unlikely(rc)) {
2f566124	201	__skb_queue_purge(&rcvq);
cb1b7280 JPM	202	return rc;
cb1b7280 JPM	203	}
52666986	204
2f566124	205	/* Broadcast to all nodes, inluding local node */
b06b281e	206	tipc_bcbase_xmit(net, &xmitq);
2f566124 JPM	207	tipc_sk_mcast_rcv(net, &rcvq, &inputq);
	208	__skb_queue_purge(list);
	209	return 0;
078bec82	210	}
52666986 JPM	211
	212	/* tipc_bcast_rcv - receive a broadcast packet, and deliver to rcv link
	213	*
	214	* RCU is locked, no other locks set
	215	*/
	216	int tipc_bcast_rcv(struct net net, struct tipc_link l, struct sk_buff *skb)
	217	{
	218	struct tipc_msg *hdr = buf_msg(skb);
	219	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	220	struct sk_buff_head xmitq;
	221	int rc;
	222
	223	__skb_queue_head_init(&xmitq);
	224
	225	if (msg_mc_netid(hdr) != tipc_netid(net) \|\| !tipc_link_is_up(l)) {
	226	kfree_skb(skb);
	227	return 0;
	228	}
	229
	230	tipc_bcast_lock(net);
	231	if (msg_user(hdr) == BCAST_PROTOCOL)
	232	rc = tipc_link_bc_nack_rcv(l, skb, &xmitq);
	233	else
	234	rc = tipc_link_rcv(l, skb, NULL);
	235	tipc_bcast_unlock(net);
	236
b06b281e	237	tipc_bcbase_xmit(net, &xmitq);
52666986 JPM	238
	239	/* Any socket wakeup messages ? */
	240	if (!skb_queue_empty(inputq))
	241	tipc_sk_rcv(net, inputq);
	242
	243	return rc;
	244	}
	245
	246	/* tipc_bcast_ack_rcv - receive and handle a broadcast acknowledge
	247	*
	248	* RCU is locked, no other locks set
	249	*/
	250	void tipc_bcast_ack_rcv(struct net net, struct tipc_link l, u32 acked)
	251	{
	252	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	253	struct sk_buff_head xmitq;
	254
	255	__skb_queue_head_init(&xmitq);
	256
	257	tipc_bcast_lock(net);
	258	tipc_link_bc_ack_rcv(l, acked, &xmitq);
	259	tipc_bcast_unlock(net);
	260
b06b281e	261	tipc_bcbase_xmit(net, &xmitq);
52666986 JPM	262
	263	/* Any socket wakeup messages ? */
	264	if (!skb_queue_empty(inputq))
	265	tipc_sk_rcv(net, inputq);
	266	}
	267
	268	/* tipc_bcast_synch_rcv - check and update rcv link with peer's send state
	269	*
	270	* RCU is locked, no other locks set
	271	*/
	272	void tipc_bcast_sync_rcv(struct net net, struct tipc_link l,
	273	struct tipc_msg *hdr)
	274	{
	275	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	276	struct sk_buff_head xmitq;
	277
	278	__skb_queue_head_init(&xmitq);
	279
	280	tipc_bcast_lock(net);
	281	if (msg_type(hdr) == STATE_MSG) {
	282	tipc_link_bc_ack_rcv(l, msg_bcast_ack(hdr), &xmitq);
	283	tipc_link_bc_sync_rcv(l, hdr, &xmitq);
	284	} else {
	285	tipc_link_bc_init_rcv(l, hdr);
	286	}
	287	tipc_bcast_unlock(net);
	288
b06b281e	289	tipc_bcbase_xmit(net, &xmitq);
52666986 JPM	290
	291	/* Any socket wakeup messages ? */
	292	if (!skb_queue_empty(inputq))
	293	tipc_sk_rcv(net, inputq);
	294	}
	295
	296	/* tipc_bcast_add_peer - add a peer node to broadcast link and bearer
	297	*
	298	* RCU is locked, node lock is set
	299	*/
b06b281e	300	void tipc_bcast_add_peer(struct net net, struct tipc_link uc_l,
52666986 JPM	301	struct sk_buff_head *xmitq)
52666986 JPM	302	{
52666986 JPM	303	struct tipc_link *snd_l = tipc_bc_sndlink(net);
52666986 JPM	304
b06b281e	305	tipc_bcast_lock(net);
52666986	306	tipc_link_add_bc_peer(snd_l, uc_l, xmitq);
b06b281e JPM	307	tipc_bcbase_select_primary(net);
b06b281e JPM	308	tipc_bcast_unlock(net);
52666986 JPM	309	}
	310
	311	/* tipc_bcast_remove_peer - remove a peer node from broadcast link and bearer
	312	*
	313	* RCU is locked, node lock is set
	314	*/
b06b281e	315	void tipc_bcast_remove_peer(struct net net, struct tipc_link rcv_l)
52666986	316	{
52666986	317	struct tipc_link *snd_l = tipc_bc_sndlink(net);
b06b281e	318	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
52666986 JPM	319	struct sk_buff_head xmitq;
	320
	321	__skb_queue_head_init(&xmitq);
	322
b06b281e	323	tipc_bcast_lock(net);
52666986	324	tipc_link_remove_bc_peer(snd_l, rcv_l, &xmitq);
b06b281e JPM	325	tipc_bcbase_select_primary(net);
b06b281e JPM	326	tipc_bcast_unlock(net);
52666986	327
b06b281e	328	tipc_bcbase_xmit(net, &xmitq);
52666986 JPM	329
	330	/* Any socket wakeup messages ? */
	331	if (!skb_queue_empty(inputq))
	332	tipc_sk_rcv(net, inputq);
	333	}
	334
1da46568	335	int tipc_bclink_reset_stats(struct net *net)
b97bf3fd	336	{
38206d59	337	struct tipc_link *l = tipc_bc_sndlink(net);
1da46568	338
38206d59	339	if (!l)
b97bf3fd PL	340	return -ENOPROTOOPT;
b97bf3fd PL	341
2af5ae37	342	tipc_bcast_lock(net);
38206d59	343	tipc_link_reset_stats(l);
2af5ae37	344	tipc_bcast_unlock(net);
0e35fd5e	345	return 0;
b97bf3fd PL	346	}
b97bf3fd PL	347
2af5ae37	348	static int tipc_bc_link_set_queue_limits(struct net *net, u32 limit)
b97bf3fd	349	{
2af5ae37	350	struct tipc_link *l = tipc_bc_sndlink(net);
1da46568	351
2af5ae37	352	if (!l)
b97bf3fd	353	return -ENOPROTOOPT;
53387c4e JPM	354	if (limit < BCLINK_WIN_MIN)
	355	limit = BCLINK_WIN_MIN;
	356	if (limit > TIPC_MAX_LINK_WIN)
b97bf3fd	357	return -EINVAL;
2af5ae37 JPM	358	tipc_bcast_lock(net);
	359	tipc_link_set_queue_limits(l, limit);
	360	tipc_bcast_unlock(net);
0e35fd5e	361	return 0;
b97bf3fd PL	362	}
b97bf3fd PL	363
670f4f88 RA	364	int tipc_nl_bc_link_set(struct net net, struct nlattr attrs[])
	365	{
	366	int err;
	367	u32 win;
	368	struct nlattr *props[TIPC_NLA_PROP_MAX + 1];
	369
	370	if (!attrs[TIPC_NLA_LINK_PROP])
	371	return -EINVAL;
	372
	373	err = tipc_nl_parse_link_prop(attrs[TIPC_NLA_LINK_PROP], props);
	374	if (err)
	375	return err;
	376
	377	if (!props[TIPC_NLA_PROP_WIN])
	378	return -EOPNOTSUPP;
	379
	380	win = nla_get_u32(props[TIPC_NLA_PROP_WIN]);
	381
2af5ae37	382	return tipc_bc_link_set_queue_limits(net, win);
670f4f88 RA	383	}
670f4f88 RA	384
6beb19a6	385	int tipc_bcast_init(struct net *net)
b97bf3fd	386	{
32301906	387	struct tipc_net *tn = tipc_net(net);
32301906 JPM	388	struct tipc_bc_base *bb = NULL;
	389	struct tipc_link *l = NULL;
	390
32301906 JPM	391	bb = kzalloc(sizeof(*bb), GFP_ATOMIC);
	392	if (!bb)
	393	goto enomem;
	394	tn->bcbase = bb;
0043550b	395	spin_lock_init(&tipc_net(net)->bclock);
32301906	396
c72fa872	397	if (!tipc_link_bc_create(net, 0, 0,
959e1781	398	U16_MAX,
32301906	399	BCLINK_WIN_DEFAULT,
fd556f20	400	0,
32301906	401	&bb->inputq,
2af5ae37	402	NULL,
52666986	403	NULL,
32301906 JPM	404	&l))
	405	goto enomem;
	406	bb->link = l;
	407	tn->bcl = l;
eb8b00f5	408	return 0;
32301906	409	enomem:
32301906 JPM	410	kfree(bb);
	411	kfree(l);
	412	return -ENOMEM;
b97bf3fd PL	413	}
b97bf3fd PL	414
6beb19a6	415	void tipc_bcast_stop(struct net *net)
b97bf3fd	416	{
7f9f95d9 YX	417	struct tipc_net *tn = net_generic(net, tipc_net_id);
7f9f95d9 YX	418
eb8b00f5	419	synchronize_net();
6beb19a6	420	kfree(tn->bcbase);
32301906	421	kfree(tn->bcl);
b97bf3fd	422	}