subscr.c revision 13a2e89873506d64d7e52f17b571da371a3e25a4
1/* 2 * net/tipc/subscr.c: TIPC network topology service 3 * 4 * Copyright (c) 2000-2006, Ericsson AB 5 * Copyright (c) 2005-2007, 2010-2013, 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 "name_table.h" 39#include "port.h" 40#include "subscr.h" 41 42/** 43 * struct tipc_subscriber - TIPC network topology subscriber 44 * @conid: connection identifier to server connecting to subscriber 45 * @lock: controll access to subscriber 46 * @subscription_list: list of subscription objects for this subscriber 47 */ 48struct tipc_subscriber { 49 int conid; 50 spinlock_t lock; 51 struct list_head subscription_list; 52}; 53 54static void subscr_conn_msg_event(int conid, struct sockaddr_tipc *addr, 55 void *usr_data, void *buf, size_t len); 56static void *subscr_named_msg_event(int conid); 57static void subscr_conn_shutdown_event(int conid, void *usr_data); 58 59static atomic_t subscription_count = ATOMIC_INIT(0); 60 61static struct sockaddr_tipc topsrv_addr __read_mostly = { 62 .family = AF_TIPC, 63 .addrtype = TIPC_ADDR_NAMESEQ, 64 .addr.nameseq.type = TIPC_TOP_SRV, 65 .addr.nameseq.lower = TIPC_TOP_SRV, 66 .addr.nameseq.upper = TIPC_TOP_SRV, 67 .scope = TIPC_NODE_SCOPE 68}; 69 70static struct tipc_server topsrv __read_mostly = { 71 .saddr = &topsrv_addr, 72 .imp = TIPC_CRITICAL_IMPORTANCE, 73 .type = SOCK_SEQPACKET, 74 .max_rcvbuf_size = sizeof(struct tipc_subscr), 75 .name = "topology_server", 76 .tipc_conn_recvmsg = subscr_conn_msg_event, 77 .tipc_conn_new = subscr_named_msg_event, 78 .tipc_conn_shutdown = subscr_conn_shutdown_event, 79}; 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 */ 88static u32 htohl(u32 in, int swap) 89{ 90 return swap ? swab32(in) : in; 91} 92 93static void subscr_send_event(struct tipc_subscription *sub, u32 found_lower, 94 u32 found_upper, u32 event, u32 port_ref, 95 u32 node) 96{ 97 struct tipc_subscriber *subscriber = sub->subscriber; 98 struct kvec msg_sect; 99 int ret; 100 101 msg_sect.iov_base = (void *)&sub->evt; 102 msg_sect.iov_len = sizeof(struct tipc_event); 103 104 sub->evt.event = htohl(event, sub->swap); 105 sub->evt.found_lower = htohl(found_lower, sub->swap); 106 sub->evt.found_upper = htohl(found_upper, sub->swap); 107 sub->evt.port.ref = htohl(port_ref, sub->swap); 108 sub->evt.port.node = htohl(node, sub->swap); 109 ret = tipc_conn_sendmsg(&topsrv, subscriber->conid, NULL, 110 msg_sect.iov_base, msg_sect.iov_len); 111 if (ret < 0) 112 pr_err("Sending subscription event failed, no memory\n"); 113} 114 115/** 116 * tipc_subscr_overlap - test for subscription overlap with the given values 117 * 118 * Returns 1 if there is overlap, otherwise 0. 119 */ 120int tipc_subscr_overlap(struct tipc_subscription *sub, 121 u32 found_lower, 122 u32 found_upper) 123 124{ 125 if (found_lower < sub->seq.lower) 126 found_lower = sub->seq.lower; 127 if (found_upper > sub->seq.upper) 128 found_upper = sub->seq.upper; 129 if (found_lower > found_upper) 130 return 0; 131 return 1; 132} 133 134/** 135 * tipc_subscr_report_overlap - issue event if there is subscription overlap 136 * 137 * Protected by nameseq.lock in name_table.c 138 */ 139void tipc_subscr_report_overlap(struct tipc_subscription *sub, 140 u32 found_lower, 141 u32 found_upper, 142 u32 event, 143 u32 port_ref, 144 u32 node, 145 int must) 146{ 147 if (!tipc_subscr_overlap(sub, found_lower, found_upper)) 148 return; 149 if (!must && !(sub->filter & TIPC_SUB_PORTS)) 150 return; 151 152 subscr_send_event(sub, found_lower, found_upper, event, port_ref, node); 153} 154 155static void subscr_timeout(struct tipc_subscription *sub) 156{ 157 struct tipc_subscriber *subscriber = sub->subscriber; 158 159 /* The spin lock per subscriber is used to protect its members */ 160 spin_lock_bh(&subscriber->lock); 161 162 /* Validate if the connection related to the subscriber is 163 * closed (in case subscriber is terminating) 164 */ 165 if (subscriber->conid == 0) { 166 spin_unlock_bh(&subscriber->lock); 167 return; 168 } 169 170 /* Validate timeout (in case subscription is being cancelled) */ 171 if (sub->timeout == TIPC_WAIT_FOREVER) { 172 spin_unlock_bh(&subscriber->lock); 173 return; 174 } 175 176 /* Unlink subscription from name table */ 177 tipc_nametbl_unsubscribe(sub); 178 179 /* Unlink subscription from subscriber */ 180 list_del(&sub->subscription_list); 181 182 spin_unlock_bh(&subscriber->lock); 183 184 /* Notify subscriber of timeout */ 185 subscr_send_event(sub, sub->evt.s.seq.lower, sub->evt.s.seq.upper, 186 TIPC_SUBSCR_TIMEOUT, 0, 0); 187 188 /* Now destroy subscription */ 189 k_term_timer(&sub->timer); 190 kfree(sub); 191 atomic_dec(&subscription_count); 192} 193 194/** 195 * subscr_del - delete a subscription within a subscription list 196 * 197 * Called with subscriber lock held. 198 */ 199static void subscr_del(struct tipc_subscription *sub) 200{ 201 tipc_nametbl_unsubscribe(sub); 202 list_del(&sub->subscription_list); 203 kfree(sub); 204 atomic_dec(&subscription_count); 205} 206 207/** 208 * subscr_terminate - terminate communication with a subscriber 209 * 210 * Note: Must call it in process context since it might sleep. 211 */ 212static void subscr_terminate(struct tipc_subscriber *subscriber) 213{ 214 tipc_conn_terminate(&topsrv, subscriber->conid); 215} 216 217static void subscr_release(struct tipc_subscriber *subscriber) 218{ 219 struct tipc_subscription *sub; 220 struct tipc_subscription *sub_temp; 221 222 spin_lock_bh(&subscriber->lock); 223 224 /* Invalidate subscriber reference */ 225 subscriber->conid = 0; 226 227 /* Destroy any existing subscriptions for subscriber */ 228 list_for_each_entry_safe(sub, sub_temp, &subscriber->subscription_list, 229 subscription_list) { 230 if (sub->timeout != TIPC_WAIT_FOREVER) { 231 spin_unlock_bh(&subscriber->lock); 232 k_cancel_timer(&sub->timer); 233 k_term_timer(&sub->timer); 234 spin_lock_bh(&subscriber->lock); 235 } 236 subscr_del(sub); 237 } 238 spin_unlock_bh(&subscriber->lock); 239 240 /* Now destroy subscriber */ 241 kfree(subscriber); 242} 243 244/** 245 * subscr_cancel - handle subscription cancellation request 246 * 247 * Called with subscriber lock held. Routine must temporarily release lock 248 * to enable the subscription timeout routine to finish without deadlocking; 249 * the lock is then reclaimed to allow caller to release it upon return. 250 * 251 * Note that fields of 's' use subscriber's endianness! 252 */ 253static void subscr_cancel(struct tipc_subscr *s, 254 struct tipc_subscriber *subscriber) 255{ 256 struct tipc_subscription *sub; 257 struct tipc_subscription *sub_temp; 258 int found = 0; 259 260 /* Find first matching subscription, exit if not found */ 261 list_for_each_entry_safe(sub, sub_temp, &subscriber->subscription_list, 262 subscription_list) { 263 if (!memcmp(s, &sub->evt.s, sizeof(struct tipc_subscr))) { 264 found = 1; 265 break; 266 } 267 } 268 if (!found) 269 return; 270 271 /* Cancel subscription timer (if used), then delete subscription */ 272 if (sub->timeout != TIPC_WAIT_FOREVER) { 273 sub->timeout = TIPC_WAIT_FOREVER; 274 spin_unlock_bh(&subscriber->lock); 275 k_cancel_timer(&sub->timer); 276 k_term_timer(&sub->timer); 277 spin_lock_bh(&subscriber->lock); 278 } 279 subscr_del(sub); 280} 281 282/** 283 * subscr_subscribe - create subscription for subscriber 284 * 285 * Called with subscriber lock held. 286 */ 287static struct tipc_subscription *subscr_subscribe(struct tipc_subscr *s, 288 struct tipc_subscriber *subscriber) 289{ 290 struct tipc_subscription *sub; 291 int swap; 292 293 /* Determine subscriber's endianness */ 294 swap = !(s->filter & (TIPC_SUB_PORTS | TIPC_SUB_SERVICE)); 295 296 /* Detect & process a subscription cancellation request */ 297 if (s->filter & htohl(TIPC_SUB_CANCEL, swap)) { 298 s->filter &= ~htohl(TIPC_SUB_CANCEL, swap); 299 subscr_cancel(s, subscriber); 300 return NULL; 301 } 302 303 /* Refuse subscription if global limit exceeded */ 304 if (atomic_read(&subscription_count) >= TIPC_MAX_SUBSCRIPTIONS) { 305 pr_warn("Subscription rejected, limit reached (%u)\n", 306 TIPC_MAX_SUBSCRIPTIONS); 307 subscr_terminate(subscriber); 308 return NULL; 309 } 310 311 /* Allocate subscription object */ 312 sub = kmalloc(sizeof(*sub), GFP_ATOMIC); 313 if (!sub) { 314 pr_warn("Subscription rejected, no memory\n"); 315 subscr_terminate(subscriber); 316 return NULL; 317 } 318 319 /* Initialize subscription object */ 320 sub->seq.type = htohl(s->seq.type, swap); 321 sub->seq.lower = htohl(s->seq.lower, swap); 322 sub->seq.upper = htohl(s->seq.upper, swap); 323 sub->timeout = htohl(s->timeout, swap); 324 sub->filter = htohl(s->filter, swap); 325 if ((!(sub->filter & TIPC_SUB_PORTS) == 326 !(sub->filter & TIPC_SUB_SERVICE)) || 327 (sub->seq.lower > sub->seq.upper)) { 328 pr_warn("Subscription rejected, illegal request\n"); 329 kfree(sub); 330 subscr_terminate(subscriber); 331 return NULL; 332 } 333 INIT_LIST_HEAD(&sub->nameseq_list); 334 list_add(&sub->subscription_list, &subscriber->subscription_list); 335 sub->subscriber = subscriber; 336 sub->swap = swap; 337 memcpy(&sub->evt.s, s, sizeof(struct tipc_subscr)); 338 atomic_inc(&subscription_count); 339 if (sub->timeout != TIPC_WAIT_FOREVER) { 340 k_init_timer(&sub->timer, 341 (Handler)subscr_timeout, (unsigned long)sub); 342 k_start_timer(&sub->timer, sub->timeout); 343 } 344 345 return sub; 346} 347 348/* Handle one termination request for the subscriber */ 349static void subscr_conn_shutdown_event(int conid, void *usr_data) 350{ 351 subscr_release((struct tipc_subscriber *)usr_data); 352} 353 354/* Handle one request to create a new subscription for the subscriber */ 355static void subscr_conn_msg_event(int conid, struct sockaddr_tipc *addr, 356 void *usr_data, void *buf, size_t len) 357{ 358 struct tipc_subscriber *subscriber = usr_data; 359 struct tipc_subscription *sub; 360 361 spin_lock_bh(&subscriber->lock); 362 sub = subscr_subscribe((struct tipc_subscr *)buf, subscriber); 363 if (sub) 364 tipc_nametbl_subscribe(sub); 365 spin_unlock_bh(&subscriber->lock); 366} 367 368 369/* Handle one request to establish a new subscriber */ 370static void *subscr_named_msg_event(int conid) 371{ 372 struct tipc_subscriber *subscriber; 373 374 /* Create subscriber object */ 375 subscriber = kzalloc(sizeof(struct tipc_subscriber), GFP_ATOMIC); 376 if (subscriber == NULL) { 377 pr_warn("Subscriber rejected, no memory\n"); 378 return NULL; 379 } 380 INIT_LIST_HEAD(&subscriber->subscription_list); 381 subscriber->conid = conid; 382 spin_lock_init(&subscriber->lock); 383 384 return (void *)subscriber; 385} 386 387int tipc_subscr_start(void) 388{ 389 return tipc_server_start(&topsrv); 390} 391 392void tipc_subscr_stop(void) 393{ 394 tipc_server_stop(&topsrv); 395} 396