bcast.c 22.7 KB
Newer Older
Per Liden's avatar
Per Liden committed
1
2
/*
 * net/tipc/bcast.c: TIPC broadcast code
3
 *
4
 * Copyright (c) 2004-2006, Ericsson AB
Per Liden's avatar
Per Liden committed
5
 * Copyright (c) 2004, Intel Corporation.
6
 * Copyright (c) 2005, 2010-2011, Wind River Systems
Per Liden's avatar
Per Liden committed
7
8
 * All rights reserved.
 *
Per Liden's avatar
Per Liden committed
9
 * Redistribution and use in source and binary forms, with or without
Per Liden's avatar
Per Liden committed
10
11
 * modification, are permitted provided that the following conditions are met:
 *
Per Liden's avatar
Per Liden committed
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.
Per Liden's avatar
Per Liden committed
20
 *
Per Liden's avatar
Per Liden committed
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
Per Liden's avatar
Per Liden committed
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"
Per Liden's avatar
Per Liden committed
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
Per Liden's avatar
Per Liden committed
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
Per Liden's avatar
Per Liden committed
54
55
56
 * to be paired.
 */

57
struct tipc_bcbearer_pair {
58
59
	struct tipc_bearer *primary;
	struct tipc_bearer *secondary;
Per Liden's avatar
Per Liden committed
60
61
62
};

/**
63
 * struct tipc_bcbearer - bearer used by broadcast link
Per Liden's avatar
Per Liden committed
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".
Per Liden's avatar
Per Liden committed
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;
Per Liden's avatar
Per Liden committed
83
84
85
};

/**
86
 * struct tipc_bclink - link used for broadcast messages
Per Liden's avatar
Per Liden committed
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
 *
Per Liden's avatar
Per Liden committed
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;
Per Liden's avatar
Per Liden committed
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;
Per Liden's avatar
Per Liden committed
107

108
static DEFINE_SPINLOCK(bc_lock);
Per Liden's avatar
Per Liden committed
109

110
const char tipc_bclink_name[] = "broadcast-link";
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
115

Sam Ravnborg's avatar
Sam Ravnborg committed
116
static u32 bcbuf_acks(struct sk_buff *buf)
Per Liden's avatar
Per Liden committed
117
{
118
	return (u32)(unsigned long)TIPC_SKB_CB(buf)->handle;
Per Liden's avatar
Per Liden committed
119
120
}

Sam Ravnborg's avatar
Sam Ravnborg committed
121
static void bcbuf_set_acks(struct sk_buff *buf, u32 acks)
Per Liden's avatar
Per Liden committed
122
{
123
	TIPC_SKB_CB(buf)->handle = (void *)(unsigned long)acks;
Per Liden's avatar
Per Liden committed
124
125
}

Sam Ravnborg's avatar
Sam Ravnborg committed
126
static void bcbuf_decr_acks(struct sk_buff *buf)
Per Liden's avatar
Per Liden committed
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);
}
Per Liden's avatar
Per Liden committed
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)
Per Liden's avatar
Per Liden committed
159
{
160
161
	node->bclink.last_sent = less_eq(node->bclink.last_sent, seqno) ?
						seqno : node->bclink.last_sent;
Per Liden's avatar
Per Liden committed
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
/**
Per Liden's avatar
Per Liden committed
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
Per Liden's avatar
Per Liden committed
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));
Per Liden's avatar
Per Liden committed
190
191
}

192
/**
193
 * tipc_bclink_acknowledge - handle acknowledgement of broadcast packets
Per Liden's avatar
Per Liden committed
194
195
 * @n_ptr: node that sent acknowledgement info
 * @acked: broadcast sequence # that has been acknowledged
196
 *
Per Liden's avatar
Per Liden committed
197
198
 * Node is locked, bc_lock unlocked.
 */
199
void tipc_bclink_acknowledge(struct tipc_node *n_ptr, u32 acked)
Per Liden's avatar
Per Liden committed
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) */
Per Liden's avatar
Per Liden committed
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))
Per Liden's avatar
Per Liden committed
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();
		}

Per Liden's avatar
Per Liden committed
251
252
253
		if (bcbuf_acks(crs) == 0) {
			bcl->first_out = next;
			bcl->out_queue_size--;
254
			kfree_skb(crs);
Per Liden's avatar
Per Liden committed
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();
	}
Per Liden's avatar
Per Liden committed
267
	if (unlikely(released && !list_empty(&bcl->waiting_ports)))
268
		tipc_link_wakeup_ports(bcl, 0);
