[thirdparty/kernel/stable.git] / net / tipc / group.c

/*
 * net/tipc/group.c: TIPC group messaging code
 *
 * Copyright (c) 2017, Ericsson AB
 * 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 "core.h"
#include "addr.h"
#include "group.h"
#include "bcast.h"
#include "server.h"
#include "msg.h"
#include "socket.h"
#include "node.h"
#include "name_table.h"
#include "subscr.h"

#define ADV_UNIT (((MAX_MSG_SIZE + MAX_H_SIZE) / FLOWCTL_BLK_SZ) + 1)
#define ADV_IDLE ADV_UNIT

enum mbr_state {
	MBR_QUARANTINED,
	MBR_DISCOVERED,
	MBR_JOINING,
	MBR_PUBLISHED,
	MBR_JOINED,
	MBR_LEAVING
};

struct tipc_member {
	struct rb_node tree_node;
	struct list_head list;
	struct sk_buff *event_msg;
	u32 node;
	u32 port;
	u32 instance;
	enum mbr_state state;
	u16 bc_rcv_nxt;
};

struct tipc_group {
	struct rb_root members;
	struct tipc_nlist dests;
	struct net *net;
	int subid;
	u32 type;
	u32 instance;
	u32 domain;
	u32 scope;
	u32 portid;
	u16 member_cnt;
	u16 bc_snd_nxt;
	bool loopback;
	bool events;
};

static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,
				  int mtyp, struct sk_buff_head *xmitq);

u16 tipc_group_bc_snd_nxt(struct tipc_group *grp)
{
	return grp->bc_snd_nxt;
}

static bool tipc_group_is_receiver(struct tipc_member *m)
{
	return m && m->state >= MBR_JOINED;
}

int tipc_group_size(struct tipc_group *grp)
{
	return grp->member_cnt;
}

struct tipc_group *tipc_group_create(struct net *net, u32 portid,
				     struct tipc_group_req *mreq)
{
	struct tipc_group *grp;
	u32 type = mreq->type;

	grp = kzalloc(sizeof(*grp), GFP_ATOMIC);
	if (!grp)
		return NULL;
	tipc_nlist_init(&grp->dests, tipc_own_addr(net));
	grp->members = RB_ROOT;
	grp->net = net;
	grp->portid = portid;
	grp->domain = addr_domain(net, mreq->scope);
	grp->type = type;
	grp->instance = mreq->instance;
	grp->scope = mreq->scope;
	grp->loopback = mreq->flags & TIPC_GROUP_LOOPBACK;
	grp->events = mreq->flags & TIPC_GROUP_MEMBER_EVTS;
	if (tipc_topsrv_kern_subscr(net, portid, type, 0, ~0, &grp->subid))
		return grp;
	kfree(grp);
	return NULL;
}

void tipc_group_delete(struct net *net, struct tipc_group *grp)
{
	struct rb_root *tree = &grp->members;
	struct tipc_member *m, *tmp;
	struct sk_buff_head xmitq;

	__skb_queue_head_init(&xmitq);

	rbtree_postorder_for_each_entry_safe(m, tmp, tree, tree_node) {
		tipc_group_proto_xmit(grp, m, GRP_LEAVE_MSG, &xmitq);
		list_del(&m->list);
		kfree(m);
	}
	tipc_node_distr_xmit(net, &xmitq);
	tipc_nlist_purge(&grp->dests);
	tipc_topsrv_kern_unsubscr(net, grp->subid);
	kfree(grp);
}

struct tipc_member *tipc_group_find_member(struct tipc_group *grp,
					   u32 node, u32 port)
{
	struct rb_node *n = grp->members.rb_node;
	u64 nkey, key = (u64)node << 32 | port;
	struct tipc_member *m;

	while (n) {
		m = container_of(n, struct tipc_member, tree_node);
		nkey = (u64)m->node << 32 | m->port;
		if (key < nkey)
			n = n->rb_left;
		else if (key > nkey)
			n = n->rb_right;
		else
			return m;
	}
	return NULL;
}

static struct tipc_member *tipc_group_find_node(struct tipc_group *grp,
						u32 node)
{
	struct tipc_member *m;
	struct rb_node *n;

	for (n = rb_first(&grp->members); n; n = rb_next(n)) {
		m = container_of(n, struct tipc_member, tree_node);
		if (m->node == node)
			return m;
	}
	return NULL;
}

static void tipc_group_add_to_tree(struct tipc_group *grp,
				   struct tipc_member *m)
{
	u64 nkey, key = (u64)m->node << 32 | m->port;
	struct rb_node **n, *parent = NULL;
	struct tipc_member *tmp;

	n = &grp->members.rb_node;
	while (*n) {
		tmp = container_of(*n, struct tipc_member, tree_node);
		parent = *n;
		tmp = container_of(parent, struct tipc_member, tree_node);
		nkey = (u64)tmp->node << 32 | tmp->port;
		if (key < nkey)
			n = &(*n)->rb_left;
		else if (key > nkey)
			n = &(*n)->rb_right;
		else
			return;
	}
	rb_link_node(&m->tree_node, parent, n);
	rb_insert_color(&m->tree_node, &grp->members);
}

static struct tipc_member *tipc_group_create_member(struct tipc_group *grp,
						    u32 node, u32 port,
						    int state)
{
	struct tipc_member *m;

	m = kzalloc(sizeof(*m), GFP_ATOMIC);
	if (!m)
		return NULL;
	INIT_LIST_HEAD(&m->list);
	m->node = node;
	m->port = port;
	grp->member_cnt++;
	tipc_group_add_to_tree(grp, m);
	tipc_nlist_add(&grp->dests, m->node);
	m->state = state;
	return m;
}

void tipc_group_add_member(struct tipc_group *grp, u32 node, u32 port)
{
	tipc_group_create_member(grp, node, port, MBR_DISCOVERED);
}

static void tipc_group_delete_member(struct tipc_group *grp,
				     struct tipc_member *m)
{
	rb_erase(&m->tree_node, &grp->members);
	grp->member_cnt--;
	list_del_init(&m->list);

	/* If last member on a node, remove node from dest list */
	if (!tipc_group_find_node(grp, m->node))
		tipc_nlist_del(&grp->dests, m->node);

	kfree(m);
}

