[linux-2.6-block.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;
	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;
};

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;
	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;

	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 *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;
		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 */
		tipc_group_delete_member(grp, m);
		return;
	default:
		pr_warn("Received unknown GROUP_PROTO message\n");
	}
}

/* tipc_group_member_evt() - receive and handle a member up/down event
 */
void tipc_group_member_evt(struct tipc_group *grp,
			   struct sk_buff *skb,
			   struct sk_buff_head *xmitq)
{
	struct tipc_msg *hdr = buf_msg(skb);
	struct tipc_event *evt = (void *)msg_data(hdr);
	u32 node = evt->port.node;
	u32 port = evt->port.ref;
	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;

	m = tipc_group_find_member(grp, node, port);

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

		/* Wait if JOIN message not yet received */
		if (m->state == MBR_DISCOVERED)
			m->state = MBR_PUBLISHED;
		else
			m->state = MBR_JOINED;
		m->instance = evt->found_lower;
		tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
	} else if (evt->event == TIPC_WITHDRAWN) {
		if (!m)
			goto drop;

		/* Keep back event if more messages might be expected */
		if (m->state != MBR_LEAVING && tipc_node_is_up(net, node))
			m->state = MBR_LEAVING;
		else
			tipc_group_delete_member(grp, m);
	}
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;
	62	u32 node;
	63	u32 port;
