2 * net/tipc/link.c: TIPC link code
4 * Copyright (c) 1996-2007, Ericsson AB
5 * Copyright (c) 2004-2007, 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.
44 #include "node_subscr.h"
45 #include "name_distr.h"
47 #include "name_table.h"
54 * Out-of-range value for link session numbers
57 #define INVALID_SESSION 0x10000
60 * Limit for deferred reception queue:
63 #define DEF_QUEUE_LIMIT 256u
69 #define STARTING_EVT 856384768 /* link processing trigger */
70 #define TRAFFIC_MSG_EVT 560815u /* rx'd ??? */
71 #define TIMEOUT_EVT 560817u /* link timer expired */
74 * The following two 'message types' is really just implementation
75 * data conveniently stored in the message header.
76 * They must not be considered part of the protocol
82 * State value stored in 'exp_msg_count'
85 #define START_CHANGEOVER 100000u
88 * struct link_name - deconstructed link name
89 * @addr_local: network address of node at this end
90 * @if_local: name of interface at this end
91 * @addr_peer: network address of node at far end
92 * @if_peer: name of interface at far end
97 char if_local
[TIPC_MAX_IF_NAME
];
99 char if_peer
[TIPC_MAX_IF_NAME
];
103 static void link_handle_out_of_seq_msg(struct link
*l_ptr
,
104 struct sk_buff
*buf
);
105 static void link_recv_proto_msg(struct link
*l_ptr
, struct sk_buff
*buf
);
106 static int link_recv_changeover_msg(struct link
**l_ptr
, struct sk_buff
**buf
);
107 static void link_set_supervision_props(struct link
*l_ptr
, u32 tolerance
);
108 static int link_send_sections_long(struct port
*sender
,
109 struct iovec
const *msg_sect
,
110 u32 num_sect
, u32 destnode
);
111 static void link_check_defragm_bufs(struct link
*l_ptr
);
112 static void link_state_event(struct link
*l_ptr
, u32 event
);
113 static void link_reset_statistics(struct link
*l_ptr
);
114 static void link_print(struct link
*l_ptr
, struct print_buf
*buf
,
118 * Debugging code used by link routines only
120 * When debugging link problems on a system that has multiple links,
121 * the standard TIPC debugging routines may not be useful since they
122 * allow the output from multiple links to be intermixed. For this reason
123 * routines of the form "dbg_link_XXX()" have been created that will capture
124 * debug info into a link's personal print buffer, which can then be dumped
125 * into the TIPC system log (TIPC_LOG) upon request.
127 * To enable per-link debugging, use LINK_LOG_BUF_SIZE to specify the size
128 * of the print buffer used by each link. If LINK_LOG_BUF_SIZE is set to 0,
129 * the dbg_link_XXX() routines simply send their output to the standard
130 * debug print buffer (DBG_OUTPUT), if it has been defined; this can be useful
131 * when there is only a single link in the system being debugged.
134 * - When enabled, LINK_LOG_BUF_SIZE should be set to at least TIPC_PB_MIN_SIZE
135 * - "l_ptr" must be valid when using dbg_link_XXX() macros
138 #define LINK_LOG_BUF_SIZE 0
140 #define dbg_link(fmt, arg...) \
142 if (LINK_LOG_BUF_SIZE) \
143 tipc_printf(&l_ptr->print_buf, fmt, ## arg); \
145 #define dbg_link_msg(msg, txt) \
147 if (LINK_LOG_BUF_SIZE) \
148 tipc_msg_dbg(&l_ptr->print_buf, msg, txt); \
150 #define dbg_link_state(txt) \
152 if (LINK_LOG_BUF_SIZE) \
153 link_print(l_ptr, &l_ptr->print_buf, txt); \
155 #define dbg_link_dump() do { \
156 if (LINK_LOG_BUF_SIZE) { \
157 tipc_printf(LOG, "\n\nDumping link <%s>:\n", l_ptr->name); \
158 tipc_printbuf_move(LOG, &l_ptr->print_buf); \
162 static void dbg_print_link(struct link
*l_ptr
, const char *str
)
164 if (DBG_OUTPUT
!= TIPC_NULL
)
165 link_print(l_ptr
, DBG_OUTPUT
, str
);
168 static void dbg_print_buf_chain(struct sk_buff
*root_buf
)
170 if (DBG_OUTPUT
!= TIPC_NULL
) {
171 struct sk_buff
*buf
= root_buf
;
174 msg_dbg(buf_msg(buf
), "In chain: ");
181 * Simple link routines
184 static unsigned int align(unsigned int i
)
186 return (i
+ 3) & ~3u;
189 static void link_init_max_pkt(struct link
*l_ptr
)
193 max_pkt
= (l_ptr
->b_ptr
->publ
.mtu
& ~3);
194 if (max_pkt
> MAX_MSG_SIZE
)
195 max_pkt
= MAX_MSG_SIZE
;
197 l_ptr
->max_pkt_target
= max_pkt
;
198 if (l_ptr
->max_pkt_target
< MAX_PKT_DEFAULT
)
199 l_ptr
->max_pkt
= l_ptr
->max_pkt_target
;
201 l_ptr
->max_pkt
= MAX_PKT_DEFAULT
;
203 l_ptr
->max_pkt_probes
= 0;
206 static u32
link_next_sent(struct link
*l_ptr
)
209 return msg_seqno(buf_msg(l_ptr
->next_out
));
210 return mod(l_ptr
->next_out_no
);
213 static u32
link_last_sent(struct link
*l_ptr
)
215 return mod(link_next_sent(l_ptr
) - 1);
219 * Simple non-static link routines (i.e. referenced outside this file)
222 int tipc_link_is_up(struct link
*l_ptr
)
226 return (link_working_working(l_ptr
) || link_working_unknown(l_ptr
));
229 int tipc_link_is_active(struct link
*l_ptr
)
231 return ((l_ptr
->owner
->active_links
[0] == l_ptr
) ||
232 (l_ptr
->owner
->active_links
[1] == l_ptr
));
236 * link_name_validate - validate & (optionally) deconstruct link name
237 * @name - ptr to link name string
238 * @name_parts - ptr to area for link name components (or NULL if not needed)
240 * Returns 1 if link name is valid, otherwise 0.
243 static int link_name_validate(const char *name
, struct link_name
*name_parts
)
245 char name_copy
[TIPC_MAX_LINK_NAME
];
251 u32 z_local
, c_local
, n_local
;
252 u32 z_peer
, c_peer
, n_peer
;
256 /* copy link name & ensure length is OK */
258 name_copy
[TIPC_MAX_LINK_NAME
- 1] = 0;
259 /* need above in case non-Posix strncpy() doesn't pad with nulls */
260 strncpy(name_copy
, name
, TIPC_MAX_LINK_NAME
);
261 if (name_copy
[TIPC_MAX_LINK_NAME
- 1] != 0)
264 /* ensure all component parts of link name are present */
266 addr_local
= name_copy
;
267 if ((if_local
= strchr(addr_local
, ':')) == NULL
)
270 if ((addr_peer
= strchr(if_local
, '-')) == NULL
)
273 if_local_len
= addr_peer
- if_local
;
274 if ((if_peer
= strchr(addr_peer
, ':')) == NULL
)
277 if_peer_len
= strlen(if_peer
) + 1;
279 /* validate component parts of link name */
281 if ((sscanf(addr_local
, "%u.%u.%u%c",
282 &z_local
, &c_local
, &n_local
, &dummy
) != 3) ||
283 (sscanf(addr_peer
, "%u.%u.%u%c",
284 &z_peer
, &c_peer
, &n_peer
, &dummy
) != 3) ||
285 (z_local
> 255) || (c_local
> 4095) || (n_local
> 4095) ||
286 (z_peer
> 255) || (c_peer
> 4095) || (n_peer
> 4095) ||
287 (if_local_len
<= 1) || (if_local_len
> TIPC_MAX_IF_NAME
) ||
288 (if_peer_len
<= 1) || (if_peer_len
> TIPC_MAX_IF_NAME
) ||
289 (strspn(if_local
, tipc_alphabet
) != (if_local_len
- 1)) ||
290 (strspn(if_peer
, tipc_alphabet
) != (if_peer_len
- 1)))
293 /* return link name components, if necessary */
296 name_parts
->addr_local
= tipc_addr(z_local
, c_local
, n_local
);
297 strcpy(name_parts
->if_local
, if_local
);
298 name_parts
->addr_peer
= tipc_addr(z_peer
, c_peer
, n_peer
);
299 strcpy(name_parts
->if_peer
, if_peer
);
305 * link_timeout - handle expiration of link timer
306 * @l_ptr: pointer to link
308 * This routine must not grab "tipc_net_lock" to avoid a potential deadlock conflict
309 * with tipc_link_delete(). (There is no risk that the node will be deleted by
310 * another thread because tipc_link_delete() always cancels the link timer before
311 * tipc_node_delete() is called.)
314 static void link_timeout(struct link
*l_ptr
)
316 tipc_node_lock(l_ptr
->owner
);
318 /* update counters used in statistical profiling of send traffic */
320 l_ptr
->stats
.accu_queue_sz
+= l_ptr
->out_queue_size
;
321 l_ptr
->stats
.queue_sz_counts
++;
323 if (l_ptr
->out_queue_size
> l_ptr
->stats
.max_queue_sz
)
324 l_ptr
->stats
.max_queue_sz
= l_ptr
->out_queue_size
;
326 if (l_ptr
->first_out
) {
327 struct tipc_msg
*msg
= buf_msg(l_ptr
->first_out
);
328 u32 length
= msg_size(msg
);
330 if ((msg_user(msg
) == MSG_FRAGMENTER
) &&
331 (msg_type(msg
) == FIRST_FRAGMENT
)) {
332 length
= msg_size(msg_get_wrapped(msg
));
335 l_ptr
->stats
.msg_lengths_total
+= length
;
336 l_ptr
->stats
.msg_length_counts
++;
338 l_ptr
->stats
.msg_length_profile
[0]++;
339 else if (length
<= 256)
340 l_ptr
->stats
.msg_length_profile
[1]++;
341 else if (length
<= 1024)
342 l_ptr
->stats
.msg_length_profile
[2]++;
343 else if (length
<= 4096)
344 l_ptr
->stats
.msg_length_profile
[3]++;
345 else if (length
<= 16384)
346 l_ptr
->stats
.msg_length_profile
[4]++;
347 else if (length
<= 32768)
348 l_ptr
->stats
.msg_length_profile
[5]++;
350 l_ptr
->stats
.msg_length_profile
[6]++;
354 /* do all other link processing performed on a periodic basis */
356 link_check_defragm_bufs(l_ptr
);
358 link_state_event(l_ptr
, TIMEOUT_EVT
);
361 tipc_link_push_queue(l_ptr
);
363 tipc_node_unlock(l_ptr
->owner
);
366 static void link_set_timer(struct link
*l_ptr
, u32 time
)
368 k_start_timer(&l_ptr
->timer
, time
);
372 * tipc_link_create - create a new link
373 * @b_ptr: pointer to associated bearer
374 * @peer: network address of node at other end of link
375 * @media_addr: media address to use when sending messages over link
377 * Returns pointer to link.
380 struct link
*tipc_link_create(struct bearer
*b_ptr
, const u32 peer
,
381 const struct tipc_media_addr
*media_addr
)
384 struct tipc_msg
*msg
;
387 l_ptr
= kzalloc(sizeof(*l_ptr
), GFP_ATOMIC
);
389 warn("Link creation failed, no memory\n");
393 if (LINK_LOG_BUF_SIZE
) {
394 char *pb
= kmalloc(LINK_LOG_BUF_SIZE
, GFP_ATOMIC
);
398 warn("Link creation failed, no memory for print buffer\n");
401 tipc_printbuf_init(&l_ptr
->print_buf
, pb
, LINK_LOG_BUF_SIZE
);
405 if_name
= strchr(b_ptr
->publ
.name
, ':') + 1;
406 sprintf(l_ptr
->name
, "%u.%u.%u:%s-%u.%u.%u:",
407 tipc_zone(tipc_own_addr
), tipc_cluster(tipc_own_addr
),
408 tipc_node(tipc_own_addr
),
410 tipc_zone(peer
), tipc_cluster(peer
), tipc_node(peer
));
411 /* note: peer i/f is appended to link name by reset/activate */
412 memcpy(&l_ptr
->media_addr
, media_addr
, sizeof(*media_addr
));
413 l_ptr
->checkpoint
= 1;
414 l_ptr
->b_ptr
= b_ptr
;
415 link_set_supervision_props(l_ptr
, b_ptr
->media
->tolerance
);
416 l_ptr
->state
= RESET_UNKNOWN
;
418 l_ptr
->pmsg
= (struct tipc_msg
*)&l_ptr
->proto_msg
;
420 tipc_msg_init(msg
, LINK_PROTOCOL
, RESET_MSG
, INT_H_SIZE
, l_ptr
->addr
);
421 msg_set_size(msg
, sizeof(l_ptr
->proto_msg
));
422 msg_set_session(msg
, (tipc_random
& 0xffff));
423 msg_set_bearer_id(msg
, b_ptr
->identity
);
424 strcpy((char *)msg_data(msg
), if_name
);
426 l_ptr
->priority
= b_ptr
->priority
;
427 tipc_link_set_queue_limits(l_ptr
, b_ptr
->media
->window
);
429 link_init_max_pkt(l_ptr
);
431 l_ptr
->next_out_no
= 1;
432 INIT_LIST_HEAD(&l_ptr
->waiting_ports
);
434 link_reset_statistics(l_ptr
);
436 l_ptr
->owner
= tipc_node_attach_link(l_ptr
);
438 if (LINK_LOG_BUF_SIZE
)
439 kfree(l_ptr
->print_buf
.buf
);
444 k_init_timer(&l_ptr
->timer
, (Handler
)link_timeout
, (unsigned long)l_ptr
);
445 list_add_tail(&l_ptr
->link_list
, &b_ptr
->links
);
446 tipc_k_signal((Handler
)tipc_link_start
, (unsigned long)l_ptr
);
448 dbg("tipc_link_create(): tolerance = %u,cont intv = %u, abort_limit = %u\n",
449 l_ptr
->tolerance
, l_ptr
->continuity_interval
, l_ptr
->abort_limit
);
455 * tipc_link_delete - delete a link
456 * @l_ptr: pointer to link
458 * Note: 'tipc_net_lock' is write_locked, bearer is locked.
459 * This routine must not grab the node lock until after link timer cancellation
460 * to avoid a potential deadlock situation.
463 void tipc_link_delete(struct link
*l_ptr
)
466 err("Attempt to delete non-existent link\n");
470 dbg("tipc_link_delete()\n");
472 k_cancel_timer(&l_ptr
->timer
);
474 tipc_node_lock(l_ptr
->owner
);
475 tipc_link_reset(l_ptr
);
476 tipc_node_detach_link(l_ptr
->owner
, l_ptr
);
477 tipc_link_stop(l_ptr
);
478 list_del_init(&l_ptr
->link_list
);
479 if (LINK_LOG_BUF_SIZE
)
480 kfree(l_ptr
->print_buf
.buf
);
481 tipc_node_unlock(l_ptr
->owner
);
482 k_term_timer(&l_ptr
->timer
);
486 void tipc_link_start(struct link
*l_ptr
)
488 dbg("tipc_link_start %x\n", l_ptr
);
489 link_state_event(l_ptr
, STARTING_EVT
);
493 * link_schedule_port - schedule port for deferred sending
494 * @l_ptr: pointer to link
495 * @origport: reference to sending port
496 * @sz: amount of data to be sent
498 * Schedules port for renewed sending of messages after link congestion
502 static int link_schedule_port(struct link
*l_ptr
, u32 origport
, u32 sz
)
506 spin_lock_bh(&tipc_port_list_lock
);
507 p_ptr
= tipc_port_lock(origport
);
511 if (!list_empty(&p_ptr
->wait_list
))
513 p_ptr
->publ
.congested
= 1;
514 p_ptr
->waiting_pkts
= 1 + ((sz
- 1) / l_ptr
->max_pkt
);
515 list_add_tail(&p_ptr
->wait_list
, &l_ptr
->waiting_ports
);
516 l_ptr
->stats
.link_congs
++;
518 tipc_port_unlock(p_ptr
);
520 spin_unlock_bh(&tipc_port_list_lock
);
524 void tipc_link_wakeup_ports(struct link
*l_ptr
, int all
)
527 struct port
*temp_p_ptr
;
528 int win
= l_ptr
->queue_limit
[0] - l_ptr
->out_queue_size
;
534 if (!spin_trylock_bh(&tipc_port_list_lock
))
536 if (link_congested(l_ptr
))
538 list_for_each_entry_safe(p_ptr
, temp_p_ptr
, &l_ptr
->waiting_ports
,
542 list_del_init(&p_ptr
->wait_list
);
543 spin_lock_bh(p_ptr
->publ
.lock
);
544 p_ptr
->publ
.congested
= 0;
545 p_ptr
->wakeup(&p_ptr
->publ
);
546 win
-= p_ptr
->waiting_pkts
;
547 spin_unlock_bh(p_ptr
->publ
.lock
);
551 spin_unlock_bh(&tipc_port_list_lock
);
555 * link_release_outqueue - purge link's outbound message queue
556 * @l_ptr: pointer to link
559 static void link_release_outqueue(struct link
*l_ptr
)
561 struct sk_buff
*buf
= l_ptr
->first_out
;
562 struct sk_buff
*next
;
569 l_ptr
->first_out
= NULL
;
570 l_ptr
->out_queue_size
= 0;
574 * tipc_link_reset_fragments - purge link's inbound message fragments queue
575 * @l_ptr: pointer to link
578 void tipc_link_reset_fragments(struct link
*l_ptr
)
580 struct sk_buff
*buf
= l_ptr
->defragm_buf
;
581 struct sk_buff
*next
;
588 l_ptr
->defragm_buf
= NULL
;
592 * tipc_link_stop - purge all inbound and outbound messages associated with link
593 * @l_ptr: pointer to link
596 void tipc_link_stop(struct link
*l_ptr
)
599 struct sk_buff
*next
;
601 buf
= l_ptr
->oldest_deferred_in
;
608 buf
= l_ptr
->first_out
;
615 tipc_link_reset_fragments(l_ptr
);
617 buf_discard(l_ptr
->proto_msg_queue
);
618 l_ptr
->proto_msg_queue
= NULL
;
622 #define link_send_event(fcn, l_ptr, up) do { } while (0)
624 void tipc_link_reset(struct link
*l_ptr
)
627 u32 prev_state
= l_ptr
->state
;
628 u32 checkpoint
= l_ptr
->next_in_no
;
629 int was_active_link
= tipc_link_is_active(l_ptr
);
631 msg_set_session(l_ptr
->pmsg
, ((msg_session(l_ptr
->pmsg
) + 1) & 0xffff));
633 /* Link is down, accept any session */
634 l_ptr
->peer_session
= INVALID_SESSION
;
636 /* Prepare for max packet size negotiation */
637 link_init_max_pkt(l_ptr
);
639 l_ptr
->state
= RESET_UNKNOWN
;
640 dbg_link_state("Resetting Link\n");
642 if ((prev_state
== RESET_UNKNOWN
) || (prev_state
== RESET_RESET
))
645 tipc_node_link_down(l_ptr
->owner
, l_ptr
);
646 tipc_bearer_remove_dest(l_ptr
->b_ptr
, l_ptr
->addr
);
647 if (was_active_link
&& tipc_node_has_active_links(l_ptr
->owner
) &&
648 l_ptr
->owner
->permit_changeover
) {
649 l_ptr
->reset_checkpoint
= checkpoint
;
650 l_ptr
->exp_msg_count
= START_CHANGEOVER
;
653 /* Clean up all queues: */
655 link_release_outqueue(l_ptr
);
656 buf_discard(l_ptr
->proto_msg_queue
);
657 l_ptr
->proto_msg_queue
= NULL
;
658 buf
= l_ptr
->oldest_deferred_in
;
660 struct sk_buff
*next
= buf
->next
;
664 if (!list_empty(&l_ptr
->waiting_ports
))
665 tipc_link_wakeup_ports(l_ptr
, 1);
667 l_ptr
->retransm_queue_head
= 0;
668 l_ptr
->retransm_queue_size
= 0;
669 l_ptr
->last_out
= NULL
;
670 l_ptr
->first_out
= NULL
;
671 l_ptr
->next_out
= NULL
;
672 l_ptr
->unacked_window
= 0;
673 l_ptr
->checkpoint
= 1;
674 l_ptr
->next_out_no
= 1;
675 l_ptr
->deferred_inqueue_sz
= 0;
676 l_ptr
->oldest_deferred_in
= NULL
;
677 l_ptr
->newest_deferred_in
= NULL
;
678 l_ptr
->fsm_msg_cnt
= 0;
679 l_ptr
->stale_count
= 0;
680 link_reset_statistics(l_ptr
);
682 link_send_event(tipc_cfg_link_event
, l_ptr
, 0);
683 if (!in_own_cluster(l_ptr
->addr
))
684 link_send_event(tipc_disc_link_event
, l_ptr
, 0);
688 static void link_activate(struct link
*l_ptr
)
690 l_ptr
->next_in_no
= l_ptr
->stats
.recv_info
= 1;
691 tipc_node_link_up(l_ptr
->owner
, l_ptr
);
692 tipc_bearer_add_dest(l_ptr
->b_ptr
, l_ptr
->addr
);
693 link_send_event(tipc_cfg_link_event
, l_ptr
, 1);
694 if (!in_own_cluster(l_ptr
->addr
))
695 link_send_event(tipc_disc_link_event
, l_ptr
, 1);
699 * link_state_event - link finite state machine
700 * @l_ptr: pointer to link
701 * @event: state machine event to process
704 static void link_state_event(struct link
*l_ptr
, unsigned event
)
707 u32 cont_intv
= l_ptr
->continuity_interval
;
709 if (!l_ptr
->started
&& (event
!= STARTING_EVT
))
710 return; /* Not yet. */
712 if (link_blocked(l_ptr
)) {
713 if (event
== TIMEOUT_EVT
) {
714 link_set_timer(l_ptr
, cont_intv
);
716 return; /* Changeover going on */
718 dbg_link("STATE_EV: <%s> ", l_ptr
->name
);
720 switch (l_ptr
->state
) {
721 case WORKING_WORKING
:
724 case TRAFFIC_MSG_EVT
:
732 if (l_ptr
->next_in_no
!= l_ptr
->checkpoint
) {
733 l_ptr
->checkpoint
= l_ptr
->next_in_no
;
734 if (tipc_bclink_acks_missing(l_ptr
->owner
)) {
735 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
737 l_ptr
->fsm_msg_cnt
++;
738 } else if (l_ptr
->max_pkt
< l_ptr
->max_pkt_target
) {
739 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
741 l_ptr
->fsm_msg_cnt
++;
743 link_set_timer(l_ptr
, cont_intv
);
746 dbg_link(" -> WU\n");
747 l_ptr
->state
= WORKING_UNKNOWN
;
748 l_ptr
->fsm_msg_cnt
= 0;
749 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
, 1, 0, 0, 0, 0);
750 l_ptr
->fsm_msg_cnt
++;
751 link_set_timer(l_ptr
, cont_intv
/ 4);
754 dbg_link("RES -> RR\n");
755 info("Resetting link <%s>, requested by peer\n",
757 tipc_link_reset(l_ptr
);
758 l_ptr
->state
= RESET_RESET
;
759 l_ptr
->fsm_msg_cnt
= 0;
760 tipc_link_send_proto_msg(l_ptr
, ACTIVATE_MSG
, 0, 0, 0, 0, 0);
761 l_ptr
->fsm_msg_cnt
++;
762 link_set_timer(l_ptr
, cont_intv
);
765 err("Unknown link event %u in WW state\n", event
);
768 case WORKING_UNKNOWN
:
771 case TRAFFIC_MSG_EVT
:
774 dbg_link("ACT -> WW\n");
775 l_ptr
->state
= WORKING_WORKING
;
776 l_ptr
->fsm_msg_cnt
= 0;
777 link_set_timer(l_ptr
, cont_intv
);
780 dbg_link("RES -> RR\n");
781 info("Resetting link <%s>, requested by peer "
782 "while probing\n", l_ptr
->name
);
783 tipc_link_reset(l_ptr
);
784 l_ptr
->state
= RESET_RESET
;
785 l_ptr
->fsm_msg_cnt
= 0;
786 tipc_link_send_proto_msg(l_ptr
, ACTIVATE_MSG
, 0, 0, 0, 0, 0);
787 l_ptr
->fsm_msg_cnt
++;
788 link_set_timer(l_ptr
, cont_intv
);
792 if (l_ptr
->next_in_no
!= l_ptr
->checkpoint
) {
794 l_ptr
->state
= WORKING_WORKING
;
795 l_ptr
->fsm_msg_cnt
= 0;
796 l_ptr
->checkpoint
= l_ptr
->next_in_no
;
797 if (tipc_bclink_acks_missing(l_ptr
->owner
)) {
798 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
800 l_ptr
->fsm_msg_cnt
++;
802 link_set_timer(l_ptr
, cont_intv
);
803 } else if (l_ptr
->fsm_msg_cnt
< l_ptr
->abort_limit
) {
804 dbg_link("Probing %u/%u,timer = %u ms)\n",
805 l_ptr
->fsm_msg_cnt
, l_ptr
->abort_limit
,
807 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
809 l_ptr
->fsm_msg_cnt
++;
810 link_set_timer(l_ptr
, cont_intv
/ 4);
811 } else { /* Link has failed */
812 dbg_link("-> RU (%u probes unanswered)\n",
814 warn("Resetting link <%s>, peer not responding\n",
816 tipc_link_reset(l_ptr
);
817 l_ptr
->state
= RESET_UNKNOWN
;
818 l_ptr
->fsm_msg_cnt
= 0;
819 tipc_link_send_proto_msg(l_ptr
, RESET_MSG
,
821 l_ptr
->fsm_msg_cnt
++;
822 link_set_timer(l_ptr
, cont_intv
);
826 err("Unknown link event %u in WU state\n", event
);
832 case TRAFFIC_MSG_EVT
:
836 other
= l_ptr
->owner
->active_links
[0];
837 if (other
&& link_working_unknown(other
)) {
841 dbg_link("ACT -> WW\n");
842 l_ptr
->state
= WORKING_WORKING
;
843 l_ptr
->fsm_msg_cnt
= 0;
844 link_activate(l_ptr
);
845 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
, 1, 0, 0, 0, 0);
846 l_ptr
->fsm_msg_cnt
++;
847 link_set_timer(l_ptr
, cont_intv
);
851 dbg_link(" -> RR\n");
852 l_ptr
->state
= RESET_RESET
;
853 l_ptr
->fsm_msg_cnt
= 0;
854 tipc_link_send_proto_msg(l_ptr
, ACTIVATE_MSG
, 1, 0, 0, 0, 0);
855 l_ptr
->fsm_msg_cnt
++;
856 link_set_timer(l_ptr
, cont_intv
);
864 tipc_link_send_proto_msg(l_ptr
, RESET_MSG
, 0, 0, 0, 0, 0);
865 l_ptr
->fsm_msg_cnt
++;
866 link_set_timer(l_ptr
, cont_intv
);
869 err("Unknown link event %u in RU state\n", event
);
875 case TRAFFIC_MSG_EVT
:
879 other
= l_ptr
->owner
->active_links
[0];
880 if (other
&& link_working_unknown(other
)) {
884 dbg_link("ACT -> WW\n");
885 l_ptr
->state
= WORKING_WORKING
;
886 l_ptr
->fsm_msg_cnt
= 0;
887 link_activate(l_ptr
);
888 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
, 1, 0, 0, 0, 0);
889 l_ptr
->fsm_msg_cnt
++;
890 link_set_timer(l_ptr
, cont_intv
);
897 tipc_link_send_proto_msg(l_ptr
, ACTIVATE_MSG
, 0, 0, 0, 0, 0);
898 l_ptr
->fsm_msg_cnt
++;
899 link_set_timer(l_ptr
, cont_intv
);
900 dbg_link("fsm_msg_cnt %u\n", l_ptr
->fsm_msg_cnt
);
903 err("Unknown link event %u in RR state\n", event
);
907 err("Unknown link state %u/%u\n", l_ptr
->state
, event
);
912 * link_bundle_buf(): Append contents of a buffer to
913 * the tail of an existing one.
916 static int link_bundle_buf(struct link
*l_ptr
,
917 struct sk_buff
*bundler
,
920 struct tipc_msg
*bundler_msg
= buf_msg(bundler
);
921 struct tipc_msg
*msg
= buf_msg(buf
);
922 u32 size
= msg_size(msg
);
923 u32 bundle_size
= msg_size(bundler_msg
);
924 u32 to_pos
= align(bundle_size
);
925 u32 pad
= to_pos
- bundle_size
;
927 if (msg_user(bundler_msg
) != MSG_BUNDLER
)
929 if (msg_type(bundler_msg
) != OPEN_MSG
)
931 if (skb_tailroom(bundler
) < (pad
+ size
))
933 if (l_ptr
->max_pkt
< (to_pos
+ size
))
936 skb_put(bundler
, pad
+ size
);
937 skb_copy_to_linear_data_offset(bundler
, to_pos
, buf
->data
, size
);
938 msg_set_size(bundler_msg
, to_pos
+ size
);
939 msg_set_msgcnt(bundler_msg
, msg_msgcnt(bundler_msg
) + 1);
940 dbg("Packed msg # %u(%u octets) into pos %u in buf(#%u)\n",
941 msg_msgcnt(bundler_msg
), size
, to_pos
, msg_seqno(bundler_msg
));
942 msg_dbg(msg
, "PACKD:");
944 l_ptr
->stats
.sent_bundled
++;
948 static void link_add_to_outqueue(struct link
*l_ptr
,
950 struct tipc_msg
*msg
)
952 u32 ack
= mod(l_ptr
->next_in_no
- 1);
953 u32 seqno
= mod(l_ptr
->next_out_no
++);
955 msg_set_word(msg
, 2, ((ack
<< 16) | seqno
));
956 msg_set_bcast_ack(msg
, l_ptr
->owner
->bclink
.last_in
);
958 if (l_ptr
->first_out
) {
959 l_ptr
->last_out
->next
= buf
;
960 l_ptr
->last_out
= buf
;
962 l_ptr
->first_out
= l_ptr
->last_out
= buf
;
963 l_ptr
->out_queue_size
++;
967 * tipc_link_send_buf() is the 'full path' for messages, called from
968 * inside TIPC when the 'fast path' in tipc_send_buf
969 * has failed, and from link_send()
972 int tipc_link_send_buf(struct link
*l_ptr
, struct sk_buff
*buf
)
974 struct tipc_msg
*msg
= buf_msg(buf
);
975 u32 size
= msg_size(msg
);
976 u32 dsz
= msg_data_sz(msg
);
977 u32 queue_size
= l_ptr
->out_queue_size
;
978 u32 imp
= tipc_msg_tot_importance(msg
);
979 u32 queue_limit
= l_ptr
->queue_limit
[imp
];
980 u32 max_packet
= l_ptr
->max_pkt
;
982 msg_set_prevnode(msg
, tipc_own_addr
); /* If routed message */
984 /* Match msg importance against queue limits: */
986 if (unlikely(queue_size
>= queue_limit
)) {
987 if (imp
<= TIPC_CRITICAL_IMPORTANCE
) {
988 return link_schedule_port(l_ptr
, msg_origport(msg
),
991 msg_dbg(msg
, "TIPC: Congestion, throwing away\n");
993 if (imp
> CONN_MANAGER
) {
994 warn("Resetting link <%s>, send queue full", l_ptr
->name
);
995 tipc_link_reset(l_ptr
);
1000 /* Fragmentation needed ? */
1002 if (size
> max_packet
)
1003 return tipc_link_send_long_buf(l_ptr
, buf
);
1005 /* Packet can be queued or sent: */
1007 if (queue_size
> l_ptr
->stats
.max_queue_sz
)
1008 l_ptr
->stats
.max_queue_sz
= queue_size
;
1010 if (likely(!tipc_bearer_congested(l_ptr
->b_ptr
, l_ptr
) &&
1011 !link_congested(l_ptr
))) {
1012 link_add_to_outqueue(l_ptr
, buf
, msg
);
1014 if (likely(tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
))) {
1015 l_ptr
->unacked_window
= 0;
1017 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
1018 l_ptr
->stats
.bearer_congs
++;
1019 l_ptr
->next_out
= buf
;
1023 /* Congestion: can message be bundled ?: */
1025 if ((msg_user(msg
) != CHANGEOVER_PROTOCOL
) &&
1026 (msg_user(msg
) != MSG_FRAGMENTER
)) {
1028 /* Try adding message to an existing bundle */
1030 if (l_ptr
->next_out
&&
1031 link_bundle_buf(l_ptr
, l_ptr
->last_out
, buf
)) {
1032 tipc_bearer_resolve_congestion(l_ptr
->b_ptr
, l_ptr
);
1036 /* Try creating a new bundle */
1038 if (size
<= max_packet
* 2 / 3) {
1039 struct sk_buff
*bundler
= buf_acquire(max_packet
);
1040 struct tipc_msg bundler_hdr
;
1043 tipc_msg_init(&bundler_hdr
, MSG_BUNDLER
, OPEN_MSG
,
1044 INT_H_SIZE
, l_ptr
->addr
);
1045 skb_copy_to_linear_data(bundler
, &bundler_hdr
,
1047 skb_trim(bundler
, INT_H_SIZE
);
1048 link_bundle_buf(l_ptr
, bundler
, buf
);
1051 l_ptr
->stats
.sent_bundles
++;
1055 if (!l_ptr
->next_out
)
1056 l_ptr
->next_out
= buf
;
1057 link_add_to_outqueue(l_ptr
, buf
, msg
);
1058 tipc_bearer_resolve_congestion(l_ptr
->b_ptr
, l_ptr
);
1063 * tipc_link_send(): same as tipc_link_send_buf(), but the link to use has
1064 * not been selected yet, and the the owner node is not locked
1065 * Called by TIPC internal users, e.g. the name distributor
1068 int tipc_link_send(struct sk_buff
*buf
, u32 dest
, u32 selector
)
1071 struct tipc_node
*n_ptr
;
1072 int res
= -ELINKCONG
;
1074 read_lock_bh(&tipc_net_lock
);
1075 n_ptr
= tipc_node_select(dest
, selector
);
1077 tipc_node_lock(n_ptr
);
1078 l_ptr
= n_ptr
->active_links
[selector
& 1];
1080 dbg("tipc_link_send: found link %x for dest %x\n", l_ptr
, dest
);
1081 res
= tipc_link_send_buf(l_ptr
, buf
);
1083 dbg("Attempt to send msg to unreachable node:\n");
1084 msg_dbg(buf_msg(buf
),">>>");
1087 tipc_node_unlock(n_ptr
);
1089 dbg("Attempt to send msg to unknown node:\n");
1090 msg_dbg(buf_msg(buf
),">>>");
1093 read_unlock_bh(&tipc_net_lock
);
1098 * link_send_buf_fast: Entry for data messages where the
1099 * destination link is known and the header is complete,
1100 * inclusive total message length. Very time critical.
1101 * Link is locked. Returns user data length.
1104 static int link_send_buf_fast(struct link
*l_ptr
, struct sk_buff
*buf
,
1107 struct tipc_msg
*msg
= buf_msg(buf
);
1108 int res
= msg_data_sz(msg
);
1110 if (likely(!link_congested(l_ptr
))) {
1111 if (likely(msg_size(msg
) <= l_ptr
->max_pkt
)) {
1112 if (likely(list_empty(&l_ptr
->b_ptr
->cong_links
))) {
1113 link_add_to_outqueue(l_ptr
, buf
, msg
);
1114 if (likely(tipc_bearer_send(l_ptr
->b_ptr
, buf
,
1115 &l_ptr
->media_addr
))) {
1116 l_ptr
->unacked_window
= 0;
1117 msg_dbg(msg
,"SENT_FAST:");
1120 dbg("failed sent fast...\n");
1121 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
1122 l_ptr
->stats
.bearer_congs
++;
1123 l_ptr
->next_out
= buf
;
1128 *used_max_pkt
= l_ptr
->max_pkt
;
1130 return tipc_link_send_buf(l_ptr
, buf
); /* All other cases */
1134 * tipc_send_buf_fast: Entry for data messages where the
1135 * destination node is known and the header is complete,
1136 * inclusive total message length.
1137 * Returns user data length.
1139 int tipc_send_buf_fast(struct sk_buff
*buf
, u32 destnode
)
1142 struct tipc_node
*n_ptr
;
1144 u32 selector
= msg_origport(buf_msg(buf
)) & 1;
1147 if (destnode
== tipc_own_addr
)
1148 return tipc_port_recv_msg(buf
);
1150 read_lock_bh(&tipc_net_lock
);
1151 n_ptr
= tipc_node_select(destnode
, selector
);
1152 if (likely(n_ptr
)) {
1153 tipc_node_lock(n_ptr
);
1154 l_ptr
= n_ptr
->active_links
[selector
];
1155 dbg("send_fast: buf %x selected %x, destnode = %x\n",
1156 buf
, l_ptr
, destnode
);
1157 if (likely(l_ptr
)) {
1158 res
= link_send_buf_fast(l_ptr
, buf
, &dummy
);
1159 tipc_node_unlock(n_ptr
);
1160 read_unlock_bh(&tipc_net_lock
);
1163 tipc_node_unlock(n_ptr
);
1165 read_unlock_bh(&tipc_net_lock
);
1166 res
= msg_data_sz(buf_msg(buf
));
1167 tipc_reject_msg(buf
, TIPC_ERR_NO_NODE
);
1173 * tipc_link_send_sections_fast: Entry for messages where the
1174 * destination processor is known and the header is complete,
1175 * except for total message length.
1176 * Returns user data length or errno.
1178 int tipc_link_send_sections_fast(struct port
*sender
,
1179 struct iovec
const *msg_sect
,
1183 struct tipc_msg
*hdr
= &sender
->publ
.phdr
;
1185 struct sk_buff
*buf
;
1186 struct tipc_node
*node
;
1188 u32 selector
= msg_origport(hdr
) & 1;
1192 * Try building message using port's max_pkt hint.
1193 * (Must not hold any locks while building message.)
1196 res
= tipc_msg_build(hdr
, msg_sect
, num_sect
, sender
->publ
.max_pkt
,
1197 !sender
->user_port
, &buf
);
1199 read_lock_bh(&tipc_net_lock
);
1200 node
= tipc_node_select(destaddr
, selector
);
1202 tipc_node_lock(node
);
1203 l_ptr
= node
->active_links
[selector
];
1204 if (likely(l_ptr
)) {
1206 res
= link_send_buf_fast(l_ptr
, buf
,
1207 &sender
->publ
.max_pkt
);
1208 if (unlikely(res
< 0))
1211 tipc_node_unlock(node
);
1212 read_unlock_bh(&tipc_net_lock
);
1216 /* Exit if build request was invalid */
1218 if (unlikely(res
< 0))
1221 /* Exit if link (or bearer) is congested */
1223 if (link_congested(l_ptr
) ||
1224 !list_empty(&l_ptr
->b_ptr
->cong_links
)) {
1225 res
= link_schedule_port(l_ptr
,
1226 sender
->publ
.ref
, res
);
1231 * Message size exceeds max_pkt hint; update hint,
1232 * then re-try fast path or fragment the message
1235 sender
->publ
.max_pkt
= l_ptr
->max_pkt
;
1236 tipc_node_unlock(node
);
1237 read_unlock_bh(&tipc_net_lock
);
1240 if ((msg_hdr_sz(hdr
) + res
) <= sender
->publ
.max_pkt
)
1243 return link_send_sections_long(sender
, msg_sect
,
1244 num_sect
, destaddr
);
1246 tipc_node_unlock(node
);
1248 read_unlock_bh(&tipc_net_lock
);
1250 /* Couldn't find a link to the destination node */
1253 return tipc_reject_msg(buf
, TIPC_ERR_NO_NODE
);
1255 return tipc_port_reject_sections(sender
, hdr
, msg_sect
, num_sect
,
1261 * link_send_sections_long(): Entry for long messages where the
1262 * destination node is known and the header is complete,
1263 * inclusive total message length.
1264 * Link and bearer congestion status have been checked to be ok,
1265 * and are ignored if they change.
1267 * Note that fragments do not use the full link MTU so that they won't have
1268 * to undergo refragmentation if link changeover causes them to be sent
1269 * over another link with an additional tunnel header added as prefix.
1270 * (Refragmentation will still occur if the other link has a smaller MTU.)
1272 * Returns user data length or errno.
1274 static int link_send_sections_long(struct port
*sender
,
1275 struct iovec
const *msg_sect
,
1280 struct tipc_node
*node
;
1281 struct tipc_msg
*hdr
= &sender
->publ
.phdr
;
1282 u32 dsz
= msg_data_sz(hdr
);
1283 u32 max_pkt
,fragm_sz
,rest
;
1284 struct tipc_msg fragm_hdr
;
1285 struct sk_buff
*buf
,*buf_chain
,*prev
;
1286 u32 fragm_crs
,fragm_rest
,hsz
,sect_rest
;
1287 const unchar
*sect_crs
;
1293 max_pkt
= sender
->publ
.max_pkt
- INT_H_SIZE
;
1294 /* leave room for tunnel header in case of link changeover */
1295 fragm_sz
= max_pkt
- INT_H_SIZE
;
1296 /* leave room for fragmentation header in each fragment */
1304 /* Prepare reusable fragment header: */
1306 msg_dbg(hdr
, ">FRAGMENTING>");
1307 tipc_msg_init(&fragm_hdr
, MSG_FRAGMENTER
, FIRST_FRAGMENT
,
1308 INT_H_SIZE
, msg_destnode(hdr
));
1309 msg_set_link_selector(&fragm_hdr
, sender
->publ
.ref
);
1310 msg_set_size(&fragm_hdr
, max_pkt
);
1311 msg_set_fragm_no(&fragm_hdr
, 1);
1313 /* Prepare header of first fragment: */
1315 buf_chain
= buf
= buf_acquire(max_pkt
);
1319 skb_copy_to_linear_data(buf
, &fragm_hdr
, INT_H_SIZE
);
1320 hsz
= msg_hdr_sz(hdr
);
1321 skb_copy_to_linear_data_offset(buf
, INT_H_SIZE
, hdr
, hsz
);
1322 msg_dbg(buf_msg(buf
), ">BUILD>");
1324 /* Chop up message: */
1326 fragm_crs
= INT_H_SIZE
+ hsz
;
1327 fragm_rest
= fragm_sz
- hsz
;
1329 do { /* For all sections */
1333 sect_rest
= msg_sect
[++curr_sect
].iov_len
;
1334 sect_crs
= (const unchar
*)msg_sect
[curr_sect
].iov_base
;
1337 if (sect_rest
< fragm_rest
)
1342 if (likely(!sender
->user_port
)) {
1343 if (copy_from_user(buf
->data
+ fragm_crs
, sect_crs
, sz
)) {
1345 for (; buf_chain
; buf_chain
= buf
) {
1346 buf
= buf_chain
->next
;
1347 buf_discard(buf_chain
);
1352 skb_copy_to_linear_data_offset(buf
, fragm_crs
,
1360 if (!fragm_rest
&& rest
) {
1362 /* Initiate new fragment: */
1363 if (rest
<= fragm_sz
) {
1365 msg_set_type(&fragm_hdr
,LAST_FRAGMENT
);
1367 msg_set_type(&fragm_hdr
, FRAGMENT
);
1369 msg_set_size(&fragm_hdr
, fragm_sz
+ INT_H_SIZE
);
1370 msg_set_fragm_no(&fragm_hdr
, ++fragm_no
);
1372 buf
= buf_acquire(fragm_sz
+ INT_H_SIZE
);
1378 skb_copy_to_linear_data(buf
, &fragm_hdr
, INT_H_SIZE
);
1379 fragm_crs
= INT_H_SIZE
;
1380 fragm_rest
= fragm_sz
;
1381 msg_dbg(buf_msg(buf
)," >BUILD>");
1387 * Now we have a buffer chain. Select a link and check
1388 * that packet size is still OK
1390 node
= tipc_node_select(destaddr
, sender
->publ
.ref
& 1);
1392 tipc_node_lock(node
);
1393 l_ptr
= node
->active_links
[sender
->publ
.ref
& 1];
1395 tipc_node_unlock(node
);
1398 if (l_ptr
->max_pkt
< max_pkt
) {
1399 sender
->publ
.max_pkt
= l_ptr
->max_pkt
;
1400 tipc_node_unlock(node
);
1401 for (; buf_chain
; buf_chain
= buf
) {
1402 buf
= buf_chain
->next
;
1403 buf_discard(buf_chain
);
1409 for (; buf_chain
; buf_chain
= buf
) {
1410 buf
= buf_chain
->next
;
1411 buf_discard(buf_chain
);
1413 return tipc_port_reject_sections(sender
, hdr
, msg_sect
, num_sect
,
1417 /* Append whole chain to send queue: */
1420 l_ptr
->long_msg_seq_no
= mod(l_ptr
->long_msg_seq_no
+ 1);
1421 if (!l_ptr
->next_out
)
1422 l_ptr
->next_out
= buf_chain
;
1423 l_ptr
->stats
.sent_fragmented
++;
1425 struct sk_buff
*next
= buf
->next
;
1426 struct tipc_msg
*msg
= buf_msg(buf
);
1428 l_ptr
->stats
.sent_fragments
++;
1429 msg_set_long_msgno(msg
, l_ptr
->long_msg_seq_no
);
1430 link_add_to_outqueue(l_ptr
, buf
, msg
);
1431 msg_dbg(msg
, ">ADD>");
1435 /* Send it, if possible: */
1437 tipc_link_push_queue(l_ptr
);
1438 tipc_node_unlock(node
);
1443 * tipc_link_push_packet: Push one unsent packet to the media
1445 u32
tipc_link_push_packet(struct link
*l_ptr
)
1447 struct sk_buff
*buf
= l_ptr
->first_out
;
1448 u32 r_q_size
= l_ptr
->retransm_queue_size
;
1449 u32 r_q_head
= l_ptr
->retransm_queue_head
;
1451 /* Step to position where retransmission failed, if any, */
1452 /* consider that buffers may have been released in meantime */
1454 if (r_q_size
&& buf
) {
1455 u32 last
= lesser(mod(r_q_head
+ r_q_size
),
1456 link_last_sent(l_ptr
));
1457 u32 first
= msg_seqno(buf_msg(buf
));
1459 while (buf
&& less(first
, r_q_head
)) {
1460 first
= mod(first
+ 1);
1463 l_ptr
->retransm_queue_head
= r_q_head
= first
;
1464 l_ptr
->retransm_queue_size
= r_q_size
= mod(last
- first
);
1467 /* Continue retransmission now, if there is anything: */
1469 if (r_q_size
&& buf
) {
1470 msg_set_ack(buf_msg(buf
), mod(l_ptr
->next_in_no
- 1));
1471 msg_set_bcast_ack(buf_msg(buf
), l_ptr
->owner
->bclink
.last_in
);
1472 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
1473 msg_dbg(buf_msg(buf
), ">DEF-RETR>");
1474 l_ptr
->retransm_queue_head
= mod(++r_q_head
);
1475 l_ptr
->retransm_queue_size
= --r_q_size
;
1476 l_ptr
->stats
.retransmitted
++;
1479 l_ptr
->stats
.bearer_congs
++;
1480 msg_dbg(buf_msg(buf
), "|>DEF-RETR>");
1485 /* Send deferred protocol message, if any: */
1487 buf
= l_ptr
->proto_msg_queue
;
1489 msg_set_ack(buf_msg(buf
), mod(l_ptr
->next_in_no
- 1));
1490 msg_set_bcast_ack(buf_msg(buf
),l_ptr
->owner
->bclink
.last_in
);
1491 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
1492 msg_dbg(buf_msg(buf
), ">DEF-PROT>");
1493 l_ptr
->unacked_window
= 0;
1495 l_ptr
->proto_msg_queue
= NULL
;
1498 msg_dbg(buf_msg(buf
), "|>DEF-PROT>");
1499 l_ptr
->stats
.bearer_congs
++;
1504 /* Send one deferred data message, if send window not full: */
1506 buf
= l_ptr
->next_out
;
1508 struct tipc_msg
*msg
= buf_msg(buf
);
1509 u32 next
= msg_seqno(msg
);
1510 u32 first
= msg_seqno(buf_msg(l_ptr
->first_out
));
1512 if (mod(next
- first
) < l_ptr
->queue_limit
[0]) {
1513 msg_set_ack(msg
, mod(l_ptr
->next_in_no
- 1));
1514 msg_set_bcast_ack(msg
, l_ptr
->owner
->bclink
.last_in
);
1515 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
1516 if (msg_user(msg
) == MSG_BUNDLER
)
1517 msg_set_type(msg
, CLOSED_MSG
);
1518 msg_dbg(msg
, ">PUSH-DATA>");
1519 l_ptr
->next_out
= buf
->next
;
1522 msg_dbg(msg
, "|PUSH-DATA|");
1523 l_ptr
->stats
.bearer_congs
++;
1528 return PUSH_FINISHED
;
1532 * push_queue(): push out the unsent messages of a link where
1533 * congestion has abated. Node is locked
1535 void tipc_link_push_queue(struct link
*l_ptr
)
1539 if (tipc_bearer_congested(l_ptr
->b_ptr
, l_ptr
))
1543 res
= tipc_link_push_packet(l_ptr
);
1546 if (res
== PUSH_FAILED
)
1547 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
1550 static void link_reset_all(unsigned long addr
)
1552 struct tipc_node
*n_ptr
;
1553 char addr_string
[16];
1556 read_lock_bh(&tipc_net_lock
);
1557 n_ptr
= tipc_node_find((u32
)addr
);
1559 read_unlock_bh(&tipc_net_lock
);
1560 return; /* node no longer exists */
1563 tipc_node_lock(n_ptr
);
1565 warn("Resetting all links to %s\n",
1566 tipc_addr_string_fill(addr_string
, n_ptr
->addr
));
1568 for (i
= 0; i
< MAX_BEARERS
; i
++) {
1569 if (n_ptr
->links
[i
]) {
1570 link_print(n_ptr
->links
[i
], TIPC_OUTPUT
,
1571 "Resetting link\n");
1572 tipc_link_reset(n_ptr
->links
[i
]);
1576 tipc_node_unlock(n_ptr
);
1577 read_unlock_bh(&tipc_net_lock
);
1580 static void link_retransmit_failure(struct link
*l_ptr
, struct sk_buff
*buf
)
1582 struct tipc_msg
*msg
= buf_msg(buf
);
1584 warn("Retransmission failure on link <%s>\n", l_ptr
->name
);
1585 tipc_msg_dbg(TIPC_OUTPUT
, msg
, ">RETR-FAIL>");
1589 /* Handle failure on standard link */
1591 link_print(l_ptr
, TIPC_OUTPUT
, "Resetting link\n");
1592 tipc_link_reset(l_ptr
);
1596 /* Handle failure on broadcast link */
1598 struct tipc_node
*n_ptr
;
1599 char addr_string
[16];
1601 tipc_printf(TIPC_OUTPUT
, "Msg seq number: %u, ", msg_seqno(msg
));
1602 tipc_printf(TIPC_OUTPUT
, "Outstanding acks: %lu\n",
1603 (unsigned long) TIPC_SKB_CB(buf
)->handle
);
1605 n_ptr
= l_ptr
->owner
->next
;
1606 tipc_node_lock(n_ptr
);
1608 tipc_addr_string_fill(addr_string
, n_ptr
->addr
);
1609 tipc_printf(TIPC_OUTPUT
, "Multicast link info for %s\n", addr_string
);
1610 tipc_printf(TIPC_OUTPUT
, "Supported: %d, ", n_ptr
->bclink
.supported
);
1611 tipc_printf(TIPC_OUTPUT
, "Acked: %u\n", n_ptr
->bclink
.acked
);
1612 tipc_printf(TIPC_OUTPUT
, "Last in: %u, ", n_ptr
->bclink
.last_in
);
1613 tipc_printf(TIPC_OUTPUT
, "Gap after: %u, ", n_ptr
->bclink
.gap_after
);
1614 tipc_printf(TIPC_OUTPUT
, "Gap to: %u\n", n_ptr
->bclink
.gap_to
);
1615 tipc_printf(TIPC_OUTPUT
, "Nack sync: %u\n\n", n_ptr
->bclink
.nack_sync
);
1617 tipc_k_signal((Handler
)link_reset_all
, (unsigned long)n_ptr
->addr
);
1619 tipc_node_unlock(n_ptr
);
1621 l_ptr
->stale_count
= 0;
1625 void tipc_link_retransmit(struct link
*l_ptr
, struct sk_buff
*buf
,
1628 struct tipc_msg
*msg
;
1635 dbg("Retransmitting %u in link %x\n", retransmits
, l_ptr
);
1637 if (tipc_bearer_congested(l_ptr
->b_ptr
, l_ptr
)) {
1638 if (l_ptr
->retransm_queue_size
== 0) {
1639 msg_dbg(msg
, ">NO_RETR->BCONG>");
1640 dbg_print_link(l_ptr
, " ");
1641 l_ptr
->retransm_queue_head
= msg_seqno(msg
);
1642 l_ptr
->retransm_queue_size
= retransmits
;
1644 err("Unexpected retransmit on link %s (qsize=%d)\n",
1645 l_ptr
->name
, l_ptr
->retransm_queue_size
);
1649 /* Detect repeated retransmit failures on uncongested bearer */
1651 if (l_ptr
->last_retransmitted
== msg_seqno(msg
)) {
1652 if (++l_ptr
->stale_count
> 100) {
1653 link_retransmit_failure(l_ptr
, buf
);
1657 l_ptr
->last_retransmitted
= msg_seqno(msg
);
1658 l_ptr
->stale_count
= 1;
1662 while (retransmits
&& (buf
!= l_ptr
->next_out
) && buf
) {
1664 msg_set_ack(msg
, mod(l_ptr
->next_in_no
- 1));
1665 msg_set_bcast_ack(msg
, l_ptr
->owner
->bclink
.last_in
);
1666 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
1667 msg_dbg(buf_msg(buf
), ">RETR>");
1670 l_ptr
->stats
.retransmitted
++;
1672 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
1673 l_ptr
->stats
.bearer_congs
++;
1674 l_ptr
->retransm_queue_head
= msg_seqno(buf_msg(buf
));
1675 l_ptr
->retransm_queue_size
= retransmits
;
1680 l_ptr
->retransm_queue_head
= l_ptr
->retransm_queue_size
= 0;
1684 * link_insert_deferred_queue - insert deferred messages back into receive chain
1687 static struct sk_buff
*link_insert_deferred_queue(struct link
*l_ptr
,
1688 struct sk_buff
*buf
)
1692 if (l_ptr
->oldest_deferred_in
== NULL
)
1695 seq_no
= msg_seqno(buf_msg(l_ptr
->oldest_deferred_in
));
1696 if (seq_no
== mod(l_ptr
->next_in_no
)) {
1697 l_ptr
->newest_deferred_in
->next
= buf
;
1698 buf
= l_ptr
->oldest_deferred_in
;
1699 l_ptr
->oldest_deferred_in
= NULL
;
1700 l_ptr
->deferred_inqueue_sz
= 0;
1706 * link_recv_buf_validate - validate basic format of received message
1708 * This routine ensures a TIPC message has an acceptable header, and at least
1709 * as much data as the header indicates it should. The routine also ensures
1710 * that the entire message header is stored in the main fragment of the message
1711 * buffer, to simplify future access to message header fields.
1713 * Note: Having extra info present in the message header or data areas is OK.
1714 * TIPC will ignore the excess, under the assumption that it is optional info
1715 * introduced by a later release of the protocol.
1718 static int link_recv_buf_validate(struct sk_buff
*buf
)
1720 static u32 min_data_hdr_size
[8] = {
1721 SHORT_H_SIZE
, MCAST_H_SIZE
, LONG_H_SIZE
, DIR_MSG_H_SIZE
,
1722 MAX_H_SIZE
, MAX_H_SIZE
, MAX_H_SIZE
, MAX_H_SIZE
1725 struct tipc_msg
*msg
;
1731 if (unlikely(buf
->len
< MIN_H_SIZE
))
1734 msg
= skb_header_pointer(buf
, 0, sizeof(tipc_hdr
), tipc_hdr
);
1738 if (unlikely(msg_version(msg
) != TIPC_VERSION
))
1741 size
= msg_size(msg
);
1742 hdr_size
= msg_hdr_sz(msg
);
1743 min_hdr_size
= msg_isdata(msg
) ?
1744 min_data_hdr_size
[msg_type(msg
)] : INT_H_SIZE
;
1746 if (unlikely((hdr_size
< min_hdr_size
) ||
1747 (size
< hdr_size
) ||
1748 (buf
->len
< size
) ||
1749 (size
- hdr_size
> TIPC_MAX_USER_MSG_SIZE
)))
1752 return pskb_may_pull(buf
, hdr_size
);
1755 void tipc_recv_msg(struct sk_buff
*head
, struct tipc_bearer
*tb_ptr
)
1757 read_lock_bh(&tipc_net_lock
);
1759 struct bearer
*b_ptr
= (struct bearer
*)tb_ptr
;
1760 struct tipc_node
*n_ptr
;
1762 struct sk_buff
*crs
;
1763 struct sk_buff
*buf
= head
;
1764 struct tipc_msg
*msg
;
1772 /* Ensure message is well-formed */
1774 if (unlikely(!link_recv_buf_validate(buf
)))
1777 /* Ensure message data is a single contiguous unit */
1779 if (unlikely(buf_linearize(buf
))) {
1783 /* Handle arrival of a non-unicast link message */
1787 if (unlikely(msg_non_seq(msg
))) {
1788 if (msg_user(msg
) == LINK_CONFIG
)
1789 tipc_disc_recv_msg(buf
, b_ptr
);
1791 tipc_bclink_recv_pkt(buf
);
1795 if (unlikely(!msg_short(msg
) &&
1796 (msg_destnode(msg
) != tipc_own_addr
)))
1799 /* Discard non-routeable messages destined for another node */
1801 if (unlikely(!msg_isdata(msg
) &&
1802 (msg_destnode(msg
) != tipc_own_addr
))) {
1803 if ((msg_user(msg
) != CONN_MANAGER
) &&
1804 (msg_user(msg
) != MSG_FRAGMENTER
))
1808 /* Locate unicast link endpoint that should handle message */
1810 n_ptr
= tipc_node_find(msg_prevnode(msg
));
1811 if (unlikely(!n_ptr
))
1813 tipc_node_lock(n_ptr
);
1815 l_ptr
= n_ptr
->links
[b_ptr
->identity
];
1816 if (unlikely(!l_ptr
)) {
1817 tipc_node_unlock(n_ptr
);
1821 /* Validate message sequence number info */
1823 seq_no
= msg_seqno(msg
);
1824 ackd
= msg_ack(msg
);
1826 /* Release acked messages */
1828 if (less(n_ptr
->bclink
.acked
, msg_bcast_ack(msg
))) {
1829 if (tipc_node_is_up(n_ptr
) && n_ptr
->bclink
.supported
)
1830 tipc_bclink_acknowledge(n_ptr
, msg_bcast_ack(msg
));
1833 crs
= l_ptr
->first_out
;
1834 while ((crs
!= l_ptr
->next_out
) &&
1835 less_eq(msg_seqno(buf_msg(crs
)), ackd
)) {
1836 struct sk_buff
*next
= crs
->next
;
1843 l_ptr
->first_out
= crs
;
1844 l_ptr
->out_queue_size
-= released
;
1847 /* Try sending any messages link endpoint has pending */
1849 if (unlikely(l_ptr
->next_out
))
1850 tipc_link_push_queue(l_ptr
);
1851 if (unlikely(!list_empty(&l_ptr
->waiting_ports
)))
1852 tipc_link_wakeup_ports(l_ptr
, 0);
1853 if (unlikely(++l_ptr
->unacked_window
>= TIPC_MIN_LINK_WIN
)) {
1854 l_ptr
->stats
.sent_acks
++;
1855 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
, 0, 0, 0, 0, 0);
1858 /* Now (finally!) process the incoming message */
1861 if (likely(link_working_working(l_ptr
))) {
1862 if (likely(seq_no
== mod(l_ptr
->next_in_no
))) {
1863 l_ptr
->next_in_no
++;
1864 if (unlikely(l_ptr
->oldest_deferred_in
))
1865 head
= link_insert_deferred_queue(l_ptr
,
1867 if (likely(msg_is_dest(msg
, tipc_own_addr
))) {
1869 if (likely(msg_isdata(msg
))) {
1870 tipc_node_unlock(n_ptr
);
1871 tipc_port_recv_msg(buf
);
1874 switch (msg_user(msg
)) {
1876 l_ptr
->stats
.recv_bundles
++;
1877 l_ptr
->stats
.recv_bundled
+=
1879 tipc_node_unlock(n_ptr
);
1880 tipc_link_recv_bundle(buf
);
1882 case ROUTE_DISTRIBUTOR
:
1883 tipc_node_unlock(n_ptr
);
1884 tipc_cltr_recv_routing_table(buf
);
1886 case NAME_DISTRIBUTOR
:
1887 tipc_node_unlock(n_ptr
);
1888 tipc_named_recv(buf
);
1891 tipc_node_unlock(n_ptr
);
1892 tipc_port_recv_proto_msg(buf
);
1894 case MSG_FRAGMENTER
:
1895 l_ptr
->stats
.recv_fragments
++;
1896 if (tipc_link_recv_fragment(&l_ptr
->defragm_buf
,
1898 l_ptr
->stats
.recv_fragmented
++;
1902 case CHANGEOVER_PROTOCOL
:
1903 type
= msg_type(msg
);
1904 if (link_recv_changeover_msg(&l_ptr
, &buf
)) {
1906 seq_no
= msg_seqno(msg
);
1907 if (type
== ORIGINAL_MSG
)
1909 goto protocol_check
;
1914 tipc_node_unlock(n_ptr
);
1915 tipc_net_route_msg(buf
);
1918 link_handle_out_of_seq_msg(l_ptr
, buf
);
1919 head
= link_insert_deferred_queue(l_ptr
, head
);
1920 tipc_node_unlock(n_ptr
);
1924 if (msg_user(msg
) == LINK_PROTOCOL
) {
1925 link_recv_proto_msg(l_ptr
, buf
);
1926 head
= link_insert_deferred_queue(l_ptr
, head
);
1927 tipc_node_unlock(n_ptr
);
1930 msg_dbg(msg
,"NSEQ<REC<");
1931 link_state_event(l_ptr
, TRAFFIC_MSG_EVT
);
1933 if (link_working_working(l_ptr
)) {
1934 /* Re-insert in front of queue */
1935 msg_dbg(msg
,"RECV-REINS:");
1938 tipc_node_unlock(n_ptr
);
1941 tipc_node_unlock(n_ptr
);
1945 read_unlock_bh(&tipc_net_lock
);
1949 * link_defer_buf(): Sort a received out-of-sequence packet
1950 * into the deferred reception queue.
1951 * Returns the increase of the queue length,i.e. 0 or 1
1954 u32
tipc_link_defer_pkt(struct sk_buff
**head
,
1955 struct sk_buff
**tail
,
1956 struct sk_buff
*buf
)
1958 struct sk_buff
*prev
= NULL
;
1959 struct sk_buff
*crs
= *head
;
1960 u32 seq_no
= msg_seqno(buf_msg(buf
));
1965 if (*head
== NULL
) {
1966 *head
= *tail
= buf
;
1971 if (less(msg_seqno(buf_msg(*tail
)), seq_no
)) {
1972 (*tail
)->next
= buf
;
1977 /* Scan through queue and sort it in */
1979 struct tipc_msg
*msg
= buf_msg(crs
);
1981 if (less(seq_no
, msg_seqno(msg
))) {
1989 if (seq_no
== msg_seqno(msg
)) {
1997 /* Message is a duplicate of an existing message */
2004 * link_handle_out_of_seq_msg - handle arrival of out-of-sequence packet
2007 static void link_handle_out_of_seq_msg(struct link
*l_ptr
,
2008 struct sk_buff
*buf
)
2010 u32 seq_no
= msg_seqno(buf_msg(buf
));
2012 if (likely(msg_user(buf_msg(buf
)) == LINK_PROTOCOL
)) {
2013 link_recv_proto_msg(l_ptr
, buf
);
2017 dbg("rx OOS msg: seq_no %u, expecting %u (%u)\n",
2018 seq_no
, mod(l_ptr
->next_in_no
), l_ptr
->next_in_no
);
2020 /* Record OOS packet arrival (force mismatch on next timeout) */
2022 l_ptr
->checkpoint
--;
2025 * Discard packet if a duplicate; otherwise add it to deferred queue
2026 * and notify peer of gap as per protocol specification
2029 if (less(seq_no
, mod(l_ptr
->next_in_no
))) {
2030 l_ptr
->stats
.duplicates
++;
2035 if (tipc_link_defer_pkt(&l_ptr
->oldest_deferred_in
,
2036 &l_ptr
->newest_deferred_in
, buf
)) {
2037 l_ptr
->deferred_inqueue_sz
++;
2038 l_ptr
->stats
.deferred_recv
++;
2039 if ((l_ptr
->deferred_inqueue_sz
% 16) == 1)
2040 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
, 0, 0, 0, 0, 0);
2042 l_ptr
->stats
.duplicates
++;
2046 * Send protocol message to the other endpoint.
2048 void tipc_link_send_proto_msg(struct link
*l_ptr
, u32 msg_typ
, int probe_msg
,
2049 u32 gap
, u32 tolerance
, u32 priority
, u32 ack_mtu
)
2051 struct sk_buff
*buf
= NULL
;
2052 struct tipc_msg
*msg
= l_ptr
->pmsg
;
2053 u32 msg_size
= sizeof(l_ptr
->proto_msg
);
2055 if (link_blocked(l_ptr
))
2057 msg_set_type(msg
, msg_typ
);
2058 msg_set_net_plane(msg
, l_ptr
->b_ptr
->net_plane
);
2059 msg_set_bcast_ack(msg
, mod(l_ptr
->owner
->bclink
.last_in
));
2060 msg_set_last_bcast(msg
, tipc_bclink_get_last_sent());
2062 if (msg_typ
== STATE_MSG
) {
2063 u32 next_sent
= mod(l_ptr
->next_out_no
);
2065 if (!tipc_link_is_up(l_ptr
))
2067 if (l_ptr
->next_out
)
2068 next_sent
= msg_seqno(buf_msg(l_ptr
->next_out
));
2069 msg_set_next_sent(msg
, next_sent
);
2070 if (l_ptr
->oldest_deferred_in
) {
2071 u32 rec
= msg_seqno(buf_msg(l_ptr
->oldest_deferred_in
));
2072 gap
= mod(rec
- mod(l_ptr
->next_in_no
));
2074 msg_set_seq_gap(msg
, gap
);
2076 l_ptr
->stats
.sent_nacks
++;
2077 msg_set_link_tolerance(msg
, tolerance
);
2078 msg_set_linkprio(msg
, priority
);
2079 msg_set_max_pkt(msg
, ack_mtu
);
2080 msg_set_ack(msg
, mod(l_ptr
->next_in_no
- 1));
2081 msg_set_probe(msg
, probe_msg
!= 0);
2083 u32 mtu
= l_ptr
->max_pkt
;
2085 if ((mtu
< l_ptr
->max_pkt_target
) &&
2086 link_working_working(l_ptr
) &&
2087 l_ptr
->fsm_msg_cnt
) {
2088 msg_size
= (mtu
+ (l_ptr
->max_pkt_target
- mtu
)/2 + 2) & ~3;
2089 if (l_ptr
->max_pkt_probes
== 10) {
2090 l_ptr
->max_pkt_target
= (msg_size
- 4);
2091 l_ptr
->max_pkt_probes
= 0;
2092 msg_size
= (mtu
+ (l_ptr
->max_pkt_target
- mtu
)/2 + 2) & ~3;
2094 l_ptr
->max_pkt_probes
++;
2097 l_ptr
->stats
.sent_probes
++;
2099 l_ptr
->stats
.sent_states
++;
2100 } else { /* RESET_MSG or ACTIVATE_MSG */
2101 msg_set_ack(msg
, mod(l_ptr
->reset_checkpoint
- 1));
2102 msg_set_seq_gap(msg
, 0);
2103 msg_set_next_sent(msg
, 1);
2104 msg_set_link_tolerance(msg
, l_ptr
->tolerance
);
2105 msg_set_linkprio(msg
, l_ptr
->priority
);
2106 msg_set_max_pkt(msg
, l_ptr
->max_pkt_target
);
2109 if (tipc_node_has_redundant_links(l_ptr
->owner
)) {
2110 msg_set_redundant_link(msg
);
2112 msg_clear_redundant_link(msg
);
2114 msg_set_linkprio(msg
, l_ptr
->priority
);
2116 /* Ensure sequence number will not fit : */
2118 msg_set_seqno(msg
, mod(l_ptr
->next_out_no
+ (0xffff/2)));
2122 if (tipc_bearer_congested(l_ptr
->b_ptr
, l_ptr
)) {
2123 if (!l_ptr
->proto_msg_queue
) {
2124 l_ptr
->proto_msg_queue
=
2125 buf_acquire(sizeof(l_ptr
->proto_msg
));
2127 buf
= l_ptr
->proto_msg_queue
;
2130 skb_copy_to_linear_data(buf
, msg
, sizeof(l_ptr
->proto_msg
));
2133 msg_set_timestamp(msg
, jiffies_to_msecs(jiffies
));
2135 /* Message can be sent */
2139 buf
= buf_acquire(msg_size
);
2143 skb_copy_to_linear_data(buf
, msg
, sizeof(l_ptr
->proto_msg
));
2144 msg_set_size(buf_msg(buf
), msg_size
);
2146 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
2147 l_ptr
->unacked_window
= 0;
2152 /* New congestion */
2153 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
2154 l_ptr
->proto_msg_queue
= buf
;
2155 l_ptr
->stats
.bearer_congs
++;
2159 * Receive protocol message :
2160 * Note that network plane id propagates through the network, and may
2161 * change at any time. The node with lowest address rules
2164 static void link_recv_proto_msg(struct link
*l_ptr
, struct sk_buff
*buf
)
2170 struct tipc_msg
*msg
= buf_msg(buf
);
2172 dbg("AT(%u):", jiffies_to_msecs(jiffies
));
2174 if (link_blocked(l_ptr
))
2177 /* record unnumbered packet arrival (force mismatch on next timeout) */
2179 l_ptr
->checkpoint
--;
2181 if (l_ptr
->b_ptr
->net_plane
!= msg_net_plane(msg
))
2182 if (tipc_own_addr
> msg_prevnode(msg
))
2183 l_ptr
->b_ptr
->net_plane
= msg_net_plane(msg
);
2185 l_ptr
->owner
->permit_changeover
= msg_redundant_link(msg
);
2187 switch (msg_type(msg
)) {
2190 if (!link_working_unknown(l_ptr
) &&
2191 (l_ptr
->peer_session
!= INVALID_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 */
2200 /* Update link settings according other endpoint's values */
2202 strcpy((strrchr(l_ptr
->name
, ':') + 1), (char *)msg_data(msg
));
2204 if ((msg_tol
= msg_link_tolerance(msg
)) &&
2205 (msg_tol
> l_ptr
->tolerance
))
2206 link_set_supervision_props(l_ptr
, msg_tol
);
2208 if (msg_linkprio(msg
) > l_ptr
->priority
)
2209 l_ptr
->priority
= msg_linkprio(msg
);
2211 max_pkt_info
= msg_max_pkt(msg
);
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
;
2218 l_ptr
->max_pkt
= l_ptr
->max_pkt_target
;
2220 l_ptr
->owner
->bclink
.supported
= (max_pkt_info
!= 0);
2222 link_state_event(l_ptr
, msg_type(msg
));
2224 l_ptr
->peer_session
= msg_session(msg
);
2225 l_ptr
->peer_bearer_id
= msg_bearer_id(msg
);
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
));
2234 if ((msg_tol
= msg_link_tolerance(msg
)))
2235 link_set_supervision_props(l_ptr
, msg_tol
);
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 */
2245 link_state_event(l_ptr
, TRAFFIC_MSG_EVT
);
2246 l_ptr
->stats
.recv_states
++;
2247 if (link_reset_unknown(l_ptr
))
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
));
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;
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
);
2271 /* Protocol message before retransmits, reduce loss risk */
2273 tipc_bclink_check_gap(l_ptr
->owner
, msg_last_bcast(msg
));
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
);
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
,
2287 msg_dbg(buf_msg(buf
), "<DISCARDING UNKNOWN<");
2295 * tipc_link_tunnel(): Send one message via a link belonging to
2296 * another bearer. Owner node is locked.
2298 void tipc_link_tunnel(struct link
*l_ptr
,
2299 struct tipc_msg
*tunnel_hdr
,
2300 struct tipc_msg
*msg
,
2303 struct link
*tunnel
;
2304 struct sk_buff
*buf
;
2305 u32 length
= msg_size(msg
);
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");
2313 msg_set_size(tunnel_hdr
, length
+ INT_H_SIZE
);
2314 buf
= buf_acquire(length
+ INT_H_SIZE
);
2316 warn("Link changeover error, "
2317 "unable to send tunnel msg\n");
2320 skb_copy_to_linear_data(buf
, tunnel_hdr
, INT_H_SIZE
);
2321 skb_copy_to_linear_data_offset(buf
, INT_H_SIZE
, 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
);
2330 * changeover(): Send whole message queue via the remaining link
2331 * Owner node is locked.
2334 void tipc_link_changeover(struct link
*l_ptr
)
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
;
2345 if (!l_ptr
->owner
->permit_changeover
) {
2346 warn("Link changeover error, "
2347 "peer did not permit changeover\n");
2351 tipc_msg_init(&tunnel_hdr
, CHANGEOVER_PROTOCOL
,
2352 ORIGINAL_MSG
, 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
);
2357 if (!l_ptr
->first_out
) {
2358 struct sk_buff
*buf
;
2360 buf
= buf_acquire(INT_H_SIZE
);
2362 skb_copy_to_linear_data(buf
, &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
);
2369 warn("Link changeover error, "
2370 "unable to send changeover msg\n");
2375 split_bundles
= (l_ptr
->owner
->active_links
[0] !=
2376 l_ptr
->owner
->active_links
[1]);
2379 struct tipc_msg
*msg
= buf_msg(crs
);
2381 if ((msg_user(msg
) == MSG_BUNDLER
) && split_bundles
) {
2382 struct tipc_msg
*m
= msg_get_wrapped(msg
);
2383 unchar
* pos
= (unchar
*)m
;
2385 msgcount
= msg_msgcnt(msg
);
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
;
2394 tipc_link_tunnel(l_ptr
, &tunnel_hdr
, msg
,
2395 msg_link_selector(msg
));
2401 void tipc_link_send_duplicate(struct link
*l_ptr
, struct link
*tunnel
)
2403 struct sk_buff
*iter
;
2404 struct tipc_msg tunnel_hdr
;
2406 tipc_msg_init(&tunnel_hdr
, CHANGEOVER_PROTOCOL
,
2407 DUPLICATE_MSG
, 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
;
2412 struct sk_buff
*outbuf
;
2413 struct tipc_msg
*msg
= buf_msg(iter
);
2414 u32 length
= msg_size(msg
);
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");
2427 skb_copy_to_linear_data(outbuf
, &tunnel_hdr
, INT_H_SIZE
);
2428 skb_copy_to_linear_data_offset(outbuf
, INT_H_SIZE
, iter
->data
,
2430 dbg("%c->%c:", l_ptr
->b_ptr
->net_plane
,
2431 tunnel
->b_ptr
->net_plane
);
2432 msg_dbg(buf_msg(outbuf
), ">SEND>");
2433 tipc_link_send_buf(tunnel
, outbuf
);
2434 if (!tipc_link_is_up(l_ptr
))
2443 * buf_extract - extracts embedded TIPC message from another message
2444 * @skb: encapsulating message buffer
2445 * @from_pos: offset to extract from
2447 * Returns a new message buffer containing an embedded message. The
2448 * encapsulating message itself is left unchanged.
2451 static struct sk_buff
*buf_extract(struct sk_buff
*skb
, u32 from_pos
)
2453 struct tipc_msg
*msg
= (struct tipc_msg
*)(skb
->data
+ from_pos
);
2454 u32 size
= msg_size(msg
);
2457 eb
= buf_acquire(size
);
2459 skb_copy_to_linear_data(eb
, msg
, size
);
2464 * link_recv_changeover_msg(): Receive tunneled packet sent
2465 * via other link. Node is locked. Return extracted buffer.
2468 static int link_recv_changeover_msg(struct link
**l_ptr
,
2469 struct sk_buff
**buf
)
2471 struct sk_buff
*tunnel_buf
= *buf
;
2472 struct link
*dest_link
;
2473 struct tipc_msg
*msg
;
2474 struct tipc_msg
*tunnel_msg
= buf_msg(tunnel_buf
);
2475 u32 msg_typ
= msg_type(tunnel_msg
);
2476 u32 msg_count
= msg_msgcnt(tunnel_msg
);
2478 dest_link
= (*l_ptr
)->owner
->links
[msg_bearer_id(tunnel_msg
)];
2480 msg_dbg(tunnel_msg
, "NOLINK/<REC<");
2483 if (dest_link
== *l_ptr
) {
2484 err("Unexpected changeover message on link <%s>\n",
2488 dbg("%c<-%c:", dest_link
->b_ptr
->net_plane
,
2489 (*l_ptr
)->b_ptr
->net_plane
);
2491 msg
= msg_get_wrapped(tunnel_msg
);
2493 if (msg_typ
== DUPLICATE_MSG
) {
2494 if (less(msg_seqno(msg
), mod(dest_link
->next_in_no
))) {
2495 msg_dbg(tunnel_msg
, "DROP/<REC<");
2498 *buf
= buf_extract(tunnel_buf
,INT_H_SIZE
);
2500 warn("Link changeover error, duplicate msg dropped\n");
2503 msg_dbg(tunnel_msg
, "TNL<REC<");
2504 buf_discard(tunnel_buf
);
2508 /* First original message ?: */
2510 if (tipc_link_is_up(dest_link
)) {
2511 msg_dbg(tunnel_msg
, "UP/FIRST/<REC<");
2512 info("Resetting link <%s>, changeover initiated by peer\n",
2514 tipc_link_reset(dest_link
);
2515 dest_link
->exp_msg_count
= msg_count
;
2516 dbg("Expecting %u tunnelled messages\n", msg_count
);
2519 } else if (dest_link
->exp_msg_count
== START_CHANGEOVER
) {
2520 msg_dbg(tunnel_msg
, "BLK/FIRST/<REC<");
2521 dest_link
->exp_msg_count
= msg_count
;
2522 dbg("Expecting %u tunnelled messages\n", msg_count
);
2527 /* Receive original message */
2529 if (dest_link
->exp_msg_count
== 0) {
2530 warn("Link switchover error, "
2531 "got too many tunnelled messages\n");
2532 msg_dbg(tunnel_msg
, "OVERDUE/DROP/<REC<");
2533 dbg_print_link(dest_link
, "LINK:");
2536 dest_link
->exp_msg_count
--;
2537 if (less(msg_seqno(msg
), dest_link
->reset_checkpoint
)) {
2538 msg_dbg(tunnel_msg
, "DROP/DUPL/<REC<");
2541 *buf
= buf_extract(tunnel_buf
, INT_H_SIZE
);
2543 msg_dbg(tunnel_msg
, "TNL<REC<");
2544 buf_discard(tunnel_buf
);
2547 warn("Link changeover error, original msg dropped\n");
2552 buf_discard(tunnel_buf
);
2557 * Bundler functionality:
2559 void tipc_link_recv_bundle(struct sk_buff
*buf
)
2561 u32 msgcount
= msg_msgcnt(buf_msg(buf
));
2562 u32 pos
= INT_H_SIZE
;
2563 struct sk_buff
*obuf
;
2565 msg_dbg(buf_msg(buf
), "<BNDL<: ");
2566 while (msgcount
--) {
2567 obuf
= buf_extract(buf
, pos
);
2569 warn("Link unable to unbundle message(s)\n");
2572 pos
+= align(msg_size(buf_msg(obuf
)));
2573 msg_dbg(buf_msg(obuf
), " /");
2574 tipc_net_route_msg(obuf
);
2580 * Fragmentation/defragmentation:
2585 * tipc_link_send_long_buf: Entry for buffers needing fragmentation.
2586 * The buffer is complete, inclusive total message length.
2587 * Returns user data length.
2589 int tipc_link_send_long_buf(struct link
*l_ptr
, struct sk_buff
*buf
)
2591 struct tipc_msg
*inmsg
= buf_msg(buf
);
2592 struct tipc_msg fragm_hdr
;
2593 u32 insize
= msg_size(inmsg
);
2594 u32 dsz
= msg_data_sz(inmsg
);
2595 unchar
*crs
= buf
->data
;
2597 u32 pack_sz
= l_ptr
->max_pkt
;
2598 u32 fragm_sz
= pack_sz
- INT_H_SIZE
;
2602 if (msg_short(inmsg
))
2603 destaddr
= l_ptr
->addr
;
2605 destaddr
= msg_destnode(inmsg
);
2607 if (msg_routed(inmsg
))
2608 msg_set_prevnode(inmsg
, tipc_own_addr
);
2610 /* Prepare reusable fragment header: */
2612 tipc_msg_init(&fragm_hdr
, MSG_FRAGMENTER
, FIRST_FRAGMENT
,
2613 INT_H_SIZE
, destaddr
);
2614 msg_set_link_selector(&fragm_hdr
, msg_link_selector(inmsg
));
2615 msg_set_long_msgno(&fragm_hdr
, mod(l_ptr
->long_msg_seq_no
++));
2616 msg_set_fragm_no(&fragm_hdr
, fragm_no
);
2617 l_ptr
->stats
.sent_fragmented
++;
2619 /* Chop up message: */
2622 struct sk_buff
*fragm
;
2624 if (rest
<= fragm_sz
) {
2626 msg_set_type(&fragm_hdr
, LAST_FRAGMENT
);
2628 fragm
= buf_acquire(fragm_sz
+ INT_H_SIZE
);
2629 if (fragm
== NULL
) {
2630 warn("Link unable to fragment message\n");
2634 msg_set_size(&fragm_hdr
, fragm_sz
+ INT_H_SIZE
);
2635 skb_copy_to_linear_data(fragm
, &fragm_hdr
, INT_H_SIZE
);
2636 skb_copy_to_linear_data_offset(fragm
, INT_H_SIZE
, crs
,
2638 /* Send queued messages first, if any: */
2640 l_ptr
->stats
.sent_fragments
++;
2641 tipc_link_send_buf(l_ptr
, fragm
);
2642 if (!tipc_link_is_up(l_ptr
))
2644 msg_set_fragm_no(&fragm_hdr
, ++fragm_no
);
2647 msg_set_type(&fragm_hdr
, FRAGMENT
);
2655 * A pending message being re-assembled must store certain values
2656 * to handle subsequent fragments correctly. The following functions
2657 * help storing these values in unused, available fields in the
2658 * pending message. This makes dynamic memory allocation unecessary.
2661 static void set_long_msg_seqno(struct sk_buff
*buf
, u32 seqno
)
2663 msg_set_seqno(buf_msg(buf
), seqno
);
2666 static u32
get_fragm_size(struct sk_buff
*buf
)
2668 return msg_ack(buf_msg(buf
));
2671 static void set_fragm_size(struct sk_buff
*buf
, u32 sz
)
2673 msg_set_ack(buf_msg(buf
), sz
);
2676 static u32
get_expected_frags(struct sk_buff
*buf
)
2678 return msg_bcast_ack(buf_msg(buf
));
2681 static void set_expected_frags(struct sk_buff
*buf
, u32 exp
)
2683 msg_set_bcast_ack(buf_msg(buf
), exp
);
2686 static u32
get_timer_cnt(struct sk_buff
*buf
)
2688 return msg_reroute_cnt(buf_msg(buf
));
2691 static void incr_timer_cnt(struct sk_buff
*buf
)
2693 msg_incr_reroute_cnt(buf_msg(buf
));
2697 * tipc_link_recv_fragment(): Called with node lock on. Returns
2698 * the reassembled buffer if message is complete.
2700 int tipc_link_recv_fragment(struct sk_buff
**pending
, struct sk_buff
**fb
,
2701 struct tipc_msg
**m
)
2703 struct sk_buff
*prev
= NULL
;
2704 struct sk_buff
*fbuf
= *fb
;
2705 struct tipc_msg
*fragm
= buf_msg(fbuf
);
2706 struct sk_buff
*pbuf
= *pending
;
2707 u32 long_msg_seq_no
= msg_long_msgno(fragm
);
2710 msg_dbg(fragm
,"FRG<REC<");
2712 /* Is there an incomplete message waiting for this fragment? */
2714 while (pbuf
&& ((msg_seqno(buf_msg(pbuf
)) != long_msg_seq_no
) ||
2715 (msg_orignode(fragm
) != msg_orignode(buf_msg(pbuf
))))) {
2720 if (!pbuf
&& (msg_type(fragm
) == FIRST_FRAGMENT
)) {
2721 struct tipc_msg
*imsg
= (struct tipc_msg
*)msg_data(fragm
);
2722 u32 msg_sz
= msg_size(imsg
);
2723 u32 fragm_sz
= msg_data_sz(fragm
);
2724 u32 exp_fragm_cnt
= msg_sz
/fragm_sz
+ !!(msg_sz
% fragm_sz
);
2725 u32 max
= TIPC_MAX_USER_MSG_SIZE
+ LONG_H_SIZE
;
2726 if (msg_type(imsg
) == TIPC_MCAST_MSG
)
2727 max
= TIPC_MAX_USER_MSG_SIZE
+ MCAST_H_SIZE
;
2728 if (msg_size(imsg
) > max
) {
2729 msg_dbg(fragm
,"<REC<Oversized: ");
2733 pbuf
= buf_acquire(msg_size(imsg
));
2735 pbuf
->next
= *pending
;
2737 skb_copy_to_linear_data(pbuf
, imsg
,
2738 msg_data_sz(fragm
));
2739 /* Prepare buffer for subsequent fragments. */
2741 set_long_msg_seqno(pbuf
, long_msg_seq_no
);
2742 set_fragm_size(pbuf
,fragm_sz
);
2743 set_expected_frags(pbuf
,exp_fragm_cnt
- 1);
2745 warn("Link unable to reassemble fragmented message\n");
2749 } else if (pbuf
&& (msg_type(fragm
) != FIRST_FRAGMENT
)) {
2750 u32 dsz
= msg_data_sz(fragm
);
2751 u32 fsz
= get_fragm_size(pbuf
);
2752 u32 crs
= ((msg_fragm_no(fragm
) - 1) * fsz
);
2753 u32 exp_frags
= get_expected_frags(pbuf
) - 1;
2754 skb_copy_to_linear_data_offset(pbuf
, crs
,
2755 msg_data(fragm
), dsz
);
2758 /* Is message complete? */
2760 if (exp_frags
== 0) {
2762 prev
->next
= pbuf
->next
;
2764 *pending
= pbuf
->next
;
2765 msg_reset_reroute_cnt(buf_msg(pbuf
));
2770 set_expected_frags(pbuf
,exp_frags
);
2773 dbg(" Discarding orphan fragment %x\n",fbuf
);
2774 msg_dbg(fragm
,"ORPHAN:");
2775 dbg("Pending long buffers:\n");
2776 dbg_print_buf_chain(*pending
);
2782 * link_check_defragm_bufs - flush stale incoming message fragments
2783 * @l_ptr: pointer to link
2786 static void link_check_defragm_bufs(struct link
*l_ptr
)
2788 struct sk_buff
*prev
= NULL
;
2789 struct sk_buff
*next
= NULL
;
2790 struct sk_buff
*buf
= l_ptr
->defragm_buf
;
2794 if (!link_working_working(l_ptr
))
2797 u32 cnt
= get_timer_cnt(buf
);
2801 incr_timer_cnt(buf
);
2804 dbg(" Discarding incomplete long buffer\n");
2805 msg_dbg(buf_msg(buf
), "LONG:");
2806 dbg_print_link(l_ptr
, "curr:");
2807 dbg("Pending long buffers:\n");
2808 dbg_print_buf_chain(l_ptr
->defragm_buf
);
2810 prev
->next
= buf
->next
;
2812 l_ptr
->defragm_buf
= buf
->next
;
2821 static void link_set_supervision_props(struct link
*l_ptr
, u32 tolerance
)
2823 l_ptr
->tolerance
= tolerance
;
2824 l_ptr
->continuity_interval
=
2825 ((tolerance
/ 4) > 500) ? 500 : tolerance
/ 4;
2826 l_ptr
->abort_limit
= tolerance
/ (l_ptr
->continuity_interval
/ 4);
2830 void tipc_link_set_queue_limits(struct link
*l_ptr
, u32 window
)
2832 /* Data messages from this node, inclusive FIRST_FRAGM */
2833 l_ptr
->queue_limit
[TIPC_LOW_IMPORTANCE
] = window
;
2834 l_ptr
->queue_limit
[TIPC_MEDIUM_IMPORTANCE
] = (window
/ 3) * 4;
2835 l_ptr
->queue_limit
[TIPC_HIGH_IMPORTANCE
] = (window
/ 3) * 5;
2836 l_ptr
->queue_limit
[TIPC_CRITICAL_IMPORTANCE
] = (window
/ 3) * 6;
2837 /* Transiting data messages,inclusive FIRST_FRAGM */
2838 l_ptr
->queue_limit
[TIPC_LOW_IMPORTANCE
+ 4] = 300;
2839 l_ptr
->queue_limit
[TIPC_MEDIUM_IMPORTANCE
+ 4] = 600;
2840 l_ptr
->queue_limit
[TIPC_HIGH_IMPORTANCE
+ 4] = 900;
2841 l_ptr
->queue_limit
[TIPC_CRITICAL_IMPORTANCE
+ 4] = 1200;
2842 l_ptr
->queue_limit
[CONN_MANAGER
] = 1200;
2843 l_ptr
->queue_limit
[ROUTE_DISTRIBUTOR
] = 1200;
2844 l_ptr
->queue_limit
[CHANGEOVER_PROTOCOL
] = 2500;
2845 l_ptr
->queue_limit
[NAME_DISTRIBUTOR
] = 3000;
2846 /* FRAGMENT and LAST_FRAGMENT packets */
2847 l_ptr
->queue_limit
[MSG_FRAGMENTER
] = 4000;
2851 * link_find_link - locate link by name
2852 * @name - ptr to link name string
2853 * @node - ptr to area to be filled with ptr to associated node
2855 * Caller must hold 'tipc_net_lock' to ensure node and bearer are not deleted;
2856 * this also prevents link deletion.
2858 * Returns pointer to link (or 0 if invalid link name).
2861 static struct link
*link_find_link(const char *name
, struct tipc_node
**node
)
2863 struct link_name link_name_parts
;
2864 struct bearer
*b_ptr
;
2867 if (!link_name_validate(name
, &link_name_parts
))
2870 b_ptr
= tipc_bearer_find_interface(link_name_parts
.if_local
);
2874 *node
= tipc_node_find(link_name_parts
.addr_peer
);
2878 l_ptr
= (*node
)->links
[b_ptr
->identity
];
2879 if (!l_ptr
|| strcmp(l_ptr
->name
, name
))
2885 struct sk_buff
*tipc_link_cmd_config(const void *req_tlv_area
, int req_tlv_space
,
2888 struct tipc_link_config
*args
;
2891 struct tipc_node
*node
;
2894 if (!TLV_CHECK(req_tlv_area
, req_tlv_space
, TIPC_TLV_LINK_CONFIG
))
2895 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR
);
2897 args
= (struct tipc_link_config
*)TLV_DATA(req_tlv_area
);
2898 new_value
= ntohl(args
->value
);
2900 if (!strcmp(args
->name
, tipc_bclink_name
)) {
2901 if ((cmd
== TIPC_CMD_SET_LINK_WINDOW
) &&
2902 (tipc_bclink_set_queue_limits(new_value
) == 0))
2903 return tipc_cfg_reply_none();
2904 return tipc_cfg_reply_error_string(TIPC_CFG_NOT_SUPPORTED
2905 " (cannot change setting on broadcast link)");
2908 read_lock_bh(&tipc_net_lock
);
2909 l_ptr
= link_find_link(args
->name
, &node
);
2911 read_unlock_bh(&tipc_net_lock
);
2912 return tipc_cfg_reply_error_string("link not found");
2915 tipc_node_lock(node
);
2918 case TIPC_CMD_SET_LINK_TOL
:
2919 if ((new_value
>= TIPC_MIN_LINK_TOL
) &&
2920 (new_value
<= TIPC_MAX_LINK_TOL
)) {
2921 link_set_supervision_props(l_ptr
, new_value
);
2922 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
2923 0, 0, new_value
, 0, 0);
2927 case TIPC_CMD_SET_LINK_PRI
:
2928 if ((new_value
>= TIPC_MIN_LINK_PRI
) &&
2929 (new_value
<= TIPC_MAX_LINK_PRI
)) {
2930 l_ptr
->priority
= new_value
;
2931 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
2932 0, 0, 0, new_value
, 0);
2936 case TIPC_CMD_SET_LINK_WINDOW
:
2937 if ((new_value
>= TIPC_MIN_LINK_WIN
) &&
2938 (new_value
<= TIPC_MAX_LINK_WIN
)) {
2939 tipc_link_set_queue_limits(l_ptr
, new_value
);
2944 tipc_node_unlock(node
);
2946 read_unlock_bh(&tipc_net_lock
);
2948 return tipc_cfg_reply_error_string("cannot change link setting");
2950 return tipc_cfg_reply_none();
2954 * link_reset_statistics - reset link statistics
2955 * @l_ptr: pointer to link
2958 static void link_reset_statistics(struct link
*l_ptr
)
2960 memset(&l_ptr
->stats
, 0, sizeof(l_ptr
->stats
));
2961 l_ptr
->stats
.sent_info
= l_ptr
->next_out_no
;
2962 l_ptr
->stats
.recv_info
= l_ptr
->next_in_no
;
2965 struct sk_buff
*tipc_link_cmd_reset_stats(const void *req_tlv_area
, int req_tlv_space
)
2969 struct tipc_node
*node
;
2971 if (!TLV_CHECK(req_tlv_area
, req_tlv_space
, TIPC_TLV_LINK_NAME
))
2972 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR
);
2974 link_name
= (char *)TLV_DATA(req_tlv_area
);
2975 if (!strcmp(link_name
, tipc_bclink_name
)) {
2976 if (tipc_bclink_reset_stats())
2977 return tipc_cfg_reply_error_string("link not found");
2978 return tipc_cfg_reply_none();
2981 read_lock_bh(&tipc_net_lock
);
2982 l_ptr
= link_find_link(link_name
, &node
);
2984 read_unlock_bh(&tipc_net_lock
);
2985 return tipc_cfg_reply_error_string("link not found");
2988 tipc_node_lock(node
);
2989 link_reset_statistics(l_ptr
);
2990 tipc_node_unlock(node
);
2991 read_unlock_bh(&tipc_net_lock
);
2992 return tipc_cfg_reply_none();
2996 * percent - convert count to a percentage of total (rounding up or down)
2999 static u32
percent(u32 count
, u32 total
)
3001 return (count
* 100 + (total
/ 2)) / total
;
3005 * tipc_link_stats - print link statistics
3007 * @buf: print buffer area
3008 * @buf_size: size of print buffer area
3010 * Returns length of print buffer data string (or 0 if error)
3013 static int tipc_link_stats(const char *name
, char *buf
, const u32 buf_size
)
3015 struct print_buf pb
;
3017 struct tipc_node
*node
;
3019 u32 profile_total
= 0;
3021 if (!strcmp(name
, tipc_bclink_name
))
3022 return tipc_bclink_stats(buf
, buf_size
);
3024 tipc_printbuf_init(&pb
, buf
, buf_size
);
3026 read_lock_bh(&tipc_net_lock
);
3027 l_ptr
= link_find_link(name
, &node
);
3029 read_unlock_bh(&tipc_net_lock
);
3032 tipc_node_lock(node
);
3034 if (tipc_link_is_active(l_ptr
))
3036 else if (tipc_link_is_up(l_ptr
))
3040 tipc_printf(&pb
, "Link <%s>\n"
3041 " %s MTU:%u Priority:%u Tolerance:%u ms"
3042 " Window:%u packets\n",
3043 l_ptr
->name
, status
, l_ptr
->max_pkt
,
3044 l_ptr
->priority
, l_ptr
->tolerance
, l_ptr
->queue_limit
[0]);
3045 tipc_printf(&pb
, " RX packets:%u fragments:%u/%u bundles:%u/%u\n",
3046 l_ptr
->next_in_no
- l_ptr
->stats
.recv_info
,
3047 l_ptr
->stats
.recv_fragments
,
3048 l_ptr
->stats
.recv_fragmented
,
3049 l_ptr
->stats
.recv_bundles
,
3050 l_ptr
->stats
.recv_bundled
);
3051 tipc_printf(&pb
, " TX packets:%u fragments:%u/%u bundles:%u/%u\n",
3052 l_ptr
->next_out_no
- l_ptr
->stats
.sent_info
,
3053 l_ptr
->stats
.sent_fragments
,
3054 l_ptr
->stats
.sent_fragmented
,
3055 l_ptr
->stats
.sent_bundles
,
3056 l_ptr
->stats
.sent_bundled
);
3057 profile_total
= l_ptr
->stats
.msg_length_counts
;
3060 tipc_printf(&pb
, " TX profile sample:%u packets average:%u octets\n"
3061 " 0-64:%u%% -256:%u%% -1024:%u%% -4096:%u%% "
3062 "-16354:%u%% -32768:%u%% -66000:%u%%\n",
3063 l_ptr
->stats
.msg_length_counts
,
3064 l_ptr
->stats
.msg_lengths_total
/ profile_total
,
3065 percent(l_ptr
->stats
.msg_length_profile
[0], profile_total
),
3066 percent(l_ptr
->stats
.msg_length_profile
[1], profile_total
),
3067 percent(l_ptr
->stats
.msg_length_profile
[2], profile_total
),
3068 percent(l_ptr
->stats
.msg_length_profile
[3], profile_total
),
3069 percent(l_ptr
->stats
.msg_length_profile
[4], profile_total
),
3070 percent(l_ptr
->stats
.msg_length_profile
[5], profile_total
),
3071 percent(l_ptr
->stats
.msg_length_profile
[6], profile_total
));
3072 tipc_printf(&pb
, " RX states:%u probes:%u naks:%u defs:%u dups:%u\n",
3073 l_ptr
->stats
.recv_states
,
3074 l_ptr
->stats
.recv_probes
,
3075 l_ptr
->stats
.recv_nacks
,
3076 l_ptr
->stats
.deferred_recv
,
3077 l_ptr
->stats
.duplicates
);
3078 tipc_printf(&pb
, " TX states:%u probes:%u naks:%u acks:%u dups:%u\n",
3079 l_ptr
->stats
.sent_states
,
3080 l_ptr
->stats
.sent_probes
,
3081 l_ptr
->stats
.sent_nacks
,
3082 l_ptr
->stats
.sent_acks
,
3083 l_ptr
->stats
.retransmitted
);
3084 tipc_printf(&pb
, " Congestion bearer:%u link:%u Send queue max:%u avg:%u\n",
3085 l_ptr
->stats
.bearer_congs
,
3086 l_ptr
->stats
.link_congs
,
3087 l_ptr
->stats
.max_queue_sz
,
3088 l_ptr
->stats
.queue_sz_counts
3089 ? (l_ptr
->stats
.accu_queue_sz
/ l_ptr
->stats
.queue_sz_counts
)
3092 tipc_node_unlock(node
);
3093 read_unlock_bh(&tipc_net_lock
);
3094 return tipc_printbuf_validate(&pb
);
3097 #define MAX_LINK_STATS_INFO 2000
3099 struct sk_buff
*tipc_link_cmd_show_stats(const void *req_tlv_area
, int req_tlv_space
)
3101 struct sk_buff
*buf
;
3102 struct tlv_desc
*rep_tlv
;
3105 if (!TLV_CHECK(req_tlv_area
, req_tlv_space
, TIPC_TLV_LINK_NAME
))
3106 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR
);
3108 buf
= tipc_cfg_reply_alloc(TLV_SPACE(MAX_LINK_STATS_INFO
));
3112 rep_tlv
= (struct tlv_desc
*)buf
->data
;
3114 str_len
= tipc_link_stats((char *)TLV_DATA(req_tlv_area
),
3115 (char *)TLV_DATA(rep_tlv
), MAX_LINK_STATS_INFO
);
3118 return tipc_cfg_reply_error_string("link not found");
3121 skb_put(buf
, TLV_SPACE(str_len
));
3122 TLV_SET(rep_tlv
, TIPC_TLV_ULTRA_STRING
, NULL
, str_len
);
3129 * tipc_link_get_max_pkt - get maximum packet size to use when sending to destination
3130 * @dest: network address of destination node
3131 * @selector: used to select from set of active links
3133 * If no active link can be found, uses default maximum packet size.
3136 u32
tipc_link_get_max_pkt(u32 dest
, u32 selector
)
3138 struct tipc_node
*n_ptr
;
3140 u32 res
= MAX_PKT_DEFAULT
;
3142 if (dest
== tipc_own_addr
)
3143 return MAX_MSG_SIZE
;
3145 read_lock_bh(&tipc_net_lock
);
3146 n_ptr
= tipc_node_select(dest
, selector
);
3148 tipc_node_lock(n_ptr
);
3149 l_ptr
= n_ptr
->active_links
[selector
& 1];
3151 res
= l_ptr
->max_pkt
;
3152 tipc_node_unlock(n_ptr
);
3154 read_unlock_bh(&tipc_net_lock
);
3159 static void link_dump_send_queue(struct link
*l_ptr
)
3161 if (l_ptr
->next_out
) {
3162 info("\nContents of unsent queue:\n");
3163 dbg_print_buf_chain(l_ptr
->next_out
);
3165 info("\nContents of send queue:\n");
3166 if (l_ptr
->first_out
) {
3167 dbg_print_buf_chain(l_ptr
->first_out
);
3169 info("Empty send queue\n");
3172 static void link_print(struct link
*l_ptr
, struct print_buf
*buf
,
3175 tipc_printf(buf
, str
);
3176 if (link_reset_reset(l_ptr
) || link_reset_unknown(l_ptr
))
3178 tipc_printf(buf
, "Link %x<%s>:",
3179 l_ptr
->addr
, l_ptr
->b_ptr
->publ
.name
);
3180 tipc_printf(buf
, ": NXO(%u):", mod(l_ptr
->next_out_no
));
3181 tipc_printf(buf
, "NXI(%u):", mod(l_ptr
->next_in_no
));
3182 tipc_printf(buf
, "SQUE");
3183 if (l_ptr
->first_out
) {
3184 tipc_printf(buf
, "[%u..", msg_seqno(buf_msg(l_ptr
->first_out
)));
3185 if (l_ptr
->next_out
)
3186 tipc_printf(buf
, "%u..",
3187 msg_seqno(buf_msg(l_ptr
->next_out
)));
3188 tipc_printf(buf
, "%u]", msg_seqno(buf_msg(l_ptr
->last_out
)));
3189 if ((mod(msg_seqno(buf_msg(l_ptr
->last_out
)) -
3190 msg_seqno(buf_msg(l_ptr
->first_out
)))
3191 != (l_ptr
->out_queue_size
- 1)) ||
3192 (l_ptr
->last_out
->next
!= NULL
)) {
3193 tipc_printf(buf
, "\nSend queue inconsistency\n");
3194 tipc_printf(buf
, "first_out= %x ", l_ptr
->first_out
);
3195 tipc_printf(buf
, "next_out= %x ", l_ptr
->next_out
);
3196 tipc_printf(buf
, "last_out= %x ", l_ptr
->last_out
);
3197 link_dump_send_queue(l_ptr
);
3200 tipc_printf(buf
, "[]");
3201 tipc_printf(buf
, "SQSIZ(%u)", l_ptr
->out_queue_size
);
3202 if (l_ptr
->oldest_deferred_in
) {
3203 u32 o
= msg_seqno(buf_msg(l_ptr
->oldest_deferred_in
));
3204 u32 n
= msg_seqno(buf_msg(l_ptr
->newest_deferred_in
));
3205 tipc_printf(buf
, ":RQUE[%u..%u]", o
, n
);
3206 if (l_ptr
->deferred_inqueue_sz
!= mod((n
+ 1) - o
)) {
3207 tipc_printf(buf
, ":RQSIZ(%u)",
3208 l_ptr
->deferred_inqueue_sz
);
3211 if (link_working_unknown(l_ptr
))
3212 tipc_printf(buf
, ":WU");
3213 if (link_reset_reset(l_ptr
))
3214 tipc_printf(buf
, ":RR");
3215 if (link_reset_unknown(l_ptr
))
3216 tipc_printf(buf
, ":RU");
3217 if (link_working_working(l_ptr
))
3218 tipc_printf(buf
, ":WW");
3219 tipc_printf(buf
, "\n");