struct tipc_nlist *tipc_group_dests(struct tipc_group *grp)
{
	return &grp->dests;
}

void tipc_group_self(struct tipc_group *grp, struct tipc_name_seq *seq,
		     int *scope)
{
	seq->type = grp->type;
	seq->lower = grp->instance;
	seq->upper = grp->instance;
	*scope = grp->scope;
}

void tipc_group_update_bc_members(struct tipc_group *grp)
{
	grp->bc_snd_nxt++;
}

/* tipc_group_filter_msg() - determine if we should accept arriving message
 */
void tipc_group_filter_msg(struct tipc_group *grp, struct sk_buff_head *inputq,
			   struct sk_buff_head *xmitq)
{
	struct sk_buff *skb = __skb_dequeue(inputq);
	struct tipc_member *m;
	struct tipc_msg *hdr;
	u32 node, port;
	int mtyp;

	if (!skb)
		return;

	hdr = buf_msg(skb);
	mtyp = msg_type(hdr);
	node =  msg_orignode(hdr);
	port = msg_origport(hdr);

	if (!msg_in_group(hdr))
		goto drop;

	if (mtyp == TIPC_GRP_MEMBER_EVT) {
		if (!grp->events)
			goto drop;
		__skb_queue_tail(inputq, skb);
		return;
	}

	m = tipc_group_find_member(grp, node, port);
	if (!tipc_group_is_receiver(m))
		goto drop;

	TIPC_SKB_CB(skb)->orig_member = m->instance;
	__skb_queue_tail(inputq, skb);

	m->bc_rcv_nxt = msg_grp_bc_seqno(hdr) + 1;
	return;
drop:
	kfree_skb(skb);
}

