bcast.c 22.7 KB
Newer Older
P
Per Liden 已提交
1 2
/*
 * net/tipc/bcast.c: TIPC broadcast code
3
 *
P
Per Liden 已提交
4
 * Copyright (c) 2004-2006, 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 38 39 40 41
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "core.h"
#include "link.h"
#include "port.h"
#include "bcast.h"
42
#include "name_distr.h"
P
Per Liden 已提交
43 44 45 46 47 48

#define MAX_PKT_DEFAULT_MCAST 1500	/* bcast link max packet size (fixed) */

#define BCLINK_WIN_DEFAULT 20		/* bcast link window size (default) */

/**
49
 * struct tipc_bcbearer_pair - a pair of bearers used by broadcast link
P
Per Liden 已提交
50 51
 * @primary: pointer to primary bearer
 * @secondary: pointer to secondary bearer
52 53
 *
 * Bearers must have same priority and same set of reachable destinations
P
Per Liden 已提交
54 55 56
 * to be paired.
 */

57
struct tipc_bcbearer_pair {
58 59
	struct tipc_bearer *primary;
	struct tipc_bearer *secondary;
P
Per Liden 已提交
60 61 62
};

/**
63
 * struct tipc_bcbearer - bearer used by broadcast link
P
Per Liden 已提交
64 65 66
 * @bearer: (non-standard) broadcast bearer structure
 * @media: (non-standard) broadcast media structure
 * @bpairs: array of bearer pairs
67 68 69
 * @bpairs_temp: temporary array of bearer pairs used by tipc_bcbearer_sort()
 * @remains: temporary node map used by tipc_bcbearer_send()
 * @remains_new: temporary node map used tipc_bcbearer_send()
70
 *
71 72 73 74
 * Note: The fields labelled "temporary" are incorporated into the bearer
 * to avoid consuming potentially limited stack space through the use of
 * large local variables within multicast routines.  Concurrent access is
 * prevented through use of the spinlock "bc_lock".
P
Per Liden 已提交
75
 */
76
struct tipc_bcbearer {
77
	struct tipc_bearer bearer;
78
	struct tipc_media media;
79 80
	struct tipc_bcbearer_pair bpairs[MAX_BEARERS];
	struct tipc_bcbearer_pair bpairs_temp[TIPC_MAX_LINK_PRI + 1];
81 82
	struct tipc_node_map remains;
	struct tipc_node_map remains_new;
P
Per Liden 已提交
83 84 85
};

/**
86
 * struct tipc_bclink - link used for broadcast messages
P
Per Liden 已提交
87 88
 * @link: (non-standard) broadcast link structure
 * @node: (non-standard) node structure representing b'cast link's peer node
89
 * @bcast_nodes: map of broadcast-capable nodes
90
 * @retransmit_to: node that most recently requested a retransmit
91
 *
P
Per Liden 已提交
92 93
 * Handles sequence numbering, fragmentation, bundling, etc.
 */
94
struct tipc_bclink {
95
	struct tipc_link link;
96
	struct tipc_node node;
97
	struct tipc_node_map bcast_nodes;
98
	struct tipc_node *retransmit_to;
P
Per Liden 已提交
99 100
};

101
static struct tipc_bcbearer bcast_bearer;
102
static struct tipc_bclink bcast_link;
103

104
static struct tipc_bcbearer *bcbearer = &bcast_bearer;
105
static struct tipc_bclink *bclink = &bcast_link;
106
static struct tipc_link *bcl = &bcast_link.link;
P
Per Liden 已提交
107

I
Ingo Molnar 已提交
108
static DEFINE_SPINLOCK(bc_lock);
P
Per Liden 已提交
109

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

112 113 114
static void tipc_nmap_diff(struct tipc_node_map *nm_a,
			   struct tipc_node_map *nm_b,
			   struct tipc_node_map *nm_diff);
P
Per Liden 已提交
115

S
Sam Ravnborg 已提交
116
static u32 bcbuf_acks(struct sk_buff *buf)
P
Per Liden 已提交
117
{
118
	return (u32)(unsigned long)TIPC_SKB_CB(buf)->handle;
P
Per Liden 已提交
119 120
}

S
Sam Ravnborg 已提交
121
static void bcbuf_set_acks(struct sk_buff *buf, u32 acks)
P
Per Liden 已提交
122
{
123
	TIPC_SKB_CB(buf)->handle = (void *)(unsigned long)acks;
P
Per Liden 已提交
124 125
}

