bcast.c 21.6 KB
Newer Older
P
Per Liden 已提交
1 2
/*
 * net/tipc/bcast.c: TIPC broadcast code
3
 *
4
 * Copyright (c) 2004-2006, 2014-2017, Ericsson AB
P
Per Liden 已提交
5
 * Copyright (c) 2004, Intel Corporation.
6
 * Copyright (c) 2005, 2010-2011, Wind River Systems
P
Per Liden 已提交
7 8
 * All rights reserved.
 *
P
Per Liden 已提交
9
 * Redistribution and use in source and binary forms, with or without
P
Per Liden 已提交
10 11
 * modification, are permitted provided that the following conditions are met:
 *
P
Per Liden 已提交
12 13 14 15 16 17 18 19
 * 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.
P
Per Liden 已提交
20
 *
P
Per Liden 已提交
21 22 23 24 25 26 27 28 29 30 31 32 33 34
 * 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
P
Per Liden 已提交
35 36 37
 * POSSIBILITY OF SUCH DAMAGE.
 */

38
#include <linux/tipc_config.h>
39 40
#include "socket.h"
#include "msg.h"
P
Per Liden 已提交
41
#include "bcast.h"
42
#include "link.h"
43
#include "name_table.h"
P
Per Liden 已提交
44

45 46
#define BCLINK_WIN_DEFAULT  50	/* bcast link window size (default) */
#define BCLINK_WIN_MIN      32	/* bcast minimum link window size */
P
Per Liden 已提交
47

48
const char tipc_bclink_name[] = "broadcast-link";
P
Per Liden 已提交
49

50
/**
51
 * struct tipc_bc_base - base structure for keeping broadcast send state
52
 * @link: broadcast send link structure
53
 * @inputq: data input queue; will only carry SOCK_WAKEUP messages
54
 * @dests: array keeping number of reachable destinations per bearer
55
 * @primary_bearer: a bearer having links to all broadcast destinations, if any
56
 * @bcast_support: indicates if primary bearer, if any, supports broadcast
57
 * @force_bcast: forces broadcast for multicast traffic
58
 * @rcast_support: indicates if all peer nodes support replicast
59
 * @force_rcast: forces replicast for multicast traffic
60
 * @rc_ratio: dest count as percentage of cluster size where send method changes
61
 * @bc_threshold: calculated from rc_ratio; if dests > threshold use broadcast
62 63
 */
struct tipc_bc_base {
64
	struct tipc_link *link;
65
	struct sk_buff_head inputq;
66 67
	int dests[MAX_BEARERS];
	int primary_bearer;
68
	bool bcast_support;
69
	bool force_bcast;
70
	bool rcast_support;
71
	bool force_rcast;
72 73
	int rc_ratio;
	int bc_threshold;
74 75
};

76 77 78 79 80
static struct tipc_bc_base *tipc_bc_base(struct net *net)
{
	return tipc_net(net)->bcbase;
}

81 82 83 84
/* tipc_bcast_get_mtu(): -get the MTU currently used by broadcast link
 * Note: the MTU is decremented to give room for a tunnel header, in
 * case the message needs to be sent as replicast
 */
85
int tipc_bcast_get_mtu(struct net *net)
86
{
87
	return tipc_link_mss(tipc_bc_sndlink(net));
88 89
}

90
void tipc_bcast_toggle_rcast(struct net *net, bool supp)
91
{
92
	tipc_bc_base(net)->rcast_support = supp;
93 94 95 96 97 98 99 100 101 102
}

static void tipc_bcbase_calc_bc_threshold(struct net *net)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);
	int cluster_size = tipc_link_bc_peers(tipc_bc_sndlink(net));

	bb->bc_threshold = 1 + (cluster_size * bb->rc_ratio / 100);
}

103 104 105 106 107 108 109
/* tipc_bcbase_select_primary(): find a bearer with links to all destinations,
 *                               if any, and make it primary bearer
 */
