group.c 21.8 KB
Newer Older
J
Jon Maloy 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48
/*
 * 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
49
#define ADV_ACTIVE (ADV_UNIT * 12)
J
Jon Maloy 已提交
50 51 52 53 54 55 56

enum mbr_state {
	MBR_QUARANTINED,
	MBR_DISCOVERED,
	MBR_JOINING,
	MBR_PUBLISHED,
	MBR_JOINED,
57 58 59 60
	MBR_PENDING,
	MBR_ACTIVE,
	MBR_RECLAIMING,
	MBR_REMITTED,
J
Jon Maloy 已提交
61 62 63 64 65 66
	MBR_LEAVING
};

struct tipc_member {
	struct rb_node tree_node;
	struct list_head list;
67
	struct list_head congested;
68
	struct sk_buff *event_msg;
69
	struct sk_buff_head deferredq;
70
	struct tipc_group *group;
J
Jon Maloy 已提交
71 72
	u32 node;
	u32 port;
73
	u32 instance;
J
Jon Maloy 已提交
74
	enum mbr_state state;
75 76
	u16 advertised;
	u16 window;
J
Jon Maloy 已提交
77
	u16 bc_rcv_nxt;
78
	u16 bc_syncpt;
79
	u16 bc_acked;
80
	bool usr_pending;
J
Jon Maloy 已提交
81 82 83 84
};

struct tipc_group {
	struct rb_root members;
85
	struct list_head congested;
86 87 88
	struct list_head pending;
	struct list_head active;
	struct list_head reclaiming;
J
Jon Maloy 已提交
89 90 91 92 93 94 95 96 97
	struct tipc_nlist dests;
	struct net *net;
	int subid;
	u32 type;
	u32 instance;
	u32 domain;
	u32 scope;
	u32 portid;
	u16 member_cnt;
98 99
	u16 active_cnt;
	u16 max_active;
J
Jon Maloy 已提交
100
	u16 bc_snd_nxt;
101
	u16 bc_ackers;
J
Jon Maloy 已提交
102
	bool loopback;
103
	bool events;
J
Jon Maloy 已提交
104 105 106 107 108
};

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

109 110 111 112 113 114 115
static void tipc_group_decr_active(struct tipc_group *grp,
				   struct tipc_member *m)
{
	if (m->state == MBR_ACTIVE || m->state == MBR_RECLAIMING)
		grp->active_cnt--;
}

116 117
static int tipc_group_rcvbuf_limit(struct tipc_group *grp)
{
118
	int max_active, active_pool, idle_pool;
119 120
	int mcnt = grp->member_cnt + 1;

121 122 123 124 125 126 127 128 129
	/* Limit simultaneous reception from other members */
	max_active = min(mcnt / 8, 64);
	max_active = max(max_active, 16);
	grp->max_active = max_active;

	/* Reserve blocks for active and idle members */
	active_pool = max_active * ADV_ACTIVE;
	idle_pool = (mcnt - max_active) * ADV_IDLE;

130
	/* Scale to bytes, considering worst-case truesize/msgsize ratio */
131
	return (active_pool + idle_pool) * FLOWCTL_BLK_SZ * 4;
132 133
}

J
Jon Maloy 已提交
134 135 136 137 138
u16 tipc_group_bc_snd_nxt(struct tipc_group *grp)
{
	return grp->bc_snd_nxt;
}

139 140 141 142 143
static bool tipc_group_is_enabled(struct tipc_member *m)
{
	return m->state != MBR_QUARANTINED && m->state != MBR_LEAVING;
}

J
Jon Maloy 已提交
144 145 146 147 148
static bool tipc_group_is_receiver(struct tipc_member *m)
{
	return m && m->state >= MBR_JOINED;
}

149 150 151 152 153 154 155
u32 tipc_group_exclude(struct tipc_group *grp)
{
	if (!grp->loopback)
		return grp->portid;
	return 0;
}

J
Jon Maloy 已提交
156 157 158 159 160 161 162 163 164 165 166 167 168 169 170
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));
171
	INIT_LIST_HEAD(&grp->congested);
172 173 174
	INIT_LIST_HEAD(&grp->active);
	INIT_LIST_HEAD(&grp->pending);
	INIT_LIST_HEAD(&grp->reclaiming);