S
Sam Ravnborg 已提交
126
static void bcbuf_decr_acks(struct sk_buff *buf)
P
Per Liden 已提交
127 128 129 130
{
	bcbuf_set_acks(buf, bcbuf_acks(buf) - 1);
}

131 132 133 134 135 136 137 138 139 140 141 142 143
void tipc_bclink_add_node(u32 addr)
{
	spin_lock_bh(&bc_lock);
	tipc_nmap_add(&bclink->bcast_nodes, addr);
	spin_unlock_bh(&bc_lock);
}

void tipc_bclink_remove_node(u32 addr)
{
	spin_lock_bh(&bc_lock);
	tipc_nmap_remove(&bclink->bcast_nodes, addr);
	spin_unlock_bh(&bc_lock);
}
P
Per Liden 已提交
144

145 146 147 148 149 150 151 152 153 154 155 156 157
static void bclink_set_last_sent(void)
{
	if (bcl->next_out)
		bcl->fsm_msg_cnt = mod(buf_seqno(bcl->next_out) - 1);
	else
		bcl->fsm_msg_cnt = mod(bcl->next_out_no - 1);
}

u32 tipc_bclink_get_last_sent(void)
{
	return bcl->fsm_msg_cnt;
}

158
static void bclink_update_last_sent(struct tipc_node *node, u32 seqno)
P
Per Liden 已提交
159
{
160 161
	node->bclink.last_sent = less_eq(node->bclink.last_sent, seqno) ?
						seqno : node->bclink.last_sent;
P
Per Liden 已提交
162 163 164
}


165
/*
166 167 168 169 170 171 172 173 174
 * tipc_bclink_retransmit_to - get most recent node to request retransmission
 *
 * Called with bc_lock locked
 */
struct tipc_node *tipc_bclink_retransmit_to(void)
{
	return bclink->retransmit_to;
}

175
/**
P
Per Liden 已提交
176 177 178
 * bclink_retransmit_pkt - retransmit broadcast packets
 * @after: sequence number of last packet to *not* retransmit
 * @to: sequence number of last packet to retransmit
179
 *
180
 * Called with bc_lock locked
P
Per Liden 已提交
181 182 183 184 185 186
 */
static void bclink_retransmit_pkt(u32 after, u32 to)
{
	struct sk_buff *buf;

	buf = bcl->first_out;
187
	while (buf && less_eq(buf_seqno(buf), after))
188
		buf = buf->next;
189
	tipc_link_retransmit(bcl, buf, mod(to - after));
P
Per Liden 已提交
190 191
}

192
/**
193
 * tipc_bclink_acknowledge - handle acknowledgement of broadcast packets
P
Per Liden 已提交
194 195
 * @n_ptr: node that sent acknowledgement info
 * @acked: broadcast sequence # that has been acknowledged
196
 *
P
Per Liden 已提交
197 198
 * Node is locked, bc_lock unlocked.
 */
199
void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
P
Per Liden 已提交
200 201 202 203 204 205 206
{
	struct sk_buff *crs;
	struct sk_buff *next;
	unsigned int released = 0;

	spin_lock_bh(&bc_lock);

207
	/* Bail out if tx queue is empty (no clean up is required) */
P
Per Liden 已提交
208
	crs = bcl->first_out;
209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234
	if (!crs)
		goto exit;

	/* Determine which messages need to be acknowledged */
	if (acked == INVALID_LINK_SEQ) {
		/*
		 * Contact with specified node has been lost, so need to
		 * acknowledge sent messages only (if other nodes still exist)
		 * or both sent and unsent messages (otherwise)
		 */
		if (bclink->bcast_nodes.count)
			acked = bcl->fsm_msg_cnt;
		else
			acked = bcl->next_out_no;
	} else {
		/*
		 * Bail out if specified sequence number does not correspond
		 * to a message that has been sent and not yet acknowledged
		 */
		if (less(acked, buf_seqno(crs)) ||
		    less(bcl->fsm_msg_cnt, acked) ||
		    less_eq(acked, n_ptr->bclink.acked))
			goto exit;
	}

	/* Skip over packets that node has previously acknowledged */
235
	while (crs && less_eq(buf_seqno(crs), n_ptr->bclink.acked))
P
Per Liden 已提交
236 237 238 239 240 241
		crs = crs->next;

	/* Update packets that node is now acknowledging */

	while (crs && less_eq(buf_seqno(crs), acked)) {
		next = crs->next;
242 243 244 245 246 247 248 249 250

		if (crs != bcl->next_out)
			bcbuf_decr_acks(crs);
		else {
			bcbuf_set_acks(crs, 0);
			bcl->next_out = next;
			bclink_set_last_sent();
		}

P
Per Liden 已提交
251 252 253
		if (bcbuf_acks(crs) == 0) {
			bcl->first_out = next;
			bcl->out_queue_size--;
254
			kfree_skb(crs);
P
Per Liden 已提交
255 256 257 258 259 260 261 262
			released = 1;
		}
		crs = next;
	}
	n_ptr->bclink.acked = acked;

	/* Try resolving broadcast link congestion, if necessary */

263
	if (unlikely(bcl->next_out)) {
264
		tipc_link_push_queue(bcl);
265 266
		bclink_set_last_sent();
	}
P
Per Liden 已提交
267
	if (unlikely(released && !list_empty(&bcl->waiting_ports)))
268
		tipc_link_wakeup_ports(bcl, 0);
269
exit:
P
Per Liden 已提交
270 271 272
	spin_unlock_bh(&bc_lock);
}