static void tipc_bcbase_select_primary(struct net *net)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);
	int all_dests =  tipc_link_bc_peers(bb->link);
110
	int i, mtu, prim;
111 112

	bb->primary_bearer = INVALID_BEARER_ID;
113
	bb->bcast_support = true;
114 115 116 117 118

	if (!all_dests)
		return;

	for (i = 0; i < MAX_BEARERS; i++) {
119 120 121 122 123 124
		if (!bb->dests[i])
			continue;

		mtu = tipc_bearer_mtu(net, i);
		if (mtu < tipc_link_mtu(bb->link))
			tipc_link_set_mtu(bb->link, mtu);
125
		bb->bcast_support &= tipc_bearer_bcast_support(net, i);
126 127 128 129 130 131 132 133 134
		if (bb->dests[i] < all_dests)
			continue;

		bb->primary_bearer = i;

		/* Reduce risk that all nodes select same primary */
		if ((i ^ tipc_own_addr(net)) & 1)
			break;
	}
135 136 137
	prim = bb->primary_bearer;
	if (prim != INVALID_BEARER_ID)
		bb->bcast_support = tipc_bearer_bcast_support(net, prim);
138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187
}

void tipc_bcast_inc_bearer_dst_cnt(struct net *net, int bearer_id)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);

	tipc_bcast_lock(net);
	bb->dests[bearer_id]++;
	tipc_bcbase_select_primary(net);
	tipc_bcast_unlock(net);
}

void tipc_bcast_dec_bearer_dst_cnt(struct net *net, int bearer_id)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);

	tipc_bcast_lock(net);
	bb->dests[bearer_id]--;
	tipc_bcbase_select_primary(net);
	tipc_bcast_unlock(net);
}

/* tipc_bcbase_xmit - broadcast a packet queue across one or more bearers
 *
 * Note that number of reachable destinations, as indicated in the dests[]
 * array, may transitionally differ from the number of destinations indicated
 * in each sent buffer. We can sustain this. Excess destination nodes will
 * drop and never acknowledge the unexpected packets, and missing destinations
 * will either require retransmission (if they are just about to be added to
 * the bearer), or be removed from the buffer's 'ackers' counter (if they
 * just went down)
 */
static void tipc_bcbase_xmit(struct net *net, struct sk_buff_head *xmitq)
{
	int bearer_id;
	struct tipc_bc_base *bb = tipc_bc_base(net);
	struct sk_buff *skb, *_skb;
	struct sk_buff_head _xmitq;

	if (skb_queue_empty(xmitq))
		return;

	/* The typical case: at least one bearer has links to all nodes */
	bearer_id = bb->primary_bearer;
	if (bearer_id >= 0) {
		tipc_bearer_bc_xmit(net, bearer_id, xmitq);
		return;
	}

	/* We have to transmit across all bearers */
188
	__skb_queue_head_init(&_xmitq);
189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204
	for (bearer_id = 0; bearer_id < MAX_BEARERS; bearer_id++) {
		if (!bb->dests[bearer_id])
			continue;

		skb_queue_walk(xmitq, skb) {
			_skb = pskb_copy_for_clone(skb, GFP_ATOMIC);
			if (!_skb)
				break;
			__skb_queue_tail(&_xmitq, _skb);
		}
		tipc_bearer_bc_xmit(net, bearer_id, &_xmitq);
	}
	__skb_queue_purge(xmitq);
	__skb_queue_purge(&_xmitq);
}

205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222
static void tipc_bcast_select_xmit_method(struct net *net, int dests,
					  struct tipc_mc_method *method)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);
	unsigned long exp = method->expires;

	/* Broadcast supported by used bearer/bearers? */
	if (!bb->bcast_support) {
		method->rcast = true;
		return;
	}
	/* Any destinations which don't support replicast ? */
	if (!bb->rcast_support) {
		method->rcast = false;
		return;
	}
	/* Can current method be changed ? */
	method->expires = jiffies + TIPC_METHOD_EXPIRE;
