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, 251 struct tipc_msg *hdr) 252 { 253 struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq; 254 u16 acked = msg_bcast_ack(hdr); 255 struct sk_buff_head xmitq; 256 257 /* Ignore bc acks sent by peer before bcast synch point was received */ 258 if (msg_bc_ack_invalid(hdr)) 259 return; 260 261 __skb_queue_head_init(&xmitq); 262 263 tipc_bcast_lock(net); 264 tipc_link_bc_ack_rcv(l, acked, &xmitq); 265 tipc_bcast_unlock(net); 266 267 tipc_bcbase_xmit(net, &xmitq); 268 269 /* Any socket wakeup messages ? */ 270 if (!skb_queue_empty(inputq)) 271 tipc_sk_rcv(net, inputq); 272 } 273 274 /* tipc_bcast_synch_rcv - check and update rcv link with peer's send state 275 * 276 * RCU is locked, no other locks set 277 */ 278 int tipc_bcast_sync_rcv(struct net *net, struct tipc_link *l, 279 struct tipc_msg *hdr) 280 { 281 struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq; 282 struct sk_buff_head xmitq; 283 int rc = 0; 284 285 __skb_queue_head_init(&xmitq); 286 287 tipc_bcast_lock(net); 288 if (msg_type(hdr) != STATE_MSG) { 289 tipc_link_bc_init_rcv(l, hdr); 290 } else if (!msg_bc_ack_invalid(hdr)) { 291 tipc_link_bc_ack_rcv(l, msg_bcast_ack(hdr), &xmitq); 292 rc = tipc_link_bc_sync_rcv(l, hdr, &xmitq); 293 } 294 tipc_bcast_unlock(net); 295 296 tipc_bcbase_xmit(net, &xmitq); 297 298 /* Any socket wakeup messages ? */ 299 if (!skb_queue_empty(inputq)) 300 tipc_sk_rcv(net, inputq); 301 return rc; 302 } 303 304 /* tipc_bcast_add_peer - add a peer node to broadcast link and bearer 305 * 306 * RCU is locked, node lock is set 307 */ 308 void tipc_bcast_add_peer(struct net *net, struct tipc_link *uc_l, 309 struct sk_buff_head *xmitq) 310 { 311 struct tipc_link *snd_l = tipc_bc_sndlink(net); 312 313 tipc_bcast_lock(net); 314 tipc_link_add_bc_peer(snd_l, uc_l, xmitq); 315 tipc_bcbase_select_primary(net); 316 tipc_bcast_unlock(net); 317 } 318 319 /* tipc_bcast_remove_peer - remove a peer node from broadcast link and bearer 320 * 321 * RCU is locked, node lock is set 322 */ 323 void tipc_bcast_remove_peer(struct net *net, struct tipc_link *rcv_l) 324 { 325 struct tipc_link *snd_l = tipc_bc_sndlink(net); 326 struct sk_buff_head *inputq = &tipc_bc_base(net)->inputq; 327 struct sk_buff_head xmitq; 328 329 __skb_queue_head_init(&xmitq); 330 331 tipc_bcast_lock(net); 332 tipc_link_remove_bc_peer(snd_l, rcv_l, &xmitq); 333 tipc_bcbase_select_primary(net); 334 tipc_bcast_unlock(net); 335 336 tipc_bcbase_xmit(net, &xmitq); 337 338 /* Any socket wakeup messages ? */ 339 if (!skb_queue_empty(inputq)) 340 tipc_sk_rcv(net, inputq); 341 } 342 343 int tipc_bclink_reset_stats(struct net *net) 344 { 345 struct tipc_link *l = tipc_bc_sndlink(net); 346 347 if (!l) 348 return -ENOPROTOOPT; 349 350 tipc_bcast_lock(net); 351 tipc_link_reset_stats(l); 352 tipc_bcast_unlock(net); 353 return 0; 354 } 355 356 static int tipc_bc_link_set_queue_limits(struct net *net, u32 limit) 357 { 358 struct tipc_link *l = tipc_bc_sndlink(net); 359 360 if (!l) 361 return -ENOPROTOOPT; 362 if (limit < BCLINK_WIN_MIN) 363 limit = BCLINK_WIN_MIN; 364 if (limit > TIPC_MAX_LINK_WIN) 365 return -EINVAL; 366 tipc_bcast_lock(net); 367 tipc_link_set_queue_limits(l, limit); 368 tipc_bcast_unlock(net); 369 return 0; 370 } 371 372 int tipc_nl_bc_link_set(struct net *net, struct nlattr *attrs[]) 373 { 374 int err; 375 u32 win; 376 struct nlattr *props[TIPC_NLA_PROP_MAX + 1]; 377 378 if (!attrs[TIPC_NLA_LINK_PROP]) 379 return -EINVAL; 380 381 err = tipc_nl_parse_link_prop(attrs[TIPC_NLA_LINK_PROP], props); 382 if (err) 383 return err; 384 385 if (!props[TIPC_NLA_PROP_WIN]) 386 return -EOPNOTSUPP; 387 388 win = nla_get_u32(props[TIPC_NLA_PROP_WIN]); 389 390 return tipc_bc_link_set_queue_limits(net, win); 391 } 392 393 int tipc_bcast_init(struct net *net) 394 { 395 struct tipc_net *tn = tipc_net(net); 396 struct tipc_bc_base *bb = NULL; 397 struct tipc_link *l = NULL; 398 399 bb = kzalloc(sizeof(*bb), GFP_ATOMIC); 400 if (!bb) 401 goto enomem; 402 tn->bcbase = bb; 403 spin_lock_init(&tipc_net(net)->bclock); 404 405 if (!tipc_link_bc_create(net, 0, 0, 406 U16_MAX, 407 BCLINK_WIN_DEFAULT, 408 0, 409 &bb->inputq, 410 NULL, 411 NULL, 412 &l)) 413 goto enomem; 414 bb->link = l; 415 tn->bcl = l; 416 return 0; 417 enomem: 418 kfree(bb); 419 kfree(l); 420 return -ENOMEM; 421 } 422 423 void tipc_bcast_stop(struct net *net) 424 { 425 struct tipc_net *tn = net_generic(net, tipc_net_id); 426 427 synchronize_net(); 428 kfree(tn->bcbase); 429 kfree(tn->bcl); 430 } 431