subscr.c revision 0da974f4f303a6842516b764507e3c0a03f41e5a 
1/*
2 * net/tipc/subscr.c: TIPC subscription service
3 *
4 * Copyright (c) 2000-2006, Ericsson AB
5 * Copyright (c) 2005, 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 "subscr.h"
40#include "name_table.h"
41#include "ref.h"
42
43/**
44 * struct subscriber - TIPC network topology subscriber
45 * @ref: object reference to subscriber object itself
46 * @lock: pointer to spinlock controlling access to subscriber object
47 * @subscriber_list: adjacent subscribers in top. server's list of subscribers
48 * @subscription_list: list of subscription objects for this subscriber
49 * @port_ref: object reference to port used to communicate with subscriber
50 * @swap: indicates if subscriber uses opposite endianness in its messages
51 */
52
53struct subscriber {
54	u32 ref;
55        spinlock_t *lock;
56	struct list_head subscriber_list;
57	struct list_head subscription_list;
58	u32 port_ref;
59	int swap;
60};
61
62/**
63 * struct top_srv - TIPC network topology subscription service
64 * @user_ref: TIPC userid of subscription service
65 * @setup_port: reference to TIPC port that handles subscription requests
66 * @subscription_count: number of active subscriptions (not subscribers!)
67 * @subscriber_list: list of ports subscribing to service
68 * @lock: spinlock govering access to subscriber list
69 */
70
71struct top_srv {
72	u32 user_ref;
73	u32 setup_port;
74	atomic_t subscription_count;
75	struct list_head subscriber_list;
76	spinlock_t lock;
77};
78
79static struct top_srv topsrv = { 0 };
80
81/**
82 * htohl - convert value to endianness used by destination
83 * @in: value to convert
84 * @swap: non-zero if endianness must be reversed
85 *
86 * Returns converted value
87 */
88
89static u32 htohl(u32 in, int swap)
90{
91	char *c = (char *)∈
92
93	return swap ? ((c[3] << 3) + (c[2] << 2) + (c[1] << 1) + c[0]) : in;
94}
95
96/**
97 * subscr_send_event - send a message containing a tipc_event to the subscriber
98 */
99
100static void subscr_send_event(struct subscription *sub,
101			      u32 found_lower,
102			      u32 found_upper,
103			      u32 event,
104			      u32 port_ref,
105			      u32 node)
106{
107	struct iovec msg_sect;
108
109	msg_sect.iov_base = (void *)&sub->evt;
110	msg_sect.iov_len = sizeof(struct tipc_event);
111
112	sub->evt.event = htohl(event, sub->owner->swap);
113	sub->evt.found_lower = htohl(found_lower, sub->owner->swap);
114	sub->evt.found_upper = htohl(found_upper, sub->owner->swap);
115	sub->evt.port.ref = htohl(port_ref, sub->owner->swap);
116	sub->evt.port.node = htohl(node, sub->owner->swap);
117	tipc_send(sub->owner->port_ref, 1, &msg_sect);
118}
119
120/**
121 * tipc_subscr_overlap - test for subscription overlap with the given values
122 *
123 * Returns 1 if there is overlap, otherwise 0.
124 */
125
126int tipc_subscr_overlap(struct subscription *sub,
127			u32 found_lower,
128			u32 found_upper)
129
130{
131	if (found_lower < sub->seq.lower)
132		found_lower = sub->seq.lower;
133	if (found_upper > sub->seq.upper)
134		found_upper = sub->seq.upper;
135	if (found_lower > found_upper)
136		return 0;
137	return 1;
138}
139
140/**
141 * tipc_subscr_report_overlap - issue event if there is subscription overlap
142 *
143 * Protected by nameseq.lock in name_table.c
144 */
145
146void tipc_subscr_report_overlap(struct subscription *sub,
147				u32 found_lower,
148				u32 found_upper,
149				u32 event,
150				u32 port_ref,
151				u32 node,
152				int must)
153{
154	dbg("Rep overlap %u:%u,%u<->%u,%u\n", sub->seq.type, sub->seq.lower,
155	    sub->seq.upper, found_lower, found_upper);
156	if (!tipc_subscr_overlap(sub, found_lower, found_upper))
157		return;
158	if (!must && (sub->filter != TIPC_SUB_PORTS))
159		return;
160	subscr_send_event(sub, found_lower, found_upper, event, port_ref, node);
161}
162
163/**
164 * subscr_timeout - subscription timeout has occurred
165 */
166
167static void subscr_timeout(struct subscription *sub)
168{
169	struct subscriber *subscriber;
170	u32 subscriber_ref;
171
172	/* Validate subscriber reference (in case subscriber is terminating) */
173
174	subscriber_ref = sub->owner->ref;
175	subscriber = (struct subscriber *)tipc_ref_lock(subscriber_ref);
176	if (subscriber == NULL)
177		return;
178
179	/* Unlink subscription from name table */
180
181	tipc_nametbl_unsubscribe(sub);
182
183	/* Notify subscriber of timeout, then unlink subscription */
184
185	subscr_send_event(sub,
186			  sub->evt.s.seq.lower,
187			  sub->evt.s.seq.upper,
188			  TIPC_SUBSCR_TIMEOUT,
189			  0,
190			  0);
191	list_del(&sub->subscription_list);
192
193	/* Now destroy subscription */
194
195	tipc_ref_unlock(subscriber_ref);
196	k_term_timer(&sub->timer);
197	kfree(sub);
198	atomic_dec(&topsrv.subscription_count);
199}
200
201/**
202 * subscr_terminate - terminate communication with a subscriber
203 *
204 * Called with subscriber locked.  Routine must temporarily release this lock
205 * to enable subscription timeout routine(s) to finish without deadlocking;
206 * the lock is then reclaimed to allow caller to release it upon return.
207 * (This should work even in the unlikely event some other thread creates
208 * a new object reference in the interim that uses this lock; this routine will
209 * simply wait for it to be released, then claim it.)
210 */
211
212static void subscr_terminate(struct subscriber *subscriber)
213{
214	struct subscription *sub;
215	struct subscription *sub_temp;
216
217	/* Invalidate subscriber reference */
218
219	tipc_ref_discard(subscriber->ref);
220	spin_unlock_bh(subscriber->lock);
221
222	/* Destroy any existing subscriptions for subscriber */
223
224	list_for_each_entry_safe(sub, sub_temp, &subscriber->subscription_list,
225				 subscription_list) {
226		if (sub->timeout != TIPC_WAIT_FOREVER) {
227			k_cancel_timer(&sub->timer);
228			k_term_timer(&sub->timer);
229		}
230		tipc_nametbl_unsubscribe(sub);
231		list_del(&sub->subscription_list);
232		dbg("Term: Removed sub %u,%u,%u from subscriber %x list\n",
233		    sub->seq.type, sub->seq.lower, sub->seq.upper, subscriber);
234		kfree(sub);
235		atomic_dec(&topsrv.subscription_count);
236	}
237
238	/* Sever connection to subscriber */
239
240	tipc_shutdown(subscriber->port_ref);
241	tipc_deleteport(subscriber->port_ref);
242
243	/* Remove subscriber from topology server's subscriber list */
244
245	spin_lock_bh(&topsrv.lock);
246	list_del(&subscriber->subscriber_list);
247	spin_unlock_bh(&topsrv.lock);
248
249	/* Now destroy subscriber */
250
251	spin_lock_bh(subscriber->lock);
252	kfree(subscriber);
253}
254
255/**
256 * subscr_subscribe - create subscription for subscriber
257 *
258 * Called with subscriber locked
259 */
260
261static void subscr_subscribe(struct tipc_subscr *s,
262			     struct subscriber *subscriber)
263{
264	struct subscription *sub;
265
266	/* Refuse subscription if global limit exceeded */
267
268	if (atomic_read(&topsrv.subscription_count) >= tipc_max_subscriptions) {
269		warn("Subscription rejected, subscription limit reached (%u)\n",
270		     tipc_max_subscriptions);
271		subscr_terminate(subscriber);
272		return;
273	}
274
275	/* Allocate subscription object */
276
277	sub = kmalloc(sizeof(*sub), GFP_ATOMIC);
278	if (!sub) {
279		warn("Subscription rejected, no memory\n");
280		subscr_terminate(subscriber);
281		return;
282	}
283
284	/* Determine/update subscriber's endianness */
285
286	if ((s->filter == TIPC_SUB_PORTS) || (s->filter == TIPC_SUB_SERVICE))
287		subscriber->swap = 0;
288	else
289		subscriber->swap = 1;
290
291	/* Initialize subscription object */
292
293	memset(sub, 0, sizeof(*sub));
294	sub->seq.type = htohl(s->seq.type, subscriber->swap);
295	sub->seq.lower = htohl(s->seq.lower, subscriber->swap);
296	sub->seq.upper = htohl(s->seq.upper, subscriber->swap);
297	sub->timeout = htohl(s->timeout, subscriber->swap);
298	sub->filter = htohl(s->filter, subscriber->swap);
299	if ((((sub->filter != TIPC_SUB_PORTS)
300	      && (sub->filter != TIPC_SUB_SERVICE)))
301	    || (sub->seq.lower > sub->seq.upper)) {
302		warn("Subscription rejected, illegal request\n");
303		kfree(sub);
304		subscr_terminate(subscriber);
305		return;
306	}
307	memcpy(&sub->evt.s, s, sizeof(struct tipc_subscr));
308	INIT_LIST_HEAD(&sub->subscription_list);
309	INIT_LIST_HEAD(&sub->nameseq_list);
310	list_add(&sub->subscription_list, &subscriber->subscription_list);
311	atomic_inc(&topsrv.subscription_count);
312	if (sub->timeout != TIPC_WAIT_FOREVER) {
313		k_init_timer(&sub->timer,
314			     (Handler)subscr_timeout, (unsigned long)sub);
315		k_start_timer(&sub->timer, sub->timeout);
316	}
317	sub->owner = subscriber;
318	tipc_nametbl_subscribe(sub);
319}
320
321/**
322 * subscr_conn_shutdown_event - handle termination request from subscriber
323 */
324
325static void subscr_conn_shutdown_event(void *usr_handle,
326				       u32 portref,
327				       struct sk_buff **buf,
328				       unsigned char const *data,
329				       unsigned int size,
330				       int reason)
331{
332	struct subscriber *subscriber;
333	spinlock_t *subscriber_lock;
334
335	subscriber = tipc_ref_lock((u32)(unsigned long)usr_handle);
336	if (subscriber == NULL)
337		return;
338
339	subscriber_lock = subscriber->lock;
340	subscr_terminate(subscriber);
341	spin_unlock_bh(subscriber_lock);
342}
343
344/**
345 * subscr_conn_msg_event - handle new subscription request from subscriber
346 */
347
348static void subscr_conn_msg_event(void *usr_handle,
349				  u32 port_ref,
350				  struct sk_buff **buf,
351				  const unchar *data,
352				  u32 size)
353{
354	struct subscriber *subscriber;
355	spinlock_t *subscriber_lock;
356
357	subscriber = tipc_ref_lock((u32)(unsigned long)usr_handle);
358	if (subscriber == NULL)
359		return;
360
361	subscriber_lock = subscriber->lock;
362	if (size != sizeof(struct tipc_subscr))
363		subscr_terminate(subscriber);
364	else
365		subscr_subscribe((struct tipc_subscr *)data, subscriber);
366
367	spin_unlock_bh(subscriber_lock);
368}
369
370/**
371 * subscr_named_msg_event - handle request to establish a new subscriber
372 */
373
374static void subscr_named_msg_event(void *usr_handle,
375				   u32 port_ref,
376				   struct sk_buff **buf,
377				   const unchar *data,
378				   u32 size,
379				   u32 importance,
380				   struct tipc_portid const *orig,
381				   struct tipc_name_seq const *dest)
382{
383	struct subscriber *subscriber;
384	struct iovec msg_sect = {NULL, 0};
385	spinlock_t *subscriber_lock;
386
387	dbg("subscr_named_msg_event: orig = %x own = %x,\n",
388	    orig->node, tipc_own_addr);
389	if (size && (size != sizeof(struct tipc_subscr))) {
390		warn("Subscriber rejected, invalid subscription size\n");
391		return;
392	}
393
394	/* Create subscriber object */
395
396	subscriber = kzalloc(sizeof(struct subscriber), GFP_ATOMIC);
397	if (subscriber == NULL) {
398		warn("Subscriber rejected, no memory\n");
399		return;
400	}
401	INIT_LIST_HEAD(&subscriber->subscription_list);
402	INIT_LIST_HEAD(&subscriber->subscriber_list);
403	subscriber->ref = tipc_ref_acquire(subscriber, &subscriber->lock);
404	if (subscriber->ref == 0) {
405		warn("Subscriber rejected, reference table exhausted\n");
406		kfree(subscriber);
407		return;
408	}
409
410	/* Establish a connection to subscriber */
411
412	tipc_createport(topsrv.user_ref,
413			(void *)(unsigned long)subscriber->ref,
414			importance,
415			NULL,
416			NULL,
417			subscr_conn_shutdown_event,
418			NULL,
419			NULL,
420			subscr_conn_msg_event,
421			NULL,
422			&subscriber->port_ref);
423	if (subscriber->port_ref == 0) {
424		warn("Subscriber rejected, unable to create port\n");
425		tipc_ref_discard(subscriber->ref);
426		kfree(subscriber);
427		return;
428	}
429	tipc_connect2port(subscriber->port_ref, orig);
430
431
432	/* Add subscriber to topology server's subscriber list */
433
434	tipc_ref_lock(subscriber->ref);
435	spin_lock_bh(&topsrv.lock);
436	list_add(&subscriber->subscriber_list, &topsrv.subscriber_list);
437	spin_unlock_bh(&topsrv.lock);
438
439	/*
440	 * Subscribe now if message contains a subscription,
441	 * otherwise send an empty response to complete connection handshaking
442	 */
443
444	subscriber_lock = subscriber->lock;
445	if (size)
446		subscr_subscribe((struct tipc_subscr *)data, subscriber);
447	else
448		tipc_send(subscriber->port_ref, 1, &msg_sect);
449
450	spin_unlock_bh(subscriber_lock);
451}
452
453int tipc_subscr_start(void)
454{
455	struct tipc_name_seq seq = {TIPC_TOP_SRV, TIPC_TOP_SRV, TIPC_TOP_SRV};
456	int res = -1;
457
458	memset(&topsrv, 0, sizeof (topsrv));
459	spin_lock_init(&topsrv.lock);
460	INIT_LIST_HEAD(&topsrv.subscriber_list);
461
462	spin_lock_bh(&topsrv.lock);
463	res = tipc_attach(&topsrv.user_ref, NULL, NULL);
464	if (res) {
465		spin_unlock_bh(&topsrv.lock);
466		return res;
467	}
468
469 	res = tipc_createport(topsrv.user_ref,
470 			      NULL,
471 			      TIPC_CRITICAL_IMPORTANCE,
472 			      NULL,
473 			      NULL,
474 			      NULL,
475 			      NULL,
476 			      subscr_named_msg_event,
477 			      NULL,
478 			      NULL,
479 			      &topsrv.setup_port);
480 	if (res)
481		goto failed;
482
483 	res = tipc_nametbl_publish_rsv(topsrv.setup_port, TIPC_NODE_SCOPE, &seq);
484 	if (res)
485		goto failed;
486
487	spin_unlock_bh(&topsrv.lock);
488	return 0;
489
490failed:
491	err("Failed to create subscription service\n");
492	tipc_detach(topsrv.user_ref);
493	topsrv.user_ref = 0;
494	spin_unlock_bh(&topsrv.lock);
495	return res;
496}
497
498void tipc_subscr_stop(void)
499{
500	struct subscriber *subscriber;
501	struct subscriber *subscriber_temp;
502	spinlock_t *subscriber_lock;
503
504	if (topsrv.user_ref) {
505		tipc_deleteport(topsrv.setup_port);
506		list_for_each_entry_safe(subscriber, subscriber_temp,
507					 &topsrv.subscriber_list,
508					 subscriber_list) {
509			tipc_ref_lock(subscriber->ref);
510			subscriber_lock = subscriber->lock;
511			subscr_terminate(subscriber);
512			spin_unlock_bh(subscriber_lock);
513		}
514		tipc_detach(topsrv.user_ref);
515		topsrv.user_ref = 0;
516	}
517}
518
519
520int tipc_ispublished(struct tipc_name const *name)
521{
522	u32 domain = 0;
523
524	return(tipc_nametbl_translate(name->type, name->instance,&domain) != 0);
525}
526
527