J
Jon Maloy 已提交
175 176 177 178 179 180 181 182
	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;
183
	grp->events = mreq->flags & TIPC_GROUP_MEMBER_EVTS;
J
Jon Maloy 已提交
184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228
	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;
}

229 230 231 232 233 234 235 236 237 238 239
static struct tipc_member *tipc_group_find_dest(struct tipc_group *grp,
						u32 node, u32 port)
{
	struct tipc_member *m;

	m = tipc_group_find_member(grp, node, port);
	if (m && tipc_group_is_enabled(m))
		return m;
	return NULL;
}

J
Jon Maloy 已提交
240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287
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);
288
	INIT_LIST_HEAD(&m->congested);
289
	__skb_queue_head_init(&m->deferredq);
290
	m->group = grp;
J
Jon Maloy 已提交
291 292
	m->node = node;
	m->port = port;
293
	m->bc_acked = grp->bc_snd_nxt - 1;
J
Jon Maloy 已提交
294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310
	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--;
311 312 313 314 315

	/* Check if we were waiting for replicast ack from this member */
	if (grp->bc_ackers && less(m->bc_acked, grp->bc_snd_nxt - 1))
		grp->bc_ackers--;

J
Jon Maloy 已提交
316
	list_del_init(&m->list);
317
	list_del_init(&m->congested);
318
	tipc_group_decr_active(grp, m);
J
Jon Maloy 已提交
319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340

	/* 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;
}

341 342 343 344 345 346 347 348 349 350 351 352 353
void tipc_group_update_member(struct tipc_member *m, int len)
{
	struct tipc_group *grp = m->group;
	struct tipc_member *_m, *tmp;

	if (!tipc_group_is_enabled(m))
		return;

	m->window -= len;

	if (m->window >= ADV_IDLE)
		return;

354
	list_del_init(&m->congested);
355 356 357 358 359 360 361 362 363 364 365

	/* Sort member into congested members' list */
	list_for_each_entry_safe(_m, tmp, &grp->congested, congested) {
		if (m->window > _m->window)
			continue;
		list_add_tail(&m->congested, &_m->congested);
		return;
	}
	list_add_tail(&m->congested, &grp->congested);
}

366
void tipc_group_update_bc_members(struct tipc_group *grp, int len, bool ack)
J
Jon Maloy 已提交
367
{
368
	u16 prev = grp->bc_snd_nxt - 1;
369 370
	struct tipc_member *m;
	struct rb_node *n;
371
	u16 ackers = 0;
372 373 374

	for (n = rb_first(&grp->members); n; n = rb_next(n)) {
		m = container_of(n, struct tipc_member, tree_node);
375
		if (tipc_group_is_enabled(m)) {
376
			tipc_group_update_member(m, len);
377
			m->bc_acked = prev;
378
			ackers++;
379
		}
380
	}
381 382 383

	/* Mark number of acknowledges to expect, if any */
	if (ack)
384
		grp->bc_ackers = ackers;
J
Jon Maloy 已提交
385 386 387
	grp->bc_snd_nxt++;
}

388 389
bool tipc_group_cong(struct tipc_group *grp, u32 dnode, u32 dport,
		     int len, struct tipc_member **mbr)
390
{
391
	struct sk_buff_head xmitq;
392
	struct tipc_member *m;
393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411
	int adv, state;

	m = tipc_group_find_dest(grp, dnode, dport);
	*mbr = m;
	if (!m)
		return false;
	if (m->usr_pending)
		return true;
	if (m->window >= len)
		return false;
	m->usr_pending = true;

	/* If not fully advertised, do it now to prevent mutual blocking */
	adv = m->advertised;
	state = m->state;
	if (state < MBR_JOINED)
		return true;
	if (state == MBR_JOINED && adv == ADV_IDLE)
		return true;
412 413 414 415
	if (state == MBR_ACTIVE && adv == ADV_ACTIVE)
		return true;
	if (state == MBR_PENDING && adv == ADV_IDLE)
		return true;
416 417 418 419 420 421 422 423 424
	skb_queue_head_init(&xmitq);
	tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, &xmitq);
	tipc_node_distr_xmit(grp->net, &xmitq);
	return true;
}

