2 * net/tipc/link.c: TIPC link code
4 * Copyright (c) 1996-2007, 2012, Ericsson AB
5 * Copyright (c) 2004-2007, 2010-2013, Wind River Systems
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are met:
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.
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.
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.
40 #include "name_distr.h"
44 #include <linux/pkt_sched.h>
47 * Error message prefixes
49 static const char *link_co_err = "Link changeover error, ";
50 static const char *link_rst_msg = "Resetting link ";
51 static const char *link_unk_evt = "Unknown link event ";
54 * Out-of-range value for link session numbers
56 #define INVALID_SESSION 0x10000
61 #define STARTING_EVT 856384768 /* link processing trigger */
62 #define TRAFFIC_MSG_EVT 560815u /* rx'd ??? */
63 #define TIMEOUT_EVT 560817u /* link timer expired */
66 * The following two 'message types' is really just implementation
67 * data conveniently stored in the message header.
68 * They must not be considered part of the protocol
74 * State value stored in 'exp_msg_count'
76 #define START_CHANGEOVER 100000u
79 * struct tipc_link_name - deconstructed link name
80 * @addr_local: network address of node at this end
81 * @if_local: name of interface at this end
82 * @addr_peer: network address of node at far end
83 * @if_peer: name of interface at far end
85 struct tipc_link_name {
87 char if_local[TIPC_MAX_IF_NAME];
89 char if_peer[TIPC_MAX_IF_NAME];
92 static void link_handle_out_of_seq_msg(struct tipc_link *l_ptr,
94 static void link_recv_proto_msg(struct tipc_link *l_ptr, struct sk_buff *buf);
95 static int link_recv_changeover_msg(struct tipc_link **l_ptr,
96 struct sk_buff **buf);
97 static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tolerance);
98 static int link_send_sections_long(struct tipc_port *sender,
99 struct iovec const *msg_sect,
100 unsigned int len, u32 destnode);
101 static void link_state_event(struct tipc_link *l_ptr, u32 event);
102 static void link_reset_statistics(struct tipc_link *l_ptr);
103 static void link_print(struct tipc_link *l_ptr, const char *str);
104 static void link_start(struct tipc_link *l_ptr);
105 static int link_send_long_buf(struct tipc_link *l_ptr, struct sk_buff *buf);
106 static void tipc_link_send_sync(struct tipc_link *l);
107 static void tipc_link_recv_sync(struct tipc_node *n, struct sk_buff *buf);
110 * Simple link routines
112 static unsigned int align(unsigned int i)
114 return (i + 3) & ~3u;
117 static void link_init_max_pkt(struct tipc_link *l_ptr)
121 max_pkt = (l_ptr->b_ptr->mtu & ~3);
122 if (max_pkt > MAX_MSG_SIZE)
123 max_pkt = MAX_MSG_SIZE;
125 l_ptr->max_pkt_target = max_pkt;
126 if (l_ptr->max_pkt_target < MAX_PKT_DEFAULT)
127 l_ptr->max_pkt = l_ptr->max_pkt_target;
129 l_ptr->max_pkt = MAX_PKT_DEFAULT;
131 l_ptr->max_pkt_probes = 0;
134 static u32 link_next_sent(struct tipc_link *l_ptr)
137 return buf_seqno(l_ptr->next_out);
138 return mod(l_ptr->next_out_no);
141 static u32 link_last_sent(struct tipc_link *l_ptr)
143 return mod(link_next_sent(l_ptr) - 1);
147 * Simple non-static link routines (i.e. referenced outside this file)
149 int tipc_link_is_up(struct tipc_link *l_ptr)
153 return link_working_working(l_ptr) || link_working_unknown(l_ptr);
156 int tipc_link_is_active(struct tipc_link *l_ptr)
158 return (l_ptr->owner->active_links[0] == l_ptr) ||
159 (l_ptr->owner->active_links[1] == l_ptr);
163 * link_name_validate - validate & (optionally) deconstruct tipc_link name
164 * @name: ptr to link name string
165 * @name_parts: ptr to area for link name components (or NULL if not needed)
167 * Returns 1 if link name is valid, otherwise 0.
169 static int link_name_validate(const char *name,
170 struct tipc_link_name *name_parts)
172 char name_copy[TIPC_MAX_LINK_NAME];
178 u32 z_local, c_local, n_local;
179 u32 z_peer, c_peer, n_peer;
183 /* copy link name & ensure length is OK */
184 name_copy[TIPC_MAX_LINK_NAME - 1] = 0;
185 /* need above in case non-Posix strncpy() doesn't pad with nulls */
186 strncpy(name_copy, name, TIPC_MAX_LINK_NAME);
187 if (name_copy[TIPC_MAX_LINK_NAME - 1] != 0)
190 /* ensure all component parts of link name are present */
191 addr_local = name_copy;
192 if_local = strchr(addr_local, ':');
193 if (if_local == NULL)
196 addr_peer = strchr(if_local, '-');
197 if (addr_peer == NULL)
200 if_local_len = addr_peer - if_local;
201 if_peer = strchr(addr_peer, ':');
205 if_peer_len = strlen(if_peer) + 1;
207 /* validate component parts of link name */
208 if ((sscanf(addr_local, "%u.%u.%u%c",
209 &z_local, &c_local, &n_local, &dummy) != 3) ||
210 (sscanf(addr_peer, "%u.%u.%u%c",
211 &z_peer, &c_peer, &n_peer, &dummy) != 3) ||
212 (z_local > 255) || (c_local > 4095) || (n_local > 4095) ||
213 (z_peer > 255) || (c_peer > 4095) || (n_peer > 4095) ||
214 (if_local_len <= 1) || (if_local_len > TIPC_MAX_IF_NAME) ||
215 (if_peer_len <= 1) || (if_peer_len > TIPC_MAX_IF_NAME))
218 /* return link name components, if necessary */
220 name_parts->addr_local = tipc_addr(z_local, c_local, n_local);
221 strcpy(name_parts->if_local, if_local);
222 name_parts->addr_peer = tipc_addr(z_peer, c_peer, n_peer);
223 strcpy(name_parts->if_peer, if_peer);
229 * link_timeout - handle expiration of link timer
230 * @l_ptr: pointer to link
232 * This routine must not grab "tipc_net_lock" to avoid a potential deadlock conflict
233 * with tipc_link_delete(). (There is no risk that the node will be deleted by
234 * another thread because tipc_link_delete() always cancels the link timer before
235 * tipc_node_delete() is called.)
237 static void link_timeout(struct tipc_link *l_ptr)
239 tipc_node_lock(l_ptr->owner);
241 /* update counters used in statistical profiling of send traffic */
242 l_ptr->stats.accu_queue_sz += l_ptr->out_queue_size;
243 l_ptr->stats.queue_sz_counts++;
245 if (l_ptr->first_out) {
246 struct tipc_msg *msg = buf_msg(l_ptr->first_out);
247 u32 length = msg_size(msg);
249 if ((msg_user(msg) == MSG_FRAGMENTER) &&
250 (msg_type(msg) == FIRST_FRAGMENT)) {
251 length = msg_size(msg_get_wrapped(msg));
254 l_ptr->stats.msg_lengths_total += length;
255 l_ptr->stats.msg_length_counts++;
257 l_ptr->stats.msg_length_profile[0]++;
258 else if (length <= 256)
259 l_ptr->stats.msg_length_profile[1]++;
260 else if (length <= 1024)
261 l_ptr->stats.msg_length_profile[2]++;
262 else if (length <= 4096)
263 l_ptr->stats.msg_length_profile[3]++;
264 else if (length <= 16384)
265 l_ptr->stats.msg_length_profile[4]++;
266 else if (length <= 32768)
267 l_ptr->stats.msg_length_profile[5]++;
269 l_ptr->stats.msg_length_profile[6]++;
273 /* do all other link processing performed on a periodic basis */
275 link_state_event(l_ptr, TIMEOUT_EVT);
278 tipc_link_push_queue(l_ptr);
280 tipc_node_unlock(l_ptr->owner);
283 static void link_set_timer(struct tipc_link *l_ptr, u32 time)
285 k_start_timer(&l_ptr->timer, time);
289 * tipc_link_create - create a new link
290 * @n_ptr: pointer to associated node
291 * @b_ptr: pointer to associated bearer
292 * @media_addr: media address to use when sending messages over link
294 * Returns pointer to link.
296 struct tipc_link *tipc_link_create(struct tipc_node *n_ptr,
297 struct tipc_bearer *b_ptr,
298 const struct tipc_media_addr *media_addr)
300 struct tipc_link *l_ptr;
301 struct tipc_msg *msg;
303 char addr_string[16];
304 u32 peer = n_ptr->addr;
306 if (n_ptr->link_cnt >= 2) {
307 tipc_addr_string_fill(addr_string, n_ptr->addr);
308 pr_err("Attempt to establish third link to %s\n", addr_string);
312 if (n_ptr->links[b_ptr->identity]) {
313 tipc_addr_string_fill(addr_string, n_ptr->addr);
314 pr_err("Attempt to establish second link on <%s> to %s\n",
315 b_ptr->name, addr_string);
319 l_ptr = kzalloc(sizeof(*l_ptr), GFP_ATOMIC);
321 pr_warn("Link creation failed, no memory\n");
326 if_name = strchr(b_ptr->name, ':') + 1;
327 sprintf(l_ptr->name, "%u.%u.%u:%s-%u.%u.%u:unknown",
328 tipc_zone(tipc_own_addr), tipc_cluster(tipc_own_addr),
329 tipc_node(tipc_own_addr),
331 tipc_zone(peer), tipc_cluster(peer), tipc_node(peer));
332 /* note: peer i/f name is updated by reset/activate message */
333 memcpy(&l_ptr->media_addr, media_addr, sizeof(*media_addr));
334 l_ptr->owner = n_ptr;
335 l_ptr->checkpoint = 1;
336 l_ptr->peer_session = INVALID_SESSION;
337 l_ptr->b_ptr = b_ptr;
338 link_set_supervision_props(l_ptr, b_ptr->tolerance);
339 l_ptr->state = RESET_UNKNOWN;
341 l_ptr->pmsg = (struct tipc_msg *)&l_ptr->proto_msg;
343 tipc_msg_init(msg, LINK_PROTOCOL, RESET_MSG, INT_H_SIZE, l_ptr->addr);
344 msg_set_size(msg, sizeof(l_ptr->proto_msg));
345 msg_set_session(msg, (tipc_random & 0xffff));
346 msg_set_bearer_id(msg, b_ptr->identity);
347 strcpy((char *)msg_data(msg), if_name);
349 l_ptr->priority = b_ptr->priority;
350 tipc_link_set_queue_limits(l_ptr, b_ptr->window);
352 link_init_max_pkt(l_ptr);
354 l_ptr->next_out_no = 1;
355 INIT_LIST_HEAD(&l_ptr->waiting_ports);
357 link_reset_statistics(l_ptr);
359 tipc_node_attach_link(n_ptr, l_ptr);
361 k_init_timer(&l_ptr->timer, (Handler)link_timeout, (unsigned long)l_ptr);
362 list_add_tail(&l_ptr->link_list, &b_ptr->links);
363 tipc_k_signal((Handler)link_start, (unsigned long)l_ptr);
369 * tipc_link_delete - delete a link
370 * @l_ptr: pointer to link
372 * Note: 'tipc_net_lock' is write_locked, bearer is locked.
373 * This routine must not grab the node lock until after link timer cancellation
374 * to avoid a potential deadlock situation.
376 void tipc_link_delete(struct tipc_link *l_ptr)
379 pr_err("Attempt to delete non-existent link\n");
383 k_cancel_timer(&l_ptr->timer);
385 tipc_node_lock(l_ptr->owner);
386 tipc_link_reset(l_ptr);
387 tipc_node_detach_link(l_ptr->owner, l_ptr);
388 tipc_link_stop(l_ptr);
389 list_del_init(&l_ptr->link_list);
390 tipc_node_unlock(l_ptr->owner);
391 k_term_timer(&l_ptr->timer);
395 static void link_start(struct tipc_link *l_ptr)
397 tipc_node_lock(l_ptr->owner);
398 link_state_event(l_ptr, STARTING_EVT);
399 tipc_node_unlock(l_ptr->owner);
403 * link_schedule_port - schedule port for deferred sending
404 * @l_ptr: pointer to link
405 * @origport: reference to sending port
406 * @sz: amount of data to be sent
408 * Schedules port for renewed sending of messages after link congestion
411 static int link_schedule_port(struct tipc_link *l_ptr, u32 origport, u32 sz)
413 struct tipc_port *p_ptr;
415 spin_lock_bh(&tipc_port_list_lock);
416 p_ptr = tipc_port_lock(origport);
420 if (!list_empty(&p_ptr->wait_list))
422 p_ptr->congested = 1;
423 p_ptr->waiting_pkts = 1 + ((sz - 1) / l_ptr->max_pkt);
424 list_add_tail(&p_ptr->wait_list, &l_ptr->waiting_ports);
425 l_ptr->stats.link_congs++;
427 tipc_port_unlock(p_ptr);
429 spin_unlock_bh(&tipc_port_list_lock);
433 void tipc_link_wakeup_ports(struct tipc_link *l_ptr, int all)
435 struct tipc_port *p_ptr;
436 struct tipc_port *temp_p_ptr;
437 int win = l_ptr->queue_limit[0] - l_ptr->out_queue_size;
443 if (!spin_trylock_bh(&tipc_port_list_lock))
445 if (link_congested(l_ptr))
447 list_for_each_entry_safe(p_ptr, temp_p_ptr, &l_ptr->waiting_ports,
451 list_del_init(&p_ptr->wait_list);
452 spin_lock_bh(p_ptr->lock);
453 p_ptr->congested = 0;
454 p_ptr->wakeup(p_ptr);
455 win -= p_ptr->waiting_pkts;
456 spin_unlock_bh(p_ptr->lock);
460 spin_unlock_bh(&tipc_port_list_lock);
464 * link_release_outqueue - purge link's outbound message queue
465 * @l_ptr: pointer to link
467 static void link_release_outqueue(struct tipc_link *l_ptr)
469 struct sk_buff *buf = l_ptr->first_out;
470 struct sk_buff *next;
477 l_ptr->first_out = NULL;
478 l_ptr->out_queue_size = 0;
482 * tipc_link_reset_fragments - purge link's inbound message fragments queue
483 * @l_ptr: pointer to link
485 void tipc_link_reset_fragments(struct tipc_link *l_ptr)
487 struct sk_buff *buf = l_ptr->defragm_buf;
488 struct sk_buff *next;
495 l_ptr->defragm_buf = NULL;
499 * tipc_link_stop - purge all inbound and outbound messages associated with link
500 * @l_ptr: pointer to link
502 void tipc_link_stop(struct tipc_link *l_ptr)
505 struct sk_buff *next;
507 buf = l_ptr->oldest_deferred_in;
514 buf = l_ptr->first_out;
521 tipc_link_reset_fragments(l_ptr);
523 kfree_skb(l_ptr->proto_msg_queue);
524 l_ptr->proto_msg_queue = NULL;
527 void tipc_link_reset(struct tipc_link *l_ptr)
530 u32 prev_state = l_ptr->state;
531 u32 checkpoint = l_ptr->next_in_no;
532 int was_active_link = tipc_link_is_active(l_ptr);
534 msg_set_session(l_ptr->pmsg, ((msg_session(l_ptr->pmsg) + 1) & 0xffff));
536 /* Link is down, accept any session */
537 l_ptr->peer_session = INVALID_SESSION;
539 /* Prepare for max packet size negotiation */
540 link_init_max_pkt(l_ptr);
542 l_ptr->state = RESET_UNKNOWN;
544 if ((prev_state == RESET_UNKNOWN) || (prev_state == RESET_RESET))
547 tipc_node_link_down(l_ptr->owner, l_ptr);
548 tipc_bearer_remove_dest(l_ptr->b_ptr, l_ptr->addr);
550 if (was_active_link && tipc_node_active_links(l_ptr->owner) &&
551 l_ptr->owner->permit_changeover) {
552 l_ptr->reset_checkpoint = checkpoint;
553 l_ptr->exp_msg_count = START_CHANGEOVER;
556 /* Clean up all queues: */
557 link_release_outqueue(l_ptr);
558 kfree_skb(l_ptr->proto_msg_queue);
559 l_ptr->proto_msg_queue = NULL;
560 buf = l_ptr->oldest_deferred_in;
562 struct sk_buff *next = buf->next;
566 if (!list_empty(&l_ptr->waiting_ports))
567 tipc_link_wakeup_ports(l_ptr, 1);
569 l_ptr->retransm_queue_head = 0;
570 l_ptr->retransm_queue_size = 0;
571 l_ptr->last_out = NULL;
572 l_ptr->first_out = NULL;
573 l_ptr->next_out = NULL;
574 l_ptr->unacked_window = 0;
575 l_ptr->checkpoint = 1;
576 l_ptr->next_out_no = 1;
577 l_ptr->deferred_inqueue_sz = 0;
578 l_ptr->oldest_deferred_in = NULL;
579 l_ptr->newest_deferred_in = NULL;
580 l_ptr->fsm_msg_cnt = 0;
581 l_ptr->stale_count = 0;
582 link_reset_statistics(l_ptr);
586 static void link_activate(struct tipc_link *l_ptr)
588 l_ptr->next_in_no = l_ptr->stats.recv_info = 1;
589 tipc_node_link_up(l_ptr->owner, l_ptr);
590 tipc_bearer_add_dest(l_ptr->b_ptr, l_ptr->addr);
594 * link_state_event - link finite state machine
595 * @l_ptr: pointer to link
596 * @event: state machine event to process
598 static void link_state_event(struct tipc_link *l_ptr, unsigned int event)
600 struct tipc_link *other;
601 u32 cont_intv = l_ptr->continuity_interval;
603 if (!l_ptr->started && (event != STARTING_EVT))
604 return; /* Not yet. */
606 if (link_blocked(l_ptr)) {
607 if (event == TIMEOUT_EVT)
608 link_set_timer(l_ptr, cont_intv);
609 return; /* Changeover going on */
612 switch (l_ptr->state) {
613 case WORKING_WORKING:
615 case TRAFFIC_MSG_EVT:
619 if (l_ptr->next_in_no != l_ptr->checkpoint) {
620 l_ptr->checkpoint = l_ptr->next_in_no;
621 if (tipc_bclink_acks_missing(l_ptr->owner)) {
622 tipc_link_send_proto_msg(l_ptr, STATE_MSG,
624 l_ptr->fsm_msg_cnt++;
625 } else if (l_ptr->max_pkt < l_ptr->max_pkt_target) {
626 tipc_link_send_proto_msg(l_ptr, STATE_MSG,
628 l_ptr->fsm_msg_cnt++;
630 link_set_timer(l_ptr, cont_intv);
633 l_ptr->state = WORKING_UNKNOWN;
634 l_ptr->fsm_msg_cnt = 0;
635 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
636 l_ptr->fsm_msg_cnt++;
637 link_set_timer(l_ptr, cont_intv / 4);
640 pr_info("%s<%s>, requested by peer\n", link_rst_msg,
642 tipc_link_reset(l_ptr);
643 l_ptr->state = RESET_RESET;
644 l_ptr->fsm_msg_cnt = 0;
645 tipc_link_send_proto_msg(l_ptr, ACTIVATE_MSG, 0, 0, 0, 0, 0);
646 l_ptr->fsm_msg_cnt++;
647 link_set_timer(l_ptr, cont_intv);
650 pr_err("%s%u in WW state\n", link_unk_evt, event);
653 case WORKING_UNKNOWN:
655 case TRAFFIC_MSG_EVT:
657 l_ptr->state = WORKING_WORKING;
658 l_ptr->fsm_msg_cnt = 0;
659 link_set_timer(l_ptr, cont_intv);
662 pr_info("%s<%s>, requested by peer while probing\n",
663 link_rst_msg, l_ptr->name);
664 tipc_link_reset(l_ptr);
665 l_ptr->state = RESET_RESET;
666 l_ptr->fsm_msg_cnt = 0;
667 tipc_link_send_proto_msg(l_ptr, ACTIVATE_MSG, 0, 0, 0, 0, 0);
668 l_ptr->fsm_msg_cnt++;
669 link_set_timer(l_ptr, cont_intv);
672 if (l_ptr->next_in_no != l_ptr->checkpoint) {
673 l_ptr->state = WORKING_WORKING;
674 l_ptr->fsm_msg_cnt = 0;
675 l_ptr->checkpoint = l_ptr->next_in_no;
676 if (tipc_bclink_acks_missing(l_ptr->owner)) {
677 tipc_link_send_proto_msg(l_ptr, STATE_MSG,
679 l_ptr->fsm_msg_cnt++;
681 link_set_timer(l_ptr, cont_intv);
682 } else if (l_ptr->fsm_msg_cnt < l_ptr->abort_limit) {
683 tipc_link_send_proto_msg(l_ptr, STATE_MSG,
685 l_ptr->fsm_msg_cnt++;
686 link_set_timer(l_ptr, cont_intv / 4);
687 } else { /* Link has failed */
688 pr_warn("%s<%s>, peer not responding\n",
689 link_rst_msg, l_ptr->name);
690 tipc_link_reset(l_ptr);
691 l_ptr->state = RESET_UNKNOWN;
692 l_ptr->fsm_msg_cnt = 0;
693 tipc_link_send_proto_msg(l_ptr, RESET_MSG,
695 l_ptr->fsm_msg_cnt++;
696 link_set_timer(l_ptr, cont_intv);
700 pr_err("%s%u in WU state\n", link_unk_evt, event);
705 case TRAFFIC_MSG_EVT:
708 other = l_ptr->owner->active_links[0];
709 if (other && link_working_unknown(other))
711 l_ptr->state = WORKING_WORKING;
712 l_ptr->fsm_msg_cnt = 0;
713 link_activate(l_ptr);
714 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
715 l_ptr->fsm_msg_cnt++;
716 if (l_ptr->owner->working_links == 1)
717 tipc_link_send_sync(l_ptr);
718 link_set_timer(l_ptr, cont_intv);
721 l_ptr->state = RESET_RESET;
722 l_ptr->fsm_msg_cnt = 0;
723 tipc_link_send_proto_msg(l_ptr, ACTIVATE_MSG, 1, 0, 0, 0, 0);
724 l_ptr->fsm_msg_cnt++;
725 link_set_timer(l_ptr, cont_intv);
731 tipc_link_send_proto_msg(l_ptr, RESET_MSG, 0, 0, 0, 0, 0);
732 l_ptr->fsm_msg_cnt++;
733 link_set_timer(l_ptr, cont_intv);
736 pr_err("%s%u in RU state\n", link_unk_evt, event);
741 case TRAFFIC_MSG_EVT:
743 other = l_ptr->owner->active_links[0];
744 if (other && link_working_unknown(other))
746 l_ptr->state = WORKING_WORKING;
747 l_ptr->fsm_msg_cnt = 0;
748 link_activate(l_ptr);
749 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 1, 0, 0, 0, 0);
750 l_ptr->fsm_msg_cnt++;
751 if (l_ptr->owner->working_links == 1)
752 tipc_link_send_sync(l_ptr);
753 link_set_timer(l_ptr, cont_intv);
758 tipc_link_send_proto_msg(l_ptr, ACTIVATE_MSG, 0, 0, 0, 0, 0);
759 l_ptr->fsm_msg_cnt++;
760 link_set_timer(l_ptr, cont_intv);
763 pr_err("%s%u in RR state\n", link_unk_evt, event);
767 pr_err("Unknown link state %u/%u\n", l_ptr->state, event);
772 * link_bundle_buf(): Append contents of a buffer to
773 * the tail of an existing one.
775 static int link_bundle_buf(struct tipc_link *l_ptr, struct sk_buff *bundler,
778 struct tipc_msg *bundler_msg = buf_msg(bundler);
779 struct tipc_msg *msg = buf_msg(buf);
780 u32 size = msg_size(msg);
781 u32 bundle_size = msg_size(bundler_msg);
782 u32 to_pos = align(bundle_size);
783 u32 pad = to_pos - bundle_size;
785 if (msg_user(bundler_msg) != MSG_BUNDLER)
787 if (msg_type(bundler_msg) != OPEN_MSG)
789 if (skb_tailroom(bundler) < (pad + size))
791 if (l_ptr->max_pkt < (to_pos + size))
794 skb_put(bundler, pad + size);
795 skb_copy_to_linear_data_offset(bundler, to_pos, buf->data, size);
796 msg_set_size(bundler_msg, to_pos + size);
797 msg_set_msgcnt(bundler_msg, msg_msgcnt(bundler_msg) + 1);
799 l_ptr->stats.sent_bundled++;
803 static void link_add_to_outqueue(struct tipc_link *l_ptr,
805 struct tipc_msg *msg)
807 u32 ack = mod(l_ptr->next_in_no - 1);
808 u32 seqno = mod(l_ptr->next_out_no++);
810 msg_set_word(msg, 2, ((ack << 16) | seqno));
811 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
813 if (l_ptr->first_out) {
814 l_ptr->last_out->next = buf;
815 l_ptr->last_out = buf;
817 l_ptr->first_out = l_ptr->last_out = buf;
819 l_ptr->out_queue_size++;
820 if (l_ptr->out_queue_size > l_ptr->stats.max_queue_sz)
821 l_ptr->stats.max_queue_sz = l_ptr->out_queue_size;
824 static void link_add_chain_to_outqueue(struct tipc_link *l_ptr,
825 struct sk_buff *buf_chain,
829 struct tipc_msg *msg;
831 if (!l_ptr->next_out)
832 l_ptr->next_out = buf_chain;
835 buf_chain = buf_chain->next;
838 msg_set_long_msgno(msg, long_msgno);
839 link_add_to_outqueue(l_ptr, buf, msg);
844 * tipc_link_send_buf() is the 'full path' for messages, called from
845 * inside TIPC when the 'fast path' in tipc_send_buf
846 * has failed, and from link_send()
848 int tipc_link_send_buf(struct tipc_link *l_ptr, struct sk_buff *buf)
850 struct tipc_msg *msg = buf_msg(buf);
851 u32 size = msg_size(msg);
852 u32 dsz = msg_data_sz(msg);
853 u32 queue_size = l_ptr->out_queue_size;
854 u32 imp = tipc_msg_tot_importance(msg);
855 u32 queue_limit = l_ptr->queue_limit[imp];
856 u32 max_packet = l_ptr->max_pkt;
858 /* Match msg importance against queue limits: */
859 if (unlikely(queue_size >= queue_limit)) {
860 if (imp <= TIPC_CRITICAL_IMPORTANCE) {
861 link_schedule_port(l_ptr, msg_origport(msg), size);
866 if (imp > CONN_MANAGER) {
867 pr_warn("%s<%s>, send queue full", link_rst_msg,
869 tipc_link_reset(l_ptr);
874 /* Fragmentation needed ? */
875 if (size > max_packet)
876 return link_send_long_buf(l_ptr, buf);
878 /* Packet can be queued or sent. */
879 if (likely(!tipc_bearer_blocked(l_ptr->b_ptr) &&
880 !link_congested(l_ptr))) {
881 link_add_to_outqueue(l_ptr, buf, msg);
883 tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr);
884 l_ptr->unacked_window = 0;
887 /* Congestion: can message be bundled ? */
888 if ((msg_user(msg) != CHANGEOVER_PROTOCOL) &&
889 (msg_user(msg) != MSG_FRAGMENTER)) {
891 /* Try adding message to an existing bundle */
892 if (l_ptr->next_out &&
893 link_bundle_buf(l_ptr, l_ptr->last_out, buf))
896 /* Try creating a new bundle */
897 if (size <= max_packet * 2 / 3) {
898 struct sk_buff *bundler = tipc_buf_acquire(max_packet);
899 struct tipc_msg bundler_hdr;
902 tipc_msg_init(&bundler_hdr, MSG_BUNDLER, OPEN_MSG,
903 INT_H_SIZE, l_ptr->addr);
904 skb_copy_to_linear_data(bundler, &bundler_hdr,
906 skb_trim(bundler, INT_H_SIZE);
907 link_bundle_buf(l_ptr, bundler, buf);
910 l_ptr->stats.sent_bundles++;
914 if (!l_ptr->next_out)
915 l_ptr->next_out = buf;
916 link_add_to_outqueue(l_ptr, buf, msg);
921 * tipc_link_send(): same as tipc_link_send_buf(), but the link to use has
922 * not been selected yet, and the the owner node is not locked
923 * Called by TIPC internal users, e.g. the name distributor
925 int tipc_link_send(struct sk_buff *buf, u32 dest, u32 selector)
927 struct tipc_link *l_ptr;
928 struct tipc_node *n_ptr;
929 int res = -ELINKCONG;
931 read_lock_bh(&tipc_net_lock);
932 n_ptr = tipc_node_find(dest);
934 tipc_node_lock(n_ptr);
935 l_ptr = n_ptr->active_links[selector & 1];
937 res = tipc_link_send_buf(l_ptr, buf);
940 tipc_node_unlock(n_ptr);
944 read_unlock_bh(&tipc_net_lock);
949 * tipc_link_send_sync - synchronize broadcast link endpoints.
951 * Give a newly added peer node the sequence number where it should
952 * start receiving and acking broadcast packets.
954 * Called with node locked
956 static void tipc_link_send_sync(struct tipc_link *l)
959 struct tipc_msg *msg;
961 buf = tipc_buf_acquire(INT_H_SIZE);
966 tipc_msg_init(msg, BCAST_PROTOCOL, STATE_MSG, INT_H_SIZE, l->addr);
967 msg_set_last_bcast(msg, l->owner->bclink.acked);
968 link_add_chain_to_outqueue(l, buf, 0);
969 tipc_link_push_queue(l);
973 * tipc_link_recv_sync - synchronize broadcast link endpoints.
974 * Receive the sequence number where we should start receiving and
975 * acking broadcast packets from a newly added peer node, and open
976 * up for reception of such packets.
978 * Called with node locked
980 static void tipc_link_recv_sync(struct tipc_node *n, struct sk_buff *buf)
982 struct tipc_msg *msg = buf_msg(buf);
984 n->bclink.last_sent = n->bclink.last_in = msg_last_bcast(msg);
985 n->bclink.recv_permitted = true;
990 * tipc_link_send_names - send name table entries to new neighbor
992 * Send routine for bulk delivery of name table messages when contact
993 * with a new neighbor occurs. No link congestion checking is performed
994 * because name table messages *must* be delivered. The messages must be
995 * small enough not to require fragmentation.
996 * Called without any locks held.
998 void tipc_link_send_names(struct list_head *message_list, u32 dest)
1000 struct tipc_node *n_ptr;
1001 struct tipc_link *l_ptr;
1002 struct sk_buff *buf;
1003 struct sk_buff *temp_buf;
1005 if (list_empty(message_list))
1008 read_lock_bh(&tipc_net_lock);
1009 n_ptr = tipc_node_find(dest);
1011 tipc_node_lock(n_ptr);
1012 l_ptr = n_ptr->active_links[0];
1014 /* convert circular list to linear list */
1015 ((struct sk_buff *)message_list->prev)->next = NULL;
1016 link_add_chain_to_outqueue(l_ptr,
1017 (struct sk_buff *)message_list->next, 0);
1018 tipc_link_push_queue(l_ptr);
1019 INIT_LIST_HEAD(message_list);
1021 tipc_node_unlock(n_ptr);
1023 read_unlock_bh(&tipc_net_lock);
1025 /* discard the messages if they couldn't be sent */
1026 list_for_each_safe(buf, temp_buf, ((struct sk_buff *)message_list)) {
1027 list_del((struct list_head *)buf);
1033 * link_send_buf_fast: Entry for data messages where the
1034 * destination link is known and the header is complete,
1035 * inclusive total message length. Very time critical.
1036 * Link is locked. Returns user data length.
1038 static int link_send_buf_fast(struct tipc_link *l_ptr, struct sk_buff *buf,
1041 struct tipc_msg *msg = buf_msg(buf);
1042 int res = msg_data_sz(msg);
1044 if (likely(!link_congested(l_ptr))) {
1045 if (likely(msg_size(msg) <= l_ptr->max_pkt)) {
1046 if (likely(!tipc_bearer_blocked(l_ptr->b_ptr))) {
1047 link_add_to_outqueue(l_ptr, buf, msg);
1048 tipc_bearer_send(l_ptr->b_ptr, buf,
1049 &l_ptr->media_addr);
1050 l_ptr->unacked_window = 0;
1054 *used_max_pkt = l_ptr->max_pkt;
1056 return tipc_link_send_buf(l_ptr, buf); /* All other cases */
1060 * tipc_link_send_sections_fast: Entry for messages where the
1061 * destination processor is known and the header is complete,
1062 * except for total message length.
1063 * Returns user data length or errno.
1065 int tipc_link_send_sections_fast(struct tipc_port *sender,
1066 struct iovec const *msg_sect,
1067 unsigned int len, u32 destaddr)
1069 struct tipc_msg *hdr = &sender->phdr;
1070 struct tipc_link *l_ptr;
1071 struct sk_buff *buf;
1072 struct tipc_node *node;
1074 u32 selector = msg_origport(hdr) & 1;
1078 * Try building message using port's max_pkt hint.
1079 * (Must not hold any locks while building message.)
1081 res = tipc_msg_build(hdr, msg_sect, len, sender->max_pkt, &buf);
1082 /* Exit if build request was invalid */
1083 if (unlikely(res < 0))
1086 read_lock_bh(&tipc_net_lock);
1087 node = tipc_node_find(destaddr);
1089 tipc_node_lock(node);
1090 l_ptr = node->active_links[selector];
1091 if (likely(l_ptr)) {
1093 res = link_send_buf_fast(l_ptr, buf,
1096 tipc_node_unlock(node);
1097 read_unlock_bh(&tipc_net_lock);
1101 /* Exit if link (or bearer) is congested */
1102 if (link_congested(l_ptr) ||
1103 tipc_bearer_blocked(l_ptr->b_ptr)) {
1104 res = link_schedule_port(l_ptr,
1110 * Message size exceeds max_pkt hint; update hint,
1111 * then re-try fast path or fragment the message
1113 sender->max_pkt = l_ptr->max_pkt;
1114 tipc_node_unlock(node);
1115 read_unlock_bh(&tipc_net_lock);
1118 if ((msg_hdr_sz(hdr) + res) <= sender->max_pkt)
1121 return link_send_sections_long(sender, msg_sect, len,
1124 tipc_node_unlock(node);
1126 read_unlock_bh(&tipc_net_lock);
1128 /* Couldn't find a link to the destination node */
1130 return tipc_reject_msg(buf, TIPC_ERR_NO_NODE);
1132 return tipc_port_reject_sections(sender, hdr, msg_sect,
1133 len, TIPC_ERR_NO_NODE);
1138 * link_send_sections_long(): Entry for long messages where the
1139 * destination node is known and the header is complete,
1140 * inclusive total message length.
1141 * Link and bearer congestion status have been checked to be ok,
1142 * and are ignored if they change.
1144 * Note that fragments do not use the full link MTU so that they won't have
1145 * to undergo refragmentation if link changeover causes them to be sent
1146 * over another link with an additional tunnel header added as prefix.
1147 * (Refragmentation will still occur if the other link has a smaller MTU.)
1149 * Returns user data length or errno.
1151 static int link_send_sections_long(struct tipc_port *sender,
1152 struct iovec const *msg_sect,
1153 unsigned int len, u32 destaddr)
1155 struct tipc_link *l_ptr;
1156 struct tipc_node *node;
1157 struct tipc_msg *hdr = &sender->phdr;
1159 u32 max_pkt, fragm_sz, rest;
1160 struct tipc_msg fragm_hdr;
1161 struct sk_buff *buf, *buf_chain, *prev;
1162 u32 fragm_crs, fragm_rest, hsz, sect_rest;
1163 const unchar __user *sect_crs;
1170 max_pkt = sender->max_pkt - INT_H_SIZE;
1171 /* leave room for tunnel header in case of link changeover */
1172 fragm_sz = max_pkt - INT_H_SIZE;
1173 /* leave room for fragmentation header in each fragment */
1181 /* Prepare reusable fragment header */
1182 tipc_msg_init(&fragm_hdr, MSG_FRAGMENTER, FIRST_FRAGMENT,
1183 INT_H_SIZE, msg_destnode(hdr));
1184 msg_set_size(&fragm_hdr, max_pkt);
1185 msg_set_fragm_no(&fragm_hdr, 1);
1187 /* Prepare header of first fragment */
1188 buf_chain = buf = tipc_buf_acquire(max_pkt);
1192 skb_copy_to_linear_data(buf, &fragm_hdr, INT_H_SIZE);
1193 hsz = msg_hdr_sz(hdr);
1194 skb_copy_to_linear_data_offset(buf, INT_H_SIZE, hdr, hsz);
1196 /* Chop up message */
1197 fragm_crs = INT_H_SIZE + hsz;
1198 fragm_rest = fragm_sz - hsz;
1200 do { /* For all sections */
1204 sect_rest = msg_sect[++curr_sect].iov_len;
1205 sect_crs = msg_sect[curr_sect].iov_base;
1208 if (sect_rest < fragm_rest)
1213 if (copy_from_user(buf->data + fragm_crs, sect_crs, sz)) {
1216 for (; buf_chain; buf_chain = buf) {
1217 buf = buf_chain->next;
1218 kfree_skb(buf_chain);
1228 if (!fragm_rest && rest) {
1230 /* Initiate new fragment: */
1231 if (rest <= fragm_sz) {
1233 msg_set_type(&fragm_hdr, LAST_FRAGMENT);
1235 msg_set_type(&fragm_hdr, FRAGMENT);
1237 msg_set_size(&fragm_hdr, fragm_sz + INT_H_SIZE);
1238 msg_set_fragm_no(&fragm_hdr, ++fragm_no);
1240 buf = tipc_buf_acquire(fragm_sz + INT_H_SIZE);
1248 skb_copy_to_linear_data(buf, &fragm_hdr, INT_H_SIZE);
1249 fragm_crs = INT_H_SIZE;
1250 fragm_rest = fragm_sz;
1255 * Now we have a buffer chain. Select a link and check
1256 * that packet size is still OK
1258 node = tipc_node_find(destaddr);
1260 tipc_node_lock(node);
1261 l_ptr = node->active_links[sender->ref & 1];
1263 tipc_node_unlock(node);
1266 if (l_ptr->max_pkt < max_pkt) {
1267 sender->max_pkt = l_ptr->max_pkt;
1268 tipc_node_unlock(node);
1269 for (; buf_chain; buf_chain = buf) {
1270 buf = buf_chain->next;
1271 kfree_skb(buf_chain);
1277 for (; buf_chain; buf_chain = buf) {
1278 buf = buf_chain->next;
1279 kfree_skb(buf_chain);
1281 return tipc_port_reject_sections(sender, hdr, msg_sect,
1282 len, TIPC_ERR_NO_NODE);
1285 /* Append chain of fragments to send queue & send them */
1286 l_ptr->long_msg_seq_no++;
1287 link_add_chain_to_outqueue(l_ptr, buf_chain, l_ptr->long_msg_seq_no);
1288 l_ptr->stats.sent_fragments += fragm_no;
1289 l_ptr->stats.sent_fragmented++;
1290 tipc_link_push_queue(l_ptr);
1291 tipc_node_unlock(node);
1296 * tipc_link_push_packet: Push one unsent packet to the media
1298 u32 tipc_link_push_packet(struct tipc_link *l_ptr)
1300 struct sk_buff *buf = l_ptr->first_out;
1301 u32 r_q_size = l_ptr->retransm_queue_size;
1302 u32 r_q_head = l_ptr->retransm_queue_head;
1304 /* Step to position where retransmission failed, if any, */
1305 /* consider that buffers may have been released in meantime */
1306 if (r_q_size && buf) {
1307 u32 last = lesser(mod(r_q_head + r_q_size),
1308 link_last_sent(l_ptr));
1309 u32 first = buf_seqno(buf);
1311 while (buf && less(first, r_q_head)) {
1312 first = mod(first + 1);
1315 l_ptr->retransm_queue_head = r_q_head = first;
1316 l_ptr->retransm_queue_size = r_q_size = mod(last - first);
1319 /* Continue retransmission now, if there is anything: */
1320 if (r_q_size && buf) {
1321 msg_set_ack(buf_msg(buf), mod(l_ptr->next_in_no - 1));
1322 msg_set_bcast_ack(buf_msg(buf), l_ptr->owner->bclink.last_in);
1323 tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr);
1324 l_ptr->retransm_queue_head = mod(++r_q_head);
1325 l_ptr->retransm_queue_size = --r_q_size;
1326 l_ptr->stats.retransmitted++;
1330 /* Send deferred protocol message, if any: */
1331 buf = l_ptr->proto_msg_queue;
1333 msg_set_ack(buf_msg(buf), mod(l_ptr->next_in_no - 1));
1334 msg_set_bcast_ack(buf_msg(buf), l_ptr->owner->bclink.last_in);
1335 tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr);
1336 l_ptr->unacked_window = 0;
1338 l_ptr->proto_msg_queue = NULL;
1342 /* Send one deferred data message, if send window not full: */
1343 buf = l_ptr->next_out;
1345 struct tipc_msg *msg = buf_msg(buf);
1346 u32 next = msg_seqno(msg);
1347 u32 first = buf_seqno(l_ptr->first_out);
1349 if (mod(next - first) < l_ptr->queue_limit[0]) {
1350 msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
1351 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
1352 tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr);
1353 if (msg_user(msg) == MSG_BUNDLER)
1354 msg_set_type(msg, CLOSED_MSG);
1355 l_ptr->next_out = buf->next;
1363 * push_queue(): push out the unsent messages of a link where
1364 * congestion has abated. Node is locked
1366 void tipc_link_push_queue(struct tipc_link *l_ptr)
1370 if (tipc_bearer_blocked(l_ptr->b_ptr))
1374 res = tipc_link_push_packet(l_ptr);
1378 static void link_reset_all(unsigned long addr)
1380 struct tipc_node *n_ptr;
1381 char addr_string[16];
1384 read_lock_bh(&tipc_net_lock);
1385 n_ptr = tipc_node_find((u32)addr);
1387 read_unlock_bh(&tipc_net_lock);
1388 return; /* node no longer exists */
1391 tipc_node_lock(n_ptr);
1393 pr_warn("Resetting all links to %s\n",
1394 tipc_addr_string_fill(addr_string, n_ptr->addr));
1396 for (i = 0; i < MAX_BEARERS; i++) {
1397 if (n_ptr->links[i]) {
1398 link_print(n_ptr->links[i], "Resetting link\n");
1399 tipc_link_reset(n_ptr->links[i]);
1403 tipc_node_unlock(n_ptr);
1404 read_unlock_bh(&tipc_net_lock);
1407 static void link_retransmit_failure(struct tipc_link *l_ptr,
1408 struct sk_buff *buf)
1410 struct tipc_msg *msg = buf_msg(buf);
1412 pr_warn("Retransmission failure on link <%s>\n", l_ptr->name);
1415 /* Handle failure on standard link */
1416 link_print(l_ptr, "Resetting link\n");
1417 tipc_link_reset(l_ptr);
1420 /* Handle failure on broadcast link */
1421 struct tipc_node *n_ptr;
1422 char addr_string[16];
1424 pr_info("Msg seq number: %u, ", msg_seqno(msg));
1425 pr_cont("Outstanding acks: %lu\n",
1426 (unsigned long) TIPC_SKB_CB(buf)->handle);
1428 n_ptr = tipc_bclink_retransmit_to();
1429 tipc_node_lock(n_ptr);
1431 tipc_addr_string_fill(addr_string, n_ptr->addr);
1432 pr_info("Broadcast link info for %s\n", addr_string);
1433 pr_info("Reception permitted: %d, Acked: %u\n",
1434 n_ptr->bclink.recv_permitted,
1435 n_ptr->bclink.acked);
1436 pr_info("Last in: %u, Oos state: %u, Last sent: %u\n",
1437 n_ptr->bclink.last_in,
1438 n_ptr->bclink.oos_state,
1439 n_ptr->bclink.last_sent);
1441 tipc_k_signal((Handler)link_reset_all, (unsigned long)n_ptr->addr);
1443 tipc_node_unlock(n_ptr);
1445 l_ptr->stale_count = 0;
1449 void tipc_link_retransmit(struct tipc_link *l_ptr, struct sk_buff *buf,
1452 struct tipc_msg *msg;
1459 if (tipc_bearer_blocked(l_ptr->b_ptr)) {
1460 if (l_ptr->retransm_queue_size == 0) {
1461 l_ptr->retransm_queue_head = msg_seqno(msg);
1462 l_ptr->retransm_queue_size = retransmits;
1464 pr_err("Unexpected retransmit on link %s (qsize=%d)\n",
1465 l_ptr->name, l_ptr->retransm_queue_size);
1469 /* Detect repeated retransmit failures on unblocked bearer */
1470 if (l_ptr->last_retransmitted == msg_seqno(msg)) {
1471 if (++l_ptr->stale_count > 100) {
1472 link_retransmit_failure(l_ptr, buf);
1476 l_ptr->last_retransmitted = msg_seqno(msg);
1477 l_ptr->stale_count = 1;
1481 while (retransmits && (buf != l_ptr->next_out) && buf) {
1483 msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
1484 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
1485 tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr);
1488 l_ptr->stats.retransmitted++;
1491 l_ptr->retransm_queue_head = l_ptr->retransm_queue_size = 0;
1495 * link_insert_deferred_queue - insert deferred messages back into receive chain
1497 static struct sk_buff *link_insert_deferred_queue(struct tipc_link *l_ptr,
1498 struct sk_buff *buf)
1502 if (l_ptr->oldest_deferred_in == NULL)
1505 seq_no = buf_seqno(l_ptr->oldest_deferred_in);
1506 if (seq_no == mod(l_ptr->next_in_no)) {
1507 l_ptr->newest_deferred_in->next = buf;
1508 buf = l_ptr->oldest_deferred_in;
1509 l_ptr->oldest_deferred_in = NULL;
1510 l_ptr->deferred_inqueue_sz = 0;
1516 * link_recv_buf_validate - validate basic format of received message
1518 * This routine ensures a TIPC message has an acceptable header, and at least
1519 * as much data as the header indicates it should. The routine also ensures
1520 * that the entire message header is stored in the main fragment of the message
1521 * buffer, to simplify future access to message header fields.
1523 * Note: Having extra info present in the message header or data areas is OK.
1524 * TIPC will ignore the excess, under the assumption that it is optional info
1525 * introduced by a later release of the protocol.
1527 static int link_recv_buf_validate(struct sk_buff *buf)
1529 static u32 min_data_hdr_size[8] = {
1530 SHORT_H_SIZE, MCAST_H_SIZE, NAMED_H_SIZE, BASIC_H_SIZE,
1531 MAX_H_SIZE, MAX_H_SIZE, MAX_H_SIZE, MAX_H_SIZE
1534 struct tipc_msg *msg;
1540 if (unlikely(buf->len < MIN_H_SIZE))
1543 msg = skb_header_pointer(buf, 0, sizeof(tipc_hdr), tipc_hdr);
1547 if (unlikely(msg_version(msg) != TIPC_VERSION))
1550 size = msg_size(msg);
1551 hdr_size = msg_hdr_sz(msg);
1552 min_hdr_size = msg_isdata(msg) ?
1553 min_data_hdr_size[msg_type(msg)] : INT_H_SIZE;
1555 if (unlikely((hdr_size < min_hdr_size) ||
1556 (size < hdr_size) ||
1557 (buf->len < size) ||
1558 (size - hdr_size > TIPC_MAX_USER_MSG_SIZE)))
1561 return pskb_may_pull(buf, hdr_size);
1565 * tipc_recv_msg - process TIPC messages arriving from off-node
1566 * @head: pointer to message buffer chain
1567 * @tb_ptr: pointer to bearer message arrived on
1569 * Invoked with no locks held. Bearer pointer must point to a valid bearer
1570 * structure (i.e. cannot be NULL), but bearer can be inactive.
1572 void tipc_recv_msg(struct sk_buff *head, struct tipc_bearer *b_ptr)
1574 read_lock_bh(&tipc_net_lock);
1576 struct tipc_node *n_ptr;
1577 struct tipc_link *l_ptr;
1578 struct sk_buff *crs;
1579 struct sk_buff *buf = head;
1580 struct tipc_msg *msg;
1588 /* Ensure bearer is still enabled */
1589 if (unlikely(!b_ptr->active))
1592 /* Ensure message is well-formed */
1593 if (unlikely(!link_recv_buf_validate(buf)))
1596 /* Ensure message data is a single contiguous unit */
1597 if (unlikely(skb_linearize(buf)))
1600 /* Handle arrival of a non-unicast link message */
1603 if (unlikely(msg_non_seq(msg))) {
1604 if (msg_user(msg) == LINK_CONFIG)
1605 tipc_disc_recv_msg(buf, b_ptr);
1607 tipc_bclink_recv_pkt(buf);
1611 /* Discard unicast link messages destined for another node */
1612 if (unlikely(!msg_short(msg) &&
1613 (msg_destnode(msg) != tipc_own_addr)))
1616 /* Locate neighboring node that sent message */
1617 n_ptr = tipc_node_find(msg_prevnode(msg));
1618 if (unlikely(!n_ptr))
1620 tipc_node_lock(n_ptr);
1622 /* Locate unicast link endpoint that should handle message */
1623 l_ptr = n_ptr->links[b_ptr->identity];
1624 if (unlikely(!l_ptr)) {
1625 tipc_node_unlock(n_ptr);
1629 /* Verify that communication with node is currently allowed */
1630 if ((n_ptr->block_setup & WAIT_PEER_DOWN) &&
1631 msg_user(msg) == LINK_PROTOCOL &&
1632 (msg_type(msg) == RESET_MSG ||
1633 msg_type(msg) == ACTIVATE_MSG) &&
1634 !msg_redundant_link(msg))
1635 n_ptr->block_setup &= ~WAIT_PEER_DOWN;
1637 if (n_ptr->block_setup) {
1638 tipc_node_unlock(n_ptr);
1642 /* Validate message sequence number info */
1643 seq_no = msg_seqno(msg);
1644 ackd = msg_ack(msg);
1646 /* Release acked messages */
1647 if (n_ptr->bclink.recv_permitted)
1648 tipc_bclink_acknowledge(n_ptr, msg_bcast_ack(msg));
1650 crs = l_ptr->first_out;
1651 while ((crs != l_ptr->next_out) &&
1652 less_eq(buf_seqno(crs), ackd)) {
1653 struct sk_buff *next = crs->next;
1660 l_ptr->first_out = crs;
1661 l_ptr->out_queue_size -= released;
1664 /* Try sending any messages link endpoint has pending */
1665 if (unlikely(l_ptr->next_out))
1666 tipc_link_push_queue(l_ptr);
1667 if (unlikely(!list_empty(&l_ptr->waiting_ports)))
1668 tipc_link_wakeup_ports(l_ptr, 0);
1669 if (unlikely(++l_ptr->unacked_window >= TIPC_MIN_LINK_WIN)) {
1670 l_ptr->stats.sent_acks++;
1671 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
1674 /* Now (finally!) process the incoming message */
1676 if (likely(link_working_working(l_ptr))) {
1677 if (likely(seq_no == mod(l_ptr->next_in_no))) {
1678 l_ptr->next_in_no++;
1679 if (unlikely(l_ptr->oldest_deferred_in))
1680 head = link_insert_deferred_queue(l_ptr,
1683 if (likely(msg_isdata(msg))) {
1684 tipc_node_unlock(n_ptr);
1685 tipc_port_recv_msg(buf);
1688 switch (msg_user(msg)) {
1691 l_ptr->stats.recv_bundles++;
1692 l_ptr->stats.recv_bundled +=
1694 tipc_node_unlock(n_ptr);
1695 tipc_link_recv_bundle(buf);
1697 case NAME_DISTRIBUTOR:
1698 n_ptr->bclink.recv_permitted = true;
1699 tipc_node_unlock(n_ptr);
1700 tipc_named_recv(buf);
1702 case BCAST_PROTOCOL:
1703 tipc_link_recv_sync(n_ptr, buf);
1704 tipc_node_unlock(n_ptr);
1707 tipc_node_unlock(n_ptr);
1708 tipc_port_recv_proto_msg(buf);
1710 case MSG_FRAGMENTER:
1711 l_ptr->stats.recv_fragments++;
1712 ret = tipc_link_recv_fragment(
1713 &l_ptr->defragm_buf,
1716 l_ptr->stats.recv_fragmented++;
1720 l_ptr->next_in_no--;
1722 case CHANGEOVER_PROTOCOL:
1723 type = msg_type(msg);
1724 if (link_recv_changeover_msg(&l_ptr,
1727 seq_no = msg_seqno(msg);
1728 if (type == ORIGINAL_MSG)
1730 goto protocol_check;
1738 tipc_node_unlock(n_ptr);
1739 tipc_net_route_msg(buf);
1742 link_handle_out_of_seq_msg(l_ptr, buf);
1743 head = link_insert_deferred_queue(l_ptr, head);
1744 tipc_node_unlock(n_ptr);
1748 /* Link is not in state WORKING_WORKING */
1749 if (msg_user(msg) == LINK_PROTOCOL) {
1750 link_recv_proto_msg(l_ptr, buf);
1751 head = link_insert_deferred_queue(l_ptr, head);
1752 tipc_node_unlock(n_ptr);
1756 /* Traffic message. Conditionally activate link */
1757 link_state_event(l_ptr, TRAFFIC_MSG_EVT);
1759 if (link_working_working(l_ptr)) {
1760 /* Re-insert buffer in front of queue */
1763 tipc_node_unlock(n_ptr);
1766 tipc_node_unlock(n_ptr);
1770 read_unlock_bh(&tipc_net_lock);
1774 * tipc_link_defer_pkt - Add out-of-sequence message to deferred reception queue
1776 * Returns increase in queue length (i.e. 0 or 1)
1778 u32 tipc_link_defer_pkt(struct sk_buff **head, struct sk_buff **tail,
1779 struct sk_buff *buf)
1781 struct sk_buff *queue_buf;
1782 struct sk_buff **prev;
1783 u32 seq_no = buf_seqno(buf);
1788 if (*head == NULL) {
1789 *head = *tail = buf;
1794 if (less(buf_seqno(*tail), seq_no)) {
1795 (*tail)->next = buf;
1800 /* Locate insertion point in queue, then insert; discard if duplicate */
1804 u32 curr_seqno = buf_seqno(queue_buf);
1806 if (seq_no == curr_seqno) {
1811 if (less(seq_no, curr_seqno))
1814 prev = &queue_buf->next;
1815 queue_buf = queue_buf->next;
1818 buf->next = queue_buf;
1824 * link_handle_out_of_seq_msg - handle arrival of out-of-sequence packet
1826 static void link_handle_out_of_seq_msg(struct tipc_link *l_ptr,
1827 struct sk_buff *buf)
1829 u32 seq_no = buf_seqno(buf);
1831 if (likely(msg_user(buf_msg(buf)) == LINK_PROTOCOL)) {
1832 link_recv_proto_msg(l_ptr, buf);
1836 /* Record OOS packet arrival (force mismatch on next timeout) */
1837 l_ptr->checkpoint--;
1840 * Discard packet if a duplicate; otherwise add it to deferred queue
1841 * and notify peer of gap as per protocol specification
1843 if (less(seq_no, mod(l_ptr->next_in_no))) {
1844 l_ptr->stats.duplicates++;
1849 if (tipc_link_defer_pkt(&l_ptr->oldest_deferred_in,
1850 &l_ptr->newest_deferred_in, buf)) {
1851 l_ptr->deferred_inqueue_sz++;
1852 l_ptr->stats.deferred_recv++;
1853 if ((l_ptr->deferred_inqueue_sz % 16) == 1)
1854 tipc_link_send_proto_msg(l_ptr, STATE_MSG, 0, 0, 0, 0, 0);
1856 l_ptr->stats.duplicates++;
1860 * Send protocol message to the other endpoint.
1862 void tipc_link_send_proto_msg(struct tipc_link *l_ptr, u32 msg_typ,
1863 int probe_msg, u32 gap, u32 tolerance,
1864 u32 priority, u32 ack_mtu)
1866 struct sk_buff *buf = NULL;
1867 struct tipc_msg *msg = l_ptr->pmsg;
1868 u32 msg_size = sizeof(l_ptr->proto_msg);
1871 /* Discard any previous message that was deferred due to congestion */
1872 if (l_ptr->proto_msg_queue) {
1873 kfree_skb(l_ptr->proto_msg_queue);
1874 l_ptr->proto_msg_queue = NULL;
1877 if (link_blocked(l_ptr))
1880 /* Abort non-RESET send if communication with node is prohibited */
1881 if ((l_ptr->owner->block_setup) && (msg_typ != RESET_MSG))
1884 /* Create protocol message with "out-of-sequence" sequence number */
1885 msg_set_type(msg, msg_typ);
1886 msg_set_net_plane(msg, l_ptr->b_ptr->net_plane);
1887 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
1888 msg_set_last_bcast(msg, tipc_bclink_get_last_sent());
1890 if (msg_typ == STATE_MSG) {
1891 u32 next_sent = mod(l_ptr->next_out_no);
1893 if (!tipc_link_is_up(l_ptr))
1895 if (l_ptr->next_out)
1896 next_sent = buf_seqno(l_ptr->next_out);
1897 msg_set_next_sent(msg, next_sent);
1898 if (l_ptr->oldest_deferred_in) {
1899 u32 rec = buf_seqno(l_ptr->oldest_deferred_in);
1900 gap = mod(rec - mod(l_ptr->next_in_no));
1902 msg_set_seq_gap(msg, gap);
1904 l_ptr->stats.sent_nacks++;
1905 msg_set_link_tolerance(msg, tolerance);
1906 msg_set_linkprio(msg, priority);
1907 msg_set_max_pkt(msg, ack_mtu);
1908 msg_set_ack(msg, mod(l_ptr->next_in_no - 1));
1909 msg_set_probe(msg, probe_msg != 0);
1911 u32 mtu = l_ptr->max_pkt;
1913 if ((mtu < l_ptr->max_pkt_target) &&
1914 link_working_working(l_ptr) &&
1915 l_ptr->fsm_msg_cnt) {
1916 msg_size = (mtu + (l_ptr->max_pkt_target - mtu)/2 + 2) & ~3;
1917 if (l_ptr->max_pkt_probes == 10) {
1918 l_ptr->max_pkt_target = (msg_size - 4);
1919 l_ptr->max_pkt_probes = 0;
1920 msg_size = (mtu + (l_ptr->max_pkt_target - mtu)/2 + 2) & ~3;
1922 l_ptr->max_pkt_probes++;
1925 l_ptr->stats.sent_probes++;
1927 l_ptr->stats.sent_states++;
1928 } else { /* RESET_MSG or ACTIVATE_MSG */
1929 msg_set_ack(msg, mod(l_ptr->reset_checkpoint - 1));
1930 msg_set_seq_gap(msg, 0);
1931 msg_set_next_sent(msg, 1);
1932 msg_set_probe(msg, 0);
1933 msg_set_link_tolerance(msg, l_ptr->tolerance);
1934 msg_set_linkprio(msg, l_ptr->priority);
1935 msg_set_max_pkt(msg, l_ptr->max_pkt_target);
1938 r_flag = (l_ptr->owner->working_links > tipc_link_is_up(l_ptr));
1939 msg_set_redundant_link(msg, r_flag);
1940 msg_set_linkprio(msg, l_ptr->priority);
1941 msg_set_size(msg, msg_size);
1943 msg_set_seqno(msg, mod(l_ptr->next_out_no + (0xffff/2)));
1945 buf = tipc_buf_acquire(msg_size);
1949 skb_copy_to_linear_data(buf, msg, sizeof(l_ptr->proto_msg));
1950 buf->priority = TC_PRIO_CONTROL;
1952 /* Defer message if bearer is already blocked */
1953 if (tipc_bearer_blocked(l_ptr->b_ptr)) {
1954 l_ptr->proto_msg_queue = buf;
1958 tipc_bearer_send(l_ptr->b_ptr, buf, &l_ptr->media_addr);
1959 l_ptr->unacked_window = 0;
1964 * Receive protocol message :
1965 * Note that network plane id propagates through the network, and may
1966 * change at any time. The node with lowest address rules
1968 static void link_recv_proto_msg(struct tipc_link *l_ptr, struct sk_buff *buf)
1974 struct tipc_msg *msg = buf_msg(buf);
1976 if (link_blocked(l_ptr))
1979 /* record unnumbered packet arrival (force mismatch on next timeout) */
1980 l_ptr->checkpoint--;
1982 if (l_ptr->b_ptr->net_plane != msg_net_plane(msg))
1983 if (tipc_own_addr > msg_prevnode(msg))
1984 l_ptr->b_ptr->net_plane = msg_net_plane(msg);
1986 l_ptr->owner->permit_changeover = msg_redundant_link(msg);
1988 switch (msg_type(msg)) {
1991 if (!link_working_unknown(l_ptr) &&
1992 (l_ptr->peer_session != INVALID_SESSION)) {
1993 if (less_eq(msg_session(msg), l_ptr->peer_session))
1994 break; /* duplicate or old reset: ignore */
1997 if (!msg_redundant_link(msg) && (link_working_working(l_ptr) ||
1998 link_working_unknown(l_ptr))) {
2000 * peer has lost contact -- don't allow peer's links
2001 * to reactivate before we recognize loss & clean up
2003 l_ptr->owner->block_setup = WAIT_NODE_DOWN;
2006 link_state_event(l_ptr, RESET_MSG);
2010 /* Update link settings according other endpoint's values */
2011 strcpy((strrchr(l_ptr->name, ':') + 1), (char *)msg_data(msg));
2013 msg_tol = msg_link_tolerance(msg);
2014 if (msg_tol > l_ptr->tolerance)
2015 link_set_supervision_props(l_ptr, msg_tol);
2017 if (msg_linkprio(msg) > l_ptr->priority)
2018 l_ptr->priority = msg_linkprio(msg);
2020 max_pkt_info = msg_max_pkt(msg);
2022 if (max_pkt_info < l_ptr->max_pkt_target)
2023 l_ptr->max_pkt_target = max_pkt_info;
2024 if (l_ptr->max_pkt > l_ptr->max_pkt_target)
2025 l_ptr->max_pkt = l_ptr->max_pkt_target;
2027 l_ptr->max_pkt = l_ptr->max_pkt_target;
2030 /* Synchronize broadcast link info, if not done previously */
2031 if (!tipc_node_is_up(l_ptr->owner)) {
2032 l_ptr->owner->bclink.last_sent =
2033 l_ptr->owner->bclink.last_in =
2034 msg_last_bcast(msg);
2035 l_ptr->owner->bclink.oos_state = 0;
2038 l_ptr->peer_session = msg_session(msg);
2039 l_ptr->peer_bearer_id = msg_bearer_id(msg);
2041 if (msg_type(msg) == ACTIVATE_MSG)
2042 link_state_event(l_ptr, ACTIVATE_MSG);
2046 msg_tol = msg_link_tolerance(msg);
2048 link_set_supervision_props(l_ptr, msg_tol);
2050 if (msg_linkprio(msg) &&
2051 (msg_linkprio(msg) != l_ptr->priority)) {
2052 pr_warn("%s<%s>, priority change %u->%u\n",
2053 link_rst_msg, l_ptr->name, l_ptr->priority,
2055 l_ptr->priority = msg_linkprio(msg);
2056 tipc_link_reset(l_ptr); /* Enforce change to take effect */
2059 link_state_event(l_ptr, TRAFFIC_MSG_EVT);
2060 l_ptr->stats.recv_states++;
2061 if (link_reset_unknown(l_ptr))
2064 if (less_eq(mod(l_ptr->next_in_no), msg_next_sent(msg))) {
2065 rec_gap = mod(msg_next_sent(msg) -
2066 mod(l_ptr->next_in_no));
2069 max_pkt_ack = msg_max_pkt(msg);
2070 if (max_pkt_ack > l_ptr->max_pkt) {
2071 l_ptr->max_pkt = max_pkt_ack;
2072 l_ptr->max_pkt_probes = 0;
2076 if (msg_probe(msg)) {
2077 l_ptr->stats.recv_probes++;
2078 if (msg_size(msg) > sizeof(l_ptr->proto_msg))
2079 max_pkt_ack = msg_size(msg);
2082 /* Protocol message before retransmits, reduce loss risk */
2083 if (l_ptr->owner->bclink.recv_permitted)
2084 tipc_bclink_update_link_state(l_ptr->owner,
2085 msg_last_bcast(msg));
2087 if (rec_gap || (msg_probe(msg))) {
2088 tipc_link_send_proto_msg(l_ptr, STATE_MSG,
2089 0, rec_gap, 0, 0, max_pkt_ack);
2091 if (msg_seq_gap(msg)) {
2092 l_ptr->stats.recv_nacks++;
2093 tipc_link_retransmit(l_ptr, l_ptr->first_out,
2104 * tipc_link_tunnel(): Send one message via a link belonging to
2105 * another bearer. Owner node is locked.
2107 static void tipc_link_tunnel(struct tipc_link *l_ptr,
2108 struct tipc_msg *tunnel_hdr, struct tipc_msg *msg,
2111 struct tipc_link *tunnel;
2112 struct sk_buff *buf;
2113 u32 length = msg_size(msg);
2115 tunnel = l_ptr->owner->active_links[selector & 1];
2116 if (!tipc_link_is_up(tunnel)) {
2117 pr_warn("%stunnel link no longer available\n", link_co_err);
2120 msg_set_size(tunnel_hdr, length + INT_H_SIZE);
2121 buf = tipc_buf_acquire(length + INT_H_SIZE);
2123 pr_warn("%sunable to send tunnel msg\n", link_co_err);
2126 skb_copy_to_linear_data(buf, tunnel_hdr, INT_H_SIZE);
2127 skb_copy_to_linear_data_offset(buf, INT_H_SIZE, msg, length);
2128 tipc_link_send_buf(tunnel, buf);
2134 * changeover(): Send whole message queue via the remaining link
2135 * Owner node is locked.
2137 void tipc_link_changeover(struct tipc_link *l_ptr)
2139 u32 msgcount = l_ptr->out_queue_size;
2140 struct sk_buff *crs = l_ptr->first_out;
2141 struct tipc_link *tunnel = l_ptr->owner->active_links[0];
2142 struct tipc_msg tunnel_hdr;
2148 if (!l_ptr->owner->permit_changeover) {
2149 pr_warn("%speer did not permit changeover\n", link_co_err);
2153 tipc_msg_init(&tunnel_hdr, CHANGEOVER_PROTOCOL,
2154 ORIGINAL_MSG, INT_H_SIZE, l_ptr->addr);
2155 msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id);
2156 msg_set_msgcnt(&tunnel_hdr, msgcount);
2158 if (!l_ptr->first_out) {
2159 struct sk_buff *buf;
2161 buf = tipc_buf_acquire(INT_H_SIZE);
2163 skb_copy_to_linear_data(buf, &tunnel_hdr, INT_H_SIZE);
2164 msg_set_size(&tunnel_hdr, INT_H_SIZE);
2165 tipc_link_send_buf(tunnel, buf);
2167 pr_warn("%sunable to send changeover msg\n",
2173 split_bundles = (l_ptr->owner->active_links[0] !=
2174 l_ptr->owner->active_links[1]);
2177 struct tipc_msg *msg = buf_msg(crs);
2179 if ((msg_user(msg) == MSG_BUNDLER) && split_bundles) {
2180 struct tipc_msg *m = msg_get_wrapped(msg);
2181 unchar *pos = (unchar *)m;
2183 msgcount = msg_msgcnt(msg);
2184 while (msgcount--) {
2185 msg_set_seqno(m, msg_seqno(msg));
2186 tipc_link_tunnel(l_ptr, &tunnel_hdr, m,
2187 msg_link_selector(m));
2188 pos += align(msg_size(m));
2189 m = (struct tipc_msg *)pos;
2192 tipc_link_tunnel(l_ptr, &tunnel_hdr, msg,
2193 msg_link_selector(msg));
2199 void tipc_link_send_duplicate(struct tipc_link *l_ptr, struct tipc_link *tunnel)
2201 struct sk_buff *iter;
2202 struct tipc_msg tunnel_hdr;
2204 tipc_msg_init(&tunnel_hdr, CHANGEOVER_PROTOCOL,
2205 DUPLICATE_MSG, INT_H_SIZE, l_ptr->addr);
2206 msg_set_msgcnt(&tunnel_hdr, l_ptr->out_queue_size);
2207 msg_set_bearer_id(&tunnel_hdr, l_ptr->peer_bearer_id);
2208 iter = l_ptr->first_out;
2210 struct sk_buff *outbuf;
2211 struct tipc_msg *msg = buf_msg(iter);
2212 u32 length = msg_size(msg);
2214 if (msg_user(msg) == MSG_BUNDLER)
2215 msg_set_type(msg, CLOSED_MSG);
2216 msg_set_ack(msg, mod(l_ptr->next_in_no - 1)); /* Update */
2217 msg_set_bcast_ack(msg, l_ptr->owner->bclink.last_in);
2218 msg_set_size(&tunnel_hdr, length + INT_H_SIZE);
2219 outbuf = tipc_buf_acquire(length + INT_H_SIZE);
2220 if (outbuf == NULL) {
2221 pr_warn("%sunable to send duplicate msg\n",
2225 skb_copy_to_linear_data(outbuf, &tunnel_hdr, INT_H_SIZE);
2226 skb_copy_to_linear_data_offset(outbuf, INT_H_SIZE, iter->data,
2228 tipc_link_send_buf(tunnel, outbuf);
2229 if (!tipc_link_is_up(l_ptr))
2236 * buf_extract - extracts embedded TIPC message from another message
2237 * @skb: encapsulating message buffer
2238 * @from_pos: offset to extract from
2240 * Returns a new message buffer containing an embedded message. The
2241 * encapsulating message itself is left unchanged.
2243 static struct sk_buff *buf_extract(struct sk_buff *skb, u32 from_pos)
2245 struct tipc_msg *msg = (struct tipc_msg *)(skb->data + from_pos);
2246 u32 size = msg_size(msg);
2249 eb = tipc_buf_acquire(size);
2251 skb_copy_to_linear_data(eb, msg, size);
2256 * link_recv_changeover_msg(): Receive tunneled packet sent
2257 * via other link. Node is locked. Return extracted buffer.
2259 static int link_recv_changeover_msg(struct tipc_link **l_ptr,
2260 struct sk_buff **buf)
2262 struct sk_buff *tunnel_buf = *buf;
2263 struct tipc_link *dest_link;
2264 struct tipc_msg *msg;
2265 struct tipc_msg *tunnel_msg = buf_msg(tunnel_buf);
2266 u32 msg_typ = msg_type(tunnel_msg);
2267 u32 msg_count = msg_msgcnt(tunnel_msg);
2268 u32 bearer_id = msg_bearer_id(tunnel_msg);
2270 if (bearer_id >= MAX_BEARERS)
2272 dest_link = (*l_ptr)->owner->links[bearer_id];
2275 if (dest_link == *l_ptr) {
2276 pr_err("Unexpected changeover message on link <%s>\n",
2281 msg = msg_get_wrapped(tunnel_msg);
2283 if (msg_typ == DUPLICATE_MSG) {
2284 if (less(msg_seqno(msg), mod(dest_link->next_in_no)))
2286 *buf = buf_extract(tunnel_buf, INT_H_SIZE);
2288 pr_warn("%sduplicate msg dropped\n", link_co_err);
2291 kfree_skb(tunnel_buf);
2295 /* First original message ?: */
2296 if (tipc_link_is_up(dest_link)) {
2297 pr_info("%s<%s>, changeover initiated by peer\n", link_rst_msg,
2299 tipc_link_reset(dest_link);
2300 dest_link->exp_msg_count = msg_count;
2303 } else if (dest_link->exp_msg_count == START_CHANGEOVER) {
2304 dest_link->exp_msg_count = msg_count;
2309 /* Receive original message */
2310 if (dest_link->exp_msg_count == 0) {
2311 pr_warn("%sgot too many tunnelled messages\n", link_co_err);
2314 dest_link->exp_msg_count--;
2315 if (less(msg_seqno(msg), dest_link->reset_checkpoint)) {
2318 *buf = buf_extract(tunnel_buf, INT_H_SIZE);
2320 kfree_skb(tunnel_buf);
2323 pr_warn("%soriginal msg dropped\n", link_co_err);
2328 kfree_skb(tunnel_buf);
2333 * Bundler functionality:
2335 void tipc_link_recv_bundle(struct sk_buff *buf)
2337 u32 msgcount = msg_msgcnt(buf_msg(buf));
2338 u32 pos = INT_H_SIZE;
2339 struct sk_buff *obuf;
2341 while (msgcount--) {
2342 obuf = buf_extract(buf, pos);
2344 pr_warn("Link unable to unbundle message(s)\n");
2347 pos += align(msg_size(buf_msg(obuf)));
2348 tipc_net_route_msg(obuf);
2354 * Fragmentation/defragmentation:
2358 * link_send_long_buf: Entry for buffers needing fragmentation.
2359 * The buffer is complete, inclusive total message length.
2360 * Returns user data length.
2362 static int link_send_long_buf(struct tipc_link *l_ptr, struct sk_buff *buf)
2364 struct sk_buff *buf_chain = NULL;
2365 struct sk_buff *buf_chain_tail = (struct sk_buff *)&buf_chain;
2366 struct tipc_msg *inmsg = buf_msg(buf);
2367 struct tipc_msg fragm_hdr;
2368 u32 insize = msg_size(inmsg);
2369 u32 dsz = msg_data_sz(inmsg);
2370 unchar *crs = buf->data;
2372 u32 pack_sz = l_ptr->max_pkt;
2373 u32 fragm_sz = pack_sz - INT_H_SIZE;
2377 if (msg_short(inmsg))
2378 destaddr = l_ptr->addr;
2380 destaddr = msg_destnode(inmsg);
2382 /* Prepare reusable fragment header: */
2383 tipc_msg_init(&fragm_hdr, MSG_FRAGMENTER, FIRST_FRAGMENT,
2384 INT_H_SIZE, destaddr);
2386 /* Chop up message: */
2388 struct sk_buff *fragm;
2390 if (rest <= fragm_sz) {
2392 msg_set_type(&fragm_hdr, LAST_FRAGMENT);
2394 fragm = tipc_buf_acquire(fragm_sz + INT_H_SIZE);
2395 if (fragm == NULL) {
2399 buf_chain = buf_chain->next;
2404 msg_set_size(&fragm_hdr, fragm_sz + INT_H_SIZE);
2406 msg_set_fragm_no(&fragm_hdr, fragm_no);
2407 skb_copy_to_linear_data(fragm, &fragm_hdr, INT_H_SIZE);
2408 skb_copy_to_linear_data_offset(fragm, INT_H_SIZE, crs,
2410 buf_chain_tail->next = fragm;
2411 buf_chain_tail = fragm;
2415 msg_set_type(&fragm_hdr, FRAGMENT);
2419 /* Append chain of fragments to send queue & send them */
2420 l_ptr->long_msg_seq_no++;
2421 link_add_chain_to_outqueue(l_ptr, buf_chain, l_ptr->long_msg_seq_no);
2422 l_ptr->stats.sent_fragments += fragm_no;
2423 l_ptr->stats.sent_fragmented++;
2424 tipc_link_push_queue(l_ptr);
2430 * A pending message being re-assembled must store certain values
2431 * to handle subsequent fragments correctly. The following functions
2432 * help storing these values in unused, available fields in the
2433 * pending message. This makes dynamic memory allocation unnecessary.
2435 static void set_long_msg_seqno(struct sk_buff *buf, u32 seqno)
2437 msg_set_seqno(buf_msg(buf), seqno);
2440 static u32 get_fragm_size(struct sk_buff *buf)
2442 return msg_ack(buf_msg(buf));
2445 static void set_fragm_size(struct sk_buff *buf, u32 sz)
2447 msg_set_ack(buf_msg(buf), sz);
2450 static u32 get_expected_frags(struct sk_buff *buf)
2452 return msg_bcast_ack(buf_msg(buf));
2455 static void set_expected_frags(struct sk_buff *buf, u32 exp)
2457 msg_set_bcast_ack(buf_msg(buf), exp);
2461 * tipc_link_recv_fragment(): Called with node lock on. Returns
2462 * the reassembled buffer if message is complete.
2464 int tipc_link_recv_fragment(struct sk_buff **pending, struct sk_buff **fb,
2465 struct tipc_msg **m)
2467 struct sk_buff *prev = NULL;
2468 struct sk_buff *fbuf = *fb;
2469 struct tipc_msg *fragm = buf_msg(fbuf);
2470 struct sk_buff *pbuf = *pending;
2471 u32 long_msg_seq_no = msg_long_msgno(fragm);
2475 /* Is there an incomplete message waiting for this fragment? */
2476 while (pbuf && ((buf_seqno(pbuf) != long_msg_seq_no) ||
2477 (msg_orignode(fragm) != msg_orignode(buf_msg(pbuf))))) {
2482 if (!pbuf && (msg_type(fragm) == FIRST_FRAGMENT)) {
2483 struct tipc_msg *imsg = (struct tipc_msg *)msg_data(fragm);
2484 u32 msg_sz = msg_size(imsg);
2485 u32 fragm_sz = msg_data_sz(fragm);
2487 u32 max = TIPC_MAX_USER_MSG_SIZE + NAMED_H_SIZE;
2489 if (msg_type(imsg) == TIPC_MCAST_MSG)
2490 max = TIPC_MAX_USER_MSG_SIZE + MCAST_H_SIZE;
2491 if (fragm_sz == 0 || msg_size(imsg) > max) {
2495 exp_fragm_cnt = msg_sz / fragm_sz + !!(msg_sz % fragm_sz);
2496 pbuf = tipc_buf_acquire(msg_size(imsg));
2498 pbuf->next = *pending;
2500 skb_copy_to_linear_data(pbuf, imsg,
2501 msg_data_sz(fragm));
2502 /* Prepare buffer for subsequent fragments. */
2503 set_long_msg_seqno(pbuf, long_msg_seq_no);
2504 set_fragm_size(pbuf, fragm_sz);
2505 set_expected_frags(pbuf, exp_fragm_cnt - 1);
2507 pr_debug("Link unable to reassemble fragmented message\n");
2513 } else if (pbuf && (msg_type(fragm) != FIRST_FRAGMENT)) {
2514 u32 dsz = msg_data_sz(fragm);
2515 u32 fsz = get_fragm_size(pbuf);
2516 u32 crs = ((msg_fragm_no(fragm) - 1) * fsz);
2517 u32 exp_frags = get_expected_frags(pbuf) - 1;
2518 skb_copy_to_linear_data_offset(pbuf, crs,
2519 msg_data(fragm), dsz);
2522 /* Is message complete? */
2523 if (exp_frags == 0) {
2525 prev->next = pbuf->next;
2527 *pending = pbuf->next;
2528 msg_reset_reroute_cnt(buf_msg(pbuf));
2533 set_expected_frags(pbuf, exp_frags);
2540 static void link_set_supervision_props(struct tipc_link *l_ptr, u32 tolerance)
2542 if ((tolerance < TIPC_MIN_LINK_TOL) || (tolerance > TIPC_MAX_LINK_TOL))
2545 l_ptr->tolerance = tolerance;
2546 l_ptr->continuity_interval =
2547 ((tolerance / 4) > 500) ? 500 : tolerance / 4;
2548 l_ptr->abort_limit = tolerance / (l_ptr->continuity_interval / 4);
2551 void tipc_link_set_queue_limits(struct tipc_link *l_ptr, u32 window)
2553 /* Data messages from this node, inclusive FIRST_FRAGM */
2554 l_ptr->queue_limit[TIPC_LOW_IMPORTANCE] = window;
2555 l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE] = (window / 3) * 4;
2556 l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE] = (window / 3) * 5;
2557 l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE] = (window / 3) * 6;
2558 /* Transiting data messages,inclusive FIRST_FRAGM */
2559 l_ptr->queue_limit[TIPC_LOW_IMPORTANCE + 4] = 300;
2560 l_ptr->queue_limit[TIPC_MEDIUM_IMPORTANCE + 4] = 600;
2561 l_ptr->queue_limit[TIPC_HIGH_IMPORTANCE + 4] = 900;
2562 l_ptr->queue_limit[TIPC_CRITICAL_IMPORTANCE + 4] = 1200;
2563 l_ptr->queue_limit[CONN_MANAGER] = 1200;
2564 l_ptr->queue_limit[CHANGEOVER_PROTOCOL] = 2500;
2565 l_ptr->queue_limit[NAME_DISTRIBUTOR] = 3000;
2566 /* FRAGMENT and LAST_FRAGMENT packets */
2567 l_ptr->queue_limit[MSG_FRAGMENTER] = 4000;
2571 * link_find_link - locate link by name
2572 * @name: ptr to link name string
2573 * @node: ptr to area to be filled with ptr to associated node
2575 * Caller must hold 'tipc_net_lock' to ensure node and bearer are not deleted;
2576 * this also prevents link deletion.
2578 * Returns pointer to link (or 0 if invalid link name).
2580 static struct tipc_link *link_find_link(const char *name,
2581 struct tipc_node **node)
2583 struct tipc_link_name link_name_parts;
2584 struct tipc_bearer *b_ptr;
2585 struct tipc_link *l_ptr;
2587 if (!link_name_validate(name, &link_name_parts))
2590 b_ptr = tipc_bearer_find_interface(link_name_parts.if_local);
2594 *node = tipc_node_find(link_name_parts.addr_peer);
2598 l_ptr = (*node)->links[b_ptr->identity];
2599 if (!l_ptr || strcmp(l_ptr->name, name))
2606 * link_value_is_valid -- validate proposed link tolerance/priority/window
2608 * @cmd: value type (TIPC_CMD_SET_LINK_*)
2609 * @new_value: the new value
2611 * Returns 1 if value is within range, 0 if not.
2613 static int link_value_is_valid(u16 cmd, u32 new_value)
2616 case TIPC_CMD_SET_LINK_TOL:
2617 return (new_value >= TIPC_MIN_LINK_TOL) &&
2618 (new_value <= TIPC_MAX_LINK_TOL);
2619 case TIPC_CMD_SET_LINK_PRI:
2620 return (new_value <= TIPC_MAX_LINK_PRI);
2621 case TIPC_CMD_SET_LINK_WINDOW:
2622 return (new_value >= TIPC_MIN_LINK_WIN) &&
2623 (new_value <= TIPC_MAX_LINK_WIN);
2629 * link_cmd_set_value - change priority/tolerance/window for link/bearer/media
2630 * @name: ptr to link, bearer, or media name
2631 * @new_value: new value of link, bearer, or media setting
2632 * @cmd: which link, bearer, or media attribute to set (TIPC_CMD_SET_LINK_*)
2634 * Caller must hold 'tipc_net_lock' to ensure link/bearer/media is not deleted.
2636 * Returns 0 if value updated and negative value on error.
2638 static int link_cmd_set_value(const char *name, u32 new_value, u16 cmd)
2640 struct tipc_node *node;
2641 struct tipc_link *l_ptr;
2642 struct tipc_bearer *b_ptr;
2643 struct tipc_media *m_ptr;
2646 l_ptr = link_find_link(name, &node);
2649 * acquire node lock for tipc_link_send_proto_msg().
2650 * see "TIPC locking policy" in net.c.
2652 tipc_node_lock(node);
2654 case TIPC_CMD_SET_LINK_TOL:
2655 link_set_supervision_props(l_ptr, new_value);
2656 tipc_link_send_proto_msg(l_ptr,
2657 STATE_MSG, 0, 0, new_value, 0, 0);
2659 case TIPC_CMD_SET_LINK_PRI:
2660 l_ptr->priority = new_value;
2661 tipc_link_send_proto_msg(l_ptr,
2662 STATE_MSG, 0, 0, 0, new_value, 0);
2664 case TIPC_CMD_SET_LINK_WINDOW:
2665 tipc_link_set_queue_limits(l_ptr, new_value);
2671 tipc_node_unlock(node);
2675 b_ptr = tipc_bearer_find(name);
2678 case TIPC_CMD_SET_LINK_TOL:
2679 b_ptr->tolerance = new_value;
2681 case TIPC_CMD_SET_LINK_PRI:
2682 b_ptr->priority = new_value;
2684 case TIPC_CMD_SET_LINK_WINDOW:
2685 b_ptr->window = new_value;
2694 m_ptr = tipc_media_find(name);
2698 case TIPC_CMD_SET_LINK_TOL:
2699 m_ptr->tolerance = new_value;
2701 case TIPC_CMD_SET_LINK_PRI:
2702 m_ptr->priority = new_value;
2704 case TIPC_CMD_SET_LINK_WINDOW:
2705 m_ptr->window = new_value;
2714 struct sk_buff *tipc_link_cmd_config(const void *req_tlv_area, int req_tlv_space,
2717 struct tipc_link_config *args;
2721 if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_LINK_CONFIG))
2722 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR);
2724 args = (struct tipc_link_config *)TLV_DATA(req_tlv_area);
2725 new_value = ntohl(args->value);
2727 if (!link_value_is_valid(cmd, new_value))
2728 return tipc_cfg_reply_error_string(
2729 "cannot change, value invalid");
2731 if (!strcmp(args->name, tipc_bclink_name)) {
2732 if ((cmd == TIPC_CMD_SET_LINK_WINDOW) &&
2733 (tipc_bclink_set_queue_limits(new_value) == 0))
2734 return tipc_cfg_reply_none();
2735 return tipc_cfg_reply_error_string(TIPC_CFG_NOT_SUPPORTED
2736 " (cannot change setting on broadcast link)");
2739 read_lock_bh(&tipc_net_lock);
2740 res = link_cmd_set_value(args->name, new_value, cmd);
2741 read_unlock_bh(&tipc_net_lock);
2743 return tipc_cfg_reply_error_string("cannot change link setting");
2745 return tipc_cfg_reply_none();
2749 * link_reset_statistics - reset link statistics
2750 * @l_ptr: pointer to link
2752 static void link_reset_statistics(struct tipc_link *l_ptr)
2754 memset(&l_ptr->stats, 0, sizeof(l_ptr->stats));
2755 l_ptr->stats.sent_info = l_ptr->next_out_no;
2756 l_ptr->stats.recv_info = l_ptr->next_in_no;
2759 struct sk_buff *tipc_link_cmd_reset_stats(const void *req_tlv_area, int req_tlv_space)
2762 struct tipc_link *l_ptr;
2763 struct tipc_node *node;
2765 if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_LINK_NAME))
2766 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR);
2768 link_name = (char *)TLV_DATA(req_tlv_area);
2769 if (!strcmp(link_name, tipc_bclink_name)) {
2770 if (tipc_bclink_reset_stats())
2771 return tipc_cfg_reply_error_string("link not found");
2772 return tipc_cfg_reply_none();
2775 read_lock_bh(&tipc_net_lock);
2776 l_ptr = link_find_link(link_name, &node);
2778 read_unlock_bh(&tipc_net_lock);
2779 return tipc_cfg_reply_error_string("link not found");
2782 tipc_node_lock(node);
2783 link_reset_statistics(l_ptr);
2784 tipc_node_unlock(node);
2785 read_unlock_bh(&tipc_net_lock);
2786 return tipc_cfg_reply_none();
2790 * percent - convert count to a percentage of total (rounding up or down)
2792 static u32 percent(u32 count, u32 total)
2794 return (count * 100 + (total / 2)) / total;
2798 * tipc_link_stats - print link statistics
2800 * @buf: print buffer area
2801 * @buf_size: size of print buffer area
2803 * Returns length of print buffer data string (or 0 if error)
2805 static int tipc_link_stats(const char *name, char *buf, const u32 buf_size)
2807 struct tipc_link *l;
2808 struct tipc_stats *s;
2809 struct tipc_node *node;
2811 u32 profile_total = 0;
2814 if (!strcmp(name, tipc_bclink_name))
2815 return tipc_bclink_stats(buf, buf_size);
2817 read_lock_bh(&tipc_net_lock);
2818 l = link_find_link(name, &node);
2820 read_unlock_bh(&tipc_net_lock);
2823 tipc_node_lock(node);
2826 if (tipc_link_is_active(l))
2828 else if (tipc_link_is_up(l))
2833 ret = tipc_snprintf(buf, buf_size, "Link <%s>\n"
2834 " %s MTU:%u Priority:%u Tolerance:%u ms"
2835 " Window:%u packets\n",
2836 l->name, status, l->max_pkt, l->priority,
2837 l->tolerance, l->queue_limit[0]);
2839 ret += tipc_snprintf(buf + ret, buf_size - ret,
2840 " RX packets:%u fragments:%u/%u bundles:%u/%u\n",
2841 l->next_in_no - s->recv_info, s->recv_fragments,
2842 s->recv_fragmented, s->recv_bundles,
2845 ret += tipc_snprintf(buf + ret, buf_size - ret,
2846 " TX packets:%u fragments:%u/%u bundles:%u/%u\n",
2847 l->next_out_no - s->sent_info, s->sent_fragments,
2848 s->sent_fragmented, s->sent_bundles,
2851 profile_total = s->msg_length_counts;
2855 ret += tipc_snprintf(buf + ret, buf_size - ret,
2856 " TX profile sample:%u packets average:%u octets\n"
2857 " 0-64:%u%% -256:%u%% -1024:%u%% -4096:%u%% "
2858 "-16384:%u%% -32768:%u%% -66000:%u%%\n",
2859 s->msg_length_counts,
2860 s->msg_lengths_total / profile_total,
2861 percent(s->msg_length_profile[0], profile_total),
2862 percent(s->msg_length_profile[1], profile_total),
2863 percent(s->msg_length_profile[2], profile_total),
2864 percent(s->msg_length_profile[3], profile_total),
2865 percent(s->msg_length_profile[4], profile_total),
2866 percent(s->msg_length_profile[5], profile_total),
2867 percent(s->msg_length_profile[6], profile_total));
2869 ret += tipc_snprintf(buf + ret, buf_size - ret,
2870 " RX states:%u probes:%u naks:%u defs:%u"
2871 " dups:%u\n", s->recv_states, s->recv_probes,
2872 s->recv_nacks, s->deferred_recv, s->duplicates);
2874 ret += tipc_snprintf(buf + ret, buf_size - ret,
2875 " TX states:%u probes:%u naks:%u acks:%u"
2876 " dups:%u\n", s->sent_states, s->sent_probes,
2877 s->sent_nacks, s->sent_acks, s->retransmitted);
2879 ret += tipc_snprintf(buf + ret, buf_size - ret,
2880 " Congestion link:%u Send queue"
2881 " max:%u avg:%u\n", s->link_congs,
2882 s->max_queue_sz, s->queue_sz_counts ?
2883 (s->accu_queue_sz / s->queue_sz_counts) : 0);
2885 tipc_node_unlock(node);
2886 read_unlock_bh(&tipc_net_lock);
2890 struct sk_buff *tipc_link_cmd_show_stats(const void *req_tlv_area, int req_tlv_space)
2892 struct sk_buff *buf;
2893 struct tlv_desc *rep_tlv;
2898 if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_LINK_NAME))
2899 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR);
2901 buf = tipc_cfg_reply_alloc(TLV_SPACE(ULTRA_STRING_MAX_LEN));
2905 rep_tlv = (struct tlv_desc *)buf->data;
2906 pb = TLV_DATA(rep_tlv);
2907 pb_len = ULTRA_STRING_MAX_LEN;
2908 str_len = tipc_link_stats((char *)TLV_DATA(req_tlv_area),
2912 return tipc_cfg_reply_error_string("link not found");
2914 str_len += 1; /* for "\0" */
2915 skb_put(buf, TLV_SPACE(str_len));
2916 TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
2922 * tipc_link_get_max_pkt - get maximum packet size to use when sending to destination
2923 * @dest: network address of destination node
2924 * @selector: used to select from set of active links
2926 * If no active link can be found, uses default maximum packet size.
2928 u32 tipc_link_get_max_pkt(u32 dest, u32 selector)
2930 struct tipc_node *n_ptr;
2931 struct tipc_link *l_ptr;
2932 u32 res = MAX_PKT_DEFAULT;
2934 if (dest == tipc_own_addr)
2935 return MAX_MSG_SIZE;
2937 read_lock_bh(&tipc_net_lock);
2938 n_ptr = tipc_node_find(dest);
2940 tipc_node_lock(n_ptr);
2941 l_ptr = n_ptr->active_links[selector & 1];
2943 res = l_ptr->max_pkt;
2944 tipc_node_unlock(n_ptr);
2946 read_unlock_bh(&tipc_net_lock);
2950 static void link_print(struct tipc_link *l_ptr, const char *str)
2952 pr_info("%s Link %x<%s>:", str, l_ptr->addr, l_ptr->b_ptr->name);
2954 if (link_working_unknown(l_ptr))
2956 else if (link_reset_reset(l_ptr))
2958 else if (link_reset_unknown(l_ptr))
2960 else if (link_working_working(l_ptr))