273 274
/*
 * tipc_bclink_update_link_state - update broadcast link state
275
 *
276
 * tipc_net_lock and node lock set
P
Per Liden 已提交
277
 */
278
void tipc_bclink_update_link_state(struct tipc_node *n_ptr, u32 last_sent)
P
Per Liden 已提交
279
{
280
	struct sk_buff *buf;
P
Per Liden 已提交
281

282
	/* Ignore "stale" link state info */
P
Per Liden 已提交
283

284 285
	if (less_eq(last_sent, n_ptr->bclink.last_in))
		return;
P
Per Liden 已提交
286

287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302
	/* Update link synchronization state; quit if in sync */

	bclink_update_last_sent(n_ptr, last_sent);

	if (n_ptr->bclink.last_sent == n_ptr->bclink.last_in)
		return;

	/* Update out-of-sync state; quit if loss is still unconfirmed */

	if ((++n_ptr->bclink.oos_state) == 1) {
		if (n_ptr->bclink.deferred_size < (TIPC_MIN_LINK_WIN / 2))
			return;
		n_ptr->bclink.oos_state++;
	}

	/* Don't NACK if one has been recently sent (or seen) */
P
Per Liden 已提交
303

304
	if (n_ptr->bclink.oos_state & 0x1)
P
Per Liden 已提交
305 306
		return;

307 308
	/* Send NACK */

309
	buf = tipc_buf_acquire(INT_H_SIZE);
P
Per Liden 已提交
310
	if (buf) {
311 312
		struct tipc_msg *msg = buf_msg(buf);

313
		tipc_msg_init(msg, BCAST_PROTOCOL, STATE_MSG,
314
			      INT_H_SIZE, n_ptr->addr);
315
		msg_set_non_seq(msg, 1);
P
Per Liden 已提交
316
		msg_set_mc_netid(msg, tipc_net_id);
317 318 319 320 321
		msg_set_bcast_ack(msg, n_ptr->bclink.last_in);
		msg_set_bcgap_after(msg, n_ptr->bclink.last_in);
		msg_set_bcgap_to(msg, n_ptr->bclink.deferred_head
				 ? buf_seqno(n_ptr->bclink.deferred_head) - 1
				 : n_ptr->bclink.last_sent);
P
Per Liden 已提交
322

323
		spin_lock_bh(&bc_lock);
324 325
		tipc_bearer_send(&bcbearer->bearer, buf, NULL);
		bcl->stats.sent_nacks++;
326
		spin_unlock_bh(&bc_lock);
327
		kfree_skb(buf);
P
Per Liden 已提交
328

329
		n_ptr->bclink.oos_state++;
P
Per Liden 已提交
330 331 332
	}
}

333 334
/*
 * bclink_peek_nack - monitor retransmission requests sent by other nodes
P
Per Liden 已提交
335
 *
336 337
 * Delay any upcoming NACK by this node if another node has already
 * requested the first message this node is going to ask for.
338
 *
339
 * Only tipc_net_lock set.
P
Per Liden 已提交
340
 */
341
static void bclink_peek_nack(struct tipc_msg *msg)
P
Per Liden 已提交
342
{
343
	struct tipc_node *n_ptr = tipc_node_find(msg_destnode(msg));
P
Per Liden 已提交
344

345
	if (unlikely(!n_ptr))
P
Per Liden 已提交
346
		return;
347

348
	tipc_node_lock(n_ptr);
349 350 351 352 353 354

	if (n_ptr->bclink.supported &&
	    (n_ptr->bclink.last_in != n_ptr->bclink.last_sent) &&
	    (n_ptr->bclink.last_in == msg_bcgap_after(msg)))
		n_ptr->bclink.oos_state = 2;

355
	tipc_node_unlock(n_ptr);
P
Per Liden 已提交
356 357
}

