tipc: eliminate message disordering during binding table update
[muen/linux.git] / net / tipc / name_table.c
1 /*
2  * net/tipc/name_table.c: TIPC name table code
3  *
4  * Copyright (c) 2000-2006, 2014-2018, Ericsson AB
5  * Copyright (c) 2004-2008, 2010-2014, Wind River Systems
6  * All rights reserved.
7  *
8  * Redistribution and use in source and binary forms, with or without
9  * modification, are permitted provided that the following conditions are met:
10  *
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.
19  *
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.
23  *
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.
35  */
36
37 #include <net/sock.h>
38 #include "core.h"
39 #include "netlink.h"
40 #include "name_table.h"
41 #include "name_distr.h"
42 #include "subscr.h"
43 #include "bcast.h"
44 #include "addr.h"
45 #include "node.h"
46 #include "group.h"
47
48 /**
49  * struct service_range - container for all bindings of a service range
50  * @lower: service range lower bound
51  * @upper: service range upper bound
52  * @tree_node: member of service range RB tree
53  * @local_publ: list of identical publications made from this node
54  *   Used by closest_first lookup and multicast lookup algorithm
55  * @all_publ: all publications identical to this one, whatever node and scope
56  *   Used by round-robin lookup algorithm
57  */
58 struct service_range {
59         u32 lower;
60         u32 upper;
61         struct rb_node tree_node;
62         struct list_head local_publ;
63         struct list_head all_publ;
64 };
65
66 /**
67  * struct tipc_service - container for all published instances of a service type
68  * @type: 32 bit 'type' value for service
69  * @ranges: rb tree containing all service ranges for this service
70  * @service_list: links to adjacent name ranges in hash chain
71  * @subscriptions: list of subscriptions for this service type
72  * @lock: spinlock controlling access to pertaining service ranges/publications
73  * @rcu: RCU callback head used for deferred freeing
74  */
75 struct tipc_service {
76         u32 type;
77         struct rb_root ranges;
78         struct hlist_node service_list;
79         struct list_head subscriptions;
80         spinlock_t lock; /* Covers service range list */
81         struct rcu_head rcu;
82 };
83
84 static int hash(int x)
85 {
86         return x & (TIPC_NAMETBL_SIZE - 1);
87 }
88
89 /**
90  * tipc_publ_create - create a publication structure
91  */
92 static struct publication *tipc_publ_create(u32 type, u32 lower, u32 upper,
93                                             u32 scope, u32 node, u32 port,
94                                             u32 key)
95 {
96         struct publication *publ = kzalloc(sizeof(*publ), GFP_ATOMIC);
97
98         if (!publ)
99                 return NULL;
100
101         publ->type = type;
102         publ->lower = lower;
103         publ->upper = upper;
104         publ->scope = scope;
105         publ->node = node;
106         publ->port = port;
107         publ->key = key;
108         INIT_LIST_HEAD(&publ->binding_sock);
109         INIT_LIST_HEAD(&publ->binding_node);
110         INIT_LIST_HEAD(&publ->local_publ);
111         INIT_LIST_HEAD(&publ->all_publ);
112         return publ;
113 }
114
115 /**
116  * tipc_service_create - create a service structure for the specified 'type'
117  *
118  * Allocates a single range structure and sets it to all 0's.
119  */
120 static struct tipc_service *tipc_service_create(u32 type, struct hlist_head *hd)
121 {
122         struct tipc_service *service = kzalloc(sizeof(*service), GFP_ATOMIC);
123
124         if (!service) {
125                 pr_warn("Service creation failed, no memory\n");
126                 return NULL;
127         }
128
129         spin_lock_init(&service->lock);
130         service->type = type;
131         service->ranges = RB_ROOT;
132         INIT_HLIST_NODE(&service->service_list);
133         INIT_LIST_HEAD(&service->subscriptions);
134         hlist_add_head_rcu(&service->service_list, hd);
135         return service;
136 }
137
138 /**
139  * tipc_service_first_range - find first service range in tree matching instance
140  *
141  * Very time-critical, so binary search through range rb tree
142  */
143 static struct service_range *tipc_service_first_range(struct tipc_service *sc,
144                                                       u32 instance)
145 {
146         struct rb_node *n = sc->ranges.rb_node;
147         struct service_range *sr;
148
149         while (n) {
150                 sr = container_of(n, struct service_range, tree_node);
151                 if (sr->lower > instance)
152                         n = n->rb_left;
153                 else if (sr->upper < instance)
154                         n = n->rb_right;
155                 else
156                         return sr;
157         }
158         return NULL;
159 }
160
161 /*  tipc_service_find_range - find service range matching publication parameters
162  */
163 static struct service_range *tipc_service_find_range(struct tipc_service *sc,
164                                                      u32 lower, u32 upper)
165 {
166         struct rb_node *n = sc->ranges.rb_node;
167         struct service_range *sr;
168
169         sr = tipc_service_first_range(sc, lower);
170         if (!sr)
171                 return NULL;
172
173         /* Look for exact match */
174         for (n = &sr->tree_node; n; n = rb_next(n)) {
175                 sr = container_of(n, struct service_range, tree_node);
176                 if (sr->upper == upper)
177                         break;
178         }
179         if (!n || sr->lower != lower || sr->upper != upper)
180                 return NULL;
181
182         return sr;
183 }
184
185 static struct service_range *tipc_service_create_range(struct tipc_service *sc,
186                                                        u32 lower, u32 upper)
187 {
188         struct rb_node **n, *parent = NULL;
189         struct service_range *sr, *tmp;
190
191         n = &sc->ranges.rb_node;
192         while (*n) {
193                 tmp = container_of(*n, struct service_range, tree_node);
194                 parent = *n;
195                 tmp = container_of(parent, struct service_range, tree_node);
196                 if (lower < tmp->lower)
197                         n = &(*n)->rb_left;
198                 else if (lower > tmp->lower)
199                         n = &(*n)->rb_right;
200                 else if (upper < tmp->upper)
201                         n = &(*n)->rb_left;
202                 else if (upper > tmp->upper)
203                         n = &(*n)->rb_right;
204                 else
205                         return tmp;
206         }
207         sr = kzalloc(sizeof(*sr), GFP_ATOMIC);
208         if (!sr)
209                 return NULL;
210         sr->lower = lower;
211         sr->upper = upper;
212         INIT_LIST_HEAD(&sr->local_publ);
213         INIT_LIST_HEAD(&sr->all_publ);
214         rb_link_node(&sr->tree_node, parent, n);
215         rb_insert_color(&sr->tree_node, &sc->ranges);
216         return sr;
217 }
218
219 static struct publication *tipc_service_insert_publ(struct net *net,
220                                                     struct tipc_service *sc,
221                                                     u32 type, u32 lower,
222                                                     u32 upper, u32 scope,
223                                                     u32 node, u32 port,
224                                                     u32 key)
225 {
226         struct tipc_subscription *sub, *tmp;
227         struct service_range *sr;
228         struct publication *p;
229         bool first = false;
230
231         sr = tipc_service_create_range(sc, lower, upper);
232         if (!sr)
233                 goto  err;
234
235         first = list_empty(&sr->all_publ);
236
237         /* Return if the publication already exists */
238         list_for_each_entry(p, &sr->all_publ, all_publ) {
239                 if (p->key == key && (!p->node || p->node == node))
240                         return NULL;
241         }
242
243         /* Create and insert publication */
244         p = tipc_publ_create(type, lower, upper, scope, node, port, key);
245         if (!p)
246                 goto err;
247         if (in_own_node(net, node))
248                 list_add(&p->local_publ, &sr->local_publ);
249         list_add(&p->all_publ, &sr->all_publ);
250
251         /* Any subscriptions waiting for notification?  */
252         list_for_each_entry_safe(sub, tmp, &sc->subscriptions, service_list) {
253                 tipc_sub_report_overlap(sub, p->lower, p->upper, TIPC_PUBLISHED,
254                                         p->port, p->node, p->scope, first);
255         }
256         return p;
257 err:
258         pr_warn("Failed to bind to %u,%u,%u, no memory\n", type, lower, upper);
259         return NULL;
260 }
261
262 /**
263  * tipc_service_remove_publ - remove a publication from a service
264  */
265 static struct publication *tipc_service_remove_publ(struct service_range *sr,
266                                                     u32 node, u32 key)
267 {
268         struct publication *p;
269
270         list_for_each_entry(p, &sr->all_publ, all_publ) {
271                 if (p->key != key || (node && node != p->node))
272                         continue;
273                 list_del(&p->all_publ);
274                 list_del(&p->local_publ);
275                 return p;
276         }
277         return NULL;
278 }
279
280 /**
281  * tipc_service_subscribe - attach a subscription, and optionally
282  * issue the prescribed number of events if there is any service
283  * range overlapping with the requested range
284  */
285 static void tipc_service_subscribe(struct tipc_service *service,
286                                    struct tipc_subscription *sub)
287 {
288         struct tipc_subscr *sb = &sub->evt.s;
289         struct service_range *sr;
290         struct tipc_name_seq ns;
291         struct publication *p;
292         struct rb_node *n;
293         bool first;
294
295         ns.type = tipc_sub_read(sb, seq.type);
296         ns.lower = tipc_sub_read(sb, seq.lower);
297         ns.upper = tipc_sub_read(sb, seq.upper);
298
299         tipc_sub_get(sub);
300         list_add(&sub->service_list, &service->subscriptions);
301
302         if (tipc_sub_read(sb, filter) & TIPC_SUB_NO_STATUS)
303                 return;
304
305         for (n = rb_first(&service->ranges); n; n = rb_next(n)) {
306                 sr = container_of(n, struct service_range, tree_node);
307                 if (sr->lower > ns.upper)
308                         break;
309                 if (!tipc_sub_check_overlap(&ns, sr->lower, sr->upper))
310                         continue;
311                 first = true;
312
313                 list_for_each_entry(p, &sr->all_publ, all_publ) {
314                         tipc_sub_report_overlap(sub, sr->lower, sr->upper,
315                                                 TIPC_PUBLISHED, p->port,
316                                                 p->node, p->scope, first);
317                         first = false;
318                 }
319         }
320 }
321
322 static struct tipc_service *tipc_service_find(struct net *net, u32 type)
323 {
324         struct name_table *nt = tipc_name_table(net);
325         struct hlist_head *service_head;
326         struct tipc_service *service;
327
328         service_head = &nt->services[hash(type)];
329         hlist_for_each_entry_rcu(service, service_head, service_list) {
330                 if (service->type == type)
331                         return service;
332         }
333         return NULL;
334 };
335
336 struct publication *tipc_nametbl_insert_publ(struct net *net, u32 type,
337                                              u32 lower, u32 upper,
338                                              u32 scope, u32 node,
339                                              u32 port, u32 key)
340 {
341         struct name_table *nt = tipc_name_table(net);
342         struct tipc_service *sc;
343         struct publication *p;
344
345         if (scope > TIPC_NODE_SCOPE || lower > upper) {
346                 pr_debug("Failed to bind illegal {%u,%u,%u} with scope %u\n",
347                          type, lower, upper, scope);
348                 return NULL;
349         }
350         sc = tipc_service_find(net, type);
351         if (!sc)
352                 sc = tipc_service_create(type, &nt->services[hash(type)]);
353         if (!sc)
354                 return NULL;
355
356         spin_lock_bh(&sc->lock);
357         p = tipc_service_insert_publ(net, sc, type, lower, upper,
358                                      scope, node, port, key);
359         spin_unlock_bh(&sc->lock);
360         return p;
361 }
362
363 struct publication *tipc_nametbl_remove_publ(struct net *net, u32 type,
364                                              u32 lower, u32 upper,
365                                              u32 node, u32 key)
366 {
367         struct tipc_service *sc = tipc_service_find(net, type);
368         struct tipc_subscription *sub, *tmp;
369         struct service_range *sr = NULL;
370         struct publication *p = NULL;
371         bool last;
372
373         if (!sc)
374                 return NULL;
375
376         spin_lock_bh(&sc->lock);
377         sr = tipc_service_find_range(sc, lower, upper);
378         if (!sr)
379                 goto exit;
380         p = tipc_service_remove_publ(sr, node, key);
381         if (!p)
382                 goto exit;
383
384         /* Notify any waiting subscriptions */
385         last = list_empty(&sr->all_publ);
386         list_for_each_entry_safe(sub, tmp, &sc->subscriptions, service_list) {
387                 tipc_sub_report_overlap(sub, lower, upper, TIPC_WITHDRAWN,
388                                         p->port, node, p->scope, last);
389         }
390
391         /* Remove service range item if this was its last publication */
392         if (list_empty(&sr->all_publ)) {
393                 rb_erase(&sr->tree_node, &sc->ranges);
394                 kfree(sr);
395         }
396
397         /* Delete service item if this no more publications and subscriptions */
398         if (RB_EMPTY_ROOT(&sc->ranges) && list_empty(&sc->subscriptions)) {
399                 hlist_del_init_rcu(&sc->service_list);
400                 kfree_rcu(sc, rcu);
401         }
402 exit:
403         spin_unlock_bh(&sc->lock);
404         return p;
405 }
406
407 /**
408  * tipc_nametbl_translate - perform service instance to socket translation
409  *
410  * On entry, 'dnode' is the search domain used during translation.
411  *
412  * On exit:
413  * - if translation is deferred to another node, leave 'dnode' unchanged and
414  *   return 0
415  * - if translation is attempted and succeeds, set 'dnode' to the publishing
416  *   node and return the published (non-zero) port number
417  * - if translation is attempted and fails, set 'dnode' to 0 and return 0
418  *
419  * Note that for legacy users (node configured with Z.C.N address format) the
420  * 'closest-first' lookup algorithm must be maintained, i.e., if dnode is 0
421  * we must look in the local binding list first
422  */
423 u32 tipc_nametbl_translate(struct net *net, u32 type, u32 instance, u32 *dnode)
424 {
425         struct tipc_net *tn = tipc_net(net);
426         bool legacy = tn->legacy_addr_format;
427         u32 self = tipc_own_addr(net);
428         struct service_range *sr;
429         struct tipc_service *sc;
430         struct list_head *list;
431         struct publication *p;
432         u32 port = 0;
433         u32 node = 0;
434
435         if (!tipc_in_scope(legacy, *dnode, self))
436                 return 0;
437
438         rcu_read_lock();
439         sc = tipc_service_find(net, type);
440         if (unlikely(!sc))
441                 goto not_found;
442
443         spin_lock_bh(&sc->lock);
444         sr = tipc_service_first_range(sc, instance);
445         if (unlikely(!sr))
446                 goto no_match;
447
448         /* Select lookup algorithm: local, closest-first or round-robin */
449         if (*dnode == self) {
450                 list = &sr->local_publ;
451                 if (list_empty(list))
452                         goto no_match;
453                 p = list_first_entry(list, struct publication, local_publ);
454                 list_move_tail(&p->local_publ, &sr->local_publ);
455         } else if (legacy && !*dnode && !list_empty(&sr->local_publ)) {
456                 list = &sr->local_publ;
457                 p = list_first_entry(list, struct publication, local_publ);
458                 list_move_tail(&p->local_publ, &sr->local_publ);
459         } else {
460                 list = &sr->all_publ;
461                 p = list_first_entry(list, struct publication, all_publ);
462                 list_move_tail(&p->all_publ, &sr->all_publ);
463         }
464         port = p->port;
465         node = p->node;
466 no_match:
467         spin_unlock_bh(&sc->lock);
468 not_found:
469         rcu_read_unlock();
470         *dnode = node;
471         return port;
472 }
473
474 bool tipc_nametbl_lookup(struct net *net, u32 type, u32 instance, u32 scope,
475                          struct list_head *dsts, int *dstcnt, u32 exclude,
476                          bool all)
477 {
478         u32 self = tipc_own_addr(net);
479         struct service_range *sr;
480         struct tipc_service *sc;
481         struct publication *p;
482
483         *dstcnt = 0;
484         rcu_read_lock();
485         sc = tipc_service_find(net, type);
486         if (unlikely(!sc))
487                 goto exit;
488
489         spin_lock_bh(&sc->lock);
490
491         sr = tipc_service_first_range(sc, instance);
492         if (!sr)
493                 goto no_match;
494
495         list_for_each_entry(p, &sr->all_publ, all_publ) {
496                 if (p->scope != scope)
497                         continue;
498                 if (p->port == exclude && p->node == self)
499                         continue;
500                 tipc_dest_push(dsts, p->node, p->port);
501                 (*dstcnt)++;
502                 if (all)
503                         continue;
504                 list_move_tail(&p->all_publ, &sr->all_publ);
505                 break;
506         }
507 no_match:
508         spin_unlock_bh(&sc->lock);
509 exit:
510         rcu_read_unlock();
511         return !list_empty(dsts);
512 }
513
514 void tipc_nametbl_mc_lookup(struct net *net, u32 type, u32 lower, u32 upper,
515                             u32 scope, bool exact, struct list_head *dports)
516 {
517         struct service_range *sr;
518         struct tipc_service *sc;
519         struct publication *p;
520         struct rb_node *n;
521
522         rcu_read_lock();
523         sc = tipc_service_find(net, type);
524         if (!sc)
525                 goto exit;
526
527         spin_lock_bh(&sc->lock);
528
529         for (n = rb_first(&sc->ranges); n; n = rb_next(n)) {
530                 sr = container_of(n, struct service_range, tree_node);
531                 if (sr->upper < lower)
532                         continue;
533                 if (sr->lower > upper)
534                         break;
535                 list_for_each_entry(p, &sr->local_publ, local_publ) {
536                         if (p->scope == scope || (!exact && p->scope < scope))
537                                 tipc_dest_push(dports, 0, p->port);
538                 }
539         }
540         spin_unlock_bh(&sc->lock);
541 exit:
542         rcu_read_unlock();
543 }
544
545 /* tipc_nametbl_lookup_dst_nodes - find broadcast destination nodes
546  * - Creates list of nodes that overlap the given multicast address
547  * - Determines if any node local destinations overlap
548  */
549 void tipc_nametbl_lookup_dst_nodes(struct net *net, u32 type, u32 lower,
550                                    u32 upper, struct tipc_nlist *nodes)
551 {
552         struct service_range *sr;
553         struct tipc_service *sc;
554         struct publication *p;
555         struct rb_node *n;
556
557         rcu_read_lock();
558         sc = tipc_service_find(net, type);
559         if (!sc)
560                 goto exit;
561
562         spin_lock_bh(&sc->lock);
563
564         for (n = rb_first(&sc->ranges); n; n = rb_next(n)) {
565                 sr = container_of(n, struct service_range, tree_node);
566                 if (sr->upper < lower)
567                         continue;
568                 if (sr->lower > upper)
569                         break;
570                 list_for_each_entry(p, &sr->all_publ, all_publ) {
571                         tipc_nlist_add(nodes, p->node);
572                 }
573         }
574         spin_unlock_bh(&sc->lock);
575 exit:
576         rcu_read_unlock();
577 }
578
579 /* tipc_nametbl_build_group - build list of communication group members
580  */
581 void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
582                               u32 type, u32 scope)
583 {
584         struct service_range *sr;
585         struct tipc_service *sc;
586         struct publication *p;
587         struct rb_node *n;
588
589         rcu_read_lock();
590         sc = tipc_service_find(net, type);
591         if (!sc)
592                 goto exit;
593
594         spin_lock_bh(&sc->lock);
595         for (n = rb_first(&sc->ranges); n; n = rb_next(n)) {
596                 sr = container_of(n, struct service_range, tree_node);
597                 list_for_each_entry(p, &sr->all_publ, all_publ) {
598                         if (p->scope != scope)
599                                 continue;
600                         tipc_group_add_member(grp, p->node, p->port, p->lower);
601                 }
602         }
603         spin_unlock_bh(&sc->lock);
604 exit:
605         rcu_read_unlock();
606 }
607
608 /* tipc_nametbl_publish - add service binding to name table
609  */
610 struct publication *tipc_nametbl_publish(struct net *net, u32 type, u32 lower,
611                                          u32 upper, u32 scope, u32 port,
612                                          u32 key)
613 {
614         struct name_table *nt = tipc_name_table(net);
615         struct tipc_net *tn = tipc_net(net);
616         struct publication *p = NULL;
617         struct sk_buff *skb = NULL;
618
619         spin_lock_bh(&tn->nametbl_lock);
620
621         if (nt->local_publ_count >= TIPC_MAX_PUBL) {
622                 pr_warn("Bind failed, max limit %u reached\n", TIPC_MAX_PUBL);
623                 goto exit;
624         }
625
626         p = tipc_nametbl_insert_publ(net, type, lower, upper, scope,
627                                      tipc_own_addr(net), port, key);
628         if (p) {
629                 nt->local_publ_count++;
630                 skb = tipc_named_publish(net, p);
631         }
632 exit:
633         spin_unlock_bh(&tn->nametbl_lock);
634
635         if (skb)
636                 tipc_node_broadcast(net, skb);
637         return p;
638 }
639
640 /**
641  * tipc_nametbl_withdraw - withdraw a service binding
642  */
643 int tipc_nametbl_withdraw(struct net *net, u32 type, u32 lower,
644                           u32 upper, u32 key)
645 {
646         struct name_table *nt = tipc_name_table(net);
647         struct tipc_net *tn = tipc_net(net);
648         u32 self = tipc_own_addr(net);
649         struct sk_buff *skb = NULL;
650         struct publication *p;
651
652         spin_lock_bh(&tn->nametbl_lock);
653
654         p = tipc_nametbl_remove_publ(net, type, lower, upper, self, key);
655         if (p) {
656                 nt->local_publ_count--;
657                 skb = tipc_named_withdraw(net, p);
658                 list_del_init(&p->binding_sock);
659                 kfree_rcu(p, rcu);
660         } else {
661                 pr_err("Failed to remove local publication {%u,%u,%u}/%u\n",
662                        type, lower, upper, key);
663         }
664         spin_unlock_bh(&tn->nametbl_lock);
665
666         if (skb) {
667                 tipc_node_broadcast(net, skb);
668                 return 1;
669         }
670         return 0;
671 }
672
673 /**
674  * tipc_nametbl_subscribe - add a subscription object to the name table
675  */
676 bool tipc_nametbl_subscribe(struct tipc_subscription *sub)
677 {
678         struct name_table *nt = tipc_name_table(sub->net);
679         struct tipc_net *tn = tipc_net(sub->net);
680         struct tipc_subscr *s = &sub->evt.s;
681         u32 type = tipc_sub_read(s, seq.type);
682         struct tipc_service *sc;
683         bool res = true;
684
685         spin_lock_bh(&tn->nametbl_lock);
686         sc = tipc_service_find(sub->net, type);
687         if (!sc)
688                 sc = tipc_service_create(type, &nt->services[hash(type)]);
689         if (sc) {
690                 spin_lock_bh(&sc->lock);
691                 tipc_service_subscribe(sc, sub);
692                 spin_unlock_bh(&sc->lock);
693         } else {
694                 pr_warn("Failed to subscribe for {%u,%u,%u}\n", type,
695                         tipc_sub_read(s, seq.lower),
696                         tipc_sub_read(s, seq.upper));
697                 res = false;
698         }
699         spin_unlock_bh(&tn->nametbl_lock);
700         return res;
701 }
702
703 /**
704  * tipc_nametbl_unsubscribe - remove a subscription object from name table
705  */
706 void tipc_nametbl_unsubscribe(struct tipc_subscription *sub)
707 {
708         struct tipc_net *tn = tipc_net(sub->net);
709         struct tipc_subscr *s = &sub->evt.s;
710         u32 type = tipc_sub_read(s, seq.type);
711         struct tipc_service *sc;
712
713         spin_lock_bh(&tn->nametbl_lock);
714         sc = tipc_service_find(sub->net, type);
715         if (!sc)
716                 goto exit;
717
718         spin_lock_bh(&sc->lock);
719         list_del_init(&sub->service_list);
720         tipc_sub_put(sub);
721
722         /* Delete service item if no more publications and subscriptions */
723         if (RB_EMPTY_ROOT(&sc->ranges) && list_empty(&sc->subscriptions)) {
724                 hlist_del_init_rcu(&sc->service_list);
725                 kfree_rcu(sc, rcu);
726         }
727         spin_unlock_bh(&sc->lock);
728 exit:
729         spin_unlock_bh(&tn->nametbl_lock);
730 }
731
732 int tipc_nametbl_init(struct net *net)
733 {
734         struct tipc_net *tn = tipc_net(net);
735         struct name_table *nt;
736         int i;
737
738         nt = kzalloc(sizeof(*nt), GFP_KERNEL);
739         if (!nt)
740                 return -ENOMEM;
741
742         for (i = 0; i < TIPC_NAMETBL_SIZE; i++)
743                 INIT_HLIST_HEAD(&nt->services[i]);
744
745         INIT_LIST_HEAD(&nt->node_scope);
746         INIT_LIST_HEAD(&nt->cluster_scope);
747         rwlock_init(&nt->cluster_scope_lock);
748         tn->nametbl = nt;
749         spin_lock_init(&tn->nametbl_lock);
750         return 0;
751 }
752
753 /**
754  *  tipc_service_delete - purge all publications for a service and delete it
755  */
756 static void tipc_service_delete(struct net *net, struct tipc_service *sc)
757 {
758         struct service_range *sr, *tmpr;
759         struct publication *p, *tmp;
760
761         spin_lock_bh(&sc->lock);
762         rbtree_postorder_for_each_entry_safe(sr, tmpr, &sc->ranges, tree_node) {
763                 list_for_each_entry_safe(p, tmp, &sr->all_publ, all_publ) {
764                         tipc_service_remove_publ(sr, p->node, p->key);
765                         kfree_rcu(p, rcu);
766                 }
767                 rb_erase(&sr->tree_node, &sc->ranges);
768                 kfree(sr);
769         }
770         hlist_del_init_rcu(&sc->service_list);
771         spin_unlock_bh(&sc->lock);
772         kfree_rcu(sc, rcu);
773 }
774
775 void tipc_nametbl_stop(struct net *net)
776 {
777         struct name_table *nt = tipc_name_table(net);
778         struct tipc_net *tn = tipc_net(net);
779         struct hlist_head *service_head;
780         struct tipc_service *service;
781         u32 i;
782
783         /* Verify name table is empty and purge any lingering
784          * publications, then release the name table
785          */
786         spin_lock_bh(&tn->nametbl_lock);
787         for (i = 0; i < TIPC_NAMETBL_SIZE; i++) {
788                 if (hlist_empty(&nt->services[i]))
789                         continue;
790                 service_head = &nt->services[i];
791                 hlist_for_each_entry_rcu(service, service_head, service_list) {
792                         tipc_service_delete(net, service);
793                 }
794         }
795         spin_unlock_bh(&tn->nametbl_lock);
796
797         synchronize_net();
798         kfree(nt);
799 }
800
801 static int __tipc_nl_add_nametable_publ(struct tipc_nl_msg *msg,
802                                         struct tipc_service *service,
803                                         struct service_range *sr,
804                                         u32 *last_key)
805 {
806         struct publication *p;
807         struct nlattr *attrs;
808         struct nlattr *b;
809         void *hdr;
810
811         if (*last_key) {
812                 list_for_each_entry(p, &sr->all_publ, all_publ)
813                         if (p->key == *last_key)
814                                 break;
815                 if (p->key != *last_key)
816                         return -EPIPE;
817         } else {
818                 p = list_first_entry(&sr->all_publ,
819                                      struct publication,
820                                      all_publ);
821         }
822
823         list_for_each_entry_from(p, &sr->all_publ, all_publ) {
824                 *last_key = p->key;
825
826                 hdr = genlmsg_put(msg->skb, msg->portid, msg->seq,
827                                   &tipc_genl_family, NLM_F_MULTI,
828                                   TIPC_NL_NAME_TABLE_GET);
829                 if (!hdr)
830                         return -EMSGSIZE;
831
832                 attrs = nla_nest_start(msg->skb, TIPC_NLA_NAME_TABLE);
833                 if (!attrs)
834                         goto msg_full;
835
836                 b = nla_nest_start(msg->skb, TIPC_NLA_NAME_TABLE_PUBL);
837                 if (!b)
838                         goto attr_msg_full;
839
840                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_TYPE, service->type))
841                         goto publ_msg_full;
842                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_LOWER, sr->lower))
843                         goto publ_msg_full;
844                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_UPPER, sr->upper))
845                         goto publ_msg_full;
846                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_SCOPE, p->scope))
847                         goto publ_msg_full;
848                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_NODE, p->node))
849                         goto publ_msg_full;
850                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_REF, p->port))
851                         goto publ_msg_full;
852                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_KEY, p->key))
853                         goto publ_msg_full;
854
855                 nla_nest_end(msg->skb, b);
856                 nla_nest_end(msg->skb, attrs);
857                 genlmsg_end(msg->skb, hdr);
858         }
859         *last_key = 0;
860
861         return 0;
862
863 publ_msg_full:
864         nla_nest_cancel(msg->skb, b);
865 attr_msg_full:
866         nla_nest_cancel(msg->skb, attrs);
867 msg_full:
868         genlmsg_cancel(msg->skb, hdr);
869
870         return -EMSGSIZE;
871 }
872
873 static int __tipc_nl_service_range_list(struct tipc_nl_msg *msg,
874                                         struct tipc_service *sc,
875                                         u32 *last_lower, u32 *last_key)
876 {
877         struct service_range *sr;
878         struct rb_node *n;
879         int err;
880
881         for (n = rb_first(&sc->ranges); n; n = rb_next(n)) {
882                 sr = container_of(n, struct service_range, tree_node);
883                 if (sr->lower < *last_lower)
884                         continue;
885                 err = __tipc_nl_add_nametable_publ(msg, sc, sr, last_key);
886                 if (err) {
887                         *last_lower = sr->lower;
888                         return err;
889                 }
890         }
891         *last_lower = 0;
892         return 0;
893 }
894
895 static int tipc_nl_service_list(struct net *net, struct tipc_nl_msg *msg,
896                                 u32 *last_type, u32 *last_lower, u32 *last_key)
897 {
898         struct tipc_net *tn = tipc_net(net);
899         struct tipc_service *service = NULL;
900         struct hlist_head *head;
901         int err;
902         int i;
903
904         if (*last_type)
905                 i = hash(*last_type);
906         else
907                 i = 0;
908
909         for (; i < TIPC_NAMETBL_SIZE; i++) {
910                 head = &tn->nametbl->services[i];
911
912                 if (*last_type) {
913                         service = tipc_service_find(net, *last_type);
914                         if (!service)
915                                 return -EPIPE;
916                 } else {
917                         hlist_for_each_entry_rcu(service, head, service_list)
918                                 break;
919                         if (!service)
920                                 continue;
921                 }
922
923                 hlist_for_each_entry_from_rcu(service, service_list) {
924                         spin_lock_bh(&service->lock);
925                         err = __tipc_nl_service_range_list(msg, service,
926                                                            last_lower,
927                                                            last_key);
928
929                         if (err) {
930                                 *last_type = service->type;
931                                 spin_unlock_bh(&service->lock);
932                                 return err;
933                         }
934                         spin_unlock_bh(&service->lock);
935                 }
936                 *last_type = 0;
937         }
938         return 0;
939 }
940
941 int tipc_nl_name_table_dump(struct sk_buff *skb, struct netlink_callback *cb)
942 {
943         struct net *net = sock_net(skb->sk);
944         u32 last_type = cb->args[0];
945         u32 last_lower = cb->args[1];
946         u32 last_key = cb->args[2];
947         int done = cb->args[3];
948         struct tipc_nl_msg msg;
949         int err;
950
951         if (done)
952                 return 0;
953
954         msg.skb = skb;
955         msg.portid = NETLINK_CB(cb->skb).portid;
956         msg.seq = cb->nlh->nlmsg_seq;
957
958         rcu_read_lock();
959         err = tipc_nl_service_list(net, &msg, &last_type,
960                                    &last_lower, &last_key);
961         if (!err) {
962                 done = 1;
963         } else if (err != -EMSGSIZE) {
964                 /* We never set seq or call nl_dump_check_consistent() this
965                  * means that setting prev_seq here will cause the consistence
966                  * check to fail in the netlink callback handler. Resulting in
967                  * the NLMSG_DONE message having the NLM_F_DUMP_INTR flag set if
968                  * we got an error.
969                  */
970                 cb->prev_seq = 1;
971         }
972         rcu_read_unlock();
973
974         cb->args[0] = last_type;
975         cb->args[1] = last_lower;
976         cb->args[2] = last_key;
977         cb->args[3] = done;
978
979         return skb->len;
980 }
981
982 struct tipc_dest *tipc_dest_find(struct list_head *l, u32 node, u32 port)
983 {
984         struct tipc_dest *dst;
985
986         list_for_each_entry(dst, l, list) {
987                 if (dst->node == node && dst->port == port)
988                         return dst;
989         }
990         return NULL;
991 }
992
993 bool tipc_dest_push(struct list_head *l, u32 node, u32 port)
994 {
995         struct tipc_dest *dst;
996
997         if (tipc_dest_find(l, node, port))
998                 return false;
999
1000         dst = kmalloc(sizeof(*dst), GFP_ATOMIC);
1001         if (unlikely(!dst))
1002                 return false;
1003         dst->node = node;
1004         dst->port = port;
1005         list_add(&dst->list, l);
1006         return true;
1007 }
1008
1009 bool tipc_dest_pop(struct list_head *l, u32 *node, u32 *port)
1010 {
1011         struct tipc_dest *dst;
1012
1013         if (list_empty(l))
1014                 return false;
1015         dst = list_first_entry(l, typeof(*dst), list);
1016         if (port)
1017                 *port = dst->port;
1018         if (node)
1019                 *node = dst->node;
1020         list_del(&dst->list);
1021         kfree(dst);
1022         return true;
1023 }
1024
1025 bool tipc_dest_del(struct list_head *l, u32 node, u32 port)
1026 {
1027         struct tipc_dest *dst;
1028
1029         dst = tipc_dest_find(l, node, port);
1030         if (!dst)
1031                 return false;
1032         list_del(&dst->list);
1033         kfree(dst);
1034         return true;
1035 }
1036
1037 void tipc_dest_list_purge(struct list_head *l)
1038 {
1039         struct tipc_dest *dst, *tmp;
1040
1041         list_for_each_entry_safe(dst, tmp, l, list) {
1042                 list_del(&dst->list);
1043                 kfree(dst);
1044         }
1045 }
1046
1047 int tipc_dest_list_len(struct list_head *l)
1048 {
1049         struct tipc_dest *dst;
1050         int i = 0;
1051
1052         list_for_each_entry(dst, l, list) {
1053                 i++;
1054         }
1055         return i;
1056 }