bool tipc_group_bc_cong(struct tipc_group *grp, int len)
{
	struct tipc_member *m = NULL;
425

426 427 428 429
	/* If prev bcast was replicast, reject until all receivers have acked */
	if (grp->bc_ackers)
		return true;

430 431 432 433 434 435 436
	if (list_empty(&grp->congested))
		return false;

	m = list_first_entry(&grp->congested, struct tipc_member, congested);
	if (m->window >= len)
		return false;

437
	return tipc_group_cong(grp, m->node, m->port, len, &m);
438 439
}

440 441 442 443 444 445 446 447 448
/* tipc_group_sort_msg() - sort msg into queue by bcast sequence number
 */
static void tipc_group_sort_msg(struct sk_buff *skb, struct sk_buff_head *defq)
{
	struct tipc_msg *_hdr, *hdr = buf_msg(skb);
	u16 bc_seqno = msg_grp_bc_seqno(hdr);
	struct sk_buff *_skb, *tmp;
	int mtyp = msg_type(hdr);

449
	/* Bcast/mcast may be bypassed by ucast or other bcast, - sort it in */
450 451 452 453 454 455 456 457 458 459 460 461 462 463
	if (mtyp == TIPC_GRP_BCAST_MSG || mtyp == TIPC_GRP_MCAST_MSG) {
		skb_queue_walk_safe(defq, _skb, tmp) {
			_hdr = buf_msg(_skb);
			if (!less(bc_seqno, msg_grp_bc_seqno(_hdr)))
				continue;
			__skb_queue_before(defq, _skb, skb);
			return;
		}
		/* Bcast was not bypassed, - add to tail */
	}
	/* Unicasts are never bypassed, - always add to tail */
	__skb_queue_tail(defq, skb);
}

J
Jon Maloy 已提交
464 465 466 467 468 469
/* 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);
470
	bool ack, deliver, update, leave = false;
471
	struct sk_buff_head *defq;
J
Jon Maloy 已提交
472 473 474
	struct tipc_member *m;
	struct tipc_msg *hdr;
	u32 node, port;
475
	int mtyp, blks;
J
Jon Maloy 已提交
476 477 478 479 480 481 482 483 484 485 486 487 488 489 490

	if (!skb)
		return;

	hdr = buf_msg(skb);
	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;

491 492
	if (less(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt))
		goto drop;
493

494 495 496 497 498 499 500
	TIPC_SKB_CB(skb)->orig_member = m->instance;
	defq = &m->deferredq;
	tipc_group_sort_msg(skb, defq);

	while ((skb = skb_peek(defq))) {
		hdr = buf_msg(skb);
		mtyp = msg_type(hdr);
501
		blks = msg_blocks(hdr);
502
		deliver = true;
503
		ack = false;
504 505 506 507 508 509 510 511 512 513 514 515 516 517 518
		update = false;

		if (more(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt))
			break;

		/* Decide what to do with message */
		switch (mtyp) {
		case TIPC_GRP_MCAST_MSG:
			if (msg_nameinst(hdr) != grp->instance) {
				update = true;
				deliver = false;
			}
			/* Fall thru */
		case TIPC_GRP_BCAST_MSG:
			m->bc_rcv_nxt++;
519
			ack = msg_grp_bc_ack_req(hdr);
520 521 522
			break;
		case TIPC_GRP_UCAST_MSG:
			break;
523 524 525 526 527 528
		case TIPC_GRP_MEMBER_EVT:
			if (m->state == MBR_LEAVING)
				leave = true;
			if (!grp->events)
				deliver = false;
			break;
529 530
		default:
			break;
531 532
		}

533 534 535 536 537 538 539
		/* Execute decisions */
		__skb_dequeue(defq);
		if (deliver)
			__skb_queue_tail(inputq, skb);
		else
			kfree_skb(skb);

540 541 542
		if (ack)
			tipc_group_proto_xmit(grp, m, GRP_ACK_MSG, xmitq);

543 544
		if (leave) {
			__skb_queue_purge(defq);
J
Jon Maloy 已提交
545
			tipc_group_delete_member(grp, m);
546 547
			break;
		}
548 549
		if (!update)
			continue;
J
Jon Maloy 已提交
550

551 552
		tipc_group_update_rcv_win(grp, blks, node, port, xmitq);
	}
