Linux Audio

Check our new training course

Yocto / OpenEmbedded training

Mar 24-27, 2025, special US time zones
Register
Loading...
Note: File does not exist in v3.1.
  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 "topsrv.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#define ADV_ACTIVE (ADV_UNIT * 12)
 50
 51enum mbr_state {
 52	MBR_JOINING,
 53	MBR_PUBLISHED,
 54	MBR_JOINED,
 55	MBR_PENDING,
 56	MBR_ACTIVE,
 57	MBR_RECLAIMING,
 58	MBR_REMITTED,
 59	MBR_LEAVING
 60};
 61
 62struct tipc_member {
 63	struct rb_node tree_node;
 64	struct list_head list;
 65	struct list_head small_win;
 66	struct sk_buff_head deferredq;
 67	struct tipc_group *group;
 68	u32 node;
 69	u32 port;
 70	u32 instance;
 71	enum mbr_state state;
 72	u16 advertised;
 73	u16 window;
 74	u16 bc_rcv_nxt;
 75	u16 bc_syncpt;
 76	u16 bc_acked;
 77};
 78
 79struct tipc_group {
 80	struct rb_root members;
 81	struct list_head small_win;
 82	struct list_head pending;
 83	struct list_head active;
 84	struct tipc_nlist dests;
 85	struct net *net;
 86	int subid;
 87	u32 type;
 88	u32 instance;
 89	u32 scope;
 90	u32 portid;
 91	u16 member_cnt;
 92	u16 active_cnt;
 93	u16 max_active;
 94	u16 bc_snd_nxt;
 95	u16 bc_ackers;
 96	bool *open;
 97	bool loopback;
 98	bool events;
 99};