358
/*
359
 * tipc_bclink_send_msg - broadcast a packet to all nodes in cluster
P
Per Liden 已提交
360
 */
361
int tipc_bclink_send_msg(struct sk_buff *buf)
P
Per Liden 已提交
362 363 364 365 366
{
	int res;

	spin_lock_bh(&bc_lock);

367 368
	if (!bclink->bcast_nodes.count) {
		res = msg_data_sz(buf_msg(buf));
369
		kfree_skb(buf);
370 371 372
		goto exit;
	}

373
	res = tipc_link_send_buf(bcl, buf);
374
	if (likely(res >= 0)) {
375
		bclink_set_last_sent();
376 377 378
		bcl->stats.queue_sz_counts++;
		bcl->stats.accu_queue_sz += bcl->out_queue_size;
	}
379
exit:
P
Per Liden 已提交
380 381 382 383
	spin_unlock_bh(&bc_lock);
	return res;
}

384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409
/*
 * bclink_accept_pkt - accept an incoming, in-sequence broadcast packet
 *
 * Called with both sending node's lock and bc_lock taken.
 */
static void bclink_accept_pkt(struct tipc_node *node, u32 seqno)
{
	bclink_update_last_sent(node, seqno);
	node->bclink.last_in = seqno;
	node->bclink.oos_state = 0;
	bcl->stats.recv_info++;

	/*
	 * Unicast an ACK periodically, ensuring that
	 * all nodes in the cluster don't ACK at the same time
	 */

	if (((seqno - tipc_own_addr) % TIPC_MIN_LINK_WIN) == 0) {
		tipc_link_send_proto_msg(
			node->active_links[node->addr & 1],
			STATE_MSG, 0, 0, 0, 0, 0);
		bcl->stats.sent_acks++;
	}
}

/*
410
 * tipc_bclink_recv_pkt - receive a broadcast packet, and deliver upwards
411
 *
412
 * tipc_net_lock is read_locked, no other locks set
P
Per Liden 已提交
413
 */