static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,
				  int mtyp, struct sk_buff_head *xmitq)
{
	struct tipc_msg *hdr;
	struct sk_buff *skb;

	skb = tipc_msg_create(GROUP_PROTOCOL, mtyp, INT_H_SIZE, 0,
			      m->node, tipc_own_addr(grp->net),
			      m->port, grp->portid, 0);
	if (!skb)
		return;

	hdr = buf_msg(skb);
	if (mtyp == GRP_JOIN_MSG)
		msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
	__skb_queue_tail(xmitq, skb);
}

void tipc_group_proto_rcv(struct tipc_group *grp, struct tipc_msg *hdr,
			  struct sk_buff_head *inputq,
			  struct sk_buff_head *xmitq)
{
	u32 node = msg_orignode(hdr);
	u32 port = msg_origport(hdr);
	struct tipc_member *m;

	if (!grp)
		return;

	m = tipc_group_find_member(grp, node, port);

	switch (msg_type(hdr)) {
	case GRP_JOIN_MSG:
		if (!m)
			m = tipc_group_create_member(grp, node, port,
						     MBR_QUARANTINED);
		if (!m)
			return;
		m->bc_rcv_nxt = msg_grp_bc_syncpt(hdr);

		/* Wait until PUBLISH event is received */
		if (m->state == MBR_DISCOVERED) {
			m->state = MBR_JOINING;
		} else if (m->state == MBR_PUBLISHED) {
			m->state = MBR_JOINED;
			__skb_queue_tail(inputq, m->event_msg);
		}
		return;
	case GRP_LEAVE_MSG:
		if (!m)
			return;

		/* Wait until WITHDRAW event is received */
		if (m->state != MBR_LEAVING) {
			m->state = MBR_LEAVING;
			return;
		}
		/* Otherwise deliver already received WITHDRAW event */
		__skb_queue_tail(inputq, m->event_msg);
		tipc_group_delete_member(grp, m);
		return;
	default:
		pr_warn("Received unknown GROUP_PROTO message\n");
	}
}

void tipc_group_member_evt(struct tipc_group *grp,
			   struct sk_buff *skb,
			   struct sk_buff_head *inputq,
			   struct sk_buff_head *xmitq)
{
	struct tipc_msg *hdr = buf_msg(skb);
	struct tipc_event *evt = (void *)msg_data(hdr);
	u32 instance = evt->found_lower;
	u32 node = evt->port.node;
	u32 port = evt->port.ref;
	int event = evt->event;
	struct tipc_member *m;
	struct net *net;
	u32 self;

	if (!grp)
		goto drop;

	net = grp->net;
	self = tipc_own_addr(net);
	if (!grp->loopback && node == self && port == grp->portid)
		goto drop;

	/* Convert message before delivery to user */
	msg_set_hdr_sz(hdr, GROUP_H_SIZE);
	msg_set_user(hdr, TIPC_CRITICAL_IMPORTANCE);
	msg_set_type(hdr, TIPC_GRP_MEMBER_EVT);
	msg_set_origport(hdr, port);
	msg_set_orignode(hdr, node);
	msg_set_nametype(hdr, grp->type);
	msg_set_grp_evt(hdr, event);

	m = tipc_group_find_member(grp, node, port);