100
101static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,
102				  int mtyp, struct sk_buff_head *xmitq);
103
104static void tipc_group_open(struct tipc_member *m, bool *wakeup)
105{
106	*wakeup = false;
107	if (list_empty(&m->small_win))
108		return;
109	list_del_init(&m->small_win);
110	*m->group->open = true;
111	*wakeup = true;
112}
113
114static void tipc_group_decr_active(struct tipc_group *grp,
115				   struct tipc_member *m)
116{
117	if (m->state == MBR_ACTIVE || m->state == MBR_RECLAIMING ||
118	    m->state == MBR_REMITTED)
119		grp->active_cnt--;
120}
121
122static int tipc_group_rcvbuf_limit(struct tipc_group *grp)
123{
124	int max_active, active_pool, idle_pool;
125	int mcnt = grp->member_cnt + 1;
126
127	/* Limit simultaneous reception from other members */
128	max_active = min(mcnt / 8, 64);
129	max_active = max(max_active, 16);
130	grp->max_active = max_active;
131
132	/* Reserve blocks for active and idle members */
133	active_pool = max_active * ADV_ACTIVE;
134	idle_pool = (mcnt - max_active) * ADV_IDLE;
135
136	/* Scale to bytes, considering worst-case truesize/msgsize ratio */
137	return (active_pool + idle_pool) * FLOWCTL_BLK_SZ * 4;
138}
139
140u16 tipc_group_bc_snd_nxt(struct tipc_group *grp)
141{
142	return grp->bc_snd_nxt;
143}
144
145static bool tipc_group_is_receiver(struct tipc_member *m)
146{
147	return m && m->state != MBR_JOINING && m->state != MBR_LEAVING;
148}
149
150static bool tipc_group_is_sender(struct tipc_member *m)
151{
152	return m && m->state != MBR_JOINING && m->state != MBR_PUBLISHED;
153}
154
155u32 tipc_group_exclude(struct tipc_group *grp)
156{
157	if (!grp->loopback)
158		return grp->portid;
159	return 0;
160}
161
162int tipc_group_size(struct tipc_group *grp)
163{
164	return grp->member_cnt;
165}
166
167struct tipc_group *tipc_group_create(struct net *net, u32 portid,
168				     struct tipc_group_req *mreq,
169				     bool *group_is_open)
170{
171	u32 filter = TIPC_SUB_PORTS | TIPC_SUB_NO_STATUS;
172	bool global = mreq->scope != TIPC_NODE_SCOPE;
173	struct tipc_group *grp;
174	u32 type = mreq->type;
175
176	grp = kzalloc(sizeof(*grp), GFP_ATOMIC);
177	if (!grp)
178		return NULL;
179	tipc_nlist_init(&grp->dests, tipc_own_addr(net));
180	INIT_LIST_HEAD(&grp->small_win);
181	INIT_LIST_HEAD(&grp->active);
182	INIT_LIST_HEAD(&grp->pending);
183	grp->members = RB_ROOT;
184	grp->net = net;
185	grp->portid = portid;
186	grp->type = type;
187	grp->instance = mreq->instance;
188	grp->scope = mreq->scope;
189	grp->loopback = mreq->flags & TIPC_GROUP_LOOPBACK;
190	grp->events = mreq->flags & TIPC_GROUP_MEMBER_EVTS;
191	grp->open = group_is_open;
192	*grp->open = false;
193	filter |= global ? TIPC_SUB_CLUSTER_SCOPE : TIPC_SUB_NODE_SCOPE;
194	if (tipc_topsrv_kern_subscr(net, portid, type, 0, ~0,
195				    filter, &grp->subid))
196		return grp;
197	kfree(grp);
198	return NULL;
199}
200
201void tipc_group_join(struct net *net, struct tipc_group *grp, int *sk_rcvbuf)
202{
203	struct rb_root *tree = &grp->members;
204	struct tipc_member *m, *tmp;
205	struct sk_buff_head xmitq;
206
207	skb_queue_head_init(&xmitq);
208	rbtree_postorder_for_each_entry_safe(m, tmp, tree, tree_node) {
209		tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, &xmitq);
210		tipc_group_update_member(m, 0);
211	}
212	tipc_node_distr_xmit(net, &xmitq);
213	*sk_rcvbuf = tipc_group_rcvbuf_limit(grp);
214}
215
216void tipc_group_delete(struct net *net, struct tipc_group *grp)
217{
218	struct rb_root *tree = &grp->members;
219	struct tipc_member *m, *tmp;
220	struct sk_buff_head xmitq;
221
222	__skb_queue_head_init(&xmitq);
223
224	rbtree_postorder_for_each_entry_safe(m, tmp, tree, tree_node) {
225		tipc_group_proto_xmit(grp, m, GRP_LEAVE_MSG, &xmitq);
226		list_del(&m->list);
227		kfree(m);
228	}
229	tipc_node_distr_xmit(net, &xmitq);
230	tipc_nlist_purge(&grp->dests);
231	tipc_topsrv_kern_unsubscr(net, grp->subid);
232	kfree(grp);
233}
234
235struct tipc_member *tipc_group_find_member(struct tipc_group *grp,
236					   u32 node, u32 port)
237{
238	struct rb_node *n = grp->members.rb_node;
239	u64 nkey, key = (u64)node << 32 | port;
240	struct tipc_member *m;
241
242	while (n) {
243		m = container_of(n, struct tipc_member, tree_node);
244		nkey = (u64)m->node << 32 | m->port;
245		if (key < nkey)
246			n = n->rb_left;
247		else if (key > nkey)
248			n = n->rb_right;
249		else
250			return m;
251	}
252	return NULL;
253}
254
255static struct tipc_member *tipc_group_find_dest(struct tipc_group *grp,
256						u32 node, u32 port)
257{
258	struct tipc_member *m;
259
260	m = tipc_group_find_member(grp, node, port);
261	if (m && tipc_group_is_receiver(m))
262		return m;
263	return NULL;
264}
265
266static struct tipc_member *tipc_group_find_node(struct tipc_group *grp,
267						u32 node)
268{
269	struct tipc_member *m;
270	struct rb_node *n;
271
272	for (n = rb_first(&grp->members); n; n = rb_next(n)) {
273		m = container_of(n, struct tipc_member, tree_node);
274		if (m->node == node)
275			return m;
276	}
277	return NULL;
278}
279
280static void tipc_group_add_to_tree(struct tipc_group *grp,
281				   struct tipc_member *m)
282{
283	u64 nkey, key = (u64)m->node << 32 | m->port;
284	struct rb_node **n, *parent = NULL;
285	struct tipc_member *tmp;
286
287	n = &grp->members.rb_node;
288	while (*n) {
289		tmp = container_of(*n, struct tipc_member, tree_node);
290		parent = *n;
291		tmp = container_of(parent, struct tipc_member, tree_node);
292		nkey = (u64)tmp->node << 32 | tmp->port;
293		if (key < nkey)
294			n = &(*n)->rb_left;
295		else if (key > nkey)
296			n = &(*n)->rb_right;
297		else
298			return;
299	}
300	rb_link_node(&m->tree_node, parent, n);
301	rb_insert_color(&m->tree_node, &grp->members);
302}
303
304static struct tipc_member *tipc_group_create_member(struct tipc_group *grp,
305						    u32 node, u32 port,
306						    u32 instance, int state)
307{
308	struct tipc_member *m;
309
310	m = kzalloc(sizeof(*m), GFP_ATOMIC);
311	if (!m)
312		return NULL;
313	INIT_LIST_HEAD(&m->list);
314	INIT_LIST_HEAD(&m->small_win);
315	__skb_queue_head_init(&m->deferredq);
316	m->group = grp;
317	m->node = node;
318	m->port = port;
319	m->instance = instance;
320	m->bc_acked = grp->bc_snd_nxt - 1;
321	grp->member_cnt++;
322	tipc_group_add_to_tree(grp, m);
323	tipc_nlist_add(&grp->dests, m->node);
324	m->state = state;
325	return m;
326}
327
328void tipc_group_add_member(struct tipc_group *grp, u32 node,
329			   u32 port, u32 instance)
330{
331	tipc_group_create_member(grp, node, port, instance, MBR_PUBLISHED);
332}
333
334static void tipc_group_delete_member(struct tipc_group *grp,
335				     struct tipc_member *m)
336{
337	rb_erase(&m->tree_node, &grp->members);
338	grp->member_cnt--;
339
340	/* Check if we were waiting for replicast ack from this member */
341	if (grp->bc_ackers && less(m->bc_acked, grp->bc_snd_nxt - 1))
342		grp->bc_ackers--;
343
344	list_del_init(&m->list);
345	list_del_init(&m->small_win);
346	tipc_group_decr_active(grp, m);
347
348	/* If last member on a node, remove node from dest list */
349	if (!tipc_group_find_node(grp, m->node))
350		tipc_nlist_del(&grp->dests, m->node);
351
352	kfree(m);
353}
354
355struct tipc_nlist *tipc_group_dests(struct tipc_group *grp)
356{
357	return &grp->dests;
358}
359
360void tipc_group_self(struct tipc_group *grp, struct tipc_name_seq *seq,
361		     int *scope)
362{
363	seq->type = grp->type;
364	seq->lower = grp->instance;
365	seq->upper = grp->instance;
366	*scope = grp->scope;
367}
368
369void tipc_group_update_member(struct tipc_member *m, int len)
370{
371	struct tipc_group *grp = m->group;
372	struct tipc_member *_m, *tmp;
373
374	if (!tipc_group_is_receiver(m))
375		return;
376
377	m->window -= len;
378
379	if (m->window >= ADV_IDLE)
380		return;
381
382	list_del_init(&m->small_win);
383
384	/* Sort member into small_window members' list */
385	list_for_each_entry_safe(_m, tmp, &grp->small_win, small_win) {
386		if (_m->window > m->window)
387			break;
388	}
389	list_add_tail(&m->small_win, &_m->small_win);
390}
391
392void tipc_group_update_bc_members(struct tipc_group *grp, int len, bool ack)
393{
394	u16 prev = grp->bc_snd_nxt - 1;
395	struct tipc_member *m;
396	struct rb_node *n;
397	u16 ackers = 0;
398
399	for (n = rb_first(&grp->members); n; n = rb_next(n)) {
400		m = container_of(n, struct tipc_member, tree_node);
401		if (tipc_group_is_receiver(m)) {
402			tipc_group_update_member(m, len);
403			m->bc_acked = prev;
404			ackers++;
405		}
406	}
407
408	/* Mark number of acknowledges to expect, if any */
409	if (ack)
410		grp->bc_ackers = ackers;
411	grp->bc_snd_nxt++;
412}
413
414bool tipc_group_cong(struct tipc_group *grp, u32 dnode, u32 dport,
415		     int len, struct tipc_member **mbr)
416{
417	struct sk_buff_head xmitq;
418	struct tipc_member *m;
419	int adv, state;
420
421	m = tipc_group_find_dest(grp, dnode, dport);
422	if (!tipc_group_is_receiver(m)) {
423		*mbr = NULL;
424		return false;
425	}
426	*mbr = m;
427
428	if (m->window >= len)
429		return false;
430
431	*grp->open = false;
432
433	/* If not fully advertised, do it now to prevent mutual blocking */
434	adv = m->advertised;
435	state = m->state;
436	if (state == MBR_JOINED && adv == ADV_IDLE)
437		return true;
438	if (state == MBR_ACTIVE && adv == ADV_ACTIVE)
439		return true;
440	if (state == MBR_PENDING && adv == ADV_IDLE)
441		return true;
442	skb_queue_head_init(&xmitq);
443	tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, &xmitq);
444	tipc_node_distr_xmit(grp->net, &xmitq);
445	return true;
446}
447
448bool tipc_group_bc_cong(struct tipc_group *grp, int len)
449{
450	struct tipc_member *m = NULL;
451
452	/* If prev bcast was replicast, reject until all receivers have acked */
453	if (grp->bc_ackers) {
454		*grp->open = false;
455		return true;
456	}
457	if (list_empty(&grp->small_win))
458		return false;
459
460	m = list_first_entry(&grp->small_win, struct tipc_member, small_win);
461	if (m->window >= len)
462		return false;
463
464	return tipc_group_cong(grp, m->node, m->port, len, &m);
465}
466
467/* tipc_group_sort_msg() - sort msg into queue by bcast sequence number
468 */
469static void tipc_group_sort_msg(struct sk_buff *skb, struct sk_buff_head *defq)
470{
471	struct tipc_msg *_hdr, *hdr = buf_msg(skb);
472	u16 bc_seqno = msg_grp_bc_seqno(hdr);
473	struct sk_buff *_skb, *tmp;
474	int mtyp = msg_type(hdr);
475
476	/* Bcast/mcast may be bypassed by ucast or other bcast, - sort it in */
477	if (mtyp == TIPC_GRP_BCAST_MSG || mtyp == TIPC_GRP_MCAST_MSG) {
478		skb_queue_walk_safe(defq, _skb, tmp) {
479			_hdr = buf_msg(_skb);
480			if (!less(bc_seqno, msg_grp_bc_seqno(_hdr)))
481				continue;
482			__skb_queue_before(defq, _skb, skb);
483			return;
484		}
485		/* Bcast was not bypassed, - add to tail */
486	}
487	/* Unicasts are never bypassed, - always add to tail */
488	__skb_queue_tail(defq, skb);
489}
490
491/* tipc_group_filter_msg() - determine if we should accept arriving message
492 */
493void tipc_group_filter_msg(struct tipc_group *grp, struct sk_buff_head *inputq,
494			   struct sk_buff_head *xmitq)
495{
496	struct sk_buff *skb = __skb_dequeue(inputq);
497	bool ack, deliver, update, leave = false;
498	struct sk_buff_head *defq;
499	struct tipc_member *m;
500	struct tipc_msg *hdr;
501	u32 node, port;
502	int mtyp, blks;
503
504	if (!skb)
505		return;
506
507	hdr = buf_msg(skb);
508	node =  msg_orignode(hdr);
509	port = msg_origport(hdr);
510
511	if (!msg_in_group(hdr))
512		goto drop;
513
514	m = tipc_group_find_member(grp, node, port);
515	if (!tipc_group_is_sender(m))
516		goto drop;
517
518	if (less(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt))
519		goto drop;
520
521	TIPC_SKB_CB(skb)->orig_member = m->instance;
522	defq = &m->deferredq;
523	tipc_group_sort_msg(skb, defq);
524
525	while ((skb = skb_peek(defq))) {
526		hdr = buf_msg(skb);
527		mtyp = msg_type(hdr);
528		blks = msg_blocks(hdr);
529		deliver = true;
530		ack = false;
531		update = false;
532
533		if (more(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt))
534			break;
535
536		/* Decide what to do with message */
537		switch (mtyp) {
538		case TIPC_GRP_MCAST_MSG:
539			if (msg_nameinst(hdr) != grp->instance) {
540				update = true;
541				deliver = false;
542			}
543			/* Fall thru */
544		case TIPC_GRP_BCAST_MSG:
545			m->bc_rcv_nxt++;
546			ack = msg_grp_bc_ack_req(hdr);
547			break;
548		case TIPC_GRP_UCAST_MSG:
549			break;
550		case TIPC_GRP_MEMBER_EVT:
551			if (m->state == MBR_LEAVING)
552				leave = true;
553			if (!grp->events)
554				deliver = false;
555			break;
556		default:
557			break;
558		}
559
560		/* Execute decisions */
561		__skb_dequeue(defq);
562		if (deliver)
563			__skb_queue_tail(inputq, skb);
564		else
565			kfree_skb(skb);
566
567		if (ack)
568			tipc_group_proto_xmit(grp, m, GRP_ACK_MSG, xmitq);
569
570		if (leave) {
571			__skb_queue_purge(defq);
572			tipc_group_delete_member(grp, m);
573			break;
574		}
575		if (!update)
576			continue;
577
578		tipc_group_update_rcv_win(grp, blks, node, port, xmitq);
579	}
580	return;
581drop:
582	kfree_skb(skb);
583}
584
585void tipc_group_update_rcv_win(struct tipc_group *grp, int blks, u32 node,
586			       u32 port, struct sk_buff_head *xmitq)
587{
588	struct list_head *active = &grp->active;
589	int max_active = grp->max_active;
590	int reclaim_limit = max_active * 3 / 4;
591	int active_cnt = grp->active_cnt;
592	struct tipc_member *m, *rm, *pm;
593
594	m = tipc_group_find_member(grp, node, port);
595	if (!m)
596		return;
597
598	m->advertised -= blks;
599
600	switch (m->state) {
601	case MBR_JOINED:
602		/* First, decide if member can go active */
603		if (active_cnt <= max_active) {
604			m->state = MBR_ACTIVE;
605			list_add_tail(&m->list, active);
606			grp->active_cnt++;
607			tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
608		} else {
609			m->state = MBR_PENDING;
610			list_add_tail(&m->list, &grp->pending);
611		}
612
613		if (active_cnt < reclaim_limit)
614			break;
615
616		/* Reclaim from oldest active member, if possible */
617		if (!list_empty(active)) {
618			rm = list_first_entry(active, struct tipc_member, list);
619			rm->state = MBR_RECLAIMING;
620			list_del_init(&rm->list);
621			tipc_group_proto_xmit(grp, rm, GRP_RECLAIM_MSG, xmitq);
622			break;
623		}
624		/* Nobody to reclaim from; - revert oldest pending to JOINED */
625		pm = list_first_entry(&grp->pending, struct tipc_member, list);
626		list_del_init(&pm->list);
627		pm->state = MBR_JOINED;
628		tipc_group_proto_xmit(grp, pm, GRP_ADV_MSG, xmitq);
629		break;
630	case MBR_ACTIVE:
631		if (!list_is_last(&m->list, &grp->active))
632			list_move_tail(&m->list, &grp->active);
633		if (m->advertised > (ADV_ACTIVE * 3 / 4))
634			break;
635		tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
636		break;
637	case MBR_REMITTED:
638		if (m->advertised > ADV_IDLE)
639			break;
640		m->state = MBR_JOINED;
641		grp->active_cnt--;
642		if (m->advertised < ADV_IDLE) {
643			pr_warn_ratelimited("Rcv unexpected msg after REMIT\n");
644			tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
645		}
646
647		if (list_empty(&grp->pending))
648			return;
649
650		/* Set oldest pending member to active and advertise */
651		pm = list_first_entry(&grp->pending, struct tipc_member, list);
652		pm->state = MBR_ACTIVE;
653		list_move_tail(&pm->list, &grp->active);
654		grp->active_cnt++;
655		tipc_group_proto_xmit(grp, pm, GRP_ADV_MSG, xmitq);
656		break;
657	case MBR_RECLAIMING:
658	case MBR_JOINING:
659	case MBR_LEAVING:
660	default:
661		break;
662	}
663}
664
665static void tipc_group_create_event(struct tipc_group *grp,
666				    struct tipc_member *m,
667				    u32 event, u16 seqno,
668				    struct sk_buff_head *inputq)
669{	u32 dnode = tipc_own_addr(grp->net);
670	struct tipc_event evt;
671	struct sk_buff *skb;
672	struct tipc_msg *hdr;
673
674	evt.event = event;
675	evt.found_lower = m->instance;
676	evt.found_upper = m->instance;
677	evt.port.ref = m->port;
678	evt.port.node = m->node;
679	evt.s.seq.type = grp->type;
680	evt.s.seq.lower = m->instance;
681	evt.s.seq.upper = m->instance;
682
683	skb = tipc_msg_create(TIPC_CRITICAL_IMPORTANCE, TIPC_GRP_MEMBER_EVT,
684			      GROUP_H_SIZE, sizeof(evt), dnode, m->node,
685			      grp->portid, m->port, 0);
686	if (!skb)
687		return;
688
689	hdr = buf_msg(skb);
690	msg_set_nametype(hdr, grp->type);
691	msg_set_grp_evt(hdr, event);
692	msg_set_dest_droppable(hdr, true);
693	msg_set_grp_bc_seqno(hdr, seqno);
694	memcpy(msg_data(hdr), &evt, sizeof(evt));
695	TIPC_SKB_CB(skb)->orig_member = m->instance;
696	__skb_queue_tail(inputq, skb);
697}
698
699static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,
700				  int mtyp, struct sk_buff_head *xmitq)
701{
702	struct tipc_msg *hdr;
703	struct sk_buff *skb;
704	int adv = 0;
705
706	skb = tipc_msg_create(GROUP_PROTOCOL, mtyp, INT_H_SIZE, 0,
707			      m->node, tipc_own_addr(grp->net),
708			      m->port, grp->portid, 0);
709	if (!skb)
710		return;
711
712	if (m->state == MBR_ACTIVE)
713		adv = ADV_ACTIVE - m->advertised;
714	else if (m->state == MBR_JOINED || m->state == MBR_PENDING)
715		adv = ADV_IDLE - m->advertised;
716
717	hdr = buf_msg(skb);
718
719	if (mtyp == GRP_JOIN_MSG) {
720		msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
721		msg_set_adv_win(hdr, adv);
722		m->advertised += adv;
723	} else if (mtyp == GRP_LEAVE_MSG) {
724		msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
725	} else if (mtyp == GRP_ADV_MSG) {
726		msg_set_adv_win(hdr, adv);
727		m->advertised += adv;
728	} else if (mtyp == GRP_ACK_MSG) {
729		msg_set_grp_bc_acked(hdr, m->bc_rcv_nxt);
730	} else if (mtyp == GRP_REMIT_MSG) {
731		msg_set_grp_remitted(hdr, m->window);
732	}
733	msg_set_dest_droppable(hdr, true);
734	__skb_queue_tail(xmitq, skb);
735}
736
737void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup,
738			  struct tipc_msg *hdr, struct sk_buff_head *inputq,
739			  struct sk_buff_head *xmitq)
740{
741	u32 node = msg_orignode(hdr);
742	u32 port = msg_origport(hdr);
743	struct tipc_member *m, *pm;
744	u16 remitted, in_flight;
745
746	if (!grp)
747		return;
748
749	if (grp->scope == TIPC_NODE_SCOPE && node != tipc_own_addr(grp->net))
750		return;
751
752	m = tipc_group_find_member(grp, node, port);
753
754	switch (msg_type(hdr)) {
755	case GRP_JOIN_MSG:
756		if (!m)
757			m = tipc_group_create_member(grp, node, port,
758						     0, MBR_JOINING);
759		if (!m)
760			return;
761		m->bc_syncpt = msg_grp_bc_syncpt(hdr);
762		m->bc_rcv_nxt = m->bc_syncpt;
763		m->window += msg_adv_win(hdr);
764
765		/* Wait until PUBLISH event is received if necessary */
766		if (m->state != MBR_PUBLISHED)
767			return;
768
769		/* Member can be taken into service */
770		m->state = MBR_JOINED;
771		tipc_group_open(m, usr_wakeup);
772		tipc_group_update_member(m, 0);
773		tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
774		tipc_group_create_event(grp, m, TIPC_PUBLISHED,
775					m->bc_syncpt, inputq);
776		return;
777	case GRP_LEAVE_MSG:
778		if (!m)
779			return;
780		m->bc_syncpt = msg_grp_bc_syncpt(hdr);
781		list_del_init(&m->list);
782		tipc_group_open(m, usr_wakeup);
783		tipc_group_decr_active(grp, m);
784		m->state = MBR_LEAVING;
785		tipc_group_create_event(grp, m, TIPC_WITHDRAWN,
786					m->bc_syncpt, inputq);
787		return;
788	case GRP_ADV_MSG:
789		if (!m)
790			return;
791		m->window += msg_adv_win(hdr);
792		tipc_group_open(m, usr_wakeup);
793		return;
794	case GRP_ACK_MSG:
795		if (!m)
796			return;
797		m->bc_acked = msg_grp_bc_acked(hdr);
798		if (--grp->bc_ackers)
799			return;
800		list_del_init(&m->small_win);
801		*m->group->open = true;
802		*usr_wakeup = true;
803		tipc_group_update_member(m, 0);
804		return;
805	case GRP_RECLAIM_MSG:
806		if (!m)
807			return;
808		tipc_group_proto_xmit(grp, m, GRP_REMIT_MSG, xmitq);
809		m->window = ADV_IDLE;
810		tipc_group_open(m, usr_wakeup);
811		return;
812	case GRP_REMIT_MSG:
813		if (!m || m->state != MBR_RECLAIMING)
814			return;
815
816		remitted = msg_grp_remitted(hdr);
817
818		/* Messages preceding the REMIT still in receive queue */
819		if (m->advertised > remitted) {
820			m->state = MBR_REMITTED;
821			in_flight = m->advertised - remitted;
822			m->advertised = ADV_IDLE + in_flight;
823			return;
824		}
825		/* This should never happen */
826		if (m->advertised < remitted)
827			pr_warn_ratelimited("Unexpected REMIT msg\n");
828
829		/* All messages preceding the REMIT have been read */
830		m->state = MBR_JOINED;
831		grp->active_cnt--;
832		m->advertised = ADV_IDLE;
833
834		/* Set oldest pending member to active and advertise */
835		if (list_empty(&grp->pending))
836			return;
837		pm = list_first_entry(&grp->pending, struct tipc_member, list);
838		pm->state = MBR_ACTIVE;
839		list_move_tail(&pm->list, &grp->active);
840		grp->active_cnt++;
841		if (pm->advertised <= (ADV_ACTIVE * 3 / 4))
842			tipc_group_proto_xmit(grp, pm, GRP_ADV_MSG, xmitq);
843		return;
844	default:
845		pr_warn("Received unknown GROUP_PROTO message\n");
846	}
847}
848
849/* tipc_group_member_evt() - receive and handle a member up/down event
850 */
851void tipc_group_member_evt(struct tipc_group *grp,
852			   bool *usr_wakeup,
853			   int *sk_rcvbuf,
854			   struct tipc_msg *hdr,
855			   struct sk_buff_head *inputq,
856			   struct sk_buff_head *xmitq)
857{
858	struct tipc_event *evt = (void *)msg_data(hdr);
859	u32 instance = evt->found_lower;
860	u32 node = evt->port.node;
861	u32 port = evt->port.ref;
862	int event = evt->event;
863	struct tipc_member *m;
864	struct net *net;
865	u32 self;
866
867	if (!grp)
868		return;
869
870	net = grp->net;
871	self = tipc_own_addr(net);
872	if (!grp->loopback && node == self && port == grp->portid)
873		return;
874
875	m = tipc_group_find_member(grp, node, port);
876
877	switch (event) {
878	case TIPC_PUBLISHED:
879		/* Send and wait for arrival of JOIN message if necessary */
880		if (!m) {
881			m = tipc_group_create_member(grp, node, port, instance,
882						     MBR_PUBLISHED);
883			if (!m)
884				break;
885			tipc_group_update_member(m, 0);
886			tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
887			break;
888		}
889
890		if (m->state != MBR_JOINING)
891			break;
892
893		/* Member can be taken into service */
894		m->instance = instance;
895		m->state = MBR_JOINED;
896		tipc_group_open(m, usr_wakeup);
897		tipc_group_update_member(m, 0);
898		tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
899		tipc_group_create_event(grp, m, TIPC_PUBLISHED,
900					m->bc_syncpt, inputq);
901		break;
902	case TIPC_WITHDRAWN:
903		if (!m)
904			break;
905
906		tipc_group_decr_active(grp, m);
907		m->state = MBR_LEAVING;
908		list_del_init(&m->list);
909		tipc_group_open(m, usr_wakeup);
910
911		/* Only send event if no LEAVE message can be expected */
912		if (!tipc_node_is_up(net, node))
913			tipc_group_create_event(grp, m, TIPC_WITHDRAWN,
914						m->bc_rcv_nxt, inputq);
915		break;
916	default:
917		break;
918	}
919	*sk_rcvbuf = tipc_group_rcvbuf_limit(grp);
920}