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