269
exit:
Per Liden's avatar
Per Liden committed
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
Per Liden's avatar
Per Liden committed
277
 */
278
void tipc_bclink_update_link_state(struct tipc_node *n_ptr, u32 last_sent)
Per Liden's avatar
Per Liden committed
279
{
280
	struct sk_buff *buf;
Per Liden's avatar
Per Liden committed
281

282
	/* Ignore "stale" link state info */
Per Liden's avatar
Per Liden committed
283

284
285
	if (less_eq(last_sent, n_ptr->bclink.last_in))
		return;
Per Liden's avatar
Per Liden committed
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) */
Per Liden's avatar
Per Liden committed
303

304
	if (n_ptr->bclink.oos_state & 0x1)
Per Liden's avatar
Per Liden committed
305
306
		return;

307
308
	/* Send NACK */

309
	buf = tipc_buf_acquire(INT_H_SIZE);
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
328

329
		n_ptr->bclink.oos_state++;
Per Liden's avatar
Per Liden committed
330
331
332
	}
}

333
/**
334
 * bclink_peek_nack - monitor retransmission requests sent by other nodes
Per Liden's avatar
Per Liden committed
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.
Per Liden's avatar
Per Liden committed
340
 */
341
static void bclink_peek_nack(struct tipc_msg *msg)
Per Liden's avatar
Per Liden committed
342
{
343
	struct tipc_node *n_ptr = tipc_node_find(msg_destnode(msg));
Per Liden's avatar
Per Liden committed
344

345
	if (unlikely(!n_ptr))
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
356
357
}

358
/*
359
 * tipc_bclink_send_msg - broadcast a packet to all nodes in cluster
Per Liden's avatar
Per Liden committed
360
 */
361
int tipc_bclink_send_msg(struct sk_buff *buf)
Per Liden's avatar
Per Liden committed
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:
Per Liden's avatar
Per Liden committed
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
 * 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++;
	}
}

409
/**
410
 * tipc_bclink_recv_pkt - receive a broadcast packet, and deliver upwards
411
 *
412
 * tipc_net_lock is read_locked, no other locks set
Per Liden's avatar
Per Liden committed
413
 */
414
void tipc_bclink_recv_pkt(struct sk_buff *buf)
415
{
Per Liden's avatar
Per Liden committed
416
	struct tipc_msg *msg = buf_msg(buf);
417
	struct tipc_node *node;
Per Liden's avatar
Per Liden committed
418
419
	u32 next_in;
	u32 seqno;
420
	int deferred;
Per Liden's avatar
Per Liden committed
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;
Per Liden's avatar
Per Liden committed
434

435
436
	/* Handle broadcast protocol message */

Per Liden's avatar
Per Liden committed
437
	if (unlikely(msg_user(msg) == BCAST_PROTOCOL)) {
438
439
		if (msg_type(msg) != STATE_MSG)
			goto unlock;
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
444
			bcl->stats.recv_nacks++;
445
			bclink->retransmit_to = node;
Per Liden's avatar
Per Liden committed
446
447
			bclink_retransmit_pkt(msg_bcgap_after(msg),
					      msg_bcgap_to(msg));
448
			spin_unlock_bh(&bc_lock);
Per Liden's avatar
Per Liden committed
449
		} else {
450
			tipc_node_unlock(node);
451
			bclink_peek_nack(msg);
Per Liden's avatar
Per Liden committed
452
		}
453
		goto exit;
Per Liden's avatar
Per Liden committed
454
455
	}

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

Per Liden's avatar
Per Liden committed
458
	seqno = msg_seqno(msg);
459
	next_in = mod(node->bclink.last_in + 1);
Per Liden's avatar
Per Liden committed
460
461

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

Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
474
		} else if (msg_user(msg) == MSG_BUNDLER) {
475
476
			spin_lock_bh(&bc_lock);
			bclink_accept_pkt(node, seqno);
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
489
			bcl->stats.recv_fragments++;
490
			if (ret > 0)
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
563
564
}

565
u32 tipc_bclink_acks_missing(struct tipc_node *n_ptr)
Per Liden's avatar
Per Liden committed
566
567
{
	return (n_ptr->bclink.supported &&
568
		(tipc_bclink_get_last_sent() != n_ptr->bclink.acked));
Per Liden's avatar
Per Liden committed
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
Per Liden's avatar
Per Liden committed
580
 */
Adrian Bunk's avatar
Adrian Bunk committed
581
582
583
static int tipc_bcbearer_send(struct sk_buff *buf,
			      struct tipc_bearer *unused1,
			      struct tipc_media_addr *unused2)
Per Liden's avatar
Per Liden committed
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
	 */
Per Liden's avatar
Per Liden committed
593
594
595
	if (likely(!msg_non_seq(buf_msg(buf)))) {
		struct tipc_msg *msg;

596
		bcbuf_set_acks(buf, bclink->bcast_nodes.count);
Per Liden's avatar
Per Liden committed
597
		msg = buf_msg(buf);
598
		msg_set_non_seq(msg, 1);
Per Liden's avatar
Per Liden committed
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;
		}
Per Liden's avatar
Per Liden committed
606
607
608
	}

	/* Send buffer over bearers until all targets reached */
