7d85cc1ace0e67714e684809271e9629c5eb1d42
[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 if (!list_empty(&info->zone_list)) {
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                 } else
584                         goto no_match;
585         }
586
587         /* Round-Robin Algorithm: */
588         else if (*destnode == tipc_own_addr) {
589                 if (list_empty(&info->node_list))
590                         goto no_match;
591                 publ = list_first_entry(&info->node_list, struct publication,
592                                         node_list);
593                 list_move_tail(&publ->node_list, &info->node_list);
594         } else if (in_own_cluster(*destnode)) {
595                 if (list_empty(&info->cluster_list))
596                         goto no_match;
597                 publ = list_first_entry(&info->cluster_list, struct publication,
598                                         cluster_list);
599                 list_move_tail(&publ->cluster_list, &info->cluster_list);
600         } else {
601                 if (list_empty(&info->zone_list))
602                         goto no_match;
603                 publ = list_first_entry(&info->zone_list, struct publication,
604                                         zone_list);
605                 list_move_tail(&publ->zone_list, &info->zone_list);
606         }
607
608         ref = publ->ref;
609         *destnode = publ->node;
610 no_match:
611         spin_unlock_bh(&seq->lock);
612 not_found:
613         read_unlock_bh(&tipc_nametbl_lock);
614         return ref;
615 }
616
617 /**
618  * tipc_nametbl_mc_translate - find multicast destinations
619  *
620  * Creates list of all local ports that overlap the given multicast address;
621  * also determines if any off-node ports overlap.
622  *
623  * Note: Publications with a scope narrower than 'limit' are ignored.
624  * (i.e. local node-scope publications mustn't receive messages arriving
625  * from another node, even if the multcast link brought it here)
626  *
627  * Returns non-zero if any off-node ports overlap
628  */
629
630 int tipc_nametbl_mc_translate(u32 type, u32 lower, u32 upper, u32 limit,
631                               struct port_list *dports)
632 {
633         struct name_seq *seq;
634         struct sub_seq *sseq;
635         struct sub_seq *sseq_stop;
636         struct name_info *info;
637         int res = 0;
638
639         read_lock_bh(&tipc_nametbl_lock);
640         seq = nametbl_find_seq(type);
641         if (!seq)
642                 goto exit;
643
644         spin_lock_bh(&seq->lock);
645
646         sseq = seq->sseqs + nameseq_locate_subseq(seq, lower);
647         sseq_stop = seq->sseqs + seq->first_free;
648         for (; sseq != sseq_stop; sseq++) {
649                 struct publication *publ;
650
651                 if (sseq->lower > upper)
652                         break;
653
654                 info = sseq->info;
655                 list_for_each_entry(publ, &info->node_list, node_list) {
656                         if (publ->scope <= limit)
657                                 tipc_port_list_add(dports, publ->ref);
658                 }
659
660                 if (info->cluster_list_size != info->node_list_size)
661                         res = 1;
662         }
663
664         spin_unlock_bh(&seq->lock);
665 exit:
666         read_unlock_bh(&tipc_nametbl_lock);
667         return res;
668 }
669
670 /**
671  * tipc_nametbl_publish_rsv - publish port name using a reserved name type
672  */
673
674 int tipc_nametbl_publish_rsv(u32 ref, unsigned int scope,
675                         struct tipc_name_seq const *seq)
676 {
677         int res;
678
679         atomic_inc(&rsv_publ_ok);
680         res = tipc_publish(ref, scope, seq);
681         atomic_dec(&rsv_publ_ok);
682         return res;
683 }
684
685 /**
686  * tipc_nametbl_publish - add name publication to network name tables
687  */
688
689 struct publication *tipc_nametbl_publish(u32 type, u32 lower, u32 upper,
690                                     u32 scope, u32 port_ref, u32 key)
691 {
692         struct publication *publ;
693
694         if (table.local_publ_count >= tipc_max_publications) {
695                 warn("Publication failed, local publication limit reached (%u)\n",
696                      tipc_max_publications);
697                 return NULL;
698         }
699         if ((type < TIPC_RESERVED_TYPES) && !atomic_read(&rsv_publ_ok)) {
700                 warn("Publication failed, reserved name {%u,%u,%u}\n",
701                      type, lower, upper);
702                 return NULL;
703         }
704
705         write_lock_bh(&tipc_nametbl_lock);
706         table.local_publ_count++;
707         publ = tipc_nametbl_insert_publ(type, lower, upper, scope,
708                                    tipc_own_addr, port_ref, key);
709         if (publ && (scope != TIPC_NODE_SCOPE))
710                 tipc_named_publish(publ);
711         write_unlock_bh(&tipc_nametbl_lock);
712         return publ;
713 }
714
715 /**
716  * tipc_nametbl_withdraw - withdraw name publication from network name tables
717  */
718
719 int tipc_nametbl_withdraw(u32 type, u32 lower, u32 ref, u32 key)
720 {
721         struct publication *publ;
722
723         write_lock_bh(&tipc_nametbl_lock);
724         publ = tipc_nametbl_remove_publ(type, lower, tipc_own_addr, ref, key);
725         if (likely(publ)) {
726                 table.local_publ_count--;
727                 if (publ->scope != TIPC_NODE_SCOPE)
728                         tipc_named_withdraw(publ);
729                 write_unlock_bh(&tipc_nametbl_lock);
730                 list_del_init(&publ->pport_list);
731                 kfree(publ);
732                 return 1;
733         }
734         write_unlock_bh(&tipc_nametbl_lock);
735         err("Unable to remove local publication\n"
736             "(type=%u, lower=%u, ref=%u, key=%u)\n",
737             type, lower, ref, key);
738         return 0;
739 }
740
741 /**
742  * tipc_nametbl_subscribe - add a subscription object to the name table
743  */
744
745 void tipc_nametbl_subscribe(struct subscription *s)
746 {
747         u32 type = s->seq.type;
748         struct name_seq *seq;
749
750         write_lock_bh(&tipc_nametbl_lock);
751         seq = nametbl_find_seq(type);
752         if (!seq)
753                 seq = tipc_nameseq_create(type, &table.types[hash(type)]);
754         if (seq) {
755                 spin_lock_bh(&seq->lock);
756                 tipc_nameseq_subscribe(seq, s);
757                 spin_unlock_bh(&seq->lock);
758         } else {
759                 warn("Failed to create subscription for {%u,%u,%u}\n",
760                      s->seq.type, s->seq.lower, s->seq.upper);
761         }
762         write_unlock_bh(&tipc_nametbl_lock);
763 }
764
765 /**
766  * tipc_nametbl_unsubscribe - remove a subscription object from name table
767  */
768
769 void tipc_nametbl_unsubscribe(struct subscription *s)
770 {
771         struct name_seq *seq;
772
773         write_lock_bh(&tipc_nametbl_lock);
774         seq = nametbl_find_seq(s->seq.type);
775         if (seq != NULL) {
776                 spin_lock_bh(&seq->lock);
777                 list_del_init(&s->nameseq_list);
778                 spin_unlock_bh(&seq->lock);
779                 if ((seq->first_free == 0) && list_empty(&seq->subscriptions)) {
780                         hlist_del_init(&seq->ns_list);
781                         kfree(seq->sseqs);
782                         kfree(seq);
783                 }
784         }
785         write_unlock_bh(&tipc_nametbl_lock);
786 }
787
788
789 /**
790  * subseq_list: print specified sub-sequence contents into the given buffer
791  */
792
793 static void subseq_list(struct sub_seq *sseq, struct print_buf *buf, u32 depth,
794                         u32 index)
795 {
796         char portIdStr[27];
797         const char *scope_str[] = {"", " zone", " cluster", " node"};
798         struct publication *publ;
799         struct name_info *info;
800
801         tipc_printf(buf, "%-10u %-10u ", sseq->lower, sseq->upper);
802
803         if (depth == 2) {
804                 tipc_printf(buf, "\n");
805                 return;
806         }
807
808         info = sseq->info;
809
810         list_for_each_entry(publ, &info->zone_list, zone_list) {
811                 sprintf(portIdStr, "<%u.%u.%u:%u>",
812                          tipc_zone(publ->node), tipc_cluster(publ->node),
813                          tipc_node(publ->node), publ->ref);
814                 tipc_printf(buf, "%-26s ", portIdStr);
815                 if (depth > 3) {
816                         tipc_printf(buf, "%-10u %s", publ->key,
817                                     scope_str[publ->scope]);
818                 }
819                 if (!list_is_last(&publ->zone_list, &info->zone_list))
820                         tipc_printf(buf, "\n%33s", " ");
821         };
822
823         tipc_printf(buf, "\n");
824 }
825
826 /**
827  * nameseq_list: print specified name sequence contents into the given buffer
828  */
829
830 static void nameseq_list(struct name_seq *seq, struct print_buf *buf, u32 depth,
831                          u32 type, u32 lowbound, u32 upbound, u32 index)
832 {
833         struct sub_seq *sseq;
834         char typearea[11];
835
836         if (seq->first_free == 0)
837                 return;
838
839         sprintf(typearea, "%-10u", seq->type);
840
841         if (depth == 1) {
842                 tipc_printf(buf, "%s\n", typearea);
843                 return;
844         }
845
846         for (sseq = seq->sseqs; sseq != &seq->sseqs[seq->first_free]; sseq++) {
847                 if ((lowbound <= sseq->upper) && (upbound >= sseq->lower)) {
848                         tipc_printf(buf, "%s ", typearea);
849                         spin_lock_bh(&seq->lock);
850                         subseq_list(sseq, buf, depth, index);
851                         spin_unlock_bh(&seq->lock);
852                         sprintf(typearea, "%10s", " ");
853                 }
854         }
855 }
856
857 /**
858  * nametbl_header - print name table header into the given buffer
859  */
860
861 static void nametbl_header(struct print_buf *buf, u32 depth)
862 {
863         const char *header[] = {
864                 "Type       ",
865                 "Lower      Upper      ",
866                 "Port Identity              ",
867                 "Publication Scope"
868         };
869
870         int i;
871
872         if (depth > 4)
873                 depth = 4;
874         for (i = 0; i < depth; i++)
875                 tipc_printf(buf, header[i]);
876         tipc_printf(buf, "\n");
877 }
878
879 /**
880  * nametbl_list - print specified name table contents into the given buffer
881  */
882
883 static void nametbl_list(struct print_buf *buf, u32 depth_info,
884                          u32 type, u32 lowbound, u32 upbound)
885 {
886         struct hlist_head *seq_head;
887         struct hlist_node *seq_node;
888         struct name_seq *seq;
889         int all_types;
890         u32 depth;
891         u32 i;
892
893         all_types = (depth_info & TIPC_NTQ_ALLTYPES);
894         depth = (depth_info & ~TIPC_NTQ_ALLTYPES);
895
896         if (depth == 0)
897                 return;
898
899         if (all_types) {
900                 /* display all entries in name table to specified depth */
901                 nametbl_header(buf, depth);
902                 lowbound = 0;
903                 upbound = ~0;
904                 for (i = 0; i < tipc_nametbl_size; i++) {
905                         seq_head = &table.types[i];
906                         hlist_for_each_entry(seq, seq_node, seq_head, ns_list) {
907                                 nameseq_list(seq, buf, depth, seq->type,
908                                              lowbound, upbound, i);
909                         }
910                 }
911         } else {
912                 /* display only the sequence that matches the specified type */
913                 if (upbound < lowbound) {
914                         tipc_printf(buf, "invalid name sequence specified\n");
915                         return;
916                 }
917                 nametbl_header(buf, depth);
918                 i = hash(type);
919                 seq_head = &table.types[i];
920                 hlist_for_each_entry(seq, seq_node, seq_head, ns_list) {
921                         if (seq->type == type) {
922                                 nameseq_list(seq, buf, depth, type,
923                                              lowbound, upbound, i);
924                                 break;
925                         }
926                 }
927         }
928 }
929
930 #define MAX_NAME_TBL_QUERY 32768
931
932 struct sk_buff *tipc_nametbl_get(const void *req_tlv_area, int req_tlv_space)
933 {
934         struct sk_buff *buf;
935         struct tipc_name_table_query *argv;
936         struct tlv_desc *rep_tlv;
937         struct print_buf b;
938         int str_len;
939
940         if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_NAME_TBL_QUERY))
941                 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR);
942
943         buf = tipc_cfg_reply_alloc(TLV_SPACE(MAX_NAME_TBL_QUERY));
944         if (!buf)
945                 return NULL;
946
947         rep_tlv = (struct tlv_desc *)buf->data;
948         tipc_printbuf_init(&b, TLV_DATA(rep_tlv), MAX_NAME_TBL_QUERY);
949         argv = (struct tipc_name_table_query *)TLV_DATA(req_tlv_area);
950         read_lock_bh(&tipc_nametbl_lock);
951         nametbl_list(&b, ntohl(argv->depth), ntohl(argv->type),
952                      ntohl(argv->lowbound), ntohl(argv->upbound));
953         read_unlock_bh(&tipc_nametbl_lock);
954         str_len = tipc_printbuf_validate(&b);
955
956         skb_put(buf, TLV_SPACE(str_len));
957         TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
958
959         return buf;
960 }
961
962 int tipc_nametbl_init(void)
963 {
964         table.types = kcalloc(tipc_nametbl_size, sizeof(struct hlist_head),
965                               GFP_ATOMIC);
966         if (!table.types)
967                 return -ENOMEM;
968
969         table.local_publ_count = 0;
970         return 0;
971 }
972
973 void tipc_nametbl_stop(void)
974 {
975         u32 i;
976
977         if (!table.types)
978                 return;
979
980         /* Verify name table is empty, then release it */
981
982         write_lock_bh(&tipc_nametbl_lock);
983         for (i = 0; i < tipc_nametbl_size; i++) {
984                 if (!hlist_empty(&table.types[i]))
985                         err("tipc_nametbl_stop(): hash chain %u is non-null\n", i);
986         }
987         kfree(table.types);
988         table.types = NULL;
989         write_unlock_bh(&tipc_nametbl_lock);
990 }
991