net/tipc/link.c

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