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