1 /* 2 * net/tipc/name_distr.c: TIPC name distribution code 3 * 4 * Copyright (c) 2000-2006, 2014, Ericsson AB 5 * Copyright (c) 2005, 2010-2011, 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 "link.h" 39 #include "name_distr.h" 40 41 /** 42 * struct publ_list - list of publications made by this node 43 * @list: circular list of publications 44 * @list_size: number of entries in list 45 */ 46 struct publ_list { 47 struct list_head list; 48 u32 size; 49 }; 50 51 static struct publ_list publ_zone = { 52 .list = LIST_HEAD_INIT(publ_zone.list), 53 .size = 0, 54 }; 55 56 static struct publ_list publ_cluster = { 57 .list = LIST_HEAD_INIT(publ_cluster.list), 58 .size = 0, 59 }; 60 61 static struct publ_list publ_node = { 62 .list = LIST_HEAD_INIT(publ_node.list), 63 .size = 0, 64 }; 65 66 static struct publ_list *publ_lists[] = { 67 NULL, 68 &publ_zone, /* publ_lists[TIPC_ZONE_SCOPE] */ 69 &publ_cluster, /* publ_lists[TIPC_CLUSTER_SCOPE] */ 70 &publ_node /* publ_lists[TIPC_NODE_SCOPE] */ 71 }; 72 73 74 int sysctl_tipc_named_timeout __read_mostly = 2000; 75 76 /** 77 * struct tipc_dist_queue - queue holding deferred name table updates 78 */ 79 static struct list_head tipc_dist_queue = LIST_HEAD_INIT(tipc_dist_queue); 80 81 struct distr_queue_item { 82 struct distr_item i; 83 u32 dtype; 84 u32 node; 85 unsigned long expires; 86 struct list_head next; 87 }; 88 89 /** 90 * publ_to_item - add publication info to a publication message 91 */ 92 static void publ_to_item(struct distr_item *i, struct publication *p) 93 { 94 i->type = htonl(p->type); 95 i->lower = htonl(p->lower); 96 i->upper = htonl(p->upper); 97 i->ref = htonl(p->ref); 98 i->key = htonl(p->key); 99 } 100 101 /** 102 * named_prepare_buf - allocate & initialize a publication message 103 */ 104 static struct sk_buff *named_prepare_buf(u32 type, u32 size, u32 dest) 105 { 106 struct sk_buff *buf = tipc_buf_acquire(INT_H_SIZE + size); 107 struct tipc_msg *msg; 108 109 if (buf != NULL) { 110 msg = buf_msg(buf); 111 tipc_msg_init(msg, NAME_DISTRIBUTOR, type, INT_H_SIZE, dest); 112 msg_set_size(msg, INT_H_SIZE + size); 113 } 114 return buf; 115 } 116 117 void named_cluster_distribute(struct sk_buff *buf) 118 { 119 struct sk_buff *obuf; 120 struct tipc_node *node; 121 u32 dnode; 122 123 rcu_read_lock(); 124 list_for_each_entry_rcu(node, &tipc_node_list, list) { 125 dnode = node->addr; 126 if (in_own_node(dnode)) 127 continue; 128 if (!tipc_node_active_links(node)) 129 continue; 130 obuf = skb_copy(buf, GFP_ATOMIC); 131 if (!obuf) 132 break; 133 msg_set_destnode(buf_msg(obuf), dnode); 134 tipc_link_xmit(obuf, dnode, dnode); 135 } 136 rcu_read_unlock(); 137 138 kfree_skb(buf); 139 } 140 141 /** 142 * tipc_named_publish - tell other nodes about a new publication by this node 143 */ 144 struct sk_buff *tipc_named_publish(struct publication *publ) 145 { 146 struct sk_buff *buf; 147 struct distr_item *item; 148 149 list_add_tail(&publ->local_list, &publ_lists[publ->scope]->list); 150 publ_lists[publ->scope]->size++; 151 152 if (publ->scope == TIPC_NODE_SCOPE) 153 return NULL; 154 155 buf = named_prepare_buf(PUBLICATION, ITEM_SIZE, 0); 156 if (!buf) { 157 pr_warn("Publication distribution failure\n"); 158 return NULL; 159 } 160 161 item = (struct distr_item *)msg_data(buf_msg(buf)); 162 publ_to_item(item, publ); 163 return buf; 164 } 165 166 /** 167 * tipc_named_withdraw - tell other nodes about a withdrawn publication by this node 168 */ 169 struct sk_buff *tipc_named_withdraw(struct publication *publ) 170 { 171 struct sk_buff *buf; 172 struct distr_item *item; 173 174 list_del(&publ->local_list); 175 publ_lists[publ->scope]->size--; 176 177 if (publ->scope == TIPC_NODE_SCOPE) 178 return NULL; 179 180 buf = named_prepare_buf(WITHDRAWAL, ITEM_SIZE, 0); 181 if (!buf) { 182 pr_warn("Withdrawal distribution failure\n"); 183 return NULL; 184 } 185 186 item = (struct distr_item *)msg_data(buf_msg(buf)); 187 publ_to_item(item, publ); 188 return buf; 189 } 190 191 /** 192 * named_distribute - prepare name info for bulk distribution to another node 193 * @msg_list: list of messages (buffers) to be returned from this function 194 * @dnode: node to be updated 195 * @pls: linked list of publication items to be packed into buffer chain 196 */ 197 static void named_distribute(struct list_head *msg_list, u32 dnode, 198 struct publ_list *pls) 199 { 200 struct publication *publ; 201 struct sk_buff *buf = NULL; 202 struct distr_item *item = NULL; 203 uint dsz = pls->size * ITEM_SIZE; 204 uint msg_dsz = (tipc_node_get_mtu(dnode, 0) / ITEM_SIZE) * ITEM_SIZE; 205 uint rem = dsz; 206 uint msg_rem = 0; 207 208 list_for_each_entry(publ, &pls->list, local_list) { 209 /* Prepare next buffer: */ 210 if (!buf) { 211 msg_rem = min_t(uint, rem, msg_dsz); 212 rem -= msg_rem; 213 buf = named_prepare_buf(PUBLICATION, msg_rem, dnode); 214 if (!buf) { 215 pr_warn("Bulk publication failure\n"); 216 return; 217 } 218 item = (struct distr_item *)msg_data(buf_msg(buf)); 219 } 220 221 /* Pack publication into message: */ 222 publ_to_item(item, publ); 223 item++; 224 msg_rem -= ITEM_SIZE; 225 226 /* Append full buffer to list: */ 227 if (!msg_rem) { 228 list_add_tail((struct list_head *)buf, msg_list); 229 buf = NULL; 230 } 231 } 232 } 233 234 /** 235 * tipc_named_node_up - tell specified node about all publications by this node 236 */ 237 void tipc_named_node_up(u32 dnode) 238 { 239 LIST_HEAD(msg_list); 240 struct sk_buff *buf_chain; 241 242 read_lock_bh(&tipc_nametbl_lock); 243 named_distribute(&msg_list, dnode, &publ_cluster); 244 named_distribute(&msg_list, dnode, &publ_zone); 245 read_unlock_bh(&tipc_nametbl_lock); 246 247 /* Convert circular list to linear list and send: */ 248 buf_chain = (struct sk_buff *)msg_list.next; 249 ((struct sk_buff *)msg_list.prev)->next = NULL; 250 tipc_link_xmit(buf_chain, dnode, dnode); 251 } 252 253 /** 254 * named_purge_publ - remove publication associated with a failed node 255 * 256 * Invoked for each publication issued by a newly failed node. 257 * Removes publication structure from name table & deletes it. 258 */ 259 static void named_purge_publ(struct publication *publ) 260 { 261 struct publication *p; 262 263 write_lock_bh(&tipc_nametbl_lock); 264 p = tipc_nametbl_remove_publ(publ->type, publ->lower, 265 publ->node, publ->ref, publ->key); 266 if (p) 267 tipc_nodesub_unsubscribe(&p->subscr); 268 write_unlock_bh(&tipc_nametbl_lock); 269 270 if (p != publ) { 271 pr_err("Unable to remove publication from failed node\n" 272 " (type=%u, lower=%u, node=0x%x, ref=%u, key=%u)\n", 273 publ->type, publ->lower, publ->node, publ->ref, 274 publ->key); 275 } 276 277 kfree(p); 278 } 279 280 /** 281 * tipc_update_nametbl - try to process a nametable update and notify 282 * subscribers 283 * 284 * tipc_nametbl_lock must be held. 285 * Returns the publication item if successful, otherwise NULL. 286 */ 287 static bool tipc_update_nametbl(struct distr_item *i, u32 node, u32 dtype) 288 { 289 struct publication *publ = NULL; 290 291 if (dtype == PUBLICATION) { 292 publ = tipc_nametbl_insert_publ(ntohl(i->type), ntohl(i->lower), 293 ntohl(i->upper), 294 TIPC_CLUSTER_SCOPE, node, 295 ntohl(i->ref), ntohl(i->key)); 296 if (publ) { 297 tipc_nodesub_subscribe(&publ->subscr, node, publ, 298 (net_ev_handler) 299 named_purge_publ); 300 return true; 301 } 302 } else if (dtype == WITHDRAWAL) { 303 publ = tipc_nametbl_remove_publ(ntohl(i->type), ntohl(i->lower), 304 node, ntohl(i->ref), 305 ntohl(i->key)); 306 if (publ) { 307 tipc_nodesub_unsubscribe(&publ->subscr); 308 kfree(publ); 309 return true; 310 } 311 } else { 312 pr_warn("Unrecognized name table message received\n"); 313 } 314 return false; 315 } 316 317 /** 318 * tipc_named_add_backlog - add a failed name table update to the backlog 319 * 320 */ 321 static void tipc_named_add_backlog(struct distr_item *i, u32 type, u32 node) 322 { 323 struct distr_queue_item *e; 324 unsigned long now = get_jiffies_64(); 325 326 e = kzalloc(sizeof(*e), GFP_ATOMIC); 327 if (!e) 328 return; 329 e->dtype = type; 330 e->node = node; 331 e->expires = now + msecs_to_jiffies(sysctl_tipc_named_timeout); 332 memcpy(e, i, sizeof(*i)); 333 list_add_tail(&e->next, &tipc_dist_queue); 334 } 335 336 /** 337 * tipc_named_process_backlog - try to process any pending name table updates 338 * from the network. 339 */ 340 void tipc_named_process_backlog(void) 341 { 342 struct distr_queue_item *e, *tmp; 343 char addr[16]; 344 unsigned long now = get_jiffies_64(); 345 346 list_for_each_entry_safe(e, tmp, &tipc_dist_queue, next) { 347 if (time_after(e->expires, now)) { 348 if (!tipc_update_nametbl(&e->i, e->node, e->dtype)) 349 continue; 350 } else { 351 tipc_addr_string_fill(addr, e->node); 352 pr_warn_ratelimited("Dropping name table update (%d) of {%u, %u, %u} from %s key=%u\n", 353 e->dtype, ntohl(e->i.type), 354 ntohl(e->i.lower), 355 ntohl(e->i.upper), 356 addr, ntohl(e->i.key)); 357 } 358 list_del(&e->next); 359 kfree(e); 360 } 361 } 362 363 /** 364 * tipc_named_rcv - process name table update message sent by another node 365 */ 366 void tipc_named_rcv(struct sk_buff *buf) 367 { 368 struct tipc_msg *msg = buf_msg(buf); 369 struct distr_item *item = (struct distr_item *)msg_data(msg); 370 u32 count = msg_data_sz(msg) / ITEM_SIZE; 371 u32 node = msg_orignode(msg); 372 373 write_lock_bh(&tipc_nametbl_lock); 374 while (count--) { 375 if (!tipc_update_nametbl(item, node, msg_type(msg))) 376 tipc_named_add_backlog(item, msg_type(msg), node); 377 item++; 378 } 379 tipc_named_process_backlog(); 380 write_unlock_bh(&tipc_nametbl_lock); 381 kfree_skb(buf); 382 } 383 384 /** 385 * tipc_named_reinit - re-initialize local publications 386 * 387 * This routine is called whenever TIPC networking is enabled. 388 * All name table entries published by this node are updated to reflect 389 * the node's new network address. 390 */ 391 void tipc_named_reinit(void) 392 { 393 struct publication *publ; 394 int scope; 395 396 write_lock_bh(&tipc_nametbl_lock); 397 398 for (scope = TIPC_ZONE_SCOPE; scope <= TIPC_NODE_SCOPE; scope++) 399 list_for_each_entry(publ, &publ_lists[scope]->list, local_list) 400 publ->node = tipc_own_addr; 401 402 write_unlock_bh(&tipc_nametbl_lock); 403 } 404