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