223
	if (method->mandatory)
224 225
		return;

226 227 228 229 230 231 232 233 234 235 236 237 238 239 240
	if (!(tipc_net(net)->capabilities & TIPC_MCAST_RBCTL) &&
	    time_before(jiffies, exp))
		return;

	/* Configuration as force 'broadcast' method */
	if (bb->force_bcast) {
		method->rcast = false;
		return;
	}
	/* Configuration as force 'replicast' method */
	if (bb->force_rcast) {
		method->rcast = true;
		return;
	}
	/* Configuration as 'autoselect' or default method */
241 242 243 244
	/* Determine method to use now */
	method->rcast = dests <= bb->bc_threshold;
}

245
/* tipc_bcast_xmit - broadcast the buffer chain to all external nodes
246
 * @net: the applicable net namespace
247 248
 * @pkts: chain of buffers containing message
 * @cong_link_cnt: set to 1 if broadcast link is congested, otherwise 0
249
 * Consumes the buffer chain.
250
 * Returns 0 if success, otherwise errno: -EHOSTUNREACH,-EMSGSIZE
251
 */
252 253
static int tipc_bcast_xmit(struct net *net, struct sk_buff_head *pkts,
			   u16 *cong_link_cnt)
254
{
255
	struct tipc_link *l = tipc_bc_sndlink(net);
256
	struct sk_buff_head xmitq;
257 258
	int rc = 0;

259
	__skb_queue_head_init(&xmitq);
260 261
	tipc_bcast_lock(net);
	if (tipc_link_bc_peers(l))
262
		rc = tipc_link_xmit(l, pkts, &xmitq);
263
	tipc_bcast_unlock(net);
264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283
	tipc_bcbase_xmit(net, &xmitq);
	__skb_queue_purge(pkts);
	if (rc == -ELINKCONG) {
		*cong_link_cnt = 1;
		rc = 0;
	}
	return rc;
}

/* tipc_rcast_xmit - replicate and send a message to given destination nodes
 * @net: the applicable net namespace
 * @pkts: chain of buffers containing message
 * @dests: list of destination nodes
 * @cong_link_cnt: returns number of congested links
 * @cong_links: returns identities of congested links
 * Returns 0 if success, otherwise errno
 */
static int tipc_rcast_xmit(struct net *net, struct sk_buff_head *pkts,
			   struct tipc_nlist *dests, u16 *cong_link_cnt)
{
J
Jon Maloy 已提交
284
	struct tipc_dest *dst, *tmp;
285
	struct sk_buff_head _pkts;
J
Jon Maloy 已提交
286
	u32 dnode, selector;
287 288

	selector = msg_link_selector(buf_msg(skb_peek(pkts)));
289
	__skb_queue_head_init(&_pkts);
290

J
Jon Maloy 已提交
291 292 293
	list_for_each_entry_safe(dst, tmp, &dests->list, list) {
		dnode = dst->node;
		if (!tipc_msg_pskb_copy(dnode, pkts, &_pkts))
294
			return -ENOMEM;
295

296
		/* Any other return value than -ELINKCONG is ignored */
J
Jon Maloy 已提交
297
		if (tipc_node_xmit(net, &_pkts, dnode, selector) == -ELINKCONG)
298
			(*cong_link_cnt)++;
299
	}
300 301
	return 0;
}
302

303 304 305 306 307 308 309 310 311
/* tipc_mcast_send_sync - deliver a dummy message with SYN bit
 * @net: the applicable net namespace
 * @skb: socket buffer to copy
 * @method: send method to be used
 * @dests: destination nodes for message.
 * Returns 0 if success, otherwise errno
 */
static int tipc_mcast_send_sync(struct net *net, struct sk_buff *skb,
				struct tipc_mc_method *method,
312
				struct tipc_nlist *dests)
