2 * net/tipc/subscr.c: TIPC network topology service
4 * Copyright (c) 2000-2006, Ericsson AB
5 * Copyright (c) 2005-2007, Wind River Systems
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are met:
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.
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.
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.
38 #include "name_table.h"
43 * struct subscriber - TIPC network topology subscriber
44 * @port_ref: object reference to server port connecting to subscriber
45 * @lock: pointer to spinlock controlling access to subscriber's server port
46 * @subscriber_list: adjacent subscribers in top. server's list of subscribers
47 * @subscription_list: list of subscription objects for this subscriber
53 struct list_head subscriber_list
;
54 struct list_head subscription_list
;
58 * struct top_srv - TIPC network topology subscription service
59 * @user_ref: TIPC userid of subscription service
60 * @setup_port: reference to TIPC port that handles subscription requests
61 * @subscription_count: number of active subscriptions (not subscribers!)
62 * @subscriber_list: list of ports subscribing to service
63 * @lock: spinlock govering access to subscriber list
69 atomic_t subscription_count
;
70 struct list_head subscriber_list
;
74 static struct top_srv topsrv
= { 0 };
77 * htohl - convert value to endianness used by destination
78 * @in: value to convert
79 * @swap: non-zero if endianness must be reversed
81 * Returns converted value
84 static u32
htohl(u32 in
, int swap
)
86 return swap
? swab32(in
) : in
;
90 * subscr_send_event - send a message containing a tipc_event to the subscriber
92 * Note: Must not hold subscriber's server port lock, since tipc_send() will
93 * try to take the lock if the message is rejected and returned!
96 static void subscr_send_event(struct subscription
*sub
,
103 struct iovec msg_sect
;
105 msg_sect
.iov_base
= (void *)&sub
->evt
;
106 msg_sect
.iov_len
= sizeof(struct tipc_event
);
108 sub
->evt
.event
= htohl(event
, sub
->swap
);
109 sub
->evt
.found_lower
= htohl(found_lower
, sub
->swap
);
110 sub
->evt
.found_upper
= htohl(found_upper
, sub
->swap
);
111 sub
->evt
.port
.ref
= htohl(port_ref
, sub
->swap
);
112 sub
->evt
.port
.node
= htohl(node
, sub
->swap
);
113 tipc_send(sub
->server_ref
, 1, &msg_sect
);
117 * tipc_subscr_overlap - test for subscription overlap with the given values
119 * Returns 1 if there is overlap, otherwise 0.
122 int tipc_subscr_overlap(struct subscription
*sub
,
127 if (found_lower
< sub
->seq
.lower
)
128 found_lower
= sub
->seq
.lower
;
129 if (found_upper
> sub
->seq
.upper
)
130 found_upper
= sub
->seq
.upper
;
131 if (found_lower
> found_upper
)
137 * tipc_subscr_report_overlap - issue event if there is subscription overlap
139 * Protected by nameseq.lock in name_table.c
142 void tipc_subscr_report_overlap(struct subscription
*sub
,
150 if (!tipc_subscr_overlap(sub
, found_lower
, found_upper
))
152 if (!must
&& !(sub
->filter
& TIPC_SUB_PORTS
))
155 sub
->event_cb(sub
, found_lower
, found_upper
, event
, port_ref
, node
);
159 * subscr_timeout - subscription timeout has occurred
162 static void subscr_timeout(struct subscription
*sub
)
164 struct port
*server_port
;
166 /* Validate server port reference (in case subscriber is terminating) */
168 server_port
= tipc_port_lock(sub
->server_ref
);
169 if (server_port
== NULL
)
172 /* Validate timeout (in case subscription is being cancelled) */
174 if (sub
->timeout
== TIPC_WAIT_FOREVER
) {
175 tipc_port_unlock(server_port
);
179 /* Unlink subscription from name table */
181 tipc_nametbl_unsubscribe(sub
);
183 /* Unlink subscription from subscriber */
185 list_del(&sub
->subscription_list
);
187 /* Release subscriber's server port */
189 tipc_port_unlock(server_port
);
191 /* Notify subscriber of timeout */
193 subscr_send_event(sub
, sub
->evt
.s
.seq
.lower
, sub
->evt
.s
.seq
.upper
,
194 TIPC_SUBSCR_TIMEOUT
, 0, 0);
196 /* Now destroy subscription */
198 k_term_timer(&sub
->timer
);
200 atomic_dec(&topsrv
.subscription_count
);
204 * subscr_del - delete a subscription within a subscription list
206 * Called with subscriber port locked.
209 static void subscr_del(struct subscription
*sub
)
211 tipc_nametbl_unsubscribe(sub
);
212 list_del(&sub
->subscription_list
);
214 atomic_dec(&topsrv
.subscription_count
);
218 * subscr_terminate - terminate communication with a subscriber
220 * Called with subscriber port locked. Routine must temporarily release lock
221 * to enable subscription timeout routine(s) to finish without deadlocking;
222 * the lock is then reclaimed to allow caller to release it upon return.
223 * (This should work even in the unlikely event some other thread creates
224 * a new object reference in the interim that uses this lock; this routine will
225 * simply wait for it to be released, then claim it.)
228 static void subscr_terminate(struct subscriber
*subscriber
)
231 struct subscription
*sub
;
232 struct subscription
*sub_temp
;
234 /* Invalidate subscriber reference */
236 port_ref
= subscriber
->port_ref
;
237 subscriber
->port_ref
= 0;
238 spin_unlock_bh(subscriber
->lock
);
240 /* Sever connection to subscriber */
242 tipc_shutdown(port_ref
);
243 tipc_deleteport(port_ref
);
245 /* Destroy any existing subscriptions for subscriber */
247 list_for_each_entry_safe(sub
, sub_temp
, &subscriber
->subscription_list
,
249 if (sub
->timeout
!= TIPC_WAIT_FOREVER
) {
250 k_cancel_timer(&sub
->timer
);
251 k_term_timer(&sub
->timer
);
253 dbg("Term: Removing sub %u,%u,%u from subscriber %x list\n",
254 sub
->seq
.type
, sub
->seq
.lower
, sub
->seq
.upper
, subscriber
);
258 /* Remove subscriber from topology server's subscriber list */
260 spin_lock_bh(&topsrv
.lock
);
261 list_del(&subscriber
->subscriber_list
);
262 spin_unlock_bh(&topsrv
.lock
);
264 /* Reclaim subscriber lock */
266 spin_lock_bh(subscriber
->lock
);
268 /* Now destroy subscriber */
274 * subscr_cancel - handle subscription cancellation request
276 * Called with subscriber port locked. Routine must temporarily release lock
277 * to enable the subscription timeout routine to finish without deadlocking;
278 * the lock is then reclaimed to allow caller to release it upon return.
280 * Note that fields of 's' use subscriber's endianness!
283 static void subscr_cancel(struct tipc_subscr
*s
,
284 struct subscriber
*subscriber
)
286 struct subscription
*sub
;
287 struct subscription
*sub_temp
;
290 /* Find first matching subscription, exit if not found */
292 list_for_each_entry_safe(sub
, sub_temp
, &subscriber
->subscription_list
,
294 if (!memcmp(s
, &sub
->evt
.s
, sizeof(struct tipc_subscr
))) {
302 /* Cancel subscription timer (if used), then delete subscription */
304 if (sub
->timeout
!= TIPC_WAIT_FOREVER
) {
305 sub
->timeout
= TIPC_WAIT_FOREVER
;
306 spin_unlock_bh(subscriber
->lock
);
307 k_cancel_timer(&sub
->timer
);
308 k_term_timer(&sub
->timer
);
309 spin_lock_bh(subscriber
->lock
);
311 dbg("Cancel: removing sub %u,%u,%u from subscriber %x list\n",
312 sub
->seq
.type
, sub
->seq
.lower
, sub
->seq
.upper
, subscriber
);
317 * subscr_subscribe - create subscription for subscriber
319 * Called with subscriber port locked.
322 static struct subscription
*subscr_subscribe(struct tipc_subscr
*s
,
323 struct subscriber
*subscriber
)
325 struct subscription
*sub
;
328 /* Determine subscriber's endianness */
330 swap
= !(s
->filter
& (TIPC_SUB_PORTS
| TIPC_SUB_SERVICE
));
332 /* Detect & process a subscription cancellation request */
334 if (s
->filter
& htohl(TIPC_SUB_CANCEL
, swap
)) {
335 s
->filter
&= ~htohl(TIPC_SUB_CANCEL
, swap
);
336 subscr_cancel(s
, subscriber
);
340 /* Refuse subscription if global limit exceeded */
342 if (atomic_read(&topsrv
.subscription_count
) >= tipc_max_subscriptions
) {
343 warn("Subscription rejected, subscription limit reached (%u)\n",
344 tipc_max_subscriptions
);
345 subscr_terminate(subscriber
);
349 /* Allocate subscription object */
351 sub
= kmalloc(sizeof(*sub
), GFP_ATOMIC
);
353 warn("Subscription rejected, no memory\n");
354 subscr_terminate(subscriber
);
358 /* Initialize subscription object */
360 sub
->seq
.type
= htohl(s
->seq
.type
, swap
);
361 sub
->seq
.lower
= htohl(s
->seq
.lower
, swap
);
362 sub
->seq
.upper
= htohl(s
->seq
.upper
, swap
);
363 sub
->timeout
= htohl(s
->timeout
, swap
);
364 sub
->filter
= htohl(s
->filter
, swap
);
365 if ((!(sub
->filter
& TIPC_SUB_PORTS
) ==
366 !(sub
->filter
& TIPC_SUB_SERVICE
)) ||
367 (sub
->seq
.lower
> sub
->seq
.upper
)) {
368 warn("Subscription rejected, illegal request\n");
370 subscr_terminate(subscriber
);
373 sub
->event_cb
= subscr_send_event
;
374 INIT_LIST_HEAD(&sub
->nameseq_list
);
375 list_add(&sub
->subscription_list
, &subscriber
->subscription_list
);
376 sub
->server_ref
= subscriber
->port_ref
;
378 memcpy(&sub
->evt
.s
, s
, sizeof(struct tipc_subscr
));
379 atomic_inc(&topsrv
.subscription_count
);
380 if (sub
->timeout
!= TIPC_WAIT_FOREVER
) {
381 k_init_timer(&sub
->timer
,
382 (Handler
)subscr_timeout
, (unsigned long)sub
);
383 k_start_timer(&sub
->timer
, sub
->timeout
);
390 * subscr_conn_shutdown_event - handle termination request from subscriber
392 * Called with subscriber's server port unlocked.
395 static void subscr_conn_shutdown_event(void *usr_handle
,
397 struct sk_buff
**buf
,
398 unsigned char const *data
,
402 struct subscriber
*subscriber
= usr_handle
;
403 spinlock_t
*subscriber_lock
;
405 if (tipc_port_lock(port_ref
) == NULL
)
408 subscriber_lock
= subscriber
->lock
;
409 subscr_terminate(subscriber
);
410 spin_unlock_bh(subscriber_lock
);
414 * subscr_conn_msg_event - handle new subscription request from subscriber
416 * Called with subscriber's server port unlocked.
419 static void subscr_conn_msg_event(void *usr_handle
,
421 struct sk_buff
**buf
,
425 struct subscriber
*subscriber
= usr_handle
;
426 spinlock_t
*subscriber_lock
;
427 struct subscription
*sub
;
430 * Lock subscriber's server port (& make a local copy of lock pointer,
431 * in case subscriber is deleted while processing subscription request)
434 if (tipc_port_lock(port_ref
) == NULL
)
437 subscriber_lock
= subscriber
->lock
;
439 if (size
!= sizeof(struct tipc_subscr
)) {
440 subscr_terminate(subscriber
);
441 spin_unlock_bh(subscriber_lock
);
443 sub
= subscr_subscribe((struct tipc_subscr
*)data
, subscriber
);
444 spin_unlock_bh(subscriber_lock
);
448 * We must release the server port lock before adding a
449 * subscription to the name table since TIPC needs to be
450 * able to (re)acquire the port lock if an event message
451 * issued by the subscription process is rejected and
452 * returned. The subscription cannot be deleted while
453 * it is being added to the name table because:
454 * a) the single-threading of the native API port code
455 * ensures the subscription cannot be cancelled and
456 * the subscriber connection cannot be broken, and
457 * b) the name table lock ensures the subscription
458 * timeout code cannot delete the subscription,
459 * so the subscription object is still protected.
462 tipc_nametbl_subscribe(sub
);
468 * subscr_named_msg_event - handle request to establish a new subscriber
471 static void subscr_named_msg_event(void *usr_handle
,
473 struct sk_buff
**buf
,
477 struct tipc_portid
const *orig
,
478 struct tipc_name_seq
const *dest
)
480 static struct iovec msg_sect
= {NULL
, 0};
482 struct subscriber
*subscriber
;
485 /* Create subscriber object */
487 subscriber
= kzalloc(sizeof(struct subscriber
), GFP_ATOMIC
);
488 if (subscriber
== NULL
) {
489 warn("Subscriber rejected, no memory\n");
492 INIT_LIST_HEAD(&subscriber
->subscription_list
);
493 INIT_LIST_HEAD(&subscriber
->subscriber_list
);
495 /* Create server port & establish connection to subscriber */
497 tipc_createport(topsrv
.user_ref
,
502 subscr_conn_shutdown_event
,
505 subscr_conn_msg_event
,
507 &subscriber
->port_ref
);
508 if (subscriber
->port_ref
== 0) {
509 warn("Subscriber rejected, unable to create port\n");
513 tipc_connect2port(subscriber
->port_ref
, orig
);
515 /* Lock server port (& save lock address for future use) */
517 subscriber
->lock
= tipc_port_lock(subscriber
->port_ref
)->publ
.lock
;
519 /* Add subscriber to topology server's subscriber list */
521 spin_lock_bh(&topsrv
.lock
);
522 list_add(&subscriber
->subscriber_list
, &topsrv
.subscriber_list
);
523 spin_unlock_bh(&topsrv
.lock
);
525 /* Unlock server port */
527 server_port_ref
= subscriber
->port_ref
;
528 spin_unlock_bh(subscriber
->lock
);
530 /* Send an ACK- to complete connection handshaking */
532 tipc_send(server_port_ref
, 1, &msg_sect
);
534 /* Handle optional subscription request */
537 subscr_conn_msg_event(subscriber
, server_port_ref
,
542 int tipc_subscr_start(void)
544 struct tipc_name_seq seq
= {TIPC_TOP_SRV
, TIPC_TOP_SRV
, TIPC_TOP_SRV
};
547 memset(&topsrv
, 0, sizeof (topsrv
));
548 spin_lock_init(&topsrv
.lock
);
549 INIT_LIST_HEAD(&topsrv
.subscriber_list
);
551 spin_lock_bh(&topsrv
.lock
);
552 res
= tipc_attach(&topsrv
.user_ref
);
554 spin_unlock_bh(&topsrv
.lock
);
558 res
= tipc_createport(topsrv
.user_ref
,
560 TIPC_CRITICAL_IMPORTANCE
,
565 subscr_named_msg_event
,
572 res
= tipc_nametbl_publish_rsv(topsrv
.setup_port
, TIPC_NODE_SCOPE
, &seq
);
576 spin_unlock_bh(&topsrv
.lock
);
580 err("Failed to create subscription service\n");
581 tipc_detach(topsrv
.user_ref
);
583 spin_unlock_bh(&topsrv
.lock
);
587 void tipc_subscr_stop(void)
589 struct subscriber
*subscriber
;
590 struct subscriber
*subscriber_temp
;
591 spinlock_t
*subscriber_lock
;
593 if (topsrv
.user_ref
) {
594 tipc_deleteport(topsrv
.setup_port
);
595 list_for_each_entry_safe(subscriber
, subscriber_temp
,
596 &topsrv
.subscriber_list
,
598 subscriber_lock
= subscriber
->lock
;
599 spin_lock_bh(subscriber_lock
);
600 subscr_terminate(subscriber
);
601 spin_unlock_bh(subscriber_lock
);
603 tipc_detach(topsrv
.user_ref
);