1 /* 2 * net/tipc/link.c: TIPC link code 3 * 4 * Copyright (c) 1996-2007, Ericsson AB 5 * Copyright (c) 2004-2007, 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 "link.h" 40 #include "net.h" 41 #include "node.h" 42 #include "port.h" 43 #include "addr.h" 44 #include "node_subscr.h" 45 #include "name_distr.h" 46 #include "bearer.h" 47 #include "name_table.h" 48 #include "discover.h" 49 #include "config.h" 50 #include "bcast.h" 51 52 53 /* 54 * Limit for deferred reception queue: 55 */ 56 57 #define DEF_QUEUE_LIMIT 256u 58 59 /* 60 * Link state events: 61 */ 62 63 #define STARTING_EVT 856384768 /* link processing trigger */ 64 #define TRAFFIC_MSG_EVT 560815u /* rx'd ??? */ 65 #define TIMEOUT_EVT 560817u /* link timer expired */ 66 67 /* 68 * The following two 'message types' is really just implementation 69 * data conveniently stored in the message header. 70 * They must not be considered part of the protocol 71 */ 72 #define OPEN_MSG 0 73 #define CLOSED_MSG 1 74 75 /* 76 * State value stored in 'exp_msg_count' 77 */ 78 79 #define START_CHANGEOVER 100000u 80 81 /** 82 * struct link_name - deconstructed link name 83 * @addr_local: network address of node at this end 84 * @if_local: name of interface at this end 85 * @addr_peer: network address of node at far end 86 * @if_peer: name of interface at far end 87 */ 88 89 struct link_name { 90 u32 addr_local; 91 char if_local[TIPC_MAX_IF_NAME]; 92 u32 addr_peer; 93 char if_peer[TIPC_MAX_IF_NAME]; 94 }; 95 96 #if 0 97 98 /* LINK EVENT CODE IS NOT SUPPORTED AT PRESENT */ 99 100 /** 101 * struct link_event - link up/down event notification 102 */ 103 104 struct link_event { 105 u32 addr; 106 int up; 107 void (*fcn)(u32, char *, int); 108 char name[TIPC_MAX_LINK_NAME]; 109 }; 110 111 #endif 112 113 static void link_handle_out_of_seq_msg(struct link *l_ptr, 114 struct sk_buff *buf); 115 static void link_recv_proto_msg(struct link *l_ptr, struct sk_buff *buf); 116 static int link_recv_changeover_msg(struct link **l_ptr, struct sk_buff **buf); 117 static void link_set_supervision_props(struct link *l_ptr, u32 tolerance); 118 static int link_send_sections_long(struct port *sender, 119 struct iovec const *msg_sect, 120 u32 num_sect, u32 destnode); 121 static void link_check_defragm_bufs(struct link *l_ptr); 122 static void link_state_event(struct link *l_ptr, u32 event); 123 static void link_reset_statistics(struct link *l_ptr); 124 static void link_print(struct link *l_ptr, struct print_buf *buf, 125 const char *str); 126 127 /* 128 * Debugging code used by link routines only 129 * 130 * When debugging link problems on a system that has multiple links, 131 * the standard TIPC debugging routines may not be useful since they 132 * allow the output from multiple links to be intermixed. For this reason 133 * routines of the form "dbg_link_XXX()" have been created that will capture 134 * debug info into a link's personal print buffer, which can then be dumped 135 * into the TIPC system log (TIPC_LOG) upon request. 136 * 137 * To enable per-link debugging, use LINK_LOG_BUF_SIZE to specify the size 138 * of the print buffer used by each link. If LINK_LOG_BUF_SIZE is set to 0, 139 * the dbg_link_XXX() routines simply send their output to the standard 140 * debug print buffer (DBG_OUTPUT), if it has been defined; this can be useful 141 * when there is only a single link in the system being debugged. 142 * 143 * Notes: 144 * - When enabled, LINK_LOG_BUF_SIZE should be set to at least TIPC_PB_MIN_SIZE 145 * - "l_ptr" must be valid when using dbg_link_XXX() macros 146 */ 147 148 #define LINK_LOG_BUF_SIZE 0 149 150 #define dbg_link(fmt, arg...) do {if (LINK_LOG_BUF_SIZE) tipc_printf(&l_ptr->print_buf, fmt, ## arg); } while(0) 151 #define dbg_link_msg(msg, txt) do {if (LINK_LOG_BUF_SIZE) tipc_msg_print(&l_ptr->print_buf, msg, txt); } while(0) 152 #define dbg_link_state(txt) do {if (LINK_LOG_BUF_SIZE) link_print(l_ptr, &l_ptr->print_buf, txt); } while(0) 153 #define dbg_link_dump() do { \ 154 if (LINK_LOG_BUF_SIZE) { \ 155 tipc_printf(LOG, "\n\nDumping link <%s>:\n", l_ptr->name); \ 156 tipc_printbuf_move(LOG, &l_ptr->print_buf); \ 157 } \ 158 } while (0) 159 160 static void dbg_print_link(struct link *l_ptr, const char *str) 161 { 162 if (DBG_OUTPUT != TIPC_NULL) 163 link_print(l_ptr, DBG_OUTPUT, str); 164 } 165 166 static void dbg_print_buf_chain(struct sk_buff *root_buf) 167 { 168 if (DBG_OUTPUT != TIPC_NULL) { 169 struct sk_buff *buf = root_buf; 170 171 while (buf) { 172 msg_dbg(buf_msg(buf), "In chain: "); 173 buf = buf->next; 174 } 175 } 176 } 177 178 /* 179 * Simple link routines 180 */ 181 182 static unsigned int align(unsigned int i) 183 { 184 return (i + 3) & ~3u; 185 } 186 187 static int link_working_working(struct link *l_ptr) 188 { 189 return (l_ptr->state == WORKING_WORKING); 190 } 191 192 static int link_working_unknown(struct link *l_ptr) 193 { 194 return (l_ptr->state == WORKING_UNKNOWN); 195 } 196 197 static int link_reset_unknown(struct link *l_ptr) 198 { 199 return (l_ptr->state == RESET_UNKNOWN); 200 } 201 202 static int link_reset_reset(struct link *l_ptr) 203 { 204 return (l_ptr->state == RESET_RESET); 205 } 206 207 static int link_blocked(struct link *l_ptr) 208 { 209 return (l_ptr->exp_msg_count || l_ptr->blocked); 210 } 211 212 static int link_congested(struct link *l_ptr) 213 { 214 return (l_ptr->out_queue_size >= l_ptr->queue_limit[0]); 215 } 216 217 static u32 link_max_pkt(struct link *l_ptr) 218 { 219 return l_ptr->max_pkt; 220 } 221 222 static void link_init_max_pkt(struct link *l_ptr) 223 { 224 u32 max_pkt; 225 226 max_pkt = (l_ptr->b_ptr->publ.mtu & ~3); 227 if (max_pkt > MAX_MSG_SIZE) 228 max_pkt = MAX_MSG_SIZE; 229 230 l_ptr->max_pkt_target = max_pkt; 231 if (l_ptr->max_pkt_target < MAX_PKT_DEFAULT) 232 l_ptr->max_pkt = l_ptr->max_pkt_target; 233 else 234 l_ptr->max_pkt = MAX_PKT_DEFAULT; 235 236 l_ptr->max_pkt_probes = 0; 237 } 238 239 static u32 link_next_sent(struct link *l_ptr) 240 { 241 if (l_ptr->next_out) 242 return msg_seqno(buf_msg(l_ptr->next_out)); 243 return mod(l_ptr->next_out_no); 244 } 245 246 static u32 link_last_sent(struct link *l_ptr) 247 { 248 return mod(link_next_sent(l_ptr) - 1); 249 } 250 251 /* 252 * Simple non-static link routines (i.e. referenced outside this file) 253 */ 254 255 int tipc_link_is_up(struct link *l_ptr) 256 { 257 if (!l_ptr) 258 return 0; 259 return (link_working_working(l_ptr) || link_working_unknown(l_ptr)); 260 } 261 262 int tipc_link_is_active(struct link *l_ptr) 263 { 264 return ((l_ptr->owner->active_links[0] == l_ptr) || 265 (l_ptr->owner->active_links[1] == l_ptr)); 266 } 267 268 /** 269 * link_name_validate - validate & (optionally) deconstruct link name 270 * @name - ptr to link name string 271 * @name_parts - ptr to area for link name components (or NULL if not needed) 272 * 273 * Returns 1 if link name is valid, otherwise 0. 274 */ 275 276 static int link_name_validate(const char *name, struct link_name *name_parts) 277 { 278 char name_copy[TIPC_MAX_LINK_NAME]; 279 char *addr_local; 280 char *if_local; 281 char *addr_peer; 282 char *if_peer; 283 char dummy; 284 u32 z_local, c_local, n_local; 285 u32 z_peer, c_peer, n_peer; 286 u32 if_local_len; 287 u32 if_peer_len; 288 289 /* copy link name & ensure length is OK */ 290 291 name_copy[TIPC_MAX_LINK_NAME - 1] = 0; 292 /* need above in case non-Posix strncpy() doesn't pad with nulls */ 293 strncpy(name_copy, name, TIPC_MAX_LINK_NAME); 294 if (name_copy[TIPC_MAX_LINK_NAME - 1] != 0) 295 return 0; 296 297 /* ensure all component parts of link name are present */ 298 299 addr_local = name_copy; 300 if ((if_local = strchr(addr_local, ':')) == NULL) 301 return 0; 302 *(if_local++) = 0; 303 if ((addr_peer = strchr(if_local, '-')) == NULL) 304 return 0; 305 *(addr_peer++) = 0; 306 if_local_len = addr_peer - if_local; 307 if ((if_peer = strchr(addr_peer, ':')) == NULL) 308 return 0; 309 *(if_peer++) = 0; 310 if_peer_len = strlen(if_peer) + 1; 311 312 /* validate component parts of link name */ 313 314 if ((sscanf(addr_local, "%u.%u.%u%c", 315 &z_local, &c_local, &n_local, &dummy) != 3) || 316 (sscanf(addr_peer, "%u.%u.%u%c", 317 &z_peer, &c_peer, &n_peer, &dummy) != 3) || 318 (z_local > 255) || (c_local > 4095) || (n_local > 4095) || 319 (z_peer > 255) || (c_peer > 4095) || (n_peer > 4095) || 320 (if_local_len <= 1) || (if_local_len > TIPC_MAX_IF_NAME) || 321 (if_peer_len <= 1) || (if_peer_len > TIPC_MAX_IF_NAME) || 322 (strspn(if_local, tipc_alphabet) != (if_local_len - 1)) || 323 (strspn(if_peer, tipc_alphabet) != (if_peer_len - 1))) 324 return 0; 325 326 /* return link name components, if necessary */ 327 328 if (name_parts) { 329 name_parts->addr_local = tipc_addr(z_local, c_local, n_local); 330 strcpy(name_parts->if_local, if_local); 331 name_parts->addr_peer = tipc_addr(z_peer, c_peer, n_peer); 332 strcpy(name_parts->if_peer, if_peer); 333 } 334 return 1; 335 } 336 337 /** 338 * link_timeout - handle expiration of link timer 339 * @l_ptr: pointer to link 340 * 341 * This routine must not grab "tipc_net_lock" to avoid a potential deadlock conflict 342 * with tipc_link_delete(). (There is no risk that the node will be deleted by 343 * another thread because tipc_link_delete() always cancels the link timer before 344 * tipc_node_delete() is called.) 345 */ 346 347 static void link_timeout(struct link *l_ptr) 348 { 349 tipc_node_lock(l_ptr->owner); 350 351 /* update counters used in statistical profiling of send traffic */ 352 353 l_ptr->stats.accu_queue_sz += l_ptr->out_queue_size; 354 l_ptr->stats.queue_sz_counts++; 355 356 if (l_ptr->out_queue_size > l_ptr->stats.max_queue_sz) 357 l_ptr->stats.max_queue_sz = l_ptr->out_queue_size; 358 359 if (l_ptr->first_out) { 360 struct tipc_msg *msg = buf_msg(l_ptr->first_out); 361 u32 length = msg_size(msg); 362 363 if ((msg_user(msg) == MSG_FRAGMENTER) 364 && (msg_type(msg) == FIRST_FRAGMENT)) { 365 length = msg_size(msg_get_wrapped(msg)); 366 } 367 if (length) { 368 l_ptr->stats.msg_lengths_total += length; 369 l_ptr->stats.msg_length_counts++; 370 if (length <= 64) 371 l_ptr->stats.msg_length_profile[0]++; 372 else if (length <= 256) 373 l_ptr->stats.msg_length_profile[1]++; 374 else if (length <= 1024) 375 l_ptr->stats.msg_length_profile[2]++; 376 else if (length <= 4096) 377 l_ptr->stats.msg_length_profile[3]++; 378 else if (length <= 16384) 379 l_ptr->stats.msg_length_profile[4]++; 380 else if (length <= 32768) 381 l_ptr->stats.msg_length_profile[5]++; 382 else 383 l_ptr->stats.msg_length_profile[6]++; 384 } 385 } 386 387 /* do all other link processing performed on a periodic basis */ 388 389 link_check_defragm_bufs(l_ptr); 390 391 link_state_event(l_ptr, TIMEOUT_EVT); 392 393 if (l_ptr->next_out) 394 tipc_link_push_queue(l_ptr); 395 396 tipc_node_unlock(l_ptr->owner); 397 } 398 399 static void link_set_timer(struct link *l_ptr, u32 time) 400 { 401 k_start_timer(&l_ptr->timer, time); 402 } 403 404 /** 405 * tipc_link_create - create a new link 406 * @b_ptr: pointer to associated bearer 407 * @peer: network address of node at other end of link 408 * @media_addr: media address to use when sending messages over link 409 * 410 * Returns pointer to link. 411 */ 412 413 struct link *tipc_link_create(struct bearer *b_ptr, const u32 peer, 414 const struct tipc_media_addr *media_addr) 415 { 416 struct link *l_ptr; 417 struct tipc_msg *msg; 418 char *if_name; 419 420 l_ptr = kzalloc(sizeof(*l_ptr), GFP_ATOMIC); 421 if (!l_ptr) { 422 warn("Link creation failed, no memory\n"); 423 return NULL; 424 } 425 426 if (LINK_LOG_BUF_SIZE) { 427 char *pb = kmalloc(LINK_LOG_BUF_SIZE, GFP_ATOMIC); 428 429 if (!pb) { 430 kfree(l_ptr); 431 warn("Link creation failed, no memory for print buffer\n"); 432 return NULL; 433 } 434 tipc_printbuf_init(&l_ptr->print_buf, pb, LINK_LOG_BUF_SIZE); 435 } 436 437 l_ptr->addr = peer; 438 if_name = strchr(b_ptr->publ.name, ':') + 1; 439 sprintf(l_ptr->name, "%u.%u.%u:%s-%u.%u.%u:", 440 tipc_zone(tipc_own_addr), tipc_cluster(tipc_own_addr), 441 tipc_node(tipc_own_addr), 442 if_name, 443 tipc_zone(peer), tipc_cluster(peer), tipc_node(peer)); 444 /* note: peer i/f is appended to link name by reset/activate */ 445 memcpy(&l_ptr->media_addr, media_addr, sizeof(*media_addr)); 446 l_ptr->checkpoint = 1; 447 l_ptr->b_ptr = b_ptr; 448 link_set_supervision_props(l_ptr, b_ptr->media->tolerance); 449 l_ptr->state = RESET_UNKNOWN; 450 451 l_ptr->pmsg = (struct tipc_msg *)&l_ptr->proto_msg; 452 msg = l_ptr->pmsg; 453 msg_init(msg, LINK_PROTOCOL, RESET_MSG, TIPC_OK, INT_H_SIZE, l_ptr->addr); 454 msg_set_size(msg, sizeof(l_ptr->proto_msg)); 455 msg_set_session(msg, tipc_random); 456 msg_set_bearer_id(msg, b_ptr->identity); 457 strcpy((char *)msg_data(msg), if_name); 458 459 l_ptr->priority = b_ptr->priority; 460 tipc_link_set_queue_limits(l_ptr, b_ptr->media->window); 461 462 link_init_max_pkt(l_ptr); 463 464 l_ptr->next_out_no = 1; 465 INIT_LIST_HEAD(&l_ptr->waiting_ports); 466 467 link_reset_statistics(l_ptr); 468 469 l_ptr->owner = tipc_node_attach_link(l_ptr); 470 if (!l_ptr->owner) { 471 if (LINK_LOG_BUF_SIZE) 472 kfree(l_ptr->print_buf.buf); 473 kfree(l_ptr); 474 return NULL; 475 } 476 477 k_init_timer(&l_ptr->timer, (Handler)link_timeout, (unsigned long)l_ptr); 478 list_add_tail(&l_ptr->link_list, &b_ptr->links); 479 tipc_k_signal((Handler)tipc_link_start, (unsigned long)l_ptr); 480 481 dbg("tipc_link_create(): tolerance = %u,cont intv = %u, abort_limit = %u\n", 482 l_ptr->tolerance, l_ptr->continuity_interval, l_ptr->abort_limit); 483 484 return l_ptr; 485 } 486 487 /** 488 * tipc_link_delete - delete a link 489 * @l_ptr: pointer to link 490 * 491 * Note: 'tipc_net_lock' is write_locked, bearer is locked. 492 * This routine must not grab the node lock until after link timer cancellation 493 * to avoid a potential deadlock situation. 494 */ 495 496 void tipc_link_delete(struct link *l_ptr) 497 { 498 if (!l_ptr) { 499 err("Attempt to delete non-existent link\n"); 500 return; 501 } 502 503 dbg("tipc_link_delete()\n"); 504 505 k_cancel_timer(&l_ptr->timer); 506 507 tipc_node_lock(l_ptr->owner); 508 tipc_link_reset(l_ptr); 509 tipc_node_detach_link(l_ptr->owner, l_ptr); 510 tipc_link_stop(l_ptr); 511 list_del_init(&l_ptr->link_list); 512 if (LINK_LOG_BUF_SIZE) 513 kfree(l_ptr->print_buf.buf); 514 tipc_node_unlock(l_ptr->owner); 515 k_term_timer(&l_ptr->timer); 516 kfree(l_ptr); 517 } 518 519 void tipc_link_start(struct link *l_ptr) 520 { 521 dbg("tipc_link_start %x\n", l_ptr); 522 link_state_event(l_ptr, STARTING_EVT); 523 } 524 525 /** 526 * link_schedule_port - schedule port for deferred sending 527 * @l_ptr: pointer to link 528 * @origport: reference to sending port 529 * @sz: amount of data to be sent 530 * 531 * Schedules port for renewed sending of messages after link congestion 532 * has abated. 533 */ 534 535 static int link_schedule_port(struct link *l_ptr, u32 origport, u32 sz) 536 { 537 struct port *p_ptr; 538 539 spin_lock_bh(&tipc_port_list_lock); 540 p_ptr = tipc_port_lock(origport); 541 if (p_ptr) { 542 if (!p_ptr->wakeup) 543 goto exit; 544 if (!list_empty(&p_ptr->wait_list)) 545 goto exit; 546 p_ptr->congested_link = l_ptr; 547 p_ptr->publ.congested = 1; 548 p_ptr->waiting_pkts = 1 + ((sz - 1) / link_max_pkt(l_ptr)); 549 list_add_tail(&p_ptr->wait_list, &l_ptr->waiting_ports); 550 l_ptr->stats.link_congs++; 551 exit: 552 tipc_port_unlock(p_ptr); 553 } 554 spin_unlock_bh(&tipc_port_list_lock); 555 return -ELINKCONG; 556 } 557 558 void tipc_link_wakeup_ports(struct link *l_ptr, int all) 559 { 560 struct port *p_ptr; 561 struct port *temp_p_ptr; 562 int win = l_ptr->queue_limit[0] - l_ptr->out_queue_size; 563 564 if (all) 565 win = 100000; 566 if (win <= 0) 567 return; 568 if (!spin_trylock_bh(&tipc_port_list_lock)) 569 return; 570 if (link_congested(l_ptr)) 571 goto exit; 572 list_for_each_entry_safe(p_ptr, temp_p_ptr, &l_ptr->waiting_ports, 573 wait_list) { 574 if (win <= 0) 575 break; 576 list_del_init(&p_ptr->wait_list); 577 p_ptr->congested_link = NULL; 578 spin_lock_bh(p_ptr->publ.lock); 579 p_ptr->publ.congested = 0; 580 p_ptr->wakeup(&p_ptr->publ); 581 win -= p_ptr->waiting_pkts; 582 spin_unlock_bh(p_ptr->publ.lock); 583 } 584 585 exit: 586 spin_unlock_bh(&tipc_port_list_lock); 587 } 588 589 /** 590 * link_release_outqueue - purge link's outbound message queue 591 * @l_ptr: pointer to link 592 */ 593 594 static void link_release_outqueue(struct link *l_ptr) 595 { 596 struct sk_buff *buf = l_ptr->first_out; 597 struct sk_buff *next; 598 599 while (buf) { 600 next = buf->next; 601 buf_discard(buf); 602 buf = next; 603 } 604 l_ptr->first_out = NULL; 605 l_ptr->out_queue_size = 0; 606 } 607 608 /** 609 * tipc_link_reset_fragments - purge link's inbound message fragments queue 610 * @l_ptr: pointer to link 611 */ 612 613 void tipc_link_reset_fragments(struct link *l_ptr) 614 { 615 struct sk_buff *buf = l_ptr->defragm_buf; 616 struct sk_buff *next; 617 618 while (buf) { 619 next = buf->next; 620 buf_discard(buf); 621 buf = next; 622 } 623 l_ptr->defragm_buf = NULL; 624 } 625 626 /** 627 * tipc_link_stop - purge all inbound and outbound messages associated with link 628 * @l_ptr: pointer to link 629 */ 630 631 void tipc_link_stop(struct link *l_ptr) 632 { 633 struct sk_buff *buf; 634 struct sk_buff *next; 635 636 buf = l_ptr->oldest_deferred_in; 637 while (buf) { 638 next = buf->next; 639 buf_discard(buf); 640 buf = next; 641 } 642 643 buf = l_ptr->first_out; 644 while (buf) { 645 next = buf->next; 646 buf_discard(buf); 647 buf = next; 648 } 649 650 tipc_link_reset_fragments(l_ptr); 651 652 buf_discard(l_ptr->proto_msg_queue); 653 l_ptr->proto_msg_queue = NULL; 654 } 655 656 #if 0 657 658 /* LINK EVENT CODE IS NOT SUPPORTED AT PRESENT */ 659 660 static void link_recv_event(struct link_event *ev) 661 { 662 ev->fcn(ev->addr, ev->name, ev->up); 663 kfree(ev); 664 } 665 666 static void link_send_event(void (*fcn)(u32 a, char *n, int up), 667 struct link *l_ptr, int up) 668 { 669 struct link_event *ev; 670 671 ev = kmalloc(sizeof(*ev), GFP_ATOMIC); 672 if (!ev) { 673 warn("Link event allocation failure\n"); 674 return; 675 } 676 ev->addr = l_ptr->addr; 677 ev->up = up; 678 ev->fcn = fcn; 679 memcpy(ev->name, l_ptr->name, TIPC_MAX_LINK_NAME); 680 tipc_k_signal((Handler)link_recv_event, (unsigned long)ev); 681 } 682 683 #else 684 685 #define link_send_event(fcn, l_ptr, up) do { } while (0) 686 687 #endif 688 689 void tipc_link_reset(struct link *l_ptr) 690 { 691 struct sk_buff *buf; 692 u32 prev_state = l_ptr->state; 693 u32 checkpoint = l_ptr->next_in_no; 694 int was_active_link = tipc_link_is_active(l_ptr); 695 696 msg_set_session(l_ptr->pmsg, msg_session(l_ptr->pmsg) + 1); 697 698 /* Link is down, accept any session: */ 699 l_ptr->peer_session = 0; 700 701 /* Prepare for max packet size negotiation */ 702 link_init_max_pkt(l_ptr); 703 704 l_ptr->state = RESET_UNKNOWN; 705 dbg_link_state("Resetting Link\n"); 706 707 if ((prev_state == RESET_UNKNOWN) || (prev_state == RESET_RESET)) 708 return; 709 710 tipc_node_link_down(l_ptr->owner, l_ptr); 711 tipc_bearer_remove_dest(l_ptr->b_ptr, l_ptr->addr); 712 #if 0 713 tipc_printf(TIPC_CONS, "\nReset link <%s>\n", l_ptr->name); 714 dbg_link_dump(); 715 #endif 716 if (was_active_link && tipc_node_has_active_links(l_ptr->owner) && 717 l_ptr->owner->permit_changeover) { 718 l_ptr->reset_checkpoint = checkpoint; 719 l_ptr->exp_msg_count = START_CHANGEOVER; 720 } 721 722 /* Clean up all queues: */ 723 724 link_release_outqueue(l_ptr); 725 buf_discard(l_ptr->proto_msg_queue); 726 l_ptr->proto_msg_queue = NULL; 727 buf = l_ptr->oldest_deferred_in; 728 while (buf) { 729 struct sk_buff *next = buf->next; 730 buf_discard(buf); 731 buf = next; 732 } 733 if (!list_empty(&l_ptr->waiting_ports)) 734 tipc_link_wakeup_ports(l_ptr, 1); 735 736 l_ptr->retransm_queue_head = 0; 737 l_ptr->retransm_queue_size = 0; 738 l_ptr->last_out = NULL; 739 l_ptr->first_out = NULL; 740 l_ptr->next_out = NULL; 741 l_ptr->unacked_window = 0; 742 l_ptr->checkpoint = 1; 743 l_ptr->next_out_no = 1; 744 l_ptr->deferred_inqueue_sz = 0; 745 l_ptr->oldest_deferred_in = NULL; 746 l_ptr->newest_deferred_in = NULL; 747 l_ptr->fsm_msg_cnt = 0; 748 l_ptr->stale_count = 0; 749 link_reset_statistics(l_ptr); 750 751 link_send_event(tipc_cfg_link_event, l_ptr, 0); 752 if (!in_own_cluster(l_ptr->addr)) 753 link_send_event(tipc_disc_link_event, l_ptr, 0); 754 } 755 756 757 static void link_activate(struct link *l_ptr) 758 { 759 l_ptr->next_in_no = l_ptr->stats.recv_info = 1; 760 tipc_node_link_up(l_ptr->owner, l_ptr); 761 tipc_bearer_add_dest(l_ptr->b_ptr, l_ptr->addr); 762 link_send_event(tipc_cfg_link_event, l_ptr, 1); 763 if (!in_own_cluster(l_ptr->addr)) 764 link_send_event(tipc_disc_link_event, l_ptr, 1); 765 } 766 767 /** 768 * link_state_event - link finite state machine 769 * @l_ptr: pointer to link 770 * @event: state machine event to process 771 */ 772 773 static void link_state_event(struct link *l_ptr, unsigned event) 774 { 775 struct link *other; 776 u32 cont_intv = l_ptr->continuity_interval; 777 778 if (!l_ptr->started && (event != STARTING_EVT)) 779 return; /* Not yet. */ 780 781 if (link_blocked(l_ptr)) { 782 if (event == TIMEOUT_EVT) { 783 link_set_timer(l_ptr, cont_intv); 784 } 785 return; /* Changeover going on */ 786 } 787 dbg_link("STATE_EV: <%s> ", l_ptr->name); 788 789 switch (l_ptr->state) { 790 case WORKING_WORKING: 791 dbg_link("WW/"); 792 switch (event) { 793 case TRAFFIC_MSG_EVT: 794 dbg_link("TRF-"); 795 /* fall through */ 796 case ACTIVATE_MSG: 797 dbg_link("ACT\n"); 798 break; 799 case TIMEOUT_EVT: 800 dbg_link("TIM "); 801 if (l_ptr->next_in_no != l_ptr->checkpoint) { 802 l_ptr->checkpoint = l_ptr->next_in_no; 803 if (tipc_bclink_acks_missing(l_ptr->owner)) { 804 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 805 0, 0, 0, 0, 0); 806 l_ptr->fsm_msg_cnt++; 807 } else if (l_ptr->max_pkt < l_ptr->max_pkt_target) { 808 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 809 1, 0, 0, 0, 0); 810 l_ptr->fsm_msg_cnt++; 811 } 812 link_set_timer(l_ptr, cont_intv); 813 break; 814 } 815 dbg_link(" -> WU\n"); 816 l_ptr->state = WORKING_UNKNOWN; 817 l_ptr->fsm_msg_cnt = 0; 818 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 1, 0, 0, 0, 0); 819 l_ptr->fsm_msg_cnt++; 820 link_set_timer(l_ptr, cont_intv / 4); 821 break; 822 case RESET_MSG: 823 dbg_link("RES -> RR\n"); 824 info("Resetting link <%s>, requested by peer\n", 825 l_ptr->name); 826 tipc_link_reset(l_ptr); 827 l_ptr->state = RESET_RESET; 828 l_ptr->fsm_msg_cnt = 0; 829 tipc_link_send_proto_msg(l_ptr, ACTIVATE_MSG, 0, 0, 0, 0, 0); 830 l_ptr->fsm_msg_cnt++; 831 link_set_timer(l_ptr, cont_intv); 832 break; 833 default: 834 err("Unknown link event %u in WW state\n", event); 835 } 836 break; 837 case WORKING_UNKNOWN: 838 dbg_link("WU/"); 839 switch (event) { 840 case TRAFFIC_MSG_EVT: 841 dbg_link("TRF-"); 842 case ACTIVATE_MSG: 843 dbg_link("ACT -> WW\n"); 844 l_ptr->state = WORKING_WORKING; 845 l_ptr->fsm_msg_cnt = 0; 846 link_set_timer(l_ptr, cont_intv); 847 break; 848 case RESET_MSG: 849 dbg_link("RES -> RR\n"); 850 info("Resetting link <%s>, requested by peer " 851 "while probing\n", l_ptr->name); 852 tipc_link_reset(l_ptr); 853 l_ptr->state = RESET_RESET; 854 l_ptr->fsm_msg_cnt = 0; 855 tipc_link_send_proto_msg(l_ptr, ACTIVATE_MSG, 0, 0, 0, 0, 0); 856 l_ptr->fsm_msg_cnt++; 857 link_set_timer(l_ptr, cont_intv); 858 break; 859 case TIMEOUT_EVT: 860 dbg_link("TIM "); 861 if (l_ptr->next_in_no != l_ptr->checkpoint) { 862 dbg_link("-> WW \n"); 863 l_ptr->state = WORKING_WORKING; 864 l_ptr->fsm_msg_cnt = 0; 865 l_ptr->checkpoint = l_ptr->next_in_no; 866 if (tipc_bclink_acks_missing(l_ptr->owner)) { 867 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 868 0, 0, 0, 0, 0); 869 l_ptr->fsm_msg_cnt++; 870 } 871 link_set_timer(l_ptr, cont_intv); 872 } else if (l_ptr->fsm_msg_cnt < l_ptr->abort_limit) { 873 dbg_link("Probing %u/%u,timer = %u ms)\n", 874 l_ptr->fsm_msg_cnt, l_ptr->abort_limit, 875 cont_intv / 4); 876 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 877 1, 0, 0, 0, 0); 878 l_ptr->fsm_msg_cnt++; 879 link_set_timer(l_ptr, cont_intv / 4); 880 } else { /* Link has failed */ 881 dbg_link("-> RU (%u probes unanswered)\n", 882 l_ptr->fsm_msg_cnt); 883 warn("Resetting link <%s>, peer not responding\n", 884 l_ptr->name); 885 tipc_link_reset(l_ptr); 886 l_ptr->state = RESET_UNKNOWN; 887 l_ptr->fsm_msg_cnt = 0; 888 tipc_link_send_proto_msg(l_ptr, RESET_MSG, 889 0, 0, 0, 0, 0); 890 l_ptr->fsm_msg_cnt++; 891 link_set_timer(l_ptr, cont_intv); 892 } 893 break; 894 default: 895 err("Unknown link event %u in WU state\n", event); 896 } 897 break; 898 case RESET_UNKNOWN: 899 dbg_link("RU/"); 900 switch (event) { 901 case TRAFFIC_MSG_EVT: 902 dbg_link("TRF-\n"); 903 break; 904 case ACTIVATE_MSG: 905 other = l_ptr->owner->active_links[0]; 906 if (other && link_working_unknown(other)) { 907 dbg_link("ACT\n"); 908 break; 909 } 910 dbg_link("ACT -> WW\n"); 911 l_ptr->state = WORKING_WORKING; 912 l_ptr->fsm_msg_cnt = 0; 913 link_activate(l_ptr); 914 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 1, 0, 0, 0, 0); 915 l_ptr->fsm_msg_cnt++; 916 link_set_timer(l_ptr, cont_intv); 917 break; 918 case RESET_MSG: 919 dbg_link("RES \n"); 920 dbg_link(" -> RR\n"); 921 l_ptr->state = RESET_RESET; 922 l_ptr->fsm_msg_cnt = 0; 923 tipc_link_send_proto_msg(l_ptr, ACTIVATE_MSG, 1, 0, 0, 0, 0); 924 l_ptr->fsm_msg_cnt++; 925 link_set_timer(l_ptr, cont_intv); 926 break; 927 case STARTING_EVT: 928 dbg_link("START-"); 929 l_ptr->started = 1; 930 /* fall through */ 931 case TIMEOUT_EVT: 932 dbg_link("TIM \n"); 933 tipc_link_send_proto_msg(l_ptr, RESET_MSG, 0, 0, 0, 0, 0); 934 l_ptr->fsm_msg_cnt++; 935 link_set_timer(l_ptr, cont_intv); 936 break; 937 default: 938 err("Unknown link event %u in RU state\n", event); 939 } 940 break; 941 case RESET_RESET: 942 dbg_link("RR/ "); 943 switch (event) { 944 case TRAFFIC_MSG_EVT: 945 dbg_link("TRF-"); 946 /* fall through */ 947 case ACTIVATE_MSG: 948 other = l_ptr->owner->active_links[0]; 949 if (other && link_working_unknown(other)) { 950 dbg_link("ACT\n"); 951 break; 952 } 953 dbg_link("ACT -> WW\n"); 954 l_ptr->state = WORKING_WORKING; 955 l_ptr->fsm_msg_cnt = 0; 956 link_activate(l_ptr); 957 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 1, 0, 0, 0, 0); 958 l_ptr->fsm_msg_cnt++; 959 link_set_timer(l_ptr, cont_intv); 960 break; 961 case RESET_MSG: 962 dbg_link("RES\n"); 963 break; 964 case TIMEOUT_EVT: 965 dbg_link("TIM\n"); 966 tipc_link_send_proto_msg(l_ptr, ACTIVATE_MSG, 0, 0, 0, 0, 0); 967 l_ptr->fsm_msg_cnt++; 968 link_set_timer(l_ptr, cont_intv); 969 dbg_link("fsm_msg_cnt %u\n", l_ptr->fsm_msg_cnt); 970 break; 971 default: 972 err("Unknown link event %u in RR state\n", event); 973 } 974 break; 975 default: 976 err("Unknown link state %u/%u\n", l_ptr->state, event); 977 } 978 } 979 980 /* 981 * link_bundle_buf(): Append contents of a buffer to 982 * the tail of an existing one. 983 */ 984 985 static int link_bundle_buf(struct link *l_ptr, 986 struct sk_buff *bundler, 987 struct sk_buff *buf) 988 { 989 struct tipc_msg *bundler_msg = buf_msg(bundler); 990 struct tipc_msg *msg = buf_msg(buf); 991 u32 size = msg_size(msg); 992 u32 bundle_size = msg_size(bundler_msg); 993 u32 to_pos = align(bundle_size); 994 u32 pad = to_pos - bundle_size; 995 996 if (msg_user(bundler_msg) != MSG_BUNDLER) 997 return 0; 998 if (msg_type(bundler_msg) != OPEN_MSG) 999 return 0; 1000 if (skb_tailroom(bundler) < (pad + size)) 1001 return 0; 1002 if (link_max_pkt(l_ptr) < (to_pos + size)) 1003 return 0; 1004 1005 skb_put(bundler, pad + size); 1006 skb_copy_to_linear_data_offset(bundler, to_pos, buf->data, size); 1007 msg_set_size(bundler_msg, to_pos + size); 1008 msg_set_msgcnt(bundler_msg, msg_msgcnt(bundler_msg) + 1); 1009 dbg("Packed msg # %u(%u octets) into pos %u in buf(#%u)\n", 1010 msg_msgcnt(bundler_msg), size, to_pos, msg_seqno(bundler_msg)); 1011 msg_dbg(msg, "PACKD:"); 1012 buf_discard(buf); 1013 l_ptr->stats.sent_bundled++; 1014 return 1; 1015 } 1016 1017 static void link_add_to_outqueue(struct link *l_ptr, 1018 struct sk_buff *buf, 1019 struct tipc_msg *msg) 1020 { 1021 u32 ack = mod(l_ptr->next_in_no - 1); 1022 u32 seqno = mod(l_ptr->next_out_no++); 1023 1024 msg_set_word(msg, 2, ((ack << 16) | seqno)); 1025 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in); 1026 buf->next = NULL; 1027 if (l_ptr->first_out) { 1028 l_ptr->last_out->next = buf; 1029 l_ptr->last_out = buf; 1030 } else 1031 l_ptr->first_out = l_ptr->last_out = buf; 1032 l_ptr->out_queue_size++; 1033 } 1034 1035 /* 1036 * tipc_link_send_buf() is the 'full path' for messages, called from 1037 * inside TIPC when the 'fast path' in tipc_send_buf 1038 * has failed, and from link_send() 1039 */ 1040 1041 int tipc_link_send_buf(struct link *l_ptr, struct sk_buff *buf) 1042 { 1043 struct tipc_msg *msg = buf_msg(buf); 1044 u32 size = msg_size(msg); 1045 u32 dsz = msg_data_sz(msg); 1046 u32 queue_size = l_ptr->out_queue_size; 1047 u32 imp = msg_tot_importance(msg); 1048 u32 queue_limit = l_ptr->queue_limit[imp]; 1049 u32 max_packet = link_max_pkt(l_ptr); 1050 1051 msg_set_prevnode(msg, tipc_own_addr); /* If routed message */ 1052 1053 /* Match msg importance against queue limits: */ 1054 1055 if (unlikely(queue_size >= queue_limit)) { 1056 if (imp <= TIPC_CRITICAL_IMPORTANCE) { 1057 return link_schedule_port(l_ptr, msg_origport(msg), 1058 size); 1059 } 1060 msg_dbg(msg, "TIPC: Congestion, throwing away\n"); 1061 buf_discard(buf); 1062 if (imp > CONN_MANAGER) { 1063 warn("Resetting link <%s>, send queue full", l_ptr->name); 1064 tipc_link_reset(l_ptr); 1065 } 1066 return dsz; 1067 } 1068 1069 /* Fragmentation needed ? */ 1070 1071 if (size > max_packet) 1072 return tipc_link_send_long_buf(l_ptr, buf); 1073 1074 /* Packet can be queued or sent: */ 1075 1076 if (queue_size > l_ptr->stats.max_queue_sz) 1077 l_ptr->stats.max_queue_sz = queue_size; 1078 1079 if (likely(!tipc_bearer_congested(l_ptr->b_ptr, l_ptr) && 1080 !link_congested(l_ptr))) { 1081 link_add_to_outqueue(l_ptr, buf, msg); 1082 1083 if (likely(tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr))) { 1084 l_ptr->unacked_window = 0; 1085 } else { 1086 tipc_bearer_schedule(l_ptr->b_ptr, l_ptr); 1087 l_ptr->stats.bearer_congs++; 1088 l_ptr->next_out = buf; 1089 } 1090 return dsz; 1091 } 1092 /* Congestion: can message be bundled ?: */ 1093 1094 if ((msg_user(msg) != CHANGEOVER_PROTOCOL) && 1095 (msg_user(msg) != MSG_FRAGMENTER)) { 1096 1097 /* Try adding message to an existing bundle */ 1098 1099 if (l_ptr->next_out && 1100 link_bundle_buf(l_ptr, l_ptr->last_out, buf)) { 1101 tipc_bearer_resolve_congestion(l_ptr->b_ptr, l_ptr); 1102 return dsz; 1103 } 1104 1105 /* Try creating a new bundle */ 1106 1107 if (size <= max_packet * 2 / 3) { 1108 struct sk_buff *bundler = buf_acquire(max_packet); 1109 struct tipc_msg bundler_hdr; 1110 1111 if (bundler) { 1112 msg_init(&bundler_hdr, MSG_BUNDLER, OPEN_MSG, 1113 TIPC_OK, INT_H_SIZE, l_ptr->addr); 1114 skb_copy_to_linear_data(bundler, &bundler_hdr, 1115 INT_H_SIZE); 1116 skb_trim(bundler, INT_H_SIZE); 1117 link_bundle_buf(l_ptr, bundler, buf); 1118 buf = bundler; 1119 msg = buf_msg(buf); 1120 l_ptr->stats.sent_bundles++; 1121 } 1122 } 1123 } 1124 if (!l_ptr->next_out) 1125 l_ptr->next_out = buf; 1126 link_add_to_outqueue(l_ptr, buf, msg); 1127 tipc_bearer_resolve_congestion(l_ptr->b_ptr, l_ptr); 1128 return dsz; 1129 } 1130 1131 /* 1132 * tipc_link_send(): same as tipc_link_send_buf(), but the link to use has 1133 * not been selected yet, and the the owner node is not locked 1134 * Called by TIPC internal users, e.g. the name distributor 1135 */ 1136 1137 int tipc_link_send(struct sk_buff *buf, u32 dest, u32 selector) 1138 { 1139 struct link *l_ptr; 1140 struct node *n_ptr; 1141 int res = -ELINKCONG; 1142 1143 read_lock_bh(&tipc_net_lock); 1144 n_ptr = tipc_node_select(dest, selector); 1145 if (n_ptr) { 1146 tipc_node_lock(n_ptr); 1147 l_ptr = n_ptr->active_links[selector & 1]; 1148 if (l_ptr) { 1149 dbg("tipc_link_send: found link %x for dest %x\n", l_ptr, dest); 1150 res = tipc_link_send_buf(l_ptr, buf); 1151 } else { 1152 dbg("Attempt to send msg to unreachable node:\n"); 1153 msg_dbg(buf_msg(buf),">>>"); 1154 buf_discard(buf); 1155 } 1156 tipc_node_unlock(n_ptr); 1157 } else { 1158 dbg("Attempt to send msg to unknown node:\n"); 1159 msg_dbg(buf_msg(buf),">>>"); 1160 buf_discard(buf); 1161 } 1162 read_unlock_bh(&tipc_net_lock); 1163 return res; 1164 } 1165 1166 /* 1167 * link_send_buf_fast: Entry for data messages where the 1168 * destination link is known and the header is complete, 1169 * inclusive total message length. Very time critical. 1170 * Link is locked. Returns user data length. 1171 */ 1172 1173 static int link_send_buf_fast(struct link *l_ptr, struct sk_buff *buf, 1174 u32 *used_max_pkt) 1175 { 1176 struct tipc_msg *msg = buf_msg(buf); 1177 int res = msg_data_sz(msg); 1178 1179 if (likely(!link_congested(l_ptr))) { 1180 if (likely(msg_size(msg) <= link_max_pkt(l_ptr))) { 1181 if (likely(list_empty(&l_ptr->b_ptr->cong_links))) { 1182 link_add_to_outqueue(l_ptr, buf, msg); 1183 if (likely(tipc_bearer_send(l_ptr->b_ptr, buf, 1184 &l_ptr->media_addr))) { 1185 l_ptr->unacked_window = 0; 1186 msg_dbg(msg,"SENT_FAST:"); 1187 return res; 1188 } 1189 dbg("failed sent fast...\n"); 1190 tipc_bearer_schedule(l_ptr->b_ptr, l_ptr); 1191 l_ptr->stats.bearer_congs++; 1192 l_ptr->next_out = buf; 1193 return res; 1194 } 1195 } 1196 else 1197 *used_max_pkt = link_max_pkt(l_ptr); 1198 } 1199 return tipc_link_send_buf(l_ptr, buf); /* All other cases */ 1200 } 1201 1202 /* 1203 * tipc_send_buf_fast: Entry for data messages where the 1204 * destination node is known and the header is complete, 1205 * inclusive total message length. 1206 * Returns user data length. 1207 */ 1208 int tipc_send_buf_fast(struct sk_buff *buf, u32 destnode) 1209 { 1210 struct link *l_ptr; 1211 struct node *n_ptr; 1212 int res; 1213 u32 selector = msg_origport(buf_msg(buf)) & 1; 1214 u32 dummy; 1215 1216 if (destnode == tipc_own_addr) 1217 return tipc_port_recv_msg(buf); 1218 1219 read_lock_bh(&tipc_net_lock); 1220 n_ptr = tipc_node_select(destnode, selector); 1221 if (likely(n_ptr)) { 1222 tipc_node_lock(n_ptr); 1223 l_ptr = n_ptr->active_links[selector]; 1224 dbg("send_fast: buf %x selected %x, destnode = %x\n", 1225 buf, l_ptr, destnode); 1226 if (likely(l_ptr)) { 1227 res = link_send_buf_fast(l_ptr, buf, &dummy); 1228 tipc_node_unlock(n_ptr); 1229 read_unlock_bh(&tipc_net_lock); 1230 return res; 1231 } 1232 tipc_node_unlock(n_ptr); 1233 } 1234 read_unlock_bh(&tipc_net_lock); 1235 res = msg_data_sz(buf_msg(buf)); 1236 tipc_reject_msg(buf, TIPC_ERR_NO_NODE); 1237 return res; 1238 } 1239 1240 1241 /* 1242 * tipc_link_send_sections_fast: Entry for messages where the 1243 * destination processor is known and the header is complete, 1244 * except for total message length. 1245 * Returns user data length or errno. 1246 */ 1247 int tipc_link_send_sections_fast(struct port *sender, 1248 struct iovec const *msg_sect, 1249 const u32 num_sect, 1250 u32 destaddr) 1251 { 1252 struct tipc_msg *hdr = &sender->publ.phdr; 1253 struct link *l_ptr; 1254 struct sk_buff *buf; 1255 struct node *node; 1256 int res; 1257 u32 selector = msg_origport(hdr) & 1; 1258 1259 again: 1260 /* 1261 * Try building message using port's max_pkt hint. 1262 * (Must not hold any locks while building message.) 1263 */ 1264 1265 res = msg_build(hdr, msg_sect, num_sect, sender->publ.max_pkt, 1266 !sender->user_port, &buf); 1267 1268 read_lock_bh(&tipc_net_lock); 1269 node = tipc_node_select(destaddr, selector); 1270 if (likely(node)) { 1271 tipc_node_lock(node); 1272 l_ptr = node->active_links[selector]; 1273 if (likely(l_ptr)) { 1274 if (likely(buf)) { 1275 res = link_send_buf_fast(l_ptr, buf, 1276 &sender->publ.max_pkt); 1277 if (unlikely(res < 0)) 1278 buf_discard(buf); 1279 exit: 1280 tipc_node_unlock(node); 1281 read_unlock_bh(&tipc_net_lock); 1282 return res; 1283 } 1284 1285 /* Exit if build request was invalid */ 1286 1287 if (unlikely(res < 0)) 1288 goto exit; 1289 1290 /* Exit if link (or bearer) is congested */ 1291 1292 if (link_congested(l_ptr) || 1293 !list_empty(&l_ptr->b_ptr->cong_links)) { 1294 res = link_schedule_port(l_ptr, 1295 sender->publ.ref, res); 1296 goto exit; 1297 } 1298 1299 /* 1300 * Message size exceeds max_pkt hint; update hint, 1301 * then re-try fast path or fragment the message 1302 */ 1303 1304 sender->publ.max_pkt = link_max_pkt(l_ptr); 1305 tipc_node_unlock(node); 1306 read_unlock_bh(&tipc_net_lock); 1307 1308 1309 if ((msg_hdr_sz(hdr) + res) <= sender->publ.max_pkt) 1310 goto again; 1311 1312 return link_send_sections_long(sender, msg_sect, 1313 num_sect, destaddr); 1314 } 1315 tipc_node_unlock(node); 1316 } 1317 read_unlock_bh(&tipc_net_lock); 1318 1319 /* Couldn't find a link to the destination node */ 1320 1321 if (buf) 1322 return tipc_reject_msg(buf, TIPC_ERR_NO_NODE); 1323 if (res >= 0) 1324 return tipc_port_reject_sections(sender, hdr, msg_sect, num_sect, 1325 TIPC_ERR_NO_NODE); 1326 return res; 1327 } 1328 1329 /* 1330 * link_send_sections_long(): Entry for long messages where the 1331 * destination node is known and the header is complete, 1332 * inclusive total message length. 1333 * Link and bearer congestion status have been checked to be ok, 1334 * and are ignored if they change. 1335 * 1336 * Note that fragments do not use the full link MTU so that they won't have 1337 * to undergo refragmentation if link changeover causes them to be sent 1338 * over another link with an additional tunnel header added as prefix. 1339 * (Refragmentation will still occur if the other link has a smaller MTU.) 1340 * 1341 * Returns user data length or errno. 1342 */ 1343 static int link_send_sections_long(struct port *sender, 1344 struct iovec const *msg_sect, 1345 u32 num_sect, 1346 u32 destaddr) 1347 { 1348 struct link *l_ptr; 1349 struct node *node; 1350 struct tipc_msg *hdr = &sender->publ.phdr; 1351 u32 dsz = msg_data_sz(hdr); 1352 u32 max_pkt,fragm_sz,rest; 1353 struct tipc_msg fragm_hdr; 1354 struct sk_buff *buf,*buf_chain,*prev; 1355 u32 fragm_crs,fragm_rest,hsz,sect_rest; 1356 const unchar *sect_crs; 1357 int curr_sect; 1358 u32 fragm_no; 1359 1360 again: 1361 fragm_no = 1; 1362 max_pkt = sender->publ.max_pkt - INT_H_SIZE; 1363 /* leave room for tunnel header in case of link changeover */ 1364 fragm_sz = max_pkt - INT_H_SIZE; 1365 /* leave room for fragmentation header in each fragment */ 1366 rest = dsz; 1367 fragm_crs = 0; 1368 fragm_rest = 0; 1369 sect_rest = 0; 1370 sect_crs = NULL; 1371 curr_sect = -1; 1372 1373 /* Prepare reusable fragment header: */ 1374 1375 msg_dbg(hdr, ">FRAGMENTING>"); 1376 msg_init(&fragm_hdr, MSG_FRAGMENTER, FIRST_FRAGMENT, 1377 TIPC_OK, INT_H_SIZE, msg_destnode(hdr)); 1378 msg_set_link_selector(&fragm_hdr, sender->publ.ref); 1379 msg_set_size(&fragm_hdr, max_pkt); 1380 msg_set_fragm_no(&fragm_hdr, 1); 1381 1382 /* Prepare header of first fragment: */ 1383 1384 buf_chain = buf = buf_acquire(max_pkt); 1385 if (!buf) 1386 return -ENOMEM; 1387 buf->next = NULL; 1388 skb_copy_to_linear_data(buf, &fragm_hdr, INT_H_SIZE); 1389 hsz = msg_hdr_sz(hdr); 1390 skb_copy_to_linear_data_offset(buf, INT_H_SIZE, hdr, hsz); 1391 msg_dbg(buf_msg(buf), ">BUILD>"); 1392 1393 /* Chop up message: */ 1394 1395 fragm_crs = INT_H_SIZE + hsz; 1396 fragm_rest = fragm_sz - hsz; 1397 1398 do { /* For all sections */ 1399 u32 sz; 1400 1401 if (!sect_rest) { 1402 sect_rest = msg_sect[++curr_sect].iov_len; 1403 sect_crs = (const unchar *)msg_sect[curr_sect].iov_base; 1404 } 1405 1406 if (sect_rest < fragm_rest) 1407 sz = sect_rest; 1408 else 1409 sz = fragm_rest; 1410 1411 if (likely(!sender->user_port)) { 1412 if (copy_from_user(buf->data + fragm_crs, sect_crs, sz)) { 1413 error: 1414 for (; buf_chain; buf_chain = buf) { 1415 buf = buf_chain->next; 1416 buf_discard(buf_chain); 1417 } 1418 return -EFAULT; 1419 } 1420 } else 1421 skb_copy_to_linear_data_offset(buf, fragm_crs, 1422 sect_crs, sz); 1423 sect_crs += sz; 1424 sect_rest -= sz; 1425 fragm_crs += sz; 1426 fragm_rest -= sz; 1427 rest -= sz; 1428 1429 if (!fragm_rest && rest) { 1430 1431 /* Initiate new fragment: */ 1432 if (rest <= fragm_sz) { 1433 fragm_sz = rest; 1434 msg_set_type(&fragm_hdr,LAST_FRAGMENT); 1435 } else { 1436 msg_set_type(&fragm_hdr, FRAGMENT); 1437 } 1438 msg_set_size(&fragm_hdr, fragm_sz + INT_H_SIZE); 1439 msg_set_fragm_no(&fragm_hdr, ++fragm_no); 1440 prev = buf; 1441 buf = buf_acquire(fragm_sz + INT_H_SIZE); 1442 if (!buf) 1443 goto error; 1444 1445 buf->next = NULL; 1446 prev->next = buf; 1447 skb_copy_to_linear_data(buf, &fragm_hdr, INT_H_SIZE); 1448 fragm_crs = INT_H_SIZE; 1449 fragm_rest = fragm_sz; 1450 msg_dbg(buf_msg(buf)," >BUILD>"); 1451 } 1452 } 1453 while (rest > 0); 1454 1455 /* 1456 * Now we have a buffer chain. Select a link and check 1457 * that packet size is still OK 1458 */ 1459 node = tipc_node_select(destaddr, sender->publ.ref & 1); 1460 if (likely(node)) { 1461 tipc_node_lock(node); 1462 l_ptr = node->active_links[sender->publ.ref & 1]; 1463 if (!l_ptr) { 1464 tipc_node_unlock(node); 1465 goto reject; 1466 } 1467 if (link_max_pkt(l_ptr) < max_pkt) { 1468 sender->publ.max_pkt = link_max_pkt(l_ptr); 1469 tipc_node_unlock(node); 1470 for (; buf_chain; buf_chain = buf) { 1471 buf = buf_chain->next; 1472 buf_discard(buf_chain); 1473 } 1474 goto again; 1475 } 1476 } else { 1477 reject: 1478 for (; buf_chain; buf_chain = buf) { 1479 buf = buf_chain->next; 1480 buf_discard(buf_chain); 1481 } 1482 return tipc_port_reject_sections(sender, hdr, msg_sect, num_sect, 1483 TIPC_ERR_NO_NODE); 1484 } 1485 1486 /* Append whole chain to send queue: */ 1487 1488 buf = buf_chain; 1489 l_ptr->long_msg_seq_no = mod(l_ptr->long_msg_seq_no + 1); 1490 if (!l_ptr->next_out) 1491 l_ptr->next_out = buf_chain; 1492 l_ptr->stats.sent_fragmented++; 1493 while (buf) { 1494 struct sk_buff *next = buf->next; 1495 struct tipc_msg *msg = buf_msg(buf); 1496 1497 l_ptr->stats.sent_fragments++; 1498 msg_set_long_msgno(msg, l_ptr->long_msg_seq_no); 1499 link_add_to_outqueue(l_ptr, buf, msg); 1500 msg_dbg(msg, ">ADD>"); 1501 buf = next; 1502 } 1503 1504 /* Send it, if possible: */ 1505 1506 tipc_link_push_queue(l_ptr); 1507 tipc_node_unlock(node); 1508 return dsz; 1509 } 1510 1511 /* 1512 * tipc_link_push_packet: Push one unsent packet to the media 1513 */ 1514 u32 tipc_link_push_packet(struct link *l_ptr) 1515 { 1516 struct sk_buff *buf = l_ptr->first_out; 1517 u32 r_q_size = l_ptr->retransm_queue_size; 1518 u32 r_q_head = l_ptr->retransm_queue_head; 1519 1520 /* Step to position where retransmission failed, if any, */ 1521 /* consider that buffers may have been released in meantime */ 1522 1523 if (r_q_size && buf) { 1524 u32 last = lesser(mod(r_q_head + r_q_size), 1525 link_last_sent(l_ptr)); 1526 u32 first = msg_seqno(buf_msg(buf)); 1527 1528 while (buf && less(first, r_q_head)) { 1529 first = mod(first + 1); 1530 buf = buf->next; 1531 } 1532 l_ptr->retransm_queue_head = r_q_head = first; 1533 l_ptr->retransm_queue_size = r_q_size = mod(last - first); 1534 } 1535 1536 /* Continue retransmission now, if there is anything: */ 1537 1538 if (r_q_size && buf && !skb_cloned(buf)) { 1539 msg_set_ack(buf_msg(buf), mod(l_ptr->next_in_no - 1)); 1540 msg_set_bcast_ack(buf_msg(buf), l_ptr->owner->bclink.last_in); 1541 if (tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr)) { 1542 msg_dbg(buf_msg(buf), ">DEF-RETR>"); 1543 l_ptr->retransm_queue_head = mod(++r_q_head); 1544 l_ptr->retransm_queue_size = --r_q_size; 1545 l_ptr->stats.retransmitted++; 1546 return TIPC_OK; 1547 } else { 1548 l_ptr->stats.bearer_congs++; 1549 msg_dbg(buf_msg(buf), "|>DEF-RETR>"); 1550 return PUSH_FAILED; 1551 } 1552 } 1553 1554 /* Send deferred protocol message, if any: */ 1555 1556 buf = l_ptr->proto_msg_queue; 1557 if (buf) { 1558 msg_set_ack(buf_msg(buf), mod(l_ptr->next_in_no - 1)); 1559 msg_set_bcast_ack(buf_msg(buf),l_ptr->owner->bclink.last_in); 1560 if (tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr)) { 1561 msg_dbg(buf_msg(buf), ">DEF-PROT>"); 1562 l_ptr->unacked_window = 0; 1563 buf_discard(buf); 1564 l_ptr->proto_msg_queue = NULL; 1565 return TIPC_OK; 1566 } else { 1567 msg_dbg(buf_msg(buf), "|>DEF-PROT>"); 1568 l_ptr->stats.bearer_congs++; 1569 return PUSH_FAILED; 1570 } 1571 } 1572 1573 /* Send one deferred data message, if send window not full: */ 1574 1575 buf = l_ptr->next_out; 1576 if (buf) { 1577 struct tipc_msg *msg = buf_msg(buf); 1578 u32 next = msg_seqno(msg); 1579 u32 first = msg_seqno(buf_msg(l_ptr->first_out)); 1580 1581 if (mod(next - first) < l_ptr->queue_limit[0]) { 1582 msg_set_ack(msg, mod(l_ptr->next_in_no - 1)); 1583 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in); 1584 if (tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr)) { 1585 if (msg_user(msg) == MSG_BUNDLER) 1586 msg_set_type(msg, CLOSED_MSG); 1587 msg_dbg(msg, ">PUSH-DATA>"); 1588 l_ptr->next_out = buf->next; 1589 return TIPC_OK; 1590 } else { 1591 msg_dbg(msg, "|PUSH-DATA|"); 1592 l_ptr->stats.bearer_congs++; 1593 return PUSH_FAILED; 1594 } 1595 } 1596 } 1597 return PUSH_FINISHED; 1598 } 1599 1600 /* 1601 * push_queue(): push out the unsent messages of a link where 1602 * congestion has abated. Node is locked 1603 */ 1604 void tipc_link_push_queue(struct link *l_ptr) 1605 { 1606 u32 res; 1607 1608 if (tipc_bearer_congested(l_ptr->b_ptr, l_ptr)) 1609 return; 1610 1611 do { 1612 res = tipc_link_push_packet(l_ptr); 1613 } 1614 while (res == TIPC_OK); 1615 if (res == PUSH_FAILED) 1616 tipc_bearer_schedule(l_ptr->b_ptr, l_ptr); 1617 } 1618 1619 static void link_reset_all(unsigned long addr) 1620 { 1621 struct node *n_ptr; 1622 char addr_string[16]; 1623 u32 i; 1624 1625 read_lock_bh(&tipc_net_lock); 1626 n_ptr = tipc_node_find((u32)addr); 1627 if (!n_ptr) { 1628 read_unlock_bh(&tipc_net_lock); 1629 return; /* node no longer exists */ 1630 } 1631 1632 tipc_node_lock(n_ptr); 1633 1634 warn("Resetting all links to %s\n", 1635 addr_string_fill(addr_string, n_ptr->addr)); 1636 1637 for (i = 0; i < MAX_BEARERS; i++) { 1638 if (n_ptr->links[i]) { 1639 link_print(n_ptr->links[i], TIPC_OUTPUT, 1640 "Resetting link\n"); 1641 tipc_link_reset(n_ptr->links[i]); 1642 } 1643 } 1644 1645 tipc_node_unlock(n_ptr); 1646 read_unlock_bh(&tipc_net_lock); 1647 } 1648 1649 static void link_retransmit_failure(struct link *l_ptr, struct sk_buff *buf) 1650 { 1651 struct tipc_msg *msg = buf_msg(buf); 1652 1653 warn("Retransmission failure on link <%s>\n", l_ptr->name); 1654 tipc_msg_print(TIPC_OUTPUT, msg, ">RETR-FAIL>"); 1655 1656 if (l_ptr->addr) { 1657 1658 /* Handle failure on standard link */ 1659 1660 link_print(l_ptr, TIPC_OUTPUT, "Resetting link\n"); 1661 tipc_link_reset(l_ptr); 1662 1663 } else { 1664 1665 /* Handle failure on broadcast link */ 1666 1667 struct node *n_ptr; 1668 char addr_string[16]; 1669 1670 tipc_printf(TIPC_OUTPUT, "Msg seq number: %u, ", msg_seqno(msg)); 1671 tipc_printf(TIPC_OUTPUT, "Outstanding acks: %lu\n", 1672 (unsigned long) TIPC_SKB_CB(buf)->handle); 1673 1674 n_ptr = l_ptr->owner->next; 1675 tipc_node_lock(n_ptr); 1676 1677 addr_string_fill(addr_string, n_ptr->addr); 1678 tipc_printf(TIPC_OUTPUT, "Multicast link info for %s\n", addr_string); 1679 tipc_printf(TIPC_OUTPUT, "Supported: %d, ", n_ptr->bclink.supported); 1680 tipc_printf(TIPC_OUTPUT, "Acked: %u\n", n_ptr->bclink.acked); 1681 tipc_printf(TIPC_OUTPUT, "Last in: %u, ", n_ptr->bclink.last_in); 1682 tipc_printf(TIPC_OUTPUT, "Gap after: %u, ", n_ptr->bclink.gap_after); 1683 tipc_printf(TIPC_OUTPUT, "Gap to: %u\n", n_ptr->bclink.gap_to); 1684 tipc_printf(TIPC_OUTPUT, "Nack sync: %u\n\n", n_ptr->bclink.nack_sync); 1685 1686 tipc_k_signal((Handler)link_reset_all, (unsigned long)n_ptr->addr); 1687 1688 tipc_node_unlock(n_ptr); 1689 1690 l_ptr->stale_count = 0; 1691 } 1692 } 1693 1694 void tipc_link_retransmit(struct link *l_ptr, struct sk_buff *buf, 1695 u32 retransmits) 1696 { 1697 struct tipc_msg *msg; 1698 1699 if (!buf) 1700 return; 1701 1702 msg = buf_msg(buf); 1703 1704 dbg("Retransmitting %u in link %x\n", retransmits, l_ptr); 1705 1706 if (tipc_bearer_congested(l_ptr->b_ptr, l_ptr)) { 1707 if (!skb_cloned(buf)) { 1708 msg_dbg(msg, ">NO_RETR->BCONG>"); 1709 dbg_print_link(l_ptr, " "); 1710 l_ptr->retransm_queue_head = msg_seqno(msg); 1711 l_ptr->retransm_queue_size = retransmits; 1712 return; 1713 } else { 1714 /* Don't retransmit if driver already has the buffer */ 1715 } 1716 } else { 1717 /* Detect repeated retransmit failures on uncongested bearer */ 1718 1719 if (l_ptr->last_retransmitted == msg_seqno(msg)) { 1720 if (++l_ptr->stale_count > 100) { 1721 link_retransmit_failure(l_ptr, buf); 1722 return; 1723 } 1724 } else { 1725 l_ptr->last_retransmitted = msg_seqno(msg); 1726 l_ptr->stale_count = 1; 1727 } 1728 } 1729 1730 while (retransmits && (buf != l_ptr->next_out) && buf && !skb_cloned(buf)) { 1731 msg = buf_msg(buf); 1732 msg_set_ack(msg, mod(l_ptr->next_in_no - 1)); 1733 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in); 1734 if (tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr)) { 1735 msg_dbg(buf_msg(buf), ">RETR>"); 1736 buf = buf->next; 1737 retransmits--; 1738 l_ptr->stats.retransmitted++; 1739 } else { 1740 tipc_bearer_schedule(l_ptr->b_ptr, l_ptr); 1741 l_ptr->stats.bearer_congs++; 1742 l_ptr->retransm_queue_head = msg_seqno(buf_msg(buf)); 1743 l_ptr->retransm_queue_size = retransmits; 1744 return; 1745 } 1746 } 1747 1748 l_ptr->retransm_queue_head = l_ptr->retransm_queue_size = 0; 1749 } 1750 1751 /* 1752 * link_recv_non_seq: Receive packets which are outside 1753 * the link sequence flow 1754 */ 1755 1756 static void link_recv_non_seq(struct sk_buff *buf) 1757 { 1758 struct tipc_msg *msg = buf_msg(buf); 1759 1760 if (msg_user(msg) == LINK_CONFIG) 1761 tipc_disc_recv_msg(buf); 1762 else 1763 tipc_bclink_recv_pkt(buf); 1764 } 1765 1766 /** 1767 * link_insert_deferred_queue - insert deferred messages back into receive chain 1768 */ 1769 1770 static struct sk_buff *link_insert_deferred_queue(struct link *l_ptr, 1771 struct sk_buff *buf) 1772 { 1773 u32 seq_no; 1774 1775 if (l_ptr->oldest_deferred_in == NULL) 1776 return buf; 1777 1778 seq_no = msg_seqno(buf_msg(l_ptr->oldest_deferred_in)); 1779 if (seq_no == mod(l_ptr->next_in_no)) { 1780 l_ptr->newest_deferred_in->next = buf; 1781 buf = l_ptr->oldest_deferred_in; 1782 l_ptr->oldest_deferred_in = NULL; 1783 l_ptr->deferred_inqueue_sz = 0; 1784 } 1785 return buf; 1786 } 1787 1788 /** 1789 * link_recv_buf_validate - validate basic format of received message 1790 * 1791 * This routine ensures a TIPC message has an acceptable header, and at least 1792 * as much data as the header indicates it should. The routine also ensures 1793 * that the entire message header is stored in the main fragment of the message 1794 * buffer, to simplify future access to message header fields. 1795 * 1796 * Note: Having extra info present in the message header or data areas is OK. 1797 * TIPC will ignore the excess, under the assumption that it is optional info 1798 * introduced by a later release of the protocol. 1799 */ 1800 1801 static int link_recv_buf_validate(struct sk_buff *buf) 1802 { 1803 static u32 min_data_hdr_size[8] = { 1804 SHORT_H_SIZE, MCAST_H_SIZE, LONG_H_SIZE, DIR_MSG_H_SIZE, 1805 MAX_H_SIZE, MAX_H_SIZE, MAX_H_SIZE, MAX_H_SIZE 1806 }; 1807 1808 struct tipc_msg *msg; 1809 u32 tipc_hdr[2]; 1810 u32 size; 1811 u32 hdr_size; 1812 u32 min_hdr_size; 1813 1814 if (unlikely(buf->len < MIN_H_SIZE)) 1815 return 0; 1816 1817 msg = skb_header_pointer(buf, 0, sizeof(tipc_hdr), tipc_hdr); 1818 if (msg == NULL) 1819 return 0; 1820 1821 if (unlikely(msg_version(msg) != TIPC_VERSION)) 1822 return 0; 1823 1824 size = msg_size(msg); 1825 hdr_size = msg_hdr_sz(msg); 1826 min_hdr_size = msg_isdata(msg) ? 1827 min_data_hdr_size[msg_type(msg)] : INT_H_SIZE; 1828 1829 if (unlikely((hdr_size < min_hdr_size) || 1830 (size < hdr_size) || 1831 (buf->len < size) || 1832 (size - hdr_size > TIPC_MAX_USER_MSG_SIZE))) 1833 return 0; 1834 1835 return pskb_may_pull(buf, hdr_size); 1836 } 1837 1838 void tipc_recv_msg(struct sk_buff *head, struct tipc_bearer *tb_ptr) 1839 { 1840 read_lock_bh(&tipc_net_lock); 1841 while (head) { 1842 struct bearer *b_ptr; 1843 struct node *n_ptr; 1844 struct link *l_ptr; 1845 struct sk_buff *crs; 1846 struct sk_buff *buf = head; 1847 struct tipc_msg *msg; 1848 u32 seq_no; 1849 u32 ackd; 1850 u32 released = 0; 1851 int type; 1852 1853 b_ptr = (struct bearer *)tb_ptr; 1854 TIPC_SKB_CB(buf)->handle = b_ptr; 1855 1856 head = head->next; 1857 1858 /* Ensure message is well-formed */ 1859 1860 if (unlikely(!link_recv_buf_validate(buf))) 1861 goto cont; 1862 1863 /* Ensure message data is a single contiguous unit */ 1864 1865 if (unlikely(buf_linearize(buf))) { 1866 goto cont; 1867 } 1868 1869 /* Handle arrival of a non-unicast link message */ 1870 1871 msg = buf_msg(buf); 1872 1873 if (unlikely(msg_non_seq(msg))) { 1874 link_recv_non_seq(buf); 1875 continue; 1876 } 1877 1878 if (unlikely(!msg_short(msg) && 1879 (msg_destnode(msg) != tipc_own_addr))) 1880 goto cont; 1881 1882 /* Locate unicast link endpoint that should handle message */ 1883 1884 n_ptr = tipc_node_find(msg_prevnode(msg)); 1885 if (unlikely(!n_ptr)) 1886 goto cont; 1887 tipc_node_lock(n_ptr); 1888 1889 l_ptr = n_ptr->links[b_ptr->identity]; 1890 if (unlikely(!l_ptr)) { 1891 tipc_node_unlock(n_ptr); 1892 goto cont; 1893 } 1894 1895 /* Validate message sequence number info */ 1896 1897 seq_no = msg_seqno(msg); 1898 ackd = msg_ack(msg); 1899 1900 /* Release acked messages */ 1901 1902 if (less(n_ptr->bclink.acked, msg_bcast_ack(msg))) { 1903 if (tipc_node_is_up(n_ptr) && n_ptr->bclink.supported) 1904 tipc_bclink_acknowledge(n_ptr, msg_bcast_ack(msg)); 1905 } 1906 1907 crs = l_ptr->first_out; 1908 while ((crs != l_ptr->next_out) && 1909 less_eq(msg_seqno(buf_msg(crs)), ackd)) { 1910 struct sk_buff *next = crs->next; 1911 1912 buf_discard(crs); 1913 crs = next; 1914 released++; 1915 } 1916 if (released) { 1917 l_ptr->first_out = crs; 1918 l_ptr->out_queue_size -= released; 1919 } 1920 1921 /* Try sending any messages link endpoint has pending */ 1922 1923 if (unlikely(l_ptr->next_out)) 1924 tipc_link_push_queue(l_ptr); 1925 if (unlikely(!list_empty(&l_ptr->waiting_ports))) 1926 tipc_link_wakeup_ports(l_ptr, 0); 1927 if (unlikely(++l_ptr->unacked_window >= TIPC_MIN_LINK_WIN)) { 1928 l_ptr->stats.sent_acks++; 1929 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 0, 0, 0, 0, 0); 1930 } 1931 1932 /* Now (finally!) process the incoming message */ 1933 1934 protocol_check: 1935 if (likely(link_working_working(l_ptr))) { 1936 if (likely(seq_no == mod(l_ptr->next_in_no))) { 1937 l_ptr->next_in_no++; 1938 if (unlikely(l_ptr->oldest_deferred_in)) 1939 head = link_insert_deferred_queue(l_ptr, 1940 head); 1941 if (likely(msg_is_dest(msg, tipc_own_addr))) { 1942 deliver: 1943 if (likely(msg_isdata(msg))) { 1944 tipc_node_unlock(n_ptr); 1945 tipc_port_recv_msg(buf); 1946 continue; 1947 } 1948 switch (msg_user(msg)) { 1949 case MSG_BUNDLER: 1950 l_ptr->stats.recv_bundles++; 1951 l_ptr->stats.recv_bundled += 1952 msg_msgcnt(msg); 1953 tipc_node_unlock(n_ptr); 1954 tipc_link_recv_bundle(buf); 1955 continue; 1956 case ROUTE_DISTRIBUTOR: 1957 tipc_node_unlock(n_ptr); 1958 tipc_cltr_recv_routing_table(buf); 1959 continue; 1960 case NAME_DISTRIBUTOR: 1961 tipc_node_unlock(n_ptr); 1962 tipc_named_recv(buf); 1963 continue; 1964 case CONN_MANAGER: 1965 tipc_node_unlock(n_ptr); 1966 tipc_port_recv_proto_msg(buf); 1967 continue; 1968 case MSG_FRAGMENTER: 1969 l_ptr->stats.recv_fragments++; 1970 if (tipc_link_recv_fragment(&l_ptr->defragm_buf, 1971 &buf, &msg)) { 1972 l_ptr->stats.recv_fragmented++; 1973 goto deliver; 1974 } 1975 break; 1976 case CHANGEOVER_PROTOCOL: 1977 type = msg_type(msg); 1978 if (link_recv_changeover_msg(&l_ptr, &buf)) { 1979 msg = buf_msg(buf); 1980 seq_no = msg_seqno(msg); 1981 TIPC_SKB_CB(buf)->handle 1982 = b_ptr; 1983 if (type == ORIGINAL_MSG) 1984 goto deliver; 1985 goto protocol_check; 1986 } 1987 break; 1988 } 1989 } 1990 tipc_node_unlock(n_ptr); 1991 tipc_net_route_msg(buf); 1992 continue; 1993 } 1994 link_handle_out_of_seq_msg(l_ptr, buf); 1995 head = link_insert_deferred_queue(l_ptr, head); 1996 tipc_node_unlock(n_ptr); 1997 continue; 1998 } 1999 2000 if (msg_user(msg) == LINK_PROTOCOL) { 2001 link_recv_proto_msg(l_ptr, buf); 2002 head = link_insert_deferred_queue(l_ptr, head); 2003 tipc_node_unlock(n_ptr); 2004 continue; 2005 } 2006 msg_dbg(msg,"NSEQ<REC<"); 2007 link_state_event(l_ptr, TRAFFIC_MSG_EVT); 2008 2009 if (link_working_working(l_ptr)) { 2010 /* Re-insert in front of queue */ 2011 msg_dbg(msg,"RECV-REINS:"); 2012 buf->next = head; 2013 head = buf; 2014 tipc_node_unlock(n_ptr); 2015 continue; 2016 } 2017 tipc_node_unlock(n_ptr); 2018 cont: 2019 buf_discard(buf); 2020 } 2021 read_unlock_bh(&tipc_net_lock); 2022 } 2023 2024 /* 2025 * link_defer_buf(): Sort a received out-of-sequence packet 2026 * into the deferred reception queue. 2027 * Returns the increase of the queue length,i.e. 0 or 1 2028 */ 2029 2030 u32 tipc_link_defer_pkt(struct sk_buff **head, 2031 struct sk_buff **tail, 2032 struct sk_buff *buf) 2033 { 2034 struct sk_buff *prev = NULL; 2035 struct sk_buff *crs = *head; 2036 u32 seq_no = msg_seqno(buf_msg(buf)); 2037 2038 buf->next = NULL; 2039 2040 /* Empty queue ? */ 2041 if (*head == NULL) { 2042 *head = *tail = buf; 2043 return 1; 2044 } 2045 2046 /* Last ? */ 2047 if (less(msg_seqno(buf_msg(*tail)), seq_no)) { 2048 (*tail)->next = buf; 2049 *tail = buf; 2050 return 1; 2051 } 2052 2053 /* Scan through queue and sort it in */ 2054 do { 2055 struct tipc_msg *msg = buf_msg(crs); 2056 2057 if (less(seq_no, msg_seqno(msg))) { 2058 buf->next = crs; 2059 if (prev) 2060 prev->next = buf; 2061 else 2062 *head = buf; 2063 return 1; 2064 } 2065 if (seq_no == msg_seqno(msg)) { 2066 break; 2067 } 2068 prev = crs; 2069 crs = crs->next; 2070 } 2071 while (crs); 2072 2073 /* Message is a duplicate of an existing message */ 2074 2075 buf_discard(buf); 2076 return 0; 2077 } 2078 2079 /** 2080 * link_handle_out_of_seq_msg - handle arrival of out-of-sequence packet 2081 */ 2082 2083 static void link_handle_out_of_seq_msg(struct link *l_ptr, 2084 struct sk_buff *buf) 2085 { 2086 u32 seq_no = msg_seqno(buf_msg(buf)); 2087 2088 if (likely(msg_user(buf_msg(buf)) == LINK_PROTOCOL)) { 2089 link_recv_proto_msg(l_ptr, buf); 2090 return; 2091 } 2092 2093 dbg("rx OOS msg: seq_no %u, expecting %u (%u)\n", 2094 seq_no, mod(l_ptr->next_in_no), l_ptr->next_in_no); 2095 2096 /* Record OOS packet arrival (force mismatch on next timeout) */ 2097 2098 l_ptr->checkpoint--; 2099 2100 /* 2101 * Discard packet if a duplicate; otherwise add it to deferred queue 2102 * and notify peer of gap as per protocol specification 2103 */ 2104 2105 if (less(seq_no, mod(l_ptr->next_in_no))) { 2106 l_ptr->stats.duplicates++; 2107 buf_discard(buf); 2108 return; 2109 } 2110 2111 if (tipc_link_defer_pkt(&l_ptr->oldest_deferred_in, 2112 &l_ptr->newest_deferred_in, buf)) { 2113 l_ptr->deferred_inqueue_sz++; 2114 l_ptr->stats.deferred_recv++; 2115 if ((l_ptr->deferred_inqueue_sz % 16) == 1) 2116 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 0, 0, 0, 0, 0); 2117 } else 2118 l_ptr->stats.duplicates++; 2119 } 2120 2121 /* 2122 * Send protocol message to the other endpoint. 2123 */ 2124 void tipc_link_send_proto_msg(struct link *l_ptr, u32 msg_typ, int probe_msg, 2125 u32 gap, u32 tolerance, u32 priority, u32 ack_mtu) 2126 { 2127 struct sk_buff *buf = NULL; 2128 struct tipc_msg *msg = l_ptr->pmsg; 2129 u32 msg_size = sizeof(l_ptr->proto_msg); 2130 2131 if (link_blocked(l_ptr)) 2132 return; 2133 msg_set_type(msg, msg_typ); 2134 msg_set_net_plane(msg, l_ptr->b_ptr->net_plane); 2135 msg_set_bcast_ack(msg, mod(l_ptr->owner->bclink.last_in)); 2136 msg_set_last_bcast(msg, tipc_bclink_get_last_sent()); 2137 2138 if (msg_typ == STATE_MSG) { 2139 u32 next_sent = mod(l_ptr->next_out_no); 2140 2141 if (!tipc_link_is_up(l_ptr)) 2142 return; 2143 if (l_ptr->next_out) 2144 next_sent = msg_seqno(buf_msg(l_ptr->next_out)); 2145 msg_set_next_sent(msg, next_sent); 2146 if (l_ptr->oldest_deferred_in) { 2147 u32 rec = msg_seqno(buf_msg(l_ptr->oldest_deferred_in)); 2148 gap = mod(rec - mod(l_ptr->next_in_no)); 2149 } 2150 msg_set_seq_gap(msg, gap); 2151 if (gap) 2152 l_ptr->stats.sent_nacks++; 2153 msg_set_link_tolerance(msg, tolerance); 2154 msg_set_linkprio(msg, priority); 2155 msg_set_max_pkt(msg, ack_mtu); 2156 msg_set_ack(msg, mod(l_ptr->next_in_no - 1)); 2157 msg_set_probe(msg, probe_msg != 0); 2158 if (probe_msg) { 2159 u32 mtu = l_ptr->max_pkt; 2160 2161 if ((mtu < l_ptr->max_pkt_target) && 2162 link_working_working(l_ptr) && 2163 l_ptr->fsm_msg_cnt) { 2164 msg_size = (mtu + (l_ptr->max_pkt_target - mtu)/2 + 2) & ~3; 2165 if (l_ptr->max_pkt_probes == 10) { 2166 l_ptr->max_pkt_target = (msg_size - 4); 2167 l_ptr->max_pkt_probes = 0; 2168 msg_size = (mtu + (l_ptr->max_pkt_target - mtu)/2 + 2) & ~3; 2169 } 2170 l_ptr->max_pkt_probes++; 2171 } 2172 2173 l_ptr->stats.sent_probes++; 2174 } 2175 l_ptr->stats.sent_states++; 2176 } else { /* RESET_MSG or ACTIVATE_MSG */ 2177 msg_set_ack(msg, mod(l_ptr->reset_checkpoint - 1)); 2178 msg_set_seq_gap(msg, 0); 2179 msg_set_next_sent(msg, 1); 2180 msg_set_link_tolerance(msg, l_ptr->tolerance); 2181 msg_set_linkprio(msg, l_ptr->priority); 2182 msg_set_max_pkt(msg, l_ptr->max_pkt_target); 2183 } 2184 2185 if (tipc_node_has_redundant_links(l_ptr->owner)) { 2186 msg_set_redundant_link(msg); 2187 } else { 2188 msg_clear_redundant_link(msg); 2189 } 2190 msg_set_linkprio(msg, l_ptr->priority); 2191 2192 /* Ensure sequence number will not fit : */ 2193 2194 msg_set_seqno(msg, mod(l_ptr->next_out_no + (0xffff/2))); 2195 2196 /* Congestion? */ 2197 2198 if (tipc_bearer_congested(l_ptr->b_ptr, l_ptr)) { 2199 if (!l_ptr->proto_msg_queue) { 2200 l_ptr->proto_msg_queue = 2201 buf_acquire(sizeof(l_ptr->proto_msg)); 2202 } 2203 buf = l_ptr->proto_msg_queue; 2204 if (!buf) 2205 return; 2206 skb_copy_to_linear_data(buf, msg, sizeof(l_ptr->proto_msg)); 2207 return; 2208 } 2209 msg_set_timestamp(msg, jiffies_to_msecs(jiffies)); 2210 2211 /* Message can be sent */ 2212 2213 msg_dbg(msg, ">>"); 2214 2215 buf = buf_acquire(msg_size); 2216 if (!buf) 2217 return; 2218 2219 skb_copy_to_linear_data(buf, msg, sizeof(l_ptr->proto_msg)); 2220 msg_set_size(buf_msg(buf), msg_size); 2221 2222 if (tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr)) { 2223 l_ptr->unacked_window = 0; 2224 buf_discard(buf); 2225 return; 2226 } 2227 2228 /* New congestion */ 2229 tipc_bearer_schedule(l_ptr->b_ptr, l_ptr); 2230 l_ptr->proto_msg_queue = buf; 2231 l_ptr->stats.bearer_congs++; 2232 } 2233 2234 /* 2235 * Receive protocol message : 2236 * Note that network plane id propagates through the network, and may 2237 * change at any time. The node with lowest address rules 2238 */ 2239 2240 static void link_recv_proto_msg(struct link *l_ptr, struct sk_buff *buf) 2241 { 2242 u32 rec_gap = 0; 2243 u32 max_pkt_info; 2244 u32 max_pkt_ack; 2245 u32 msg_tol; 2246 struct tipc_msg *msg = buf_msg(buf); 2247 2248 dbg("AT(%u):", jiffies_to_msecs(jiffies)); 2249 msg_dbg(msg, "<<"); 2250 if (link_blocked(l_ptr)) 2251 goto exit; 2252 2253 /* record unnumbered packet arrival (force mismatch on next timeout) */ 2254 2255 l_ptr->checkpoint--; 2256 2257 if (l_ptr->b_ptr->net_plane != msg_net_plane(msg)) 2258 if (tipc_own_addr > msg_prevnode(msg)) 2259 l_ptr->b_ptr->net_plane = msg_net_plane(msg); 2260 2261 l_ptr->owner->permit_changeover = msg_redundant_link(msg); 2262 2263 switch (msg_type(msg)) { 2264 2265 case RESET_MSG: 2266 if (!link_working_unknown(l_ptr) && l_ptr->peer_session) { 2267 if (msg_session(msg) == l_ptr->peer_session) { 2268 dbg("Duplicate RESET: %u<->%u\n", 2269 msg_session(msg), l_ptr->peer_session); 2270 break; /* duplicate: ignore */ 2271 } 2272 } 2273 /* fall thru' */ 2274 case ACTIVATE_MSG: 2275 /* Update link settings according other endpoint's values */ 2276 2277 strcpy((strrchr(l_ptr->name, ':') + 1), (char *)msg_data(msg)); 2278 2279 if ((msg_tol = msg_link_tolerance(msg)) && 2280 (msg_tol > l_ptr->tolerance)) 2281 link_set_supervision_props(l_ptr, msg_tol); 2282 2283 if (msg_linkprio(msg) > l_ptr->priority) 2284 l_ptr->priority = msg_linkprio(msg); 2285 2286 max_pkt_info = msg_max_pkt(msg); 2287 if (max_pkt_info) { 2288 if (max_pkt_info < l_ptr->max_pkt_target) 2289 l_ptr->max_pkt_target = max_pkt_info; 2290 if (l_ptr->max_pkt > l_ptr->max_pkt_target) 2291 l_ptr->max_pkt = l_ptr->max_pkt_target; 2292 } else { 2293 l_ptr->max_pkt = l_ptr->max_pkt_target; 2294 } 2295 l_ptr->owner->bclink.supported = (max_pkt_info != 0); 2296 2297 link_state_event(l_ptr, msg_type(msg)); 2298 2299 l_ptr->peer_session = msg_session(msg); 2300 l_ptr->peer_bearer_id = msg_bearer_id(msg); 2301 2302 /* Synchronize broadcast sequence numbers */ 2303 if (!tipc_node_has_redundant_links(l_ptr->owner)) { 2304 l_ptr->owner->bclink.last_in = mod(msg_last_bcast(msg)); 2305 } 2306 break; 2307 case STATE_MSG: 2308 2309 if ((msg_tol = msg_link_tolerance(msg))) 2310 link_set_supervision_props(l_ptr, msg_tol); 2311 2312 if (msg_linkprio(msg) && 2313 (msg_linkprio(msg) != l_ptr->priority)) { 2314 warn("Resetting link <%s>, priority change %u->%u\n", 2315 l_ptr->name, l_ptr->priority, msg_linkprio(msg)); 2316 l_ptr->priority = msg_linkprio(msg); 2317 tipc_link_reset(l_ptr); /* Enforce change to take effect */ 2318 break; 2319 } 2320 link_state_event(l_ptr, TRAFFIC_MSG_EVT); 2321 l_ptr->stats.recv_states++; 2322 if (link_reset_unknown(l_ptr)) 2323 break; 2324 2325 if (less_eq(mod(l_ptr->next_in_no), msg_next_sent(msg))) { 2326 rec_gap = mod(msg_next_sent(msg) - 2327 mod(l_ptr->next_in_no)); 2328 } 2329 2330 max_pkt_ack = msg_max_pkt(msg); 2331 if (max_pkt_ack > l_ptr->max_pkt) { 2332 dbg("Link <%s> updated MTU %u -> %u\n", 2333 l_ptr->name, l_ptr->max_pkt, max_pkt_ack); 2334 l_ptr->max_pkt = max_pkt_ack; 2335 l_ptr->max_pkt_probes = 0; 2336 } 2337 2338 max_pkt_ack = 0; 2339 if (msg_probe(msg)) { 2340 l_ptr->stats.recv_probes++; 2341 if (msg_size(msg) > sizeof(l_ptr->proto_msg)) { 2342 max_pkt_ack = msg_size(msg); 2343 } 2344 } 2345 2346 /* Protocol message before retransmits, reduce loss risk */ 2347 2348 tipc_bclink_check_gap(l_ptr->owner, msg_last_bcast(msg)); 2349 2350 if (rec_gap || (msg_probe(msg))) { 2351 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 2352 0, rec_gap, 0, 0, max_pkt_ack); 2353 } 2354 if (msg_seq_gap(msg)) { 2355 msg_dbg(msg, "With Gap:"); 2356 l_ptr->stats.recv_nacks++; 2357 tipc_link_retransmit(l_ptr, l_ptr->first_out, 2358 msg_seq_gap(msg)); 2359 } 2360 break; 2361 default: 2362 msg_dbg(buf_msg(buf), "<DISCARDING UNKNOWN<"); 2363 } 2364 exit: 2365 buf_discard(buf); 2366 } 2367 2368 2369 /* 2370 * tipc_link_tunnel(): Send one message via a link belonging to 2371 * another bearer. Owner node is locked. 2372 */ 2373 void tipc_link_tunnel(struct link *l_ptr, 2374 struct tipc_msg *tunnel_hdr, 2375 struct tipc_msg *msg, 2376 u32 selector) 2377 { 2378 struct link *tunnel; 2379 struct sk_buff *buf; 2380 u32 length = msg_size(msg); 2381 2382 tunnel = l_ptr->owner->active_links[selector & 1]; 2383 if (!tipc_link_is_up(tunnel)) { 2384 warn("Link changeover error, " 2385 "tunnel link no longer available\n"); 2386 return; 2387 } 2388 msg_set_size(tunnel_hdr, length + INT_H_SIZE); 2389 buf = buf_acquire(length + INT_H_SIZE); 2390 if (!buf) { 2391 warn("Link changeover error, " 2392 "unable to send tunnel msg\n"); 2393 return; 2394 } 2395 skb_copy_to_linear_data(buf, tunnel_hdr, INT_H_SIZE); 2396 skb_copy_to_linear_data_offset(buf, INT_H_SIZE, msg, length); 2397 dbg("%c->%c:", l_ptr->b_ptr->net_plane, tunnel->b_ptr->net_plane); 2398 msg_dbg(buf_msg(buf), ">SEND>"); 2399 tipc_link_send_buf(tunnel, buf); 2400 } 2401 2402 2403 2404 /* 2405 * changeover(): Send whole message queue via the remaining link 2406 * Owner node is locked. 2407 */ 2408 2409 void tipc_link_changeover(struct link *l_ptr) 2410 { 2411 u32 msgcount = l_ptr->out_queue_size; 2412 struct sk_buff *crs = l_ptr->first_out; 2413 struct link *tunnel = l_ptr->owner->active_links[0]; 2414 struct tipc_msg tunnel_hdr; 2415 int split_bundles; 2416 2417 if (!tunnel) 2418 return; 2419 2420 if (!l_ptr->owner->permit_changeover) { 2421 warn("Link changeover error, " 2422 "peer did not permit changeover\n"); 2423 return; 2424 } 2425 2426 msg_init(&tunnel_hdr, CHANGEOVER_PROTOCOL, 2427 ORIGINAL_MSG, TIPC_OK, INT_H_SIZE, l_ptr->addr); 2428 msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id); 2429 msg_set_msgcnt(&tunnel_hdr, msgcount); 2430 dbg("Link changeover requires %u tunnel messages\n", msgcount); 2431 2432 if (!l_ptr->first_out) { 2433 struct sk_buff *buf; 2434 2435 buf = buf_acquire(INT_H_SIZE); 2436 if (buf) { 2437 skb_copy_to_linear_data(buf, &tunnel_hdr, INT_H_SIZE); 2438 msg_set_size(&tunnel_hdr, INT_H_SIZE); 2439 dbg("%c->%c:", l_ptr->b_ptr->net_plane, 2440 tunnel->b_ptr->net_plane); 2441 msg_dbg(&tunnel_hdr, "EMPTY>SEND>"); 2442 tipc_link_send_buf(tunnel, buf); 2443 } else { 2444 warn("Link changeover error, " 2445 "unable to send changeover msg\n"); 2446 } 2447 return; 2448 } 2449 2450 split_bundles = (l_ptr->owner->active_links[0] != 2451 l_ptr->owner->active_links[1]); 2452 2453 while (crs) { 2454 struct tipc_msg *msg = buf_msg(crs); 2455 2456 if ((msg_user(msg) == MSG_BUNDLER) && split_bundles) { 2457 struct tipc_msg *m = msg_get_wrapped(msg); 2458 unchar* pos = (unchar*)m; 2459 2460 msgcount = msg_msgcnt(msg); 2461 while (msgcount--) { 2462 msg_set_seqno(m,msg_seqno(msg)); 2463 tipc_link_tunnel(l_ptr, &tunnel_hdr, m, 2464 msg_link_selector(m)); 2465 pos += align(msg_size(m)); 2466 m = (struct tipc_msg *)pos; 2467 } 2468 } else { 2469 tipc_link_tunnel(l_ptr, &tunnel_hdr, msg, 2470 msg_link_selector(msg)); 2471 } 2472 crs = crs->next; 2473 } 2474 } 2475 2476 void tipc_link_send_duplicate(struct link *l_ptr, struct link *tunnel) 2477 { 2478 struct sk_buff *iter; 2479 struct tipc_msg tunnel_hdr; 2480 2481 msg_init(&tunnel_hdr, CHANGEOVER_PROTOCOL, 2482 DUPLICATE_MSG, TIPC_OK, INT_H_SIZE, l_ptr->addr); 2483 msg_set_msgcnt(&tunnel_hdr, l_ptr->out_queue_size); 2484 msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id); 2485 iter = l_ptr->first_out; 2486 while (iter) { 2487 struct sk_buff *outbuf; 2488 struct tipc_msg *msg = buf_msg(iter); 2489 u32 length = msg_size(msg); 2490 2491 if (msg_user(msg) == MSG_BUNDLER) 2492 msg_set_type(msg, CLOSED_MSG); 2493 msg_set_ack(msg, mod(l_ptr->next_in_no - 1)); /* Update */ 2494 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in); 2495 msg_set_size(&tunnel_hdr, length + INT_H_SIZE); 2496 outbuf = buf_acquire(length + INT_H_SIZE); 2497 if (outbuf == NULL) { 2498 warn("Link changeover error, " 2499 "unable to send duplicate msg\n"); 2500 return; 2501 } 2502 skb_copy_to_linear_data(outbuf, &tunnel_hdr, INT_H_SIZE); 2503 skb_copy_to_linear_data_offset(outbuf, INT_H_SIZE, iter->data, 2504 length); 2505 dbg("%c->%c:", l_ptr->b_ptr->net_plane, 2506 tunnel->b_ptr->net_plane); 2507 msg_dbg(buf_msg(outbuf), ">SEND>"); 2508 tipc_link_send_buf(tunnel, outbuf); 2509 if (!tipc_link_is_up(l_ptr)) 2510 return; 2511 iter = iter->next; 2512 } 2513 } 2514 2515 2516 2517 /** 2518 * buf_extract - extracts embedded TIPC message from another message 2519 * @skb: encapsulating message buffer 2520 * @from_pos: offset to extract from 2521 * 2522 * Returns a new message buffer containing an embedded message. The 2523 * encapsulating message itself is left unchanged. 2524 */ 2525 2526 static struct sk_buff *buf_extract(struct sk_buff *skb, u32 from_pos) 2527 { 2528 struct tipc_msg *msg = (struct tipc_msg *)(skb->data + from_pos); 2529 u32 size = msg_size(msg); 2530 struct sk_buff *eb; 2531 2532 eb = buf_acquire(size); 2533 if (eb) 2534 skb_copy_to_linear_data(eb, msg, size); 2535 return eb; 2536 } 2537 2538 /* 2539 * link_recv_changeover_msg(): Receive tunneled packet sent 2540 * via other link. Node is locked. Return extracted buffer. 2541 */ 2542 2543 static int link_recv_changeover_msg(struct link **l_ptr, 2544 struct sk_buff **buf) 2545 { 2546 struct sk_buff *tunnel_buf = *buf; 2547 struct link *dest_link; 2548 struct tipc_msg *msg; 2549 struct tipc_msg *tunnel_msg = buf_msg(tunnel_buf); 2550 u32 msg_typ = msg_type(tunnel_msg); 2551 u32 msg_count = msg_msgcnt(tunnel_msg); 2552 2553 dest_link = (*l_ptr)->owner->links[msg_bearer_id(tunnel_msg)]; 2554 if (!dest_link) { 2555 msg_dbg(tunnel_msg, "NOLINK/<REC<"); 2556 goto exit; 2557 } 2558 if (dest_link == *l_ptr) { 2559 err("Unexpected changeover message on link <%s>\n", 2560 (*l_ptr)->name); 2561 goto exit; 2562 } 2563 dbg("%c<-%c:", dest_link->b_ptr->net_plane, 2564 (*l_ptr)->b_ptr->net_plane); 2565 *l_ptr = dest_link; 2566 msg = msg_get_wrapped(tunnel_msg); 2567 2568 if (msg_typ == DUPLICATE_MSG) { 2569 if (less(msg_seqno(msg), mod(dest_link->next_in_no))) { 2570 msg_dbg(tunnel_msg, "DROP/<REC<"); 2571 goto exit; 2572 } 2573 *buf = buf_extract(tunnel_buf,INT_H_SIZE); 2574 if (*buf == NULL) { 2575 warn("Link changeover error, duplicate msg dropped\n"); 2576 goto exit; 2577 } 2578 msg_dbg(tunnel_msg, "TNL<REC<"); 2579 buf_discard(tunnel_buf); 2580 return 1; 2581 } 2582 2583 /* First original message ?: */ 2584 2585 if (tipc_link_is_up(dest_link)) { 2586 msg_dbg(tunnel_msg, "UP/FIRST/<REC<"); 2587 info("Resetting link <%s>, changeover initiated by peer\n", 2588 dest_link->name); 2589 tipc_link_reset(dest_link); 2590 dest_link->exp_msg_count = msg_count; 2591 dbg("Expecting %u tunnelled messages\n", msg_count); 2592 if (!msg_count) 2593 goto exit; 2594 } else if (dest_link->exp_msg_count == START_CHANGEOVER) { 2595 msg_dbg(tunnel_msg, "BLK/FIRST/<REC<"); 2596 dest_link->exp_msg_count = msg_count; 2597 dbg("Expecting %u tunnelled messages\n", msg_count); 2598 if (!msg_count) 2599 goto exit; 2600 } 2601 2602 /* Receive original message */ 2603 2604 if (dest_link->exp_msg_count == 0) { 2605 warn("Link switchover error, " 2606 "got too many tunnelled messages\n"); 2607 msg_dbg(tunnel_msg, "OVERDUE/DROP/<REC<"); 2608 dbg_print_link(dest_link, "LINK:"); 2609 goto exit; 2610 } 2611 dest_link->exp_msg_count--; 2612 if (less(msg_seqno(msg), dest_link->reset_checkpoint)) { 2613 msg_dbg(tunnel_msg, "DROP/DUPL/<REC<"); 2614 goto exit; 2615 } else { 2616 *buf = buf_extract(tunnel_buf, INT_H_SIZE); 2617 if (*buf != NULL) { 2618 msg_dbg(tunnel_msg, "TNL<REC<"); 2619 buf_discard(tunnel_buf); 2620 return 1; 2621 } else { 2622 warn("Link changeover error, original msg dropped\n"); 2623 } 2624 } 2625 exit: 2626 *buf = NULL; 2627 buf_discard(tunnel_buf); 2628 return 0; 2629 } 2630 2631 /* 2632 * Bundler functionality: 2633 */ 2634 void tipc_link_recv_bundle(struct sk_buff *buf) 2635 { 2636 u32 msgcount = msg_msgcnt(buf_msg(buf)); 2637 u32 pos = INT_H_SIZE; 2638 struct sk_buff *obuf; 2639 2640 msg_dbg(buf_msg(buf), "<BNDL<: "); 2641 while (msgcount--) { 2642 obuf = buf_extract(buf, pos); 2643 if (obuf == NULL) { 2644 warn("Link unable to unbundle message(s)\n"); 2645 break; 2646 } 2647 pos += align(msg_size(buf_msg(obuf))); 2648 msg_dbg(buf_msg(obuf), " /"); 2649 tipc_net_route_msg(obuf); 2650 } 2651 buf_discard(buf); 2652 } 2653 2654 /* 2655 * Fragmentation/defragmentation: 2656 */ 2657 2658 2659 /* 2660 * tipc_link_send_long_buf: Entry for buffers needing fragmentation. 2661 * The buffer is complete, inclusive total message length. 2662 * Returns user data length. 2663 */ 2664 int tipc_link_send_long_buf(struct link *l_ptr, struct sk_buff *buf) 2665 { 2666 struct tipc_msg *inmsg = buf_msg(buf); 2667 struct tipc_msg fragm_hdr; 2668 u32 insize = msg_size(inmsg); 2669 u32 dsz = msg_data_sz(inmsg); 2670 unchar *crs = buf->data; 2671 u32 rest = insize; 2672 u32 pack_sz = link_max_pkt(l_ptr); 2673 u32 fragm_sz = pack_sz - INT_H_SIZE; 2674 u32 fragm_no = 1; 2675 u32 destaddr = msg_destnode(inmsg); 2676 2677 if (msg_short(inmsg)) 2678 destaddr = l_ptr->addr; 2679 2680 if (msg_routed(inmsg)) 2681 msg_set_prevnode(inmsg, tipc_own_addr); 2682 2683 /* Prepare reusable fragment header: */ 2684 2685 msg_init(&fragm_hdr, MSG_FRAGMENTER, FIRST_FRAGMENT, 2686 TIPC_OK, INT_H_SIZE, destaddr); 2687 msg_set_link_selector(&fragm_hdr, msg_link_selector(inmsg)); 2688 msg_set_long_msgno(&fragm_hdr, mod(l_ptr->long_msg_seq_no++)); 2689 msg_set_fragm_no(&fragm_hdr, fragm_no); 2690 l_ptr->stats.sent_fragmented++; 2691 2692 /* Chop up message: */ 2693 2694 while (rest > 0) { 2695 struct sk_buff *fragm; 2696 2697 if (rest <= fragm_sz) { 2698 fragm_sz = rest; 2699 msg_set_type(&fragm_hdr, LAST_FRAGMENT); 2700 } 2701 fragm = buf_acquire(fragm_sz + INT_H_SIZE); 2702 if (fragm == NULL) { 2703 warn("Link unable to fragment message\n"); 2704 dsz = -ENOMEM; 2705 goto exit; 2706 } 2707 msg_set_size(&fragm_hdr, fragm_sz + INT_H_SIZE); 2708 skb_copy_to_linear_data(fragm, &fragm_hdr, INT_H_SIZE); 2709 skb_copy_to_linear_data_offset(fragm, INT_H_SIZE, crs, 2710 fragm_sz); 2711 /* Send queued messages first, if any: */ 2712 2713 l_ptr->stats.sent_fragments++; 2714 tipc_link_send_buf(l_ptr, fragm); 2715 if (!tipc_link_is_up(l_ptr)) 2716 return dsz; 2717 msg_set_fragm_no(&fragm_hdr, ++fragm_no); 2718 rest -= fragm_sz; 2719 crs += fragm_sz; 2720 msg_set_type(&fragm_hdr, FRAGMENT); 2721 } 2722 exit: 2723 buf_discard(buf); 2724 return dsz; 2725 } 2726 2727 /* 2728 * A pending message being re-assembled must store certain values 2729 * to handle subsequent fragments correctly. The following functions 2730 * help storing these values in unused, available fields in the 2731 * pending message. This makes dynamic memory allocation unecessary. 2732 */ 2733 2734 static void set_long_msg_seqno(struct sk_buff *buf, u32 seqno) 2735 { 2736 msg_set_seqno(buf_msg(buf), seqno); 2737 } 2738 2739 static u32 get_fragm_size(struct sk_buff *buf) 2740 { 2741 return msg_ack(buf_msg(buf)); 2742 } 2743 2744 static void set_fragm_size(struct sk_buff *buf, u32 sz) 2745 { 2746 msg_set_ack(buf_msg(buf), sz); 2747 } 2748 2749 static u32 get_expected_frags(struct sk_buff *buf) 2750 { 2751 return msg_bcast_ack(buf_msg(buf)); 2752 } 2753 2754 static void set_expected_frags(struct sk_buff *buf, u32 exp) 2755 { 2756 msg_set_bcast_ack(buf_msg(buf), exp); 2757 } 2758 2759 static u32 get_timer_cnt(struct sk_buff *buf) 2760 { 2761 return msg_reroute_cnt(buf_msg(buf)); 2762 } 2763 2764 static void incr_timer_cnt(struct sk_buff *buf) 2765 { 2766 msg_incr_reroute_cnt(buf_msg(buf)); 2767 } 2768 2769 /* 2770 * tipc_link_recv_fragment(): Called with node lock on. Returns 2771 * the reassembled buffer if message is complete. 2772 */ 2773 int tipc_link_recv_fragment(struct sk_buff **pending, struct sk_buff **fb, 2774 struct tipc_msg **m) 2775 { 2776 struct sk_buff *prev = NULL; 2777 struct sk_buff *fbuf = *fb; 2778 struct tipc_msg *fragm = buf_msg(fbuf); 2779 struct sk_buff *pbuf = *pending; 2780 u32 long_msg_seq_no = msg_long_msgno(fragm); 2781 2782 *fb = NULL; 2783 msg_dbg(fragm,"FRG<REC<"); 2784 2785 /* Is there an incomplete message waiting for this fragment? */ 2786 2787 while (pbuf && ((msg_seqno(buf_msg(pbuf)) != long_msg_seq_no) 2788 || (msg_orignode(fragm) != msg_orignode(buf_msg(pbuf))))) { 2789 prev = pbuf; 2790 pbuf = pbuf->next; 2791 } 2792 2793 if (!pbuf && (msg_type(fragm) == FIRST_FRAGMENT)) { 2794 struct tipc_msg *imsg = (struct tipc_msg *)msg_data(fragm); 2795 u32 msg_sz = msg_size(imsg); 2796 u32 fragm_sz = msg_data_sz(fragm); 2797 u32 exp_fragm_cnt = msg_sz/fragm_sz + !!(msg_sz % fragm_sz); 2798 u32 max = TIPC_MAX_USER_MSG_SIZE + LONG_H_SIZE; 2799 if (msg_type(imsg) == TIPC_MCAST_MSG) 2800 max = TIPC_MAX_USER_MSG_SIZE + MCAST_H_SIZE; 2801 if (msg_size(imsg) > max) { 2802 msg_dbg(fragm,"<REC<Oversized: "); 2803 buf_discard(fbuf); 2804 return 0; 2805 } 2806 pbuf = buf_acquire(msg_size(imsg)); 2807 if (pbuf != NULL) { 2808 pbuf->next = *pending; 2809 *pending = pbuf; 2810 skb_copy_to_linear_data(pbuf, imsg, 2811 msg_data_sz(fragm)); 2812 /* Prepare buffer for subsequent fragments. */ 2813 2814 set_long_msg_seqno(pbuf, long_msg_seq_no); 2815 set_fragm_size(pbuf,fragm_sz); 2816 set_expected_frags(pbuf,exp_fragm_cnt - 1); 2817 } else { 2818 warn("Link unable to reassemble fragmented message\n"); 2819 } 2820 buf_discard(fbuf); 2821 return 0; 2822 } else if (pbuf && (msg_type(fragm) != FIRST_FRAGMENT)) { 2823 u32 dsz = msg_data_sz(fragm); 2824 u32 fsz = get_fragm_size(pbuf); 2825 u32 crs = ((msg_fragm_no(fragm) - 1) * fsz); 2826 u32 exp_frags = get_expected_frags(pbuf) - 1; 2827 skb_copy_to_linear_data_offset(pbuf, crs, 2828 msg_data(fragm), dsz); 2829 buf_discard(fbuf); 2830 2831 /* Is message complete? */ 2832 2833 if (exp_frags == 0) { 2834 if (prev) 2835 prev->next = pbuf->next; 2836 else 2837 *pending = pbuf->next; 2838 msg_reset_reroute_cnt(buf_msg(pbuf)); 2839 *fb = pbuf; 2840 *m = buf_msg(pbuf); 2841 return 1; 2842 } 2843 set_expected_frags(pbuf,exp_frags); 2844 return 0; 2845 } 2846 dbg(" Discarding orphan fragment %x\n",fbuf); 2847 msg_dbg(fragm,"ORPHAN:"); 2848 dbg("Pending long buffers:\n"); 2849 dbg_print_buf_chain(*pending); 2850 buf_discard(fbuf); 2851 return 0; 2852 } 2853 2854 /** 2855 * link_check_defragm_bufs - flush stale incoming message fragments 2856 * @l_ptr: pointer to link 2857 */ 2858 2859 static void link_check_defragm_bufs(struct link *l_ptr) 2860 { 2861 struct sk_buff *prev = NULL; 2862 struct sk_buff *next = NULL; 2863 struct sk_buff *buf = l_ptr->defragm_buf; 2864 2865 if (!buf) 2866 return; 2867 if (!link_working_working(l_ptr)) 2868 return; 2869 while (buf) { 2870 u32 cnt = get_timer_cnt(buf); 2871 2872 next = buf->next; 2873 if (cnt < 4) { 2874 incr_timer_cnt(buf); 2875 prev = buf; 2876 } else { 2877 dbg(" Discarding incomplete long buffer\n"); 2878 msg_dbg(buf_msg(buf), "LONG:"); 2879 dbg_print_link(l_ptr, "curr:"); 2880 dbg("Pending long buffers:\n"); 2881 dbg_print_buf_chain(l_ptr->defragm_buf); 2882 if (prev) 2883 prev->next = buf->next; 2884 else 2885 l_ptr->defragm_buf = buf->next; 2886 buf_discard(buf); 2887 } 2888 buf = next; 2889 } 2890 } 2891 2892 2893 2894 static void link_set_supervision_props(struct link *l_ptr, u32 tolerance) 2895 { 2896 l_ptr->tolerance = tolerance; 2897 l_ptr->continuity_interval = 2898 ((tolerance / 4) > 500) ? 500 : tolerance / 4; 2899 l_ptr->abort_limit = tolerance / (l_ptr->continuity_interval / 4); 2900 } 2901 2902 2903 void tipc_link_set_queue_limits(struct link *l_ptr, u32 window) 2904 { 2905 /* Data messages from this node, inclusive FIRST_FRAGM */ 2906 l_ptr->queue_limit[TIPC_LOW_IMPORTANCE] = window; 2907 l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE] = (window / 3) * 4; 2908 l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE] = (window / 3) * 5; 2909 l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE] = (window / 3) * 6; 2910 /* Transiting data messages,inclusive FIRST_FRAGM */ 2911 l_ptr->queue_limit[TIPC_LOW_IMPORTANCE + 4] = 300; 2912 l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE + 4] = 600; 2913 l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE + 4] = 900; 2914 l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE + 4] = 1200; 2915 l_ptr->queue_limit[CONN_MANAGER] = 1200; 2916 l_ptr->queue_limit[ROUTE_DISTRIBUTOR] = 1200; 2917 l_ptr->queue_limit[CHANGEOVER_PROTOCOL] = 2500; 2918 l_ptr->queue_limit[NAME_DISTRIBUTOR] = 3000; 2919 /* FRAGMENT and LAST_FRAGMENT packets */ 2920 l_ptr->queue_limit[MSG_FRAGMENTER] = 4000; 2921 } 2922 2923 /** 2924 * link_find_link - locate link by name 2925 * @name - ptr to link name string 2926 * @node - ptr to area to be filled with ptr to associated node 2927 * 2928 * Caller must hold 'tipc_net_lock' to ensure node and bearer are not deleted; 2929 * this also prevents link deletion. 2930 * 2931 * Returns pointer to link (or 0 if invalid link name). 2932 */ 2933 2934 static struct link *link_find_link(const char *name, struct node **node) 2935 { 2936 struct link_name link_name_parts; 2937 struct bearer *b_ptr; 2938 struct link *l_ptr; 2939 2940 if (!link_name_validate(name, &link_name_parts)) 2941 return NULL; 2942 2943 b_ptr = tipc_bearer_find_interface(link_name_parts.if_local); 2944 if (!b_ptr) 2945 return NULL; 2946 2947 *node = tipc_node_find(link_name_parts.addr_peer); 2948 if (!*node) 2949 return NULL; 2950 2951 l_ptr = (*node)->links[b_ptr->identity]; 2952 if (!l_ptr || strcmp(l_ptr->name, name)) 2953 return NULL; 2954 2955 return l_ptr; 2956 } 2957 2958 struct sk_buff *tipc_link_cmd_config(const void *req_tlv_area, int req_tlv_space, 2959 u16 cmd) 2960 { 2961 struct tipc_link_config *args; 2962 u32 new_value; 2963 struct link *l_ptr; 2964 struct node *node; 2965 int res; 2966 2967 if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_LINK_CONFIG)) 2968 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR); 2969 2970 args = (struct tipc_link_config *)TLV_DATA(req_tlv_area); 2971 new_value = ntohl(args->value); 2972 2973 if (!strcmp(args->name, tipc_bclink_name)) { 2974 if ((cmd == TIPC_CMD_SET_LINK_WINDOW) && 2975 (tipc_bclink_set_queue_limits(new_value) == 0)) 2976 return tipc_cfg_reply_none(); 2977 return tipc_cfg_reply_error_string(TIPC_CFG_NOT_SUPPORTED 2978 " (cannot change setting on broadcast link)"); 2979 } 2980 2981 read_lock_bh(&tipc_net_lock); 2982 l_ptr = link_find_link(args->name, &node); 2983 if (!l_ptr) { 2984 read_unlock_bh(&tipc_net_lock); 2985 return tipc_cfg_reply_error_string("link not found"); 2986 } 2987 2988 tipc_node_lock(node); 2989 res = -EINVAL; 2990 switch (cmd) { 2991 case TIPC_CMD_SET_LINK_TOL: 2992 if ((new_value >= TIPC_MIN_LINK_TOL) && 2993 (new_value <= TIPC_MAX_LINK_TOL)) { 2994 link_set_supervision_props(l_ptr, new_value); 2995 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 2996 0, 0, new_value, 0, 0); 2997 res = TIPC_OK; 2998 } 2999 break; 3000 case TIPC_CMD_SET_LINK_PRI: 3001 if ((new_value >= TIPC_MIN_LINK_PRI) && 3002 (new_value <= TIPC_MAX_LINK_PRI)) { 3003 l_ptr->priority = new_value; 3004 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 3005 0, 0, 0, new_value, 0); 3006 res = TIPC_OK; 3007 } 3008 break; 3009 case TIPC_CMD_SET_LINK_WINDOW: 3010 if ((new_value >= TIPC_MIN_LINK_WIN) && 3011 (new_value <= TIPC_MAX_LINK_WIN)) { 3012 tipc_link_set_queue_limits(l_ptr, new_value); 3013 res = TIPC_OK; 3014 } 3015 break; 3016 } 3017 tipc_node_unlock(node); 3018 3019 read_unlock_bh(&tipc_net_lock); 3020 if (res) 3021 return tipc_cfg_reply_error_string("cannot change link setting"); 3022 3023 return tipc_cfg_reply_none(); 3024 } 3025 3026 /** 3027 * link_reset_statistics - reset link statistics 3028 * @l_ptr: pointer to link 3029 */ 3030 3031 static void link_reset_statistics(struct link *l_ptr) 3032 { 3033 memset(&l_ptr->stats, 0, sizeof(l_ptr->stats)); 3034 l_ptr->stats.sent_info = l_ptr->next_out_no; 3035 l_ptr->stats.recv_info = l_ptr->next_in_no; 3036 } 3037 3038 struct sk_buff *tipc_link_cmd_reset_stats(const void *req_tlv_area, int req_tlv_space) 3039 { 3040 char *link_name; 3041 struct link *l_ptr; 3042 struct node *node; 3043 3044 if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_LINK_NAME)) 3045 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR); 3046 3047 link_name = (char *)TLV_DATA(req_tlv_area); 3048 if (!strcmp(link_name, tipc_bclink_name)) { 3049 if (tipc_bclink_reset_stats()) 3050 return tipc_cfg_reply_error_string("link not found"); 3051 return tipc_cfg_reply_none(); 3052 } 3053 3054 read_lock_bh(&tipc_net_lock); 3055 l_ptr = link_find_link(link_name, &node); 3056 if (!l_ptr) { 3057 read_unlock_bh(&tipc_net_lock); 3058 return tipc_cfg_reply_error_string("link not found"); 3059 } 3060 3061 tipc_node_lock(node); 3062 link_reset_statistics(l_ptr); 3063 tipc_node_unlock(node); 3064 read_unlock_bh(&tipc_net_lock); 3065 return tipc_cfg_reply_none(); 3066 } 3067 3068 /** 3069 * percent - convert count to a percentage of total (rounding up or down) 3070 */ 3071 3072 static u32 percent(u32 count, u32 total) 3073 { 3074 return (count * 100 + (total / 2)) / total; 3075 } 3076 3077 /** 3078 * tipc_link_stats - print link statistics 3079 * @name: link name 3080 * @buf: print buffer area 3081 * @buf_size: size of print buffer area 3082 * 3083 * Returns length of print buffer data string (or 0 if error) 3084 */ 3085 3086 static int tipc_link_stats(const char *name, char *buf, const u32 buf_size) 3087 { 3088 struct print_buf pb; 3089 struct link *l_ptr; 3090 struct node *node; 3091 char *status; 3092 u32 profile_total = 0; 3093 3094 if (!strcmp(name, tipc_bclink_name)) 3095 return tipc_bclink_stats(buf, buf_size); 3096 3097 tipc_printbuf_init(&pb, buf, buf_size); 3098 3099 read_lock_bh(&tipc_net_lock); 3100 l_ptr = link_find_link(name, &node); 3101 if (!l_ptr) { 3102 read_unlock_bh(&tipc_net_lock); 3103 return 0; 3104 } 3105 tipc_node_lock(node); 3106 3107 if (tipc_link_is_active(l_ptr)) 3108 status = "ACTIVE"; 3109 else if (tipc_link_is_up(l_ptr)) 3110 status = "STANDBY"; 3111 else 3112 status = "DEFUNCT"; 3113 tipc_printf(&pb, "Link <%s>\n" 3114 " %s MTU:%u Priority:%u Tolerance:%u ms" 3115 " Window:%u packets\n", 3116 l_ptr->name, status, link_max_pkt(l_ptr), 3117 l_ptr->priority, l_ptr->tolerance, l_ptr->queue_limit[0]); 3118 tipc_printf(&pb, " RX packets:%u fragments:%u/%u bundles:%u/%u\n", 3119 l_ptr->next_in_no - l_ptr->stats.recv_info, 3120 l_ptr->stats.recv_fragments, 3121 l_ptr->stats.recv_fragmented, 3122 l_ptr->stats.recv_bundles, 3123 l_ptr->stats.recv_bundled); 3124 tipc_printf(&pb, " TX packets:%u fragments:%u/%u bundles:%u/%u\n", 3125 l_ptr->next_out_no - l_ptr->stats.sent_info, 3126 l_ptr->stats.sent_fragments, 3127 l_ptr->stats.sent_fragmented, 3128 l_ptr->stats.sent_bundles, 3129 l_ptr->stats.sent_bundled); 3130 profile_total = l_ptr->stats.msg_length_counts; 3131 if (!profile_total) 3132 profile_total = 1; 3133 tipc_printf(&pb, " TX profile sample:%u packets average:%u octets\n" 3134 " 0-64:%u%% -256:%u%% -1024:%u%% -4096:%u%% " 3135 "-16354:%u%% -32768:%u%% -66000:%u%%\n", 3136 l_ptr->stats.msg_length_counts, 3137 l_ptr->stats.msg_lengths_total / profile_total, 3138 percent(l_ptr->stats.msg_length_profile[0], profile_total), 3139 percent(l_ptr->stats.msg_length_profile[1], profile_total), 3140 percent(l_ptr->stats.msg_length_profile[2], profile_total), 3141 percent(l_ptr->stats.msg_length_profile[3], profile_total), 3142 percent(l_ptr->stats.msg_length_profile[4], profile_total), 3143 percent(l_ptr->stats.msg_length_profile[5], profile_total), 3144 percent(l_ptr->stats.msg_length_profile[6], profile_total)); 3145 tipc_printf(&pb, " RX states:%u probes:%u naks:%u defs:%u dups:%u\n", 3146 l_ptr->stats.recv_states, 3147 l_ptr->stats.recv_probes, 3148 l_ptr->stats.recv_nacks, 3149 l_ptr->stats.deferred_recv, 3150 l_ptr->stats.duplicates); 3151 tipc_printf(&pb, " TX states:%u probes:%u naks:%u acks:%u dups:%u\n", 3152 l_ptr->stats.sent_states, 3153 l_ptr->stats.sent_probes, 3154 l_ptr->stats.sent_nacks, 3155 l_ptr->stats.sent_acks, 3156 l_ptr->stats.retransmitted); 3157 tipc_printf(&pb, " Congestion bearer:%u link:%u Send queue max:%u avg:%u\n", 3158 l_ptr->stats.bearer_congs, 3159 l_ptr->stats.link_congs, 3160 l_ptr->stats.max_queue_sz, 3161 l_ptr->stats.queue_sz_counts 3162 ? (l_ptr->stats.accu_queue_sz / l_ptr->stats.queue_sz_counts) 3163 : 0); 3164 3165 tipc_node_unlock(node); 3166 read_unlock_bh(&tipc_net_lock); 3167 return tipc_printbuf_validate(&pb); 3168 } 3169 3170 #define MAX_LINK_STATS_INFO 2000 3171 3172 struct sk_buff *tipc_link_cmd_show_stats(const void *req_tlv_area, int req_tlv_space) 3173 { 3174 struct sk_buff *buf; 3175 struct tlv_desc *rep_tlv; 3176 int str_len; 3177 3178 if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_LINK_NAME)) 3179 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR); 3180 3181 buf = tipc_cfg_reply_alloc(TLV_SPACE(MAX_LINK_STATS_INFO)); 3182 if (!buf) 3183 return NULL; 3184 3185 rep_tlv = (struct tlv_desc *)buf->data; 3186 3187 str_len = tipc_link_stats((char *)TLV_DATA(req_tlv_area), 3188 (char *)TLV_DATA(rep_tlv), MAX_LINK_STATS_INFO); 3189 if (!str_len) { 3190 buf_discard(buf); 3191 return tipc_cfg_reply_error_string("link not found"); 3192 } 3193 3194 skb_put(buf, TLV_SPACE(str_len)); 3195 TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len); 3196 3197 return buf; 3198 } 3199 3200 #if 0 3201 int link_control(const char *name, u32 op, u32 val) 3202 { 3203 int res = -EINVAL; 3204 struct link *l_ptr; 3205 u32 bearer_id; 3206 struct node * node; 3207 u32 a; 3208 3209 a = link_name2addr(name, &bearer_id); 3210 read_lock_bh(&tipc_net_lock); 3211 node = tipc_node_find(a); 3212 if (node) { 3213 tipc_node_lock(node); 3214 l_ptr = node->links[bearer_id]; 3215 if (l_ptr) { 3216 if (op == TIPC_REMOVE_LINK) { 3217 struct bearer *b_ptr = l_ptr->b_ptr; 3218 spin_lock_bh(&b_ptr->publ.lock); 3219 tipc_link_delete(l_ptr); 3220 spin_unlock_bh(&b_ptr->publ.lock); 3221 } 3222 if (op == TIPC_CMD_BLOCK_LINK) { 3223 tipc_link_reset(l_ptr); 3224 l_ptr->blocked = 1; 3225 } 3226 if (op == TIPC_CMD_UNBLOCK_LINK) { 3227 l_ptr->blocked = 0; 3228 } 3229 res = TIPC_OK; 3230 } 3231 tipc_node_unlock(node); 3232 } 3233 read_unlock_bh(&tipc_net_lock); 3234 return res; 3235 } 3236 #endif 3237 3238 /** 3239 * tipc_link_get_max_pkt - get maximum packet size to use when sending to destination 3240 * @dest: network address of destination node 3241 * @selector: used to select from set of active links 3242 * 3243 * If no active link can be found, uses default maximum packet size. 3244 */ 3245 3246 u32 tipc_link_get_max_pkt(u32 dest, u32 selector) 3247 { 3248 struct node *n_ptr; 3249 struct link *l_ptr; 3250 u32 res = MAX_PKT_DEFAULT; 3251 3252 if (dest == tipc_own_addr) 3253 return MAX_MSG_SIZE; 3254 3255 read_lock_bh(&tipc_net_lock); 3256 n_ptr = tipc_node_select(dest, selector); 3257 if (n_ptr) { 3258 tipc_node_lock(n_ptr); 3259 l_ptr = n_ptr->active_links[selector & 1]; 3260 if (l_ptr) 3261 res = link_max_pkt(l_ptr); 3262 tipc_node_unlock(n_ptr); 3263 } 3264 read_unlock_bh(&tipc_net_lock); 3265 return res; 3266 } 3267 3268 #if 0 3269 static void link_dump_rec_queue(struct link *l_ptr) 3270 { 3271 struct sk_buff *crs; 3272 3273 if (!l_ptr->oldest_deferred_in) { 3274 info("Reception queue empty\n"); 3275 return; 3276 } 3277 info("Contents of Reception queue:\n"); 3278 crs = l_ptr->oldest_deferred_in; 3279 while (crs) { 3280 if (crs->data == (void *)0x0000a3a3) { 3281 info("buffer %x invalid\n", crs); 3282 return; 3283 } 3284 msg_dbg(buf_msg(crs), "In rec queue: \n"); 3285 crs = crs->next; 3286 } 3287 } 3288 #endif 3289 3290 static void link_dump_send_queue(struct link *l_ptr) 3291 { 3292 if (l_ptr->next_out) { 3293 info("\nContents of unsent queue:\n"); 3294 dbg_print_buf_chain(l_ptr->next_out); 3295 } 3296 info("\nContents of send queue:\n"); 3297 if (l_ptr->first_out) { 3298 dbg_print_buf_chain(l_ptr->first_out); 3299 } 3300 info("Empty send queue\n"); 3301 } 3302 3303 static void link_print(struct link *l_ptr, struct print_buf *buf, 3304 const char *str) 3305 { 3306 tipc_printf(buf, str); 3307 if (link_reset_reset(l_ptr) || link_reset_unknown(l_ptr)) 3308 return; 3309 tipc_printf(buf, "Link %x<%s>:", 3310 l_ptr->addr, l_ptr->b_ptr->publ.name); 3311 tipc_printf(buf, ": NXO(%u):", mod(l_ptr->next_out_no)); 3312 tipc_printf(buf, "NXI(%u):", mod(l_ptr->next_in_no)); 3313 tipc_printf(buf, "SQUE"); 3314 if (l_ptr->first_out) { 3315 tipc_printf(buf, "[%u..", msg_seqno(buf_msg(l_ptr->first_out))); 3316 if (l_ptr->next_out) 3317 tipc_printf(buf, "%u..", 3318 msg_seqno(buf_msg(l_ptr->next_out))); 3319 tipc_printf(buf, "%u]", 3320 msg_seqno(buf_msg 3321 (l_ptr->last_out)), l_ptr->out_queue_size); 3322 if ((mod(msg_seqno(buf_msg(l_ptr->last_out)) - 3323 msg_seqno(buf_msg(l_ptr->first_out))) 3324 != (l_ptr->out_queue_size - 1)) 3325 || (l_ptr->last_out->next != NULL)) { 3326 tipc_printf(buf, "\nSend queue inconsistency\n"); 3327 tipc_printf(buf, "first_out= %x ", l_ptr->first_out); 3328 tipc_printf(buf, "next_out= %x ", l_ptr->next_out); 3329 tipc_printf(buf, "last_out= %x ", l_ptr->last_out); 3330 link_dump_send_queue(l_ptr); 3331 } 3332 } else 3333 tipc_printf(buf, "[]"); 3334 tipc_printf(buf, "SQSIZ(%u)", l_ptr->out_queue_size); 3335 if (l_ptr->oldest_deferred_in) { 3336 u32 o = msg_seqno(buf_msg(l_ptr->oldest_deferred_in)); 3337 u32 n = msg_seqno(buf_msg(l_ptr->newest_deferred_in)); 3338 tipc_printf(buf, ":RQUE[%u..%u]", o, n); 3339 if (l_ptr->deferred_inqueue_sz != mod((n + 1) - o)) { 3340 tipc_printf(buf, ":RQSIZ(%u)", 3341 l_ptr->deferred_inqueue_sz); 3342 } 3343 } 3344 if (link_working_unknown(l_ptr)) 3345 tipc_printf(buf, ":WU"); 3346 if (link_reset_reset(l_ptr)) 3347 tipc_printf(buf, ":RR"); 3348 if (link_reset_unknown(l_ptr)) 3349 tipc_printf(buf, ":RU"); 3350 if (link_working_working(l_ptr)) 3351 tipc_printf(buf, ":WW"); 3352 tipc_printf(buf, "\n"); 3353 } 3354 3355