414
void tipc_bclink_recv_pkt(struct sk_buff *buf)
415
{
P
Per Liden 已提交
416
	struct tipc_msg *msg = buf_msg(buf);
417
	struct tipc_node *node;
P
Per Liden 已提交
418 419
	u32 next_in;
	u32 seqno;
420
	int deferred;
P
Per Liden 已提交
421

422 423 424 425 426 427 428 429 430 431 432 433
	/* Screen out unwanted broadcast messages */

	if (msg_mc_netid(msg) != tipc_net_id)
		goto exit;

	node = tipc_node_find(msg_prevnode(msg));
	if (unlikely(!node))
		goto exit;

	tipc_node_lock(node);
	if (unlikely(!node->bclink.supported))
		goto unlock;
P
Per Liden 已提交
434

435 436
	/* Handle broadcast protocol message */

P
Per Liden 已提交
437
	if (unlikely(msg_user(msg) == BCAST_PROTOCOL)) {
438 439
		if (msg_type(msg) != STATE_MSG)
			goto unlock;
P
Per Liden 已提交
440
		if (msg_destnode(msg) == tipc_own_addr) {
441 442
			tipc_bclink_acknowledge(node, msg_bcast_ack(msg));
			tipc_node_unlock(node);
443
			spin_lock_bh(&bc_lock);
P
Per Liden 已提交
444
			bcl->stats.recv_nacks++;
445
			bclink->retransmit_to = node;
P
Per Liden 已提交
446 447
			bclink_retransmit_pkt(msg_bcgap_after(msg),
					      msg_bcgap_to(msg));
448
			spin_unlock_bh(&bc_lock);
P
Per Liden 已提交
449
		} else {
450
			tipc_node_unlock(node);
451
			bclink_peek_nack(msg);
P
Per Liden 已提交
452
		}
453
		goto exit;
P
Per Liden 已提交
454 455
	}

456 457
	/* Handle in-sequence broadcast message */

P
Per Liden 已提交
458
	seqno = msg_seqno(msg);
459
	next_in = mod(node->bclink.last_in + 1);
P
Per Liden 已提交
460 461

	if (likely(seqno == next_in)) {
462
receive:
463 464
		/* Deliver message to destination */

P
Per Liden 已提交
465
		if (likely(msg_isdata(msg))) {
466 467
			spin_lock_bh(&bc_lock);
			bclink_accept_pkt(node, seqno);
468
			spin_unlock_bh(&bc_lock);
469
			tipc_node_unlock(node);
470 471 472
			if (likely(msg_mcast(msg)))
				tipc_port_recv_mcast(buf, NULL);
			else
473
				kfree_skb(buf);
P
Per Liden 已提交
474
		} else if (msg_user(msg) == MSG_BUNDLER) {
475 476
			spin_lock_bh(&bc_lock);
			bclink_accept_pkt(node, seqno);
P
Per Liden 已提交
477 478
			bcl->stats.recv_bundles++;
			bcl->stats.recv_bundled += msg_msgcnt(msg);
479
			spin_unlock_bh(&bc_lock);
480 481
			tipc_node_unlock(node);
			tipc_link_recv_bundle(buf);
P
Per Liden 已提交
482
		} else if (msg_user(msg) == MSG_FRAGMENTER) {
483 484 485 486 487 488
			int ret = tipc_link_recv_fragment(&node->bclink.defragm,
						      &buf, &msg);
			if (ret < 0)
				goto unlock;
			spin_lock_bh(&bc_lock);
			bclink_accept_pkt(node, seqno);
P
Per Liden 已提交
489
			bcl->stats.recv_fragments++;
490
			if (ret > 0)
P
Per Liden 已提交
491
				bcl->stats.recv_fragmented++;
492
			spin_unlock_bh(&bc_lock);
493 494
			tipc_node_unlock(node);
			tipc_net_route_msg(buf);
495
		} else if (msg_user(msg) == NAME_DISTRIBUTOR) {
496 497
			spin_lock_bh(&bc_lock);
			bclink_accept_pkt(node, seqno);
498
			spin_unlock_bh(&bc_lock);
499 500
			tipc_node_unlock(node);
			tipc_named_recv(buf);
P
Per Liden 已提交
501
		} else {
502 503
			spin_lock_bh(&bc_lock);
			bclink_accept_pkt(node, seqno);
504
			spin_unlock_bh(&bc_lock);
505
			tipc_node_unlock(node);
506
			kfree_skb(buf);
P
Per Liden 已提交
507
		}
508
		buf = NULL;
509 510 511

		/* Determine new synchronization state */

512
		tipc_node_lock(node);
513 514 515
		if (unlikely(!tipc_node_is_up(node)))
			goto unlock;

516
		if (node->bclink.last_in == node->bclink.last_sent)
517 518
			goto unlock;

519 520 521 522 523
		if (!node->bclink.deferred_head) {
			node->bclink.oos_state = 1;
			goto unlock;
		}

524 525 526 527 528 529 530 531 532 533
		msg = buf_msg(node->bclink.deferred_head);
		seqno = msg_seqno(msg);
		next_in = mod(next_in + 1);
		if (seqno != next_in)
			goto unlock;

		/* Take in-sequence message from deferred queue & deliver it */

		buf = node->bclink.deferred_head;
		node->bclink.deferred_head = buf->next;
534
		node->bclink.deferred_size--;
535 536 537 538 539 540 541 542 543
		goto receive;
	}

	/* Handle out-of-sequence broadcast message */

	if (less(next_in, seqno)) {
		deferred = tipc_link_defer_pkt(&node->bclink.deferred_head,
					       &node->bclink.deferred_tail,
					       buf);
544 545
		node->bclink.deferred_size += deferred;
		bclink_update_last_sent(node, seqno);
546
		buf = NULL;
547 548 549
	} else
		deferred = 0;

550 551
	spin_lock_bh(&bc_lock);

552 553
	if (deferred)
		bcl->stats.deferred_recv++;
554 555
	else
		bcl->stats.duplicates++;
556

557 558
	spin_unlock_bh(&bc_lock);

559
unlock:
560
	tipc_node_unlock(node);
561
exit:
562
	kfree_skb(buf);
P
Per Liden 已提交
563 564
}

565
u32 tipc_bclink_acks_missing(struct tipc_node *n_ptr)
P
Per Liden 已提交
566 567
{
	return (n_ptr->bclink.supported &&
568
		(tipc_bclink_get_last_sent() != n_ptr->bclink.acked));
P
Per Liden 已提交
569 570 571 572
}


