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 #include <net/genetlink.h>
  48
  49 #define TIPC_NAMETBL_SIZE 1024          /* must be a power of 2 */
  50
  51 /**
  52  * struct name_info - name sequence publication info
  53  * @node_list: list of publications on own node of this <type,lower,upper>
  54  * @all_publ: list of all publications of this <type,lower,upper>
  55  */
  56 struct name_info {
  57         struct list_head local_publ;
  58         struct list_head all_publ;
  59 };
  60
  61 /**
  62  * struct sub_seq - container for all published instances of a name sequence
  63  * @lower: name sequence lower bound
  64  * @upper: name sequence upper bound
  65  * @info: pointer to name sequence publication info
  66  */
  67 struct sub_seq {
  68         u32 lower;
  69         u32 upper;
  70         struct name_info *info;
  71 };
  72
  73 /**
  74  * struct name_seq - container for all published instances of a name type
  75  * @type: 32 bit 'type' value for name sequence
  76  * @sseq: pointer to dynamically-sized array of sub-sequences of this 'type';
  77  *        sub-sequences are sorted in ascending order
  78  * @alloc: number of sub-sequences currently in array
  79  * @first_free: array index of first unused sub-sequence entry
  80  * @ns_list: links to adjacent name sequences in hash chain
  81  * @subscriptions: list of subscriptions for this 'type'
  82  * @lock: spinlock controlling access to publication lists of all sub-sequences
  83  * @rcu: RCU callback head used for deferred freeing
  84  */
  85 struct name_seq {
  86         u32 type;
  87         struct sub_seq *sseqs;
  88         u32 alloc;
  89         u32 first_free;
  90         struct hlist_node ns_list;
  91         struct list_head subscriptions;
  92         spinlock_t lock;
  93         struct rcu_head rcu;
  94 };
  95
  96 static int hash(int x)
  97 {
  98         return x & (TIPC_NAMETBL_SIZE - 1);
  99 }
 100
 101 /**
 102  * publ_create - create a publication structure
 103  */
 104 static struct publication *publ_create(u32 type, u32 lower, u32 upper,
 105                                        u32 scope, u32 node, u32 port,
 106                                        u32 key)
 107 {
 108         struct publication *publ = kzalloc(sizeof(*publ), GFP_ATOMIC);
 109         if (publ == NULL) {
 110                 pr_warn("Publication creation failure, no memory\n");
 111                 return NULL;
 112         }
 113
 114         publ->type = type;
 115         publ->lower = lower;
 116         publ->upper = upper;
 117         publ->scope = scope;
 118         publ->node = node;
 119         publ->port = port;
 120         publ->key = key;
 121         INIT_LIST_HEAD(&publ->binding_sock);
 122         return publ;
 123 }
 124
 125 /**
 126  * tipc_subseq_alloc - allocate a specified number of sub-sequence structures
 127  */
 128 static struct sub_seq *tipc_subseq_alloc(u32 cnt)
 129 {
 130         return kcalloc(cnt, sizeof(struct sub_seq), GFP_ATOMIC);
 131 }
 132
 133 /**
 134  * tipc_nameseq_create - create a name sequence structure for the specified 'type'
 135  *
 136  * Allocates a single sub-sequence structure and sets it to all 0's.
 137  */
 138 static struct name_seq *tipc_nameseq_create(u32 type, struct hlist_head *seq_head)
 139 {
 140         struct name_seq *nseq = kzalloc(sizeof(*nseq), GFP_ATOMIC);
 141         struct sub_seq *sseq = tipc_subseq_alloc(1);
 142
 143         if (!nseq || !sseq) {
 144                 pr_warn("Name sequence creation failed, no memory\n");
 145                 kfree(nseq);
 146                 kfree(sseq);
 147                 return NULL;
 148         }
 149
 150         spin_lock_init(&nseq->lock);
 151         nseq->type = type;
 152         nseq->sseqs = sseq;
 153         nseq->alloc = 1;
 154         INIT_HLIST_NODE(&nseq->ns_list);
 155         INIT_LIST_HEAD(&nseq->subscriptions);
 156         hlist_add_head_rcu(&nseq->ns_list, seq_head);
 157         return nseq;
 158 }
 159
 160 /**
 161  * nameseq_find_subseq - find sub-sequence (if any) matching a name instance
 162  *
 163  * Very time-critical, so binary searches through sub-sequence array.
 164  */
 165 static struct sub_seq *nameseq_find_subseq(struct name_seq *nseq,
 166                                            u32 instance)
 167 {
 168         struct sub_seq *sseqs = nseq->sseqs;
 169         int low = 0;
 170         int high = nseq->first_free - 1;
 171         int mid;
 172
 173         while (low <= high) {
 174                 mid = (low + high) / 2;
 175                 if (instance < sseqs[mid].lower)
 176                         high = mid - 1;
 177                 else if (instance > sseqs[mid].upper)
 178                         low = mid + 1;
 179                 else
 180                         return &sseqs[mid];
 181         }
 182         return NULL;
 183 }
 184
 185 /**
 186  * nameseq_locate_subseq - determine position of name instance in sub-sequence
 187  *
 188  * Returns index in sub-sequence array of the entry that contains the specified
 189  * instance value; if no entry contains that value, returns the position
 190  * where a new entry for it would be inserted in the array.
 191  *
 192  * Note: Similar to binary search code for locating a sub-sequence.
 193  */
 194 static u32 nameseq_locate_subseq(struct name_seq *nseq, u32 instance)
 195 {
 196         struct sub_seq *sseqs = nseq->sseqs;
 197         int low = 0;
 198         int high = nseq->first_free - 1;
 199         int mid;
 200
 201         while (low <= high) {
 202                 mid = (low + high) / 2;
 203                 if (instance < sseqs[mid].lower)
 204                         high = mid - 1;
 205                 else if (instance > sseqs[mid].upper)
 206                         low = mid + 1;
 207                 else
 208                         return mid;
 209         }
 210         return low;
 211 }
 212
 213 /**
 214  * tipc_nameseq_insert_publ
 215  */
 216 static struct publication *tipc_nameseq_insert_publ(struct net *net,
 217                                                     struct name_seq *nseq,
 218                                                     u32 type, u32 lower,
 219                                                     u32 upper, u32 scope,
 220                                                     u32 node, u32 port, u32 key)
 221 {
 222         struct tipc_subscription *s;
 223         struct tipc_subscription *st;
 224         struct publication *publ;
 225         struct sub_seq *sseq;
 226         struct name_info *info;
 227         int created_subseq = 0;
 228
 229         sseq = nameseq_find_subseq(nseq, lower);
 230         if (sseq) {
 231
 232                 /* Lower end overlaps existing entry => need an exact match */
 233                 if ((sseq->lower != lower) || (sseq->upper != upper)) {
 234                         return NULL;
 235                 }
 236
 237                 info = sseq->info;
 238
 239                 /* Check if an identical publication already exists */
 240                 list_for_each_entry(publ, &info->all_publ, all_publ) {
 241                         if (publ->port == port && publ->key == key &&
 242                             (!publ->node || publ->node == node))
 243                                 return NULL;
 244                 }
 245         } else {
 246                 u32 inspos;
 247                 struct sub_seq *freesseq;
 248
 249                 /* Find where lower end should be inserted */
 250                 inspos = nameseq_locate_subseq(nseq, lower);
 251
 252                 /* Fail if upper end overlaps into an existing entry */
 253                 if ((inspos < nseq->first_free) &&
 254                     (upper >= nseq->sseqs[inspos].lower)) {
 255                         return NULL;
 256                 }
 257
 258                 /* Ensure there is space for new sub-sequence */
 259                 if (nseq->first_free == nseq->alloc) {
 260                         struct sub_seq *sseqs = tipc_subseq_alloc(nseq->alloc * 2);
 261
 262                         if (!sseqs) {
 263                                 pr_warn("Cannot publish {%u,%u,%u}, no memory\n",
 264                                         type, lower, upper);
 265                                 return NULL;
 266                         }
 267                         memcpy(sseqs, nseq->sseqs,
 268                                nseq->alloc * sizeof(struct sub_seq));
 269                         kfree(nseq->sseqs);
 270                         nseq->sseqs = sseqs;
 271                         nseq->alloc *= 2;
 272                 }
 273
 274                 info = kzalloc(sizeof(*info), GFP_ATOMIC);
 275                 if (!info) {
 276                         pr_warn("Cannot publish {%u,%u,%u}, no memory\n",
 277                                 type, lower, upper);
 278                         return NULL;
 279                 }
 280
 281                 INIT_LIST_HEAD(&info->local_publ);
 282                 INIT_LIST_HEAD(&info->all_publ);
 283
 284                 /* Insert new sub-sequence */
 285                 sseq = &nseq->sseqs[inspos];
 286                 freesseq = &nseq->sseqs[nseq->first_free];
 287                 memmove(sseq + 1, sseq, (freesseq - sseq) * sizeof(*sseq));
 288                 memset(sseq, 0, sizeof(*sseq));
 289                 nseq->first_free++;
 290                 sseq->lower = lower;
 291                 sseq->upper = upper;
 292                 sseq->info = info;
 293                 created_subseq = 1;
 294         }
 295
 296         /* Insert a publication */
 297         publ = publ_create(type, lower, upper, scope, node, port, key);
 298         if (!publ)
 299                 return NULL;
 300
 301         list_add(&publ->all_publ, &info->all_publ);
 302
 303         if (in_own_node(net, node))
 304                 list_add(&publ->local_publ, &info->local_publ);
 305
 306         /* Any subscriptions waiting for notification?  */
 307         list_for_each_entry_safe(s, st, &nseq->subscriptions, nameseq_list) {
 308                 tipc_sub_report_overlap(s, publ->lower, publ->upper,
 309                                         TIPC_PUBLISHED, publ->port,
 310                                         publ->node, publ->scope,
 311                                         created_subseq);
 312         }
 313         return publ;
 314 }
 315
 316 /**
 317  * tipc_nameseq_remove_publ
 318  *
 319  * NOTE: There may be cases where TIPC is asked to remove a publication
 320  * that is not in the name table.  For example, if another node issues a
 321  * publication for a name sequence that overlaps an existing name sequence
 322  * the publication will not be recorded, which means the publication won't
 323  * be found when the name sequence is later withdrawn by that node.
 324  * A failed withdraw request simply returns a failure indication and lets the
 325  * caller issue any error or warning messages associated with such a problem.
 326  */
 327 static struct publication *tipc_nameseq_remove_publ(struct net *net,
 328                                                     struct name_seq *nseq,
 329                                                     u32 inst, u32 node,
 330                                                     u32 port, u32 key)
 331 {
 332         struct publication *publ;
 333         struct sub_seq *sseq = nameseq_find_subseq(nseq, inst);
 334         struct name_info *info;
 335         struct sub_seq *free;
 336         struct tipc_subscription *s, *st;
 337         int removed_subseq = 0;
 338
 339         if (!sseq)
 340                 return NULL;
 341
 342         info = sseq->info;
 343
 344         /* Locate publication, if it exists */
 345         list_for_each_entry(publ, &info->all_publ, all_publ) {
 346                 if (publ->key == key && publ->port == port &&
 347                     (!publ->node || publ->node == node))
 348                         goto found;
 349         }
 350         return NULL;
 351
 352 found:
 353         list_del(&publ->all_publ);
 354         if (in_own_node(net, node))
 355                 list_del(&publ->local_publ);
 356
 357         /* Contract subseq list if no more publications for that subseq */
 358         if (list_empty(&info->all_publ)) {
 359                 kfree(info);
 360                 free = &nseq->sseqs[nseq->first_free--];
 361                 memmove(sseq, sseq + 1, (free - (sseq + 1)) * sizeof(*sseq));
 362                 removed_subseq = 1;
 363         }
 364
 365         /* Notify any waiting subscriptions */
 366         list_for_each_entry_safe(s, st, &nseq->subscriptions, nameseq_list) {
 367                 tipc_sub_report_overlap(s, publ->lower, publ->upper,
 368                                         TIPC_WITHDRAWN, publ->port,
 369                                         publ->node, publ->scope,
 370                                         removed_subseq);
 371         }
 372
 373         return publ;
 374 }
 375
 376 /**
 377  * tipc_nameseq_subscribe - attach a subscription, and optionally
 378  * issue the prescribed number of events if there is any sub-
 379  * sequence overlapping with the requested sequence
 380  */
 381 static void tipc_nameseq_subscribe(struct name_seq *nseq,
 382                                    struct tipc_subscription *sub)
 383 {
 384         struct sub_seq *sseq = nseq->sseqs;
 385         struct tipc_name_seq ns;
 386         struct tipc_subscr *s = &sub->evt.s;
 387         bool no_status;
 388
 389         ns.type = tipc_sub_read(s, seq.type);
 390         ns.lower = tipc_sub_read(s, seq.lower);
 391         ns.upper = tipc_sub_read(s, seq.upper);
 392         no_status = tipc_sub_read(s, filter) & TIPC_SUB_NO_STATUS;
 393
 394         tipc_sub_get(sub);
 395         list_add(&sub->nameseq_list, &nseq->subscriptions);
 396
 397         if (no_status || !sseq)
 398                 return;
 399
 400         while (sseq != &nseq->sseqs[nseq->first_free]) {
 401                 if (tipc_sub_check_overlap(&ns, sseq->lower, sseq->upper)) {
 402                         struct publication *crs;
 403                         struct name_info *info = sseq->info;
 404                         int must_report = 1;
 405
 406                         list_for_each_entry(crs, &info->all_publ, all_publ) {
 407                                 tipc_sub_report_overlap(sub, sseq->lower,
 408                                                         sseq->upper,
 409                                                         TIPC_PUBLISHED,
 410                                                         crs->port,
 411                                                         crs->node,
 412                                                         crs->scope,
 413                                                         must_report);
 414                                 must_report = 0;
 415                         }
 416                 }
 417                 sseq++;
 418         }
 419 }
 420
 421 static struct name_seq *nametbl_find_seq(struct net *net, u32 type)
 422 {
 423         struct tipc_net *tn = net_generic(net, tipc_net_id);
 424         struct hlist_head *seq_head;
 425         struct name_seq *ns;
 426
 427         seq_head = &tn->nametbl->seq_hlist[hash(type)];
 428         hlist_for_each_entry_rcu(ns, seq_head, ns_list) {
 429                 if (ns->type == type)
 430                         return ns;
 431         }
 432
 433         return NULL;
 434 };
 435
 436 struct publication *tipc_nametbl_insert_publ(struct net *net, u32 type,
 437                                              u32 lower, u32 upper, u32 scope,
 438                                              u32 node, u32 port, u32 key)
 439 {
 440         struct tipc_net *tn = net_generic(net, tipc_net_id);
 441         struct publication *publ;
 442         struct name_seq *seq = nametbl_find_seq(net, type);
 443         int index = hash(type);
 444
 445         if (scope > TIPC_NODE_SCOPE || lower > upper) {
 446                 pr_debug("Failed to publish illegal {%u,%u,%u} with scope %u\n",
 447                          type, lower, upper, scope);
 448                 return NULL;
 449         }
 450
 451         if (!seq)
 452                 seq = tipc_nameseq_create(type, &tn->nametbl->seq_hlist[index]);
 453         if (!seq)
 454                 return NULL;
 455
 456         spin_lock_bh(&seq->lock);
 457         publ = tipc_nameseq_insert_publ(net, seq, type, lower, upper,
 458                                         scope, node, port, key);
 459         spin_unlock_bh(&seq->lock);
 460         return publ;
 461 }
 462
 463 struct publication *tipc_nametbl_remove_publ(struct net *net, u32 type,
 464                                              u32 lower, u32 node, u32 port,
 465                                              u32 key)
 466 {
 467         struct publication *publ;
 468         struct name_seq *seq = nametbl_find_seq(net, type);
 469
 470         if (!seq)
 471                 return NULL;
 472
 473         spin_lock_bh(&seq->lock);
 474         publ = tipc_nameseq_remove_publ(net, seq, lower, node, port, key);
 475         if (!seq->first_free && list_empty(&seq->subscriptions)) {
 476                 hlist_del_init_rcu(&seq->ns_list);
 477                 kfree(seq->sseqs);
 478                 spin_unlock_bh(&seq->lock);
 479                 kfree_rcu(seq, rcu);
 480                 return publ;
 481         }
 482         spin_unlock_bh(&seq->lock);
 483         return publ;
 484 }
 485
 486 /**
 487  * tipc_nametbl_translate - perform name translation
 488  *
 489  * On entry, 'destnode' is the search domain used during translation.
 490  *
 491  * On exit:
 492  * - if name translation is deferred to another node/cluster/zone,
 493  *   leaves 'destnode' unchanged (will be non-zero) and returns 0
 494  * - if name translation is attempted and succeeds, sets 'destnode'
 495  *   to publishing node and returns port reference (will be non-zero)
 496  * - if name translation is attempted and fails, sets 'destnode' to 0
 497  *   and returns 0
 498  */
 499 u32 tipc_nametbl_translate(struct net *net, u32 type, u32 instance,
 500                            u32 *destnode)
 501 {
 502         struct tipc_net *tn = tipc_net(net);
 503         bool legacy = tn->legacy_addr_format;
 504         u32 self = tipc_own_addr(net);
 505         struct sub_seq *sseq;
 506         struct name_info *info;
 507         struct publication *publ;
 508         struct name_seq *seq;
 509         u32 port = 0;
 510         u32 node = 0;
 511
 512         if (!tipc_in_scope(legacy, *destnode, self))
 513                 return 0;
 514
 515         rcu_read_lock();
 516         seq = nametbl_find_seq(net, type);
 517         if (unlikely(!seq))
 518                 goto not_found;
 519         spin_lock_bh(&seq->lock);
 520         sseq = nameseq_find_subseq(seq, instance);
 521         if (unlikely(!sseq))
 522                 goto no_match;
 523         info = sseq->info;
 524
 525         /* Closest-First Algorithm */
 526         if (legacy && !*destnode) {
 527                 if (!list_empty(&info->local_publ)) {
 528                         publ = list_first_entry(&info->local_publ,
 529                                                 struct publication,
 530                                                 local_publ);
 531                         list_move_tail(&publ->local_publ,
 532                                        &info->local_publ);
 533                 } else {
 534                         publ = list_first_entry(&info->all_publ,
 535                                                 struct publication,
 536                                                 all_publ);
 537                         list_move_tail(&publ->all_publ,
 538                                        &info->all_publ);
 539                 }
 540         }
 541
 542         /* Round-Robin Algorithm */
 543         else if (*destnode == tipc_own_addr(net)) {
 544                 if (list_empty(&info->local_publ))
 545                         goto no_match;
 546                 publ = list_first_entry(&info->local_publ, struct publication,
 547                                         local_publ);
 548                 list_move_tail(&publ->local_publ, &info->local_publ);
 549         } else {
 550                 publ = list_first_entry(&info->all_publ, struct publication,
 551                                         all_publ);
 552                 list_move_tail(&publ->all_publ, &info->all_publ);
 553         }
 554
 555         port = publ->port;
 556         node = publ->node;
 557 no_match:
 558         spin_unlock_bh(&seq->lock);
 559 not_found:
 560         rcu_read_unlock();
 561         *destnode = node;
 562         return port;
 563 }
 564
 565 bool tipc_nametbl_lookup(struct net *net, u32 type, u32 instance, u32 scope,
 566                          struct list_head *dsts, int *dstcnt, u32 exclude,
 567                          bool all)
 568 {
 569         u32 self = tipc_own_addr(net);
 570         struct publication *publ;
 571         struct name_info *info;
 572         struct name_seq *seq;
 573         struct sub_seq *sseq;
 574
 575         *dstcnt = 0;
 576         rcu_read_lock();
 577         seq = nametbl_find_seq(net, type);
 578         if (unlikely(!seq))
 579                 goto exit;
 580         spin_lock_bh(&seq->lock);
 581         sseq = nameseq_find_subseq(seq, instance);
 582         if (likely(sseq)) {
 583                 info = sseq->info;
 584                 list_for_each_entry(publ, &info->all_publ, all_publ) {
 585                         if (publ->scope != scope)
 586                                 continue;
 587                         if (publ->port == exclude && publ->node == self)
 588                                 continue;
 589                         tipc_dest_push(dsts, publ->node, publ->port);
 590                         (*dstcnt)++;
 591                         if (all)
 592                                 continue;
 593                         list_move_tail(&publ->all_publ, &info->all_publ);
 594                         break;
 595                 }
 596         }
 597         spin_unlock_bh(&seq->lock);
 598 exit:
 599         rcu_read_unlock();
 600         return !list_empty(dsts);
 601 }
 602
 603 void tipc_nametbl_mc_lookup(struct net *net, u32 type, u32 lower, u32 upper,
 604                             u32 scope, bool exact, struct list_head *dports)
 605 {
 606         struct sub_seq *sseq_stop;
 607         struct name_info *info;
 608         struct publication *p;
 609         struct name_seq *seq;
 610         struct sub_seq *sseq;
 611
 612         rcu_read_lock();
 613         seq = nametbl_find_seq(net, type);
 614         if (!seq)
 615                 goto exit;
 616
 617         spin_lock_bh(&seq->lock);
 618         sseq = seq->sseqs + nameseq_locate_subseq(seq, lower);
 619         sseq_stop = seq->sseqs + seq->first_free;
 620         for (; sseq != sseq_stop; sseq++) {
 621                 if (sseq->lower > upper)
 622                         break;
 623                 info = sseq->info;
 624                 list_for_each_entry(p, &info->local_publ, local_publ) {
 625                         if (p->scope == scope || (!exact && p->scope < scope))
 626                                 tipc_dest_push(dports, 0, p->port);
 627                 }
 628         }
 629         spin_unlock_bh(&seq->lock);
 630 exit:
 631         rcu_read_unlock();
 632 }
 633
 634 /* tipc_nametbl_lookup_dst_nodes - find broadcast destination nodes
 635  * - Creates list of nodes that overlap the given multicast address
 636  * - Determines if any node local ports overlap
 637  */
 638 void tipc_nametbl_lookup_dst_nodes(struct net *net, u32 type, u32 lower,
 639                                    u32 upper, struct tipc_nlist *nodes)
 640 {
 641         struct sub_seq *sseq, *stop;
 642         struct publication *publ;
 643         struct name_info *info;
 644         struct name_seq *seq;
 645
 646         rcu_read_lock();
 647         seq = nametbl_find_seq(net, type);
 648         if (!seq)
 649                 goto exit;
 650
 651         spin_lock_bh(&seq->lock);
 652         sseq = seq->sseqs + nameseq_locate_subseq(seq, lower);
 653         stop = seq->sseqs + seq->first_free;
 654         for (; sseq != stop && sseq->lower <= upper; sseq++) {
 655                 info = sseq->info;
 656                 list_for_each_entry(publ, &info->all_publ, all_publ) {
 657                         tipc_nlist_add(nodes, publ->node);
 658                 }
 659         }
 660         spin_unlock_bh(&seq->lock);
 661 exit:
 662         rcu_read_unlock();
 663 }
 664
 665 /* tipc_nametbl_build_group - build list of communication group members
 666  */
 667 void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
 668                               u32 type, u32 scope)
 669 {
 670         struct sub_seq *sseq, *stop;
 671         struct name_info *info;
 672         struct publication *p;
 673         struct name_seq *seq;
 674
 675         rcu_read_lock();
 676         seq = nametbl_find_seq(net, type);
 677         if (!seq)
 678                 goto exit;
 679
 680         spin_lock_bh(&seq->lock);
 681         sseq = seq->sseqs;
 682         stop = seq->sseqs + seq->first_free;
 683         for (; sseq != stop; sseq++) {
 684                 info = sseq->info;
 685                 list_for_each_entry(p, &info->all_publ, all_publ) {
 686                         if (p->scope != scope)
 687                                 continue;
 688                         tipc_group_add_member(grp, p->node, p->port, p->lower);
 689                 }
 690         }
 691         spin_unlock_bh(&seq->lock);
 692 exit:
 693         rcu_read_unlock();
 694 }
 695
 696 /*
 697  * tipc_nametbl_publish - add name publication to network name tables
 698  */
 699 struct publication *tipc_nametbl_publish(struct net *net, u32 type, u32 lower,
 700                                          u32 upper, u32 scope, u32 port_ref,
 701                                          u32 key)
 702 {
 703         struct publication *publ;
 704         struct sk_buff *buf = NULL;
 705         struct tipc_net *tn = net_generic(net, tipc_net_id);
 706
 707         spin_lock_bh(&tn->nametbl_lock);
 708         if (tn->nametbl->local_publ_count >= TIPC_MAX_PUBLICATIONS) {
 709                 pr_warn("Publication failed, local publication limit reached (%u)\n",
 710                         TIPC_MAX_PUBLICATIONS);
 711                 spin_unlock_bh(&tn->nametbl_lock);
 712                 return NULL;
 713         }
 714
 715         publ = tipc_nametbl_insert_publ(net, type, lower, upper, scope,
 716                                         tipc_own_addr(net), port_ref, key);
 717         if (likely(publ)) {
 718                 tn->nametbl->local_publ_count++;
 719                 buf = tipc_named_publish(net, publ);
 720                 /* Any pending external events? */
 721                 tipc_named_process_backlog(net);
 722         }
 723         spin_unlock_bh(&tn->nametbl_lock);
 724
 725         if (buf)
 726                 tipc_node_broadcast(net, buf);
 727         return publ;
 728 }
 729
 730 /**
 731  * tipc_nametbl_withdraw - withdraw name publication from network name tables
 732  */
 733 int tipc_nametbl_withdraw(struct net *net, u32 type, u32 lower, u32 port,
 734                           u32 key)
 735 {
 736         struct publication *publ;
 737         struct sk_buff *skb = NULL;
 738         struct tipc_net *tn = net_generic(net, tipc_net_id);
 739
 740         spin_lock_bh(&tn->nametbl_lock);
 741         publ = tipc_nametbl_remove_publ(net, type, lower, tipc_own_addr(net),
 742                                         port, key);
 743         if (likely(publ)) {
 744                 tn->nametbl->local_publ_count--;
 745                 skb = tipc_named_withdraw(net, publ);
 746                 /* Any pending external events? */
 747                 tipc_named_process_backlog(net);
 748                 list_del_init(&publ->binding_sock);
 749                 kfree_rcu(publ, rcu);
 750         } else {
 751                 pr_err("Unable to remove local publication\n"
 752                        "(type=%u, lower=%u, port=%u, key=%u)\n",
 753                        type, lower, port, key);
 754         }
 755         spin_unlock_bh(&tn->nametbl_lock);
 756
 757         if (skb) {
 758                 tipc_node_broadcast(net, skb);
 759                 return 1;
 760         }
 761         return 0;
 762 }
 763
 764 /**
 765  * tipc_nametbl_subscribe - add a subscription object to the name table
 766  */
 767 void tipc_nametbl_subscribe(struct tipc_subscription *sub)
 768 {
 769         struct tipc_net *tn = tipc_net(sub->net);
 770         struct tipc_subscr *s = &sub->evt.s;
 771         u32 type = tipc_sub_read(s, seq.type);
 772         int index = hash(type);
 773         struct name_seq *seq;
 774         struct tipc_name_seq ns;
 775
 776         spin_lock_bh(&tn->nametbl_lock);
 777         seq = nametbl_find_seq(sub->net, type);
 778         if (!seq)
 779                 seq = tipc_nameseq_create(type, &tn->nametbl->seq_hlist[index]);
 780         if (seq) {
 781                 spin_lock_bh(&seq->lock);
 782                 tipc_nameseq_subscribe(seq, sub);
 783                 spin_unlock_bh(&seq->lock);
 784         } else {
 785                 ns.type = tipc_sub_read(s, seq.type);
 786                 ns.lower = tipc_sub_read(s, seq.lower);
 787                 ns.upper = tipc_sub_read(s, seq.upper);
 788                 pr_warn("Failed to create subscription for {%u,%u,%u}\n",
 789                         ns.type, ns.lower, ns.upper);
 790         }
 791         spin_unlock_bh(&tn->nametbl_lock);
 792 }
 793
 794 /**
 795  * tipc_nametbl_unsubscribe - remove a subscription object from name table
 796  */
 797 void tipc_nametbl_unsubscribe(struct tipc_subscription *sub)
 798 {
 799         struct tipc_subscr *s = &sub->evt.s;
 800         struct tipc_net *tn = tipc_net(sub->net);
 801         struct name_seq *seq;
 802         u32 type = tipc_sub_read(s, seq.type);
 803
 804         spin_lock_bh(&tn->nametbl_lock);
 805         seq = nametbl_find_seq(sub->net, type);
 806         if (seq != NULL) {
 807                 spin_lock_bh(&seq->lock);
 808                 list_del_init(&sub->nameseq_list);
 809                 tipc_sub_put(sub);
 810                 if (!seq->first_free && list_empty(&seq->subscriptions)) {
 811                         hlist_del_init_rcu(&seq->ns_list);
 812                         kfree(seq->sseqs);
 813                         spin_unlock_bh(&seq->lock);
 814                         kfree_rcu(seq, rcu);
 815                 } else {
 816                         spin_unlock_bh(&seq->lock);
 817                 }
 818         }
 819         spin_unlock_bh(&tn->nametbl_lock);
 820 }
 821
 822 int tipc_nametbl_init(struct net *net)
 823 {
 824         struct tipc_net *tn = net_generic(net, tipc_net_id);
 825         struct name_table *tipc_nametbl;
 826         int i;
 827
 828         tipc_nametbl = kzalloc(sizeof(*tipc_nametbl), GFP_ATOMIC);
 829         if (!tipc_nametbl)
 830                 return -ENOMEM;
 831
 832         for (i = 0; i < TIPC_NAMETBL_SIZE; i++)
 833                 INIT_HLIST_HEAD(&tipc_nametbl->seq_hlist[i]);
 834
 835         INIT_LIST_HEAD(&tipc_nametbl->node_scope);
 836         INIT_LIST_HEAD(&tipc_nametbl->cluster_scope);
 837         tn->nametbl = tipc_nametbl;
 838         spin_lock_init(&tn->nametbl_lock);
 839         return 0;
 840 }
 841
 842 /**
 843  * tipc_purge_publications - remove all publications for a given type
 844  *
 845  * tipc_nametbl_lock must be held when calling this function
 846  */
 847 static void tipc_purge_publications(struct net *net, struct name_seq *seq)
 848 {
 849         struct publication *publ, *safe;
 850         struct sub_seq *sseq;
 851         struct name_info *info;
 852
 853         spin_lock_bh(&seq->lock);
 854         sseq = seq->sseqs;
 855         info = sseq->info;
 856         list_for_each_entry_safe(publ, safe, &info->all_publ, all_publ) {
 857                 tipc_nameseq_remove_publ(net, seq, publ->lower, publ->node,
 858                                          publ->port, publ->key);
 859                 kfree_rcu(publ, rcu);
 860         }
 861         hlist_del_init_rcu(&seq->ns_list);
 862         kfree(seq->sseqs);
 863         spin_unlock_bh(&seq->lock);
 864
 865         kfree_rcu(seq, rcu);
 866 }
 867
 868 void tipc_nametbl_stop(struct net *net)
 869 {
 870         u32 i;
 871         struct name_seq *seq;
 872         struct hlist_head *seq_head;
 873         struct tipc_net *tn = net_generic(net, tipc_net_id);
 874         struct name_table *tipc_nametbl = tn->nametbl;
 875
 876         /* Verify name table is empty and purge any lingering
 877          * publications, then release the name table
 878          */
 879         spin_lock_bh(&tn->nametbl_lock);
 880         for (i = 0; i < TIPC_NAMETBL_SIZE; i++) {
 881                 if (hlist_empty(&tipc_nametbl->seq_hlist[i]))
 882                         continue;
 883                 seq_head = &tipc_nametbl->seq_hlist[i];
 884                 hlist_for_each_entry_rcu(seq, seq_head, ns_list) {
 885                         tipc_purge_publications(net, seq);
 886                 }
 887         }
 888         spin_unlock_bh(&tn->nametbl_lock);
 889
 890         synchronize_net();
 891         kfree(tipc_nametbl);
 892
 893 }
 894
 895 static int __tipc_nl_add_nametable_publ(struct tipc_nl_msg *msg,
 896                                         struct name_seq *seq,
 897                                         struct sub_seq *sseq, u32 *last_publ)
 898 {
 899         void *hdr;
 900         struct nlattr *attrs;
 901         struct nlattr *publ;
 902         struct publication *p;
 903
 904         if (*last_publ) {
 905                 list_for_each_entry(p, &sseq->info->all_publ, all_publ)
 906                         if (p->key == *last_publ)
 907                                 break;
 908                 if (p->key != *last_publ)
 909                         return -EPIPE;
 910         } else {
 911                 p = list_first_entry(&sseq->info->all_publ, struct publication,
 912                                      all_publ);
 913         }
 914
 915         list_for_each_entry_from(p, &sseq->info->all_publ, all_publ) {
 916                 *last_publ = p->key;
 917
 918                 hdr = genlmsg_put(msg->skb, msg->portid, msg->seq,
 919                                   &tipc_genl_family, NLM_F_MULTI,
 920                                   TIPC_NL_NAME_TABLE_GET);
 921                 if (!hdr)
 922                         return -EMSGSIZE;
 923
 924                 attrs = nla_nest_start(msg->skb, TIPC_NLA_NAME_TABLE);
 925                 if (!attrs)
 926                         goto msg_full;
 927
 928                 publ = nla_nest_start(msg->skb, TIPC_NLA_NAME_TABLE_PUBL);
 929                 if (!publ)
 930                         goto attr_msg_full;
 931
 932                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_TYPE, seq->type))
 933                         goto publ_msg_full;
 934                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_LOWER, sseq->lower))
 935                         goto publ_msg_full;
 936                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_UPPER, sseq->upper))
 937                         goto publ_msg_full;
 938                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_SCOPE, p->scope))
 939                         goto publ_msg_full;
 940                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_NODE, p->node))
 941                         goto publ_msg_full;
 942                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_REF, p->port))
 943                         goto publ_msg_full;
 944                 if (nla_put_u32(msg->skb, TIPC_NLA_PUBL_KEY, p->key))
 945                         goto publ_msg_full;
 946
 947                 nla_nest_end(msg->skb, publ);
 948                 nla_nest_end(msg->skb, attrs);
 949                 genlmsg_end(msg->skb, hdr);
 950         }
 951         *last_publ = 0;
 952
 953         return 0;
 954
 955 publ_msg_full:
 956         nla_nest_cancel(msg->skb, publ);
 957 attr_msg_full:
 958         nla_nest_cancel(msg->skb, attrs);
 959 msg_full:
 960         genlmsg_cancel(msg->skb, hdr);
 961
 962         return -EMSGSIZE;
 963 }
 964
 965 static int __tipc_nl_subseq_list(struct tipc_nl_msg *msg, struct name_seq *seq,
 966                                  u32 *last_lower, u32 *last_publ)
 967 {
 968         struct sub_seq *sseq;
 969         struct sub_seq *sseq_start;
 970         int err;
 971
 972         if (*last_lower) {
 973                 sseq_start = nameseq_find_subseq(seq, *last_lower);
 974                 if (!sseq_start)
 975                         return -EPIPE;
 976         } else {
 977                 sseq_start = seq->sseqs;
 978         }
 979
 980         for (sseq = sseq_start; sseq != &seq->sseqs[seq->first_free]; sseq++) {
 981                 err = __tipc_nl_add_nametable_publ(msg, seq, sseq, last_publ);
 982                 if (err) {
 983                         *last_lower = sseq->lower;
 984                         return err;
 985                 }
 986         }
 987         *last_lower = 0;
 988
 989         return 0;
 990 }
 991
 992 static int tipc_nl_seq_list(struct net *net, struct tipc_nl_msg *msg,
 993                             u32 *last_type, u32 *last_lower, u32 *last_publ)
 994 {
 995         struct tipc_net *tn = net_generic(net, tipc_net_id);
 996         struct hlist_head *seq_head;
 997         struct name_seq *seq = NULL;
 998         int err;
 999         int i;
1000
1001         if (*last_type)
1002                 i = hash(*last_type);
1003         else
1004                 i = 0;
1005
1006         for (; i < TIPC_NAMETBL_SIZE; i++) {
1007                 seq_head = &tn->nametbl->seq_hlist[i];
1008
1009                 if (*last_type) {
1010                         seq = nametbl_find_seq(net, *last_type);
1011                         if (!seq)
1012                                 return -EPIPE;
1013                 } else {
1014                         hlist_for_each_entry_rcu(seq, seq_head, ns_list)
1015                                 break;
1016                         if (!seq)
1017                                 continue;
1018                 }
1019
1020                 hlist_for_each_entry_from_rcu(seq, ns_list) {
1021                         spin_lock_bh(&seq->lock);
1022                         err = __tipc_nl_subseq_list(msg, seq, last_lower,
1023                                                     last_publ);
1024
1025                         if (err) {
1026                                 *last_type = seq->type;
1027                                 spin_unlock_bh(&seq->lock);
1028                                 return err;
1029                         }
1030                         spin_unlock_bh(&seq->lock);
1031                 }
1032                 *last_type = 0;
1033         }
1034         return 0;
1035 }
1036
1037 int tipc_nl_name_table_dump(struct sk_buff *skb, struct netlink_callback *cb)
1038 {
1039         int err;
1040         int done = cb->args[3];
1041         u32 last_type = cb->args[0];
1042         u32 last_lower = cb->args[1];
1043         u32 last_publ = cb->args[2];
1044         struct net *net = sock_net(skb->sk);
1045         struct tipc_nl_msg msg;
1046
1047         if (done)
1048                 return 0;
1049
1050         msg.skb = skb;
1051         msg.portid = NETLINK_CB(cb->skb).portid;
1052         msg.seq = cb->nlh->nlmsg_seq;
1053
1054         rcu_read_lock();
1055         err = tipc_nl_seq_list(net, &msg, &last_type, &last_lower, &last_publ);
1056         if (!err) {
1057                 done = 1;
1058         } else if (err != -EMSGSIZE) {
1059                 /* We never set seq or call nl_dump_check_consistent() this
1060                  * means that setting prev_seq here will cause the consistence
1061                  * check to fail in the netlink callback handler. Resulting in
1062                  * the NLMSG_DONE message having the NLM_F_DUMP_INTR flag set if
1063                  * we got an error.
1064                  */
1065                 cb->prev_seq = 1;
1066         }
1067         rcu_read_unlock();
1068
1069         cb->args[0] = last_type;
1070         cb->args[1] = last_lower;
1071         cb->args[2] = last_publ;
1072         cb->args[3] = done;
1073
1074         return skb->len;
1075 }
1076
1077 struct tipc_dest *tipc_dest_find(struct list_head *l, u32 node, u32 port)
1078 {
1079         u64 value = (u64)node << 32 | port;
1080         struct tipc_dest *dst;
1081
1082         list_for_each_entry(dst, l, list) {
1083                 if (dst->value != value)
1084                         continue;
1085                 return dst;
1086         }
1087         return NULL;
1088 }
1089
1090 bool tipc_dest_push(struct list_head *l, u32 node, u32 port)
1091 {
1092         u64 value = (u64)node << 32 | port;
1093         struct tipc_dest *dst;
1094
1095         if (tipc_dest_find(l, node, port))
1096                 return false;
1097
1098         dst = kmalloc(sizeof(*dst), GFP_ATOMIC);
1099         if (unlikely(!dst))
1100                 return false;
1101         dst->value = value;
1102         list_add(&dst->list, l);
1103         return true;
1104 }
1105
1106 bool tipc_dest_pop(struct list_head *l, u32 *node, u32 *port)
1107 {
1108         struct tipc_dest *dst;
1109
1110         if (list_empty(l))
1111                 return false;
1112         dst = list_first_entry(l, typeof(*dst), list);
1113         if (port)
1114                 *port = dst->port;
1115         if (node)
1116                 *node = dst->node;
1117         list_del(&dst->list);
1118         kfree(dst);
1119         return true;
1120 }
1121
1122 bool tipc_dest_del(struct list_head *l, u32 node, u32 port)
1123 {
1124         struct tipc_dest *dst;
1125
1126         dst = tipc_dest_find(l, node, port);
1127         if (!dst)
1128                 return false;
1129         list_del(&dst->list);
1130         kfree(dst);
1131         return true;
1132 }
1133
1134 void tipc_dest_list_purge(struct list_head *l)
1135 {
1136         struct tipc_dest *dst, *tmp;
1137
1138         list_for_each_entry_safe(dst, tmp, l, list) {
1139                 list_del(&dst->list);
1140                 kfree(dst);
1141         }
1142 }
1143
1144 int tipc_dest_list_len(struct list_head *l)
1145 {
1146         struct tipc_dest *dst;
1147         int i = 0;
1148
1149         list_for_each_entry(dst, l, list) {
1150                 i++;
1151         }
1152         return i;
1153 }