J
Jon Maloy 已提交
553 554 555 556 557
	return;
drop:
	kfree_skb(skb);
}

558 559 560
void tipc_group_update_rcv_win(struct tipc_group *grp, int blks, u32 node,
			       u32 port, struct sk_buff_head *xmitq)
{
561 562 563 564 565
	struct list_head *active = &grp->active;
	int max_active = grp->max_active;
	int reclaim_limit = max_active * 3 / 4;
	int active_cnt = grp->active_cnt;
	struct tipc_member *m, *rm;
566 567 568 569 570 571 572 573 574

	m = tipc_group_find_member(grp, node, port);
	if (!m)
		return;

	m->advertised -= blks;

	switch (m->state) {
	case MBR_JOINED:
575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605
		/* Reclaim advertised space from least active member */
		if (!list_empty(active) && active_cnt >= reclaim_limit) {
			rm = list_first_entry(active, struct tipc_member, list);
			rm->state = MBR_RECLAIMING;
			list_move_tail(&rm->list, &grp->reclaiming);
			tipc_group_proto_xmit(grp, rm, GRP_RECLAIM_MSG, xmitq);
		}
		/* If max active, become pending and wait for reclaimed space */
		if (active_cnt >= max_active) {
			m->state = MBR_PENDING;
			list_add_tail(&m->list, &grp->pending);
			break;
		}
		/* Otherwise become active */
		m->state = MBR_ACTIVE;
		list_add_tail(&m->list, &grp->active);
		grp->active_cnt++;
		/* Fall through */
	case MBR_ACTIVE:
		if (!list_is_last(&m->list, &grp->active))
			list_move_tail(&m->list, &grp->active);
		if (m->advertised > (ADV_ACTIVE * 3 / 4))
			break;
		tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
		break;
	case MBR_REMITTED:
		if (m->advertised > ADV_IDLE)
			break;
		m->state = MBR_JOINED;
		if (m->advertised < ADV_IDLE) {
			pr_warn_ratelimited("Rcv unexpected msg after REMIT\n");
606
			tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
607
		}
608
		break;
609
	case MBR_RECLAIMING:
610 611 612 613 614 615 616 617
	case MBR_DISCOVERED:
	case MBR_JOINING:
	case MBR_LEAVING:
	default:
		break;
	}
}

J
Jon Maloy 已提交
618 619 620 621 622
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;
623
	int adv = 0;
J
Jon Maloy 已提交
624 625 626 627 628 629 630

	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;

631
	if (m->state == MBR_ACTIVE)
632
		adv = ADV_ACTIVE - m->advertised;
633 634
	else if (m->state == MBR_JOINED || m->state == MBR_PENDING)
		adv = ADV_IDLE - m->advertised;
635

J
Jon Maloy 已提交
636
	hdr = buf_msg(skb);
637 638

	if (mtyp == GRP_JOIN_MSG) {
J
Jon Maloy 已提交
639
		msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
640 641
		msg_set_adv_win(hdr, adv);
		m->advertised += adv;
642 643
	} else if (mtyp == GRP_LEAVE_MSG) {
		msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
644 645 646
	} else if (mtyp == GRP_ADV_MSG) {
		msg_set_adv_win(hdr, adv);
		m->advertised += adv;
647 648
	} else if (mtyp == GRP_ACK_MSG) {
		msg_set_grp_bc_acked(hdr, m->bc_rcv_nxt);
649 650
	} else if (mtyp == GRP_REMIT_MSG) {
		msg_set_grp_remitted(hdr, m->window);
651
	}
J
Jon Maloy 已提交
652
	msg_set_dest_droppable(hdr, true);
J
Jon Maloy 已提交
653 654 655
	__skb_queue_tail(xmitq, skb);
}

656 657
void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup,
			  struct tipc_msg *hdr, struct sk_buff_head *inputq,
J
Jon Maloy 已提交
658 659 660 661
			  struct sk_buff_head *xmitq)
{
	u32 node = msg_orignode(hdr);
	u32 port = msg_origport(hdr);
662
	struct tipc_member *m, *pm;
663
	struct tipc_msg *ehdr;
664
	u16 remitted, in_flight;
J
Jon Maloy 已提交
665 666 667 668 669 670 671 672 673 674 675 676 677

	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;
678 679
		m->bc_syncpt = msg_grp_bc_syncpt(hdr);
		m->bc_rcv_nxt = m->bc_syncpt;
680
		m->window += msg_adv_win(hdr);
J
Jon Maloy 已提交
681 682