/**
573
 * tipc_bcbearer_send - send a packet through the broadcast pseudo-bearer
574
 *
575 576
 * Send packet over as many bearers as necessary to reach all nodes
 * that have joined the broadcast link.
577
 *
578 579
 * Returns 0 (packet sent successfully) under all circumstances,
 * since the broadcast link's pseudo-bearer never blocks
P
Per Liden 已提交
580
 */
A
Adrian Bunk 已提交
581 582 583
static int tipc_bcbearer_send(struct sk_buff *buf,
			      struct tipc_bearer *unused1,
			      struct tipc_media_addr *unused2)
P
Per Liden 已提交
584 585 586
{
	int bp_index;

587 588 589 590 591 592
	/*
	 * Prepare broadcast link message for reliable transmission,
	 * if first time trying to send it;
	 * preparation is skipped for broadcast link protocol messages
	 * since they are sent in an unreliable manner and don't need it
	 */
P
Per Liden 已提交
593 594 595
	if (likely(!msg_non_seq(buf_msg(buf)))) {
		struct tipc_msg *msg;

596
		bcbuf_set_acks(buf, bclink->bcast_nodes.count);
P
Per Liden 已提交
597
		msg = buf_msg(buf);
598
		msg_set_non_seq(msg, 1);
P
Per Liden 已提交
599
		msg_set_mc_netid(msg, tipc_net_id);
600
		bcl->stats.sent_info++;
601

602
		if (WARN_ON(!bclink->bcast_nodes.count)) {
603 604 605
			dump_stack();
			return 0;
		}
P
Per Liden 已提交
606 607 608
	}

	/* Send buffer over bearers until all targets reached */
609
	bcbearer->remains = bclink->bcast_nodes;
P
Per Liden 已提交
610 611

	for (bp_index = 0; bp_index < MAX_BEARERS; bp_index++) {
612 613
		struct tipc_bearer *p = bcbearer->bpairs[bp_index].primary;
		struct tipc_bearer *s = bcbearer->bpairs[bp_index].secondary;
P
Per Liden 已提交
614 615 616 617

		if (!p)
			break;	/* no more bearers to try */

618 619
		tipc_nmap_diff(&bcbearer->remains, &p->nodes, &bcbearer->remains_new);
		if (bcbearer->remains_new.count == bcbearer->remains.count)
P
Per Liden 已提交
620 621
			continue;	/* bearer pair doesn't add anything */

622 623
		if (p->blocked ||
		    p->media->send_msg(buf, p, &p->media->bcast_addr)) {
624
			/* unable to send on primary bearer */
625 626
			if (!s || s->blocked ||
			    s->media->send_msg(buf, s,
627 628 629 630 631 632 633 634 635
					       &s->media->bcast_addr)) {
				/* unable to send on either bearer */
				continue;
			}
		}

		if (s) {
			bcbearer->bpairs[bp_index].primary = s;
			bcbearer->bpairs[bp_index].secondary = p;
P
Per Liden 已提交
636 637
		}

638
		if (bcbearer->remains_new.count == 0)
639
			break;	/* all targets reached */
P
Per Liden 已提交
640

641
		bcbearer->remains = bcbearer->remains_new;
P
Per Liden 已提交
642
	}
643

644
	return 0;
P
Per Liden 已提交
645 646 647
}

/**
648
 * tipc_bcbearer_sort - create sets of bearer pairs used by broadcast bearer
P
Per Liden 已提交
649
 */
650
void tipc_bcbearer_sort(void)
P
Per Liden 已提交
651
{
652 653
	struct tipc_bcbearer_pair *bp_temp = bcbearer->bpairs_temp;
	struct tipc_bcbearer_pair *bp_curr;
P
Per Liden 已提交
654 655 656 657 658 659 660 661 662
	int b_index;
	int pri;

	spin_lock_bh(&bc_lock);

	/* Group bearers by priority (can assume max of two per priority) */
	memset(bp_temp, 0, sizeof(bcbearer->bpairs_temp));

	for (b_index = 0; b_index < MAX_BEARERS; b_index++) {
663
		struct tipc_bearer *b = &tipc_bearers[b_index];
P
Per Liden 已提交
664 665 666 667 668 669 670 671 672 673 674 675 676 677

		if (!b->active || !b->nodes.count)
			continue;

		if (!bp_temp[b->priority].primary)
			bp_temp[b->priority].primary = b;
		else
			bp_temp[b->priority].secondary = b;
	}

	/* Create array of bearer pairs for broadcasting */
	bp_curr = bcbearer->bpairs;
	memset(bcbearer->bpairs, 0, sizeof(bcbearer->bpairs));

P
Per Liden 已提交
678
	for (pri = TIPC_MAX_LINK_PRI; pri >= 0; pri--) {
P
Per Liden 已提交
679 680 681 682 683 684 685

		if (!bp_temp[pri].primary)
			continue;

		bp_curr->primary = bp_temp[pri].primary;

		if (bp_temp[pri].secondary) {
686 687
			if (tipc_nmap_equal(&bp_temp[pri].primary->nodes,
					    &bp_temp[pri].secondary->nodes)) {
P
Per Liden 已提交
688 689 690 691 692 693 694 695 696 697 698 699 700 701
				bp_curr->secondary = bp_temp[pri].secondary;
			} else {
				bp_curr++;
				bp_curr->primary = bp_temp[pri].secondary;
			}
		}

		bp_curr++;
	}

	spin_unlock_bh(&bc_lock);
}


