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