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