Pileus Git - ~andy/linux/blob - net/tipc/link.c

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