	if (event == TIPC_PUBLISHED) {
		if (!m)
			m = tipc_group_create_member(grp, node, port,
						     MBR_DISCOVERED);
		if (!m)
			goto drop;

		/* Hold back event if JOIN message not yet received */
		if (m->state == MBR_DISCOVERED) {
			m->event_msg = skb;
			m->state = MBR_PUBLISHED;
		} else {
			__skb_queue_tail(inputq, skb);
			m->state = MBR_JOINED;
		}
		m->instance = instance;
		TIPC_SKB_CB(skb)->orig_member = m->instance;
		tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
	} else if (event == TIPC_WITHDRAWN) {
		if (!m)
			goto drop;

		TIPC_SKB_CB(skb)->orig_member = m->instance;

		/* Hold back event if more messages might be expected */
		if (m->state != MBR_LEAVING && tipc_node_is_up(net, node)) {
			m->event_msg = skb;
			m->state = MBR_LEAVING;
		} else {
			__skb_queue_tail(inputq, skb);
			tipc_group_delete_member(grp, m);
		}
	}
	return;
drop:
	kfree_skb(skb);
}
Commit	Line	Data
75da2163 JM	1	/*
	2	* net/tipc/group.c: TIPC group messaging code
	3	*
	4	* Copyright (c) 2017, Ericsson AB
	5	* All rights reserved.
	6	*
	7	* Redistribution and use in source and binary forms, with or without
	8	* modification, are permitted provided that the following conditions are met:
	9	*
	10	* 1. Redistributions of source code must retain the above copyright
	11	* notice, this list of conditions and the following disclaimer.
	12	* 2. Redistributions in binary form must reproduce the above copyright
	13	* notice, this list of conditions and the following disclaimer in the
	14	* documentation and/or other materials provided with the distribution.
	15	* 3. Neither the names of the copyright holders nor the names of its
	16	* contributors may be used to endorse or promote products derived from
	17	* this software without specific prior written permission.
	18	*
	19	* Alternatively, this software may be distributed under the terms of the
	20	* GNU General Public License ("GPL") version 2 as published by the Free
	21	* Software Foundation.
	22	*
	23	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
	24	* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
	25	* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
	26	* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
	27	* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
	28	* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
	29	* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
	30	* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
	31	* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
	32	* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
	33	* POSSIBILITY OF SUCH DAMAGE.
	34	*/
	35
	36	#include "core.h"
	37	#include "addr.h"
	38	#include "group.h"
	39	#include "bcast.h"
	40	#include "server.h"
	41	#include "msg.h"
	42	#include "socket.h"
	43	#include "node.h"
	44	#include "name_table.h"
	45	#include "subscr.h"
	46
	47	#define ADV_UNIT (((MAX_MSG_SIZE + MAX_H_SIZE) / FLOWCTL_BLK_SZ) + 1)
	48	#define ADV_IDLE ADV_UNIT
	49
	50	enum mbr_state {
	51	MBR_QUARANTINED,
	52	MBR_DISCOVERED,
	53	MBR_JOINING,
	54	MBR_PUBLISHED,
	55	MBR_JOINED,
	56	MBR_LEAVING
	57	};
	58
	59	struct tipc_member {
	60	struct rb_node tree_node;
	61	struct list_head list;
ae236fb2	62	struct sk_buff *event_msg;
75da2163 JM	63	u32 node;
75da2163 JM	64	u32 port;
31c82a2d	65	u32 instance;
75da2163 JM	66	enum mbr_state state;
	67	u16 bc_rcv_nxt;
	68	};
	69
	70	struct tipc_group {
	71	struct rb_root members;
	72	struct tipc_nlist dests;
	73	struct net *net;
	74	int subid;
	75	u32 type;
	76	u32 instance;
	77	u32 domain;
	78	u32 scope;
	79	u32 portid;
	80	u16 member_cnt;
	81	u16 bc_snd_nxt;
	82	bool loopback;
ae236fb2	83	bool events;
75da2163 JM	84	};
	85
	86	static void tipc_group_proto_xmit(struct tipc_group grp, struct tipc_member m,
	87	int mtyp, struct sk_buff_head *xmitq);
	88
	89	u16 tipc_group_bc_snd_nxt(struct tipc_group *grp)
	90	{
	91	return grp->bc_snd_nxt;
	92	}
	93
	94	static bool tipc_group_is_receiver(struct tipc_member *m)
	95	{
	96	return m && m->state >= MBR_JOINED;
	97	}
	98
	99	int tipc_group_size(struct tipc_group *grp)
	100	{
	101	return grp->member_cnt;
	102	}
	103
	104	struct tipc_group tipc_group_create(struct net net, u32 portid,
	105	struct tipc_group_req *mreq)
	106	{
	107	struct tipc_group *grp;
	108	u32 type = mreq->type;
	109
	110	grp = kzalloc(sizeof(*grp), GFP_ATOMIC);
	111	if (!grp)
	112	return NULL;
	113	tipc_nlist_init(&grp->dests, tipc_own_addr(net));
	114	grp->members = RB_ROOT;
	115	grp->net = net;
	116	grp->portid = portid;
	117	grp->domain = addr_domain(net, mreq->scope);
	118	grp->type = type;
	119	grp->instance = mreq->instance;
	120	grp->scope = mreq->scope;
	121	grp->loopback = mreq->flags & TIPC_GROUP_LOOPBACK;
ae236fb2	122	grp->events = mreq->flags & TIPC_GROUP_MEMBER_EVTS;
75da2163 JM	123	if (tipc_topsrv_kern_subscr(net, portid, type, 0, ~0, &grp->subid))
	124	return grp;
	125	kfree(grp);
	126	return NULL;
	127	}
	128
	129	void tipc_group_delete(struct net net, struct tipc_group grp)
	130	{
	131	struct rb_root *tree = &grp->members;
	132	struct tipc_member m, tmp;
	133	struct sk_buff_head xmitq;
	134
	135	__skb_queue_head_init(&xmitq);
	136
	137	rbtree_postorder_for_each_entry_safe(m, tmp, tree, tree_node) {
	138	tipc_group_proto_xmit(grp, m, GRP_LEAVE_MSG, &xmitq);
	139	list_del(&m->list);
	140	kfree(m);
	141	}
	142	tipc_node_distr_xmit(net, &xmitq);
	143	tipc_nlist_purge(&grp->dests);
	144	tipc_topsrv_kern_unsubscr(net, grp->subid);
	145	kfree(grp);
	146	}
	147
	148	struct tipc_member tipc_group_find_member(struct tipc_group grp,
	149	u32 node, u32 port)
	150	{
	151	struct rb_node *n = grp->members.rb_node;
	152	u64 nkey, key = (u64)node << 32 \| port;
	153	struct tipc_member *m;
	154
	155	while (n) {
	156	m = container_of(n, struct tipc_member, tree_node);
	157	nkey = (u64)m->node << 32 \| m->port;
	158	if (key < nkey)
	159	n = n->rb_left;
	160	else if (key > nkey)
	161	n = n->rb_right;
	162	else
	163	return m;
	164	}
	165	return NULL;
	166	}
	167
	168	static struct tipc_member tipc_group_find_node(struct tipc_group grp,
	169	u32 node)
	170	{
	171	struct tipc_member *m;
	172	struct rb_node *n;
	173
	174	for (n = rb_first(&grp->members); n; n = rb_next(n)) {
	175	m = container_of(n, struct tipc_member, tree_node);
	176	if (m->node == node)
	177	return m;
	178	}
	179	return NULL;
	180	}
	181
	182	static void tipc_group_add_to_tree(struct tipc_group *grp,
	183	struct tipc_member *m)
	184	{
	185	u64 nkey, key = (u64)m->node << 32 \| m->port;
	186	struct rb_node *n, parent = NULL;
187	struct tipc_member *tmp;
188
189	n = &grp->members.rb_node;
190	while (*n) {
191	tmp = container_of(*n, struct tipc_member, tree_node);
192	parent = *n;
193	tmp = container_of(parent, struct tipc_member, tree_node);
194	nkey = (u64)tmp->node << 32 \| tmp->port;
195	if (key < nkey)
196	n = &(*n)->rb_left;
197	else if (key > nkey)
198	n = &(*n)->rb_right;
199	else
200	return;
201	}
202	rb_link_node(&m->tree_node, parent, n);
203	rb_insert_color(&m->tree_node, &grp->members);
204	}
205
206	static struct tipc_member tipc_group_create_member(struct tipc_group grp,
207	u32 node, u32 port,
208	int state)
209	{
210	struct tipc_member *m;
211
212	m = kzalloc(sizeof(*m), GFP_ATOMIC);
213	if (!m)
214	return NULL;
215	INIT_LIST_HEAD(&m->list);
216	m->node = node;
217	m->port = port;
218	grp->member_cnt++;
219	tipc_group_add_to_tree(grp, m);
220	tipc_nlist_add(&grp->dests, m->node);
221	m->state = state;
222	return m;
223	}
224
225	void tipc_group_add_member(struct tipc_group *grp, u32 node, u32 port)
226	{
227	tipc_group_create_member(grp, node, port, MBR_DISCOVERED);
228	}
229
230	static void tipc_group_delete_member(struct tipc_group *grp,
231	struct tipc_member *m)
232	{
233	rb_erase(&m->tree_node, &grp->members);
234	grp->member_cnt--;
235	list_del_init(&m->list);
236
237	/* If last member on a node, remove node from dest list */
238	if (!tipc_group_find_node(grp, m->node))
239	tipc_nlist_del(&grp->dests, m->node);
240
241	kfree(m);
242	}
243
244	struct tipc_nlist tipc_group_dests(struct tipc_group grp)
245	{
246	return &grp->dests;
247	}
248
249	void tipc_group_self(struct tipc_group grp, struct tipc_name_seq seq,
250	int *scope)
251	{
252	seq->type = grp->type;
253	seq->lower = grp->instance;
254	seq->upper = grp->instance;
255	*scope = grp->scope;
256	}
257
258	void tipc_group_update_bc_members(struct tipc_group *grp)
259	{
260	grp->bc_snd_nxt++;
261	}
262
263	/* tipc_group_filter_msg() - determine if we should accept arriving message
264	*/
265	void tipc_group_filter_msg(struct tipc_group grp, struct sk_buff_head inputq,
266	struct sk_buff_head *xmitq)
267	{
268	struct sk_buff *skb = __skb_dequeue(inputq);
269	struct tipc_member *m;
270	struct tipc_msg *hdr;
271	u32 node, port;
272	int mtyp;
273
274	if (!skb)
275	return;
276
277	hdr = buf_msg(skb);
278	mtyp = msg_type(hdr);
279	node = msg_orignode(hdr);
280	port = msg_origport(hdr);
281
282	if (!msg_in_group(hdr))
283	goto drop;
284
ae236fb2 JM	285	if (mtyp == TIPC_GRP_MEMBER_EVT) {
	286	if (!grp->events)
	287	goto drop;
	288	__skb_queue_tail(inputq, skb);
	289	return;
	290	}
	291
75da2163 JM	292	m = tipc_group_find_member(grp, node, port);
	293	if (!tipc_group_is_receiver(m))
	294	goto drop;
	295
31c82a2d	296	TIPC_SKB_CB(skb)->orig_member = m->instance;
75da2163 JM	297	__skb_queue_tail(inputq, skb);
	298
	299	m->bc_rcv_nxt = msg_grp_bc_seqno(hdr) + 1;
	300	return;
	301	drop:
	302	kfree_skb(skb);
	303	}
	304
	305	static void tipc_group_proto_xmit(struct tipc_group grp, struct tipc_member m,
	306	int mtyp, struct sk_buff_head *xmitq)
	307	{
	308	struct tipc_msg *hdr;
	309	struct sk_buff *skb;
	310
	311	skb = tipc_msg_create(GROUP_PROTOCOL, mtyp, INT_H_SIZE, 0,
	312	m->node, tipc_own_addr(grp->net),
	313	m->port, grp->portid, 0);
	314	if (!skb)
	315	return;
	316
	317	hdr = buf_msg(skb);
	318	if (mtyp == GRP_JOIN_MSG)
	319	msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
	320	__skb_queue_tail(xmitq, skb);
	321	}
	322
	323	void tipc_group_proto_rcv(struct tipc_group grp, struct tipc_msg hdr,
ae236fb2	324	struct sk_buff_head *inputq,
75da2163 JM	325	struct sk_buff_head *xmitq)
	326	{
	327	u32 node = msg_orignode(hdr);
	328	u32 port = msg_origport(hdr);
	329	struct tipc_member *m;
	330
	331	if (!grp)
	332	return;
	333
	334	m = tipc_group_find_member(grp, node, port);
	335
	336	switch (msg_type(hdr)) {
	337	case GRP_JOIN_MSG:
	338	if (!m)
	339	m = tipc_group_create_member(grp, node, port,
	340	MBR_QUARANTINED);
	341	if (!m)
	342	return;
	343	m->bc_rcv_nxt = msg_grp_bc_syncpt(hdr);
	344
	345	/* Wait until PUBLISH event is received */
ae236fb2	346	if (m->state == MBR_DISCOVERED) {
75da2163	347	m->state = MBR_JOINING;
ae236fb2	348	} else if (m->state == MBR_PUBLISHED) {
75da2163	349	m->state = MBR_JOINED;
ae236fb2 JM	350	__skb_queue_tail(inputq, m->event_msg);
ae236fb2 JM	351	}
75da2163 JM	352	return;
	353	case GRP_LEAVE_MSG:
	354	if (!m)
	355	return;
	356
	357	/* Wait until WITHDRAW event is received */
	358	if (m->state != MBR_LEAVING) {
	359	m->state = MBR_LEAVING;
	360	return;
	361	}
	362	/* Otherwise deliver already received WITHDRAW event */
ae236fb2	363	__skb_queue_tail(inputq, m->event_msg);
75da2163 JM	364	tipc_group_delete_member(grp, m);
	365	return;
	366	default:
	367	pr_warn("Received unknown GROUP_PROTO message\n");
	368	}
	369	}
	370
