net/tipc/link.c

   1 /*
   2  * net/tipc/link.c: TIPC link code
   3  *
   4  * Copyright (c) 1996-2007, 2012-2015, 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 "bcast.h"
  40 #include "socket.h"
  41 #include "name_distr.h"
  42 #include "discover.h"
  43 #include "netlink.h"
  44
  45 #include <linux/pkt_sched.h>
  46
  47 /*
  48  * Error message prefixes
  49  */
  50 static const char *link_co_err = "Link changeover error, ";
  51 static const char *link_rst_msg = "Resetting link ";
  52 static const char *link_unk_evt = "Unknown link event ";
  53
  54 static const struct nla_policy tipc_nl_link_policy[TIPC_NLA_LINK_MAX + 1] = {
  55         [TIPC_NLA_LINK_UNSPEC]          = { .type = NLA_UNSPEC },
  56         [TIPC_NLA_LINK_NAME] = {
  57                 .type = NLA_STRING,
  58                 .len = TIPC_MAX_LINK_NAME
  59         },
  60         [TIPC_NLA_LINK_MTU]             = { .type = NLA_U32 },
  61         [TIPC_NLA_LINK_BROADCAST]       = { .type = NLA_FLAG },
  62         [TIPC_NLA_LINK_UP]              = { .type = NLA_FLAG },
  63         [TIPC_NLA_LINK_ACTIVE]          = { .type = NLA_FLAG },
  64         [TIPC_NLA_LINK_PROP]            = { .type = NLA_NESTED },
  65         [TIPC_NLA_LINK_STATS]           = { .type = NLA_NESTED },
  66         [TIPC_NLA_LINK_RX]              = { .type = NLA_U32 },
  67         [TIPC_NLA_LINK_TX]              = { .type = NLA_U32 }
  68 };
  69
  70 /* Properties valid for media, bearar and link */
  71 static const struct nla_policy tipc_nl_prop_policy[TIPC_NLA_PROP_MAX + 1] = {
  72         [TIPC_NLA_PROP_UNSPEC]          = { .type = NLA_UNSPEC },
  73         [TIPC_NLA_PROP_PRIO]            = { .type = NLA_U32 },
  74         [TIPC_NLA_PROP_TOL]             = { .type = NLA_U32 },
  75         [TIPC_NLA_PROP_WIN]             = { .type = NLA_U32 }
  76 };
  77
  78 /*
  79  * Out-of-range value for link session numbers
  80  */
  81 #define INVALID_SESSION 0x10000
  82
  83 /*
  84  * Link state events:
  85  */
  86 #define  STARTING_EVT    856384768      /* link processing trigger */
  87 #define  TRAFFIC_MSG_EVT 560815u        /* rx'd ??? */
  88 #define  TIMEOUT_EVT     560817u        /* link timer expired */
  89
  90 /*
  91  * The following two 'message types' is really just implementation
  92  * data conveniently stored in the message header.
  93  * They must not be considered part of the protocol
  94  */
  95 #define OPEN_MSG   0
  96 #define CLOSED_MSG 1
  97
  98 /*
  99  * State value stored in 'exp_msg_count'
 100  */
 101 #define START_CHANGEOVER 100000u
 102
 103 static void link_handle_out_of_seq_msg(struct tipc_link *link,
 104                                        struct sk_buff *skb);
 105 static void tipc_link_proto_rcv(struct tipc_link *link,
 106                                 struct sk_buff *skb);
 107 static int  tipc_link_tunnel_rcv(struct tipc_node *node,
 108                                  struct sk_buff **skb);
 109 static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol);
 110 static void link_state_event(struct tipc_link *l_ptr, u32 event);
 111 static void link_reset_statistics(struct tipc_link *l_ptr);
 112 static void link_print(struct tipc_link *l_ptr, const char *str);
 113 static void tipc_link_sync_xmit(struct tipc_link *l);
 114 static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf);
 115 static void tipc_link_input(struct tipc_link *l, struct sk_buff *skb);
 116 static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb);
 117
 118 /*
 119  *  Simple link routines
 120  */
 121 static unsigned int align(unsigned int i)
 122 {
 123         return (i + 3) & ~3u;
 124 }
 125
 126 static void tipc_link_release(struct kref *kref)
 127 {
 128         kfree(container_of(kref, struct tipc_link, ref));
 129 }
 130
 131 static void tipc_link_get(struct tipc_link *l_ptr)
 132 {
 133         kref_get(&l_ptr->ref);
 134 }
 135
 136 static void tipc_link_put(struct tipc_link *l_ptr)
 137 {
 138         kref_put(&l_ptr->ref, tipc_link_release);
 139 }
 140
 141 static void link_init_max_pkt(struct tipc_link *l_ptr)
 142 {
 143         struct tipc_node *node = l_ptr->owner;
 144         struct tipc_net *tn = net_generic(node->net, tipc_net_id);
 145         struct tipc_bearer *b_ptr;
 146         u32 max_pkt;
 147
 148         rcu_read_lock();
 149         b_ptr = rcu_dereference_rtnl(tn->bearer_list[l_ptr->bearer_id]);
 150         if (!b_ptr) {
 151                 rcu_read_unlock();
 152                 return;
 153         }
 154         max_pkt = (b_ptr->mtu & ~3);
 155         rcu_read_unlock();
 156
 157         if (max_pkt > MAX_MSG_SIZE)
 158                 max_pkt = MAX_MSG_SIZE;
 159
 160         l_ptr->max_pkt_target = max_pkt;
 161         if (l_ptr->max_pkt_target < MAX_PKT_DEFAULT)
 162                 l_ptr->max_pkt = l_ptr->max_pkt_target;
 163         else
 164                 l_ptr->max_pkt = MAX_PKT_DEFAULT;
 165
 166         l_ptr->max_pkt_probes = 0;
 167 }
 168
 169 /*
 170  *  Simple non-static link routines (i.e. referenced outside this file)
 171  */
 172 int tipc_link_is_up(struct tipc_link *l_ptr)
 173 {
 174         if (!l_ptr)
 175                 return 0;
 176         return link_working_working(l_ptr) || link_working_unknown(l_ptr);
 177 }
 178
 179 int tipc_link_is_active(struct tipc_link *l_ptr)
 180 {
 181         return  (l_ptr->owner->active_links[0] == l_ptr) ||
 182                 (l_ptr->owner->active_links[1] == l_ptr);
 183 }
 184
 185 /**
 186  * link_timeout - handle expiration of link timer
 187  * @l_ptr: pointer to link
 188  */
 189 static void link_timeout(unsigned long data)
 190 {
 191         struct tipc_link *l_ptr = (struct tipc_link *)data;
 192         struct sk_buff *skb;
 193
 194         tipc_node_lock(l_ptr->owner);
 195
 196         /* update counters used in statistical profiling of send traffic */
 197         l_ptr->stats.accu_queue_sz += skb_queue_len(&l_ptr->outqueue);
 198         l_ptr->stats.queue_sz_counts++;
 199
 200         skb = skb_peek(&l_ptr->outqueue);
 201         if (skb) {
 202                 struct tipc_msg *msg = buf_msg(skb);
 203                 u32 length = msg_size(msg);
 204
 205                 if ((msg_user(msg) == MSG_FRAGMENTER) &&
 206                     (msg_type(msg) == FIRST_FRAGMENT)) {
 207                         length = msg_size(msg_get_wrapped(msg));
 208                 }
 209                 if (length) {
 210                         l_ptr->stats.msg_lengths_total += length;
 211                         l_ptr->stats.msg_length_counts++;
 212                         if (length <= 64)
 213                                 l_ptr->stats.msg_length_profile[0]++;
 214                         else if (length <= 256)
 215                                 l_ptr->stats.msg_length_profile[1]++;
 216                         else if (length <= 1024)
 217                                 l_ptr->stats.msg_length_profile[2]++;
 218                         else if (length <= 4096)
 219                                 l_ptr->stats.msg_length_profile[3]++;
 220                         else if (length <= 16384)
 221                                 l_ptr->stats.msg_length_profile[4]++;
 222                         else if (length <= 32768)
 223                                 l_ptr->stats.msg_length_profile[5]++;
 224                         else
 225                                 l_ptr->stats.msg_length_profile[6]++;
 226                 }
 227         }
 228
 229         /* do all other link processing performed on a periodic basis */
 230         link_state_event(l_ptr, TIMEOUT_EVT);
 231
 232         if (l_ptr->next_out)
 233                 tipc_link_push_packets(l_ptr);
 234
 235         tipc_node_unlock(l_ptr->owner);
 236         tipc_link_put(l_ptr);
 237 }
 238
 239 static void link_set_timer(struct tipc_link *link, unsigned long time)
 240 {
 241         if (!mod_timer(&link->timer, jiffies + time))
 242                 tipc_link_get(link);
 243 }
 244
 245 /**
 246  * tipc_link_create - create a new link
 247  * @n_ptr: pointer to associated node
 248  * @b_ptr: pointer to associated bearer
 249  * @media_addr: media address to use when sending messages over link
 250  *
 251  * Returns pointer to link.
 252  */
 253 struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
 254                                    struct tipc_bearer *b_ptr,
 255                                    const struct tipc_media_addr *media_addr)
 256 {
 257         struct tipc_net *tn = net_generic(n_ptr->net, tipc_net_id);
 258         struct tipc_link *l_ptr;
 259         struct tipc_msg *msg;
 260         char *if_name;
 261         char addr_string[16];
 262         u32 peer = n_ptr->addr;
 263
 264         if (n_ptr->link_cnt >= MAX_BEARERS) {
 265                 tipc_addr_string_fill(addr_string, n_ptr->addr);
 266                 pr_err("Attempt to establish %uth link to %s. Max %u allowed.\n",
 267                         n_ptr->link_cnt, addr_string, MAX_BEARERS);
 268                 return NULL;
 269         }
 270
 271         if (n_ptr->links[b_ptr->identity]) {
 272                 tipc_addr_string_fill(addr_string, n_ptr->addr);
 273                 pr_err("Attempt to establish second link on <%s> to %s\n",
 274                        b_ptr->name, addr_string);
 275                 return NULL;
 276         }
 277
 278         l_ptr = kzalloc(sizeof(*l_ptr), GFP_ATOMIC);
 279         if (!l_ptr) {
 280                 pr_warn("Link creation failed, no memory\n");
 281                 return NULL;
 282         }
 283         kref_init(&l_ptr->ref);
 284         l_ptr->addr = peer;
 285         if_name = strchr(b_ptr->name, ':') + 1;
 286         sprintf(l_ptr->name, "%u.%u.%u:%s-%u.%u.%u:unknown",
 287                 tipc_zone(tn->own_addr), tipc_cluster(tn->own_addr),
 288                 tipc_node(tn->own_addr),
 289                 if_name,
 290                 tipc_zone(peer), tipc_cluster(peer), tipc_node(peer));
 291                 /* note: peer i/f name is updated by reset/activate message */
 292         memcpy(&l_ptr->media_addr, media_addr, sizeof(*media_addr));
 293         l_ptr->owner = n_ptr;
 294         l_ptr->checkpoint = 1;
 295         l_ptr->peer_session = INVALID_SESSION;
 296         l_ptr->bearer_id = b_ptr->identity;
 297         link_set_supervision_props(l_ptr, b_ptr->tolerance);
 298         l_ptr->state = RESET_UNKNOWN;
 299
 300         l_ptr->pmsg = (struct tipc_msg *)&l_ptr->proto_msg;
 301         msg = l_ptr->pmsg;
 302         tipc_msg_init(tn->own_addr, msg, LINK_PROTOCOL, RESET_MSG, INT_H_SIZE,
 303                       l_ptr->addr);
 304         msg_set_size(msg, sizeof(l_ptr->proto_msg));
 305         msg_set_session(msg, (tn->random & 0xffff));
 306         msg_set_bearer_id(msg, b_ptr->identity);
 307         strcpy((char *)msg_data(msg), if_name);
 308
 309         l_ptr->priority = b_ptr->priority;
 310         tipc_link_set_queue_limits(l_ptr, b_ptr->window);
 311
 312         l_ptr->net_plane = b_ptr->net_plane;
 313         link_init_max_pkt(l_ptr);
 314
 315         l_ptr->next_out_no = 1;
 316         __skb_queue_head_init(&l_ptr->outqueue);
 317         __skb_queue_head_init(&l_ptr->deferred_queue);
 318         skb_queue_head_init(&l_ptr->wakeupq);
 319         skb_queue_head_init(&l_ptr->inputq);
 320         skb_queue_head_init(&l_ptr->namedq);
 321         link_reset_statistics(l_ptr);
 322         tipc_node_attach_link(n_ptr, l_ptr);
 323         setup_timer(&l_ptr->timer, link_timeout, (unsigned long)l_ptr);
 324         link_state_event(l_ptr, STARTING_EVT);
 325
 326         return l_ptr;
 327 }
 328
 329 /**
 330  * link_delete - Conditional deletion of link.
 331  *               If timer still running, real delete is done when it expires
 332  * @link: link to be deleted
 333  */
 334 void tipc_link_delete(struct tipc_link *link)
 335 {
 336         tipc_link_reset_fragments(link);
 337         tipc_node_detach_link(link->owner, link);
 338         tipc_link_put(link);
 339 }
 340
 341 void tipc_link_delete_list(struct net *net, unsigned int bearer_id,
 342                            bool shutting_down)
 343 {
 344         struct tipc_net *tn = net_generic(net, tipc_net_id);
 345         struct tipc_link *link;
 346         struct tipc_node *node;
 347         bool del_link;
 348
 349         rcu_read_lock();
 350         list_for_each_entry_rcu(node, &tn->node_list, list) {
 351                 tipc_node_lock(node);
 352                 link = node->links[bearer_id];
 353                 if (!link) {
 354                         tipc_node_unlock(node);
 355                         continue;
 356                 }
 357                 del_link = !tipc_link_is_up(link) && !link->exp_msg_count;
 358                 tipc_link_reset(link);
 359                 if (del_timer(&link->timer))
 360                         tipc_link_put(link);
 361                 link->flags |= LINK_STOPPED;
 362                 /* Delete link now, or when failover is finished: */
 363                 if (shutting_down || !tipc_node_is_up(node) || del_link)
 364                         tipc_link_delete(link);
 365                 tipc_node_unlock(node);
 366         }
 367         rcu_read_unlock();
 368 }
 369
 370 /**
 371  * link_schedule_user - schedule user for wakeup after congestion
 372  * @link: congested link
 373  * @oport: sending port
 374  * @chain_sz: size of buffer chain that was attempted sent
 375  * @imp: importance of message attempted sent
 376  * Create pseudo msg to send back to user when congestion abates
 377  */
 378 static bool link_schedule_user(struct tipc_link *link, u32 oport,
 379                                uint chain_sz, uint imp)
 380 {
 381         struct sk_buff *buf;
 382
 383         buf = tipc_msg_create(SOCK_WAKEUP, 0, INT_H_SIZE, 0,
 384                               link_own_addr(link), link_own_addr(link),
 385                               oport, 0, 0);
 386         if (!buf)
 387                 return false;
 388         TIPC_SKB_CB(buf)->chain_sz = chain_sz;
 389         TIPC_SKB_CB(buf)->chain_imp = imp;
 390         skb_queue_tail(&link->wakeupq, buf);
 391         link->stats.link_congs++;
 392         return true;
 393 }
 394
 395 /**
 396  * link_prepare_wakeup - prepare users for wakeup after congestion
 397  * @link: congested link
 398  * Move a number of waiting users, as permitted by available space in
 399  * the send queue, from link wait queue to node wait queue for wakeup
 400  */
 401 void link_prepare_wakeup(struct tipc_link *link)
 402 {
 403         uint pend_qsz = skb_queue_len(&link->outqueue);
 404         struct sk_buff *skb, *tmp;
 405
 406         skb_queue_walk_safe(&link->wakeupq, skb, tmp) {
 407                 if (pend_qsz >= link->queue_limit[TIPC_SKB_CB(skb)->chain_imp])
 408                         break;
 409                 pend_qsz += TIPC_SKB_CB(skb)->chain_sz;
 410                 skb_unlink(skb, &link->wakeupq);
 411                 skb_queue_tail(&link->inputq, skb);
 412                 link->owner->inputq = &link->inputq;
 413                 link->owner->action_flags |= TIPC_MSG_EVT;
 414         }
 415 }
 416
 417 /**
 418  * tipc_link_reset_fragments - purge link's inbound message fragments queue
 419  * @l_ptr: pointer to link
 420  */
 421 void tipc_link_reset_fragments(struct tipc_link *l_ptr)
 422 {
 423         kfree_skb(l_ptr->reasm_buf);
 424         l_ptr->reasm_buf = NULL;
 425 }
 426
 427 /**
 428  * tipc_link_purge_queues - purge all pkt queues associated with link
 429  * @l_ptr: pointer to link
 430  */
 431 void tipc_link_purge_queues(struct tipc_link *l_ptr)
 432 {
 433         __skb_queue_purge(&l_ptr->deferred_queue);
 434         __skb_queue_purge(&l_ptr->outqueue);
 435         tipc_link_reset_fragments(l_ptr);
 436 }
 437
 438 void tipc_link_reset(struct tipc_link *l_ptr)
 439 {
 440         u32 prev_state = l_ptr->state;
 441         u32 checkpoint = l_ptr->next_in_no;
 442         int was_active_link = tipc_link_is_active(l_ptr);
 443         struct tipc_node *owner = l_ptr->owner;
 444
 445         msg_set_session(l_ptr->pmsg, ((msg_session(l_ptr->pmsg) + 1) & 0xffff));
 446
 447         /* Link is down, accept any session */
 448         l_ptr->peer_session = INVALID_SESSION;
 449
 450         /* Prepare for max packet size negotiation */
 451         link_init_max_pkt(l_ptr);
 452
 453         l_ptr->state = RESET_UNKNOWN;
 454
 455         if ((prev_state == RESET_UNKNOWN) || (prev_state == RESET_RESET))
 456                 return;
 457
 458         tipc_node_link_down(l_ptr->owner, l_ptr);
 459         tipc_bearer_remove_dest(owner->net, l_ptr->bearer_id, l_ptr->addr);
 460
 461         if (was_active_link && tipc_node_active_links(l_ptr->owner)) {
 462                 l_ptr->reset_checkpoint = checkpoint;
 463                 l_ptr->exp_msg_count = START_CHANGEOVER;
 464         }
 465
 466         /* Clean up all queues, except inputq: */
 467         __skb_queue_purge(&l_ptr->outqueue);
 468         __skb_queue_purge(&l_ptr->deferred_queue);
 469         if (!owner->inputq)
 470                 owner->inputq = &l_ptr->inputq;
 471         skb_queue_splice_init(&l_ptr->wakeupq, owner->inputq);
 472         if (!skb_queue_empty(owner->inputq))
 473                 owner->action_flags |= TIPC_MSG_EVT;
 474         l_ptr->next_out = NULL;
 475         l_ptr->unacked_window = 0;
 476         l_ptr->checkpoint = 1;
 477         l_ptr->next_out_no = 1;
 478         l_ptr->fsm_msg_cnt = 0;
 479         l_ptr->stale_count = 0;
 480         link_reset_statistics(l_ptr);
 481 }
 482
 483 void tipc_link_reset_list(struct net *net, unsigned int bearer_id)
 484 {
 485         struct tipc_net *tn = net_generic(net, tipc_net_id);
 486         struct tipc_link *l_ptr;
 487         struct tipc_node *n_ptr;
 488
 489         rcu_read_lock();
 490         list_for_each_entry_rcu(n_ptr, &tn->node_list, list) {
 491                 tipc_node_lock(n_ptr);
 492                 l_ptr = n_ptr->links[bearer_id];
 493                 if (l_ptr)
 494                         tipc_link_reset(l_ptr);
 495                 tipc_node_unlock(n_ptr);
 496         }
 497         rcu_read_unlock();
 498 }
 499
 500 static void link_activate(struct tipc_link *link)
 501 {
 502         struct tipc_node *node = link->owner;
 503
 504         link->next_in_no = 1;
 505         link->stats.recv_info = 1;
 506         tipc_node_link_up(node, link);
 507         tipc_bearer_add_dest(node->net, link->bearer_id, link->addr);
 508 }
 509
 510 /**
 511  * link_state_event - link finite state machine
 512  * @l_ptr: pointer to link
 513  * @event: state machine event to process
 514  */
 515 static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
 516 {
 517         struct tipc_link *other;
 518         unsigned long cont_intv = l_ptr->cont_intv;
 519
 520         if (l_ptr->flags & LINK_STOPPED)
 521                 return;
 522
 523         if (!(l_ptr->flags & LINK_STARTED) && (event != STARTING_EVT))
 524                 return;         /* Not yet. */
 525
 526         /* Check whether changeover is going on */
 527         if (l_ptr->exp_msg_count) {
 528                 if (event == TIMEOUT_EVT)
 529                         link_set_timer(l_ptr, cont_intv);
 530                 return;
 531         }
 532
 533         switch (l_ptr->state) {
 534         case WORKING_WORKING:
 535                 switch (event) {
 536                 case TRAFFIC_MSG_EVT:
 537                 case ACTIVATE_MSG:
 538                         break;
 539                 case TIMEOUT_EVT:
 540                         if (l_ptr->next_in_no != l_ptr->checkpoint) {
 541                                 l_ptr->checkpoint = l_ptr->next_in_no;
 542                                 if (tipc_bclink_acks_missing(l_ptr->owner)) {
 543                                         tipc_link_proto_xmit(l_ptr, STATE_MSG,
 544                                                              0, 0, 0, 0, 0);
 545                                         l_ptr->fsm_msg_cnt++;
 546                                 } else if (l_ptr->max_pkt < l_ptr->max_pkt_target) {
 547                                         tipc_link_proto_xmit(l_ptr, STATE_MSG,
 548                                                              1, 0, 0, 0, 0);
 549                                         l_ptr->fsm_msg_cnt++;
 550                                 }
 551                                 link_set_timer(l_ptr, cont_intv);
 552                                 break;
 553                         }
 554                         l_ptr->state = WORKING_UNKNOWN;
 555                         l_ptr->fsm_msg_cnt = 0;
 556                         tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
 557                         l_ptr->fsm_msg_cnt++;
 558                         link_set_timer(l_ptr, cont_intv / 4);
 559                         break;
 560                 case RESET_MSG:
 561                         pr_debug("%s<%s>, requested by peer\n",
 562                                  link_rst_msg, l_ptr->name);
 563                         tipc_link_reset(l_ptr);
 564                         l_ptr->state = RESET_RESET;
 565                         l_ptr->fsm_msg_cnt = 0;
 566                         tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
 567                                              0, 0, 0, 0, 0);
 568                         l_ptr->fsm_msg_cnt++;
 569                         link_set_timer(l_ptr, cont_intv);
 570                         break;
 571                 default:
 572                         pr_debug("%s%u in WW state\n", link_unk_evt, event);
 573                 }
 574                 break;
 575         case WORKING_UNKNOWN:
 576                 switch (event) {
 577                 case TRAFFIC_MSG_EVT:
 578                 case ACTIVATE_MSG:
 579                         l_ptr->state = WORKING_WORKING;
 580                         l_ptr->fsm_msg_cnt = 0;
 581                         link_set_timer(l_ptr, cont_intv);
 582                         break;
 583                 case RESET_MSG:
 584                         pr_debug("%s<%s>, requested by peer while probing\n",
 585                                  link_rst_msg, l_ptr->name);
 586                         tipc_link_reset(l_ptr);
 587                         l_ptr->state = RESET_RESET;
 588                         l_ptr->fsm_msg_cnt = 0;
 589                         tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
 590                                              0, 0, 0, 0, 0);
 591                         l_ptr->fsm_msg_cnt++;
 592                         link_set_timer(l_ptr, cont_intv);
 593                         break;
 594                 case TIMEOUT_EVT:
 595                         if (l_ptr->next_in_no != l_ptr->checkpoint) {
 596                                 l_ptr->state = WORKING_WORKING;
 597                                 l_ptr->fsm_msg_cnt = 0;
 598                                 l_ptr->checkpoint = l_ptr->next_in_no;
 599                                 if (tipc_bclink_acks_missing(l_ptr->owner)) {
 600                                         tipc_link_proto_xmit(l_ptr, STATE_MSG,
 601                                                              0, 0, 0, 0, 0);
 602                                         l_ptr->fsm_msg_cnt++;
 603                                 }
 604                                 link_set_timer(l_ptr, cont_intv);
 605                         } else if (l_ptr->fsm_msg_cnt < l_ptr->abort_limit) {
 606                                 tipc_link_proto_xmit(l_ptr, STATE_MSG,
 607                                                      1, 0, 0, 0, 0);
 608                                 l_ptr->fsm_msg_cnt++;
 609                                 link_set_timer(l_ptr, cont_intv / 4);
 610                         } else {        /* Link has failed */
 611                                 pr_debug("%s<%s>, peer not responding\n",
 612                                          link_rst_msg, l_ptr->name);
 613                                 tipc_link_reset(l_ptr);
 614                                 l_ptr->state = RESET_UNKNOWN;
 615                                 l_ptr->fsm_msg_cnt = 0;
 616                                 tipc_link_proto_xmit(l_ptr, RESET_MSG,
 617                                                      0, 0, 0, 0, 0);
 618                                 l_ptr->fsm_msg_cnt++;
 619                                 link_set_timer(l_ptr, cont_intv);
 620                         }
 621                         break;
 622                 default:
 623                         pr_err("%s%u in WU state\n", link_unk_evt, event);
 624                 }
 625                 break;
 626         case RESET_UNKNOWN:
 627                 switch (event) {
 628                 case TRAFFIC_MSG_EVT:
 629                         break;
 630                 case ACTIVATE_MSG:
 631                         other = l_ptr->owner->active_links[0];
 632                         if (other && link_working_unknown(other))
 633                                 break;
 634                         l_ptr->state = WORKING_WORKING;
 635                         l_ptr->fsm_msg_cnt = 0;
 636                         link_activate(l_ptr);
 637                         tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
 638                         l_ptr->fsm_msg_cnt++;
 639                         if (l_ptr->owner->working_links == 1)
 640                                 tipc_link_sync_xmit(l_ptr);
 641                         link_set_timer(l_ptr, cont_intv);
 642                         break;
 643                 case RESET_MSG:
 644                         l_ptr->state = RESET_RESET;
 645                         l_ptr->fsm_msg_cnt = 0;
 646                         tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
 647                                              1, 0, 0, 0, 0);
 648                         l_ptr->fsm_msg_cnt++;
 649                         link_set_timer(l_ptr, cont_intv);
 650                         break;
 651                 case STARTING_EVT:
 652                         l_ptr->flags |= LINK_STARTED;
 653                         l_ptr->fsm_msg_cnt++;
 654                         link_set_timer(l_ptr, cont_intv);
 655                         break;
 656                 case TIMEOUT_EVT:
 657                         tipc_link_proto_xmit(l_ptr, RESET_MSG, 0, 0, 0, 0, 0);
 658                         l_ptr->fsm_msg_cnt++;
 659                         link_set_timer(l_ptr, cont_intv);
 660                         break;
 661                 default:
 662                         pr_err("%s%u in RU state\n", link_unk_evt, event);
 663                 }
 664                 break;
 665         case RESET_RESET:
 666                 switch (event) {
 667                 case TRAFFIC_MSG_EVT:
 668                 case ACTIVATE_MSG:
 669                         other = l_ptr->owner->active_links[0];
 670                         if (other && link_working_unknown(other))
 671                                 break;
 672                         l_ptr->state = WORKING_WORKING;
 673                         l_ptr->fsm_msg_cnt = 0;
 674                         link_activate(l_ptr);
 675                         tipc_link_proto_xmit(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
 676                         l_ptr->fsm_msg_cnt++;
 677                         if (l_ptr->owner->working_links == 1)
 678                                 tipc_link_sync_xmit(l_ptr);
 679                         link_set_timer(l_ptr, cont_intv);
 680                         break;
 681                 case RESET_MSG:
 682                         break;
 683                 case TIMEOUT_EVT:
 684                         tipc_link_proto_xmit(l_ptr, ACTIVATE_MSG,
 685                                              0, 0, 0, 0, 0);
 686                         l_ptr->fsm_msg_cnt++;
 687                         link_set_timer(l_ptr, cont_intv);
 688                         break;
 689                 default:
 690                         pr_err("%s%u in RR state\n", link_unk_evt, event);
 691                 }
 692                 break;
 693         default:
 694                 pr_err("Unknown link state %u/%u\n", l_ptr->state, event);
 695         }
 696 }
 697
 698 /* tipc_link_cong: determine return value and how to treat the
 699  * sent buffer during link congestion.
 700  * - For plain, errorless user data messages we keep the buffer and
 701  *   return -ELINKONG.
 702  * - For all other messages we discard the buffer and return -EHOSTUNREACH
 703  * - For TIPC internal messages we also reset the link
 704  */
 705 static int tipc_link_cong(struct tipc_link *link, struct sk_buff_head *list)
 706 {
 707         struct sk_buff *skb = skb_peek(list);
 708         struct tipc_msg *msg = buf_msg(skb);
 709         uint imp = tipc_msg_tot_importance(msg);
 710         u32 oport = msg_tot_origport(msg);
 711
 712         if (unlikely(imp > TIPC_CRITICAL_IMPORTANCE)) {
 713                 pr_warn("%s<%s>, send queue full", link_rst_msg, link->name);
 714                 tipc_link_reset(link);
 715                 goto drop;
 716         }
 717         if (unlikely(msg_errcode(msg)))
 718                 goto drop;
 719         if (unlikely(msg_reroute_cnt(msg)))
 720                 goto drop;
 721         if (TIPC_SKB_CB(skb)->wakeup_pending)
 722                 return -ELINKCONG;
 723         if (link_schedule_user(link, oport, skb_queue_len(list), imp))
 724                 return -ELINKCONG;
 725 drop:
 726         __skb_queue_purge(list);
 727         return -EHOSTUNREACH;
 728 }
 729
 730 /**
 731  * __tipc_link_xmit(): same as tipc_link_xmit, but destlink is known & locked
 732  * @link: link to use
 733  * @list: chain of buffers containing message
 734  *
 735  * Consumes the buffer chain, except when returning -ELINKCONG
 736  * Returns 0 if success, otherwise errno: -ELINKCONG, -EMSGSIZE (plain socket
 737  * user data messages) or -EHOSTUNREACH (all other messages/senders)
 738  * Only the socket functions tipc_send_stream() and tipc_send_packet() need
 739  * to act on the return value, since they may need to do more send attempts.
 740  */
 741 int __tipc_link_xmit(struct net *net, struct tipc_link *link,
 742                      struct sk_buff_head *list)
 743 {
 744         struct tipc_msg *msg = buf_msg(skb_peek(list));
 745         uint psz = msg_size(msg);
 746         uint sndlim = link->queue_limit[0];
 747         uint imp = tipc_msg_tot_importance(msg);
 748         uint mtu = link->max_pkt;
 749         uint ack = mod(link->next_in_no - 1);
 750         uint seqno = link->next_out_no;
 751         uint bc_last_in = link->owner->bclink.last_in;
 752         struct tipc_media_addr *addr = &link->media_addr;
 753         struct sk_buff_head *outqueue = &link->outqueue;
 754         struct sk_buff *skb, *tmp;
 755
 756         /* Match queue limits against msg importance: */
 757         if (unlikely(skb_queue_len(outqueue) >= link->queue_limit[imp]))
 758                 return tipc_link_cong(link, list);
 759
 760         /* Has valid packet limit been used ? */
 761         if (unlikely(psz > mtu)) {
 762                 __skb_queue_purge(list);
 763                 return -EMSGSIZE;
 764         }
 765
 766         /* Prepare each packet for sending, and add to outqueue: */
 767         skb_queue_walk_safe(list, skb, tmp) {
 768                 __skb_unlink(skb, list);
 769                 msg = buf_msg(skb);
 770                 msg_set_word(msg, 2, ((ack << 16) | mod(seqno)));
 771                 msg_set_bcast_ack(msg, bc_last_in);
 772
 773                 if (skb_queue_len(outqueue) < sndlim) {
 774                         __skb_queue_tail(outqueue, skb);
 775                         tipc_bearer_send(net, link->bearer_id,
 776                                          skb, addr);
 777                         link->next_out = NULL;
 778                         link->unacked_window = 0;
 779                 } else if (tipc_msg_bundle(outqueue, skb, mtu)) {
 780                         link->stats.sent_bundled++;
 781                         continue;
 782                 } else if (tipc_msg_make_bundle(outqueue, skb, mtu,
 783                                                 link->addr)) {
 784                         link->stats.sent_bundled++;
 785                         link->stats.sent_bundles++;
 786                         if (!link->next_out)
 787                                 link->next_out = skb_peek_tail(outqueue);
 788                 } else {
 789                         __skb_queue_tail(outqueue, skb);
 790                         if (!link->next_out)
 791                                 link->next_out = skb;
 792                 }
 793                 seqno++;
 794         }
 795         link->next_out_no = seqno;
 796         return 0;
 797 }
 798
 799 static void skb2list(struct sk_buff *skb, struct sk_buff_head *list)
 800 {
 801         skb_queue_head_init(list);
 802         __skb_queue_tail(list, skb);
 803 }
 804
 805 static int __tipc_link_xmit_skb(struct tipc_link *link, struct sk_buff *skb)
 806 {
 807         struct sk_buff_head head;
 808
 809         skb2list(skb, &head);
 810         return __tipc_link_xmit(link->owner->net, link, &head);
 811 }
 812
 813 int tipc_link_xmit_skb(struct net *net, struct sk_buff *skb, u32 dnode,
 814                        u32 selector)
 815 {
 816         struct sk_buff_head head;
 817
 818         skb2list(skb, &head);
 819         return tipc_link_xmit(net, &head, dnode, selector);
 820 }
 821
 822 /**
 823  * tipc_link_xmit() is the general link level function for message sending
 824  * @net: the applicable net namespace
 825  * @list: chain of buffers containing message
 826  * @dsz: amount of user data to be sent
 827  * @dnode: address of destination node
 828  * @selector: a number used for deterministic link selection
 829  * Consumes the buffer chain, except when returning -ELINKCONG
 830  * Returns 0 if success, otherwise errno: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE
 831  */
 832 int tipc_link_xmit(struct net *net, struct sk_buff_head *list, u32 dnode,
 833                    u32 selector)
 834 {
 835         struct tipc_link *link = NULL;
 836         struct tipc_node *node;
 837         int rc = -EHOSTUNREACH;
 838
 839         node = tipc_node_find(net, dnode);
 840         if (node) {
 841                 tipc_node_lock(node);
 842                 link = node->active_links[selector & 1];
 843                 if (link)
 844                         rc = __tipc_link_xmit(net, link, list);
 845                 tipc_node_unlock(node);
 846         }
 847         if (link)
 848                 return rc;
 849
 850         if (likely(in_own_node(net, dnode)))
 851                 return tipc_sk_rcv(net, list);
 852
 853         __skb_queue_purge(list);
 854         return rc;
 855 }
 856
 857 /*
 858  * tipc_link_sync_xmit - synchronize broadcast link endpoints.
 859  *
 860  * Give a newly added peer node the sequence number where it should
 861  * start receiving and acking broadcast packets.
 862  *
 863  * Called with node locked
 864  */
 865 static void tipc_link_sync_xmit(struct tipc_link *link)
 866 {
 867         struct sk_buff *skb;
 868         struct tipc_msg *msg;
 869
 870         skb = tipc_buf_acquire(INT_H_SIZE);
 871         if (!skb)
 872                 return;
 873
 874         msg = buf_msg(skb);
 875         tipc_msg_init(link_own_addr(link), msg, BCAST_PROTOCOL, STATE_MSG,
 876                       INT_H_SIZE, link->addr);
 877         msg_set_last_bcast(msg, link->owner->bclink.acked);
 878         __tipc_link_xmit_skb(link, skb);
 879 }
 880
 881 /*
 882  * tipc_link_sync_rcv - synchronize broadcast link endpoints.
 883  * Receive the sequence number where we should start receiving and
 884  * acking broadcast packets from a newly added peer node, and open
 885  * up for reception of such packets.
 886  *
 887  * Called with node locked
 888  */
 889 static void tipc_link_sync_rcv(struct tipc_node *n, struct sk_buff *buf)
 890 {
 891         struct tipc_msg *msg = buf_msg(buf);
 892
 893         n->bclink.last_sent = n->bclink.last_in = msg_last_bcast(msg);
 894         n->bclink.recv_permitted = true;
 895         kfree_skb(buf);
 896 }
 897
 898 struct sk_buff *tipc_skb_queue_next(const struct sk_buff_head *list,
 899                                     const struct sk_buff *skb)
 900 {
 901         if (skb_queue_is_last(list, skb))
 902                 return NULL;
 903         return skb->next;
 904 }
 905
 906 /*
 907  * tipc_link_push_packets - push unsent packets to bearer
 908  *
 909  * Push out the unsent messages of a link where congestion
 910  * has abated. Node is locked.
 911  *
 912  * Called with node locked
 913  */
 914 void tipc_link_push_packets(struct tipc_link *l_ptr)
 915 {
 916         struct sk_buff_head *outqueue = &l_ptr->outqueue;
 917         struct sk_buff *skb = l_ptr->next_out;
 918         struct tipc_msg *msg;
 919         u32 next, first;
 920
 921         skb_queue_walk_from(outqueue, skb) {
 922                 msg = buf_msg(skb);
 923                 next = msg_seqno(msg);
 924                 first = buf_seqno(skb_peek(outqueue));
 925
 926                 if (mod(next - first) < l_ptr->queue_limit[0]) {
 927                         msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
 928                         msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
 929                         if (msg_user(msg) == MSG_BUNDLER)
 930                                 TIPC_SKB_CB(skb)->bundling = false;
 931                         tipc_bearer_send(l_ptr->owner->net,
 932                                          l_ptr->bearer_id, skb,
 933                                          &l_ptr->media_addr);
 934                         l_ptr->next_out = tipc_skb_queue_next(outqueue, skb);
 935                 } else {
 936                         break;
 937                 }
 938         }
 939 }
 940
 941 void tipc_link_reset_all(struct tipc_node *node)
 942 {
 943         char addr_string[16];
 944         u32 i;
 945
 946         tipc_node_lock(node);
 947
 948         pr_warn("Resetting all links to %s\n",
 949                 tipc_addr_string_fill(addr_string, node->addr));
 950
 951         for (i = 0; i < MAX_BEARERS; i++) {
 952                 if (node->links[i]) {
 953                         link_print(node->links[i], "Resetting link\n");
 954                         tipc_link_reset(node->links[i]);
 955                 }
 956         }
 957
 958         tipc_node_unlock(node);
 959 }
 960
 961 static void link_retransmit_failure(struct tipc_link *l_ptr,
 962                                     struct sk_buff *buf)
 963 {
 964         struct tipc_msg *msg = buf_msg(buf);
 965         struct net *net = l_ptr->owner->net;
 966
 967         pr_warn("Retransmission failure on link <%s>\n", l_ptr->name);
 968
 969         if (l_ptr->addr) {
 970                 /* Handle failure on standard link */
 971                 link_print(l_ptr, "Resetting link\n");
 972                 tipc_link_reset(l_ptr);
 973
 974         } else {
 975                 /* Handle failure on broadcast link */
 976                 struct tipc_node *n_ptr;
 977                 char addr_string[16];
 978
 979                 pr_info("Msg seq number: %u,  ", msg_seqno(msg));
 980                 pr_cont("Outstanding acks: %lu\n",
 981                         (unsigned long) TIPC_SKB_CB(buf)->handle);
 982
 983                 n_ptr = tipc_bclink_retransmit_to(net);
 984                 tipc_node_lock(n_ptr);
 985
 986                 tipc_addr_string_fill(addr_string, n_ptr->addr);
 987                 pr_info("Broadcast link info for %s\n", addr_string);
 988                 pr_info("Reception permitted: %d,  Acked: %u\n",
 989                         n_ptr->bclink.recv_permitted,
 990                         n_ptr->bclink.acked);
 991                 pr_info("Last in: %u,  Oos state: %u,  Last sent: %u\n",
 992                         n_ptr->bclink.last_in,
 993                         n_ptr->bclink.oos_state,
 994                         n_ptr->bclink.last_sent);
 995
 996                 tipc_node_unlock(n_ptr);
 997
 998                 tipc_bclink_set_flags(net, TIPC_BCLINK_RESET);
 999                 l_ptr->stale_count = 0;
1000         }
1001 }
1002
1003 void tipc_link_retransmit(struct tipc_link *l_ptr, struct sk_buff *skb,
1004                           u32 retransmits)
1005 {
1006         struct tipc_msg *msg;
1007
1008         if (!skb)
1009                 return;
1010
1011         msg = buf_msg(skb);
1012
1013         /* Detect repeated retransmit failures */
1014         if (l_ptr->last_retransmitted == msg_seqno(msg)) {
1015                 if (++l_ptr->stale_count > 100) {
1016                         link_retransmit_failure(l_ptr, skb);
1017                         return;
1018                 }
1019         } else {
1020                 l_ptr->last_retransmitted = msg_seqno(msg);
1021                 l_ptr->stale_count = 1;
1022         }
1023
1024         skb_queue_walk_from(&l_ptr->outqueue, skb) {
1025                 if (!retransmits || skb == l_ptr->next_out)
1026                         break;
1027                 msg = buf_msg(skb);
1028                 msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
1029                 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
1030                 tipc_bearer_send(l_ptr->owner->net, l_ptr->bearer_id, skb,
1031                                  &l_ptr->media_addr);
1032                 retransmits--;
1033                 l_ptr->stats.retransmitted++;
1034         }
1035 }
1036
1037 static void link_retrieve_defq(struct tipc_link *link,
1038                                struct sk_buff_head *list)
1039 {
1040         u32 seq_no;
1041
1042         if (skb_queue_empty(&link->deferred_queue))
1043                 return;
1044
1045         seq_no = buf_seqno(skb_peek(&link->deferred_queue));
1046         if (seq_no == mod(link->next_in_no))
1047                 skb_queue_splice_tail_init(&link->deferred_queue, list);
1048 }
1049
1050 /**
1051  * tipc_rcv - process TIPC packets/messages arriving from off-node
1052  * @net: the applicable net namespace
1053  * @skb: TIPC packet
1054  * @b_ptr: pointer to bearer message arrived on
1055  *
1056  * Invoked with no locks held.  Bearer pointer must point to a valid bearer
1057  * structure (i.e. cannot be NULL), but bearer can be inactive.
1058  */
1059 void tipc_rcv(struct net *net, struct sk_buff *skb, struct tipc_bearer *b_ptr)
1060 {
1061         struct tipc_net *tn = net_generic(net, tipc_net_id);
1062         struct sk_buff_head head;
1063         struct tipc_node *n_ptr;
1064         struct tipc_link *l_ptr;
1065         struct sk_buff *skb1, *tmp;
1066         struct tipc_msg *msg;
1067         u32 seq_no;
1068         u32 ackd;
1069         u32 released;
1070
1071         skb2list(skb, &head);
1072
1073         while ((skb = __skb_dequeue(&head))) {
1074                 /* Ensure message is well-formed */
1075                 if (unlikely(!tipc_msg_validate(skb)))
1076                         goto discard;
1077
1078                 /* Handle arrival of a non-unicast link message */
1079                 msg = buf_msg(skb);
1080                 if (unlikely(msg_non_seq(msg))) {
1081                         if (msg_user(msg) ==  LINK_CONFIG)
1082                                 tipc_disc_rcv(net, skb, b_ptr);
1083                         else
1084                                 tipc_bclink_rcv(net, skb);
1085                         continue;
1086                 }
1087
1088                 /* Discard unicast link messages destined for another node */
1089                 if (unlikely(!msg_short(msg) &&
1090                              (msg_destnode(msg) != tn->own_addr)))
1091                         goto discard;
1092
1093                 /* Locate neighboring node that sent message */
1094                 n_ptr = tipc_node_find(net, msg_prevnode(msg));
1095                 if (unlikely(!n_ptr))
1096                         goto discard;
1097                 tipc_node_lock(n_ptr);
1098
1099                 /* Locate unicast link endpoint that should handle message */
1100                 l_ptr = n_ptr->links[b_ptr->identity];
1101                 if (unlikely(!l_ptr))
1102                         goto unlock;
1103
1104                 /* Verify that communication with node is currently allowed */
1105                 if ((n_ptr->action_flags & TIPC_WAIT_PEER_LINKS_DOWN) &&
1106                     msg_user(msg) == LINK_PROTOCOL &&
1107                     (msg_type(msg) == RESET_MSG ||
1108                     msg_type(msg) == ACTIVATE_MSG) &&
1109                     !msg_redundant_link(msg))
1110                         n_ptr->action_flags &= ~TIPC_WAIT_PEER_LINKS_DOWN;
1111
1112                 if (tipc_node_blocked(n_ptr))
1113                         goto unlock;
1114
1115                 /* Validate message sequence number info */
1116                 seq_no = msg_seqno(msg);
1117                 ackd = msg_ack(msg);
1118
1119                 /* Release acked messages */
1120                 if (unlikely(n_ptr->bclink.acked != msg_bcast_ack(msg)))
1121                         tipc_bclink_acknowledge(n_ptr, msg_bcast_ack(msg));
1122
1123                 released = 0;
1124                 skb_queue_walk_safe(&l_ptr->outqueue, skb1, tmp) {
1125                         if (skb1 == l_ptr->next_out ||
1126                             more(buf_seqno(skb1), ackd))
1127                                 break;
1128                          __skb_unlink(skb1, &l_ptr->outqueue);
1129                          kfree_skb(skb1);
1130                          released = 1;
1131                 }
1132
1133                 /* Try sending any messages link endpoint has pending */
1134                 if (unlikely(l_ptr->next_out))
1135                         tipc_link_push_packets(l_ptr);
1136
1137                 if (released && !skb_queue_empty(&l_ptr->wakeupq))
1138                         link_prepare_wakeup(l_ptr);
1139
1140                 /* Process the incoming packet */
1141                 if (unlikely(!link_working_working(l_ptr))) {
1142                         if (msg_user(msg) == LINK_PROTOCOL) {
1143                                 tipc_link_proto_rcv(l_ptr, skb);
1144                                 link_retrieve_defq(l_ptr, &head);
1145                                 skb = NULL;
1146                                 goto unlock;
1147                         }
1148
1149                         /* Traffic message. Conditionally activate link */
1150                         link_state_event(l_ptr, TRAFFIC_MSG_EVT);
1151
1152                         if (link_working_working(l_ptr)) {
1153                                 /* Re-insert buffer in front of queue */
1154                                 __skb_queue_head(&head, skb);
1155                                 skb = NULL;
1156                                 goto unlock;
1157                         }
1158                         goto unlock;
1159                 }
1160
1161                 /* Link is now in state WORKING_WORKING */
1162                 if (unlikely(seq_no != mod(l_ptr->next_in_no))) {
1163                         link_handle_out_of_seq_msg(l_ptr, skb);
1164                         link_retrieve_defq(l_ptr, &head);
1165                         skb = NULL;
1166                         goto unlock;
1167                 }
1168                 l_ptr->next_in_no++;
1169                 if (unlikely(!skb_queue_empty(&l_ptr->deferred_queue)))
1170                         link_retrieve_defq(l_ptr, &head);
1171
1172                 if (unlikely(++l_ptr->unacked_window >= TIPC_MIN_LINK_WIN)) {
1173                         l_ptr->stats.sent_acks++;
1174                         tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
1175                 }
1176                 tipc_link_input(l_ptr, skb);
1177                 skb = NULL;
1178 unlock:
1179                 tipc_node_unlock(n_ptr);
1180 discard:
1181                 if (unlikely(skb))
1182                         kfree_skb(skb);
1183         }
1184 }
1185
1186 /* tipc_data_input - deliver data and name distr msgs to upper layer
1187  *
1188  * Consumes buffer if message is of right type
1189  * Node lock must be held
1190  */
1191 static bool tipc_data_input(struct tipc_link *link, struct sk_buff *skb)
1192 {
1193         struct tipc_node *node = link->owner;
1194         struct tipc_msg *msg = buf_msg(skb);
1195         u32 dport = msg_destport(msg);
1196
1197         switch (msg_user(msg)) {
1198         case TIPC_LOW_IMPORTANCE:
1199         case TIPC_MEDIUM_IMPORTANCE:
1200         case TIPC_HIGH_IMPORTANCE:
1201         case TIPC_CRITICAL_IMPORTANCE:
1202         case CONN_MANAGER:
1203                 if (tipc_skb_queue_tail(&link->inputq, skb, dport)) {
1204                         node->inputq = &link->inputq;
1205                         node->action_flags |= TIPC_MSG_EVT;
1206                 }
1207                 return true;
1208         case NAME_DISTRIBUTOR:
1209                 node->bclink.recv_permitted = true;
1210                 node->namedq = &link->namedq;
1211                 skb_queue_tail(&link->namedq, skb);
1212                 if (skb_queue_len(&link->namedq) == 1)
1213                         node->action_flags |= TIPC_NAMED_MSG_EVT;
1214                 return true;
1215         case MSG_BUNDLER:
1216         case CHANGEOVER_PROTOCOL:
1217         case MSG_FRAGMENTER:
1218         case BCAST_PROTOCOL:
1219                 return false;
1220         default:
1221                 pr_warn("Dropping received illegal msg type\n");
1222                 kfree_skb(skb);
1223                 return false;
1224         };
1225 }
1226
1227 /* tipc_link_input - process packet that has passed link protocol check
1228  *
1229  * Consumes buffer
1230  * Node lock must be held
1231  */
1232 static void tipc_link_input(struct tipc_link *link, struct sk_buff *skb)
1233 {
1234         struct tipc_node *node = link->owner;
1235         struct tipc_msg *msg = buf_msg(skb);
1236         struct sk_buff *iskb;
1237         int pos = 0;
1238
1239         if (likely(tipc_data_input(link, skb)))
1240                 return;
1241
1242         switch (msg_user(msg)) {
1243         case CHANGEOVER_PROTOCOL:
1244                 if (!tipc_link_tunnel_rcv(node, &skb))
1245                         break;
1246                 if (msg_user(buf_msg(skb)) != MSG_BUNDLER) {
1247                         tipc_data_input(link, skb);
1248                         break;
1249                 }
1250         case MSG_BUNDLER:
1251                 link->stats.recv_bundles++;
1252                 link->stats.recv_bundled += msg_msgcnt(msg);
1253
1254                 while (tipc_msg_extract(skb, &iskb, &pos))
1255                         tipc_data_input(link, iskb);
1256                 break;
1257         case MSG_FRAGMENTER:
1258                 link->stats.recv_fragments++;
1259                 if (tipc_buf_append(&link->reasm_buf, &skb)) {
1260                         link->stats.recv_fragmented++;
1261                         tipc_data_input(link, skb);
1262                 } else if (!link->reasm_buf) {
1263                         tipc_link_reset(link);
1264                 }
1265                 break;
1266         case BCAST_PROTOCOL:
1267                 tipc_link_sync_rcv(node, skb);
1268                 break;
1269         default:
1270                 break;
1271         };
1272 }
1273
1274 /**
1275  * tipc_link_defer_pkt - Add out-of-sequence message to deferred reception queue
1276  *
1277  * Returns increase in queue length (i.e. 0 or 1)
1278  */
1279 u32 tipc_link_defer_pkt(struct sk_buff_head *list, struct sk_buff *skb)
1280 {
1281         struct sk_buff *skb1;
1282         u32 seq_no = buf_seqno(skb);
1283
1284         /* Empty queue ? */
1285         if (skb_queue_empty(list)) {
1286                 __skb_queue_tail(list, skb);
1287                 return 1;
1288         }
1289
1290         /* Last ? */
1291         if (less(buf_seqno(skb_peek_tail(list)), seq_no)) {
1292                 __skb_queue_tail(list, skb);
1293                 return 1;
1294         }
1295
1296         /* Locate insertion point in queue, then insert; discard if duplicate */
1297         skb_queue_walk(list, skb1) {
1298                 u32 curr_seqno = buf_seqno(skb1);
1299
1300                 if (seq_no == curr_seqno) {
1301                         kfree_skb(skb);
1302                         return 0;
1303                 }
1304
1305                 if (less(seq_no, curr_seqno))
1306                         break;
1307         }
1308
1309         __skb_queue_before(list, skb1, skb);
1310         return 1;
1311 }
1312
1313 /*
1314  * link_handle_out_of_seq_msg - handle arrival of out-of-sequence packet
1315  */
1316 static void link_handle_out_of_seq_msg(struct tipc_link *l_ptr,
1317                                        struct sk_buff *buf)
1318 {
1319         u32 seq_no = buf_seqno(buf);
1320
1321         if (likely(msg_user(buf_msg(buf)) == LINK_PROTOCOL)) {
1322                 tipc_link_proto_rcv(l_ptr, buf);
1323                 return;
1324         }
1325
1326         /* Record OOS packet arrival (force mismatch on next timeout) */
1327         l_ptr->checkpoint--;
1328
1329         /*
1330          * Discard packet if a duplicate; otherwise add it to deferred queue
1331          * and notify peer of gap as per protocol specification
1332          */
1333         if (less(seq_no, mod(l_ptr->next_in_no))) {
1334                 l_ptr->stats.duplicates++;
1335                 kfree_skb(buf);
1336                 return;
1337         }
1338
1339         if (tipc_link_defer_pkt(&l_ptr->deferred_queue, buf)) {
1340                 l_ptr->stats.deferred_recv++;
1341                 if ((skb_queue_len(&l_ptr->deferred_queue) % 16) == 1)
1342                         tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
1343         } else {
1344                 l_ptr->stats.duplicates++;
1345         }
1346 }
1347
1348 /*
1349  * Send protocol message to the other endpoint.
1350  */
1351 void tipc_link_proto_xmit(struct tipc_link *l_ptr, u32 msg_typ, int probe_msg,
1352                           u32 gap, u32 tolerance, u32 priority, u32 ack_mtu)
1353 {
1354         struct sk_buff *buf = NULL;
1355         struct tipc_msg *msg = l_ptr->pmsg;
1356         u32 msg_size = sizeof(l_ptr->proto_msg);
1357         int r_flag;
1358
1359         /* Don't send protocol message during link changeover */
1360         if (l_ptr->exp_msg_count)
1361                 return;
1362
1363         /* Abort non-RESET send if communication with node is prohibited */
1364         if ((tipc_node_blocked(l_ptr->owner)) && (msg_typ != RESET_MSG))
1365                 return;
1366
1367         /* Create protocol message with "out-of-sequence" sequence number */
1368         msg_set_type(msg, msg_typ);
1369         msg_set_net_plane(msg, l_ptr->net_plane);
1370         msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
1371         msg_set_last_bcast(msg, tipc_bclink_get_last_sent(l_ptr->owner->net));
1372
1373         if (msg_typ == STATE_MSG) {
1374                 u32 next_sent = mod(l_ptr->next_out_no);
1375
1376                 if (!tipc_link_is_up(l_ptr))
1377                         return;
1378                 if (l_ptr->next_out)
1379                         next_sent = buf_seqno(l_ptr->next_out);
1380                 msg_set_next_sent(msg, next_sent);
1381                 if (!skb_queue_empty(&l_ptr->deferred_queue)) {
1382                         u32 rec = buf_seqno(skb_peek(&l_ptr->deferred_queue));
1383                         gap = mod(rec - mod(l_ptr->next_in_no));
1384                 }
1385                 msg_set_seq_gap(msg, gap);
1386                 if (gap)
1387                         l_ptr->stats.sent_nacks++;
1388                 msg_set_link_tolerance(msg, tolerance);
1389                 msg_set_linkprio(msg, priority);
1390                 msg_set_max_pkt(msg, ack_mtu);
1391                 msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
1392                 msg_set_probe(msg, probe_msg != 0);
1393                 if (probe_msg) {
1394                         u32 mtu = l_ptr->max_pkt;
1395
1396                         if ((mtu < l_ptr->max_pkt_target) &&
1397                             link_working_working(l_ptr) &&
1398                             l_ptr->fsm_msg_cnt) {
1399                                 msg_size = (mtu + (l_ptr->max_pkt_target - mtu)/2 + 2) & ~3;
1400                                 if (l_ptr->max_pkt_probes == 10) {
1401                                         l_ptr->max_pkt_target = (msg_size - 4);
1402                                         l_ptr->max_pkt_probes = 0;
1403                                         msg_size = (mtu + (l_ptr->max_pkt_target - mtu)/2 + 2) & ~3;
1404                                 }
1405                                 l_ptr->max_pkt_probes++;
1406                         }
1407
1408                         l_ptr->stats.sent_probes++;
1409                 }
1410                 l_ptr->stats.sent_states++;
1411         } else {                /* RESET_MSG or ACTIVATE_MSG */
1412                 msg_set_ack(msg, mod(l_ptr->reset_checkpoint - 1));
1413                 msg_set_seq_gap(msg, 0);
1414                 msg_set_next_sent(msg, 1);
1415                 msg_set_probe(msg, 0);
1416                 msg_set_link_tolerance(msg, l_ptr->tolerance);
1417                 msg_set_linkprio(msg, l_ptr->priority);
1418                 msg_set_max_pkt(msg, l_ptr->max_pkt_target);
1419         }
1420
1421         r_flag = (l_ptr->owner->working_links > tipc_link_is_up(l_ptr));
1422         msg_set_redundant_link(msg, r_flag);
1423         msg_set_linkprio(msg, l_ptr->priority);
1424         msg_set_size(msg, msg_size);
1425
1426         msg_set_seqno(msg, mod(l_ptr->next_out_no + (0xffff/2)));
1427
1428         buf = tipc_buf_acquire(msg_size);
1429         if (!buf)
1430                 return;
1431
1432         skb_copy_to_linear_data(buf, msg, sizeof(l_ptr->proto_msg));
1433         buf->priority = TC_PRIO_CONTROL;
1434
1435         tipc_bearer_send(l_ptr->owner->net, l_ptr->bearer_id, buf,
1436                          &l_ptr->media_addr);
1437         l_ptr->unacked_window = 0;
1438         kfree_skb(buf);
1439 }
1440
1441 /*
1442  * Receive protocol message :
1443  * Note that network plane id propagates through the network, and may
1444  * change at any time. The node with lowest address rules
1445  */
1446 static void tipc_link_proto_rcv(struct tipc_link *l_ptr,
1447                                 struct sk_buff *buf)
1448 {
1449         u32 rec_gap = 0;
1450         u32 max_pkt_info;
1451         u32 max_pkt_ack;
1452         u32 msg_tol;
1453         struct tipc_msg *msg = buf_msg(buf);
1454
1455         /* Discard protocol message during link changeover */
1456         if (l_ptr->exp_msg_count)
1457                 goto exit;
1458
1459         if (l_ptr->net_plane != msg_net_plane(msg))
1460                 if (link_own_addr(l_ptr) > msg_prevnode(msg))
1461                         l_ptr->net_plane = msg_net_plane(msg);
1462
1463         switch (msg_type(msg)) {
1464
1465         case RESET_MSG:
1466                 if (!link_working_unknown(l_ptr) &&
1467                     (l_ptr->peer_session != INVALID_SESSION)) {
1468                         if (less_eq(msg_session(msg), l_ptr->peer_session))
1469                                 break; /* duplicate or old reset: ignore */
1470                 }
1471
1472                 if (!msg_redundant_link(msg) && (link_working_working(l_ptr) ||
1473                                 link_working_unknown(l_ptr))) {
1474                         /*
1475                          * peer has lost contact -- don't allow peer's links
1476                          * to reactivate before we recognize loss & clean up
1477                          */
1478                         l_ptr->owner->action_flags |= TIPC_WAIT_OWN_LINKS_DOWN;
1479                 }
1480
1481                 link_state_event(l_ptr, RESET_MSG);
1482
1483                 /* fall thru' */
1484         case ACTIVATE_MSG:
1485                 /* Update link settings according other endpoint's values */
1486                 strcpy((strrchr(l_ptr->name, ':') + 1), (char *)msg_data(msg));
1487
1488                 msg_tol = msg_link_tolerance(msg);
1489                 if (msg_tol > l_ptr->tolerance)
1490                         link_set_supervision_props(l_ptr, msg_tol);
1491
1492                 if (msg_linkprio(msg) > l_ptr->priority)
1493                         l_ptr->priority = msg_linkprio(msg);
1494
1495                 max_pkt_info = msg_max_pkt(msg);
1496                 if (max_pkt_info) {
1497                         if (max_pkt_info < l_ptr->max_pkt_target)
1498                                 l_ptr->max_pkt_target = max_pkt_info;
1499                         if (l_ptr->max_pkt > l_ptr->max_pkt_target)
1500                                 l_ptr->max_pkt = l_ptr->max_pkt_target;
1501                 } else {
1502                         l_ptr->max_pkt = l_ptr->max_pkt_target;
1503                 }
1504
1505                 /* Synchronize broadcast link info, if not done previously */
1506                 if (!tipc_node_is_up(l_ptr->owner)) {
1507                         l_ptr->owner->bclink.last_sent =
1508                                 l_ptr->owner->bclink.last_in =
1509                                 msg_last_bcast(msg);
1510                         l_ptr->owner->bclink.oos_state = 0;
1511                 }
1512
1513                 l_ptr->peer_session = msg_session(msg);
1514                 l_ptr->peer_bearer_id = msg_bearer_id(msg);
1515
1516                 if (msg_type(msg) == ACTIVATE_MSG)
1517                         link_state_event(l_ptr, ACTIVATE_MSG);
1518                 break;
1519         case STATE_MSG:
1520
1521                 msg_tol = msg_link_tolerance(msg);
1522                 if (msg_tol)
1523                         link_set_supervision_props(l_ptr, msg_tol);
1524
1525                 if (msg_linkprio(msg) &&
1526                     (msg_linkprio(msg) != l_ptr->priority)) {
1527                         pr_debug("%s<%s>, priority change %u->%u\n",
1528                                  link_rst_msg, l_ptr->name,
1529                                  l_ptr->priority, msg_linkprio(msg));
1530                         l_ptr->priority = msg_linkprio(msg);
1531                         tipc_link_reset(l_ptr); /* Enforce change to take effect */
1532                         break;
1533                 }
1534
1535                 /* Record reception; force mismatch at next timeout: */
1536                 l_ptr->checkpoint--;
1537
1538                 link_state_event(l_ptr, TRAFFIC_MSG_EVT);
1539                 l_ptr->stats.recv_states++;
1540                 if (link_reset_unknown(l_ptr))
1541                         break;
1542
1543                 if (less_eq(mod(l_ptr->next_in_no), msg_next_sent(msg))) {
1544                         rec_gap = mod(msg_next_sent(msg) -
1545                                       mod(l_ptr->next_in_no));
1546                 }
1547
1548                 max_pkt_ack = msg_max_pkt(msg);
1549                 if (max_pkt_ack > l_ptr->max_pkt) {
1550                         l_ptr->max_pkt = max_pkt_ack;
1551                         l_ptr->max_pkt_probes = 0;
1552                 }
1553
1554                 max_pkt_ack = 0;
1555                 if (msg_probe(msg)) {
1556                         l_ptr->stats.recv_probes++;
1557                         if (msg_size(msg) > sizeof(l_ptr->proto_msg))
1558                                 max_pkt_ack = msg_size(msg);
1559                 }
1560
1561                 /* Protocol message before retransmits, reduce loss risk */
1562                 if (l_ptr->owner->bclink.recv_permitted)
1563                         tipc_bclink_update_link_state(l_ptr->owner,
1564                                                       msg_last_bcast(msg));
1565
1566                 if (rec_gap || (msg_probe(msg))) {
1567                         tipc_link_proto_xmit(l_ptr, STATE_MSG, 0, rec_gap, 0,
1568                                              0, max_pkt_ack);
1569                 }
1570                 if (msg_seq_gap(msg)) {
1571                         l_ptr->stats.recv_nacks++;
1572                         tipc_link_retransmit(l_ptr, skb_peek(&l_ptr->outqueue),
1573                                              msg_seq_gap(msg));
1574                 }
1575                 break;
1576         }
1577 exit:
1578         kfree_skb(buf);
1579 }
1580
1581
1582 /* tipc_link_tunnel_xmit(): Tunnel one packet via a link belonging to
1583  * a different bearer. Owner node is locked.
1584  */
1585 static void tipc_link_tunnel_xmit(struct tipc_link *l_ptr,
1586                                   struct tipc_msg *tunnel_hdr,
1587                                   struct tipc_msg *msg,
1588                                   u32 selector)
1589 {
1590         struct tipc_link *tunnel;
1591         struct sk_buff *skb;
1592         u32 length = msg_size(msg);
1593
1594         tunnel = l_ptr->owner->active_links[selector & 1];
1595         if (!tipc_link_is_up(tunnel)) {
1596                 pr_warn("%stunnel link no longer available\n", link_co_err);
1597                 return;
1598         }
1599         msg_set_size(tunnel_hdr, length + INT_H_SIZE);
1600         skb = tipc_buf_acquire(length + INT_H_SIZE);
1601         if (!skb) {
1602                 pr_warn("%sunable to send tunnel msg\n", link_co_err);
1603                 return;
1604         }
1605         skb_copy_to_linear_data(skb, tunnel_hdr, INT_H_SIZE);
1606         skb_copy_to_linear_data_offset(skb, INT_H_SIZE, msg, length);
1607         __tipc_link_xmit_skb(tunnel, skb);
1608 }
1609
1610
1611 /* tipc_link_failover_send_queue(): A link has gone down, but a second
1612  * link is still active. We can do failover. Tunnel the failing link's
1613  * whole send queue via the remaining link. This way, we don't lose
1614  * any packets, and sequence order is preserved for subsequent traffic
1615  * sent over the remaining link. Owner node is locked.
1616  */
1617 void tipc_link_failover_send_queue(struct tipc_link *l_ptr)
1618 {
1619         u32 msgcount = skb_queue_len(&l_ptr->outqueue);
1620         struct tipc_link *tunnel = l_ptr->owner->active_links[0];
1621         struct tipc_msg tunnel_hdr;
1622         struct sk_buff *skb;
1623         int split_bundles;
1624
1625         if (!tunnel)
1626                 return;
1627
1628         tipc_msg_init(link_own_addr(l_ptr), &tunnel_hdr, CHANGEOVER_PROTOCOL,
1629                       ORIGINAL_MSG, INT_H_SIZE, l_ptr->addr);
1630         msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id);
1631         msg_set_msgcnt(&tunnel_hdr, msgcount);
1632
1633         if (skb_queue_empty(&l_ptr->outqueue)) {
1634                 skb = tipc_buf_acquire(INT_H_SIZE);
1635                 if (skb) {
1636                         skb_copy_to_linear_data(skb, &tunnel_hdr, INT_H_SIZE);
1637                         msg_set_size(&tunnel_hdr, INT_H_SIZE);
1638                         __tipc_link_xmit_skb(tunnel, skb);
1639                 } else {
1640                         pr_warn("%sunable to send changeover msg\n",
1641                                 link_co_err);
1642                 }
1643                 return;
1644         }
1645
1646         split_bundles = (l_ptr->owner->active_links[0] !=
1647                          l_ptr->owner->active_links[1]);
1648
1649         skb_queue_walk(&l_ptr->outqueue, skb) {
1650                 struct tipc_msg *msg = buf_msg(skb);
1651
1652                 if ((msg_user(msg) == MSG_BUNDLER) && split_bundles) {
1653                         struct tipc_msg *m = msg_get_wrapped(msg);
1654                         unchar *pos = (unchar *)m;
1655
1656                         msgcount = msg_msgcnt(msg);
1657                         while (msgcount--) {
1658                                 msg_set_seqno(m, msg_seqno(msg));
1659                                 tipc_link_tunnel_xmit(l_ptr, &tunnel_hdr, m,
1660                                                       msg_link_selector(m));
1661                                 pos += align(msg_size(m));
1662                                 m = (struct tipc_msg *)pos;
1663                         }
1664                 } else {
1665                         tipc_link_tunnel_xmit(l_ptr, &tunnel_hdr, msg,
1666                                               msg_link_selector(msg));
1667                 }
1668         }
1669 }
1670
1671 /* tipc_link_dup_queue_xmit(): A second link has become active. Tunnel a
1672  * duplicate of the first link's send queue via the new link. This way, we
1673  * are guaranteed that currently queued packets from a socket are delivered
1674  * before future traffic from the same socket, even if this is using the
1675  * new link. The last arriving copy of each duplicate packet is dropped at
1676  * the receiving end by the regular protocol check, so packet cardinality
1677  * and sequence order is preserved per sender/receiver socket pair.
1678  * Owner node is locked.
1679  */
1680 void tipc_link_dup_queue_xmit(struct tipc_link *l_ptr,
1681                               struct tipc_link *tunnel)
1682 {
1683         struct sk_buff *skb;
1684         struct tipc_msg tunnel_hdr;
1685
1686         tipc_msg_init(link_own_addr(l_ptr), &tunnel_hdr, CHANGEOVER_PROTOCOL,
1687                       DUPLICATE_MSG, INT_H_SIZE, l_ptr->addr);
1688         msg_set_msgcnt(&tunnel_hdr, skb_queue_len(&l_ptr->outqueue));
1689         msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id);
1690         skb_queue_walk(&l_ptr->outqueue, skb) {
1691                 struct sk_buff *outskb;
1692                 struct tipc_msg *msg = buf_msg(skb);
1693                 u32 length = msg_size(msg);
1694
1695                 if (msg_user(msg) == MSG_BUNDLER)
1696                         msg_set_type(msg, CLOSED_MSG);
1697                 msg_set_ack(msg, mod(l_ptr->next_in_no - 1));   /* Update */
1698                 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
1699                 msg_set_size(&tunnel_hdr, length + INT_H_SIZE);
1700                 outskb = tipc_buf_acquire(length + INT_H_SIZE);
1701                 if (outskb == NULL) {
1702                         pr_warn("%sunable to send duplicate msg\n",
1703                                 link_co_err);
1704                         return;
1705                 }
1706                 skb_copy_to_linear_data(outskb, &tunnel_hdr, INT_H_SIZE);
1707                 skb_copy_to_linear_data_offset(outskb, INT_H_SIZE, skb->data,
1708                                                length);
1709                 __tipc_link_xmit_skb(tunnel, outskb);
1710                 if (!tipc_link_is_up(l_ptr))
1711                         return;
1712         }
1713 }
1714
1715 /* tipc_link_dup_rcv(): Receive a tunnelled DUPLICATE_MSG packet.
1716  * Owner node is locked.
1717  */
1718 static void tipc_link_dup_rcv(struct tipc_link *link,
1719                               struct sk_buff *skb)
1720 {
1721         struct sk_buff *iskb;
1722         int pos = 0;
1723
1724         if (!tipc_link_is_up(link))
1725                 return;
1726
1727         if (!tipc_msg_extract(skb, &iskb, &pos)) {
1728                 pr_warn("%sfailed to extract inner dup pkt\n", link_co_err);
1729                 return;
1730         }
1731         /* Append buffer to deferred queue, if applicable: */
1732         link_handle_out_of_seq_msg(link, iskb);
1733 }
1734
1735 /*  tipc_link_failover_rcv(): Receive a tunnelled ORIGINAL_MSG packet
1736  *  Owner node is locked.
1737  */
1738 static struct sk_buff *tipc_link_failover_rcv(struct tipc_link *l_ptr,
1739                                               struct sk_buff *t_buf)
1740 {
1741         struct tipc_msg *t_msg = buf_msg(t_buf);
1742         struct sk_buff *buf = NULL;
1743         struct tipc_msg *msg;
1744         int pos = 0;
1745
1746         if (tipc_link_is_up(l_ptr))
1747                 tipc_link_reset(l_ptr);
1748
1749         /* First failover packet? */
1750         if (l_ptr->exp_msg_count == START_CHANGEOVER)
1751                 l_ptr->exp_msg_count = msg_msgcnt(t_msg);
1752
1753         /* Should there be an inner packet? */
1754         if (l_ptr->exp_msg_count) {
1755                 l_ptr->exp_msg_count--;
1756                 if (!tipc_msg_extract(t_buf, &buf, &pos)) {
1757                         pr_warn("%sno inner failover pkt\n", link_co_err);
1758                         goto exit;
1759                 }
1760                 msg = buf_msg(buf);
1761
1762                 if (less(msg_seqno(msg), l_ptr->reset_checkpoint)) {
1763                         kfree_skb(buf);
1764                         buf = NULL;
1765                         goto exit;
1766                 }
1767                 if (msg_user(msg) == MSG_FRAGMENTER) {
1768                         l_ptr->stats.recv_fragments++;
1769                         tipc_buf_append(&l_ptr->reasm_buf, &buf);
1770                 }
1771         }
1772 exit:
1773         if ((!l_ptr->exp_msg_count) && (l_ptr->flags & LINK_STOPPED))
1774                 tipc_link_delete(l_ptr);
1775         return buf;
1776 }
1777
1778 /*  tipc_link_tunnel_rcv(): Receive a tunnelled packet, sent
1779  *  via other link as result of a failover (ORIGINAL_MSG) or
1780  *  a new active link (DUPLICATE_MSG). Failover packets are
1781  *  returned to the active link for delivery upwards.
1782  *  Owner node is locked.
1783  */
1784 static int tipc_link_tunnel_rcv(struct tipc_node *n_ptr,
1785                                 struct sk_buff **buf)
1786 {
1787         struct sk_buff *t_buf = *buf;
1788         struct tipc_link *l_ptr;
1789         struct tipc_msg *t_msg = buf_msg(t_buf);
1790         u32 bearer_id = msg_bearer_id(t_msg);
1791
1792         *buf = NULL;
1793
1794         if (bearer_id >= MAX_BEARERS)
1795                 goto exit;
1796
1797         l_ptr = n_ptr->links[bearer_id];
1798         if (!l_ptr)
1799                 goto exit;
1800
1801         if (msg_type(t_msg) == DUPLICATE_MSG)
1802                 tipc_link_dup_rcv(l_ptr, t_buf);
1803         else if (msg_type(t_msg) == ORIGINAL_MSG)
1804                 *buf = tipc_link_failover_rcv(l_ptr, t_buf);
1805         else
1806                 pr_warn("%sunknown tunnel pkt received\n", link_co_err);
1807 exit:
1808         kfree_skb(t_buf);
1809         return *buf != NULL;
1810 }
1811
1812 static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tol)
1813 {
1814         unsigned long intv = ((tol / 4) > 500) ? 500 : tol / 4;
1815
1816         if ((tol < TIPC_MIN_LINK_TOL) || (tol > TIPC_MAX_LINK_TOL))
1817                 return;
1818
1819         l_ptr->tolerance = tol;
1820         l_ptr->cont_intv = msecs_to_jiffies(intv);
1821         l_ptr->abort_limit = tol / (jiffies_to_msecs(l_ptr->cont_intv) / 4);
1822 }
1823
1824 void tipc_link_set_queue_limits(struct tipc_link *l_ptr, u32 window)
1825 {
1826         /* Data messages from this node, inclusive FIRST_FRAGM */
1827         l_ptr->queue_limit[TIPC_LOW_IMPORTANCE] = window;
1828         l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE] = (window / 3) * 4;
1829         l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE] = (window / 3) * 5;
1830         l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE] = (window / 3) * 6;
1831         /* Transiting data messages,inclusive FIRST_FRAGM */
1832         l_ptr->queue_limit[TIPC_LOW_IMPORTANCE + 4] = 300;
1833         l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE + 4] = 600;
1834         l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE + 4] = 900;
1835         l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE + 4] = 1200;
1836         l_ptr->queue_limit[CONN_MANAGER] = 1200;
1837         l_ptr->queue_limit[CHANGEOVER_PROTOCOL] = 2500;
1838         l_ptr->queue_limit[NAME_DISTRIBUTOR] = 3000;
1839         /* FRAGMENT and LAST_FRAGMENT packets */
1840         l_ptr->queue_limit[MSG_FRAGMENTER] = 4000;
1841 }
1842
1843 /* tipc_link_find_owner - locate owner node of link by link's name
1844  * @net: the applicable net namespace
1845  * @name: pointer to link name string
1846  * @bearer_id: pointer to index in 'node->links' array where the link was found.
1847  *
1848  * Returns pointer to node owning the link, or 0 if no matching link is found.
1849  */
1850 static struct tipc_node *tipc_link_find_owner(struct net *net,
1851                                               const char *link_name,
1852                                               unsigned int *bearer_id)
1853 {
1854         struct tipc_net *tn = net_generic(net, tipc_net_id);
1855         struct tipc_link *l_ptr;
1856         struct tipc_node *n_ptr;
1857         struct tipc_node *found_node = NULL;
1858         int i;
1859
1860         *bearer_id = 0;
1861         rcu_read_lock();
1862         list_for_each_entry_rcu(n_ptr, &tn->node_list, list) {
1863                 tipc_node_lock(n_ptr);
1864                 for (i = 0; i < MAX_BEARERS; i++) {
1865                         l_ptr = n_ptr->links[i];
1866                         if (l_ptr && !strcmp(l_ptr->name, link_name)) {
1867                                 *bearer_id = i;
1868                                 found_node = n_ptr;
1869                                 break;
1870                         }
1871                 }
1872                 tipc_node_unlock(n_ptr);
1873                 if (found_node)
1874                         break;
1875         }
1876         rcu_read_unlock();
1877
1878         return found_node;
1879 }
1880
1881 /**
1882  * link_reset_statistics - reset link statistics
1883  * @l_ptr: pointer to link
1884  */
1885 static void link_reset_statistics(struct tipc_link *l_ptr)
1886 {
1887         memset(&l_ptr->stats, 0, sizeof(l_ptr->stats));
1888         l_ptr->stats.sent_info = l_ptr->next_out_no;
1889         l_ptr->stats.recv_info = l_ptr->next_in_no;
1890 }
1891
1892 static void link_print(struct tipc_link *l_ptr, const char *str)
1893 {
1894         struct tipc_net *tn = net_generic(l_ptr->owner->net, tipc_net_id);
1895         struct tipc_bearer *b_ptr;
1896
1897         rcu_read_lock();
1898         b_ptr = rcu_dereference_rtnl(tn->bearer_list[l_ptr->bearer_id]);
1899         if (b_ptr)
1900                 pr_info("%s Link %x<%s>:", str, l_ptr->addr, b_ptr->name);
1901         rcu_read_unlock();
1902
1903         if (link_working_unknown(l_ptr))
1904                 pr_cont(":WU\n");
1905         else if (link_reset_reset(l_ptr))
1906                 pr_cont(":RR\n");
1907         else if (link_reset_unknown(l_ptr))
1908                 pr_cont(":RU\n");
1909         else if (link_working_working(l_ptr))
1910                 pr_cont(":WW\n");
1911         else
1912                 pr_cont("\n");
1913 }
1914
1915 /* Parse and validate nested (link) properties valid for media, bearer and link
1916  */
1917 int tipc_nl_parse_link_prop(struct nlattr *prop, struct nlattr *props[])
1918 {
1919         int err;
1920
1921         err = nla_parse_nested(props, TIPC_NLA_PROP_MAX, prop,
1922                                tipc_nl_prop_policy);
1923         if (err)
1924                 return err;
1925
1926         if (props[TIPC_NLA_PROP_PRIO]) {
1927                 u32 prio;
1928
1929                 prio = nla_get_u32(props[TIPC_NLA_PROP_PRIO]);
1930                 if (prio > TIPC_MAX_LINK_PRI)
1931                         return -EINVAL;
1932         }
1933
1934         if (props[TIPC_NLA_PROP_TOL]) {
1935                 u32 tol;
1936
1937                 tol = nla_get_u32(props[TIPC_NLA_PROP_TOL]);
1938                 if ((tol < TIPC_MIN_LINK_TOL) || (tol > TIPC_MAX_LINK_TOL))
1939                         return -EINVAL;
1940         }
1941
1942         if (props[TIPC_NLA_PROP_WIN]) {
1943                 u32 win;
1944
1945                 win = nla_get_u32(props[TIPC_NLA_PROP_WIN]);
1946                 if ((win < TIPC_MIN_LINK_WIN) || (win > TIPC_MAX_LINK_WIN))
1947                         return -EINVAL;
1948         }
1949
1950         return 0;
1951 }
1952
1953 int tipc_nl_link_set(struct sk_buff *skb, struct genl_info *info)
1954 {
1955         int err;
1956         int res = 0;
1957         int bearer_id;
1958         char *name;
1959         struct tipc_link *link;
1960         struct tipc_node *node;
1961         struct nlattr *attrs[TIPC_NLA_LINK_MAX + 1];
1962         struct net *net = sock_net(skb->sk);
1963
1964         if (!info->attrs[TIPC_NLA_LINK])
1965                 return -EINVAL;
1966
1967         err = nla_parse_nested(attrs, TIPC_NLA_LINK_MAX,
1968                                info->attrs[TIPC_NLA_LINK],
1969                                tipc_nl_link_policy);
1970         if (err)
1971                 return err;
1972
1973         if (!attrs[TIPC_NLA_LINK_NAME])
1974                 return -EINVAL;
1975
1976         name = nla_data(attrs[TIPC_NLA_LINK_NAME]);
1977
1978         node = tipc_link_find_owner(net, name, &bearer_id);
1979         if (!node)
1980                 return -EINVAL;
1981
1982         tipc_node_lock(node);
1983
1984         link = node->links[bearer_id];
1985         if (!link) {
1986                 res = -EINVAL;
1987                 goto out;
1988         }
1989
1990         if (attrs[TIPC_NLA_LINK_PROP]) {
1991                 struct nlattr *props[TIPC_NLA_PROP_MAX + 1];
1992
1993                 err = tipc_nl_parse_link_prop(attrs[TIPC_NLA_LINK_PROP],
1994                                               props);
1995                 if (err) {
1996                         res = err;
1997                         goto out;
1998                 }
1999
2000                 if (props[TIPC_NLA_PROP_TOL]) {
2001                         u32 tol;
2002
2003                         tol = nla_get_u32(props[TIPC_NLA_PROP_TOL]);
2004                         link_set_supervision_props(link, tol);
2005                         tipc_link_proto_xmit(link, STATE_MSG, 0, 0, tol, 0, 0);
2006                 }
2007                 if (props[TIPC_NLA_PROP_PRIO]) {
2008                         u32 prio;
2009
2010                         prio = nla_get_u32(props[TIPC_NLA_PROP_PRIO]);
2011                         link->priority = prio;
2012                         tipc_link_proto_xmit(link, STATE_MSG, 0, 0, 0, prio, 0);
2013                 }
2014                 if (props[TIPC_NLA_PROP_WIN]) {
2015                         u32 win;
2016
2017                         win = nla_get_u32(props[TIPC_NLA_PROP_WIN]);
2018                         tipc_link_set_queue_limits(link, win);
2019                 }
2020         }
2021
2022 out:
2023         tipc_node_unlock(node);
2024
2025         return res;
2026 }
2027
2028 static int __tipc_nl_add_stats(struct sk_buff *skb, struct tipc_stats *s)
2029 {
2030         int i;
2031         struct nlattr *stats;
2032
2033         struct nla_map {
2034                 u32 key;
2035                 u32 val;
2036         };
2037
2038         struct nla_map map[] = {
2039                 {TIPC_NLA_STATS_RX_INFO, s->recv_info},
2040                 {TIPC_NLA_STATS_RX_FRAGMENTS, s->recv_fragments},
2041                 {TIPC_NLA_STATS_RX_FRAGMENTED, s->recv_fragmented},
2042                 {TIPC_NLA_STATS_RX_BUNDLES, s->recv_bundles},
2043                 {TIPC_NLA_STATS_RX_BUNDLED, s->recv_bundled},
2044                 {TIPC_NLA_STATS_TX_INFO, s->sent_info},
2045                 {TIPC_NLA_STATS_TX_FRAGMENTS, s->sent_fragments},
2046                 {TIPC_NLA_STATS_TX_FRAGMENTED, s->sent_fragmented},
2047                 {TIPC_NLA_STATS_TX_BUNDLES, s->sent_bundles},
2048                 {TIPC_NLA_STATS_TX_BUNDLED, s->sent_bundled},
2049                 {TIPC_NLA_STATS_MSG_PROF_TOT, (s->msg_length_counts) ?
2050                         s->msg_length_counts : 1},
2051                 {TIPC_NLA_STATS_MSG_LEN_CNT, s->msg_length_counts},
2052                 {TIPC_NLA_STATS_MSG_LEN_TOT, s->msg_lengths_total},
2053                 {TIPC_NLA_STATS_MSG_LEN_P0, s->msg_length_profile[0]},
2054                 {TIPC_NLA_STATS_MSG_LEN_P1, s->msg_length_profile[1]},
2055                 {TIPC_NLA_STATS_MSG_LEN_P2, s->msg_length_profile[2]},
2056                 {TIPC_NLA_STATS_MSG_LEN_P3, s->msg_length_profile[3]},
2057                 {TIPC_NLA_STATS_MSG_LEN_P4, s->msg_length_profile[4]},
2058                 {TIPC_NLA_STATS_MSG_LEN_P5, s->msg_length_profile[5]},
2059                 {TIPC_NLA_STATS_MSG_LEN_P6, s->msg_length_profile[6]},
2060                 {TIPC_NLA_STATS_RX_STATES, s->recv_states},
2061                 {TIPC_NLA_STATS_RX_PROBES, s->recv_probes},
2062                 {TIPC_NLA_STATS_RX_NACKS, s->recv_nacks},
2063                 {TIPC_NLA_STATS_RX_DEFERRED, s->deferred_recv},
2064                 {TIPC_NLA_STATS_TX_STATES, s->sent_states},
2065                 {TIPC_NLA_STATS_TX_PROBES, s->sent_probes},
2066                 {TIPC_NLA_STATS_TX_NACKS, s->sent_nacks},
2067                 {TIPC_NLA_STATS_TX_ACKS, s->sent_acks},
2068                 {TIPC_NLA_STATS_RETRANSMITTED, s->retransmitted},
2069                 {TIPC_NLA_STATS_DUPLICATES, s->duplicates},
2070                 {TIPC_NLA_STATS_LINK_CONGS, s->link_congs},
2071                 {TIPC_NLA_STATS_MAX_QUEUE, s->max_queue_sz},
2072                 {TIPC_NLA_STATS_AVG_QUEUE, s->queue_sz_counts ?
2073                         (s->accu_queue_sz / s->queue_sz_counts) : 0}
2074         };
2075
2076         stats = nla_nest_start(skb, TIPC_NLA_LINK_STATS);
2077         if (!stats)
2078                 return -EMSGSIZE;
2079
2080         for (i = 0; i <  ARRAY_SIZE(map); i++)
2081                 if (nla_put_u32(skb, map[i].key, map[i].val))
2082                         goto msg_full;
2083
2084         nla_nest_end(skb, stats);
2085
2086         return 0;
2087 msg_full:
2088         nla_nest_cancel(skb, stats);
2089
2090         return -EMSGSIZE;
2091 }
2092
2093 /* Caller should hold appropriate locks to protect the link */
2094 static int __tipc_nl_add_link(struct net *net, struct tipc_nl_msg *msg,
2095                               struct tipc_link *link)
2096 {
2097         int err;
2098         void *hdr;
2099         struct nlattr *attrs;
2100         struct nlattr *prop;
2101         struct tipc_net *tn = net_generic(net, tipc_net_id);
2102
2103         hdr = genlmsg_put(msg->skb, msg->portid, msg->seq, &tipc_genl_family,
2104                           NLM_F_MULTI, TIPC_NL_LINK_GET);
2105         if (!hdr)
2106                 return -EMSGSIZE;
2107
2108         attrs = nla_nest_start(msg->skb, TIPC_NLA_LINK);
2109         if (!attrs)
2110                 goto msg_full;
2111
2112         if (nla_put_string(msg->skb, TIPC_NLA_LINK_NAME, link->name))
2113                 goto attr_msg_full;
2114         if (nla_put_u32(msg->skb, TIPC_NLA_LINK_DEST,
2115                         tipc_cluster_mask(tn->own_addr)))
2116                 goto attr_msg_full;
2117         if (nla_put_u32(msg->skb, TIPC_NLA_LINK_MTU, link->max_pkt))
2118                 goto attr_msg_full;
2119         if (nla_put_u32(msg->skb, TIPC_NLA_LINK_RX, link->next_in_no))
2120                 goto attr_msg_full;
2121         if (nla_put_u32(msg->skb, TIPC_NLA_LINK_TX, link->next_out_no))
2122                 goto attr_msg_full;
2123
2124         if (tipc_link_is_up(link))
2125                 if (nla_put_flag(msg->skb, TIPC_NLA_LINK_UP))
2126                         goto attr_msg_full;
2127         if (tipc_link_is_active(link))
2128                 if (nla_put_flag(msg->skb, TIPC_NLA_LINK_ACTIVE))
2129                         goto attr_msg_full;
2130
2131         prop = nla_nest_start(msg->skb, TIPC_NLA_LINK_PROP);
2132         if (!prop)
2133                 goto attr_msg_full;
2134         if (nla_put_u32(msg->skb, TIPC_NLA_PROP_PRIO, link->priority))
2135                 goto prop_msg_full;
2136         if (nla_put_u32(msg->skb, TIPC_NLA_PROP_TOL, link->tolerance))
2137                 goto prop_msg_full;
2138         if (nla_put_u32(msg->skb, TIPC_NLA_PROP_WIN,
2139                         link->queue_limit[TIPC_LOW_IMPORTANCE]))
2140                 goto prop_msg_full;
2141         if (nla_put_u32(msg->skb, TIPC_NLA_PROP_PRIO, link->priority))
2142                 goto prop_msg_full;
2143         nla_nest_end(msg->skb, prop);
2144
2145         err = __tipc_nl_add_stats(msg->skb, &link->stats);
2146         if (err)
2147                 goto attr_msg_full;
2148
2149         nla_nest_end(msg->skb, attrs);
2150         genlmsg_end(msg->skb, hdr);
2151
2152         return 0;
2153
2154 prop_msg_full:
2155         nla_nest_cancel(msg->skb, prop);
2156 attr_msg_full:
2157         nla_nest_cancel(msg->skb, attrs);
2158 msg_full:
2159         genlmsg_cancel(msg->skb, hdr);
2160
2161         return -EMSGSIZE;
2162 }
2163
2164 /* Caller should hold node lock  */
2165 static int __tipc_nl_add_node_links(struct net *net, struct tipc_nl_msg *msg,
2166                                     struct tipc_node *node, u32 *prev_link)
2167 {
2168         u32 i;
2169         int err;
2170
2171         for (i = *prev_link; i < MAX_BEARERS; i++) {
2172                 *prev_link = i;
2173
2174                 if (!node->links[i])
2175                         continue;
2176
2177                 err = __tipc_nl_add_link(net, msg, node->links[i]);
2178                 if (err)
2179                         return err;
2180         }
2181         *prev_link = 0;
2182
2183         return 0;
2184 }
2185
2186 int tipc_nl_link_dump(struct sk_buff *skb, struct netlink_callback *cb)
2187 {
2188         struct net *net = sock_net(skb->sk);
2189         struct tipc_net *tn = net_generic(net, tipc_net_id);
2190         struct tipc_node *node;
2191         struct tipc_nl_msg msg;
2192         u32 prev_node = cb->args[0];
2193         u32 prev_link = cb->args[1];
2194         int done = cb->args[2];
2195         int err;
2196
2197         if (done)
2198                 return 0;
2199
2200         msg.skb = skb;
2201         msg.portid = NETLINK_CB(cb->skb).portid;
2202         msg.seq = cb->nlh->nlmsg_seq;
2203
2204         rcu_read_lock();
2205
2206         if (prev_node) {
2207                 node = tipc_node_find(net, prev_node);
2208                 if (!node) {
2209                         /* We never set seq or call nl_dump_check_consistent()
2210                          * this means that setting prev_seq here will cause the
2211                          * consistence check to fail in the netlink callback
2212                          * handler. Resulting in the last NLMSG_DONE message
2213                          * having the NLM_F_DUMP_INTR flag set.
2214                          */
2215                         cb->prev_seq = 1;
2216                         goto out;
2217                 }
2218
2219                 list_for_each_entry_continue_rcu(node, &tn->node_list,
2220                                                  list) {
2221                         tipc_node_lock(node);
2222                         err = __tipc_nl_add_node_links(net, &msg, node,
2223                                                        &prev_link);
2224                         tipc_node_unlock(node);
2225                         if (err)
2226                                 goto out;
2227
2228                         prev_node = node->addr;
2229                 }
2230         } else {
2231                 err = tipc_nl_add_bc_link(net, &msg);
2232                 if (err)
2233                         goto out;
2234
2235                 list_for_each_entry_rcu(node, &tn->node_list, list) {
2236                         tipc_node_lock(node);
2237                         err = __tipc_nl_add_node_links(net, &msg, node,
2238                                                        &prev_link);
2239                         tipc_node_unlock(node);
2240                         if (err)
2241                                 goto out;
2242
2243                         prev_node = node->addr;
2244                 }
2245         }
2246         done = 1;
2247 out:
2248         rcu_read_unlock();
2249
2250         cb->args[0] = prev_node;
2251         cb->args[1] = prev_link;
2252         cb->args[2] = done;
2253
2254         return skb->len;
2255 }
2256
2257 int tipc_nl_link_get(struct sk_buff *skb, struct genl_info *info)
2258 {
2259         struct net *net = genl_info_net(info);
2260         struct sk_buff *ans_skb;
2261         struct tipc_nl_msg msg;
2262         struct tipc_link *link;
2263         struct tipc_node *node;
2264         char *name;
2265         int bearer_id;
2266         int err;
2267
2268         if (!info->attrs[TIPC_NLA_LINK_NAME])
2269                 return -EINVAL;
2270
2271         name = nla_data(info->attrs[TIPC_NLA_LINK_NAME]);
2272         node = tipc_link_find_owner(net, name, &bearer_id);
2273         if (!node)
2274                 return -EINVAL;
2275
2276         ans_skb = nlmsg_new(NLMSG_GOODSIZE, GFP_KERNEL);
2277         if (!ans_skb)
2278                 return -ENOMEM;
2279
2280         msg.skb = ans_skb;
2281         msg.portid = info->snd_portid;
2282         msg.seq = info->snd_seq;
2283
2284         tipc_node_lock(node);
2285         link = node->links[bearer_id];
2286         if (!link) {
2287                 err = -EINVAL;
2288                 goto err_out;
2289         }
2290
2291         err = __tipc_nl_add_link(net, &msg, link);
2292         if (err)
2293                 goto err_out;
2294
2295         tipc_node_unlock(node);
2296
2297         return genlmsg_reply(ans_skb, info);
2298
2299 err_out:
2300         tipc_node_unlock(node);
2301         nlmsg_free(ans_skb);
2302
2303         return err;
2304 }
2305
2306 int tipc_nl_link_reset_stats(struct sk_buff *skb, struct genl_info *info)
2307 {
2308         int err;
2309         char *link_name;
2310         unsigned int bearer_id;
2311         struct tipc_link *link;
2312         struct tipc_node *node;
2313         struct nlattr *attrs[TIPC_NLA_LINK_MAX + 1];
2314         struct net *net = sock_net(skb->sk);
2315
2316         if (!info->attrs[TIPC_NLA_LINK])
2317                 return -EINVAL;
2318
2319         err = nla_parse_nested(attrs, TIPC_NLA_LINK_MAX,
2320                                info->attrs[TIPC_NLA_LINK],
2321                                tipc_nl_link_policy);
2322         if (err)
2323                 return err;
2324
2325         if (!attrs[TIPC_NLA_LINK_NAME])
2326                 return -EINVAL;
2327
2328         link_name = nla_data(attrs[TIPC_NLA_LINK_NAME]);
2329
2330         if (strcmp(link_name, tipc_bclink_name) == 0) {
2331                 err = tipc_bclink_reset_stats(net);
2332                 if (err)
2333                         return err;
2334                 return 0;
2335         }
2336
2337         node = tipc_link_find_owner(net, link_name, &bearer_id);
2338         if (!node)
2339                 return -EINVAL;
2340
2341         tipc_node_lock(node);
2342
2343         link = node->links[bearer_id];
2344         if (!link) {
2345                 tipc_node_unlock(node);
2346                 return -EINVAL;
2347         }
2348
2349         link_reset_statistics(link);
2350
2351         tipc_node_unlock(node);
2352
2353         return 0;
2354 }