1 /* 2 * net/tipc/bcast.c: TIPC broadcast code 3 * 4 * Copyright (c) 2004-2006, 2014-2015, Ericsson AB 5 * Copyright (c) 2004, Intel Corporation. 6 * Copyright (c) 2005, 2010-2011, Wind River Systems 7 * All rights reserved. 8 * 9 * Redistribution and use in source and binary forms, with or without 10 * modification, are permitted provided that the following conditions are met: 11 * 12 * 1. Redistributions of source code must retain the above copyright 13 * notice, this list of conditions and the following disclaimer. 14 * 2. Redistributions in binary form must reproduce the above copyright 15 * notice, this list of conditions and the following disclaimer in the 16 * documentation and/or other materials provided with the distribution. 17 * 3. Neither the names of the copyright holders nor the names of its 18 * contributors may be used to endorse or promote products derived from 19 * this software without specific prior written permission. 20 * 21 * Alternatively, this software may be distributed under the terms of the 22 * GNU General Public License ("GPL") version 2 as published by the Free 23 * Software Foundation. 24 * 25 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 26 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 27 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 28 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 29 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 30 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 31 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 32 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 33 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 34 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 35 * POSSIBILITY OF SUCH DAMAGE. 36 */ 37 38 #include <linux/tipc_config.h> 39 #include "socket.h" 40 #include "msg.h" 41 #include "bcast.h" 42 #include "name_distr.h" 43 #include "link.h" 44 #include "node.h" 45 46 #define BCLINK_WIN_DEFAULT 50 /* bcast link window size (default) */ 47 #define BCLINK_WIN_MIN 32 /* bcast minimum link window size */ 48 49 const char tipc_bclink_name[] = "broadcast-link"; 50 51 /** 52 * struct tipc_bc_base - base structure for keeping broadcast send state 53 * @link: broadcast send link structure 54 * @inputq: data input queue; will only carry SOCK_WAKEUP messages 55 * @dest: array keeping number of reachable destinations per bearer 56 * @primary_bearer: a bearer having links to all broadcast destinations, if any 57 */ 58 struct tipc_bc_base { 59 struct tipc_link *link; 60 struct sk_buff_head inputq; 61 int dests[MAX_BEARERS]; 62 int primary_bearer; 63 }; 64 65 static struct tipc_bc_base *tipc_bc_base(struct net *net) 66 { 67 return tipc_net(net)->bcbase; 68 } 69 70 int tipc_bcast_get_mtu(struct net *net) 71 { 72 return tipc_link_mtu(tipc_bc_sndlink(net)); 73 } 74 75 /* tipc_bcbase_select_primary(): find a bearer with links to all destinations, 76 * if any, and make it primary bearer 77 */ 78 static void tipc_bcbase_select_primary(struct net *net) 79 { 80 struct tipc_bc_base *bb = tipc_bc_base(net); 81 int all_dests = tipc_link_bc_peers(bb->link); 82 int i, mtu; 83 84 bb->primary_bearer = INVALID_BEARER_ID; 85 86 if (!all_dests) 87 return; 88 89 for (i = 0; i < MAX_BEARERS; i++) { 90 if (!bb->dests[i]) 91 continue; 92 93 mtu = tipc_bearer_mtu(net, i); 94 if (mtu < tipc_link_mtu(bb->link)) 95 tipc_link_set_mtu(bb->link, mtu); 96 97 if (bb->dests[i] < all_dests) 98 continue; 99 100 bb->primary_bearer = i; 101 102 /* Reduce risk that all nodes select same primary */ 103 if ((i ^ tipc_own_addr(net)) & 1) 104 break; 105 } 106 } 107 108 void tipc_bcast_inc_bearer_dst_cnt(struct net *net, int bearer_id) 109 { 110 struct tipc_bc_base *bb = tipc_bc_base(net); 111 112 tipc_bcast_lock(net); 113 bb->dests[bearer_id]++; 114 tipc_bcbase_select_primary(net); 115 tipc_bcast_unlock(net); 116 } 117 118 void tipc_bcast_dec_bearer_dst_cnt(struct net *net, int bearer_id) 119 { 120 struct tipc_bc_base *bb = tipc_bc_base(net); 121 122 tipc_bcast_lock(net); 123 bb->dests[bearer_id]--; 124 tipc_bcbase_select_primary(net); 125 tipc_bcast_unlock(net); 126 } 127 128 /* tipc_bcbase_xmit - broadcast a packet queue across one or more bearers 129 * 130 * Note that number of reachable destinations, as indicated in the dests[] 131 * array, may transitionally differ from the number of destinations indicated 132 * in each sent buffer. We can sustain this. Excess destination nodes will 133 * drop and never acknowledge the unexpected packets, and missing destinations 134 * will either require retransmission (if they are just about to be added to 135 * the bearer), or be removed from the buffer's 'ackers' counter (if they 136 * just went down) 137 */ 138 static void tipc_bcbase_xmit(struct net *net, struct sk_buff_head *xmitq) 139 { 140 int bearer_id; 141 struct tipc_bc_base *bb = tipc_bc_base(net); 142 struct sk_buff *skb, *_skb; 143 struct sk_buff_head _xmitq; 144 145 if (skb_queue_empty(xmitq)) 146 return; 147 148 /* The typical case: at least one bearer has links to all nodes */ 149 bearer_id = bb->primary_bearer; 150 if (bearer_id >= 0) { 151 tipc_bearer_bc_xmit(net, bearer_id, xmitq); 152 return; 153 } 154 155 /* We have to transmit across all bearers */ 156 skb_queue_head_init(&_xmitq); 157 for (bearer_id = 0; bearer_id < MAX_BEARERS; bearer_id++) { 158 if (!bb->dests[bearer_id]) 159 continue; 160 161 skb_queue_walk(xmitq, skb) { 162 _skb = pskb_copy_for_clone(skb, GFP_ATOMIC); 163 if (!_skb) 164 break; 165 __skb_queue_tail(&_xmitq, _skb); 166 } 167 tipc_bearer_bc_xmit(net, bearer_id, &_xmitq); 168 } 169 __skb_queue_purge(xmitq); 170 __skb_queue_purge(&_xmitq); 171 } 172 173 /* tipc_bcast_xmit - deliver buffer chain to all nodes in cluster 174 * and to identified node local sockets 175 * @net: the applicable net namespace 176 * @list: chain of buffers containing message 177 * Consumes the buffer chain, except when returning -ELINKCONG 178 * Returns 0 if success, otherwise errno: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE 179 */ 180 int tipc_bcast_xmit(struct net *net, struct sk_buff_head *list) 181 { 182 struct tipc_link *l = tipc_bc_sndlink(net); 183 struct sk_buff_head xmitq, inputq, rcvq; 184 int rc = 0; 185 186 __skb_queue_head_init(&rcvq); 187 __skb_queue_head_init(&xmitq); 188 skb_queue_head_init(&inputq); 189 190 /* Prepare message clone for local node */ 191 if (unlikely(!tipc_msg_reassemble(list, &rcvq))) 192 return -EHOSTUNREACH; 193 194 tipc_bcast_lock(net); 195 if (tipc_link_bc_peers(l)) 196 rc = tipc_link_xmit(l, list, &xmitq); 197 tipc_bcast_unlock(net); 198 199 /* Don't send to local node if adding to link failed */ 200 if (unlikely(rc)) { 201 __skb_queue_purge(&rcvq); 202 return rc; 203 } 204 205 /* Broadcast to all nodes, inluding local node */ 206 tipc_bcbase_xmit(net, &xmitq); 207 tipc_sk_mcast_rcv(net, &rcvq, &inputq); 208 __skb_queue_purge(list); 209 return 0; 210 } 211 212 /* tipc_bcast_rcv - receive a broadcast packet, and deliver to rcv link 213 * 214 * RCU is locked, no other locks set 215 */ 216 int tipc_bcast_rcv(struct net *net, struct tipc_link *l, struct sk_buff *skb) 217 { 218 struct tipc_msg *hdr = buf_msg(skb); 219 struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq; 220 struct sk_buff_head xmitq; 221 int rc; 222 223 __skb_queue_head_init(&xmitq); 224 225 if (msg_mc_netid(hdr) != tipc_netid(net) || !tipc_link_is_up(l)) { 226 kfree_skb(skb); 227 return 0; 228 } 229 230 tipc_bcast_lock(net); 231 if (msg_user(hdr) == BCAST_PROTOCOL) 232 rc = tipc_link_bc_nack_rcv(l, skb, &xmitq); 233 else 234 rc = tipc_link_rcv(l, skb, NULL); 235 tipc_bcast_unlock(net); 236 237 tipc_bcbase_xmit(net, &xmitq); 238 239 /* Any socket wakeup messages ? */ 240 if (!skb_queue_empty(inputq)) 241 tipc_sk_rcv(net, inputq); 242 243 return rc; 244 } 245 246 /* tipc_bcast_ack_rcv - receive and handle a broadcast acknowledge 247 * 248 * RCU is locked, no other locks set 249 */ 250 void tipc_bcast_ack_rcv(struct net *net, struct tipc_link *l, u32 acked) 251 { 252 struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq; 253 struct sk_buff_head xmitq; 254 255 __skb_queue_head_init(&xmitq); 256 257 tipc_bcast_lock(net); 258 tipc_link_bc_ack_rcv(l, acked, &xmitq); 259 tipc_bcast_unlock(net); 260 261 tipc_bcbase_xmit(net, &xmitq); 262 263 /* Any socket wakeup messages ? */ 264 if (!skb_queue_empty(inputq)) 265 tipc_sk_rcv(net, inputq); 266 } 267 268 /* tipc_bcast_synch_rcv - check and update rcv link with peer's send state 269 * 270 * RCU is locked, no other locks set 271 */ 272 void tipc_bcast_sync_rcv(struct net *net, struct tipc_link *l, 273 struct tipc_msg *hdr) 274 { 275 struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq; 276 struct sk_buff_head xmitq; 277 278 __skb_queue_head_init(&xmitq); 279 280 tipc_bcast_lock(net); 281 if (msg_type(hdr) == STATE_MSG) { 282 tipc_link_bc_ack_rcv(l, msg_bcast_ack(hdr), &xmitq); 283 tipc_link_bc_sync_rcv(l, hdr, &xmitq); 284 } else { 285 tipc_link_bc_init_rcv(l, hdr); 286 } 287 tipc_bcast_unlock(net); 288 289 tipc_bcbase_xmit(net, &xmitq); 290 291 /* Any socket wakeup messages ? */ 292 if (!skb_queue_empty(inputq)) 293 tipc_sk_rcv(net, inputq); 294 } 295 296 /* tipc_bcast_add_peer - add a peer node to broadcast link and bearer 297 * 298 * RCU is locked, node lock is set 299 */ 300 void tipc_bcast_add_peer(struct net *net, struct tipc_link *uc_l, 301 struct sk_buff_head *xmitq) 302 { 303 struct tipc_link *snd_l = tipc_bc_sndlink(net); 304 305 tipc_bcast_lock(net); 306 tipc_link_add_bc_peer(snd_l, uc_l, xmitq); 307 tipc_bcbase_select_primary(net); 308 tipc_bcast_unlock(net); 309 } 310 311 /* tipc_bcast_remove_peer - remove a peer node from broadcast link and bearer 312 * 313 * RCU is locked, node lock is set 314 */ 315 void tipc_bcast_remove_peer(struct net *net, struct tipc_link *rcv_l) 316 { 317 struct tipc_link *snd_l = tipc_bc_sndlink(net); 318 struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq; 319 struct sk_buff_head xmitq; 320 321 __skb_queue_head_init(&xmitq); 322 323 tipc_bcast_lock(net); 324 tipc_link_remove_bc_peer(snd_l, rcv_l, &xmitq); 325 tipc_bcbase_select_primary(net); 326 tipc_bcast_unlock(net); 327 328 tipc_bcbase_xmit(net, &xmitq); 329 330 /* Any socket wakeup messages ? */ 331 if (!skb_queue_empty(inputq)) 332 tipc_sk_rcv(net, inputq); 333 } 334 335 static int __tipc_nl_add_bc_link_stat(struct sk_buff *skb, 336 struct tipc_stats *stats) 337 { 338 int i; 339 struct nlattr *nest; 340 341 struct nla_map { 342 __u32 key; 343 __u32 val; 344 }; 345 346 struct nla_map map[] = { 347 {TIPC_NLA_STATS_RX_INFO, stats->recv_info}, 348 {TIPC_NLA_STATS_RX_FRAGMENTS, stats->recv_fragments}, 349 {TIPC_NLA_STATS_RX_FRAGMENTED, stats->recv_fragmented}, 350 {TIPC_NLA_STATS_RX_BUNDLES, stats->recv_bundles}, 351 {TIPC_NLA_STATS_RX_BUNDLED, stats->recv_bundled}, 352 {TIPC_NLA_STATS_TX_INFO, stats->sent_info}, 353 {TIPC_NLA_STATS_TX_FRAGMENTS, stats->sent_fragments}, 354 {TIPC_NLA_STATS_TX_FRAGMENTED, stats->sent_fragmented}, 355 {TIPC_NLA_STATS_TX_BUNDLES, stats->sent_bundles}, 356 {TIPC_NLA_STATS_TX_BUNDLED, stats->sent_bundled}, 357 {TIPC_NLA_STATS_RX_NACKS, stats->recv_nacks}, 358 {TIPC_NLA_STATS_RX_DEFERRED, stats->deferred_recv}, 359 {TIPC_NLA_STATS_TX_NACKS, stats->sent_nacks}, 360 {TIPC_NLA_STATS_TX_ACKS, stats->sent_acks}, 361 {TIPC_NLA_STATS_RETRANSMITTED, stats->retransmitted}, 362 {TIPC_NLA_STATS_DUPLICATES, stats->duplicates}, 363 {TIPC_NLA_STATS_LINK_CONGS, stats->link_congs}, 364 {TIPC_NLA_STATS_MAX_QUEUE, stats->max_queue_sz}, 365 {TIPC_NLA_STATS_AVG_QUEUE, stats->queue_sz_counts ? 366 (stats->accu_queue_sz / stats->queue_sz_counts) : 0} 367 }; 368 369 nest = nla_nest_start(skb, TIPC_NLA_LINK_STATS); 370 if (!nest) 371 return -EMSGSIZE; 372 373 for (i = 0; i < ARRAY_SIZE(map); i++) 374 if (nla_put_u32(skb, map[i].key, map[i].val)) 375 goto msg_full; 376 377 nla_nest_end(skb, nest); 378 379 return 0; 380 msg_full: 381 nla_nest_cancel(skb, nest); 382 383 return -EMSGSIZE; 384 } 385 386 int tipc_nl_add_bc_link(struct net *net, struct tipc_nl_msg *msg) 387 { 388 int err; 389 void *hdr; 390 struct nlattr *attrs; 391 struct nlattr *prop; 392 struct tipc_net *tn = net_generic(net, tipc_net_id); 393 struct tipc_link *bcl = tn->bcl; 394 395 if (!bcl) 396 return 0; 397 398 tipc_bcast_lock(net); 399 400 hdr = genlmsg_put(msg->skb, msg->portid, msg->seq, &tipc_genl_family, 401 NLM_F_MULTI, TIPC_NL_LINK_GET); 402 if (!hdr) 403 return -EMSGSIZE; 404 405 attrs = nla_nest_start(msg->skb, TIPC_NLA_LINK); 406 if (!attrs) 407 goto msg_full; 408 409 /* The broadcast link is always up */ 410 if (nla_put_flag(msg->skb, TIPC_NLA_LINK_UP)) 411 goto attr_msg_full; 412 413 if (nla_put_flag(msg->skb, TIPC_NLA_LINK_BROADCAST)) 414 goto attr_msg_full; 415 if (nla_put_string(msg->skb, TIPC_NLA_LINK_NAME, bcl->name)) 416 goto attr_msg_full; 417 if (nla_put_u32(msg->skb, TIPC_NLA_LINK_RX, bcl->rcv_nxt)) 418 goto attr_msg_full; 419 if (nla_put_u32(msg->skb, TIPC_NLA_LINK_TX, bcl->snd_nxt)) 420 goto attr_msg_full; 421 422 prop = nla_nest_start(msg->skb, TIPC_NLA_LINK_PROP); 423 if (!prop) 424 goto attr_msg_full; 425 if (nla_put_u32(msg->skb, TIPC_NLA_PROP_WIN, bcl->window)) 426 goto prop_msg_full; 427 nla_nest_end(msg->skb, prop); 428 429 err = __tipc_nl_add_bc_link_stat(msg->skb, &bcl->stats); 430 if (err) 431 goto attr_msg_full; 432 433 tipc_bcast_unlock(net); 434 nla_nest_end(msg->skb, attrs); 435 genlmsg_end(msg->skb, hdr); 436 437 return 0; 438 439 prop_msg_full: 440 nla_nest_cancel(msg->skb, prop); 441 attr_msg_full: 442 nla_nest_cancel(msg->skb, attrs); 443 msg_full: 444 tipc_bcast_unlock(net); 445 genlmsg_cancel(msg->skb, hdr); 446 447 return -EMSGSIZE; 448 } 449 450 int tipc_bclink_reset_stats(struct net *net) 451 { 452 struct tipc_net *tn = net_generic(net, tipc_net_id); 453 struct tipc_link *bcl = tn->bcl; 454 455 if (!bcl) 456 return -ENOPROTOOPT; 457 458 tipc_bcast_lock(net); 459 memset(&bcl->stats, 0, sizeof(bcl->stats)); 460 tipc_bcast_unlock(net); 461 return 0; 462 } 463 464 static int tipc_bc_link_set_queue_limits(struct net *net, u32 limit) 465 { 466 struct tipc_link *l = tipc_bc_sndlink(net); 467 468 if (!l) 469 return -ENOPROTOOPT; 470 if (limit < BCLINK_WIN_MIN) 471 limit = BCLINK_WIN_MIN; 472 if (limit > TIPC_MAX_LINK_WIN) 473 return -EINVAL; 474 tipc_bcast_lock(net); 475 tipc_link_set_queue_limits(l, limit); 476 tipc_bcast_unlock(net); 477 return 0; 478 } 479 480 int tipc_nl_bc_link_set(struct net *net, struct nlattr *attrs[]) 481 { 482 int err; 483 u32 win; 484 struct nlattr *props[TIPC_NLA_PROP_MAX + 1]; 485 486 if (!attrs[TIPC_NLA_LINK_PROP]) 487 return -EINVAL; 488 489 err = tipc_nl_parse_link_prop(attrs[TIPC_NLA_LINK_PROP], props); 490 if (err) 491 return err; 492 493 if (!props[TIPC_NLA_PROP_WIN]) 494 return -EOPNOTSUPP; 495 496 win = nla_get_u32(props[TIPC_NLA_PROP_WIN]); 497 498 return tipc_bc_link_set_queue_limits(net, win); 499 } 500 501 int tipc_bcast_init(struct net *net) 502 { 503 struct tipc_net *tn = tipc_net(net); 504 struct tipc_bc_base *bb = NULL; 505 struct tipc_link *l = NULL; 506 507 bb = kzalloc(sizeof(*bb), GFP_ATOMIC); 508 if (!bb) 509 goto enomem; 510 tn->bcbase = bb; 511 spin_lock_init(&tipc_net(net)->bclock); 512 513 if (!tipc_link_bc_create(net, 0, 0, 514 U16_MAX, 515 BCLINK_WIN_DEFAULT, 516 0, 517 &bb->inputq, 518 NULL, 519 NULL, 520 &l)) 521 goto enomem; 522 bb->link = l; 523 tn->bcl = l; 524 return 0; 525 enomem: 526 kfree(bb); 527 kfree(l); 528 return -ENOMEM; 529 } 530 531 void tipc_bcast_reinit(struct net *net) 532 { 533 struct tipc_bc_base *b = tipc_bc_base(net); 534 535 msg_set_prevnode(b->link->pmsg, tipc_own_addr(net)); 536 } 537 538 void tipc_bcast_stop(struct net *net) 539 { 540 struct tipc_net *tn = net_generic(net, tipc_net_id); 541 542 synchronize_net(); 543 kfree(tn->bcbase); 544 kfree(tn->bcl); 545 } 546