		/* Wait until PUBLISH event is received */
683
		if (m->state == MBR_DISCOVERED) {
J
Jon Maloy 已提交
684
			m->state = MBR_JOINING;
685
		} else if (m->state == MBR_PUBLISHED) {
J
Jon Maloy 已提交
686
			m->state = MBR_JOINED;
687 688 689
			*usr_wakeup = true;
			m->usr_pending = false;
			tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
690 691
			ehdr = buf_msg(m->event_msg);
			msg_set_grp_bc_seqno(ehdr, m->bc_syncpt);
692 693
			__skb_queue_tail(inputq, m->event_msg);
		}
694 695
		list_del_init(&m->congested);
		tipc_group_update_member(m, 0);
J
Jon Maloy 已提交
696 697 698 699
		return;
	case GRP_LEAVE_MSG:
		if (!m)
			return;
700
		m->bc_syncpt = msg_grp_bc_syncpt(hdr);
701 702 703
		list_del_init(&m->list);
		list_del_init(&m->congested);
		*usr_wakeup = true;
J
Jon Maloy 已提交
704 705 706

		/* Wait until WITHDRAW event is received */
		if (m->state != MBR_LEAVING) {
707
			tipc_group_decr_active(grp, m);
J
Jon Maloy 已提交
708 709 710 711
			m->state = MBR_LEAVING;
			return;
		}
		/* Otherwise deliver already received WITHDRAW event */
712 713
		ehdr = buf_msg(m->event_msg);
		msg_set_grp_bc_seqno(ehdr, m->bc_syncpt);
714
		__skb_queue_tail(inputq, m->event_msg);
715 716 717 718 719 720 721 722
		return;
	case GRP_ADV_MSG:
		if (!m)
			return;
		m->window += msg_adv_win(hdr);
		*usr_wakeup = m->usr_pending;
		m->usr_pending = false;
		list_del_init(&m->congested);
J
Jon Maloy 已提交
723
		return;
724 725 726 727 728 729 730 731 732
	case GRP_ACK_MSG:
		if (!m)
			return;
		m->bc_acked = msg_grp_bc_acked(hdr);
		if (--grp->bc_ackers)
			break;
		*usr_wakeup = true;
		m->usr_pending = false;
		return;
733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774
	case GRP_RECLAIM_MSG:
		if (!m)
			return;
		*usr_wakeup = m->usr_pending;
		m->usr_pending = false;
		tipc_group_proto_xmit(grp, m, GRP_REMIT_MSG, xmitq);
		m->window = ADV_IDLE;
		return;
	case GRP_REMIT_MSG:
		if (!m || m->state != MBR_RECLAIMING)
			return;

		list_del_init(&m->list);
		grp->active_cnt--;
		remitted = msg_grp_remitted(hdr);

		/* Messages preceding the REMIT still in receive queue */
		if (m->advertised > remitted) {
			m->state = MBR_REMITTED;
			in_flight = m->advertised - remitted;
		}
		/* All messages preceding the REMIT have been read */
		if (m->advertised <= remitted) {
			m->state = MBR_JOINED;
			in_flight = 0;
		}
		/* ..and the REMIT overtaken by more messages => re-advertise */
		if (m->advertised < remitted)
			tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);

		m->advertised = ADV_IDLE + in_flight;

		/* Set oldest pending member to active and advertise */
		if (list_empty(&grp->pending))
			return;
		pm = list_first_entry(&grp->pending, struct tipc_member, list);
		pm->state = MBR_ACTIVE;
		list_move_tail(&pm->list, &grp->active);
		grp->active_cnt++;
		if (pm->advertised <= (ADV_ACTIVE * 3 / 4))
			tipc_group_proto_xmit(grp, pm, GRP_ADV_MSG, xmitq);
		return;
J
Jon Maloy 已提交
775 776 777 778 779
	default:
		pr_warn("Received unknown GROUP_PROTO message\n");
	}
}

780 781
/* tipc_group_member_evt() - receive and handle a member up/down event
 */
