tipc: Remove obsolete native API files and exports
[linux-2.6.git] / net / tipc / subscr.c
1 /*
2  * net/tipc/subscr.c: TIPC network topology service
3  *
4  * Copyright (c) 2000-2006, Ericsson AB
5  * Copyright (c) 2005-2007, 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 "dbg.h"
39 #include "name_table.h"
40 #include "user_reg.h"
41 #include "port.h"
42 #include "ref.h"
43 #include "subscr.h"
44
45 /**
46  * struct subscriber - TIPC network topology subscriber
47  * @port_ref: object reference to server port connecting to subscriber
48  * @lock: pointer to spinlock controlling access to subscriber's server port
49  * @subscriber_list: adjacent subscribers in top. server's list of subscribers
50  * @subscription_list: list of subscription objects for this subscriber
51  */
52
53 struct subscriber {
54         u32 port_ref;
55         spinlock_t *lock;
56         struct list_head subscriber_list;
57         struct list_head subscription_list;
58 };
59
60 /**
61  * struct top_srv - TIPC network topology subscription service
62  * @user_ref: TIPC userid of subscription service
63  * @setup_port: reference to TIPC port that handles subscription requests
64  * @subscription_count: number of active subscriptions (not subscribers!)
65  * @subscriber_list: list of ports subscribing to service
66  * @lock: spinlock govering access to subscriber list
67  */
68
69 struct top_srv {
70         u32 user_ref;
71         u32 setup_port;
72         atomic_t subscription_count;
73         struct list_head subscriber_list;
74         spinlock_t lock;
75 };
76
77 static struct top_srv topsrv = { 0 };
78
79 /**
80  * htohl - convert value to endianness used by destination
81  * @in: value to convert
82  * @swap: non-zero if endianness must be reversed
83  *
84  * Returns converted value
85  */
86
87 static u32 htohl(u32 in, int swap)
88 {
89         return swap ? swab32(in) : in;
90 }
91
92 /**
93  * subscr_send_event - send a message containing a tipc_event to the subscriber
94  *
95  * Note: Must not hold subscriber's server port lock, since tipc_send() will
96  *       try to take the lock if the message is rejected and returned!
97  */
98
99 static void subscr_send_event(struct subscription *sub,
100                               u32 found_lower,
101                               u32 found_upper,
102                               u32 event,
103                               u32 port_ref,
104                               u32 node)
105 {
106         struct iovec msg_sect;
107
108         msg_sect.iov_base = (void *)&sub->evt;
109         msg_sect.iov_len = sizeof(struct tipc_event);
110
111         sub->evt.event = htohl(event, sub->swap);
112         sub->evt.found_lower = htohl(found_lower, sub->swap);
113         sub->evt.found_upper = htohl(found_upper, sub->swap);
114         sub->evt.port.ref = htohl(port_ref, sub->swap);
115         sub->evt.port.node = htohl(node, sub->swap);
116         tipc_send(sub->server_ref, 1, &msg_sect);
117 }
118
119 /**
120  * tipc_subscr_overlap - test for subscription overlap with the given values
121  *
122  * Returns 1 if there is overlap, otherwise 0.
123  */
124
125 int tipc_subscr_overlap(struct subscription *sub,
126                         u32 found_lower,
127                         u32 found_upper)
128
129 {
130         if (found_lower < sub->seq.lower)
131                 found_lower = sub->seq.lower;
132         if (found_upper > sub->seq.upper)
133                 found_upper = sub->seq.upper;
134         if (found_lower > found_upper)
135                 return 0;
136         return 1;
137 }
138
139 /**
140  * tipc_subscr_report_overlap - issue event if there is subscription overlap
141  *
142  * Protected by nameseq.lock in name_table.c
143  */
144
145 void tipc_subscr_report_overlap(struct subscription *sub,
146                                 u32 found_lower,
147                                 u32 found_upper,
148                                 u32 event,
149                                 u32 port_ref,
150                                 u32 node,
151                                 int must)
152 {
153         if (!tipc_subscr_overlap(sub, found_lower, found_upper))
154                 return;
155         if (!must && !(sub->filter & TIPC_SUB_PORTS))
156                 return;
157
158         sub->event_cb(sub, found_lower, found_upper, event, port_ref, node);
159 }
160
161 /**
162  * subscr_timeout - subscription timeout has occurred
163  */
164
165 static void subscr_timeout(struct subscription *sub)
166 {
167         struct port *server_port;
168
169         /* Validate server port reference (in case subscriber is terminating) */
170
171         server_port = tipc_port_lock(sub->server_ref);
172         if (server_port == NULL)
173                 return;
174
175         /* Validate timeout (in case subscription is being cancelled) */
176
177         if (sub->timeout == TIPC_WAIT_FOREVER) {
178                 tipc_port_unlock(server_port);
179                 return;
180         }
181
182         /* Unlink subscription from name table */
183
184         tipc_nametbl_unsubscribe(sub);
185
186         /* Unlink subscription from subscriber */
187
188         list_del(&sub->subscription_list);
189
190         /* Release subscriber's server port */
191
192         tipc_port_unlock(server_port);
193
194         /* Notify subscriber of timeout */
195
196         subscr_send_event(sub, sub->evt.s.seq.lower, sub->evt.s.seq.upper,
197                           TIPC_SUBSCR_TIMEOUT, 0, 0);
198
199         /* Now destroy subscription */
200
201         k_term_timer(&sub->timer);
202         kfree(sub);
203         atomic_dec(&topsrv.subscription_count);
204 }
205
206 /**
207  * subscr_del - delete a subscription within a subscription list
208  *
209  * Called with subscriber port locked.
210  */
211
212 static void subscr_del(struct subscription *sub)
213 {
214         tipc_nametbl_unsubscribe(sub);
215         list_del(&sub->subscription_list);
216         kfree(sub);
217         atomic_dec(&topsrv.subscription_count);
218 }
219
220 /**
221  * subscr_terminate - terminate communication with a subscriber
222  *
223  * Called with subscriber port locked.  Routine must temporarily release lock
224  * to enable subscription timeout routine(s) to finish without deadlocking;
225  * the lock is then reclaimed to allow caller to release it upon return.
226  * (This should work even in the unlikely event some other thread creates
227  * a new object reference in the interim that uses this lock; this routine will
228  * simply wait for it to be released, then claim it.)
229  */
230
231 static void subscr_terminate(struct subscriber *subscriber)
232 {
233         u32 port_ref;
234         struct subscription *sub;
235         struct subscription *sub_temp;
236
237         /* Invalidate subscriber reference */
238
239         port_ref = subscriber->port_ref;
240         subscriber->port_ref = 0;
241         spin_unlock_bh(subscriber->lock);
242
243         /* Sever connection to subscriber */
244
245         tipc_shutdown(port_ref);
246         tipc_deleteport(port_ref);
247
248         /* Destroy any existing subscriptions for subscriber */
249
250         list_for_each_entry_safe(sub, sub_temp, &subscriber->subscription_list,
251                                  subscription_list) {
252                 if (sub->timeout != TIPC_WAIT_FOREVER) {
253                         k_cancel_timer(&sub->timer);
254                         k_term_timer(&sub->timer);
255                 }
256                 dbg("Term: Removing sub %u,%u,%u from subscriber %x list\n",
257                     sub->seq.type, sub->seq.lower, sub->seq.upper, subscriber);
258                 subscr_del(sub);
259         }
260
261         /* Remove subscriber from topology server's subscriber list */
262
263         spin_lock_bh(&topsrv.lock);
264         list_del(&subscriber->subscriber_list);
265         spin_unlock_bh(&topsrv.lock);
266
267         /* Reclaim subscriber lock */
268
269         spin_lock_bh(subscriber->lock);
270
271         /* Now destroy subscriber */
272
273         kfree(subscriber);
274 }
275
276 /**
277  * subscr_cancel - handle subscription cancellation request
278  *
279  * Called with subscriber port locked.  Routine must temporarily release lock
280  * to enable the subscription timeout routine to finish without deadlocking;
281  * the lock is then reclaimed to allow caller to release it upon return.
282  *
283  * Note that fields of 's' use subscriber's endianness!
284  */
285
286 static void subscr_cancel(struct tipc_subscr *s,
287                           struct subscriber *subscriber)
288 {
289         struct subscription *sub;
290         struct subscription *sub_temp;
291         int found = 0;
292
293         /* Find first matching subscription, exit if not found */
294
295         list_for_each_entry_safe(sub, sub_temp, &subscriber->subscription_list,
296                                  subscription_list) {
297                 if (!memcmp(s, &sub->evt.s, sizeof(struct tipc_subscr))) {
298                         found = 1;
299                         break;
300                 }
301         }
302         if (!found)
303                 return;
304
305         /* Cancel subscription timer (if used), then delete subscription */
306
307         if (sub->timeout != TIPC_WAIT_FOREVER) {
308                 sub->timeout = TIPC_WAIT_FOREVER;
309                 spin_unlock_bh(subscriber->lock);
310                 k_cancel_timer(&sub->timer);
311                 k_term_timer(&sub->timer);
312                 spin_lock_bh(subscriber->lock);
313         }
314         dbg("Cancel: removing sub %u,%u,%u from subscriber %x list\n",
315             sub->seq.type, sub->seq.lower, sub->seq.upper, subscriber);
316         subscr_del(sub);
317 }
318
319 /**
320  * subscr_subscribe - create subscription for subscriber
321  *
322  * Called with subscriber port locked.
323  */
324
325 static struct subscription *subscr_subscribe(struct tipc_subscr *s,
326                                              struct subscriber *subscriber)
327 {
328         struct subscription *sub;
329         int swap;
330
331         /* Determine subscriber's endianness */
332
333         swap = !(s->filter & (TIPC_SUB_PORTS | TIPC_SUB_SERVICE));
334
335         /* Detect & process a subscription cancellation request */
336
337         if (s->filter & htohl(TIPC_SUB_CANCEL, swap)) {
338                 s->filter &= ~htohl(TIPC_SUB_CANCEL, swap);
339                 subscr_cancel(s, subscriber);
340                 return NULL;
341         }
342
343         /* Refuse subscription if global limit exceeded */
344
345         if (atomic_read(&topsrv.subscription_count) >= tipc_max_subscriptions) {
346                 warn("Subscription rejected, subscription limit reached (%u)\n",
347                      tipc_max_subscriptions);
348                 subscr_terminate(subscriber);
349                 return NULL;
350         }
351
352         /* Allocate subscription object */
353
354         sub = kmalloc(sizeof(*sub), GFP_ATOMIC);
355         if (!sub) {
356                 warn("Subscription rejected, no memory\n");
357                 subscr_terminate(subscriber);
358                 return NULL;
359         }
360
361         /* Initialize subscription object */
362
363         sub->seq.type = htohl(s->seq.type, swap);
364         sub->seq.lower = htohl(s->seq.lower, swap);
365         sub->seq.upper = htohl(s->seq.upper, swap);
366         sub->timeout = htohl(s->timeout, swap);
367         sub->filter = htohl(s->filter, swap);
368         if ((!(sub->filter & TIPC_SUB_PORTS) ==
369              !(sub->filter & TIPC_SUB_SERVICE)) ||
370             (sub->seq.lower > sub->seq.upper)) {
371                 warn("Subscription rejected, illegal request\n");
372                 kfree(sub);
373                 subscr_terminate(subscriber);
374                 return NULL;
375         }
376         sub->event_cb = subscr_send_event;
377         INIT_LIST_HEAD(&sub->nameseq_list);
378         list_add(&sub->subscription_list, &subscriber->subscription_list);
379         sub->server_ref = subscriber->port_ref;
380         sub->swap = swap;
381         memcpy(&sub->evt.s, s, sizeof(struct tipc_subscr));
382         atomic_inc(&topsrv.subscription_count);
383         if (sub->timeout != TIPC_WAIT_FOREVER) {
384                 k_init_timer(&sub->timer,
385                              (Handler)subscr_timeout, (unsigned long)sub);
386                 k_start_timer(&sub->timer, sub->timeout);
387         }
388
389         return sub;
390 }
391
392 /**
393  * subscr_conn_shutdown_event - handle termination request from subscriber
394  *
395  * Called with subscriber's server port unlocked.
396  */
397
398 static void subscr_conn_shutdown_event(void *usr_handle,
399                                        u32 port_ref,
400                                        struct sk_buff **buf,
401                                        unsigned char const *data,
402                                        unsigned int size,
403                                        int reason)
404 {
405         struct subscriber *subscriber = usr_handle;
406         spinlock_t *subscriber_lock;
407
408         if (tipc_port_lock(port_ref) == NULL)
409                 return;
410
411         subscriber_lock = subscriber->lock;
412         subscr_terminate(subscriber);
413         spin_unlock_bh(subscriber_lock);
414 }
415
416 /**
417  * subscr_conn_msg_event - handle new subscription request from subscriber
418  *
419  * Called with subscriber's server port unlocked.
420  */
421
422 static void subscr_conn_msg_event(void *usr_handle,
423                                   u32 port_ref,
424                                   struct sk_buff **buf,
425                                   const unchar *data,
426                                   u32 size)
427 {
428         struct subscriber *subscriber = usr_handle;
429         spinlock_t *subscriber_lock;
430         struct subscription *sub;
431
432         /*
433          * Lock subscriber's server port (& make a local copy of lock pointer,
434          * in case subscriber is deleted while processing subscription request)
435          */
436
437         if (tipc_port_lock(port_ref) == NULL)
438                 return;
439
440         subscriber_lock = subscriber->lock;
441
442         if (size != sizeof(struct tipc_subscr)) {
443                 subscr_terminate(subscriber);
444                 spin_unlock_bh(subscriber_lock);
445         } else {
446                 sub = subscr_subscribe((struct tipc_subscr *)data, subscriber);
447                 spin_unlock_bh(subscriber_lock);
448                 if (sub != NULL) {
449
450                         /*
451                          * We must release the server port lock before adding a
452                          * subscription to the name table since TIPC needs to be
453                          * able to (re)acquire the port lock if an event message
454                          * issued by the subscription process is rejected and
455                          * returned.  The subscription cannot be deleted while
456                          * it is being added to the name table because:
457                          * a) the single-threading of the native API port code
458                          *    ensures the subscription cannot be cancelled and
459                          *    the subscriber connection cannot be broken, and
460                          * b) the name table lock ensures the subscription
461                          *    timeout code cannot delete the subscription,
462                          * so the subscription object is still protected.
463                          */
464
465                         tipc_nametbl_subscribe(sub);
466                 }
467         }
468 }
469
470 /**
471  * subscr_named_msg_event - handle request to establish a new subscriber
472  */
473
474 static void subscr_named_msg_event(void *usr_handle,
475                                    u32 port_ref,
476                                    struct sk_buff **buf,
477                                    const unchar *data,
478                                    u32 size,
479                                    u32 importance,
480                                    struct tipc_portid const *orig,
481                                    struct tipc_name_seq const *dest)
482 {
483         static struct iovec msg_sect = {NULL, 0};
484
485         struct subscriber *subscriber;
486         u32 server_port_ref;
487
488         /* Create subscriber object */
489
490         subscriber = kzalloc(sizeof(struct subscriber), GFP_ATOMIC);
491         if (subscriber == NULL) {
492                 warn("Subscriber rejected, no memory\n");
493                 return;
494         }
495         INIT_LIST_HEAD(&subscriber->subscription_list);
496         INIT_LIST_HEAD(&subscriber->subscriber_list);
497
498         /* Create server port & establish connection to subscriber */
499
500         tipc_createport(topsrv.user_ref,
501                         subscriber,
502                         importance,
503                         NULL,
504                         NULL,
505                         subscr_conn_shutdown_event,
506                         NULL,
507                         NULL,
508                         subscr_conn_msg_event,
509                         NULL,
510                         &subscriber->port_ref);
511         if (subscriber->port_ref == 0) {
512                 warn("Subscriber rejected, unable to create port\n");
513                 kfree(subscriber);
514                 return;
515         }
516         tipc_connect2port(subscriber->port_ref, orig);
517
518         /* Lock server port (& save lock address for future use) */
519
520         subscriber->lock = tipc_port_lock(subscriber->port_ref)->publ.lock;
521
522         /* Add subscriber to topology server's subscriber list */
523
524         spin_lock_bh(&topsrv.lock);
525         list_add(&subscriber->subscriber_list, &topsrv.subscriber_list);
526         spin_unlock_bh(&topsrv.lock);
527
528         /* Unlock server port */
529
530         server_port_ref = subscriber->port_ref;
531         spin_unlock_bh(subscriber->lock);
532
533         /* Send an ACK- to complete connection handshaking */
534
535         tipc_send(server_port_ref, 1, &msg_sect);
536
537         /* Handle optional subscription request */
538
539         if (size != 0) {
540                 subscr_conn_msg_event(subscriber, server_port_ref,
541                                       buf, data, size);
542         }
543 }
544
545 int tipc_subscr_start(void)
546 {
547         struct tipc_name_seq seq = {TIPC_TOP_SRV, TIPC_TOP_SRV, TIPC_TOP_SRV};
548         int res = -1;
549
550         memset(&topsrv, 0, sizeof (topsrv));
551         spin_lock_init(&topsrv.lock);
552         INIT_LIST_HEAD(&topsrv.subscriber_list);
553
554         spin_lock_bh(&topsrv.lock);
555         res = tipc_attach(&topsrv.user_ref, NULL, NULL);
556         if (res) {
557                 spin_unlock_bh(&topsrv.lock);
558                 return res;
559         }
560
561         res = tipc_createport(topsrv.user_ref,
562                               NULL,
563                               TIPC_CRITICAL_IMPORTANCE,
564                               NULL,
565                               NULL,
566                               NULL,
567                               NULL,
568                               subscr_named_msg_event,
569                               NULL,
570                               NULL,
571                               &topsrv.setup_port);
572         if (res)
573                 goto failed;
574
575         res = tipc_nametbl_publish_rsv(topsrv.setup_port, TIPC_NODE_SCOPE, &seq);
576         if (res)
577                 goto failed;
578
579         spin_unlock_bh(&topsrv.lock);
580         return 0;
581
582 failed:
583         err("Failed to create subscription service\n");
584         tipc_detach(topsrv.user_ref);
585         topsrv.user_ref = 0;
586         spin_unlock_bh(&topsrv.lock);
587         return res;
588 }
589
590 void tipc_subscr_stop(void)
591 {
592         struct subscriber *subscriber;
593         struct subscriber *subscriber_temp;
594         spinlock_t *subscriber_lock;
595
596         if (topsrv.user_ref) {
597                 tipc_deleteport(topsrv.setup_port);
598                 list_for_each_entry_safe(subscriber, subscriber_temp,
599                                          &topsrv.subscriber_list,
600                                          subscriber_list) {
601                         subscriber_lock = subscriber->lock;
602                         spin_lock_bh(subscriber_lock);
603                         subscr_terminate(subscriber);
604                         spin_unlock_bh(subscriber_lock);
605                 }
606                 tipc_detach(topsrv.user_ref);
607                 topsrv.user_ref = 0;
608         }
609 }