Linux Audio

Check our new training course

Linux kernel drivers training

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