2 * net/tipc/link.c: TIPC link code
4 * Copyright (c) 1996-2006, Ericsson AB
5 * Copyright (c) 2004-2005, 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 * Limit for deferred reception queue:
57 #define DEF_QUEUE_LIMIT 256u
63 #define STARTING_EVT 856384768 /* link processing trigger */
64 #define TRAFFIC_MSG_EVT 560815u /* rx'd ??? */
65 #define TIMEOUT_EVT 560817u /* link timer expired */
68 * The following two 'message types' is really just implementation
69 * data conveniently stored in the message header.
70 * They must not be considered part of the protocol
76 * State value stored in 'exp_msg_count'
79 #define START_CHANGEOVER 100000u
82 * struct link_name - deconstructed link name
83 * @addr_local: network address of node at this end
84 * @if_local: name of interface at this end
85 * @addr_peer: network address of node at far end
86 * @if_peer: name of interface at far end
91 char if_local
[TIPC_MAX_IF_NAME
];
93 char if_peer
[TIPC_MAX_IF_NAME
];
98 /* LINK EVENT CODE IS NOT SUPPORTED AT PRESENT */
101 * struct link_event - link up/down event notification
107 void (*fcn
)(u32
, char *, int);
108 char name
[TIPC_MAX_LINK_NAME
];
113 static void link_handle_out_of_seq_msg(struct link
*l_ptr
,
114 struct sk_buff
*buf
);
115 static void link_recv_proto_msg(struct link
*l_ptr
, struct sk_buff
*buf
);
116 static int link_recv_changeover_msg(struct link
**l_ptr
, struct sk_buff
**buf
);
117 static void link_set_supervision_props(struct link
*l_ptr
, u32 tolerance
);
118 static int link_send_sections_long(struct port
*sender
,
119 struct iovec
const *msg_sect
,
120 u32 num_sect
, u32 destnode
);
121 static void link_check_defragm_bufs(struct link
*l_ptr
);
122 static void link_state_event(struct link
*l_ptr
, u32 event
);
123 static void link_reset_statistics(struct link
*l_ptr
);
124 static void link_print(struct link
*l_ptr
, struct print_buf
*buf
,
128 * Debugging code used by link routines only
130 * When debugging link problems on a system that has multiple links,
131 * the standard TIPC debugging routines may not be useful since they
132 * allow the output from multiple links to be intermixed. For this reason
133 * routines of the form "dbg_link_XXX()" have been created that will capture
134 * debug info into a link's personal print buffer, which can then be dumped
135 * into the TIPC system log (LOG) upon request.
137 * To enable per-link debugging, use LINK_LOG_BUF_SIZE to specify the size
138 * of the print buffer used by each link. If LINK_LOG_BUF_SIZE is set to 0,
139 * the dbg_link_XXX() routines simply send their output to the standard
140 * debug print buffer (DBG_OUTPUT), if it has been defined; this can be useful
141 * when there is only a single link in the system being debugged.
144 * - When enabled, LINK_LOG_BUF_SIZE should be set to at least 1000 (bytes)
145 * - "l_ptr" must be valid when using dbg_link_XXX() macros
148 #define LINK_LOG_BUF_SIZE 0
150 #define dbg_link(fmt, arg...) do {if (LINK_LOG_BUF_SIZE) tipc_printf(&l_ptr->print_buf, fmt, ## arg); } while(0)
151 #define dbg_link_msg(msg, txt) do {if (LINK_LOG_BUF_SIZE) tipc_msg_print(&l_ptr->print_buf, msg, txt); } while(0)
152 #define dbg_link_state(txt) do {if (LINK_LOG_BUF_SIZE) link_print(l_ptr, &l_ptr->print_buf, txt); } while(0)
153 #define dbg_link_dump() do { \
154 if (LINK_LOG_BUF_SIZE) { \
155 tipc_printf(LOG, "\n\nDumping link <%s>:\n", l_ptr->name); \
156 tipc_printbuf_move(LOG, &l_ptr->print_buf); \
160 static void dbg_print_link(struct link
*l_ptr
, const char *str
)
163 link_print(l_ptr
, DBG_OUTPUT
, str
);
166 static void dbg_print_buf_chain(struct sk_buff
*root_buf
)
169 struct sk_buff
*buf
= root_buf
;
172 msg_dbg(buf_msg(buf
), "In chain: ");
179 * Simple link routines
182 static unsigned int align(unsigned int i
)
184 return (i
+ 3) & ~3u;
187 static int link_working_working(struct link
*l_ptr
)
189 return (l_ptr
->state
== WORKING_WORKING
);
192 static int link_working_unknown(struct link
*l_ptr
)
194 return (l_ptr
->state
== WORKING_UNKNOWN
);
197 static int link_reset_unknown(struct link
*l_ptr
)
199 return (l_ptr
->state
== RESET_UNKNOWN
);
202 static int link_reset_reset(struct link
*l_ptr
)
204 return (l_ptr
->state
== RESET_RESET
);
207 static int link_blocked(struct link
*l_ptr
)
209 return (l_ptr
->exp_msg_count
|| l_ptr
->blocked
);
212 static int link_congested(struct link
*l_ptr
)
214 return (l_ptr
->out_queue_size
>= l_ptr
->queue_limit
[0]);
217 static u32
link_max_pkt(struct link
*l_ptr
)
219 return l_ptr
->max_pkt
;
222 static void link_init_max_pkt(struct link
*l_ptr
)
226 max_pkt
= (l_ptr
->b_ptr
->publ
.mtu
& ~3);
227 if (max_pkt
> MAX_MSG_SIZE
)
228 max_pkt
= MAX_MSG_SIZE
;
230 l_ptr
->max_pkt_target
= max_pkt
;
231 if (l_ptr
->max_pkt_target
< MAX_PKT_DEFAULT
)
232 l_ptr
->max_pkt
= l_ptr
->max_pkt_target
;
234 l_ptr
->max_pkt
= MAX_PKT_DEFAULT
;
236 l_ptr
->max_pkt_probes
= 0;
239 static u32
link_next_sent(struct link
*l_ptr
)
242 return msg_seqno(buf_msg(l_ptr
->next_out
));
243 return mod(l_ptr
->next_out_no
);
246 static u32
link_last_sent(struct link
*l_ptr
)
248 return mod(link_next_sent(l_ptr
) - 1);
252 * Simple non-static link routines (i.e. referenced outside this file)
255 int tipc_link_is_up(struct link
*l_ptr
)
259 return (link_working_working(l_ptr
) || link_working_unknown(l_ptr
));
262 int tipc_link_is_active(struct link
*l_ptr
)
264 return ((l_ptr
->owner
->active_links
[0] == l_ptr
) ||
265 (l_ptr
->owner
->active_links
[1] == l_ptr
));
269 * link_name_validate - validate & (optionally) deconstruct link name
270 * @name - ptr to link name string
271 * @name_parts - ptr to area for link name components (or NULL if not needed)
273 * Returns 1 if link name is valid, otherwise 0.
276 static int link_name_validate(const char *name
, struct link_name
*name_parts
)
278 char name_copy
[TIPC_MAX_LINK_NAME
];
284 u32 z_local
, c_local
, n_local
;
285 u32 z_peer
, c_peer
, n_peer
;
289 /* copy link name & ensure length is OK */
291 name_copy
[TIPC_MAX_LINK_NAME
- 1] = 0;
292 /* need above in case non-Posix strncpy() doesn't pad with nulls */
293 strncpy(name_copy
, name
, TIPC_MAX_LINK_NAME
);
294 if (name_copy
[TIPC_MAX_LINK_NAME
- 1] != 0)
297 /* ensure all component parts of link name are present */
299 addr_local
= name_copy
;
300 if ((if_local
= strchr(addr_local
, ':')) == NULL
)
303 if ((addr_peer
= strchr(if_local
, '-')) == NULL
)
306 if_local_len
= addr_peer
- if_local
;
307 if ((if_peer
= strchr(addr_peer
, ':')) == NULL
)
310 if_peer_len
= strlen(if_peer
) + 1;
312 /* validate component parts of link name */
314 if ((sscanf(addr_local
, "%u.%u.%u%c",
315 &z_local
, &c_local
, &n_local
, &dummy
) != 3) ||
316 (sscanf(addr_peer
, "%u.%u.%u%c",
317 &z_peer
, &c_peer
, &n_peer
, &dummy
) != 3) ||
318 (z_local
> 255) || (c_local
> 4095) || (n_local
> 4095) ||
319 (z_peer
> 255) || (c_peer
> 4095) || (n_peer
> 4095) ||
320 (if_local_len
<= 1) || (if_local_len
> TIPC_MAX_IF_NAME
) ||
321 (if_peer_len
<= 1) || (if_peer_len
> TIPC_MAX_IF_NAME
) ||
322 (strspn(if_local
, tipc_alphabet
) != (if_local_len
- 1)) ||
323 (strspn(if_peer
, tipc_alphabet
) != (if_peer_len
- 1)))
326 /* return link name components, if necessary */
329 name_parts
->addr_local
= tipc_addr(z_local
, c_local
, n_local
);
330 strcpy(name_parts
->if_local
, if_local
);
331 name_parts
->addr_peer
= tipc_addr(z_peer
, c_peer
, n_peer
);
332 strcpy(name_parts
->if_peer
, if_peer
);
338 * link_timeout - handle expiration of link timer
339 * @l_ptr: pointer to link
341 * This routine must not grab "tipc_net_lock" to avoid a potential deadlock conflict
342 * with tipc_link_delete(). (There is no risk that the node will be deleted by
343 * another thread because tipc_link_delete() always cancels the link timer before
344 * tipc_node_delete() is called.)
347 static void link_timeout(struct link
*l_ptr
)
349 tipc_node_lock(l_ptr
->owner
);
351 /* update counters used in statistical profiling of send traffic */
353 l_ptr
->stats
.accu_queue_sz
+= l_ptr
->out_queue_size
;
354 l_ptr
->stats
.queue_sz_counts
++;
356 if (l_ptr
->out_queue_size
> l_ptr
->stats
.max_queue_sz
)
357 l_ptr
->stats
.max_queue_sz
= l_ptr
->out_queue_size
;
359 if (l_ptr
->first_out
) {
360 struct tipc_msg
*msg
= buf_msg(l_ptr
->first_out
);
361 u32 length
= msg_size(msg
);
363 if ((msg_user(msg
) == MSG_FRAGMENTER
)
364 && (msg_type(msg
) == FIRST_FRAGMENT
)) {
365 length
= msg_size(msg_get_wrapped(msg
));
368 l_ptr
->stats
.msg_lengths_total
+= length
;
369 l_ptr
->stats
.msg_length_counts
++;
371 l_ptr
->stats
.msg_length_profile
[0]++;
372 else if (length
<= 256)
373 l_ptr
->stats
.msg_length_profile
[1]++;
374 else if (length
<= 1024)
375 l_ptr
->stats
.msg_length_profile
[2]++;
376 else if (length
<= 4096)
377 l_ptr
->stats
.msg_length_profile
[3]++;
378 else if (length
<= 16384)
379 l_ptr
->stats
.msg_length_profile
[4]++;
380 else if (length
<= 32768)
381 l_ptr
->stats
.msg_length_profile
[5]++;
383 l_ptr
->stats
.msg_length_profile
[6]++;
387 /* do all other link processing performed on a periodic basis */
389 link_check_defragm_bufs(l_ptr
);
391 link_state_event(l_ptr
, TIMEOUT_EVT
);
394 tipc_link_push_queue(l_ptr
);
396 tipc_node_unlock(l_ptr
->owner
);
399 static void link_set_timer(struct link
*l_ptr
, u32 time
)
401 k_start_timer(&l_ptr
->timer
, time
);
405 * tipc_link_create - create a new link
406 * @b_ptr: pointer to associated bearer
407 * @peer: network address of node at other end of link
408 * @media_addr: media address to use when sending messages over link
410 * Returns pointer to link.
413 struct link
*tipc_link_create(struct bearer
*b_ptr
, const u32 peer
,
414 const struct tipc_media_addr
*media_addr
)
417 struct tipc_msg
*msg
;
420 l_ptr
= (struct link
*)kmalloc(sizeof(*l_ptr
), GFP_ATOMIC
);
422 warn("Link creation failed, no memory\n");
425 memset(l_ptr
, 0, sizeof(*l_ptr
));
428 if_name
= strchr(b_ptr
->publ
.name
, ':') + 1;
429 sprintf(l_ptr
->name
, "%u.%u.%u:%s-%u.%u.%u:",
430 tipc_zone(tipc_own_addr
), tipc_cluster(tipc_own_addr
),
431 tipc_node(tipc_own_addr
),
433 tipc_zone(peer
), tipc_cluster(peer
), tipc_node(peer
));
434 /* note: peer i/f is appended to link name by reset/activate */
435 memcpy(&l_ptr
->media_addr
, media_addr
, sizeof(*media_addr
));
436 k_init_timer(&l_ptr
->timer
, (Handler
)link_timeout
, (unsigned long)l_ptr
);
437 list_add_tail(&l_ptr
->link_list
, &b_ptr
->links
);
438 l_ptr
->checkpoint
= 1;
439 l_ptr
->b_ptr
= b_ptr
;
440 link_set_supervision_props(l_ptr
, b_ptr
->media
->tolerance
);
441 l_ptr
->state
= RESET_UNKNOWN
;
443 l_ptr
->pmsg
= (struct tipc_msg
*)&l_ptr
->proto_msg
;
445 msg_init(msg
, LINK_PROTOCOL
, RESET_MSG
, TIPC_OK
, INT_H_SIZE
, l_ptr
->addr
);
446 msg_set_size(msg
, sizeof(l_ptr
->proto_msg
));
447 msg_set_session(msg
, tipc_random
);
448 msg_set_bearer_id(msg
, b_ptr
->identity
);
449 strcpy((char *)msg_data(msg
), if_name
);
451 l_ptr
->priority
= b_ptr
->priority
;
452 tipc_link_set_queue_limits(l_ptr
, b_ptr
->media
->window
);
454 link_init_max_pkt(l_ptr
);
456 l_ptr
->next_out_no
= 1;
457 INIT_LIST_HEAD(&l_ptr
->waiting_ports
);
459 link_reset_statistics(l_ptr
);
461 l_ptr
->owner
= tipc_node_attach_link(l_ptr
);
467 if (LINK_LOG_BUF_SIZE
) {
468 char *pb
= kmalloc(LINK_LOG_BUF_SIZE
, GFP_ATOMIC
);
472 warn("Link creation failed, no memory for print buffer\n");
475 tipc_printbuf_init(&l_ptr
->print_buf
, pb
, LINK_LOG_BUF_SIZE
);
478 tipc_k_signal((Handler
)tipc_link_start
, (unsigned long)l_ptr
);
480 dbg("tipc_link_create(): tolerance = %u,cont intv = %u, abort_limit = %u\n",
481 l_ptr
->tolerance
, l_ptr
->continuity_interval
, l_ptr
->abort_limit
);
487 * tipc_link_delete - delete a link
488 * @l_ptr: pointer to link
490 * Note: 'tipc_net_lock' is write_locked, bearer is locked.
491 * This routine must not grab the node lock until after link timer cancellation
492 * to avoid a potential deadlock situation.
495 void tipc_link_delete(struct link
*l_ptr
)
498 err("Attempt to delete non-existent link\n");
502 dbg("tipc_link_delete()\n");
504 k_cancel_timer(&l_ptr
->timer
);
506 tipc_node_lock(l_ptr
->owner
);
507 tipc_link_reset(l_ptr
);
508 tipc_node_detach_link(l_ptr
->owner
, l_ptr
);
509 tipc_link_stop(l_ptr
);
510 list_del_init(&l_ptr
->link_list
);
511 if (LINK_LOG_BUF_SIZE
)
512 kfree(l_ptr
->print_buf
.buf
);
513 tipc_node_unlock(l_ptr
->owner
);
514 k_term_timer(&l_ptr
->timer
);
518 void tipc_link_start(struct link
*l_ptr
)
520 dbg("tipc_link_start %x\n", l_ptr
);
521 link_state_event(l_ptr
, STARTING_EVT
);
525 * link_schedule_port - schedule port for deferred sending
526 * @l_ptr: pointer to link
527 * @origport: reference to sending port
528 * @sz: amount of data to be sent
530 * Schedules port for renewed sending of messages after link congestion
534 static int link_schedule_port(struct link
*l_ptr
, u32 origport
, u32 sz
)
538 spin_lock_bh(&tipc_port_list_lock
);
539 p_ptr
= tipc_port_lock(origport
);
543 if (!list_empty(&p_ptr
->wait_list
))
545 p_ptr
->congested_link
= l_ptr
;
546 p_ptr
->publ
.congested
= 1;
547 p_ptr
->waiting_pkts
= 1 + ((sz
- 1) / link_max_pkt(l_ptr
));
548 list_add_tail(&p_ptr
->wait_list
, &l_ptr
->waiting_ports
);
549 l_ptr
->stats
.link_congs
++;
551 tipc_port_unlock(p_ptr
);
553 spin_unlock_bh(&tipc_port_list_lock
);
557 void tipc_link_wakeup_ports(struct link
*l_ptr
, int all
)
560 struct port
*temp_p_ptr
;
561 int win
= l_ptr
->queue_limit
[0] - l_ptr
->out_queue_size
;
567 if (!spin_trylock_bh(&tipc_port_list_lock
))
569 if (link_congested(l_ptr
))
571 list_for_each_entry_safe(p_ptr
, temp_p_ptr
, &l_ptr
->waiting_ports
,
575 list_del_init(&p_ptr
->wait_list
);
576 p_ptr
->congested_link
= NULL
;
577 spin_lock_bh(p_ptr
->publ
.lock
);
578 p_ptr
->publ
.congested
= 0;
579 p_ptr
->wakeup(&p_ptr
->publ
);
580 win
-= p_ptr
->waiting_pkts
;
581 spin_unlock_bh(p_ptr
->publ
.lock
);
585 spin_unlock_bh(&tipc_port_list_lock
);
589 * link_release_outqueue - purge link's outbound message queue
590 * @l_ptr: pointer to link
593 static void link_release_outqueue(struct link
*l_ptr
)
595 struct sk_buff
*buf
= l_ptr
->first_out
;
596 struct sk_buff
*next
;
603 l_ptr
->first_out
= NULL
;
604 l_ptr
->out_queue_size
= 0;
608 * tipc_link_reset_fragments - purge link's inbound message fragments queue
609 * @l_ptr: pointer to link
612 void tipc_link_reset_fragments(struct link
*l_ptr
)
614 struct sk_buff
*buf
= l_ptr
->defragm_buf
;
615 struct sk_buff
*next
;
622 l_ptr
->defragm_buf
= NULL
;
626 * tipc_link_stop - purge all inbound and outbound messages associated with link
627 * @l_ptr: pointer to link
630 void tipc_link_stop(struct link
*l_ptr
)
633 struct sk_buff
*next
;
635 buf
= l_ptr
->oldest_deferred_in
;
642 buf
= l_ptr
->first_out
;
649 tipc_link_reset_fragments(l_ptr
);
651 buf_discard(l_ptr
->proto_msg_queue
);
652 l_ptr
->proto_msg_queue
= NULL
;
657 /* LINK EVENT CODE IS NOT SUPPORTED AT PRESENT */
659 static void link_recv_event(struct link_event
*ev
)
661 ev
->fcn(ev
->addr
, ev
->name
, ev
->up
);
665 static void link_send_event(void (*fcn
)(u32 a
, char *n
, int up
),
666 struct link
*l_ptr
, int up
)
668 struct link_event
*ev
;
670 ev
= kmalloc(sizeof(*ev
), GFP_ATOMIC
);
672 warn("Link event allocation failure\n");
675 ev
->addr
= l_ptr
->addr
;
678 memcpy(ev
->name
, l_ptr
->name
, TIPC_MAX_LINK_NAME
);
679 tipc_k_signal((Handler
)link_recv_event
, (unsigned long)ev
);
684 #define link_send_event(fcn, l_ptr, up) do { } while (0)
688 void tipc_link_reset(struct link
*l_ptr
)
691 u32 prev_state
= l_ptr
->state
;
692 u32 checkpoint
= l_ptr
->next_in_no
;
693 int was_active_link
= tipc_link_is_active(l_ptr
);
695 msg_set_session(l_ptr
->pmsg
, msg_session(l_ptr
->pmsg
) + 1);
697 /* Link is down, accept any session: */
698 l_ptr
->peer_session
= 0;
700 /* Prepare for max packet size negotiation */
701 link_init_max_pkt(l_ptr
);
703 l_ptr
->state
= RESET_UNKNOWN
;
704 dbg_link_state("Resetting Link\n");
706 if ((prev_state
== RESET_UNKNOWN
) || (prev_state
== RESET_RESET
))
709 tipc_node_link_down(l_ptr
->owner
, l_ptr
);
710 tipc_bearer_remove_dest(l_ptr
->b_ptr
, l_ptr
->addr
);
712 tipc_printf(TIPC_CONS
, "\nReset link <%s>\n", l_ptr
->name
);
715 if (was_active_link
&& tipc_node_has_active_links(l_ptr
->owner
) &&
716 l_ptr
->owner
->permit_changeover
) {
717 l_ptr
->reset_checkpoint
= checkpoint
;
718 l_ptr
->exp_msg_count
= START_CHANGEOVER
;
721 /* Clean up all queues: */
723 link_release_outqueue(l_ptr
);
724 buf_discard(l_ptr
->proto_msg_queue
);
725 l_ptr
->proto_msg_queue
= NULL
;
726 buf
= l_ptr
->oldest_deferred_in
;
728 struct sk_buff
*next
= buf
->next
;
732 if (!list_empty(&l_ptr
->waiting_ports
))
733 tipc_link_wakeup_ports(l_ptr
, 1);
735 l_ptr
->retransm_queue_head
= 0;
736 l_ptr
->retransm_queue_size
= 0;
737 l_ptr
->last_out
= NULL
;
738 l_ptr
->first_out
= NULL
;
739 l_ptr
->next_out
= NULL
;
740 l_ptr
->unacked_window
= 0;
741 l_ptr
->checkpoint
= 1;
742 l_ptr
->next_out_no
= 1;
743 l_ptr
->deferred_inqueue_sz
= 0;
744 l_ptr
->oldest_deferred_in
= NULL
;
745 l_ptr
->newest_deferred_in
= NULL
;
746 l_ptr
->fsm_msg_cnt
= 0;
747 l_ptr
->stale_count
= 0;
748 link_reset_statistics(l_ptr
);
750 link_send_event(tipc_cfg_link_event
, l_ptr
, 0);
751 if (!in_own_cluster(l_ptr
->addr
))
752 link_send_event(tipc_disc_link_event
, l_ptr
, 0);
756 static void link_activate(struct link
*l_ptr
)
758 l_ptr
->next_in_no
= l_ptr
->stats
.recv_info
= 1;
759 tipc_node_link_up(l_ptr
->owner
, l_ptr
);
760 tipc_bearer_add_dest(l_ptr
->b_ptr
, l_ptr
->addr
);
761 link_send_event(tipc_cfg_link_event
, l_ptr
, 1);
762 if (!in_own_cluster(l_ptr
->addr
))
763 link_send_event(tipc_disc_link_event
, l_ptr
, 1);
767 * link_state_event - link finite state machine
768 * @l_ptr: pointer to link
769 * @event: state machine event to process
772 static void link_state_event(struct link
*l_ptr
, unsigned event
)
775 u32 cont_intv
= l_ptr
->continuity_interval
;
777 if (!l_ptr
->started
&& (event
!= STARTING_EVT
))
778 return; /* Not yet. */
780 if (link_blocked(l_ptr
)) {
781 if (event
== TIMEOUT_EVT
) {
782 link_set_timer(l_ptr
, cont_intv
);
784 return; /* Changeover going on */
786 dbg_link("STATE_EV: <%s> ", l_ptr
->name
);
788 switch (l_ptr
->state
) {
789 case WORKING_WORKING
:
792 case TRAFFIC_MSG_EVT
:
800 if (l_ptr
->next_in_no
!= l_ptr
->checkpoint
) {
801 l_ptr
->checkpoint
= l_ptr
->next_in_no
;
802 if (tipc_bclink_acks_missing(l_ptr
->owner
)) {
803 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
805 l_ptr
->fsm_msg_cnt
++;
806 } else if (l_ptr
->max_pkt
< l_ptr
->max_pkt_target
) {
807 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
809 l_ptr
->fsm_msg_cnt
++;
811 link_set_timer(l_ptr
, cont_intv
);
814 dbg_link(" -> WU\n");
815 l_ptr
->state
= WORKING_UNKNOWN
;
816 l_ptr
->fsm_msg_cnt
= 0;
817 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
, 1, 0, 0, 0, 0);
818 l_ptr
->fsm_msg_cnt
++;
819 link_set_timer(l_ptr
, cont_intv
/ 4);
822 dbg_link("RES -> RR\n");
823 info("Resetting link <%s>, requested by peer\n",
825 tipc_link_reset(l_ptr
);
826 l_ptr
->state
= RESET_RESET
;
827 l_ptr
->fsm_msg_cnt
= 0;
828 tipc_link_send_proto_msg(l_ptr
, ACTIVATE_MSG
, 0, 0, 0, 0, 0);
829 l_ptr
->fsm_msg_cnt
++;
830 link_set_timer(l_ptr
, cont_intv
);
833 err("Unknown link event %u in WW state\n", event
);
836 case WORKING_UNKNOWN
:
839 case TRAFFIC_MSG_EVT
:
842 dbg_link("ACT -> WW\n");
843 l_ptr
->state
= WORKING_WORKING
;
844 l_ptr
->fsm_msg_cnt
= 0;
845 link_set_timer(l_ptr
, cont_intv
);
848 dbg_link("RES -> RR\n");
849 info("Resetting link <%s>, requested by peer "
850 "while probing\n", l_ptr
->name
);
851 tipc_link_reset(l_ptr
);
852 l_ptr
->state
= RESET_RESET
;
853 l_ptr
->fsm_msg_cnt
= 0;
854 tipc_link_send_proto_msg(l_ptr
, ACTIVATE_MSG
, 0, 0, 0, 0, 0);
855 l_ptr
->fsm_msg_cnt
++;
856 link_set_timer(l_ptr
, cont_intv
);
860 if (l_ptr
->next_in_no
!= l_ptr
->checkpoint
) {
861 dbg_link("-> WW \n");
862 l_ptr
->state
= WORKING_WORKING
;
863 l_ptr
->fsm_msg_cnt
= 0;
864 l_ptr
->checkpoint
= l_ptr
->next_in_no
;
865 if (tipc_bclink_acks_missing(l_ptr
->owner
)) {
866 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
868 l_ptr
->fsm_msg_cnt
++;
870 link_set_timer(l_ptr
, cont_intv
);
871 } else if (l_ptr
->fsm_msg_cnt
< l_ptr
->abort_limit
) {
872 dbg_link("Probing %u/%u,timer = %u ms)\n",
873 l_ptr
->fsm_msg_cnt
, l_ptr
->abort_limit
,
875 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
877 l_ptr
->fsm_msg_cnt
++;
878 link_set_timer(l_ptr
, cont_intv
/ 4);
879 } else { /* Link has failed */
880 dbg_link("-> RU (%u probes unanswered)\n",
882 warn("Resetting link <%s>, peer not responding\n",
884 tipc_link_reset(l_ptr
);
885 l_ptr
->state
= RESET_UNKNOWN
;
886 l_ptr
->fsm_msg_cnt
= 0;
887 tipc_link_send_proto_msg(l_ptr
, RESET_MSG
,
889 l_ptr
->fsm_msg_cnt
++;
890 link_set_timer(l_ptr
, cont_intv
);
894 err("Unknown link event %u in WU state\n", event
);
900 case TRAFFIC_MSG_EVT
:
904 other
= l_ptr
->owner
->active_links
[0];
905 if (other
&& link_working_unknown(other
)) {
909 dbg_link("ACT -> WW\n");
910 l_ptr
->state
= WORKING_WORKING
;
911 l_ptr
->fsm_msg_cnt
= 0;
912 link_activate(l_ptr
);
913 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
, 1, 0, 0, 0, 0);
914 l_ptr
->fsm_msg_cnt
++;
915 link_set_timer(l_ptr
, cont_intv
);
919 dbg_link(" -> RR\n");
920 l_ptr
->state
= RESET_RESET
;
921 l_ptr
->fsm_msg_cnt
= 0;
922 tipc_link_send_proto_msg(l_ptr
, ACTIVATE_MSG
, 1, 0, 0, 0, 0);
923 l_ptr
->fsm_msg_cnt
++;
924 link_set_timer(l_ptr
, cont_intv
);
932 tipc_link_send_proto_msg(l_ptr
, RESET_MSG
, 0, 0, 0, 0, 0);
933 l_ptr
->fsm_msg_cnt
++;
934 link_set_timer(l_ptr
, cont_intv
);
937 err("Unknown link event %u in RU state\n", event
);
943 case TRAFFIC_MSG_EVT
:
947 other
= l_ptr
->owner
->active_links
[0];
948 if (other
&& link_working_unknown(other
)) {
952 dbg_link("ACT -> WW\n");
953 l_ptr
->state
= WORKING_WORKING
;
954 l_ptr
->fsm_msg_cnt
= 0;
955 link_activate(l_ptr
);
956 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
, 1, 0, 0, 0, 0);
957 l_ptr
->fsm_msg_cnt
++;
958 link_set_timer(l_ptr
, cont_intv
);
965 tipc_link_send_proto_msg(l_ptr
, ACTIVATE_MSG
, 0, 0, 0, 0, 0);
966 l_ptr
->fsm_msg_cnt
++;
967 link_set_timer(l_ptr
, cont_intv
);
968 dbg_link("fsm_msg_cnt %u\n", l_ptr
->fsm_msg_cnt
);
971 err("Unknown link event %u in RR state\n", event
);
975 err("Unknown link state %u/%u\n", l_ptr
->state
, event
);
980 * link_bundle_buf(): Append contents of a buffer to
981 * the tail of an existing one.
984 static int link_bundle_buf(struct link
*l_ptr
,
985 struct sk_buff
*bundler
,
988 struct tipc_msg
*bundler_msg
= buf_msg(bundler
);
989 struct tipc_msg
*msg
= buf_msg(buf
);
990 u32 size
= msg_size(msg
);
991 u32 to_pos
= align(msg_size(bundler_msg
));
992 u32 rest
= link_max_pkt(l_ptr
) - to_pos
;
994 if (msg_user(bundler_msg
) != MSG_BUNDLER
)
996 if (msg_type(bundler_msg
) != OPEN_MSG
)
998 if (rest
< align(size
))
1001 skb_put(bundler
, (to_pos
- msg_size(bundler_msg
)) + size
);
1002 memcpy(bundler
->data
+ to_pos
, buf
->data
, size
);
1003 msg_set_size(bundler_msg
, to_pos
+ size
);
1004 msg_set_msgcnt(bundler_msg
, msg_msgcnt(bundler_msg
) + 1);
1005 dbg("Packed msg # %u(%u octets) into pos %u in buf(#%u)\n",
1006 msg_msgcnt(bundler_msg
), size
, to_pos
, msg_seqno(bundler_msg
));
1007 msg_dbg(msg
, "PACKD:");
1009 l_ptr
->stats
.sent_bundled
++;
1013 static void link_add_to_outqueue(struct link
*l_ptr
,
1014 struct sk_buff
*buf
,
1015 struct tipc_msg
*msg
)
1017 u32 ack
= mod(l_ptr
->next_in_no
- 1);
1018 u32 seqno
= mod(l_ptr
->next_out_no
++);
1020 msg_set_word(msg
, 2, ((ack
<< 16) | seqno
));
1021 msg_set_bcast_ack(msg
, l_ptr
->owner
->bclink
.last_in
);
1023 if (l_ptr
->first_out
) {
1024 l_ptr
->last_out
->next
= buf
;
1025 l_ptr
->last_out
= buf
;
1027 l_ptr
->first_out
= l_ptr
->last_out
= buf
;
1028 l_ptr
->out_queue_size
++;
1032 * tipc_link_send_buf() is the 'full path' for messages, called from
1033 * inside TIPC when the 'fast path' in tipc_send_buf
1034 * has failed, and from link_send()
1037 int tipc_link_send_buf(struct link
*l_ptr
, struct sk_buff
*buf
)
1039 struct tipc_msg
*msg
= buf_msg(buf
);
1040 u32 size
= msg_size(msg
);
1041 u32 dsz
= msg_data_sz(msg
);
1042 u32 queue_size
= l_ptr
->out_queue_size
;
1043 u32 imp
= msg_tot_importance(msg
);
1044 u32 queue_limit
= l_ptr
->queue_limit
[imp
];
1045 u32 max_packet
= link_max_pkt(l_ptr
);
1047 msg_set_prevnode(msg
, tipc_own_addr
); /* If routed message */
1049 /* Match msg importance against queue limits: */
1051 if (unlikely(queue_size
>= queue_limit
)) {
1052 if (imp
<= TIPC_CRITICAL_IMPORTANCE
) {
1053 return link_schedule_port(l_ptr
, msg_origport(msg
),
1056 msg_dbg(msg
, "TIPC: Congestion, throwing away\n");
1058 if (imp
> CONN_MANAGER
) {
1059 warn("Resetting link <%s>, send queue full", l_ptr
->name
);
1060 tipc_link_reset(l_ptr
);
1065 /* Fragmentation needed ? */
1067 if (size
> max_packet
)
1068 return tipc_link_send_long_buf(l_ptr
, buf
);
1070 /* Packet can be queued or sent: */
1072 if (queue_size
> l_ptr
->stats
.max_queue_sz
)
1073 l_ptr
->stats
.max_queue_sz
= queue_size
;
1075 if (likely(!tipc_bearer_congested(l_ptr
->b_ptr
, l_ptr
) &&
1076 !link_congested(l_ptr
))) {
1077 link_add_to_outqueue(l_ptr
, buf
, msg
);
1079 if (likely(tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
))) {
1080 l_ptr
->unacked_window
= 0;
1082 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
1083 l_ptr
->stats
.bearer_congs
++;
1084 l_ptr
->next_out
= buf
;
1088 /* Congestion: can message be bundled ?: */
1090 if ((msg_user(msg
) != CHANGEOVER_PROTOCOL
) &&
1091 (msg_user(msg
) != MSG_FRAGMENTER
)) {
1093 /* Try adding message to an existing bundle */
1095 if (l_ptr
->next_out
&&
1096 link_bundle_buf(l_ptr
, l_ptr
->last_out
, buf
)) {
1097 tipc_bearer_resolve_congestion(l_ptr
->b_ptr
, l_ptr
);
1101 /* Try creating a new bundle */
1103 if (size
<= max_packet
* 2 / 3) {
1104 struct sk_buff
*bundler
= buf_acquire(max_packet
);
1105 struct tipc_msg bundler_hdr
;
1108 msg_init(&bundler_hdr
, MSG_BUNDLER
, OPEN_MSG
,
1109 TIPC_OK
, INT_H_SIZE
, l_ptr
->addr
);
1110 memcpy(bundler
->data
, (unchar
*)&bundler_hdr
,
1112 skb_trim(bundler
, INT_H_SIZE
);
1113 link_bundle_buf(l_ptr
, bundler
, buf
);
1116 l_ptr
->stats
.sent_bundles
++;
1120 if (!l_ptr
->next_out
)
1121 l_ptr
->next_out
= buf
;
1122 link_add_to_outqueue(l_ptr
, buf
, msg
);
1123 tipc_bearer_resolve_congestion(l_ptr
->b_ptr
, l_ptr
);
1128 * tipc_link_send(): same as tipc_link_send_buf(), but the link to use has
1129 * not been selected yet, and the the owner node is not locked
1130 * Called by TIPC internal users, e.g. the name distributor
1133 int tipc_link_send(struct sk_buff
*buf
, u32 dest
, u32 selector
)
1137 int res
= -ELINKCONG
;
1139 read_lock_bh(&tipc_net_lock
);
1140 n_ptr
= tipc_node_select(dest
, selector
);
1142 tipc_node_lock(n_ptr
);
1143 l_ptr
= n_ptr
->active_links
[selector
& 1];
1145 dbg("tipc_link_send: found link %x for dest %x\n", l_ptr
, dest
);
1146 res
= tipc_link_send_buf(l_ptr
, buf
);
1148 dbg("Attempt to send msg to unreachable node:\n");
1149 msg_dbg(buf_msg(buf
),">>>");
1152 tipc_node_unlock(n_ptr
);
1154 dbg("Attempt to send msg to unknown node:\n");
1155 msg_dbg(buf_msg(buf
),">>>");
1158 read_unlock_bh(&tipc_net_lock
);
1163 * link_send_buf_fast: Entry for data messages where the
1164 * destination link is known and the header is complete,
1165 * inclusive total message length. Very time critical.
1166 * Link is locked. Returns user data length.
1169 static int link_send_buf_fast(struct link
*l_ptr
, struct sk_buff
*buf
,
1172 struct tipc_msg
*msg
= buf_msg(buf
);
1173 int res
= msg_data_sz(msg
);
1175 if (likely(!link_congested(l_ptr
))) {
1176 if (likely(msg_size(msg
) <= link_max_pkt(l_ptr
))) {
1177 if (likely(list_empty(&l_ptr
->b_ptr
->cong_links
))) {
1178 link_add_to_outqueue(l_ptr
, buf
, msg
);
1179 if (likely(tipc_bearer_send(l_ptr
->b_ptr
, buf
,
1180 &l_ptr
->media_addr
))) {
1181 l_ptr
->unacked_window
= 0;
1182 msg_dbg(msg
,"SENT_FAST:");
1185 dbg("failed sent fast...\n");
1186 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
1187 l_ptr
->stats
.bearer_congs
++;
1188 l_ptr
->next_out
= buf
;
1193 *used_max_pkt
= link_max_pkt(l_ptr
);
1195 return tipc_link_send_buf(l_ptr
, buf
); /* All other cases */
1199 * tipc_send_buf_fast: Entry for data messages where the
1200 * destination node is known and the header is complete,
1201 * inclusive total message length.
1202 * Returns user data length.
1204 int tipc_send_buf_fast(struct sk_buff
*buf
, u32 destnode
)
1209 u32 selector
= msg_origport(buf_msg(buf
)) & 1;
1212 if (destnode
== tipc_own_addr
)
1213 return tipc_port_recv_msg(buf
);
1215 read_lock_bh(&tipc_net_lock
);
1216 n_ptr
= tipc_node_select(destnode
, selector
);
1217 if (likely(n_ptr
)) {
1218 tipc_node_lock(n_ptr
);
1219 l_ptr
= n_ptr
->active_links
[selector
];
1220 dbg("send_fast: buf %x selected %x, destnode = %x\n",
1221 buf
, l_ptr
, destnode
);
1222 if (likely(l_ptr
)) {
1223 res
= link_send_buf_fast(l_ptr
, buf
, &dummy
);
1224 tipc_node_unlock(n_ptr
);
1225 read_unlock_bh(&tipc_net_lock
);
1228 tipc_node_unlock(n_ptr
);
1230 read_unlock_bh(&tipc_net_lock
);
1231 res
= msg_data_sz(buf_msg(buf
));
1232 tipc_reject_msg(buf
, TIPC_ERR_NO_NODE
);
1238 * tipc_link_send_sections_fast: Entry for messages where the
1239 * destination processor is known and the header is complete,
1240 * except for total message length.
1241 * Returns user data length or errno.
1243 int tipc_link_send_sections_fast(struct port
*sender
,
1244 struct iovec
const *msg_sect
,
1248 struct tipc_msg
*hdr
= &sender
->publ
.phdr
;
1250 struct sk_buff
*buf
;
1253 u32 selector
= msg_origport(hdr
) & 1;
1257 * Try building message using port's max_pkt hint.
1258 * (Must not hold any locks while building message.)
1261 res
= msg_build(hdr
, msg_sect
, num_sect
, sender
->max_pkt
,
1262 !sender
->user_port
, &buf
);
1264 read_lock_bh(&tipc_net_lock
);
1265 node
= tipc_node_select(destaddr
, selector
);
1267 tipc_node_lock(node
);
1268 l_ptr
= node
->active_links
[selector
];
1269 if (likely(l_ptr
)) {
1271 res
= link_send_buf_fast(l_ptr
, buf
,
1273 if (unlikely(res
< 0))
1276 tipc_node_unlock(node
);
1277 read_unlock_bh(&tipc_net_lock
);
1281 /* Exit if build request was invalid */
1283 if (unlikely(res
< 0))
1286 /* Exit if link (or bearer) is congested */
1288 if (link_congested(l_ptr
) ||
1289 !list_empty(&l_ptr
->b_ptr
->cong_links
)) {
1290 res
= link_schedule_port(l_ptr
,
1291 sender
->publ
.ref
, res
);
1296 * Message size exceeds max_pkt hint; update hint,
1297 * then re-try fast path or fragment the message
1300 sender
->max_pkt
= link_max_pkt(l_ptr
);
1301 tipc_node_unlock(node
);
1302 read_unlock_bh(&tipc_net_lock
);
1305 if ((msg_hdr_sz(hdr
) + res
) <= sender
->max_pkt
)
1308 return link_send_sections_long(sender
, msg_sect
,
1309 num_sect
, destaddr
);
1311 tipc_node_unlock(node
);
1313 read_unlock_bh(&tipc_net_lock
);
1315 /* Couldn't find a link to the destination node */
1318 return tipc_reject_msg(buf
, TIPC_ERR_NO_NODE
);
1320 return tipc_port_reject_sections(sender
, hdr
, msg_sect
, num_sect
,
1326 * link_send_sections_long(): Entry for long messages where the
1327 * destination node is known and the header is complete,
1328 * inclusive total message length.
1329 * Link and bearer congestion status have been checked to be ok,
1330 * and are ignored if they change.
1332 * Note that fragments do not use the full link MTU so that they won't have
1333 * to undergo refragmentation if link changeover causes them to be sent
1334 * over another link with an additional tunnel header added as prefix.
1335 * (Refragmentation will still occur if the other link has a smaller MTU.)
1337 * Returns user data length or errno.
1339 static int link_send_sections_long(struct port
*sender
,
1340 struct iovec
const *msg_sect
,
1346 struct tipc_msg
*hdr
= &sender
->publ
.phdr
;
1347 u32 dsz
= msg_data_sz(hdr
);
1348 u32 max_pkt
,fragm_sz
,rest
;
1349 struct tipc_msg fragm_hdr
;
1350 struct sk_buff
*buf
,*buf_chain
,*prev
;
1351 u32 fragm_crs
,fragm_rest
,hsz
,sect_rest
;
1352 const unchar
*sect_crs
;
1358 max_pkt
= sender
->max_pkt
- INT_H_SIZE
;
1359 /* leave room for tunnel header in case of link changeover */
1360 fragm_sz
= max_pkt
- INT_H_SIZE
;
1361 /* leave room for fragmentation header in each fragment */
1369 /* Prepare reusable fragment header: */
1371 msg_dbg(hdr
, ">FRAGMENTING>");
1372 msg_init(&fragm_hdr
, MSG_FRAGMENTER
, FIRST_FRAGMENT
,
1373 TIPC_OK
, INT_H_SIZE
, msg_destnode(hdr
));
1374 msg_set_link_selector(&fragm_hdr
, sender
->publ
.ref
);
1375 msg_set_size(&fragm_hdr
, max_pkt
);
1376 msg_set_fragm_no(&fragm_hdr
, 1);
1378 /* Prepare header of first fragment: */
1380 buf_chain
= buf
= buf_acquire(max_pkt
);
1384 memcpy(buf
->data
, (unchar
*)&fragm_hdr
, INT_H_SIZE
);
1385 hsz
= msg_hdr_sz(hdr
);
1386 memcpy(buf
->data
+ INT_H_SIZE
, (unchar
*)hdr
, hsz
);
1387 msg_dbg(buf_msg(buf
), ">BUILD>");
1389 /* Chop up message: */
1391 fragm_crs
= INT_H_SIZE
+ hsz
;
1392 fragm_rest
= fragm_sz
- hsz
;
1394 do { /* For all sections */
1398 sect_rest
= msg_sect
[++curr_sect
].iov_len
;
1399 sect_crs
= (const unchar
*)msg_sect
[curr_sect
].iov_base
;
1402 if (sect_rest
< fragm_rest
)
1407 if (likely(!sender
->user_port
)) {
1408 if (copy_from_user(buf
->data
+ fragm_crs
, sect_crs
, sz
)) {
1410 for (; buf_chain
; buf_chain
= buf
) {
1411 buf
= buf_chain
->next
;
1412 buf_discard(buf_chain
);
1417 memcpy(buf
->data
+ fragm_crs
, sect_crs
, sz
);
1425 if (!fragm_rest
&& rest
) {
1427 /* Initiate new fragment: */
1428 if (rest
<= fragm_sz
) {
1430 msg_set_type(&fragm_hdr
,LAST_FRAGMENT
);
1432 msg_set_type(&fragm_hdr
, FRAGMENT
);
1434 msg_set_size(&fragm_hdr
, fragm_sz
+ INT_H_SIZE
);
1435 msg_set_fragm_no(&fragm_hdr
, ++fragm_no
);
1437 buf
= buf_acquire(fragm_sz
+ INT_H_SIZE
);
1443 memcpy(buf
->data
, (unchar
*)&fragm_hdr
, INT_H_SIZE
);
1444 fragm_crs
= INT_H_SIZE
;
1445 fragm_rest
= fragm_sz
;
1446 msg_dbg(buf_msg(buf
)," >BUILD>");
1452 * Now we have a buffer chain. Select a link and check
1453 * that packet size is still OK
1455 node
= tipc_node_select(destaddr
, sender
->publ
.ref
& 1);
1457 tipc_node_lock(node
);
1458 l_ptr
= node
->active_links
[sender
->publ
.ref
& 1];
1460 tipc_node_unlock(node
);
1463 if (link_max_pkt(l_ptr
) < max_pkt
) {
1464 sender
->max_pkt
= link_max_pkt(l_ptr
);
1465 tipc_node_unlock(node
);
1466 for (; buf_chain
; buf_chain
= buf
) {
1467 buf
= buf_chain
->next
;
1468 buf_discard(buf_chain
);
1474 for (; buf_chain
; buf_chain
= buf
) {
1475 buf
= buf_chain
->next
;
1476 buf_discard(buf_chain
);
1478 return tipc_port_reject_sections(sender
, hdr
, msg_sect
, num_sect
,
1482 /* Append whole chain to send queue: */
1485 l_ptr
->long_msg_seq_no
= mod(l_ptr
->long_msg_seq_no
+ 1);
1486 if (!l_ptr
->next_out
)
1487 l_ptr
->next_out
= buf_chain
;
1488 l_ptr
->stats
.sent_fragmented
++;
1490 struct sk_buff
*next
= buf
->next
;
1491 struct tipc_msg
*msg
= buf_msg(buf
);
1493 l_ptr
->stats
.sent_fragments
++;
1494 msg_set_long_msgno(msg
, l_ptr
->long_msg_seq_no
);
1495 link_add_to_outqueue(l_ptr
, buf
, msg
);
1496 msg_dbg(msg
, ">ADD>");
1500 /* Send it, if possible: */
1502 tipc_link_push_queue(l_ptr
);
1503 tipc_node_unlock(node
);
1508 * tipc_link_push_packet: Push one unsent packet to the media
1510 u32
tipc_link_push_packet(struct link
*l_ptr
)
1512 struct sk_buff
*buf
= l_ptr
->first_out
;
1513 u32 r_q_size
= l_ptr
->retransm_queue_size
;
1514 u32 r_q_head
= l_ptr
->retransm_queue_head
;
1516 /* Step to position where retransmission failed, if any, */
1517 /* consider that buffers may have been released in meantime */
1519 if (r_q_size
&& buf
) {
1520 u32 last
= lesser(mod(r_q_head
+ r_q_size
),
1521 link_last_sent(l_ptr
));
1522 u32 first
= msg_seqno(buf_msg(buf
));
1524 while (buf
&& less(first
, r_q_head
)) {
1525 first
= mod(first
+ 1);
1528 l_ptr
->retransm_queue_head
= r_q_head
= first
;
1529 l_ptr
->retransm_queue_size
= r_q_size
= mod(last
- first
);
1532 /* Continue retransmission now, if there is anything: */
1534 if (r_q_size
&& buf
&& !skb_cloned(buf
)) {
1535 msg_set_ack(buf_msg(buf
), mod(l_ptr
->next_in_no
- 1));
1536 msg_set_bcast_ack(buf_msg(buf
), l_ptr
->owner
->bclink
.last_in
);
1537 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
1538 msg_dbg(buf_msg(buf
), ">DEF-RETR>");
1539 l_ptr
->retransm_queue_head
= mod(++r_q_head
);
1540 l_ptr
->retransm_queue_size
= --r_q_size
;
1541 l_ptr
->stats
.retransmitted
++;
1544 l_ptr
->stats
.bearer_congs
++;
1545 msg_dbg(buf_msg(buf
), "|>DEF-RETR>");
1550 /* Send deferred protocol message, if any: */
1552 buf
= l_ptr
->proto_msg_queue
;
1554 msg_set_ack(buf_msg(buf
), mod(l_ptr
->next_in_no
- 1));
1555 msg_set_bcast_ack(buf_msg(buf
),l_ptr
->owner
->bclink
.last_in
);
1556 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
1557 msg_dbg(buf_msg(buf
), ">DEF-PROT>");
1558 l_ptr
->unacked_window
= 0;
1560 l_ptr
->proto_msg_queue
= NULL
;
1563 msg_dbg(buf_msg(buf
), "|>DEF-PROT>");
1564 l_ptr
->stats
.bearer_congs
++;
1569 /* Send one deferred data message, if send window not full: */
1571 buf
= l_ptr
->next_out
;
1573 struct tipc_msg
*msg
= buf_msg(buf
);
1574 u32 next
= msg_seqno(msg
);
1575 u32 first
= msg_seqno(buf_msg(l_ptr
->first_out
));
1577 if (mod(next
- first
) < l_ptr
->queue_limit
[0]) {
1578 msg_set_ack(msg
, mod(l_ptr
->next_in_no
- 1));
1579 msg_set_bcast_ack(msg
, l_ptr
->owner
->bclink
.last_in
);
1580 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
1581 if (msg_user(msg
) == MSG_BUNDLER
)
1582 msg_set_type(msg
, CLOSED_MSG
);
1583 msg_dbg(msg
, ">PUSH-DATA>");
1584 l_ptr
->next_out
= buf
->next
;
1587 msg_dbg(msg
, "|PUSH-DATA|");
1588 l_ptr
->stats
.bearer_congs
++;
1593 return PUSH_FINISHED
;
1597 * push_queue(): push out the unsent messages of a link where
1598 * congestion has abated. Node is locked
1600 void tipc_link_push_queue(struct link
*l_ptr
)
1604 if (tipc_bearer_congested(l_ptr
->b_ptr
, l_ptr
))
1608 res
= tipc_link_push_packet(l_ptr
);
1610 while (res
== TIPC_OK
);
1611 if (res
== PUSH_FAILED
)
1612 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
1615 static void link_reset_all(unsigned long addr
)
1618 char addr_string
[16];
1621 read_lock_bh(&tipc_net_lock
);
1622 n_ptr
= tipc_node_find((u32
)addr
);
1624 read_unlock_bh(&tipc_net_lock
);
1625 return; /* node no longer exists */
1628 tipc_node_lock(n_ptr
);
1630 warn("Resetting all links to %s\n",
1631 addr_string_fill(addr_string
, n_ptr
->addr
));
1633 for (i
= 0; i
< MAX_BEARERS
; i
++) {
1634 if (n_ptr
->links
[i
]) {
1635 link_print(n_ptr
->links
[i
], TIPC_OUTPUT
,
1636 "Resetting link\n");
1637 tipc_link_reset(n_ptr
->links
[i
]);
1641 tipc_node_unlock(n_ptr
);
1642 read_unlock_bh(&tipc_net_lock
);
1645 static void link_retransmit_failure(struct link
*l_ptr
, struct sk_buff
*buf
)
1647 struct tipc_msg
*msg
= buf_msg(buf
);
1649 warn("Retransmission failure on link <%s>\n", l_ptr
->name
);
1650 tipc_msg_print(TIPC_OUTPUT
, msg
, ">RETR-FAIL>");
1654 /* Handle failure on standard link */
1656 link_print(l_ptr
, TIPC_OUTPUT
, "Resetting link\n");
1657 tipc_link_reset(l_ptr
);
1661 /* Handle failure on broadcast link */
1664 char addr_string
[16];
1666 tipc_printf(TIPC_OUTPUT
, "Msg seq number: %u, ", msg_seqno(msg
));
1667 tipc_printf(TIPC_OUTPUT
, "Outstanding acks: %u\n", (u32
)TIPC_SKB_CB(buf
)->handle
);
1669 n_ptr
= l_ptr
->owner
->next
;
1670 tipc_node_lock(n_ptr
);
1672 addr_string_fill(addr_string
, n_ptr
->addr
);
1673 tipc_printf(TIPC_OUTPUT
, "Multicast link info for %s\n", addr_string
);
1674 tipc_printf(TIPC_OUTPUT
, "Supported: %d, ", n_ptr
->bclink
.supported
);
1675 tipc_printf(TIPC_OUTPUT
, "Acked: %u\n", n_ptr
->bclink
.acked
);
1676 tipc_printf(TIPC_OUTPUT
, "Last in: %u, ", n_ptr
->bclink
.last_in
);
1677 tipc_printf(TIPC_OUTPUT
, "Gap after: %u, ", n_ptr
->bclink
.gap_after
);
1678 tipc_printf(TIPC_OUTPUT
, "Gap to: %u\n", n_ptr
->bclink
.gap_to
);
1679 tipc_printf(TIPC_OUTPUT
, "Nack sync: %u\n\n", n_ptr
->bclink
.nack_sync
);
1681 tipc_k_signal((Handler
)link_reset_all
, (unsigned long)n_ptr
->addr
);
1683 tipc_node_unlock(n_ptr
);
1685 l_ptr
->stale_count
= 0;
1689 void tipc_link_retransmit(struct link
*l_ptr
, struct sk_buff
*buf
,
1692 struct tipc_msg
*msg
;
1699 dbg("Retransmitting %u in link %x\n", retransmits
, l_ptr
);
1701 if (tipc_bearer_congested(l_ptr
->b_ptr
, l_ptr
)) {
1702 if (!skb_cloned(buf
)) {
1703 msg_dbg(msg
, ">NO_RETR->BCONG>");
1704 dbg_print_link(l_ptr
, " ");
1705 l_ptr
->retransm_queue_head
= msg_seqno(msg
);
1706 l_ptr
->retransm_queue_size
= retransmits
;
1709 /* Don't retransmit if driver already has the buffer */
1712 /* Detect repeated retransmit failures on uncongested bearer */
1714 if (l_ptr
->last_retransmitted
== msg_seqno(msg
)) {
1715 if (++l_ptr
->stale_count
> 100) {
1716 link_retransmit_failure(l_ptr
, buf
);
1720 l_ptr
->last_retransmitted
= msg_seqno(msg
);
1721 l_ptr
->stale_count
= 1;
1725 while (retransmits
&& (buf
!= l_ptr
->next_out
) && buf
&& !skb_cloned(buf
)) {
1727 msg_set_ack(msg
, mod(l_ptr
->next_in_no
- 1));
1728 msg_set_bcast_ack(msg
, l_ptr
->owner
->bclink
.last_in
);
1729 if (tipc_bearer_send(l_ptr
->b_ptr
, buf
, &l_ptr
->media_addr
)) {
1730 msg_dbg(buf_msg(buf
), ">RETR>");
1733 l_ptr
->stats
.retransmitted
++;
1735 tipc_bearer_schedule(l_ptr
->b_ptr
, l_ptr
);
1736 l_ptr
->stats
.bearer_congs
++;
1737 l_ptr
->retransm_queue_head
= msg_seqno(buf_msg(buf
));
1738 l_ptr
->retransm_queue_size
= retransmits
;
1743 l_ptr
->retransm_queue_head
= l_ptr
->retransm_queue_size
= 0;
1747 * link_recv_non_seq: Receive packets which are outside
1748 * the link sequence flow
1751 static void link_recv_non_seq(struct sk_buff
*buf
)
1753 struct tipc_msg
*msg
= buf_msg(buf
);
1755 if (msg_user(msg
) == LINK_CONFIG
)
1756 tipc_disc_recv_msg(buf
);
1758 tipc_bclink_recv_pkt(buf
);
1762 * link_insert_deferred_queue - insert deferred messages back into receive chain
1765 static struct sk_buff
*link_insert_deferred_queue(struct link
*l_ptr
,
1766 struct sk_buff
*buf
)
1770 if (l_ptr
->oldest_deferred_in
== NULL
)
1773 seq_no
= msg_seqno(buf_msg(l_ptr
->oldest_deferred_in
));
1774 if (seq_no
== mod(l_ptr
->next_in_no
)) {
1775 l_ptr
->newest_deferred_in
->next
= buf
;
1776 buf
= l_ptr
->oldest_deferred_in
;
1777 l_ptr
->oldest_deferred_in
= NULL
;
1778 l_ptr
->deferred_inqueue_sz
= 0;
1783 void tipc_recv_msg(struct sk_buff
*head
, struct tipc_bearer
*tb_ptr
)
1785 read_lock_bh(&tipc_net_lock
);
1787 struct bearer
*b_ptr
;
1790 struct sk_buff
*crs
;
1791 struct sk_buff
*buf
= head
;
1792 struct tipc_msg
*msg
= buf_msg(buf
);
1793 u32 seq_no
= msg_seqno(msg
);
1794 u32 ackd
= msg_ack(msg
);
1798 b_ptr
= (struct bearer
*)tb_ptr
;
1799 TIPC_SKB_CB(buf
)->handle
= b_ptr
;
1802 if (unlikely(msg_version(msg
) != TIPC_VERSION
))
1805 if (msg_user(msg
) != LINK_PROTOCOL
)
1807 msg_dbg(msg
,"<REC<");
1809 if (unlikely(msg_non_seq(msg
))) {
1810 link_recv_non_seq(buf
);
1814 if (unlikely(!msg_short(msg
) &&
1815 (msg_destnode(msg
) != tipc_own_addr
)))
1818 n_ptr
= tipc_node_find(msg_prevnode(msg
));
1819 if (unlikely(!n_ptr
))
1822 tipc_node_lock(n_ptr
);
1823 l_ptr
= n_ptr
->links
[b_ptr
->identity
];
1824 if (unlikely(!l_ptr
)) {
1825 tipc_node_unlock(n_ptr
);
1829 * Release acked messages
1831 if (less(n_ptr
->bclink
.acked
, msg_bcast_ack(msg
))) {
1832 if (tipc_node_is_up(n_ptr
) && n_ptr
->bclink
.supported
)
1833 tipc_bclink_acknowledge(n_ptr
, msg_bcast_ack(msg
));
1836 crs
= l_ptr
->first_out
;
1837 while ((crs
!= l_ptr
->next_out
) &&
1838 less_eq(msg_seqno(buf_msg(crs
)), ackd
)) {
1839 struct sk_buff
*next
= crs
->next
;
1846 l_ptr
->first_out
= crs
;
1847 l_ptr
->out_queue_size
-= released
;
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);
1859 if (likely(link_working_working(l_ptr
))) {
1860 if (likely(seq_no
== mod(l_ptr
->next_in_no
))) {
1861 l_ptr
->next_in_no
++;
1862 if (unlikely(l_ptr
->oldest_deferred_in
))
1863 head
= link_insert_deferred_queue(l_ptr
,
1865 if (likely(msg_is_dest(msg
, tipc_own_addr
))) {
1867 if (likely(msg_isdata(msg
))) {
1868 tipc_node_unlock(n_ptr
);
1869 tipc_port_recv_msg(buf
);
1872 switch (msg_user(msg
)) {
1874 l_ptr
->stats
.recv_bundles
++;
1875 l_ptr
->stats
.recv_bundled
+=
1877 tipc_node_unlock(n_ptr
);
1878 tipc_link_recv_bundle(buf
);
1880 case ROUTE_DISTRIBUTOR
:
1881 tipc_node_unlock(n_ptr
);
1882 tipc_cltr_recv_routing_table(buf
);
1884 case NAME_DISTRIBUTOR
:
1885 tipc_node_unlock(n_ptr
);
1886 tipc_named_recv(buf
);
1889 tipc_node_unlock(n_ptr
);
1890 tipc_port_recv_proto_msg(buf
);
1892 case MSG_FRAGMENTER
:
1893 l_ptr
->stats
.recv_fragments
++;
1894 if (tipc_link_recv_fragment(&l_ptr
->defragm_buf
,
1896 l_ptr
->stats
.recv_fragmented
++;
1900 case CHANGEOVER_PROTOCOL
:
1901 type
= msg_type(msg
);
1902 if (link_recv_changeover_msg(&l_ptr
, &buf
)) {
1904 seq_no
= msg_seqno(msg
);
1905 TIPC_SKB_CB(buf
)->handle
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 memcpy(buf
->data
, (unchar
*)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 memcpy(buf
->data
, (unchar
*)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
) && l_ptr
->peer_session
) {
2191 if (msg_session(msg
) == l_ptr
->peer_session
) {
2192 dbg("Duplicate RESET: %u<->%u\n",
2193 msg_session(msg
), l_ptr
->peer_session
);
2194 break; /* duplicate: ignore */
2199 /* Update link settings according other endpoint's values */
2201 strcpy((strrchr(l_ptr
->name
, ':') + 1), (char *)msg_data(msg
));
2203 if ((msg_tol
= msg_link_tolerance(msg
)) &&
2204 (msg_tol
> l_ptr
->tolerance
))
2205 link_set_supervision_props(l_ptr
, msg_tol
);
2207 if (msg_linkprio(msg
) > l_ptr
->priority
)
2208 l_ptr
->priority
= msg_linkprio(msg
);
2210 max_pkt_info
= msg_max_pkt(msg
);
2212 if (max_pkt_info
< l_ptr
->max_pkt_target
)
2213 l_ptr
->max_pkt_target
= max_pkt_info
;
2214 if (l_ptr
->max_pkt
> l_ptr
->max_pkt_target
)
2215 l_ptr
->max_pkt
= l_ptr
->max_pkt_target
;
2217 l_ptr
->max_pkt
= l_ptr
->max_pkt_target
;
2219 l_ptr
->owner
->bclink
.supported
= (max_pkt_info
!= 0);
2221 link_state_event(l_ptr
, msg_type(msg
));
2223 l_ptr
->peer_session
= msg_session(msg
);
2224 l_ptr
->peer_bearer_id
= msg_bearer_id(msg
);
2226 /* Synchronize broadcast sequence numbers */
2227 if (!tipc_node_has_redundant_links(l_ptr
->owner
)) {
2228 l_ptr
->owner
->bclink
.last_in
= mod(msg_last_bcast(msg
));
2233 if ((msg_tol
= msg_link_tolerance(msg
)))
2234 link_set_supervision_props(l_ptr
, msg_tol
);
2236 if (msg_linkprio(msg
) &&
2237 (msg_linkprio(msg
) != l_ptr
->priority
)) {
2238 warn("Resetting link <%s>, priority change %u->%u\n",
2239 l_ptr
->name
, l_ptr
->priority
, msg_linkprio(msg
));
2240 l_ptr
->priority
= msg_linkprio(msg
);
2241 tipc_link_reset(l_ptr
); /* Enforce change to take effect */
2244 link_state_event(l_ptr
, TRAFFIC_MSG_EVT
);
2245 l_ptr
->stats
.recv_states
++;
2246 if (link_reset_unknown(l_ptr
))
2249 if (less_eq(mod(l_ptr
->next_in_no
), msg_next_sent(msg
))) {
2250 rec_gap
= mod(msg_next_sent(msg
) -
2251 mod(l_ptr
->next_in_no
));
2254 max_pkt_ack
= msg_max_pkt(msg
);
2255 if (max_pkt_ack
> l_ptr
->max_pkt
) {
2256 dbg("Link <%s> updated MTU %u -> %u\n",
2257 l_ptr
->name
, l_ptr
->max_pkt
, max_pkt_ack
);
2258 l_ptr
->max_pkt
= max_pkt_ack
;
2259 l_ptr
->max_pkt_probes
= 0;
2263 if (msg_probe(msg
)) {
2264 l_ptr
->stats
.recv_probes
++;
2265 if (msg_size(msg
) > sizeof(l_ptr
->proto_msg
)) {
2266 max_pkt_ack
= msg_size(msg
);
2270 /* Protocol message before retransmits, reduce loss risk */
2272 tipc_bclink_check_gap(l_ptr
->owner
, msg_last_bcast(msg
));
2274 if (rec_gap
|| (msg_probe(msg
))) {
2275 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
2276 0, rec_gap
, 0, 0, max_pkt_ack
);
2278 if (msg_seq_gap(msg
)) {
2279 msg_dbg(msg
, "With Gap:");
2280 l_ptr
->stats
.recv_nacks
++;
2281 tipc_link_retransmit(l_ptr
, l_ptr
->first_out
,
2286 msg_dbg(buf_msg(buf
), "<DISCARDING UNKNOWN<");
2294 * tipc_link_tunnel(): Send one message via a link belonging to
2295 * another bearer. Owner node is locked.
2297 void tipc_link_tunnel(struct link
*l_ptr
,
2298 struct tipc_msg
*tunnel_hdr
,
2299 struct tipc_msg
*msg
,
2302 struct link
*tunnel
;
2303 struct sk_buff
*buf
;
2304 u32 length
= msg_size(msg
);
2306 tunnel
= l_ptr
->owner
->active_links
[selector
& 1];
2307 if (!tipc_link_is_up(tunnel
)) {
2308 warn("Link changeover error, "
2309 "tunnel link no longer available\n");
2312 msg_set_size(tunnel_hdr
, length
+ INT_H_SIZE
);
2313 buf
= buf_acquire(length
+ INT_H_SIZE
);
2315 warn("Link changeover error, "
2316 "unable to send tunnel msg\n");
2319 memcpy(buf
->data
, (unchar
*)tunnel_hdr
, INT_H_SIZE
);
2320 memcpy(buf
->data
+ INT_H_SIZE
, (unchar
*)msg
, length
);
2321 dbg("%c->%c:", l_ptr
->b_ptr
->net_plane
, tunnel
->b_ptr
->net_plane
);
2322 msg_dbg(buf_msg(buf
), ">SEND>");
2323 tipc_link_send_buf(tunnel
, buf
);
2329 * changeover(): Send whole message queue via the remaining link
2330 * Owner node is locked.
2333 void tipc_link_changeover(struct link
*l_ptr
)
2335 u32 msgcount
= l_ptr
->out_queue_size
;
2336 struct sk_buff
*crs
= l_ptr
->first_out
;
2337 struct link
*tunnel
= l_ptr
->owner
->active_links
[0];
2338 struct tipc_msg tunnel_hdr
;
2344 if (!l_ptr
->owner
->permit_changeover
) {
2345 warn("Link changeover error, "
2346 "peer did not permit changeover\n");
2350 msg_init(&tunnel_hdr
, CHANGEOVER_PROTOCOL
,
2351 ORIGINAL_MSG
, TIPC_OK
, INT_H_SIZE
, l_ptr
->addr
);
2352 msg_set_bearer_id(&tunnel_hdr
, l_ptr
->peer_bearer_id
);
2353 msg_set_msgcnt(&tunnel_hdr
, msgcount
);
2354 dbg("Link changeover requires %u tunnel messages\n", msgcount
);
2356 if (!l_ptr
->first_out
) {
2357 struct sk_buff
*buf
;
2359 buf
= buf_acquire(INT_H_SIZE
);
2361 memcpy(buf
->data
, (unchar
*)&tunnel_hdr
, INT_H_SIZE
);
2362 msg_set_size(&tunnel_hdr
, INT_H_SIZE
);
2363 dbg("%c->%c:", l_ptr
->b_ptr
->net_plane
,
2364 tunnel
->b_ptr
->net_plane
);
2365 msg_dbg(&tunnel_hdr
, "EMPTY>SEND>");
2366 tipc_link_send_buf(tunnel
, buf
);
2368 warn("Link changeover error, "
2369 "unable to send changeover msg\n");
2374 split_bundles
= (l_ptr
->owner
->active_links
[0] !=
2375 l_ptr
->owner
->active_links
[1]);
2378 struct tipc_msg
*msg
= buf_msg(crs
);
2380 if ((msg_user(msg
) == MSG_BUNDLER
) && split_bundles
) {
2381 u32 msgcount
= msg_msgcnt(msg
);
2382 struct tipc_msg
*m
= msg_get_wrapped(msg
);
2383 unchar
* pos
= (unchar
*)m
;
2385 while (msgcount
--) {
2386 msg_set_seqno(m
,msg_seqno(msg
));
2387 tipc_link_tunnel(l_ptr
, &tunnel_hdr
, m
,
2388 msg_link_selector(m
));
2389 pos
+= align(msg_size(m
));
2390 m
= (struct tipc_msg
*)pos
;
2393 tipc_link_tunnel(l_ptr
, &tunnel_hdr
, msg
,
2394 msg_link_selector(msg
));
2400 void tipc_link_send_duplicate(struct link
*l_ptr
, struct link
*tunnel
)
2402 struct sk_buff
*iter
;
2403 struct tipc_msg tunnel_hdr
;
2405 msg_init(&tunnel_hdr
, CHANGEOVER_PROTOCOL
,
2406 DUPLICATE_MSG
, TIPC_OK
, INT_H_SIZE
, l_ptr
->addr
);
2407 msg_set_msgcnt(&tunnel_hdr
, l_ptr
->out_queue_size
);
2408 msg_set_bearer_id(&tunnel_hdr
, l_ptr
->peer_bearer_id
);
2409 iter
= l_ptr
->first_out
;
2411 struct sk_buff
*outbuf
;
2412 struct tipc_msg
*msg
= buf_msg(iter
);
2413 u32 length
= msg_size(msg
);
2415 if (msg_user(msg
) == MSG_BUNDLER
)
2416 msg_set_type(msg
, CLOSED_MSG
);
2417 msg_set_ack(msg
, mod(l_ptr
->next_in_no
- 1)); /* Update */
2418 msg_set_bcast_ack(msg
, l_ptr
->owner
->bclink
.last_in
);
2419 msg_set_size(&tunnel_hdr
, length
+ INT_H_SIZE
);
2420 outbuf
= buf_acquire(length
+ INT_H_SIZE
);
2421 if (outbuf
== NULL
) {
2422 warn("Link changeover error, "
2423 "unable to send duplicate msg\n");
2426 memcpy(outbuf
->data
, (unchar
*)&tunnel_hdr
, INT_H_SIZE
);
2427 memcpy(outbuf
->data
+ INT_H_SIZE
, iter
->data
, length
);
2428 dbg("%c->%c:", l_ptr
->b_ptr
->net_plane
,
2429 tunnel
->b_ptr
->net_plane
);
2430 msg_dbg(buf_msg(outbuf
), ">SEND>");
2431 tipc_link_send_buf(tunnel
, outbuf
);
2432 if (!tipc_link_is_up(l_ptr
))
2441 * buf_extract - extracts embedded TIPC message from another message
2442 * @skb: encapsulating message buffer
2443 * @from_pos: offset to extract from
2445 * Returns a new message buffer containing an embedded message. The
2446 * encapsulating message itself is left unchanged.
2449 static struct sk_buff
*buf_extract(struct sk_buff
*skb
, u32 from_pos
)
2451 struct tipc_msg
*msg
= (struct tipc_msg
*)(skb
->data
+ from_pos
);
2452 u32 size
= msg_size(msg
);
2455 eb
= buf_acquire(size
);
2457 memcpy(eb
->data
, (unchar
*)msg
, size
);
2462 * link_recv_changeover_msg(): Receive tunneled packet sent
2463 * via other link. Node is locked. Return extracted buffer.
2466 static int link_recv_changeover_msg(struct link
**l_ptr
,
2467 struct sk_buff
**buf
)
2469 struct sk_buff
*tunnel_buf
= *buf
;
2470 struct link
*dest_link
;
2471 struct tipc_msg
*msg
;
2472 struct tipc_msg
*tunnel_msg
= buf_msg(tunnel_buf
);
2473 u32 msg_typ
= msg_type(tunnel_msg
);
2474 u32 msg_count
= msg_msgcnt(tunnel_msg
);
2476 dest_link
= (*l_ptr
)->owner
->links
[msg_bearer_id(tunnel_msg
)];
2478 msg_dbg(tunnel_msg
, "NOLINK/<REC<");
2481 if (dest_link
== *l_ptr
) {
2482 err("Unexpected changeover message on link <%s>\n",
2486 dbg("%c<-%c:", dest_link
->b_ptr
->net_plane
,
2487 (*l_ptr
)->b_ptr
->net_plane
);
2489 msg
= msg_get_wrapped(tunnel_msg
);
2491 if (msg_typ
== DUPLICATE_MSG
) {
2492 if (less(msg_seqno(msg
), mod(dest_link
->next_in_no
))) {
2493 msg_dbg(tunnel_msg
, "DROP/<REC<");
2496 *buf
= buf_extract(tunnel_buf
,INT_H_SIZE
);
2498 warn("Link changeover error, duplicate msg dropped\n");
2501 msg_dbg(tunnel_msg
, "TNL<REC<");
2502 buf_discard(tunnel_buf
);
2506 /* First original message ?: */
2508 if (tipc_link_is_up(dest_link
)) {
2509 msg_dbg(tunnel_msg
, "UP/FIRST/<REC<");
2510 info("Resetting link <%s>, changeover initiated by peer\n",
2512 tipc_link_reset(dest_link
);
2513 dest_link
->exp_msg_count
= msg_count
;
2514 dbg("Expecting %u tunnelled messages\n", msg_count
);
2517 } else if (dest_link
->exp_msg_count
== START_CHANGEOVER
) {
2518 msg_dbg(tunnel_msg
, "BLK/FIRST/<REC<");
2519 dest_link
->exp_msg_count
= msg_count
;
2520 dbg("Expecting %u tunnelled messages\n", msg_count
);
2525 /* Receive original message */
2527 if (dest_link
->exp_msg_count
== 0) {
2528 warn("Link switchover error, "
2529 "got too many tunnelled messages\n");
2530 msg_dbg(tunnel_msg
, "OVERDUE/DROP/<REC<");
2531 dbg_print_link(dest_link
, "LINK:");
2534 dest_link
->exp_msg_count
--;
2535 if (less(msg_seqno(msg
), dest_link
->reset_checkpoint
)) {
2536 msg_dbg(tunnel_msg
, "DROP/DUPL/<REC<");
2539 *buf
= buf_extract(tunnel_buf
, INT_H_SIZE
);
2541 msg_dbg(tunnel_msg
, "TNL<REC<");
2542 buf_discard(tunnel_buf
);
2545 warn("Link changeover error, original msg dropped\n");
2550 buf_discard(tunnel_buf
);
2555 * Bundler functionality:
2557 void tipc_link_recv_bundle(struct sk_buff
*buf
)
2559 u32 msgcount
= msg_msgcnt(buf_msg(buf
));
2560 u32 pos
= INT_H_SIZE
;
2561 struct sk_buff
*obuf
;
2563 msg_dbg(buf_msg(buf
), "<BNDL<: ");
2564 while (msgcount
--) {
2565 obuf
= buf_extract(buf
, pos
);
2567 warn("Link unable to unbundle message(s)\n");
2570 pos
+= align(msg_size(buf_msg(obuf
)));
2571 msg_dbg(buf_msg(obuf
), " /");
2572 tipc_net_route_msg(obuf
);
2578 * Fragmentation/defragmentation:
2583 * tipc_link_send_long_buf: Entry for buffers needing fragmentation.
2584 * The buffer is complete, inclusive total message length.
2585 * Returns user data length.
2587 int tipc_link_send_long_buf(struct link
*l_ptr
, struct sk_buff
*buf
)
2589 struct tipc_msg
*inmsg
= buf_msg(buf
);
2590 struct tipc_msg fragm_hdr
;
2591 u32 insize
= msg_size(inmsg
);
2592 u32 dsz
= msg_data_sz(inmsg
);
2593 unchar
*crs
= buf
->data
;
2595 u32 pack_sz
= link_max_pkt(l_ptr
);
2596 u32 fragm_sz
= pack_sz
- INT_H_SIZE
;
2598 u32 destaddr
= msg_destnode(inmsg
);
2600 if (msg_short(inmsg
))
2601 destaddr
= l_ptr
->addr
;
2603 if (msg_routed(inmsg
))
2604 msg_set_prevnode(inmsg
, tipc_own_addr
);
2606 /* Prepare reusable fragment header: */
2608 msg_init(&fragm_hdr
, MSG_FRAGMENTER
, FIRST_FRAGMENT
,
2609 TIPC_OK
, INT_H_SIZE
, destaddr
);
2610 msg_set_link_selector(&fragm_hdr
, msg_link_selector(inmsg
));
2611 msg_set_long_msgno(&fragm_hdr
, mod(l_ptr
->long_msg_seq_no
++));
2612 msg_set_fragm_no(&fragm_hdr
, fragm_no
);
2613 l_ptr
->stats
.sent_fragmented
++;
2615 /* Chop up message: */
2618 struct sk_buff
*fragm
;
2620 if (rest
<= fragm_sz
) {
2622 msg_set_type(&fragm_hdr
, LAST_FRAGMENT
);
2624 fragm
= buf_acquire(fragm_sz
+ INT_H_SIZE
);
2625 if (fragm
== NULL
) {
2626 warn("Link unable to fragment message\n");
2630 msg_set_size(&fragm_hdr
, fragm_sz
+ INT_H_SIZE
);
2631 memcpy(fragm
->data
, (unchar
*)&fragm_hdr
, INT_H_SIZE
);
2632 memcpy(fragm
->data
+ INT_H_SIZE
, crs
, fragm_sz
);
2634 /* Send queued messages first, if any: */
2636 l_ptr
->stats
.sent_fragments
++;
2637 tipc_link_send_buf(l_ptr
, fragm
);
2638 if (!tipc_link_is_up(l_ptr
))
2640 msg_set_fragm_no(&fragm_hdr
, ++fragm_no
);
2643 msg_set_type(&fragm_hdr
, FRAGMENT
);
2651 * A pending message being re-assembled must store certain values
2652 * to handle subsequent fragments correctly. The following functions
2653 * help storing these values in unused, available fields in the
2654 * pending message. This makes dynamic memory allocation unecessary.
2657 static void set_long_msg_seqno(struct sk_buff
*buf
, u32 seqno
)
2659 msg_set_seqno(buf_msg(buf
), seqno
);
2662 static u32
get_fragm_size(struct sk_buff
*buf
)
2664 return msg_ack(buf_msg(buf
));
2667 static void set_fragm_size(struct sk_buff
*buf
, u32 sz
)
2669 msg_set_ack(buf_msg(buf
), sz
);
2672 static u32
get_expected_frags(struct sk_buff
*buf
)
2674 return msg_bcast_ack(buf_msg(buf
));
2677 static void set_expected_frags(struct sk_buff
*buf
, u32 exp
)
2679 msg_set_bcast_ack(buf_msg(buf
), exp
);
2682 static u32
get_timer_cnt(struct sk_buff
*buf
)
2684 return msg_reroute_cnt(buf_msg(buf
));
2687 static void incr_timer_cnt(struct sk_buff
*buf
)
2689 msg_incr_reroute_cnt(buf_msg(buf
));
2693 * tipc_link_recv_fragment(): Called with node lock on. Returns
2694 * the reassembled buffer if message is complete.
2696 int tipc_link_recv_fragment(struct sk_buff
**pending
, struct sk_buff
**fb
,
2697 struct tipc_msg
**m
)
2699 struct sk_buff
*prev
= NULL
;
2700 struct sk_buff
*fbuf
= *fb
;
2701 struct tipc_msg
*fragm
= buf_msg(fbuf
);
2702 struct sk_buff
*pbuf
= *pending
;
2703 u32 long_msg_seq_no
= msg_long_msgno(fragm
);
2706 msg_dbg(fragm
,"FRG<REC<");
2708 /* Is there an incomplete message waiting for this fragment? */
2710 while (pbuf
&& ((msg_seqno(buf_msg(pbuf
)) != long_msg_seq_no
)
2711 || (msg_orignode(fragm
) != msg_orignode(buf_msg(pbuf
))))) {
2716 if (!pbuf
&& (msg_type(fragm
) == FIRST_FRAGMENT
)) {
2717 struct tipc_msg
*imsg
= (struct tipc_msg
*)msg_data(fragm
);
2718 u32 msg_sz
= msg_size(imsg
);
2719 u32 fragm_sz
= msg_data_sz(fragm
);
2720 u32 exp_fragm_cnt
= msg_sz
/fragm_sz
+ !!(msg_sz
% fragm_sz
);
2721 u32 max
= TIPC_MAX_USER_MSG_SIZE
+ LONG_H_SIZE
;
2722 if (msg_type(imsg
) == TIPC_MCAST_MSG
)
2723 max
= TIPC_MAX_USER_MSG_SIZE
+ MCAST_H_SIZE
;
2724 if (msg_size(imsg
) > max
) {
2725 msg_dbg(fragm
,"<REC<Oversized: ");
2729 pbuf
= buf_acquire(msg_size(imsg
));
2731 pbuf
->next
= *pending
;
2733 memcpy(pbuf
->data
, (unchar
*)imsg
, msg_data_sz(fragm
));
2735 /* Prepare buffer for subsequent fragments. */
2737 set_long_msg_seqno(pbuf
, long_msg_seq_no
);
2738 set_fragm_size(pbuf
,fragm_sz
);
2739 set_expected_frags(pbuf
,exp_fragm_cnt
- 1);
2741 warn("Link unable to reassemble fragmented message\n");
2745 } else if (pbuf
&& (msg_type(fragm
) != FIRST_FRAGMENT
)) {
2746 u32 dsz
= msg_data_sz(fragm
);
2747 u32 fsz
= get_fragm_size(pbuf
);
2748 u32 crs
= ((msg_fragm_no(fragm
) - 1) * fsz
);
2749 u32 exp_frags
= get_expected_frags(pbuf
) - 1;
2750 memcpy(pbuf
->data
+ crs
, msg_data(fragm
), dsz
);
2753 /* Is message complete? */
2755 if (exp_frags
== 0) {
2757 prev
->next
= pbuf
->next
;
2759 *pending
= pbuf
->next
;
2760 msg_reset_reroute_cnt(buf_msg(pbuf
));
2765 set_expected_frags(pbuf
,exp_frags
);
2768 dbg(" Discarding orphan fragment %x\n",fbuf
);
2769 msg_dbg(fragm
,"ORPHAN:");
2770 dbg("Pending long buffers:\n");
2771 dbg_print_buf_chain(*pending
);
2777 * link_check_defragm_bufs - flush stale incoming message fragments
2778 * @l_ptr: pointer to link
2781 static void link_check_defragm_bufs(struct link
*l_ptr
)
2783 struct sk_buff
*prev
= NULL
;
2784 struct sk_buff
*next
= NULL
;
2785 struct sk_buff
*buf
= l_ptr
->defragm_buf
;
2789 if (!link_working_working(l_ptr
))
2792 u32 cnt
= get_timer_cnt(buf
);
2796 incr_timer_cnt(buf
);
2799 dbg(" Discarding incomplete long buffer\n");
2800 msg_dbg(buf_msg(buf
), "LONG:");
2801 dbg_print_link(l_ptr
, "curr:");
2802 dbg("Pending long buffers:\n");
2803 dbg_print_buf_chain(l_ptr
->defragm_buf
);
2805 prev
->next
= buf
->next
;
2807 l_ptr
->defragm_buf
= buf
->next
;
2816 static void link_set_supervision_props(struct link
*l_ptr
, u32 tolerance
)
2818 l_ptr
->tolerance
= tolerance
;
2819 l_ptr
->continuity_interval
=
2820 ((tolerance
/ 4) > 500) ? 500 : tolerance
/ 4;
2821 l_ptr
->abort_limit
= tolerance
/ (l_ptr
->continuity_interval
/ 4);
2825 void tipc_link_set_queue_limits(struct link
*l_ptr
, u32 window
)
2827 /* Data messages from this node, inclusive FIRST_FRAGM */
2828 l_ptr
->queue_limit
[DATA_LOW
] = window
;
2829 l_ptr
->queue_limit
[DATA_MEDIUM
] = (window
/ 3) * 4;
2830 l_ptr
->queue_limit
[DATA_HIGH
] = (window
/ 3) * 5;
2831 l_ptr
->queue_limit
[DATA_CRITICAL
] = (window
/ 3) * 6;
2832 /* Transiting data messages,inclusive FIRST_FRAGM */
2833 l_ptr
->queue_limit
[DATA_LOW
+ 4] = 300;
2834 l_ptr
->queue_limit
[DATA_MEDIUM
+ 4] = 600;
2835 l_ptr
->queue_limit
[DATA_HIGH
+ 4] = 900;
2836 l_ptr
->queue_limit
[DATA_CRITICAL
+ 4] = 1200;
2837 l_ptr
->queue_limit
[CONN_MANAGER
] = 1200;
2838 l_ptr
->queue_limit
[ROUTE_DISTRIBUTOR
] = 1200;
2839 l_ptr
->queue_limit
[CHANGEOVER_PROTOCOL
] = 2500;
2840 l_ptr
->queue_limit
[NAME_DISTRIBUTOR
] = 3000;
2841 /* FRAGMENT and LAST_FRAGMENT packets */
2842 l_ptr
->queue_limit
[MSG_FRAGMENTER
] = 4000;
2846 * link_find_link - locate link by name
2847 * @name - ptr to link name string
2848 * @node - ptr to area to be filled with ptr to associated node
2850 * Caller must hold 'tipc_net_lock' to ensure node and bearer are not deleted;
2851 * this also prevents link deletion.
2853 * Returns pointer to link (or 0 if invalid link name).
2856 static struct link
*link_find_link(const char *name
, struct node
**node
)
2858 struct link_name link_name_parts
;
2859 struct bearer
*b_ptr
;
2862 if (!link_name_validate(name
, &link_name_parts
))
2865 b_ptr
= tipc_bearer_find_interface(link_name_parts
.if_local
);
2869 *node
= tipc_node_find(link_name_parts
.addr_peer
);
2873 l_ptr
= (*node
)->links
[b_ptr
->identity
];
2874 if (!l_ptr
|| strcmp(l_ptr
->name
, name
))
2880 struct sk_buff
*tipc_link_cmd_config(const void *req_tlv_area
, int req_tlv_space
,
2883 struct tipc_link_config
*args
;
2889 if (!TLV_CHECK(req_tlv_area
, req_tlv_space
, TIPC_TLV_LINK_CONFIG
))
2890 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR
);
2892 args
= (struct tipc_link_config
*)TLV_DATA(req_tlv_area
);
2893 new_value
= ntohl(args
->value
);
2895 if (!strcmp(args
->name
, tipc_bclink_name
)) {
2896 if ((cmd
== TIPC_CMD_SET_LINK_WINDOW
) &&
2897 (tipc_bclink_set_queue_limits(new_value
) == 0))
2898 return tipc_cfg_reply_none();
2899 return tipc_cfg_reply_error_string(TIPC_CFG_NOT_SUPPORTED
2900 " (cannot change setting on broadcast link)");
2903 read_lock_bh(&tipc_net_lock
);
2904 l_ptr
= link_find_link(args
->name
, &node
);
2906 read_unlock_bh(&tipc_net_lock
);
2907 return tipc_cfg_reply_error_string("link not found");
2910 tipc_node_lock(node
);
2913 case TIPC_CMD_SET_LINK_TOL
:
2914 if ((new_value
>= TIPC_MIN_LINK_TOL
) &&
2915 (new_value
<= TIPC_MAX_LINK_TOL
)) {
2916 link_set_supervision_props(l_ptr
, new_value
);
2917 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
2918 0, 0, new_value
, 0, 0);
2922 case TIPC_CMD_SET_LINK_PRI
:
2923 if ((new_value
>= TIPC_MIN_LINK_PRI
) &&
2924 (new_value
<= TIPC_MAX_LINK_PRI
)) {
2925 l_ptr
->priority
= new_value
;
2926 tipc_link_send_proto_msg(l_ptr
, STATE_MSG
,
2927 0, 0, 0, new_value
, 0);
2931 case TIPC_CMD_SET_LINK_WINDOW
:
2932 if ((new_value
>= TIPC_MIN_LINK_WIN
) &&
2933 (new_value
<= TIPC_MAX_LINK_WIN
)) {
2934 tipc_link_set_queue_limits(l_ptr
, new_value
);
2939 tipc_node_unlock(node
);
2941 read_unlock_bh(&tipc_net_lock
);
2943 return tipc_cfg_reply_error_string("cannot change link setting");
2945 return tipc_cfg_reply_none();
2949 * link_reset_statistics - reset link statistics
2950 * @l_ptr: pointer to link
2953 static void link_reset_statistics(struct link
*l_ptr
)
2955 memset(&l_ptr
->stats
, 0, sizeof(l_ptr
->stats
));
2956 l_ptr
->stats
.sent_info
= l_ptr
->next_out_no
;
2957 l_ptr
->stats
.recv_info
= l_ptr
->next_in_no
;
2960 struct sk_buff
*tipc_link_cmd_reset_stats(const void *req_tlv_area
, int req_tlv_space
)
2966 if (!TLV_CHECK(req_tlv_area
, req_tlv_space
, TIPC_TLV_LINK_NAME
))
2967 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR
);
2969 link_name
= (char *)TLV_DATA(req_tlv_area
);
2970 if (!strcmp(link_name
, tipc_bclink_name
)) {
2971 if (tipc_bclink_reset_stats())
2972 return tipc_cfg_reply_error_string("link not found");
2973 return tipc_cfg_reply_none();
2976 read_lock_bh(&tipc_net_lock
);
2977 l_ptr
= link_find_link(link_name
, &node
);
2979 read_unlock_bh(&tipc_net_lock
);
2980 return tipc_cfg_reply_error_string("link not found");
2983 tipc_node_lock(node
);
2984 link_reset_statistics(l_ptr
);
2985 tipc_node_unlock(node
);
2986 read_unlock_bh(&tipc_net_lock
);
2987 return tipc_cfg_reply_none();
2991 * percent - convert count to a percentage of total (rounding up or down)
2994 static u32
percent(u32 count
, u32 total
)
2996 return (count
* 100 + (total
/ 2)) / total
;
3000 * tipc_link_stats - print link statistics
3002 * @buf: print buffer area
3003 * @buf_size: size of print buffer area
3005 * Returns length of print buffer data string (or 0 if error)
3008 static int tipc_link_stats(const char *name
, char *buf
, const u32 buf_size
)
3010 struct print_buf pb
;
3014 u32 profile_total
= 0;
3016 if (!strcmp(name
, tipc_bclink_name
))
3017 return tipc_bclink_stats(buf
, buf_size
);
3019 tipc_printbuf_init(&pb
, buf
, buf_size
);
3021 read_lock_bh(&tipc_net_lock
);
3022 l_ptr
= link_find_link(name
, &node
);
3024 read_unlock_bh(&tipc_net_lock
);
3027 tipc_node_lock(node
);
3029 if (tipc_link_is_active(l_ptr
))
3031 else if (tipc_link_is_up(l_ptr
))
3035 tipc_printf(&pb
, "Link <%s>\n"
3036 " %s MTU:%u Priority:%u Tolerance:%u ms"
3037 " Window:%u packets\n",
3038 l_ptr
->name
, status
, link_max_pkt(l_ptr
),
3039 l_ptr
->priority
, l_ptr
->tolerance
, l_ptr
->queue_limit
[0]);
3040 tipc_printf(&pb
, " RX packets:%u fragments:%u/%u bundles:%u/%u\n",
3041 l_ptr
->next_in_no
- l_ptr
->stats
.recv_info
,
3042 l_ptr
->stats
.recv_fragments
,
3043 l_ptr
->stats
.recv_fragmented
,
3044 l_ptr
->stats
.recv_bundles
,
3045 l_ptr
->stats
.recv_bundled
);
3046 tipc_printf(&pb
, " TX packets:%u fragments:%u/%u bundles:%u/%u\n",
3047 l_ptr
->next_out_no
- l_ptr
->stats
.sent_info
,
3048 l_ptr
->stats
.sent_fragments
,
3049 l_ptr
->stats
.sent_fragmented
,
3050 l_ptr
->stats
.sent_bundles
,
3051 l_ptr
->stats
.sent_bundled
);
3052 profile_total
= l_ptr
->stats
.msg_length_counts
;
3055 tipc_printf(&pb
, " TX profile sample:%u packets average:%u octets\n"
3056 " 0-64:%u%% -256:%u%% -1024:%u%% -4096:%u%% "
3057 "-16354:%u%% -32768:%u%% -66000:%u%%\n",
3058 l_ptr
->stats
.msg_length_counts
,
3059 l_ptr
->stats
.msg_lengths_total
/ profile_total
,
3060 percent(l_ptr
->stats
.msg_length_profile
[0], profile_total
),
3061 percent(l_ptr
->stats
.msg_length_profile
[1], profile_total
),
3062 percent(l_ptr
->stats
.msg_length_profile
[2], profile_total
),
3063 percent(l_ptr
->stats
.msg_length_profile
[3], profile_total
),
3064 percent(l_ptr
->stats
.msg_length_profile
[4], profile_total
),
3065 percent(l_ptr
->stats
.msg_length_profile
[5], profile_total
),
3066 percent(l_ptr
->stats
.msg_length_profile
[6], profile_total
));
3067 tipc_printf(&pb
, " RX states:%u probes:%u naks:%u defs:%u dups:%u\n",
3068 l_ptr
->stats
.recv_states
,
3069 l_ptr
->stats
.recv_probes
,
3070 l_ptr
->stats
.recv_nacks
,
3071 l_ptr
->stats
.deferred_recv
,
3072 l_ptr
->stats
.duplicates
);
3073 tipc_printf(&pb
, " TX states:%u probes:%u naks:%u acks:%u dups:%u\n",
3074 l_ptr
->stats
.sent_states
,
3075 l_ptr
->stats
.sent_probes
,
3076 l_ptr
->stats
.sent_nacks
,
3077 l_ptr
->stats
.sent_acks
,
3078 l_ptr
->stats
.retransmitted
);
3079 tipc_printf(&pb
, " Congestion bearer:%u link:%u Send queue max:%u avg:%u\n",
3080 l_ptr
->stats
.bearer_congs
,
3081 l_ptr
->stats
.link_congs
,
3082 l_ptr
->stats
.max_queue_sz
,
3083 l_ptr
->stats
.queue_sz_counts
3084 ? (l_ptr
->stats
.accu_queue_sz
/ l_ptr
->stats
.queue_sz_counts
)
3087 tipc_node_unlock(node
);
3088 read_unlock_bh(&tipc_net_lock
);
3089 return tipc_printbuf_validate(&pb
);
3092 #define MAX_LINK_STATS_INFO 2000
3094 struct sk_buff
*tipc_link_cmd_show_stats(const void *req_tlv_area
, int req_tlv_space
)
3096 struct sk_buff
*buf
;
3097 struct tlv_desc
*rep_tlv
;
3100 if (!TLV_CHECK(req_tlv_area
, req_tlv_space
, TIPC_TLV_LINK_NAME
))
3101 return tipc_cfg_reply_error_string(TIPC_CFG_TLV_ERROR
);
3103 buf
= tipc_cfg_reply_alloc(TLV_SPACE(MAX_LINK_STATS_INFO
));
3107 rep_tlv
= (struct tlv_desc
*)buf
->data
;
3109 str_len
= tipc_link_stats((char *)TLV_DATA(req_tlv_area
),
3110 (char *)TLV_DATA(rep_tlv
), MAX_LINK_STATS_INFO
);
3113 return tipc_cfg_reply_error_string("link not found");
3116 skb_put(buf
, TLV_SPACE(str_len
));
3117 TLV_SET(rep_tlv
, TIPC_TLV_ULTRA_STRING
, NULL
, str_len
);
3123 int link_control(const char *name
, u32 op
, u32 val
)
3131 a
= link_name2addr(name
, &bearer_id
);
3132 read_lock_bh(&tipc_net_lock
);
3133 node
= tipc_node_find(a
);
3135 tipc_node_lock(node
);
3136 l_ptr
= node
->links
[bearer_id
];
3138 if (op
== TIPC_REMOVE_LINK
) {
3139 struct bearer
*b_ptr
= l_ptr
->b_ptr
;
3140 spin_lock_bh(&b_ptr
->publ
.lock
);
3141 tipc_link_delete(l_ptr
);
3142 spin_unlock_bh(&b_ptr
->publ
.lock
);
3144 if (op
== TIPC_CMD_BLOCK_LINK
) {
3145 tipc_link_reset(l_ptr
);
3148 if (op
== TIPC_CMD_UNBLOCK_LINK
) {
3153 tipc_node_unlock(node
);
3155 read_unlock_bh(&tipc_net_lock
);
3161 * tipc_link_get_max_pkt - get maximum packet size to use when sending to destination
3162 * @dest: network address of destination node
3163 * @selector: used to select from set of active links
3165 * If no active link can be found, uses default maximum packet size.
3168 u32
tipc_link_get_max_pkt(u32 dest
, u32 selector
)
3172 u32 res
= MAX_PKT_DEFAULT
;
3174 if (dest
== tipc_own_addr
)
3175 return MAX_MSG_SIZE
;
3177 read_lock_bh(&tipc_net_lock
);
3178 n_ptr
= tipc_node_select(dest
, selector
);
3180 tipc_node_lock(n_ptr
);
3181 l_ptr
= n_ptr
->active_links
[selector
& 1];
3183 res
= link_max_pkt(l_ptr
);
3184 tipc_node_unlock(n_ptr
);
3186 read_unlock_bh(&tipc_net_lock
);
3191 static void link_dump_rec_queue(struct link
*l_ptr
)
3193 struct sk_buff
*crs
;
3195 if (!l_ptr
->oldest_deferred_in
) {
3196 info("Reception queue empty\n");
3199 info("Contents of Reception queue:\n");
3200 crs
= l_ptr
->oldest_deferred_in
;
3202 if (crs
->data
== (void *)0x0000a3a3) {
3203 info("buffer %x invalid\n", crs
);
3206 msg_dbg(buf_msg(crs
), "In rec queue: \n");
3212 static void link_dump_send_queue(struct link
*l_ptr
)
3214 if (l_ptr
->next_out
) {
3215 info("\nContents of unsent queue:\n");
3216 dbg_print_buf_chain(l_ptr
->next_out
);
3218 info("\nContents of send queue:\n");
3219 if (l_ptr
->first_out
) {
3220 dbg_print_buf_chain(l_ptr
->first_out
);
3222 info("Empty send queue\n");
3225 static void link_print(struct link
*l_ptr
, struct print_buf
*buf
,
3228 tipc_printf(buf
, str
);
3229 if (link_reset_reset(l_ptr
) || link_reset_unknown(l_ptr
))
3231 tipc_printf(buf
, "Link %x<%s>:",
3232 l_ptr
->addr
, l_ptr
->b_ptr
->publ
.name
);
3233 tipc_printf(buf
, ": NXO(%u):", mod(l_ptr
->next_out_no
));
3234 tipc_printf(buf
, "NXI(%u):", mod(l_ptr
->next_in_no
));
3235 tipc_printf(buf
, "SQUE");
3236 if (l_ptr
->first_out
) {
3237 tipc_printf(buf
, "[%u..", msg_seqno(buf_msg(l_ptr
->first_out
)));
3238 if (l_ptr
->next_out
)
3239 tipc_printf(buf
, "%u..",
3240 msg_seqno(buf_msg(l_ptr
->next_out
)));
3241 tipc_printf(buf
, "%u]",
3243 (l_ptr
->last_out
)), l_ptr
->out_queue_size
);
3244 if ((mod(msg_seqno(buf_msg(l_ptr
->last_out
)) -
3245 msg_seqno(buf_msg(l_ptr
->first_out
)))
3246 != (l_ptr
->out_queue_size
- 1))
3247 || (l_ptr
->last_out
->next
!= 0)) {
3248 tipc_printf(buf
, "\nSend queue inconsistency\n");
3249 tipc_printf(buf
, "first_out= %x ", l_ptr
->first_out
);
3250 tipc_printf(buf
, "next_out= %x ", l_ptr
->next_out
);
3251 tipc_printf(buf
, "last_out= %x ", l_ptr
->last_out
);
3252 link_dump_send_queue(l_ptr
);
3255 tipc_printf(buf
, "[]");
3256 tipc_printf(buf
, "SQSIZ(%u)", l_ptr
->out_queue_size
);
3257 if (l_ptr
->oldest_deferred_in
) {
3258 u32 o
= msg_seqno(buf_msg(l_ptr
->oldest_deferred_in
));
3259 u32 n
= msg_seqno(buf_msg(l_ptr
->newest_deferred_in
));
3260 tipc_printf(buf
, ":RQUE[%u..%u]", o
, n
);
3261 if (l_ptr
->deferred_inqueue_sz
!= mod((n
+ 1) - o
)) {
3262 tipc_printf(buf
, ":RQSIZ(%u)",
3263 l_ptr
->deferred_inqueue_sz
);
3266 if (link_working_unknown(l_ptr
))
3267 tipc_printf(buf
, ":WU");
3268 if (link_reset_reset(l_ptr
))
3269 tipc_printf(buf
, ":RR");
3270 if (link_reset_unknown(l_ptr
))
3271 tipc_printf(buf
, ":RU");
3272 if (link_working_working(l_ptr
))
3273 tipc_printf(buf
, ":WW");
3274 tipc_printf(buf
, "\n");