net/tipc/name_table.c

   1 /*
   2  * net/tipc/name_table.c: TIPC name table code
   3  *
   4  * Copyright (c) 2000-2006, Ericsson AB
   5  * Copyright (c) 2004-2008, 2010-2011, 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 "core.h"
  38 #include "config.h"
  39 #include "name_table.h"
  40 #include "name_distr.h"
  41 #include "subscr.h"
  42 #include "port.h"
  43
  44 #define TIPC_NAMETBL_SIZE 1024          /* must be a power of 2 */
  45
  46 /**
  47  * struct name_info - name sequence publication info
  48  * @node_list: circular list of publications made by own node
  49  * @cluster_list: circular list of publications made by own cluster
  50  * @zone_list: circular list of publications made by own zone
  51  * @node_list_size: number of entries in "node_list"
  52  * @cluster_list_size: number of entries in "cluster_list"
  53  * @zone_list_size: number of entries in "zone_list"
  54  *
  55  * Note: The zone list always contains at least one entry, since all
  56  *       publications of the associated name sequence belong to it.
  57  *       (The cluster and node lists may be empty.)
  58  */
  59 struct name_info {
  60         struct list_head node_list;
  61         struct list_head cluster_list;
  62         struct list_head zone_list;
  63         u32 node_list_size;
  64         u32 cluster_list_size;
  65         u32 zone_list_size;
  66 };
  67
  68 /**
  69  * struct sub_seq - container for all published instances of a name sequence
  70  * @lower: name sequence lower bound
  71  * @upper: name sequence upper bound
  72  * @info: pointer to name sequence publication info
  73  */
  74 struct sub_seq {
  75         u32 lower;
  76         u32 upper;
  77         struct name_info *info;
  78 };
  79
  80 /**
  81  * struct name_seq - container for all published instances of a name type
  82  * @type: 32 bit 'type' value for name sequence
  83  * @sseq: pointer to dynamically-sized array of sub-sequences of this 'type';
  84  *        sub-sequences are sorted in ascending order
  85  * @alloc: number of sub-sequences currently in array
  86  * @first_free: array index of first unused sub-sequence entry
  87  * @ns_list: links to adjacent name sequences in hash chain
  88  * @subscriptions: list of subscriptions for this 'type'
  89  * @lock: spinlock controlling access to publication lists of all sub-sequences
  90  */
  91 struct name_seq {
  92         u32 type;
  93         struct sub_seq *sseqs;
  94         u32 alloc;
  95         u32 first_free;
  96         struct hlist_node ns_list;
  97         struct list_head subscriptions;
  98         spinlock_t lock;
  99 };
 100
 101 /**
 102  * struct name_table - table containing all existing port name publications
 103  * @types: pointer to fixed-sized array of name sequence lists,
 104  *         accessed via hashing on 'type'; name sequence lists are *not* sorted
 105  * @local_publ_count: number of publications issued by this node
 106  */
 107 struct name_table {
 108         struct hlist_head *types;
 109         u32 local_publ_count;
 110 };
 111
 112 static struct name_table table;
 113 DEFINE_RWLOCK(tipc_nametbl_lock);
 114
 115 static int hash(int x)
 116 {
 117         return x & (TIPC_NAMETBL_SIZE - 1);
 118 }
 119
 120 /**
 121  * publ_create - create a publication structure
 122  */
 123 static struct publication *publ_create(u32 type, u32 lower, u32 upper,
 124                                        u32 scope, u32 node, u32 port_ref,
 125                                        u32 key)
 126 {
 127         struct publication *publ = kzalloc(sizeof(*publ), GFP_ATOMIC);
 128         if (publ == NULL) {
 129                 pr_warn("Publication creation failure, no memory\n");
 130                 return NULL;
 131         }
 132
 133         publ->type = type;
 134         publ->lower = lower;
 135         publ->upper = upper;
 136         publ->scope = scope;
 137         publ->node = node;
 138         publ->ref = port_ref;
 139         publ->key = key;
 140         INIT_LIST_HEAD(&publ->local_list);
 141         INIT_LIST_HEAD(&publ->pport_list);
 142         INIT_LIST_HEAD(&publ->subscr.nodesub_list);
 143         return publ;
 144 }
 145
 146 /**
 147  * tipc_subseq_alloc - allocate a specified number of sub-sequence structures
 148  */
 149 static struct sub_seq *tipc_subseq_alloc(u32 cnt)
 150 {
 151         return kcalloc(cnt, sizeof(struct sub_seq), GFP_ATOMIC);
 152 }
 153
 154 /**
 155  * tipc_nameseq_create - create a name sequence structure for the specified 'type'
 156  *
 157  * Allocates a single sub-sequence structure and sets it to all 0's.
 158  */
 159 static struct name_seq *tipc_nameseq_create(u32 type, struct hlist_head *seq_head)
 160 {
 161         struct name_seq *nseq = kzalloc(sizeof(*nseq), GFP_ATOMIC);
 162         struct sub_seq *sseq = tipc_subseq_alloc(1);
 163
 164         if (!nseq || !sseq) {
 165                 pr_warn("Name sequence creation failed, no memory\n");
 166                 kfree(nseq);
 167                 kfree(sseq);
 168                 return NULL;
 169         }
 170
 171         spin_lock_init(&nseq->lock);
 172         nseq->type = type;
 173         nseq->sseqs = sseq;
 174         nseq->alloc = 1;
 175         INIT_HLIST_NODE(&nseq->ns_list);
 176         INIT_LIST_HEAD(&nseq->subscriptions);
 177         hlist_add_head(&nseq->ns_list, seq_head);
 178         return nseq;
 179 }
 180
 181 /*
 182  * nameseq_delete_empty - deletes a name sequence structure if now unused
 183  */
 184 static void nameseq_delete_empty(struct name_seq *seq)
 185 {
 186         if (!seq->first_free && list_empty(&seq->subscriptions)) {
 187                 hlist_del_init(&seq->ns_list);
 188                 kfree(seq->sseqs);
 189                 kfree(seq);
 190         }
 191 }
 192
 193 /**
 194  * nameseq_find_subseq - find sub-sequence (if any) matching a name instance
 195  *
 196  * Very time-critical, so binary searches through sub-sequence array.
 197  */
 198 static struct sub_seq *nameseq_find_subseq(struct name_seq *nseq,
 199                                            u32 instance)
 200 {
 201         struct sub_seq *sseqs = nseq->sseqs;
 202         int low = 0;
 203         int high = nseq->first_free - 1;
 204         int mid;
 205
 206         while (low <= high) {
 207                 mid = (low + high) / 2;
 208                 if (instance < sseqs[mid].lower)
 209                         high = mid - 1;
 210                 else if (instance > sseqs[mid].upper)
 211                         low = mid + 1;
 212                 else
 213                         return &sseqs[mid];
 214         }
 215         return NULL;
 216 }
 217
 218 /**
 219  * nameseq_locate_subseq - determine position of name instance in sub-sequence
 220  *
 221  * Returns index in sub-sequence array of the entry that contains the specified
 222  * instance value; if no entry contains that value, returns the position
 223  * where a new entry for it would be inserted in the array.
 224  *
 225  * Note: Similar to binary search code for locating a sub-sequence.
 226  */
 227 static u32 nameseq_locate_subseq(struct name_seq *nseq, u32 instance)
 228 {
 229         struct sub_seq *sseqs = nseq->sseqs;
 230         int low = 0;
 231         int high = nseq->first_free - 1;
 232         int mid;
 233
 234         while (low <= high) {
 235                 mid = (low + high) / 2;
 236                 if (instance < sseqs[mid].lower)
 237                         high = mid - 1;
 238                 else if (instance > sseqs[mid].upper)
 239                         low = mid + 1;
 240                 else
 241                         return mid;
 242         }
 243         return low;
 244 }
 245
 246 /**
 247  * tipc_nameseq_insert_publ
 248  */
 249 static struct publication *tipc_nameseq_insert_publ(struct name_seq *nseq,
 250                                                     u32 type, u32 lower, u32 upper,
 251                                                     u32 scope, u32 node, u32 port, u32 key)
 252 {
 253         struct tipc_subscription *s;
 254         struct tipc_subscription *st;
 255         struct publication *publ;
 256         struct sub_seq *sseq;
 257         struct name_info *info;
 258         int created_subseq = 0;
 259
 260         sseq = nameseq_find_subseq(nseq, lower);
 261         if (sseq) {
 262
 263                 /* Lower end overlaps existing entry => need an exact match */
 264                 if ((sseq->lower != lower) || (sseq->upper != upper)) {
 265                         pr_warn("Cannot publish {%u,%u,%u}, overlap error\n",
 266                                 type, lower, upper);
 267                         return NULL;
 268                 }
 269
 270                 info = sseq->info;
 271
 272                 /* Check if an identical publication already exists */
 273                 list_for_each_entry(publ, &info->zone_list, zone_list) {
 274                         if ((publ->ref == port) && (publ->key == key) &&
 275                             (!publ->node || (publ->node == node)))
 276                                 return NULL;
 277                 }
 278         } else {
 279                 u32 inspos;
 280                 struct sub_seq *freesseq;
 281
 282                 /* Find where lower end should be inserted */
 283                 inspos = nameseq_locate_subseq(nseq, lower);
 284
 285                 /* Fail if upper end overlaps into an existing entry */
 286                 if ((inspos < nseq->first_free) &&
 287                     (upper >= nseq->sseqs[inspos].lower)) {
 288                         pr_warn("Cannot publish {%u,%u,%u}, overlap error\n",
 289                                 type, lower, upper);
 290                         return NULL;
 291                 }
 292
 293                 /* Ensure there is space for new sub-sequence */
 294                 if (nseq->first_free == nseq->alloc) {
 295                         struct sub_seq *sseqs = tipc_subseq_alloc(nseq->alloc * 2);
 296
 297                         if (!sseqs) {
 298                                 pr_warn("Cannot publish {%u,%u,%u}, no memory\n",
 299                                         type, lower, upper);
 300                                 return NULL;
 301                         }
 302                         memcpy(sseqs, nseq->sseqs,
 303                                nseq->alloc * sizeof(struct sub_seq));
 304                         kfree(nseq->sseqs);
 305                         nseq->sseqs = sseqs;
 306                         nseq->alloc *= 2;
 307                 }
 308
 309                 info = kzalloc(sizeof(*info), GFP_ATOMIC);
 310                 if (!info) {
 311                         pr_warn("Cannot publish {%u,%u,%u}, no memory\n",
 312                                 type, lower, upper);
 313                         return NULL;
 314                 }
 315
 316                 INIT_LIST_HEAD(&info->node_list);
 317                 INIT_LIST_HEAD(&info->cluster_list);
 318                 INIT_LIST_HEAD(&info->zone_list);
 319
 320                 /* Insert new sub-sequence */
 321                 sseq = &nseq->sseqs[inspos];
 322                 freesseq = &nseq->sseqs[nseq->first_free];
 323                 memmove(sseq + 1, sseq, (freesseq - sseq) * sizeof(*sseq));
 324                 memset(sseq, 0, sizeof(*sseq));
 325                 nseq->first_free++;
 326                 sseq->lower = lower;
 327                 sseq->upper = upper;
 328                 sseq->info = info;
 329                 created_subseq = 1;
 330         }
 331
 332         /* Insert a publication */
 333         publ = publ_create(type, lower, upper, scope, node, port, key);
 334         if (!publ)
 335                 return NULL;
 336
 337         list_add(&publ->zone_list, &info->zone_list);
 338         info->zone_list_size++;
 339
 340         if (in_own_cluster(node)) {
 341                 list_add(&publ->cluster_list, &info->cluster_list);
 342                 info->cluster_list_size++;
 343         }
 344
 345         if (in_own_node(node)) {
 346                 list_add(&publ->node_list, &info->node_list);
 347                 info->node_list_size++;
 348         }
 349
 350         /* Any subscriptions waiting for notification?  */
 351         list_for_each_entry_safe(s, st, &nseq->subscriptions, nameseq_list) {
 352                 tipc_subscr_report_overlap(s,
 353                                            publ->lower,
 354                                            publ->upper,
 355                                            TIPC_PUBLISHED,
 356                                            publ->ref,
 357                                            publ->node,
 358                                            created_subseq);
 359         }
 360         return publ;
 361 }
 362
 363 /**
 364  * tipc_nameseq_remove_publ
 365  *
 366  * NOTE: There may be cases where TIPC is asked to remove a publication
 367  * that is not in the name table.  For example, if another node issues a
 368  * publication for a name sequence that overlaps an existing name sequence
 369  * the publication will not be recorded, which means the publication won't
 370  * be found when the name sequence is later withdrawn by that node.
 371  * A failed withdraw request simply returns a failure indication and lets the
 372  * caller issue any error or warning messages associated with such a problem.
 373  */
 374 static struct publication *tipc_nameseq_remove_publ(struct name_seq *nseq, u32 inst,
 375                                                     u32 node, u32 ref, u32 key)
 376 {
 377         struct publication *publ;
 378         struct sub_seq *sseq = nameseq_find_subseq(nseq, inst);
 379         struct name_info *info;
 380         struct sub_seq *free;
 381         struct tipc_subscription *s, *st;
 382         int removed_subseq = 0;
 383
 384         if (!sseq)
 385                 return NULL;
 386
 387         info = sseq->info;
 388
 389         /* Locate publication, if it exists */
 390         list_for_each_entry(publ, &info->zone_list, zone_list) {
 391                 if ((publ->key == key) && (publ->ref == ref) &&
 392                     (!publ->node || (publ->node == node)))
 393                         goto found;
 394         }
 395         return NULL;
 396
 397 found:
 398         /* Remove publication from zone scope list */
 399         list_del(&publ->zone_list);
 400         info->zone_list_size--;
 401
 402         /* Remove publication from cluster scope list, if present */
 403         if (in_own_cluster(node)) {
 404                 list_del(&publ->cluster_list);
 405                 info->cluster_list_size--;
 406         }
 407
 408         /* Remove publication from node scope list, if present */
 409         if (in_own_node(node)) {
 410                 list_del(&publ->node_list);
 411                 info->node_list_size--;
 412         }
 413
 414         /* Contract subseq list if no more publications for that subseq */
 415         if (list_empty(&info->zone_list)) {
 416                 kfree(info);
 417                 free = &nseq->sseqs[nseq->first_free--];
 418                 memmove(sseq, sseq + 1, (free - (sseq + 1)) * sizeof(*sseq));
 419                 removed_subseq = 1;
 420         }
 421
 422         /* Notify any waiting subscriptions */
 423         list_for_each_entry_safe(s, st, &nseq->subscriptions, nameseq_list) {
 424                 tipc_subscr_report_overlap(s,
 425                                            publ->lower,
 426                                            publ->upper,
 427                                            TIPC_WITHDRAWN,
 428                                            publ->ref,
 429                                            publ->node,
 430                                            removed_subseq);
 431         }
 432
 433         return publ;
 434 }
 435
 436 /**
 437  * tipc_nameseq_subscribe - attach a subscription, and issue
 438  * the prescribed number of events if there is any sub-
 439  * sequence overlapping with the requested sequence
 440  */
 441 static void tipc_nameseq_subscribe(struct name_seq *nseq,
 442                                    struct tipc_subscription *s)
 443 {
 444         struct sub_seq *sseq = nseq->sseqs;
 445
 446         list_add(&s->nameseq_list, &nseq->subscriptions);
 447
 448         if (!sseq)
 449                 return;
 450
 451         while (sseq != &nseq->sseqs[nseq->first_free]) {
 452                 if (tipc_subscr_overlap(s, sseq->lower, sseq->upper)) {
 453                         struct publication *crs;
 454                         struct name_info *info = sseq->info;
 455                         int must_report = 1;
 456
 457                         list_for_each_entry(crs, &info->zone_list, zone_list) {
 458                                 tipc_subscr_report_overlap(s,
 459                                                            sseq->lower,
 460                                                            sseq->upper,
 461                                                            TIPC_PUBLISHED,
 462                                                            crs->ref,
 463                                                            crs->node,
 464                                                            must_report);
 465                                 must_report = 0;
 466                         }
 467                 }
 468                 sseq++;
 469         }
 470 }
 471
 472 static struct name_seq *nametbl_find_seq(u32 type)
 473 {
 474         struct hlist_head *seq_head;
 475         struct name_seq *ns;
 476
 477         seq_head = &table.types[hash(type)];
 478         hlist_for_each_entry(ns, seq_head, ns_list) {
 479                 if (ns->type == type)
 480                         return ns;
 481         }
 482
 483         return NULL;
 484 };
 485
 486 struct publication *tipc_nametbl_insert_publ(u32 type, u32 lower, u32 upper,
 487                                              u32 scope, u32 node, u32 port, u32 key)
 488 {
 489         struct name_seq *seq = nametbl_find_seq(type);
 490
 491         if ((scope < TIPC_ZONE_SCOPE) || (scope > TIPC_NODE_SCOPE) ||
 492             (lower > upper)) {
 493                 pr_debug("Failed to publish illegal {%u,%u,%u} with scope %u\n",
 494                          type, lower, upper, scope);
 495                 return NULL;
 496         }
 497
 498         if (!seq)
 499                 seq = tipc_nameseq_create(type, &table.types[hash(type)]);
 500         if (!seq)
 501                 return NULL;
 502
 503         return tipc_nameseq_insert_publ(seq, type, lower, upper,
 504                                         scope, node, port, key);
 505 }
 506
 507 struct publication *tipc_nametbl_remove_publ(u32 type, u32 lower,
 508                                              u32 node, u32 ref, u32 key)
 509 {
 510         struct publication *publ;
 511         struct name_seq *seq = nametbl_find_seq(type);
 512
 513         if (!seq)
 514                 return NULL;
 515
 516         publ = tipc_nameseq_remove_publ(seq, lower, node, ref, key);
 517         nameseq_delete_empty(seq);
 518         return publ;
 519 }
 520
 521 /**
 522  * tipc_nametbl_translate - perform name translation
 523  *
 524  * On entry, 'destnode' is the search domain used during translation.
 525  *
 526  * On exit:
 527  * - if name translation is deferred to another node/cluster/zone,
 528  *   leaves 'destnode' unchanged (will be non-zero) and returns 0
 529  * - if name translation is attempted and succeeds, sets 'destnode'
 530  *   to publishing node and returns port reference (will be non-zero)
 531  * - if name translation is attempted and fails, sets 'destnode' to 0
 532  *   and returns 0
 533  */
 534 u32 tipc_nametbl_translate(u32 type, u32 instance, u32 *destnode)
 535 {
 536         struct sub_seq *sseq;
 537         struct name_info *info;
 538         struct publication *publ;
 539         struct name_seq *seq;
 540         u32 ref = 0;
 541         u32 node = 0;
 542
 543         if (!tipc_in_scope(*destnode, tipc_own_addr))
 544                 return 0;
 545
 546         read_lock_bh(&tipc_nametbl_lock);
 547         seq = nametbl_find_seq(type);
 548         if (unlikely(!seq))
 549                 goto not_found;
 550         sseq = nameseq_find_subseq(seq, instance);
 551         if (unlikely(!sseq))
 552                 goto not_found;
 553         spin_lock_bh(&seq->lock);
 554         info = sseq->info;
 555
 556         /* Closest-First Algorithm */
 557         if (likely(!*destnode)) {
 558                 if (!list_empty(&info->node_list)) {
 559                         publ = list_first_entry(&info->node_list,
 560                                                 struct publication,
 561                                                 node_list);
 562                         list_move_tail(&publ->node_list,
 563                                        &info->node_list);
 564                 } else if (!list_empty(&info->cluster_list)) {
 565                         publ = list_first_entry(&info->cluster_list,
 566                                                 struct publication,
 567                                                 cluster_list);
 568                         list_move_tail(&publ->cluster_list,
 569                                        &info->cluster_list);
 570                 } else {
 571                         publ = list_first_entry(&info->zone_list,
 572                                                 struct publication,
 573                                                 zone_list);
 574                         list_move_tail(&publ->zone_list,
 575                                        &info->zone_list);
 576                 }
 577         }
 578
 579         /* Round-Robin Algorithm */
 580         else if (*destnode == tipc_own_addr) {
 581                 if (list_empty(&info->node_list))
 582                         goto no_match;
 583                 publ = list_first_entry(&info->node_list, struct publication,
 584                                         node_list);
 585                 list_move_tail(&publ->node_list, &info->node_list);
 586         } else if (in_own_cluster_exact(*destnode)) {
 587                 if (list_empty(&info->cluster_list))
 588                         goto no_match;
 589                 publ = list_first_entry(&info->cluster_list, struct publication,
 590                                         cluster_list);
 591                 list_move_tail(&publ->cluster_list, &info->cluster_list);
 592         } else {
 593                 publ = list_first_entry(&info->zone_list, struct publication,
 594                                         zone_list);
 595                 list_move_tail(&publ->zone_list, &info->zone_list);
 596         }
 597
 598         ref = publ->ref;
 599         node = publ->node;
 600 no_match:
 601         spin_unlock_bh(&seq->lock);
 602 not_found:
 603         read_unlock_bh(&tipc_nametbl_lock);
 604         *destnode = node;
 605         return ref;
 606 }
 607
 608 /**
 609  * tipc_nametbl_mc_translate - find multicast destinations
 610  *
 611  * Creates list of all local ports that overlap the given multicast address;
 612  * also determines if any off-node ports overlap.
 613  *
 614  * Note: Publications with a scope narrower than 'limit' are ignored.
 615  * (i.e. local node-scope publications mustn't receive messages arriving
 616  * from another node, even if the multcast link brought it here)
 617  *
 618  * Returns non-zero if any off-node ports overlap
 619  */
 620 int tipc_nametbl_mc_translate(u32 type, u32 lower, u32 upper, u32 limit,
 621                               struct tipc_port_list *dports)
 622 {
 623         struct name_seq *seq;
 624         struct sub_seq *sseq;
 625         struct sub_seq *sseq_stop;
 626         struct name_info *info;
 627         int res = 0;
 628
 629         read_lock_bh(&tipc_nametbl_lock);
 630         seq = nametbl_find_seq(type);
 631         if (!seq)
 632                 goto exit;
 633
 634         spin_lock_bh(&seq->lock);
 635
 636         sseq = seq->sseqs + nameseq_locate_subseq(seq, lower);
 637         sseq_stop = seq->sseqs + seq->first_free;
 638         for (; sseq != sseq_stop; sseq++) {
 639                 struct publication *publ;
 640
 641                 if (sseq->lower > upper)
 642                         break;
 643
 644                 info = sseq->info;
 645                 list_for_each_entry(publ, &info->node_list, node_list) {
 646                         if (publ->scope <= limit)
 647                                 tipc_port_list_add(dports, publ->ref);
 648                 }
 649
 650                 if (info->cluster_list_size != info->node_list_size)
 651                         res = 1;
 652         }
 653
 654         spin_unlock_bh(&seq->lock);
 655 exit:
 656         read_unlock_bh(&tipc_nametbl_lock);
 657         return res;
 658 }
 659
 660 /*
 661  * tipc_nametbl_publish - add name publication to network name tables
 662  */
 663 struct publication *tipc_nametbl_publish(u32 type, u32 lower, u32 upper,
 664                                          u32 scope, u32 port_ref, u32 key)
 665 {
 666         struct publication *publ;
 667
 668         if (table.local_publ_count >= TIPC_MAX_PUBLICATIONS) {
 669                 pr_warn("Publication failed, local publication limit reached (%u)\n",
 670                         TIPC_MAX_PUBLICATIONS);
 671                 return NULL;
 672         }
 673
 674         write_lock_bh(&tipc_nametbl_lock);
 675         publ = tipc_nametbl_insert_publ(type, lower, upper, scope,
 676                                    tipc_own_addr, port_ref, key);
 677         if (likely(publ)) {
 678                 table.local_publ_count++;
 679                 tipc_named_publish(publ);
 680         }
 681         write_unlock_bh(&tipc_nametbl_lock);
 682         return publ;
 683 }
 684
 685 /**
 686  * tipc_nametbl_withdraw - withdraw name publication from network name tables
 687  */
 688 int tipc_nametbl_withdraw(u32 type, u32 lower, u32 ref, u32 key)
 689 {
 690         struct publication *publ;
 691
 692         write_lock_bh(&tipc_nametbl_lock);
 693         publ = tipc_nametbl_remove_publ(type, lower, tipc_own_addr, ref, key);
 694         if (likely(publ)) {
 695                 table.local_publ_count--;
 696                 tipc_named_withdraw(publ);
 697                 write_unlock_bh(&tipc_nametbl_lock);
 698                 list_del_init(&publ->pport_list);
 699                 kfree(publ);
 700                 return 1;
 701         }
 702         write_unlock_bh(&tipc_nametbl_lock);
 703         pr_err("Unable to remove local publication\n"
 704                "(type=%u, lower=%u, ref=%u, key=%u)\n",
 705                type, lower, ref, key);
 706         return 0;
 707 }
 708
 709 /**
 710  * tipc_nametbl_subscribe - add a subscription object to the name table
 711  */
 712 void tipc_nametbl_subscribe(struct tipc_subscription *s)
 713 {
 714         u32 type = s->seq.type;
 715         struct name_seq *seq;
 716
 717         write_lock_bh(&tipc_nametbl_lock);
 718         seq = nametbl_find_seq(type);
 719         if (!seq)
 720                 seq = tipc_nameseq_create(type, &table.types[hash(type)]);
 721         if (seq) {
 722                 spin_lock_bh(&seq->lock);
 723                 tipc_nameseq_subscribe(seq, s);
 724                 spin_unlock_bh(&seq->lock);
 725         } else {
 726                 pr_warn("Failed to create subscription for {%u,%u,%u}\n",
 727                         s->seq.type, s->seq.lower, s->seq.upper);
 728         }
 729         write_unlock_bh(&tipc_nametbl_lock);
 730 }
 731
 732 /**
 733  * tipc_nametbl_unsubscribe - remove a subscription object from name table
 734  */
 735 void tipc_nametbl_unsubscribe(struct tipc_subscription *s)
 736 {
 737         struct name_seq *seq;
 738
 739         write_lock_bh(&tipc_nametbl_lock);
 740         seq = nametbl_find_seq(s->seq.type);
 741         if (seq != NULL) {
 742                 spin_lock_bh(&seq->lock);
 743                 list_del_init(&s->nameseq_list);
 744                 spin_unlock_bh(&seq->lock);
 745                 nameseq_delete_empty(seq);
 746         }
 747         write_unlock_bh(&tipc_nametbl_lock);
 748 }
 749
 750
 751 /**
 752  * subseq_list - print specified sub-sequence contents into the given buffer
 753  */
 754 static int subseq_list(struct sub_seq *sseq, char *buf, int len, u32 depth,
 755                        u32 index)
 756 {
 757         char portIdStr[27];
 758         const char *scope_str[] = {"", " zone", " cluster", " node"};
 759         struct publication *publ;
 760         struct name_info *info;
 761         int ret;
 762
 763         ret = tipc_snprintf(buf, len, "%-10u %-10u ", sseq->lower, sseq->upper);
 764
 765         if (depth == 2) {
 766                 ret += tipc_snprintf(buf - ret, len + ret, "\n");
 767                 return ret;
 768         }
 769
 770         info = sseq->info;
 771
 772         list_for_each_entry(publ, &info->zone_list, zone_list) {
 773                 sprintf(portIdStr, "<%u.%u.%u:%u>",
 774                          tipc_zone(publ->node), tipc_cluster(publ->node),
 775                          tipc_node(publ->node), publ->ref);
 776                 ret += tipc_snprintf(buf + ret, len - ret, "%-26s ", portIdStr);
 777                 if (depth > 3) {
 778                         ret += tipc_snprintf(buf + ret, len - ret, "%-10u %s",
 779                                              publ->key, scope_str[publ->scope]);
 780                 }
 781                 if (!list_is_last(&publ->zone_list, &info->zone_list))
 782                         ret += tipc_snprintf(buf + ret, len - ret,
 783                                              "\n%33s", " ");
 784         }
 785
 786         ret += tipc_snprintf(buf + ret, len - ret, "\n");
 787         return ret;
 788 }
 789
 790 /**
 791  * nameseq_list - print specified name sequence contents into the given buffer
 792  */
 793 static int nameseq_list(struct name_seq *seq, char *buf, int len, u32 depth,
 794                         u32 type, u32 lowbound, u32 upbound, u32 index)
 795 {
 796         struct sub_seq *sseq;
 797         char typearea[11];
 798         int ret = 0;
 799
 800         if (seq->first_free == 0)
 801                 return 0;
 802
 803         sprintf(typearea, "%-10u", seq->type);
 804
 805         if (depth == 1) {
 806                 ret += tipc_snprintf(buf, len, "%s\n", typearea);
 807                 return ret;
 808         }
 809
 810         for (sseq = seq->sseqs; sseq != &seq->sseqs[seq->first_free]; sseq++) {
 811                 if ((lowbound <= sseq->upper) && (upbound >= sseq->lower)) {
 812                         ret += tipc_snprintf(buf + ret, len - ret, "%s ",
 813                                             typearea);
 814                         spin_lock_bh(&seq->lock);
 815                         ret += subseq_list(sseq, buf + ret, len - ret,
 816                                           depth, index);
 817                         spin_unlock_bh(&seq->lock);
 818                         sprintf(typearea, "%10s", " ");
 819                 }
 820         }
 821         return ret;
 822 }
 823
 824 /**
 825  * nametbl_header - print name table header into the given buffer
 826  */
 827 static int nametbl_header(char *buf, int len, u32 depth)
 828 {
 829         const char *header[] = {
 830                 "Type       ",
 831                 "Lower      Upper      ",
 832                 "Port Identity              ",
 833                 "Publication Scope"
 834         };
 835
 836         int i;
 837         int ret = 0;
 838
 839         if (depth > 4)
 840                 depth = 4;
 841         for (i = 0; i < depth; i++)
 842                 ret += tipc_snprintf(buf + ret, len - ret, header[i]);
 843         ret += tipc_snprintf(buf + ret, len - ret, "\n");
 844         return ret;
 845 }
 846
 847 /**
 848  * nametbl_list - print specified name table contents into the given buffer
 849  */
 850 static int nametbl_list(char *buf, int len, u32 depth_info,
 851                         u32 type, u32 lowbound, u32 upbound)
 852 {
 853         struct hlist_head *seq_head;
 854         struct name_seq *seq;
 855         int all_types;
 856         int ret = 0;
 857         u32 depth;
 858         u32 i;
 859
 860         all_types = (depth_info & TIPC_NTQ_ALLTYPES);
 861         depth = (depth_info & ~TIPC_NTQ_ALLTYPES);
 862
 863         if (depth == 0)
 864                 return 0;
 865
 866         if (all_types) {
 867                 /* display all entries in name table to specified depth */
 868                 ret += nametbl_header(buf, len, depth);
 869                 lowbound = 0;
 870                 upbound = ~0;
 871                 for (i = 0; i < TIPC_NAMETBL_SIZE; i++) {
 872                         seq_head = &table.types[i];
 873                         hlist_for_each_entry(seq, seq_head, ns_list) {
 874                                 ret += nameseq_list(seq, buf + ret, len - ret,
 875                                                    depth, seq->type,
 876                                                    lowbound, upbound, i);
 877                         }
 878                 }
 879         } else {
 880                 /* display only the sequence that matches the specified type */
 881                 if (upbound < lowbound) {
 882                         ret += tipc_snprintf(buf + ret, len - ret,
 883                                         "invalid name sequence specified\n");
 884                         return ret;
 885                 }
 886                 ret += nametbl_header(buf + ret, len - ret, depth);
 887                 i = hash(type);
 888                 seq_head = &table.types[i];
 889                 hlist_for_each_entry(seq, seq_head, ns_list) {
 890                         if (seq->type == type) {
 891                                 ret += nameseq_list(seq, buf + ret, len - ret,
 892                                                    depth, type,
 893                                                    lowbound, upbound, i);
 894                                 break;
 895                         }
 896                 }
 897         }
 898         return ret;
 899 }
 900
 901 struct sk_buff *tipc_nametbl_get(const void *req_tlv_area, int req_tlv_space)
 902 {
 903         struct sk_buff *buf;
 904         struct tipc_name_table_query *argv;
 905         struct tlv_desc *rep_tlv;
 906         char *pb;
 907         int pb_len;
 908         int str_len;
 909
 910         if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_NAME_TBL_QUERY))
 911                 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR);
 912
 913         buf = tipc_cfg_reply_alloc(TLV_SPACE(ULTRA_STRING_MAX_LEN));
 914         if (!buf)
 915                 return NULL;
 916
 917         rep_tlv = (struct tlv_desc *)buf->data;
 918         pb = TLV_DATA(rep_tlv);
 919         pb_len = ULTRA_STRING_MAX_LEN;
 920         argv = (struct tipc_name_table_query *)TLV_DATA(req_tlv_area);
 921         read_lock_bh(&tipc_nametbl_lock);
 922         str_len = nametbl_list(pb, pb_len, ntohl(argv->depth),
 923                                ntohl(argv->type),
 924                                ntohl(argv->lowbound), ntohl(argv->upbound));
 925         read_unlock_bh(&tipc_nametbl_lock);
 926         str_len += 1;   /* for "\0" */
 927         skb_put(buf, TLV_SPACE(str_len));
 928         TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
 929
 930         return buf;
 931 }
 932
 933 int tipc_nametbl_init(void)
 934 {
 935         table.types = kcalloc(TIPC_NAMETBL_SIZE, sizeof(struct hlist_head),
 936                               GFP_ATOMIC);
 937         if (!table.types)
 938                 return -ENOMEM;
 939
 940         table.local_publ_count = 0;
 941         return 0;
 942 }
 943
 944 /**
 945  * tipc_purge_publications - remove all publications for a given type
 946  *
 947  * tipc_nametbl_lock must be held when calling this function
 948  */
 949 static void tipc_purge_publications(struct name_seq *seq)
 950 {
 951         struct publication *publ, *safe;
 952         struct sub_seq *sseq;
 953         struct name_info *info;
 954
 955         if (!seq->sseqs) {
 956                 nameseq_delete_empty(seq);
 957                 return;
 958         }
 959         sseq = seq->sseqs;
 960         info = sseq->info;
 961         list_for_each_entry_safe(publ, safe, &info->zone_list, zone_list) {
 962                 tipc_nametbl_remove_publ(publ->type, publ->lower, publ->node,
 963                                          publ->ref, publ->key);
 964         }
 965 }
 966
 967 void tipc_nametbl_stop(void)
 968 {
 969         u32 i;
 970         struct name_seq *seq;
 971         struct hlist_head *seq_head;
 972         struct hlist_node *safe;
 973
 974         /* Verify name table is empty and purge any lingering
 975          * publications, then release the name table
 976          */
 977         write_lock_bh(&tipc_nametbl_lock);
 978         for (i = 0; i < TIPC_NAMETBL_SIZE; i++) {
 979                 if (hlist_empty(&table.types[i]))
 980                         continue;
 981                 seq_head = &table.types[i];
 982                 hlist_for_each_entry_safe(seq, safe, seq_head, ns_list) {
 983                         tipc_purge_publications(seq);
 984                 }
 985                 continue;
 986         }
 987         kfree(table.types);
 988         table.types = NULL;
 989         write_unlock_bh(&tipc_nametbl_lock);
 990 }