313 314 315 316
{
	struct tipc_msg *hdr, *_hdr;
	struct sk_buff_head tmpq;
	struct sk_buff *_skb;
317 318
	u16 cong_link_cnt;
	int rc = 0;
319 320 321 322 323 324 325

	/* Is a cluster supporting with new capabilities ? */
	if (!(tipc_net(net)->capabilities & TIPC_MCAST_RBCTL))
		return 0;

	hdr = buf_msg(skb);
	if (msg_user(hdr) == MSG_FRAGMENTER)
326
		hdr = msg_inner_hdr(hdr);
327 328 329 330 331
	if (msg_type(hdr) != TIPC_MCAST_MSG)
		return 0;

	/* Allocate dummy message */
	_skb = tipc_buf_acquire(MCAST_H_SIZE, GFP_KERNEL);
332
	if (!_skb)
333 334 335 336 337 338 339 340 341 342 343 344 345
		return -ENOMEM;

	/* Preparing for 'synching' header */
	msg_set_syn(hdr, 1);

	/* Copy skb's header into a dummy header */
	skb_copy_to_linear_data(_skb, hdr, MCAST_H_SIZE);
	skb_orphan(_skb);

	/* Reverse method for dummy message */
	_hdr = buf_msg(_skb);
	msg_set_size(_hdr, MCAST_H_SIZE);
	msg_set_is_rcast(_hdr, !msg_is_rcast(hdr));
346
	msg_set_errcode(_hdr, TIPC_ERR_NO_PORT);
347

348
	__skb_queue_head_init(&tmpq);
349 350
	__skb_queue_tail(&tmpq, _skb);
	if (method->rcast)
351
		rc = tipc_bcast_xmit(net, &tmpq, &cong_link_cnt);
352
	else
353
		rc = tipc_rcast_xmit(net, &tmpq, dests, &cong_link_cnt);
354 355 356 357

	/* This queue should normally be empty by now */
	__skb_queue_purge(&tmpq);

358
	return rc;
359 360
}

361 362 363 364
/* tipc_mcast_xmit - deliver message to indicated destination nodes
 *                   and to identified node local sockets
 * @net: the applicable net namespace
 * @pkts: chain of buffers containing message
365 366
 * @method: send method to be used
 * @dests: destination nodes for message.
367 368 369 370 371
 * @cong_link_cnt: returns number of encountered congested destination links
 * Consumes buffer chain.
 * Returns 0 if success, otherwise errno
 */
int tipc_mcast_xmit(struct net *net, struct sk_buff_head *pkts,
372 373
		    struct tipc_mc_method *method, struct tipc_nlist *dests,
		    u16 *cong_link_cnt)
374 375
{
	struct sk_buff_head inputq, localq;
376 377 378
	bool rcast = method->rcast;
	struct tipc_msg *hdr;
	struct sk_buff *skb;
379 380 381
	int rc = 0;

	skb_queue_head_init(&inputq);
382
	__skb_queue_head_init(&localq);
383 384 385 386 387 388

	/* Clone packets before they are consumed by next call */
	if (dests->local && !tipc_msg_reassemble(pkts, &localq)) {
		rc = -ENOMEM;
		goto exit;
	}
389
	/* Send according to determined transmit method */
390
	if (dests->remote) {
391
		tipc_bcast_select_xmit_method(net, dests->remote, method);
392 393 394 395

		skb = skb_peek(pkts);
		hdr = buf_msg(skb);
		if (msg_user(hdr) == MSG_FRAGMENTER)
396
			hdr = msg_inner_hdr(hdr);
397 398 399
		msg_set_is_rcast(hdr, method->rcast);

		/* Switch method ? */
400 401 402 403 404 405 406 407
		if (rcast != method->rcast) {
			rc = tipc_mcast_send_sync(net, skb, method, dests);
			if (unlikely(rc)) {
				pr_err("Unable to send SYN: method %d, rc %d\n",
				       rcast, rc);
				goto exit;
			}
		}
408

409
		if (method->rcast)
410 411 412 413 414
			rc = tipc_rcast_xmit(net, pkts, dests, cong_link_cnt);
		else
			rc = tipc_bcast_xmit(net, pkts, cong_link_cnt);
	}

415 416
	if (dests->local) {
		tipc_loopback_trace(net, &localq);
417
		tipc_sk_mcast_rcv(net, &localq, &inputq);
418
	}
419
exit:
420
	/* This queue should normally be empty by now */
421
	__skb_queue_purge(pkts);
422
	return rc;
423
}
424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449

