net/tipc/link.c

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