609
	bcbearer->remains = bclink->bcast_nodes;
Per Liden's avatar
Per Liden committed
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;
Per Liden's avatar
Per Liden committed
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)
Per Liden's avatar
Per Liden committed
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;
Per Liden's avatar
Per Liden committed
636
637
		}

638
		if (bcbearer->remains_new.count == 0)
639
			break;	/* all targets reached */
Per Liden's avatar
Per Liden committed
640

641
		bcbearer->remains = bcbearer->remains_new;
Per Liden's avatar
Per Liden committed
642
	}
643

644
	return 0;
Per Liden's avatar
Per Liden committed
645
646
647
}

/**
648
 * tipc_bcbearer_sort - create sets of bearer pairs used by broadcast bearer
Per Liden's avatar
Per Liden committed
649
 */
650
void tipc_bcbearer_sort(void)
Per Liden's avatar
Per Liden committed
651
{
652
653
	struct tipc_bcbearer_pair *bp_temp = bcbearer->bpairs_temp;
	struct tipc_bcbearer_pair *bp_curr;
Per Liden's avatar
Per Liden committed
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];
Per Liden's avatar
Per Liden committed
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));

678
	for (pri = TIPC_MAX_LINK_PRI; pri >= 0; pri--) {
Per Liden's avatar
Per Liden committed
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)) {
Per Liden's avatar
Per Liden committed
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)
Per Liden's avatar
Per Liden committed
703
704
705
706
707
708
{
	struct print_buf pb;

	if (!bcl)
		return 0;

709
	tipc_printbuf_init(&pb, buf, buf_size);
Per Liden's avatar
Per Liden committed
710
711
712
713

	spin_lock_bh(&bc_lock);

	tipc_printf(&pb, "Link <%s>\n"
714
			 "  Window:%u packets\n",
Per Liden's avatar
Per Liden committed
715
		    bcl->name, bcl->queue_limit[0]);
716
	tipc_printf(&pb, "  RX packets:%u fragments:%u/%u bundles:%u/%u\n",
Per Liden's avatar
Per Liden committed
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",
Per Liden's avatar
Per Liden committed
723
724
		    bcl->stats.sent_info,
		    bcl->stats.sent_fragments,
725
		    bcl->stats.sent_fragmented,
Per Liden's avatar
Per Liden committed
726
727
		    bcl->stats.sent_bundles,
		    bcl->stats.sent_bundled);
728
	tipc_printf(&pb, "  RX naks:%u defs:%u dups:%u\n",
Per Liden's avatar
Per Liden committed
729
		    bcl->stats.recv_nacks,
730
		    bcl->stats.deferred_recv,
Per Liden's avatar
Per Liden committed
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,
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
746
747
}

748
int tipc_bclink_reset_stats(void)
Per Liden's avatar
Per Liden committed
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;
Per Liden's avatar
Per Liden committed
757
758
}

759
int tipc_bclink_set_queue_limits(u32 limit)
Per Liden's avatar
Per Liden committed
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);
Per Liden's avatar
Per Liden committed
768
	spin_unlock_bh(&bc_lock);
769
	return 0;
Per Liden's avatar
Per Liden committed
770
771
}

772
void tipc_bclink_init(void)
Per Liden's avatar
Per Liden committed
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");
Per Liden's avatar
Per Liden committed
778
779
780

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

790
void tipc_bclink_stop(void)
Per Liden's avatar
Per Liden committed
791
792
{
	spin_lock_bh(&bc_lock);
793
	tipc_link_stop(bcl);
Per Liden's avatar
Per Liden committed
794
	spin_unlock_bh(&bc_lock);
795
796
797

	memset(bclink, 0, sizeof(*bclink));
	memset(bcbearer, 0, sizeof(*bcbearer));
Per Liden's avatar
Per Liden committed
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
	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) {
883
				pr_warn("Incomplete multicast delivery, no memory\n");
884
885
886
887
888
889
890
891
892
893
894
				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);
	}
}