31c82a2d	64	u32 instance;
75da2163 JM	65	enum mbr_state state;
	66	u16 bc_rcv_nxt;
	67	};
	68
	69	struct tipc_group {
	70	struct rb_root members;
	71	struct tipc_nlist dests;
	72	struct net *net;
	73	int subid;
	74	u32 type;
	75	u32 instance;
	76	u32 domain;
	77	u32 scope;
	78	u32 portid;
	79	u16 member_cnt;
	80	u16 bc_snd_nxt;
	81	bool loopback;
	82	};
	83
	84	static void tipc_group_proto_xmit(struct tipc_group grp, struct tipc_member m,
	85	int mtyp, struct sk_buff_head *xmitq);
	86
	87	u16 tipc_group_bc_snd_nxt(struct tipc_group *grp)
	88	{
	89	return grp->bc_snd_nxt;
	90	}
	91
	92	static bool tipc_group_is_receiver(struct tipc_member *m)
	93	{
	94	return m && m->state >= MBR_JOINED;
	95	}
	96
	97	int tipc_group_size(struct tipc_group *grp)
	98	{
	99	return grp->member_cnt;
	100	}
	101
	102	struct tipc_group tipc_group_create(struct net net, u32 portid,
	103	struct tipc_group_req *mreq)
	104	{
	105	struct tipc_group *grp;
	106	u32 type = mreq->type;
	107
	108	grp = kzalloc(sizeof(*grp), GFP_ATOMIC);
	109	if (!grp)
	110	return NULL;
	111	tipc_nlist_init(&grp->dests, tipc_own_addr(net));
	112	grp->members = RB_ROOT;
	113	grp->net = net;
	114	grp->portid = portid;
	115	grp->domain = addr_domain(net, mreq->scope);
	116	grp->type = type;
	117	grp->instance = mreq->instance;
	118	grp->scope = mreq->scope;
	119	grp->loopback = mreq->flags & TIPC_GROUP_LOOPBACK;
	120	if (tipc_topsrv_kern_subscr(net, portid, type, 0, ~0, &grp->subid))
	121	return grp;
	122	kfree(grp);
	123	return NULL;
	124	}
	125
	126	void tipc_group_delete(struct net net, struct tipc_group grp)
	127	{
	128	struct rb_root *tree = &grp->members;
129	struct tipc_member m, tmp;
130	struct sk_buff_head xmitq;
131
132	__skb_queue_head_init(&xmitq);
133
134	rbtree_postorder_for_each_entry_safe(m, tmp, tree, tree_node) {
135	tipc_group_proto_xmit(grp, m, GRP_LEAVE_MSG, &xmitq);
136	list_del(&m->list);
137	kfree(m);
138	}
139	tipc_node_distr_xmit(net, &xmitq);
140	tipc_nlist_purge(&grp->dests);
141	tipc_topsrv_kern_unsubscr(net, grp->subid);
142	kfree(grp);
143	}
144
145	struct tipc_member tipc_group_find_member(struct tipc_group grp,
146	u32 node, u32 port)
147	{
148	struct rb_node *n = grp->members.rb_node;
149	u64 nkey, key = (u64)node << 32 \| port;
150	struct tipc_member *m;
151
152	while (n) {
153	m = container_of(n, struct tipc_member, tree_node);
154	nkey = (u64)m->node << 32 \| m->port;
155	if (key < nkey)
156	n = n->rb_left;
157	else if (key > nkey)
158	n = n->rb_right;
159	else
160	return m;
161	}
162	return NULL;
163	}
164
165	static struct tipc_member tipc_group_find_node(struct tipc_group grp,
166	u32 node)
167	{
168	struct tipc_member *m;
169	struct rb_node *n;
170
171	for (n = rb_first(&grp->members); n; n = rb_next(n)) {
172	m = container_of(n, struct tipc_member, tree_node);
173	if (m->node == node)
174	return m;
175	}
176	return NULL;
177	}
178
179	static void tipc_group_add_to_tree(struct tipc_group *grp,
180	struct tipc_member *m)
181	{
182	u64 nkey, key = (u64)m->node << 32 \| m->port;
183	struct rb_node *n, parent = NULL;
184	struct tipc_member *tmp;
185
186	n = &grp->members.rb_node;
187	while (*n) {
188	tmp = container_of(*n, struct tipc_member, tree_node);
189	parent = *n;
190	tmp = container_of(parent, struct tipc_member, tree_node);
191	nkey = (u64)tmp->node << 32 \| tmp->port;
192	if (key < nkey)
193	n = &(*n)->rb_left;
194	else if (key > nkey)
195	n = &(*n)->rb_right;
196	else
197	return;
198	}
199	rb_link_node(&m->tree_node, parent, n);
200	rb_insert_color(&m->tree_node, &grp->members);
201	}
202
203	static struct tipc_member tipc_group_create_member(struct tipc_group grp,
204	u32 node, u32 port,
205	int state)
206	{
207	struct tipc_member *m;
208
209	m = kzalloc(sizeof(*m), GFP_ATOMIC);
210	if (!m)
211	return NULL;
212	INIT_LIST_HEAD(&m->list);
213	m->node = node;
214	m->port = port;
215	grp->member_cnt++;
216	tipc_group_add_to_tree(grp, m);
217	tipc_nlist_add(&grp->dests, m->node);
218	m->state = state;
219	return m;
220	}
221
222	void tipc_group_add_member(struct tipc_group *grp, u32 node, u32 port)
223	{
224	tipc_group_create_member(grp, node, port, MBR_DISCOVERED);
225	}
226
227	static void tipc_group_delete_member(struct tipc_group *grp,
228	struct tipc_member *m)
229	{
230	rb_erase(&m->tree_node, &grp->members);
231	grp->member_cnt--;
232	list_del_init(&m->list);
233
234	/* If last member on a node, remove node from dest list */
235	if (!tipc_group_find_node(grp, m->node))
236	tipc_nlist_del(&grp->dests, m->node);
237
238	kfree(m);
239	}
240
241	struct tipc_nlist tipc_group_dests(struct tipc_group grp)
242	{
243	return &grp->dests;
244	}
245
246	void tipc_group_self(struct tipc_group grp, struct tipc_name_seq seq,
247	int *scope)
248	{
249	seq->type = grp->type;
250	seq->lower = grp->instance;
251	seq->upper = grp->instance;
252	*scope = grp->scope;
253	}
254
255	void tipc_group_update_bc_members(struct tipc_group *grp)
256	{
257	grp->bc_snd_nxt++;
258	}
259
260	/* tipc_group_filter_msg() - determine if we should accept arriving message
261	*/
262	void tipc_group_filter_msg(struct tipc_group grp, struct sk_buff_head inputq,
263	struct sk_buff_head *xmitq)
264	{
265	struct sk_buff *skb = __skb_dequeue(inputq);
266	struct tipc_member *m;
267	struct tipc_msg *hdr;
268	u32 node, port;
269	int mtyp;
270
271	if (!skb)
272	return;
273
274	hdr = buf_msg(skb);
275	mtyp = msg_type(hdr);
276	node = msg_orignode(hdr);
277	port = msg_origport(hdr);
278
279	if (!msg_in_group(hdr))
280	goto drop;
281
282	m = tipc_group_find_member(grp, node, port);
283	if (!tipc_group_is_receiver(m))
284	goto drop;
285
31c82a2d	286	TIPC_SKB_CB(skb)->orig_member = m->instance;
75da2163 JM	287	__skb_queue_tail(inputq, skb);
	288
	289	m->bc_rcv_nxt = msg_grp_bc_seqno(hdr) + 1;
	290	return;
	291	drop:
	292	kfree_skb(skb);
	293	}
	294
	295	static void tipc_group_proto_xmit(struct tipc_group grp, struct tipc_member m,
	296	int mtyp, struct sk_buff_head *xmitq)
	297	{
	298	struct tipc_msg *hdr;
	299	struct sk_buff *skb;
	300
	301	skb = tipc_msg_create(GROUP_PROTOCOL, mtyp, INT_H_SIZE, 0,
	302	m->node, tipc_own_addr(grp->net),
	303	m->port, grp->portid, 0);
	304	if (!skb)
	305	return;
	306
	307	hdr = buf_msg(skb);
	308	if (mtyp == GRP_JOIN_MSG)
	309	msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
	310	__skb_queue_tail(xmitq, skb);
	311	}
	312
	313	void tipc_group_proto_rcv(struct tipc_group grp, struct tipc_msg hdr,
	314	struct sk_buff_head *xmitq)
	315	{
	316	u32 node = msg_orignode(hdr);
	317	u32 port = msg_origport(hdr);
	318	struct tipc_member *m;
	319
	320	if (!grp)
	321	return;
	322
	323	m = tipc_group_find_member(grp, node, port);
	324
	325	switch (msg_type(hdr)) {
	326	case GRP_JOIN_MSG:
	327	if (!m)
	328	m = tipc_group_create_member(grp, node, port,
	329	MBR_QUARANTINED);
	330	if (!m)
	331	return;
	332	m->bc_rcv_nxt = msg_grp_bc_syncpt(hdr);
	333
	334	/* Wait until PUBLISH event is received */
	335	if (m->state == MBR_DISCOVERED)
	336	m->state = MBR_JOINING;
	337	else if (m->state == MBR_PUBLISHED)
	338	m->state = MBR_JOINED;
	339	return;
	340	case GRP_LEAVE_MSG:
	341	if (!m)
	342	return;
	343
	344	/* Wait until WITHDRAW event is received */
	345	if (m->state != MBR_LEAVING) {
	346	m->state = MBR_LEAVING;
	347	return;
	348	}
	349	/* Otherwise deliver already received WITHDRAW event */
	350	tipc_group_delete_member(grp, m);
351	return;
352	default:
353	pr_warn("Received unknown GROUP_PROTO message\n");
354	}
355	}
356
357	/* tipc_group_member_evt() - receive and handle a member up/down event
358	*/
359	void tipc_group_member_evt(struct tipc_group *grp,
360	struct sk_buff *skb,
361	struct sk_buff_head *xmitq)
362	{
363	struct tipc_msg *hdr = buf_msg(skb);
364	struct tipc_event evt = (void )msg_data(hdr);
365	u32 node = evt->port.node;
366	u32 port = evt->port.ref;
367	struct tipc_member *m;
368	struct net *net;
369	u32 self;
370
371	if (!grp)
372	goto drop;
373
374	net = grp->net;
375	self = tipc_own_addr(net);
376	if (!grp->loopback && node == self && port == grp->portid)
377	goto drop;
378
379	m = tipc_group_find_member(grp, node, port);
380
381	if (evt->event == TIPC_PUBLISHED) {
382	if (!m)
383	m = tipc_group_create_member(grp, node, port,
384	MBR_DISCOVERED);
385	if (!m)
386	goto drop;
387
388	/* Wait if JOIN message not yet received */
389	if (m->state == MBR_DISCOVERED)
390	m->state = MBR_PUBLISHED;
391	else
392	m->state = MBR_JOINED;
31c82a2d	393	m->instance = evt->found_lower;
75da2163 JM	394	tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
	395	} else if (evt->event == TIPC_WITHDRAWN) {
	396	if (!m)
	397	goto drop;
	398
	399	/* Keep back event if more messages might be expected */
	400	if (m->state != MBR_LEAVING && tipc_node_is_up(net, node))
	401	m->state = MBR_LEAVING;
	402	else
	403	tipc_group_delete_member(grp, m);
	404	}
	405	drop:
	406	kfree_skb(skb);
	407	}