/* tipc_bcast_rcv - receive a broadcast packet, and deliver to rcv link
 *
 * RCU is locked, no other locks set
 */
int tipc_bcast_rcv(struct net *net, struct tipc_link *l, struct sk_buff *skb)
{
	struct tipc_msg *hdr = buf_msg(skb);
	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	struct sk_buff_head xmitq;
	int rc;

	__skb_queue_head_init(&xmitq);

	if (msg_mc_netid(hdr) != tipc_netid(net) || !tipc_link_is_up(l)) {
		kfree_skb(skb);
		return 0;
	}

	tipc_bcast_lock(net);
	if (msg_user(hdr) == BCAST_PROTOCOL)
		rc = tipc_link_bc_nack_rcv(l, skb, &xmitq);
	else
		rc = tipc_link_rcv(l, skb, NULL);
	tipc_bcast_unlock(net);

450
	tipc_bcbase_xmit(net, &xmitq);
451 452 453 454 455 456 457 458 459 460 461 462

	/* Any socket wakeup messages ? */
	if (!skb_queue_empty(inputq))
		tipc_sk_rcv(net, inputq);

	return rc;
}

/* tipc_bcast_ack_rcv - receive and handle a broadcast acknowledge
 *
 * RCU is locked, no other locks set
 */
463 464
void tipc_bcast_ack_rcv(struct net *net, struct tipc_link *l,
			struct tipc_msg *hdr)
465 466
{
	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
467
	u16 acked = msg_bcast_ack(hdr);
468 469
	struct sk_buff_head xmitq;

470 471 472 473
	/* Ignore bc acks sent by peer before bcast synch point was received */
	if (msg_bc_ack_invalid(hdr))
		return;

474 475 476 477 478 479
	__skb_queue_head_init(&xmitq);

	tipc_bcast_lock(net);
	tipc_link_bc_ack_rcv(l, acked, &xmitq);
	tipc_bcast_unlock(net);

480
	tipc_bcbase_xmit(net, &xmitq);
481 482 483 484 485 486 487 488 489 490

	/* Any socket wakeup messages ? */
	if (!skb_queue_empty(inputq))
		tipc_sk_rcv(net, inputq);
}

/* tipc_bcast_synch_rcv -  check and update rcv link with peer's send state
 *
 * RCU is locked, no other locks set
 */
491 492
int tipc_bcast_sync_rcv(struct net *net, struct tipc_link *l,
			struct tipc_msg *hdr)
493 494 495
{
	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
	struct sk_buff_head xmitq;
496
	int rc = 0;
497 498 499 500

	__skb_queue_head_init(&xmitq);

	tipc_bcast_lock(net);
501 502 503
	if (msg_type(hdr) != STATE_MSG) {
		tipc_link_bc_init_rcv(l, hdr);
	} else if (!msg_bc_ack_invalid(hdr)) {
504
		tipc_link_bc_ack_rcv(l, msg_bcast_ack(hdr), &xmitq);
505
		rc = tipc_link_bc_sync_rcv(l, hdr, &xmitq);
506 507 508
	}
	tipc_bcast_unlock(net);

509
	tipc_bcbase_xmit(net, &xmitq);
510 511 512 513

	/* Any socket wakeup messages ? */
	if (!skb_queue_empty(inputq))
		tipc_sk_rcv(net, inputq);
514
	return rc;
515 516 517 518 519 520
}

