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