702
int tipc_bclink_stats(char *buf, const u32 buf_size)
P
Per Liden 已提交
703 704 705 706 707 708
{
	struct print_buf pb;

	if (!bcl)
		return 0;

709
	tipc_printbuf_init(&pb, buf, buf_size);
P
Per Liden 已提交
710 711 712 713

	spin_lock_bh(&bc_lock);

	tipc_printf(&pb, "Link <%s>\n"
714
			 "  Window:%u packets\n",
P
Per Liden 已提交
715
		    bcl->name, bcl->queue_limit[0]);
716
	tipc_printf(&pb, "  RX packets:%u fragments:%u/%u bundles:%u/%u\n",
P
Per Liden 已提交
717 718 719 720 721
		    bcl->stats.recv_info,
		    bcl->stats.recv_fragments,
		    bcl->stats.recv_fragmented,
		    bcl->stats.recv_bundles,
		    bcl->stats.recv_bundled);
722
	tipc_printf(&pb, "  TX packets:%u fragments:%u/%u bundles:%u/%u\n",
P
Per Liden 已提交
723 724
		    bcl->stats.sent_info,
		    bcl->stats.sent_fragments,
725
		    bcl->stats.sent_fragmented,
P
Per Liden 已提交
726 727
		    bcl->stats.sent_bundles,
		    bcl->stats.sent_bundled);
728
	tipc_printf(&pb, "  RX naks:%u defs:%u dups:%u\n",
P
Per Liden 已提交
729
		    bcl->stats.recv_nacks,
730
		    bcl->stats.deferred_recv,
P
Per Liden 已提交
731
		    bcl->stats.duplicates);
732 733 734
	tipc_printf(&pb, "  TX naks:%u acks:%u dups:%u\n",
		    bcl->stats.sent_nacks,
		    bcl->stats.sent_acks,
P
Per Liden 已提交
735 736 737 738 739 740 741 742 743 744
		    bcl->stats.retransmitted);
	tipc_printf(&pb, "  Congestion bearer:%u link:%u  Send queue max:%u avg:%u\n",
		    bcl->stats.bearer_congs,
		    bcl->stats.link_congs,
		    bcl->stats.max_queue_sz,
		    bcl->stats.queue_sz_counts
		    ? (bcl->stats.accu_queue_sz / bcl->stats.queue_sz_counts)
		    : 0);

	spin_unlock_bh(&bc_lock);
745
	return tipc_printbuf_validate(&pb);
P
Per Liden 已提交
746 747
}

748
int tipc_bclink_reset_stats(void)
P
Per Liden 已提交
749 750 751 752 753 754 755
{
	if (!bcl)
		return -ENOPROTOOPT;

	spin_lock_bh(&bc_lock);
	memset(&bcl->stats, 0, sizeof(bcl->stats));
	spin_unlock_bh(&bc_lock);
756
	return 0;
P
Per Liden 已提交
757 758
}

759
int tipc_bclink_set_queue_limits(u32 limit)
P
Per Liden 已提交
760 761 762 763 764 765 766
{
	if (!bcl)
		return -ENOPROTOOPT;
	if ((limit < TIPC_MIN_LINK_WIN) || (limit > TIPC_MAX_LINK_WIN))
		return -EINVAL;

	spin_lock_bh(&bc_lock);
767
	tipc_link_set_queue_limits(bcl, limit);
P
Per Liden 已提交
768
	spin_unlock_bh(&bc_lock);
769
	return 0;
P
Per Liden 已提交
770 771
}