/* tipc_bcast_add_peer - add a peer node to broadcast link and bearer
 *
 * RCU is locked, node lock is set
 */
521
void tipc_bcast_add_peer(struct net *net, struct tipc_link *uc_l,
522 523 524 525
			 struct sk_buff_head *xmitq)
{
	struct tipc_link *snd_l = tipc_bc_sndlink(net);

526
	tipc_bcast_lock(net);
527
	tipc_link_add_bc_peer(snd_l, uc_l, xmitq);
528
	tipc_bcbase_select_primary(net);
529
	tipc_bcbase_calc_bc_threshold(net);
530
	tipc_bcast_unlock(net);
531 532 533 534 535 536
}

/* tipc_bcast_remove_peer - remove a peer node from broadcast link and bearer
 *
 * RCU is locked, node lock is set
 */
537
void tipc_bcast_remove_peer(struct net *net, struct tipc_link *rcv_l)
538 539
{
	struct tipc_link *snd_l = tipc_bc_sndlink(net);
540
	struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq;
541 542 543 544
	struct sk_buff_head xmitq;

	__skb_queue_head_init(&xmitq);

545
	tipc_bcast_lock(net);
546
	tipc_link_remove_bc_peer(snd_l, rcv_l, &xmitq);
547
	tipc_bcbase_select_primary(net);
548
	tipc_bcbase_calc_bc_threshold(net);
549
	tipc_bcast_unlock(net);
550

551
	tipc_bcbase_xmit(net, &xmitq);
552 553 554 555 556 557

	/* Any socket wakeup messages ? */
	if (!skb_queue_empty(inputq))
		tipc_sk_rcv(net, inputq);
}

558
int tipc_bclink_reset_stats(struct net *net)
P
Per Liden 已提交
559
{
560
	struct tipc_link *l = tipc_bc_sndlink(net);
561

562
	if (!l)
P
Per Liden 已提交
563 564
		return -ENOPROTOOPT;

565
	tipc_bcast_lock(net);
566
	tipc_link_reset_stats(l);
567
	tipc_bcast_unlock(net);
568
	return 0;
P
Per Liden 已提交
569 570
}

571
static int tipc_bc_link_set_queue_limits(struct net *net, u32 limit)
P
Per Liden 已提交
572
{
573
	struct tipc_link *l = tipc_bc_sndlink(net);
574

575
	if (!l)
P
Per Liden 已提交
576
		return -ENOPROTOOPT;
577 578 579
	if (limit < BCLINK_WIN_MIN)
		limit = BCLINK_WIN_MIN;
	if (limit > TIPC_MAX_LINK_WIN)
P
Per Liden 已提交
580
		return -EINVAL;
581 582 583
	tipc_bcast_lock(net);
	tipc_link_set_queue_limits(l, limit);
	tipc_bcast_unlock(net);
584
	return 0;
P
Per Liden 已提交
585 586
}

587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637
static int tipc_bc_link_set_broadcast_mode(struct net *net, u32 bc_mode)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);

	switch (bc_mode) {
	case BCLINK_MODE_BCAST:
		if (!bb->bcast_support)
			return -ENOPROTOOPT;

		bb->force_bcast = true;
		bb->force_rcast = false;
		break;
	case BCLINK_MODE_RCAST:
		if (!bb->rcast_support)
			return -ENOPROTOOPT;

		bb->force_bcast = false;
		bb->force_rcast = true;
		break;
	case BCLINK_MODE_SEL:
		if (!bb->bcast_support || !bb->rcast_support)
			return -ENOPROTOOPT;

		bb->force_bcast = false;
		bb->force_rcast = false;
		break;
	default:
		return -EINVAL;
	}

	return 0;
}