75da2163 JM	371	void tipc_group_member_evt(struct tipc_group *grp,
75da2163 JM	372	struct sk_buff *skb,
ae236fb2	373	struct sk_buff_head *inputq,
75da2163 JM	374	struct sk_buff_head *xmitq)
	375	{
	376	struct tipc_msg *hdr = buf_msg(skb);
	377	struct tipc_event evt = (void )msg_data(hdr);
ae236fb2	378	u32 instance = evt->found_lower;
75da2163 JM	379	u32 node = evt->port.node;
75da2163 JM	380	u32 port = evt->port.ref;
ae236fb2	381	int event = evt->event;
75da2163 JM	382	struct tipc_member *m;
	383	struct net *net;
	384	u32 self;
	385
	386	if (!grp)
	387	goto drop;
	388
	389	net = grp->net;
	390	self = tipc_own_addr(net);
	391	if (!grp->loopback && node == self && port == grp->portid)
	392	goto drop;
	393
ae236fb2 JM	394	/* Convert message before delivery to user */
	395	msg_set_hdr_sz(hdr, GROUP_H_SIZE);
	396	msg_set_user(hdr, TIPC_CRITICAL_IMPORTANCE);
	397	msg_set_type(hdr, TIPC_GRP_MEMBER_EVT);
	398	msg_set_origport(hdr, port);
	399	msg_set_orignode(hdr, node);
	400	msg_set_nametype(hdr, grp->type);
	401	msg_set_grp_evt(hdr, event);
	402
75da2163 JM	403	m = tipc_group_find_member(grp, node, port);
75da2163 JM	404
ae236fb2	405	if (event == TIPC_PUBLISHED) {
75da2163 JM	406	if (!m)
	407	m = tipc_group_create_member(grp, node, port,
	408	MBR_DISCOVERED);
	409	if (!m)
	410	goto drop;
	411
ae236fb2 JM	412	/* Hold back event if JOIN message not yet received */
	413	if (m->state == MBR_DISCOVERED) {
	414	m->event_msg = skb;
75da2163	415	m->state = MBR_PUBLISHED;
ae236fb2 JM	416	} else {
ae236fb2 JM	417	__skb_queue_tail(inputq, skb);
75da2163	418	m->state = MBR_JOINED;
ae236fb2 JM	419	}
	420	m->instance = instance;
	421	TIPC_SKB_CB(skb)->orig_member = m->instance;
75da2163	422	tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
ae236fb2	423	} else if (event == TIPC_WITHDRAWN) {
75da2163 JM	424	if (!m)
	425	goto drop;
	426
ae236fb2 JM	427	TIPC_SKB_CB(skb)->orig_member = m->instance;
	428
	429	/* Hold back event if more messages might be expected */
	430	if (m->state != MBR_LEAVING && tipc_node_is_up(net, node)) {
	431	m->event_msg = skb;
75da2163	432	m->state = MBR_LEAVING;
ae236fb2 JM	433	} else {
ae236fb2 JM	434	__skb_queue_tail(inputq, skb);
75da2163	435	tipc_group_delete_member(grp, m);
ae236fb2	436	}
75da2163	437	}
ae236fb2	438	return;
75da2163 JM	439	drop:
	440	kfree_skb(skb);
	441	}