group.c 21.5 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 354 355 356 357 358 359 360 361 362 363 364 365 366
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;

	if (!list_empty(&m->congested))
		return;

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

367
void tipc_group_update_bc_members(struct tipc_group *grp, int len, bool ack)
J
Jon Maloy 已提交
368
{
369
	u16 prev = grp->bc_snd_nxt - 1;
370 371 372 373 374
	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);
375
		if (tipc_group_is_enabled(m)) {
376
			tipc_group_update_member(m, len);
377 378
			m->bc_acked = prev;
		}
379
	}
380 381 382 383

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

387 388
bool tipc_group_cong(struct tipc_group *grp, u32 dnode, u32 dport,
		     int len, struct tipc_member **mbr)
389
{
390
	struct sk_buff_head xmitq;
391
	struct tipc_member *m;
392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410
	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;
411 412 413 414
	if (state == MBR_ACTIVE && adv == ADV_ACTIVE)
		return true;
	if (state == MBR_PENDING && adv == ADV_IDLE)
		return true;
415 416 417 418 419 420 421 422 423
	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;
424

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

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

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

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

439 440 441 442 443 444 445 446 447
/* 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);

448
	/* Bcast/mcast may be bypassed by ucast or other bcast, - sort it in */
449 450 451 452 453 454 455 456 457 458 459 460 461 462
	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 已提交
463 464 465 466 467 468
/* 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);
469
	bool ack, deliver, update, leave = false;
470
	struct sk_buff_head *defq;
J
Jon Maloy 已提交
471 472 473
	struct tipc_member *m;
	struct tipc_msg *hdr;
	u32 node, port;
474
	int mtyp, blks;
J
Jon Maloy 已提交
475 476 477 478 479 480 481 482 483 484 485 486 487 488 489

	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;

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

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);
		deliver = true;
501
		ack = false;
502 503 504 505 506 507 508 509 510 511 512 513 514 515 516
		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++;
517
			ack = msg_grp_bc_ack_req(hdr);
518 519 520
			break;
		case TIPC_GRP_UCAST_MSG:
			break;
521 522 523 524 525 526
		case TIPC_GRP_MEMBER_EVT:
			if (m->state == MBR_LEAVING)
				leave = true;
			if (!grp->events)
				deliver = false;
			break;
527 528
		default:
			break;
529 530
		}

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

538 539 540
		if (ack)
			tipc_group_proto_xmit(grp, m, GRP_ACK_MSG, xmitq);

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

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

557 558 559
void tipc_group_update_rcv_win(struct tipc_group *grp, int blks, u32 node,
			       u32 port, struct sk_buff_head *xmitq)
{
560 561 562 563 564
	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;
565 566 567 568 569 570 571 572 573

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

	m->advertised -= blks;

	switch (m->state) {
	case MBR_JOINED:
574 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
		/* 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");
605
			tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
606
		}
607
		break;
608
	case MBR_RECLAIMING:
609 610 611 612 613 614 615 616
	case MBR_DISCOVERED:
	case MBR_JOINING:
	case MBR_LEAVING:
	default:
		break;
	}
}

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

	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;

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

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

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

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

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

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

		/* Wait until WITHDRAW event is received */
		if (m->state != MBR_LEAVING) {
704
			tipc_group_decr_active(grp, m);
J
Jon Maloy 已提交
705 706 707 708
			m->state = MBR_LEAVING;
			return;
		}
		/* Otherwise deliver already received WITHDRAW event */
709 710
		ehdr = buf_msg(m->event_msg);
		msg_set_grp_bc_seqno(ehdr, m->bc_syncpt);
711
		__skb_queue_tail(inputq, m->event_msg);
712
		*usr_wakeup = true;
713 714 715 716 717 718 719 720 721
		list_del_init(&m->congested);
		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 已提交
722
		return;
723 724 725 726 727 728 729 730 731
	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;
732 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
	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 已提交
774 775 776 777 778
	default:
		pr_warn("Received unknown GROUP_PROTO message\n");
	}
}

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

	if (!grp)
		goto drop;

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

807 808 809 810 811 812 813 814 815
	/* 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 已提交
816 817
	m = tipc_group_find_member(grp, node, port);

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

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

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

849
		*usr_wakeup = true;
850
		m->usr_pending = false;
851
		node_up = tipc_node_is_up(net, node);
852

853
		/* Hold back event if more messages might be expected */
854
		if (m->state != MBR_LEAVING && node_up) {
855
			m->event_msg = skb;
856
			tipc_group_decr_active(grp, m);
J
Jon Maloy 已提交
857
			m->state = MBR_LEAVING;
858
		} else {
859 860 861 862
			if (node_up)
				msg_set_grp_bc_seqno(hdr, m->bc_syncpt);
			else
				msg_set_grp_bc_seqno(hdr, m->bc_rcv_nxt);
863 864
			__skb_queue_tail(inputq, skb);
		}
865
		list_del_init(&m->congested);
J
Jon Maloy 已提交
866
	}
867
	*sk_rcvbuf = tipc_group_rcvbuf_limit(grp);
868
	return;
J
Jon Maloy 已提交
869 870 871
drop:
	kfree_skb(skb);
}