static int tipc_bc_link_set_broadcast_ratio(struct net *net, u32 bc_ratio)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);

	if (!bb->bcast_support || !bb->rcast_support)
		return -ENOPROTOOPT;

	if (bc_ratio > 100 || bc_ratio <= 0)
		return -EINVAL;

	bb->rc_ratio = bc_ratio;
	tipc_bcast_lock(net);
	tipc_bcbase_calc_bc_threshold(net);
	tipc_bcast_unlock(net);

	return 0;
}

638 639 640 641
int tipc_nl_bc_link_set(struct net *net, struct nlattr *attrs[])
{
	int err;
	u32 win;
642 643
	u32 bc_mode;
	u32 bc_ratio;
644 645 646 647 648 649 650 651 652
	struct nlattr *props[TIPC_NLA_PROP_MAX + 1];

	if (!attrs[TIPC_NLA_LINK_PROP])
		return -EINVAL;

	err = tipc_nl_parse_link_prop(attrs[TIPC_NLA_LINK_PROP], props);
	if (err)
		return err;

653 654 655
	if (!props[TIPC_NLA_PROP_WIN] &&
	    !props[TIPC_NLA_PROP_BROADCAST] &&
	    !props[TIPC_NLA_PROP_BROADCAST_RATIO]) {
656
		return -EOPNOTSUPP;
657 658 659 660 661 662
	}

	if (props[TIPC_NLA_PROP_BROADCAST]) {
		bc_mode = nla_get_u32(props[TIPC_NLA_PROP_BROADCAST]);
		err = tipc_bc_link_set_broadcast_mode(net, bc_mode);
	}
663

664 665 666 667
	if (!err && props[TIPC_NLA_PROP_BROADCAST_RATIO]) {
		bc_ratio = nla_get_u32(props[TIPC_NLA_PROP_BROADCAST_RATIO]);
		err = tipc_bc_link_set_broadcast_ratio(net, bc_ratio);
	}
668

669 670 671 672 673 674
	if (!err && props[TIPC_NLA_PROP_WIN]) {
		win = nla_get_u32(props[TIPC_NLA_PROP_WIN]);
		err = tipc_bc_link_set_queue_limits(net, win);
	}

	return err;
675 676
}

677
int tipc_bcast_init(struct net *net)
P
Per Liden 已提交
678
{
679 680 681 682
	struct tipc_net *tn = tipc_net(net);
	struct tipc_bc_base *bb = NULL;
	struct tipc_link *l = NULL;

683
	bb = kzalloc(sizeof(*bb), GFP_KERNEL);
684 685 686
	if (!bb)
		goto enomem;
	tn->bcbase = bb;
687
	spin_lock_init(&tipc_net(net)->bclock);
688

689
	if (!tipc_link_bc_create(net, 0, 0,
690
				 FB_MTU,
691
				 BCLINK_WIN_DEFAULT,
692
				 0,
693
				 &bb->inputq,
694
				 NULL,
695
				 NULL,
696 697 698 699
				 &l))
		goto enomem;
	bb->link = l;
	tn->bcl = l;
700
	bb->rc_ratio = 10;
701
	bb->rcast_support = true;
702
	return 0;
703 704 705 706
enomem:
	kfree(bb);
	kfree(l);
	return -ENOMEM;
P
Per Liden 已提交
707 708
}

709
void tipc_bcast_stop(struct net *net)
P
Per Liden 已提交
710
{
711 712
	struct tipc_net *tn = net_generic(net, tipc_net_id);

713
	synchronize_net();
714
	kfree(tn->bcbase);
715
	kfree(tn->bcl);
P
Per Liden 已提交
716
}
717 718 719 720 721 722 723 724 725 726 727 728

void tipc_nlist_init(struct tipc_nlist *nl, u32 self)
{
	memset(nl, 0, sizeof(*nl));
	INIT_LIST_HEAD(&nl->list);
	nl->self = self;
}

