1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37#include "core.h"
38#include "link.h"
39#include "name_distr.h"
40
41int sysctl_tipc_named_timeout __read_mostly = 2000;
42
43
44
45
46static struct list_head tipc_dist_queue = LIST_HEAD_INIT(tipc_dist_queue);
47
48struct distr_queue_item {
49 struct distr_item i;
50 u32 dtype;
51 u32 node;
52 unsigned long expires;
53 struct list_head next;
54};
55
56
57
58
59static void publ_to_item(struct distr_item *i, struct publication *p)
60{
61 i->type = htonl(p->type);
62 i->lower = htonl(p->lower);
63 i->upper = htonl(p->upper);
64 i->ref = htonl(p->ref);
65 i->key = htonl(p->key);
66}
67
68
69
70
71static struct sk_buff *named_prepare_buf(struct net *net, u32 type, u32 size,
72 u32 dest)
73{
74 struct tipc_net *tn = net_generic(net, tipc_net_id);
75 struct sk_buff *buf = tipc_buf_acquire(INT_H_SIZE + size);
76 struct tipc_msg *msg;
77
78 if (buf != NULL) {
79 msg = buf_msg(buf);
80 tipc_msg_init(tn->own_addr, msg, NAME_DISTRIBUTOR, type,
81 INT_H_SIZE, dest);
82 msg_set_size(msg, INT_H_SIZE + size);
83 }
84 return buf;
85}
86
87void named_cluster_distribute(struct net *net, struct sk_buff *skb)
88{
89 struct tipc_net *tn = net_generic(net, tipc_net_id);
90 struct sk_buff *oskb;
91 struct tipc_node *node;
92 u32 dnode;
93
94 rcu_read_lock();
95 list_for_each_entry_rcu(node, &tn->node_list, list) {
96 dnode = node->addr;
97 if (in_own_node(net, dnode))
98 continue;
99 if (!tipc_node_is_up(node))
100 continue;
101 oskb = pskb_copy(skb, GFP_ATOMIC);
102 if (!oskb)
103 break;
104 msg_set_destnode(buf_msg(oskb), dnode);
105 tipc_node_xmit_skb(net, oskb, dnode, 0);
106 }
107 rcu_read_unlock();
108
109 kfree_skb(skb);
110}
111
112
113
114
115struct sk_buff *tipc_named_publish(struct net *net, struct publication *publ)
116{
117 struct tipc_net *tn = net_generic(net, tipc_net_id);
118 struct sk_buff *buf;
119 struct distr_item *item;
120
121 list_add_tail_rcu(&publ->local_list,
122 &tn->nametbl->publ_list[publ->scope]);
123
124 if (publ->scope == TIPC_NODE_SCOPE)
125 return NULL;
126
127 buf = named_prepare_buf(net, PUBLICATION, ITEM_SIZE, 0);
128 if (!buf) {
129 pr_warn("Publication distribution failure\n");
130 return NULL;
131 }
132
133 item = (struct distr_item *)msg_data(buf_msg(buf));
134 publ_to_item(item, publ);
135 return buf;
136}
137
138
139
140
141struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *publ)
142{
143 struct sk_buff *buf;
144 struct distr_item *item;
145
146 list_del(&publ->local_list);
147
148 if (publ->scope == TIPC_NODE_SCOPE)
149 return NULL;
150
151 buf = named_prepare_buf(net, WITHDRAWAL, ITEM_SIZE, 0);
152 if (!buf) {
153 pr_warn("Withdrawal distribution failure\n");
154 return NULL;
155 }
156
157 item = (struct distr_item *)msg_data(buf_msg(buf));
158 publ_to_item(item, publ);
159 return buf;
160}
161
162
163
164
165
166
167
168static void named_distribute(struct net *net, struct sk_buff_head *list,
169 u32 dnode, struct list_head *pls)
170{
171 struct publication *publ;
172 struct sk_buff *skb = NULL;
173 struct distr_item *item = NULL;
174 uint msg_dsz = (tipc_node_get_mtu(net, dnode, 0) / ITEM_SIZE) *
175 ITEM_SIZE;
176 uint msg_rem = msg_dsz;
177
178 list_for_each_entry(publ, pls, local_list) {
179
180 if (!skb) {
181 skb = named_prepare_buf(net, PUBLICATION, msg_rem,
182 dnode);
183 if (!skb) {
184 pr_warn("Bulk publication failure\n");
185 return;
186 }
187 item = (struct distr_item *)msg_data(buf_msg(skb));
188 }
189
190
191 publ_to_item(item, publ);
192 item++;
193 msg_rem -= ITEM_SIZE;
194
195
196 if (!msg_rem) {
197 __skb_queue_tail(list, skb);
198 skb = NULL;
199 msg_rem = msg_dsz;
200 }
201 }
202 if (skb) {
203 msg_set_size(buf_msg(skb), INT_H_SIZE + (msg_dsz - msg_rem));
204 skb_trim(skb, INT_H_SIZE + (msg_dsz - msg_rem));
205 __skb_queue_tail(list, skb);
206 }
207}
208
209
210
211
212void tipc_named_node_up(struct net *net, u32 dnode)
213{
214 struct tipc_net *tn = net_generic(net, tipc_net_id);
215 struct sk_buff_head head;
216
217 __skb_queue_head_init(&head);
218
219 rcu_read_lock();
220 named_distribute(net, &head, dnode,
221 &tn->nametbl->publ_list[TIPC_CLUSTER_SCOPE]);
222 named_distribute(net, &head, dnode,
223 &tn->nametbl->publ_list[TIPC_ZONE_SCOPE]);
224 rcu_read_unlock();
225
226 tipc_node_xmit(net, &head, dnode, 0);
227}
228
229static void tipc_publ_subscribe(struct net *net, struct publication *publ,
230 u32 addr)
231{
232 struct tipc_node *node;
233
234 if (in_own_node(net, addr))
235 return;
236
237 node = tipc_node_find(net, addr);
238 if (!node) {
239 pr_warn("Node subscription rejected, unknown node 0x%x\n",
240 addr);
241 return;
242 }
243
244 tipc_node_lock(node);
245 list_add_tail(&publ->nodesub_list, &node->publ_list);
246 tipc_node_unlock(node);
247 tipc_node_put(node);
248}
249
250static void tipc_publ_unsubscribe(struct net *net, struct publication *publ,
251 u32 addr)
252{
253 struct tipc_node *node;
254
255 node = tipc_node_find(net, addr);
256 if (!node)
257 return;
258
259 tipc_node_lock(node);
260 list_del_init(&publ->nodesub_list);
261 tipc_node_unlock(node);
262 tipc_node_put(node);
263}
264
265
266
267
268
269
270
271static void tipc_publ_purge(struct net *net, struct publication *publ, u32 addr)
272{
273 struct tipc_net *tn = net_generic(net, tipc_net_id);
274 struct publication *p;
275
276 spin_lock_bh(&tn->nametbl_lock);
277 p = tipc_nametbl_remove_publ(net, publ->type, publ->lower,
278 publ->node, publ->ref, publ->key);
279 if (p)
280 tipc_publ_unsubscribe(net, p, addr);
281 spin_unlock_bh(&tn->nametbl_lock);
282
283 if (p != publ) {
284 pr_err("Unable to remove publication from failed node\n"
285 " (type=%u, lower=%u, node=0x%x, ref=%u, key=%u)\n",
286 publ->type, publ->lower, publ->node, publ->ref,
287 publ->key);
288 }
289
290 kfree_rcu(p, rcu);
291}
292
293void tipc_publ_notify(struct net *net, struct list_head *nsub_list, u32 addr)
294{
295 struct publication *publ, *tmp;
296
297 list_for_each_entry_safe(publ, tmp, nsub_list, nodesub_list)
298 tipc_publ_purge(net, publ, addr);
299}
300
301
302
303
304
305
306
307
308static bool tipc_update_nametbl(struct net *net, struct distr_item *i,
309 u32 node, u32 dtype)
310{
311 struct publication *publ = NULL;
312
313 if (dtype == PUBLICATION) {
314 publ = tipc_nametbl_insert_publ(net, ntohl(i->type),
315 ntohl(i->lower),
316 ntohl(i->upper),
317 TIPC_CLUSTER_SCOPE, node,
318 ntohl(i->ref), ntohl(i->key));
319 if (publ) {
320 tipc_publ_subscribe(net, publ, node);
321 return true;
322 }
323 } else if (dtype == WITHDRAWAL) {
324 publ = tipc_nametbl_remove_publ(net, ntohl(i->type),
325 ntohl(i->lower),
326 node, ntohl(i->ref),
327 ntohl(i->key));
328 if (publ) {
329 tipc_publ_unsubscribe(net, publ, node);
330 kfree_rcu(publ, rcu);
331 return true;
332 }
333 } else {
334 pr_warn("Unrecognized name table message received\n");
335 }
336 return false;
337}
338
339
340
341
342
343static void tipc_named_add_backlog(struct distr_item *i, u32 type, u32 node)
344{
345 struct distr_queue_item *e;
346 unsigned long now = get_jiffies_64();
347
348 e = kzalloc(sizeof(*e), GFP_ATOMIC);
349 if (!e)
350 return;
351 e->dtype = type;
352 e->node = node;
353 e->expires = now + msecs_to_jiffies(sysctl_tipc_named_timeout);
354 memcpy(e, i, sizeof(*i));
355 list_add_tail(&e->next, &tipc_dist_queue);
356}
357
358
359
360
361
362void tipc_named_process_backlog(struct net *net)
363{
364 struct distr_queue_item *e, *tmp;
365 char addr[16];
366 unsigned long now = get_jiffies_64();
367
368 list_for_each_entry_safe(e, tmp, &tipc_dist_queue, next) {
369 if (time_after(e->expires, now)) {
370 if (!tipc_update_nametbl(net, &e->i, e->node, e->dtype))
371 continue;
372 } else {
373 tipc_addr_string_fill(addr, e->node);
374 pr_warn_ratelimited("Dropping name table update (%d) of {%u, %u, %u} from %s key=%u\n",
375 e->dtype, ntohl(e->i.type),
376 ntohl(e->i.lower),
377 ntohl(e->i.upper),
378 addr, ntohl(e->i.key));
379 }
380 list_del(&e->next);
381 kfree(e);
382 }
383}
384
385
386
387
388void tipc_named_rcv(struct net *net, struct sk_buff_head *inputq)
389{
390 struct tipc_net *tn = net_generic(net, tipc_net_id);
391 struct tipc_msg *msg;
392 struct distr_item *item;
393 uint count;
394 u32 node;
395 struct sk_buff *skb;
396 int mtype;
397
398 spin_lock_bh(&tn->nametbl_lock);
399 for (skb = skb_dequeue(inputq); skb; skb = skb_dequeue(inputq)) {
400 msg = buf_msg(skb);
401 mtype = msg_type(msg);
402 item = (struct distr_item *)msg_data(msg);
403 count = msg_data_sz(msg) / ITEM_SIZE;
404 node = msg_orignode(msg);
405 while (count--) {
406 if (!tipc_update_nametbl(net, item, node, mtype))
407 tipc_named_add_backlog(item, mtype, node);
408 item++;
409 }
410 kfree_skb(skb);
411 tipc_named_process_backlog(net);
412 }
413 spin_unlock_bh(&tn->nametbl_lock);
414}
415
416
417
418
419
420
421
422
423void tipc_named_reinit(struct net *net)
424{
425 struct tipc_net *tn = net_generic(net, tipc_net_id);
426 struct publication *publ;
427 int scope;
428
429 spin_lock_bh(&tn->nametbl_lock);
430
431 for (scope = TIPC_ZONE_SCOPE; scope <= TIPC_NODE_SCOPE; scope++)
432 list_for_each_entry_rcu(publ, &tn->nametbl->publ_list[scope],
433 local_list)
434 publ->node = tn->own_addr;
435
436 spin_unlock_bh(&tn->nametbl_lock);
437}
438