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 int tipc_bclink_reset_stats(struct net *net) 336 { 337 struct tipc_link *l = tipc_bc_sndlink(net); 338 339 if (!l) 340 return -ENOPROTOOPT; 341 342 tipc_bcast_lock(net); 343 tipc_link_reset_stats(l); 344 tipc_bcast_unlock(net); 345 return 0; 346 } 347 348 static int tipc_bc_link_set_queue_limits(struct net *net, u32 limit) 349 { 350 struct tipc_link *l = tipc_bc_sndlink(net); 351 352 if (!l) 353 return -ENOPROTOOPT; 354 if (limit < BCLINK_WIN_MIN) 355 limit = BCLINK_WIN_MIN; 356 if (limit > TIPC_MAX_LINK_WIN) 357 return -EINVAL; 358 tipc_bcast_lock(net); 359 tipc_link_set_queue_limits(l, limit); 360 tipc_bcast_unlock(net); 361 return 0; 362 } 363 364 int tipc_nl_bc_link_set(struct net *net, struct nlattr *attrs[]) 365 { 366 int err; 367 u32 win; 368 struct nlattr *props[TIPC_NLA_PROP_MAX + 1]; 369 370 if (!attrs[TIPC_NLA_LINK_PROP]) 371 return -EINVAL; 372 373 err = tipc_nl_parse_link_prop(attrs[TIPC_NLA_LINK_PROP], props); 374 if (err) 375 return err; 376 377 if (!props[TIPC_NLA_PROP_WIN]) 378 return -EOPNOTSUPP; 379 380 win = nla_get_u32(props[TIPC_NLA_PROP_WIN]); 381 382 return tipc_bc_link_set_queue_limits(net, win); 383 } 384 385 int tipc_bcast_init(struct net *net) 386 { 387 struct tipc_net *tn = tipc_net(net); 388 struct tipc_bc_base *bb = NULL; 389 struct tipc_link *l = NULL; 390 391 bb = kzalloc(sizeof(*bb), GFP_ATOMIC); 392 if (!bb) 393 goto enomem; 394 tn->bcbase = bb; 395 spin_lock_init(&tipc_net(net)->bclock); 396 397 if (!tipc_link_bc_create(net, 0, 0, 398 U16_MAX, 399 BCLINK_WIN_DEFAULT, 400 0, 401 &bb->inputq, 402 NULL, 403 NULL, 404 &l)) 405 goto enomem; 406 bb->link = l; 407 tn->bcl = l; 408 return 0; 409 enomem: 410 kfree(bb); 411 kfree(l); 412 return -ENOMEM; 413 } 414 415 void tipc_bcast_stop(struct net *net) 416 { 417 struct tipc_net *tn = net_generic(net, tipc_net_id); 418 419 synchronize_net(); 420 kfree(tn->bcbase); 421 kfree(tn->bcl); 422 } 423