]>
git.proxmox.com Git - mirror_ubuntu-artful-kernel.git/blob - net/tipc/bcast.c
2 * net/tipc/bcast.c: TIPC broadcast code
4 * Copyright (c) 2004-2006, 2014-2015, Ericsson AB
5 * Copyright (c) 2004, Intel Corporation.
6 * Copyright (c) 2005, 2010-2011, Wind River Systems
9 * Redistribution and use in source and binary forms, with or without
10 * modification, are permitted provided that the following conditions are met:
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.
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.
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.
38 #include <linux/tipc_config.h>
42 #include "name_distr.h"
46 #define BCLINK_WIN_DEFAULT 50 /* bcast link window size (default) */
47 #define BCLINK_WIN_MIN 32 /* bcast minimum link window size */
49 const char tipc_bclink_name
[] = "broadcast-link";
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
59 struct tipc_link
*link
;
60 struct sk_buff_head inputq
;
61 int dests
[MAX_BEARERS
];
65 static struct tipc_bc_base
*tipc_bc_base(struct net
*net
)
67 return tipc_net(net
)->bcbase
;
70 int tipc_bcast_get_mtu(struct net
*net
)
72 return tipc_link_mtu(tipc_bc_sndlink(net
));
75 /* tipc_bcbase_select_primary(): find a bearer with links to all destinations,
76 * if any, and make it primary bearer
78 static void tipc_bcbase_select_primary(struct net
*net
)
80 struct tipc_bc_base
*bb
= tipc_bc_base(net
);
81 int all_dests
= tipc_link_bc_peers(bb
->link
);
84 bb
->primary_bearer
= INVALID_BEARER_ID
;
89 for (i
= 0; i
< MAX_BEARERS
; i
++) {
93 mtu
= tipc_bearer_mtu(net
, i
);
94 if (mtu
< tipc_link_mtu(bb
->link
))
95 tipc_link_set_mtu(bb
->link
, mtu
);
97 if (bb
->dests
[i
] < all_dests
)
100 bb
->primary_bearer
= i
;
102 /* Reduce risk that all nodes select same primary */
103 if ((i
^ tipc_own_addr(net
)) & 1)
108 void tipc_bcast_inc_bearer_dst_cnt(struct net
*net
, int bearer_id
)
110 struct tipc_bc_base
*bb
= tipc_bc_base(net
);
112 tipc_bcast_lock(net
);
113 bb
->dests
[bearer_id
]++;
114 tipc_bcbase_select_primary(net
);
115 tipc_bcast_unlock(net
);
118 void tipc_bcast_dec_bearer_dst_cnt(struct net
*net
, int bearer_id
)
120 struct tipc_bc_base
*bb
= tipc_bc_base(net
);
122 tipc_bcast_lock(net
);
123 bb
->dests
[bearer_id
]--;
124 tipc_bcbase_select_primary(net
);
125 tipc_bcast_unlock(net
);
128 /* tipc_bcbase_xmit - broadcast a packet queue across one or more bearers
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
138 static void tipc_bcbase_xmit(struct net
*net
, struct sk_buff_head
*xmitq
)
141 struct tipc_bc_base
*bb
= tipc_bc_base(net
);
142 struct sk_buff
*skb
, *_skb
;
143 struct sk_buff_head _xmitq
;
145 if (skb_queue_empty(xmitq
))
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
);
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
])
161 skb_queue_walk(xmitq
, skb
) {
162 _skb
= pskb_copy_for_clone(skb
, GFP_ATOMIC
);
165 __skb_queue_tail(&_xmitq
, _skb
);
167 tipc_bearer_bc_xmit(net
, bearer_id
, &_xmitq
);
169 __skb_queue_purge(xmitq
);
170 __skb_queue_purge(&_xmitq
);
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
180 int tipc_bcast_xmit(struct net
*net
, struct sk_buff_head
*list
)
182 struct tipc_link
*l
= tipc_bc_sndlink(net
);
183 struct sk_buff_head xmitq
, inputq
, rcvq
;
186 __skb_queue_head_init(&rcvq
);
187 __skb_queue_head_init(&xmitq
);
188 skb_queue_head_init(&inputq
);
190 /* Prepare message clone for local node */
191 if (unlikely(!tipc_msg_reassemble(list
, &rcvq
)))
192 return -EHOSTUNREACH
;
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
);
199 /* Don't send to local node if adding to link failed */
201 __skb_queue_purge(&rcvq
);
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
);
212 /* tipc_bcast_rcv - receive a broadcast packet, and deliver to rcv link
214 * RCU is locked, no other locks set
216 int tipc_bcast_rcv(struct net
*net
, struct tipc_link
*l
, struct sk_buff
*skb
)
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
;
223 __skb_queue_head_init(&xmitq
);
225 if (msg_mc_netid(hdr
) != tipc_netid(net
) || !tipc_link_is_up(l
)) {
230 tipc_bcast_lock(net
);
231 if (msg_user(hdr
) == BCAST_PROTOCOL
)
232 rc
= tipc_link_bc_nack_rcv(l
, skb
, &xmitq
);
234 rc
= tipc_link_rcv(l
, skb
, NULL
);
235 tipc_bcast_unlock(net
);
237 tipc_bcbase_xmit(net
, &xmitq
);
239 /* Any socket wakeup messages ? */
240 if (!skb_queue_empty(inputq
))
241 tipc_sk_rcv(net
, inputq
);
246 /* tipc_bcast_ack_rcv - receive and handle a broadcast acknowledge
248 * RCU is locked, no other locks set
250 void tipc_bcast_ack_rcv(struct net
*net
, struct tipc_link
*l
, u32 acked
)
252 struct sk_buff_head
*inputq
= &tipc_bc_base(net
)->inputq
;
253 struct sk_buff_head xmitq
;
255 __skb_queue_head_init(&xmitq
);
257 tipc_bcast_lock(net
);
258 tipc_link_bc_ack_rcv(l
, acked
, &xmitq
);
259 tipc_bcast_unlock(net
);
261 tipc_bcbase_xmit(net
, &xmitq
);
263 /* Any socket wakeup messages ? */
264 if (!skb_queue_empty(inputq
))
265 tipc_sk_rcv(net
, inputq
);
268 /* tipc_bcast_synch_rcv - check and update rcv link with peer's send state
270 * RCU is locked, no other locks set
272 void tipc_bcast_sync_rcv(struct net
*net
, struct tipc_link
*l
,
273 struct tipc_msg
*hdr
)
275 struct sk_buff_head
*inputq
= &tipc_bc_base(net
)->inputq
;
276 struct sk_buff_head xmitq
;
278 __skb_queue_head_init(&xmitq
);
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
);
285 tipc_link_bc_init_rcv(l
, hdr
);
287 tipc_bcast_unlock(net
);
289 tipc_bcbase_xmit(net
, &xmitq
);
291 /* Any socket wakeup messages ? */
292 if (!skb_queue_empty(inputq
))
293 tipc_sk_rcv(net
, inputq
);
296 /* tipc_bcast_add_peer - add a peer node to broadcast link and bearer
298 * RCU is locked, node lock is set
300 void tipc_bcast_add_peer(struct net
*net
, struct tipc_link
*uc_l
,
301 struct sk_buff_head
*xmitq
)
303 struct tipc_link
*snd_l
= tipc_bc_sndlink(net
);
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
);
311 /* tipc_bcast_remove_peer - remove a peer node from broadcast link and bearer
313 * RCU is locked, node lock is set
315 void tipc_bcast_remove_peer(struct net
*net
, struct tipc_link
*rcv_l
)
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
;
321 __skb_queue_head_init(&xmitq
);
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
);
328 tipc_bcbase_xmit(net
, &xmitq
);
330 /* Any socket wakeup messages ? */
331 if (!skb_queue_empty(inputq
))
332 tipc_sk_rcv(net
, inputq
);
335 int tipc_bclink_reset_stats(struct net
*net
)
337 struct tipc_link
*l
= tipc_bc_sndlink(net
);
342 tipc_bcast_lock(net
);
343 tipc_link_reset_stats(l
);
344 tipc_bcast_unlock(net
);
348 static int tipc_bc_link_set_queue_limits(struct net
*net
, u32 limit
)
350 struct tipc_link
*l
= tipc_bc_sndlink(net
);
354 if (limit
< BCLINK_WIN_MIN
)
355 limit
= BCLINK_WIN_MIN
;
356 if (limit
> TIPC_MAX_LINK_WIN
)
358 tipc_bcast_lock(net
);
359 tipc_link_set_queue_limits(l
, limit
);
360 tipc_bcast_unlock(net
);
364 int tipc_nl_bc_link_set(struct net
*net
, struct nlattr
*attrs
[])
368 struct nlattr
*props
[TIPC_NLA_PROP_MAX
+ 1];
370 if (!attrs
[TIPC_NLA_LINK_PROP
])
373 err
= tipc_nl_parse_link_prop(attrs
[TIPC_NLA_LINK_PROP
], props
);
377 if (!props
[TIPC_NLA_PROP_WIN
])
380 win
= nla_get_u32(props
[TIPC_NLA_PROP_WIN
]);
382 return tipc_bc_link_set_queue_limits(net
, win
);
385 int tipc_bcast_init(struct net
*net
)
387 struct tipc_net
*tn
= tipc_net(net
);
388 struct tipc_bc_base
*bb
= NULL
;
389 struct tipc_link
*l
= NULL
;
391 bb
= kzalloc(sizeof(*bb
), GFP_ATOMIC
);
395 spin_lock_init(&tipc_net(net
)->bclock
);
397 if (!tipc_link_bc_create(net
, 0, 0,
415 void tipc_bcast_stop(struct net
*net
)
417 struct tipc_net
*tn
= net_generic(net
, tipc_net_id
);