2 * net/tipc/name_distr.c: TIPC name distribution code
4 * Copyright (c) 2000-2006, Ericsson AB
5 * Copyright (c) 2005, 2010-2011, Wind River Systems
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are met:
11 * 1. Redistributions of source code must retain the above copyright
12 * notice, this list of conditions and the following disclaimer.
13 * 2. Redistributions in binary form must reproduce the above copyright
14 * notice, this list of conditions and the following disclaimer in the
15 * documentation and/or other materials provided with the distribution.
16 * 3. Neither the names of the copyright holders nor the names of its
17 * contributors may be used to endorse or promote products derived from
18 * this software without specific prior written permission.
20 * Alternatively, this software may be distributed under the terms of the
21 * GNU General Public License ("GPL") version 2 as published by the Free
22 * Software Foundation.
24 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
25 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
26 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
27 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
28 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
29 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
30 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
31 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
32 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
33 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
34 * POSSIBILITY OF SUCH DAMAGE.
39 #include "name_distr.h"
41 #define ITEM_SIZE sizeof(struct distr_item)
44 * struct distr_item - publication info distributed to other nodes
45 * @type: name sequence type
46 * @lower: name sequence lower bound
47 * @upper: name sequence upper bound
48 * @ref: publishing port reference
49 * @key: publication key
51 * ===> All fields are stored in network byte order. <===
53 * First 3 fields identify (name or) name sequence being published.
54 * Reference field uniquely identifies port that published name sequence.
55 * Key field uniquely identifies publication, in the event a port has
56 * multiple publications of the same name sequence.
58 * Note: There is no field that identifies the publishing node because it is
59 * the same for all items contained within a publication message.
70 * struct publ_list - list of publications made by this node
71 * @list: circular list of publications
72 * @list_size: number of entries in list
75 struct list_head list
;
79 static struct publ_list publ_zone
= {
80 .list
= LIST_HEAD_INIT(publ_zone
.list
),
84 static struct publ_list publ_cluster
= {
85 .list
= LIST_HEAD_INIT(publ_cluster
.list
),
89 static struct publ_list publ_node
= {
90 .list
= LIST_HEAD_INIT(publ_node
.list
),
94 static struct publ_list
*publ_lists
[] = {
96 &publ_zone
, /* publ_lists[TIPC_ZONE_SCOPE] */
97 &publ_cluster
, /* publ_lists[TIPC_CLUSTER_SCOPE] */
98 &publ_node
/* publ_lists[TIPC_NODE_SCOPE] */
103 * publ_to_item - add publication info to a publication message
105 static void publ_to_item(struct distr_item
*i
, struct publication
*p
)
107 i
->type
= htonl(p
->type
);
108 i
->lower
= htonl(p
->lower
);
109 i
->upper
= htonl(p
->upper
);
110 i
->ref
= htonl(p
->ref
);
111 i
->key
= htonl(p
->key
);
115 * named_prepare_buf - allocate & initialize a publication message
117 static struct sk_buff
*named_prepare_buf(u32 type
, u32 size
, u32 dest
)
119 struct sk_buff
*buf
= tipc_buf_acquire(INT_H_SIZE
+ size
);
120 struct tipc_msg
*msg
;
124 tipc_msg_init(msg
, NAME_DISTRIBUTOR
, type
, INT_H_SIZE
, dest
);
125 msg_set_size(msg
, INT_H_SIZE
+ size
);
130 static void named_cluster_distribute(struct sk_buff
*buf
)
132 struct sk_buff
*buf_copy
;
133 struct tipc_node
*n_ptr
;
134 struct tipc_link
*l_ptr
;
137 list_for_each_entry_rcu(n_ptr
, &tipc_node_list
, list
) {
138 spin_lock_bh(&n_ptr
->lock
);
139 l_ptr
= n_ptr
->active_links
[n_ptr
->addr
& 1];
141 buf_copy
= skb_copy(buf
, GFP_ATOMIC
);
143 spin_unlock_bh(&n_ptr
->lock
);
146 msg_set_destnode(buf_msg(buf_copy
), n_ptr
->addr
);
147 __tipc_link_xmit(l_ptr
, buf_copy
);
149 spin_unlock_bh(&n_ptr
->lock
);
157 * tipc_named_publish - tell other nodes about a new publication by this node
159 void tipc_named_publish(struct publication
*publ
)
162 struct distr_item
*item
;
164 list_add_tail(&publ
->local_list
, &publ_lists
[publ
->scope
]->list
);
165 publ_lists
[publ
->scope
]->size
++;
167 if (publ
->scope
== TIPC_NODE_SCOPE
)
170 buf
= named_prepare_buf(PUBLICATION
, ITEM_SIZE
, 0);
172 pr_warn("Publication distribution failure\n");
176 item
= (struct distr_item
*)msg_data(buf_msg(buf
));
177 publ_to_item(item
, publ
);
178 named_cluster_distribute(buf
);
182 * tipc_named_withdraw - tell other nodes about a withdrawn publication by this node
184 void tipc_named_withdraw(struct publication
*publ
)
187 struct distr_item
*item
;
189 list_del(&publ
->local_list
);
190 publ_lists
[publ
->scope
]->size
--;
192 if (publ
->scope
== TIPC_NODE_SCOPE
)
195 buf
= named_prepare_buf(WITHDRAWAL
, ITEM_SIZE
, 0);
197 pr_warn("Withdrawal distribution failure\n");
201 item
= (struct distr_item
*)msg_data(buf_msg(buf
));
202 publ_to_item(item
, publ
);
203 named_cluster_distribute(buf
);
207 * named_distribute - prepare name info for bulk distribution to another node
209 static void named_distribute(struct list_head
*message_list
, u32 node
,
210 struct publ_list
*pls
, u32 max_item_buf
)
212 struct publication
*publ
;
213 struct sk_buff
*buf
= NULL
;
214 struct distr_item
*item
= NULL
;
216 u32 rest
= pls
->size
* ITEM_SIZE
;
218 list_for_each_entry(publ
, &pls
->list
, local_list
) {
220 left
= (rest
<= max_item_buf
) ? rest
: max_item_buf
;
222 buf
= named_prepare_buf(PUBLICATION
, left
, node
);
224 pr_warn("Bulk publication failure\n");
227 item
= (struct distr_item
*)msg_data(buf_msg(buf
));
229 publ_to_item(item
, publ
);
233 list_add_tail((struct list_head
*)buf
, message_list
);
240 * tipc_named_node_up - tell specified node about all publications by this node
242 void tipc_named_node_up(unsigned long nodearg
)
244 struct tipc_node
*n_ptr
;
245 struct tipc_link
*l_ptr
;
246 struct list_head message_list
;
247 u32 node
= (u32
)nodearg
;
248 u32 max_item_buf
= 0;
250 /* compute maximum amount of publication data to send per message */
251 read_lock_bh(&tipc_net_lock
);
252 n_ptr
= tipc_node_find(node
);
254 tipc_node_lock(n_ptr
);
255 l_ptr
= n_ptr
->active_links
[0];
257 max_item_buf
= ((l_ptr
->max_pkt
- INT_H_SIZE
) /
258 ITEM_SIZE
) * ITEM_SIZE
;
259 tipc_node_unlock(n_ptr
);
261 read_unlock_bh(&tipc_net_lock
);
265 /* create list of publication messages, then send them as a unit */
266 INIT_LIST_HEAD(&message_list
);
268 read_lock_bh(&tipc_nametbl_lock
);
269 named_distribute(&message_list
, node
, &publ_cluster
, max_item_buf
);
270 named_distribute(&message_list
, node
, &publ_zone
, max_item_buf
);
271 read_unlock_bh(&tipc_nametbl_lock
);
273 tipc_link_names_xmit(&message_list
, node
);
277 * named_purge_publ - remove publication associated with a failed node
279 * Invoked for each publication issued by a newly failed node.
280 * Removes publication structure from name table & deletes it.
282 static void named_purge_publ(struct publication
*publ
)
284 struct publication
*p
;
286 write_lock_bh(&tipc_nametbl_lock
);
287 p
= tipc_nametbl_remove_publ(publ
->type
, publ
->lower
,
288 publ
->node
, publ
->ref
, publ
->key
);
290 tipc_nodesub_unsubscribe(&p
->subscr
);
291 write_unlock_bh(&tipc_nametbl_lock
);
294 pr_err("Unable to remove publication from failed node\n"
295 " (type=%u, lower=%u, node=0x%x, ref=%u, key=%u)\n",
296 publ
->type
, publ
->lower
, publ
->node
, publ
->ref
,
304 * tipc_named_rcv - process name table update message sent by another node
306 void tipc_named_rcv(struct sk_buff
*buf
)
308 struct publication
*publ
;
309 struct tipc_msg
*msg
= buf_msg(buf
);
310 struct distr_item
*item
= (struct distr_item
*)msg_data(msg
);
311 u32 count
= msg_data_sz(msg
) / ITEM_SIZE
;
313 write_lock_bh(&tipc_nametbl_lock
);
315 if (msg_type(msg
) == PUBLICATION
) {
316 publ
= tipc_nametbl_insert_publ(ntohl(item
->type
),
324 tipc_nodesub_subscribe(&publ
->subscr
,
330 } else if (msg_type(msg
) == WITHDRAWAL
) {
331 publ
= tipc_nametbl_remove_publ(ntohl(item
->type
),
338 tipc_nodesub_unsubscribe(&publ
->subscr
);
341 pr_err("Unable to remove publication by node 0x%x\n"
342 " (type=%u, lower=%u, ref=%u, key=%u)\n",
343 msg_orignode(msg
), ntohl(item
->type
),
344 ntohl(item
->lower
), ntohl(item
->ref
),
348 pr_warn("Unrecognized name table message received\n");
352 write_unlock_bh(&tipc_nametbl_lock
);
357 * tipc_named_reinit - re-initialize local publications
359 * This routine is called whenever TIPC networking is enabled.
360 * All name table entries published by this node are updated to reflect
361 * the node's new network address.
363 void tipc_named_reinit(void)
365 struct publication
*publ
;
368 write_lock_bh(&tipc_nametbl_lock
);
370 for (scope
= TIPC_ZONE_SCOPE
; scope
<= TIPC_NODE_SCOPE
; scope
++)
371 list_for_each_entry(publ
, &publ_lists
[scope
]->list
, local_list
)
372 publ
->node
= tipc_own_addr
;
374 write_unlock_bh(&tipc_nametbl_lock
);
This page took 0.047147 seconds and 5 git commands to generate.