772
void tipc_bclink_init(void)
P
Per Liden 已提交
773 774 775
{
	INIT_LIST_HEAD(&bcbearer->bearer.cong_links);
	bcbearer->bearer.media = &bcbearer->media;
776
	bcbearer->media.send_msg = tipc_bcbearer_send;
777
	sprintf(bcbearer->media.name, "tipc-broadcast");
P
Per Liden 已提交
778 779 780

	INIT_LIST_HEAD(&bcl->waiting_ports);
	bcl->next_out_no = 1;
I
Ingo Molnar 已提交
781
	spin_lock_init(&bclink->node.lock);
P
Per Liden 已提交
782
	bcl->owner = &bclink->node;
783
	bcl->max_pkt = MAX_PKT_DEFAULT_MCAST;
784
	tipc_link_set_queue_limits(bcl, BCLINK_WIN_DEFAULT);
P
Per Liden 已提交
785 786
	bcl->b_ptr = &bcbearer->bearer;
	bcl->state = WORKING_WORKING;
787
	strlcpy(bcl->name, tipc_bclink_name, TIPC_MAX_LINK_NAME);
P
Per Liden 已提交
788 789
}

790
void tipc_bclink_stop(void)
P
Per Liden 已提交
791 792
{
	spin_lock_bh(&bc_lock);
793
	tipc_link_stop(bcl);
P
Per Liden 已提交
794
	spin_unlock_bh(&bc_lock);
795 796 797

	memset(bclink, 0, sizeof(*bclink));
	memset(bcbearer, 0, sizeof(*bcbearer));
P
Per Liden 已提交
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

/**
 * tipc_nmap_add - add a node to a node map
 */
void tipc_nmap_add(struct tipc_node_map *nm_ptr, u32 node)
{
	int n = tipc_node(node);
	int w = n / WSIZE;
	u32 mask = (1 << (n % WSIZE));

	if ((nm_ptr->map[w] & mask) == 0) {
		nm_ptr->count++;
		nm_ptr->map[w] |= mask;
	}
}

/**
 * tipc_nmap_remove - remove a node from a node map
 */
void tipc_nmap_remove(struct tipc_node_map *nm_ptr, u32 node)
{
	int n = tipc_node(node);
	int w = n / WSIZE;
	u32 mask = (1 << (n % WSIZE));

	if ((nm_ptr->map[w] & mask) != 0) {
		nm_ptr->map[w] &= ~mask;
		nm_ptr->count--;
	}
}

/**
 * tipc_nmap_diff - find differences between node maps
 * @nm_a: input node map A
 * @nm_b: input node map B
 * @nm_diff: output node map A-B (i.e. nodes of A that are not in B)
 */
837 838 839
static void tipc_nmap_diff(struct tipc_node_map *nm_a,
			   struct tipc_node_map *nm_b,
			   struct tipc_node_map *nm_diff)
840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857
{
	int stop = ARRAY_SIZE(nm_a->map);
	int w;
	int b;
	u32 map;

	memset(nm_diff, 0, sizeof(*nm_diff));
	for (w = 0; w < stop; w++) {
		map = nm_a->map[w] ^ (nm_a->map[w] & nm_b->map[w]);
		nm_diff->map[w] = map;
		if (map != 0) {
			for (b = 0 ; b < WSIZE; b++) {
				if (map & (1 << b))
					nm_diff->count++;
			}
		}
	}
}
858 859 860 861

/**
 * tipc_port_list_add - add a port to a port list, ensuring no duplicates
 */
862
void tipc_port_list_add(struct tipc_port_list *pl_ptr, u32 port)
863
{
864
	struct tipc_port_list *item = pl_ptr;
865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894
	int i;
	int item_sz = PLSIZE;
	int cnt = pl_ptr->count;

	for (; ; cnt -= item_sz, item = item->next) {
		if (cnt < PLSIZE)
			item_sz = cnt;
		for (i = 0; i < item_sz; i++)
			if (item->ports[i] == port)
				return;
		if (i < PLSIZE) {
			item->ports[i] = port;
			pl_ptr->count++;
			return;
		}
		if (!item->next) {
			item->next = kmalloc(sizeof(*item), GFP_ATOMIC);
			if (!item->next) {
				warn("Incomplete multicast delivery, no memory\n");
				return;
			}
			item->next->next = NULL;
		}
	}
}

/**
 * tipc_port_list_free - free dynamically created entries in port_list chain
 *
 */
895
void tipc_port_list_free(struct tipc_port_list *pl_ptr)
896
{
897 898
	struct tipc_port_list *item;
	struct tipc_port_list *next;
899 900 901 902 903 904

	for (item = pl_ptr->next; item; item = next) {
		next = item->next;
		kfree(item);
	}
}