void tipc_nlist_add(struct tipc_nlist *nl, u32 node)
{
	if (node == nl->self)
		nl->local = true;
J
Jon Maloy 已提交
729
	else if (tipc_dest_push(&nl->list, node, 0))
730 731 732 733 734 735 736
		nl->remote++;
}

void tipc_nlist_del(struct tipc_nlist *nl, u32 node)
{
	if (node == nl->self)
		nl->local = false;
J
Jon Maloy 已提交
737
	else if (tipc_dest_del(&nl->list, node, 0))
738 739 740 741 742
		nl->remote--;
}

void tipc_nlist_purge(struct tipc_nlist *nl)
{
J
Jon Maloy 已提交
743
	tipc_dest_list_purge(&nl->list);
744
	nl->remote = 0;
745
	nl->local = false;
746
}
747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769

u32 tipc_bcast_get_broadcast_mode(struct net *net)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);

	if (bb->force_bcast)
		return BCLINK_MODE_BCAST;

	if (bb->force_rcast)
		return BCLINK_MODE_RCAST;

	if (bb->bcast_support && bb->rcast_support)
		return BCLINK_MODE_SEL;

	return 0;
}

u32 tipc_bcast_get_broadcast_ratio(struct net *net)
{
	struct tipc_bc_base *bb = tipc_bc_base(net);

	return bb->rc_ratio;
}
770

H
Hoang Le 已提交
771
void tipc_mcast_filter_msg(struct net *net, struct sk_buff_head *defq,
772 773 774 775 776 777 778 779
			   struct sk_buff_head *inputq)
{
	struct sk_buff *skb, *_skb, *tmp;
	struct tipc_msg *hdr, *_hdr;
	bool match = false;
	u32 node, port;

	skb = skb_peek(inputq);
H
Hoang Le 已提交
780 781 782
	if (!skb)
		return;

783 784 785 786 787 788
	hdr = buf_msg(skb);

	if (likely(!msg_is_syn(hdr) && skb_queue_empty(defq)))
		return;

	node = msg_orignode(hdr);
H
Hoang Le 已提交
789 790 791
	if (node == tipc_own_addr(net))
		return;

792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851
	port = msg_origport(hdr);

	/* Has the twin SYN message already arrived ? */
	skb_queue_walk(defq, _skb) {
		_hdr = buf_msg(_skb);
		if (msg_orignode(_hdr) != node)
			continue;
		if (msg_origport(_hdr) != port)
			continue;
		match = true;
		break;
	}

	if (!match) {
		if (!msg_is_syn(hdr))
			return;
		__skb_dequeue(inputq);
		__skb_queue_tail(defq, skb);
		return;
	}

	/* Deliver non-SYN message from other link, otherwise queue it */
	if (!msg_is_syn(hdr)) {
		if (msg_is_rcast(hdr) != msg_is_rcast(_hdr))
			return;
		__skb_dequeue(inputq);
		__skb_queue_tail(defq, skb);
		return;
	}

	/* Queue non-SYN/SYN message from same link */
	if (msg_is_rcast(hdr) == msg_is_rcast(_hdr)) {
		__skb_dequeue(inputq);
		__skb_queue_tail(defq, skb);
		return;
	}

	/* Matching SYN messages => return the one with data, if any */
	__skb_unlink(_skb, defq);
	if (msg_data_sz(hdr)) {
		kfree_skb(_skb);
	} else {
		__skb_dequeue(inputq);
		kfree_skb(skb);
		__skb_queue_tail(inputq, _skb);
	}

	/* Deliver subsequent non-SYN messages from same peer */
	skb_queue_walk_safe(defq, _skb, tmp) {
		_hdr = buf_msg(_skb);
		if (msg_orignode(_hdr) != node)
			continue;
		if (msg_origport(_hdr) != port)
			continue;
		if (msg_is_syn(_hdr))
			break;
		__skb_unlink(_skb, defq);
		__skb_queue_tail(inputq, _skb);
	}
}