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 53 struct 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 71 struct 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 79 static 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 89 static 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 100 static 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 126 int 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 146 void 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 167 static 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 212 static 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 261 static 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("Failed: max %u subscriptions\n", tipc_max_subscriptions); 270 subscr_terminate(subscriber); 271 return; 272 } 273 274 /* Allocate subscription object */ 275 276 sub = kmalloc(sizeof(*sub), GFP_ATOMIC); 277 if (sub == NULL) { 278 warn("Memory squeeze; ignoring subscription\n"); 279 subscr_terminate(subscriber); 280 return; 281 } 282 283 /* Determine/update subscriber's endianness */ 284 285 if ((s->filter == TIPC_SUB_PORTS) || (s->filter == TIPC_SUB_SERVICE)) 286 subscriber->swap = 0; 287 else 288 subscriber->swap = 1; 289 290 /* Initialize subscription object */ 291 292 memset(sub, 0, sizeof(*sub)); 293 sub->seq.type = htohl(s->seq.type, subscriber->swap); 294 sub->seq.lower = htohl(s->seq.lower, subscriber->swap); 295 sub->seq.upper = htohl(s->seq.upper, subscriber->swap); 296 sub->timeout = htohl(s->timeout, subscriber->swap); 297 sub->filter = htohl(s->filter, subscriber->swap); 298 if ((((sub->filter != TIPC_SUB_PORTS) 299 && (sub->filter != TIPC_SUB_SERVICE))) 300 || (sub->seq.lower > sub->seq.upper)) { 301 warn("Rejecting illegal subscription %u,%u,%u\n", 302 sub->seq.type, sub->seq.lower, sub->seq.upper); 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 325 static 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 348 static 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 374 static 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("Received tipc_subscr of invalid size\n"); 391 return; 392 } 393 394 /* Create subscriber object */ 395 396 subscriber = kmalloc(sizeof(struct subscriber), GFP_ATOMIC); 397 if (subscriber == NULL) { 398 warn("Memory squeeze; ignoring subscriber setup\n"); 399 return; 400 } 401 memset(subscriber, 0, sizeof(struct subscriber)); 402 INIT_LIST_HEAD(&subscriber->subscription_list); 403 INIT_LIST_HEAD(&subscriber->subscriber_list); 404 subscriber->ref = tipc_ref_acquire(subscriber, &subscriber->lock); 405 if (subscriber->ref == 0) { 406 warn("Failed to acquire subscriber reference\n"); 407 kfree(subscriber); 408 return; 409 } 410 411 /* Establish a connection to subscriber */ 412 413 tipc_createport(topsrv.user_ref, 414 (void *)(unsigned long)subscriber->ref, 415 importance, 416 NULL, 417 NULL, 418 subscr_conn_shutdown_event, 419 NULL, 420 NULL, 421 subscr_conn_msg_event, 422 NULL, 423 &subscriber->port_ref); 424 if (subscriber->port_ref == 0) { 425 warn("Memory squeeze; failed to create subscription port\n"); 426 tipc_ref_discard(subscriber->ref); 427 kfree(subscriber); 428 return; 429 } 430 tipc_connect2port(subscriber->port_ref, orig); 431 432 433 /* Add subscriber to topology server's subscriber list */ 434 435 tipc_ref_lock(subscriber->ref); 436 spin_lock_bh(&topsrv.lock); 437 list_add(&subscriber->subscriber_list, &topsrv.subscriber_list); 438 spin_unlock_bh(&topsrv.lock); 439 440 /* 441 * Subscribe now if message contains a subscription, 442 * otherwise send an empty response to complete connection handshaking 443 */ 444 445 subscriber_lock = subscriber->lock; 446 if (size) 447 subscr_subscribe((struct tipc_subscr *)data, subscriber); 448 else 449 tipc_send(subscriber->port_ref, 1, &msg_sect); 450 451 spin_unlock_bh(subscriber_lock); 452 } 453 454 int tipc_subscr_start(void) 455 { 456 struct tipc_name_seq seq = {TIPC_TOP_SRV, TIPC_TOP_SRV, TIPC_TOP_SRV}; 457 int res = -1; 458 459 memset(&topsrv, 0, sizeof (topsrv)); 460 topsrv.lock = SPIN_LOCK_UNLOCKED; 461 INIT_LIST_HEAD(&topsrv.subscriber_list); 462 463 spin_lock_bh(&topsrv.lock); 464 res = tipc_attach(&topsrv.user_ref, NULL, NULL); 465 if (res) { 466 spin_unlock_bh(&topsrv.lock); 467 return res; 468 } 469 470 res = tipc_createport(topsrv.user_ref, 471 NULL, 472 TIPC_CRITICAL_IMPORTANCE, 473 NULL, 474 NULL, 475 NULL, 476 NULL, 477 subscr_named_msg_event, 478 NULL, 479 NULL, 480 &topsrv.setup_port); 481 if (res) 482 goto failed; 483 484 res = tipc_nametbl_publish_rsv(topsrv.setup_port, TIPC_NODE_SCOPE, &seq); 485 if (res) 486 goto failed; 487 488 spin_unlock_bh(&topsrv.lock); 489 return 0; 490 491 failed: 492 err("Failed to create subscription service\n"); 493 tipc_detach(topsrv.user_ref); 494 topsrv.user_ref = 0; 495 spin_unlock_bh(&topsrv.lock); 496 return res; 497 } 498 499 void tipc_subscr_stop(void) 500 { 501 struct subscriber *subscriber; 502 struct subscriber *subscriber_temp; 503 spinlock_t *subscriber_lock; 504 505 if (topsrv.user_ref) { 506 tipc_deleteport(topsrv.setup_port); 507 list_for_each_entry_safe(subscriber, subscriber_temp, 508 &topsrv.subscriber_list, 509 subscriber_list) { 510 tipc_ref_lock(subscriber->ref); 511 subscriber_lock = subscriber->lock; 512 subscr_terminate(subscriber); 513 spin_unlock_bh(subscriber_lock); 514 } 515 tipc_detach(topsrv.user_ref); 516 topsrv.user_ref = 0; 517 } 518 } 519 520 521 int tipc_ispublished(struct tipc_name const *name) 522 { 523 u32 domain = 0; 524 525 return(tipc_nametbl_translate(name->type, name->instance,&domain) != 0); 526 } 527 528