J
Jon Maloy 已提交
782
void tipc_group_member_evt(struct tipc_group *grp,
783 784
			   bool *usr_wakeup,
			   int *sk_rcvbuf,
J
Jon Maloy 已提交
785
			   struct sk_buff *skb,
786
			   struct sk_buff_head *inputq,
J
Jon Maloy 已提交
787 788 789 790
			   struct sk_buff_head *xmitq)
{
	struct tipc_msg *hdr = buf_msg(skb);
	struct tipc_event *evt = (void *)msg_data(hdr);
791
	u32 instance = evt->found_lower;
J
Jon Maloy 已提交
792 793
	u32 node = evt->port.node;
	u32 port = evt->port.ref;
794
	int event = evt->event;
J
Jon Maloy 已提交
795 796
	struct tipc_member *m;
	struct net *net;
797
	bool node_up;
J
Jon Maloy 已提交
798 799 800 801 802 803 804 805 806 807
	u32 self;

	if (!grp)
		goto drop;

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

808 809 810 811 812 813 814 815 816
	/* Convert message before delivery to user */
	msg_set_hdr_sz(hdr, GROUP_H_SIZE);
	msg_set_user(hdr, TIPC_CRITICAL_IMPORTANCE);
	msg_set_type(hdr, TIPC_GRP_MEMBER_EVT);
	msg_set_origport(hdr, port);
	msg_set_orignode(hdr, node);
	msg_set_nametype(hdr, grp->type);
	msg_set_grp_evt(hdr, event);

J
Jon Maloy 已提交
817 818
	m = tipc_group_find_member(grp, node, port);

819
	if (event == TIPC_PUBLISHED) {
J
Jon Maloy 已提交
820 821 822 823 824 825
		if (!m)
			m = tipc_group_create_member(grp, node, port,
						     MBR_DISCOVERED);
		if (!m)
			goto drop;

826 827 828
		/* Hold back event if JOIN message not yet received */
		if (m->state == MBR_DISCOVERED) {
			m->event_msg = skb;
J
Jon Maloy 已提交
829
			m->state = MBR_PUBLISHED;
830
		} else {
831
			msg_set_grp_bc_seqno(hdr, m->bc_syncpt);
832
			__skb_queue_tail(inputq, skb);
J
Jon Maloy 已提交
833
			m->state = MBR_JOINED;
834 835
			*usr_wakeup = true;
			m->usr_pending = false;
836 837 838
		}
		m->instance = instance;
		TIPC_SKB_CB(skb)->orig_member = m->instance;
J
Jon Maloy 已提交
839
		tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
840 841 842 843
		if (m->window < ADV_IDLE)
			tipc_group_update_member(m, 0);
		else
			list_del_init(&m->congested);
844
	} else if (event == TIPC_WITHDRAWN) {
J
Jon Maloy 已提交
845 846 847
		if (!m)
			goto drop;

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

850
		*usr_wakeup = true;
851
		m->usr_pending = false;
852
		node_up = tipc_node_is_up(net, node);
853 854 855 856 857 858 859 860 861
		m->event_msg = NULL;

		if (node_up) {
			/* Hold back event if a LEAVE msg should be expected */
			if (m->state != MBR_LEAVING) {
				m->event_msg = skb;
				tipc_group_decr_active(grp, m);
				m->state = MBR_LEAVING;
			} else {
862
				msg_set_grp_bc_seqno(hdr, m->bc_syncpt);
863 864 865 866 867 868
				__skb_queue_tail(inputq, skb);
			}
		} else {
			if (m->state != MBR_LEAVING) {
				tipc_group_decr_active(grp, m);
				m->state = MBR_LEAVING;
869
				msg_set_grp_bc_seqno(hdr, m->bc_rcv_nxt);
870 871 872
			} else {
				msg_set_grp_bc_seqno(hdr, m->bc_syncpt);
			}
873 874
			__skb_queue_tail(inputq, skb);
		}
875
		list_del_init(&m->list);
876
		list_del_init(&m->congested);
J
Jon Maloy 已提交
877
	}
878
	*sk_rcvbuf = tipc_group_rcvbuf_limit(grp);
879
	return;
J
Jon Maloy 已提交
880 881 882
drop:
	kfree_skb(skb);
}