2 * net/tipc/group.c: TIPC group messaging code
4 * Copyright (c) 2017, Ericsson AB
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions are met:
10 * 1. Redistributions of source code must retain the above copyright
11 * notice, this list of conditions and the following disclaimer.
12 * 2. Redistributions in binary form must reproduce the above copyright
13 * notice, this list of conditions and the following disclaimer in the
14 * documentation and/or other materials provided with the distribution.
15 * 3. Neither the names of the copyright holders nor the names of its
16 * contributors may be used to endorse or promote products derived from
17 * this software without specific prior written permission.
19 * Alternatively, this software may be distributed under the terms of the
20 * GNU General Public License ("GPL") version 2 as published by the Free
21 * Software Foundation.
23 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
24 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
25 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
26 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
27 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
28 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
29 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
30 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
31 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
32 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
33 * POSSIBILITY OF SUCH DAMAGE.
44 #include "name_table.h"
47 #define ADV_UNIT (((MAX_MSG_SIZE + MAX_H_SIZE) / FLOWCTL_BLK_SZ) + 1)
48 #define ADV_IDLE ADV_UNIT
49 #define ADV_ACTIVE (ADV_UNIT * 12)
65 struct rb_node tree_node
;
66 struct list_head list
;
67 struct list_head small_win
;
68 struct sk_buff
*event_msg
;
69 struct sk_buff_head deferredq
;
70 struct tipc_group
*group
;
84 struct rb_root members
;
85 struct list_head small_win
;
86 struct list_head pending
;
87 struct list_head active
;
88 struct tipc_nlist dests
;
105 static void tipc_group_proto_xmit(struct tipc_group
*grp
, struct tipc_member
*m
,
106 int mtyp
, struct sk_buff_head
*xmitq
);
108 static void tipc_group_decr_active(struct tipc_group
*grp
,
109 struct tipc_member
*m
)
111 if (m
->state
== MBR_ACTIVE
|| m
->state
== MBR_RECLAIMING
||
112 m
->state
== MBR_REMITTED
)
116 static int tipc_group_rcvbuf_limit(struct tipc_group
*grp
)
118 int max_active
, active_pool
, idle_pool
;
119 int mcnt
= grp
->member_cnt
+ 1;
121 /* Limit simultaneous reception from other members */
122 max_active
= min(mcnt
/ 8, 64);
123 max_active
= max(max_active
, 16);
124 grp
->max_active
= max_active
;
126 /* Reserve blocks for active and idle members */
127 active_pool
= max_active
* ADV_ACTIVE
;
128 idle_pool
= (mcnt
- max_active
) * ADV_IDLE
;
130 /* Scale to bytes, considering worst-case truesize/msgsize ratio */
131 return (active_pool
+ idle_pool
) * FLOWCTL_BLK_SZ
* 4;
134 u16
tipc_group_bc_snd_nxt(struct tipc_group
*grp
)
136 return grp
->bc_snd_nxt
;
139 static bool tipc_group_is_receiver(struct tipc_member
*m
)
141 return m
->state
!= MBR_QUARANTINED
&& m
->state
!= MBR_LEAVING
;
144 static bool tipc_group_is_sender(struct tipc_member
*m
)
146 return m
&& m
->state
>= MBR_JOINED
;
149 u32
tipc_group_exclude(struct tipc_group
*grp
)
156 int tipc_group_size(struct tipc_group
*grp
)
158 return grp
->member_cnt
;
161 struct tipc_group
*tipc_group_create(struct net
*net
, u32 portid
,
162 struct tipc_group_req
*mreq
)
164 struct tipc_group
*grp
;
165 u32 type
= mreq
->type
;
167 grp
= kzalloc(sizeof(*grp
), GFP_ATOMIC
);
170 tipc_nlist_init(&grp
->dests
, tipc_own_addr(net
));
171 INIT_LIST_HEAD(&grp
->small_win
);
172 INIT_LIST_HEAD(&grp
->active
);
173 INIT_LIST_HEAD(&grp
->pending
);
174 grp
->members
= RB_ROOT
;
176 grp
->portid
= portid
;
177 grp
->domain
= addr_domain(net
, mreq
->scope
);
179 grp
->instance
= mreq
->instance
;
180 grp
->scope
= mreq
->scope
;
181 grp
->loopback
= mreq
->flags
& TIPC_GROUP_LOOPBACK
;
182 grp
->events
= mreq
->flags
& TIPC_GROUP_MEMBER_EVTS
;
183 if (tipc_topsrv_kern_subscr(net
, portid
, type
, 0, ~0, &grp
->subid
))
189 void tipc_group_delete(struct net
*net
, struct tipc_group
*grp
)
191 struct rb_root
*tree
= &grp
->members
;
192 struct tipc_member
*m
, *tmp
;
193 struct sk_buff_head xmitq
;
195 __skb_queue_head_init(&xmitq
);
197 rbtree_postorder_for_each_entry_safe(m
, tmp
, tree
, tree_node
) {
198 tipc_group_proto_xmit(grp
, m
, GRP_LEAVE_MSG
, &xmitq
);
202 tipc_node_distr_xmit(net
, &xmitq
);
203 tipc_nlist_purge(&grp
->dests
);
204 tipc_topsrv_kern_unsubscr(net
, grp
->subid
);
208 struct tipc_member
*tipc_group_find_member(struct tipc_group
*grp
,
211 struct rb_node
*n
= grp
->members
.rb_node
;
212 u64 nkey
, key
= (u64
)node
<< 32 | port
;
213 struct tipc_member
*m
;
216 m
= container_of(n
, struct tipc_member
, tree_node
);
217 nkey
= (u64
)m
->node
<< 32 | m
->port
;
228 static struct tipc_member
*tipc_group_find_dest(struct tipc_group
*grp
,
231 struct tipc_member
*m
;
233 m
= tipc_group_find_member(grp
, node
, port
);
234 if (m
&& tipc_group_is_receiver(m
))
239 static struct tipc_member
*tipc_group_find_node(struct tipc_group
*grp
,
242 struct tipc_member
*m
;
245 for (n
= rb_first(&grp
->members
); n
; n
= rb_next(n
)) {
246 m
= container_of(n
, struct tipc_member
, tree_node
);
253 static void tipc_group_add_to_tree(struct tipc_group
*grp
,
254 struct tipc_member
*m
)
256 u64 nkey
, key
= (u64
)m
->node
<< 32 | m
->port
;
257 struct rb_node
**n
, *parent
= NULL
;
258 struct tipc_member
*tmp
;
260 n
= &grp
->members
.rb_node
;
262 tmp
= container_of(*n
, struct tipc_member
, tree_node
);
264 tmp
= container_of(parent
, struct tipc_member
, tree_node
);
265 nkey
= (u64
)tmp
->node
<< 32 | tmp
->port
;
273 rb_link_node(&m
->tree_node
, parent
, n
);
274 rb_insert_color(&m
->tree_node
, &grp
->members
);
277 static struct tipc_member
*tipc_group_create_member(struct tipc_group
*grp
,
281 struct tipc_member
*m
;
283 m
= kzalloc(sizeof(*m
), GFP_ATOMIC
);
286 INIT_LIST_HEAD(&m
->list
);
287 INIT_LIST_HEAD(&m
->small_win
);
288 __skb_queue_head_init(&m
->deferredq
);
292 m
->bc_acked
= grp
->bc_snd_nxt
- 1;
294 tipc_group_add_to_tree(grp
, m
);
295 tipc_nlist_add(&grp
->dests
, m
->node
);
300 void tipc_group_add_member(struct tipc_group
*grp
, u32 node
, u32 port
)
302 tipc_group_create_member(grp
, node
, port
, MBR_DISCOVERED
);
305 static void tipc_group_delete_member(struct tipc_group
*grp
,
306 struct tipc_member
*m
)
308 rb_erase(&m
->tree_node
, &grp
->members
);
311 /* Check if we were waiting for replicast ack from this member */
312 if (grp
->bc_ackers
&& less(m
->bc_acked
, grp
->bc_snd_nxt
- 1))
315 list_del_init(&m
->list
);
316 list_del_init(&m
->small_win
);
317 tipc_group_decr_active(grp
, m
);
319 /* If last member on a node, remove node from dest list */
320 if (!tipc_group_find_node(grp
, m
->node
))
321 tipc_nlist_del(&grp
->dests
, m
->node
);
326 struct tipc_nlist
*tipc_group_dests(struct tipc_group
*grp
)
331 void tipc_group_self(struct tipc_group
*grp
, struct tipc_name_seq
*seq
,
334 seq
->type
= grp
->type
;
335 seq
->lower
= grp
->instance
;
336 seq
->upper
= grp
->instance
;
340 void tipc_group_update_member(struct tipc_member
*m
, int len
)
342 struct tipc_group
*grp
= m
->group
;
343 struct tipc_member
*_m
, *tmp
;
345 if (!tipc_group_is_receiver(m
))
350 if (m
->window
>= ADV_IDLE
)
353 list_del_init(&m
->small_win
);
355 /* Sort member into small_window members' list */
356 list_for_each_entry_safe(_m
, tmp
, &grp
->small_win
, small_win
) {
357 if (_m
->window
> m
->window
)
360 list_add_tail(&m
->small_win
, &_m
->small_win
);
363 void tipc_group_update_bc_members(struct tipc_group
*grp
, int len
, bool ack
)
365 u16 prev
= grp
->bc_snd_nxt
- 1;
366 struct tipc_member
*m
;
370 for (n
= rb_first(&grp
->members
); n
; n
= rb_next(n
)) {
371 m
= container_of(n
, struct tipc_member
, tree_node
);
372 if (tipc_group_is_receiver(m
)) {
373 tipc_group_update_member(m
, len
);
379 /* Mark number of acknowledges to expect, if any */
381 grp
->bc_ackers
= ackers
;
385 bool tipc_group_cong(struct tipc_group
*grp
, u32 dnode
, u32 dport
,
386 int len
, struct tipc_member
**mbr
)
388 struct sk_buff_head xmitq
;
389 struct tipc_member
*m
;
392 m
= tipc_group_find_dest(grp
, dnode
, dport
);
398 if (m
->window
>= len
)
400 m
->usr_pending
= true;
402 /* If not fully advertised, do it now to prevent mutual blocking */
405 if (state
< MBR_JOINED
)
407 if (state
== MBR_JOINED
&& adv
== ADV_IDLE
)
409 if (state
== MBR_ACTIVE
&& adv
== ADV_ACTIVE
)
411 if (state
== MBR_PENDING
&& adv
== ADV_IDLE
)
413 skb_queue_head_init(&xmitq
);
414 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, &xmitq
);
415 tipc_node_distr_xmit(grp
->net
, &xmitq
);
419 bool tipc_group_bc_cong(struct tipc_group
*grp
, int len
)
421 struct tipc_member
*m
= NULL
;
423 /* If prev bcast was replicast, reject until all receivers have acked */
427 if (list_empty(&grp
->small_win
))
430 m
= list_first_entry(&grp
->small_win
, struct tipc_member
, small_win
);
431 if (m
->window
>= len
)
434 return tipc_group_cong(grp
, m
->node
, m
->port
, len
, &m
);
437 /* tipc_group_sort_msg() - sort msg into queue by bcast sequence number
439 static void tipc_group_sort_msg(struct sk_buff
*skb
, struct sk_buff_head
*defq
)
441 struct tipc_msg
*_hdr
, *hdr
= buf_msg(skb
);
442 u16 bc_seqno
= msg_grp_bc_seqno(hdr
);
443 struct sk_buff
*_skb
, *tmp
;
444 int mtyp
= msg_type(hdr
);
446 /* Bcast/mcast may be bypassed by ucast or other bcast, - sort it in */
447 if (mtyp
== TIPC_GRP_BCAST_MSG
|| mtyp
== TIPC_GRP_MCAST_MSG
) {
448 skb_queue_walk_safe(defq
, _skb
, tmp
) {
449 _hdr
= buf_msg(_skb
);
450 if (!less(bc_seqno
, msg_grp_bc_seqno(_hdr
)))
452 __skb_queue_before(defq
, _skb
, skb
);
455 /* Bcast was not bypassed, - add to tail */
457 /* Unicasts are never bypassed, - always add to tail */
458 __skb_queue_tail(defq
, skb
);
461 /* tipc_group_filter_msg() - determine if we should accept arriving message
463 void tipc_group_filter_msg(struct tipc_group
*grp
, struct sk_buff_head
*inputq
,
464 struct sk_buff_head
*xmitq
)
466 struct sk_buff
*skb
= __skb_dequeue(inputq
);
467 bool ack
, deliver
, update
, leave
= false;
468 struct sk_buff_head
*defq
;
469 struct tipc_member
*m
;
470 struct tipc_msg
*hdr
;
478 node
= msg_orignode(hdr
);
479 port
= msg_origport(hdr
);
481 if (!msg_in_group(hdr
))
484 m
= tipc_group_find_member(grp
, node
, port
);
485 if (!tipc_group_is_sender(m
))
488 if (less(msg_grp_bc_seqno(hdr
), m
->bc_rcv_nxt
))
491 TIPC_SKB_CB(skb
)->orig_member
= m
->instance
;
492 defq
= &m
->deferredq
;
493 tipc_group_sort_msg(skb
, defq
);
495 while ((skb
= skb_peek(defq
))) {
497 mtyp
= msg_type(hdr
);
498 blks
= msg_blocks(hdr
);
503 if (more(msg_grp_bc_seqno(hdr
), m
->bc_rcv_nxt
))
506 /* Decide what to do with message */
508 case TIPC_GRP_MCAST_MSG
:
509 if (msg_nameinst(hdr
) != grp
->instance
) {
514 case TIPC_GRP_BCAST_MSG
:
516 ack
= msg_grp_bc_ack_req(hdr
);
518 case TIPC_GRP_UCAST_MSG
:
520 case TIPC_GRP_MEMBER_EVT
:
521 if (m
->state
== MBR_LEAVING
)
530 /* Execute decisions */
533 __skb_queue_tail(inputq
, skb
);
538 tipc_group_proto_xmit(grp
, m
, GRP_ACK_MSG
, xmitq
);
541 __skb_queue_purge(defq
);
542 tipc_group_delete_member(grp
, m
);
548 tipc_group_update_rcv_win(grp
, blks
, node
, port
, xmitq
);
555 void tipc_group_update_rcv_win(struct tipc_group
*grp
, int blks
, u32 node
,
556 u32 port
, struct sk_buff_head
*xmitq
)
558 struct list_head
*active
= &grp
->active
;
559 int max_active
= grp
->max_active
;
560 int reclaim_limit
= max_active
* 3 / 4;
561 int active_cnt
= grp
->active_cnt
;
562 struct tipc_member
*m
, *rm
, *pm
;
564 m
= tipc_group_find_member(grp
, node
, port
);
568 m
->advertised
-= blks
;
572 /* First, decide if member can go active */
573 if (active_cnt
<= max_active
) {
574 m
->state
= MBR_ACTIVE
;
575 list_add_tail(&m
->list
, active
);
577 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, xmitq
);
579 m
->state
= MBR_PENDING
;
580 list_add_tail(&m
->list
, &grp
->pending
);
583 if (active_cnt
< reclaim_limit
)
586 /* Reclaim from oldest active member, if possible */
587 if (!list_empty(active
)) {
588 rm
= list_first_entry(active
, struct tipc_member
, list
);
589 rm
->state
= MBR_RECLAIMING
;
590 list_del_init(&rm
->list
);
591 tipc_group_proto_xmit(grp
, rm
, GRP_RECLAIM_MSG
, xmitq
);
594 /* Nobody to reclaim from; - revert oldest pending to JOINED */
595 pm
= list_first_entry(&grp
->pending
, struct tipc_member
, list
);
596 list_del_init(&pm
->list
);
597 pm
->state
= MBR_JOINED
;
598 tipc_group_proto_xmit(grp
, pm
, GRP_ADV_MSG
, xmitq
);
601 if (!list_is_last(&m
->list
, &grp
->active
))
602 list_move_tail(&m
->list
, &grp
->active
);
603 if (m
->advertised
> (ADV_ACTIVE
* 3 / 4))
605 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, xmitq
);
608 if (m
->advertised
> ADV_IDLE
)
610 m
->state
= MBR_JOINED
;
612 if (m
->advertised
< ADV_IDLE
) {
613 pr_warn_ratelimited("Rcv unexpected msg after REMIT\n");
614 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, xmitq
);
617 if (list_empty(&grp
->pending
))
620 /* Set oldest pending member to active and advertise */
621 pm
= list_first_entry(&grp
->pending
, struct tipc_member
, list
);
622 pm
->state
= MBR_ACTIVE
;
623 list_move_tail(&pm
->list
, &grp
->active
);
625 tipc_group_proto_xmit(grp
, pm
, GRP_ADV_MSG
, xmitq
);
636 static void tipc_group_proto_xmit(struct tipc_group
*grp
, struct tipc_member
*m
,
637 int mtyp
, struct sk_buff_head
*xmitq
)
639 struct tipc_msg
*hdr
;
643 skb
= tipc_msg_create(GROUP_PROTOCOL
, mtyp
, INT_H_SIZE
, 0,
644 m
->node
, tipc_own_addr(grp
->net
),
645 m
->port
, grp
->portid
, 0);
649 if (m
->state
== MBR_ACTIVE
)
650 adv
= ADV_ACTIVE
- m
->advertised
;
651 else if (m
->state
== MBR_JOINED
|| m
->state
== MBR_PENDING
)
652 adv
= ADV_IDLE
- m
->advertised
;
656 if (mtyp
== GRP_JOIN_MSG
) {
657 msg_set_grp_bc_syncpt(hdr
, grp
->bc_snd_nxt
);
658 msg_set_adv_win(hdr
, adv
);
659 m
->advertised
+= adv
;
660 } else if (mtyp
== GRP_LEAVE_MSG
) {
661 msg_set_grp_bc_syncpt(hdr
, grp
->bc_snd_nxt
);
662 } else if (mtyp
== GRP_ADV_MSG
) {
663 msg_set_adv_win(hdr
, adv
);
664 m
->advertised
+= adv
;
665 } else if (mtyp
== GRP_ACK_MSG
) {
666 msg_set_grp_bc_acked(hdr
, m
->bc_rcv_nxt
);
667 } else if (mtyp
== GRP_REMIT_MSG
) {
668 msg_set_grp_remitted(hdr
, m
->window
);
670 msg_set_dest_droppable(hdr
, true);
671 __skb_queue_tail(xmitq
, skb
);
674 void tipc_group_proto_rcv(struct tipc_group
*grp
, bool *usr_wakeup
,
675 struct tipc_msg
*hdr
, struct sk_buff_head
*inputq
,
676 struct sk_buff_head
*xmitq
)
678 u32 node
= msg_orignode(hdr
);
679 u32 port
= msg_origport(hdr
);
680 struct tipc_member
*m
, *pm
;
681 struct tipc_msg
*ehdr
;
682 u16 remitted
, in_flight
;
687 m
= tipc_group_find_member(grp
, node
, port
);
689 switch (msg_type(hdr
)) {
692 m
= tipc_group_create_member(grp
, node
, port
,
696 m
->bc_syncpt
= msg_grp_bc_syncpt(hdr
);
697 m
->bc_rcv_nxt
= m
->bc_syncpt
;
698 m
->window
+= msg_adv_win(hdr
);
700 /* Wait until PUBLISH event is received */
701 if (m
->state
== MBR_DISCOVERED
) {
702 m
->state
= MBR_JOINING
;
703 } else if (m
->state
== MBR_PUBLISHED
) {
704 m
->state
= MBR_JOINED
;
706 m
->usr_pending
= false;
707 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, xmitq
);
708 ehdr
= buf_msg(m
->event_msg
);
709 msg_set_grp_bc_seqno(ehdr
, m
->bc_syncpt
);
710 __skb_queue_tail(inputq
, m
->event_msg
);
712 list_del_init(&m
->small_win
);
713 tipc_group_update_member(m
, 0);
718 m
->bc_syncpt
= msg_grp_bc_syncpt(hdr
);
719 list_del_init(&m
->list
);
720 list_del_init(&m
->small_win
);
723 /* Wait until WITHDRAW event is received */
724 if (m
->state
!= MBR_LEAVING
) {
725 tipc_group_decr_active(grp
, m
);
726 m
->state
= MBR_LEAVING
;
729 /* Otherwise deliver already received WITHDRAW event */
730 ehdr
= buf_msg(m
->event_msg
);
731 msg_set_grp_bc_seqno(ehdr
, m
->bc_syncpt
);
732 __skb_queue_tail(inputq
, m
->event_msg
);
737 m
->window
+= msg_adv_win(hdr
);
738 *usr_wakeup
= m
->usr_pending
;
739 m
->usr_pending
= false;
740 list_del_init(&m
->small_win
);
745 m
->bc_acked
= msg_grp_bc_acked(hdr
);
746 if (--grp
->bc_ackers
)
749 m
->usr_pending
= false;
751 case GRP_RECLAIM_MSG
:
754 *usr_wakeup
= m
->usr_pending
;
755 m
->usr_pending
= false;
756 tipc_group_proto_xmit(grp
, m
, GRP_REMIT_MSG
, xmitq
);
757 m
->window
= ADV_IDLE
;
760 if (!m
|| m
->state
!= MBR_RECLAIMING
)
763 remitted
= msg_grp_remitted(hdr
);
765 /* Messages preceding the REMIT still in receive queue */
766 if (m
->advertised
> remitted
) {
767 m
->state
= MBR_REMITTED
;
768 in_flight
= m
->advertised
- remitted
;
769 m
->advertised
= ADV_IDLE
+ in_flight
;
772 /* This should never happen */
773 if (m
->advertised
< remitted
)
774 pr_warn_ratelimited("Unexpected REMIT msg\n");
776 /* All messages preceding the REMIT have been read */
777 m
->state
= MBR_JOINED
;
779 m
->advertised
= ADV_IDLE
;
781 /* Set oldest pending member to active and advertise */
782 if (list_empty(&grp
->pending
))
784 pm
= list_first_entry(&grp
->pending
, struct tipc_member
, list
);
785 pm
->state
= MBR_ACTIVE
;
786 list_move_tail(&pm
->list
, &grp
->active
);
788 if (pm
->advertised
<= (ADV_ACTIVE
* 3 / 4))
789 tipc_group_proto_xmit(grp
, pm
, GRP_ADV_MSG
, xmitq
);
792 pr_warn("Received unknown GROUP_PROTO message\n");
796 /* tipc_group_member_evt() - receive and handle a member up/down event
798 void tipc_group_member_evt(struct tipc_group
*grp
,
802 struct sk_buff_head
*inputq
,
803 struct sk_buff_head
*xmitq
)
805 struct tipc_msg
*hdr
= buf_msg(skb
);
806 struct tipc_event
*evt
= (void *)msg_data(hdr
);
807 u32 instance
= evt
->found_lower
;
808 u32 node
= evt
->port
.node
;
809 u32 port
= evt
->port
.ref
;
810 int event
= evt
->event
;
811 struct tipc_member
*m
;
820 self
= tipc_own_addr(net
);
821 if (!grp
->loopback
&& node
== self
&& port
== grp
->portid
)
824 /* Convert message before delivery to user */
825 msg_set_hdr_sz(hdr
, GROUP_H_SIZE
);
826 msg_set_user(hdr
, TIPC_CRITICAL_IMPORTANCE
);
827 msg_set_type(hdr
, TIPC_GRP_MEMBER_EVT
);
828 msg_set_origport(hdr
, port
);
829 msg_set_orignode(hdr
, node
);
830 msg_set_nametype(hdr
, grp
->type
);
831 msg_set_grp_evt(hdr
, event
);
833 m
= tipc_group_find_member(grp
, node
, port
);
835 if (event
== TIPC_PUBLISHED
) {
837 m
= tipc_group_create_member(grp
, node
, port
,
842 /* Hold back event if JOIN message not yet received */
843 if (m
->state
== MBR_DISCOVERED
) {
845 m
->state
= MBR_PUBLISHED
;
847 msg_set_grp_bc_seqno(hdr
, m
->bc_syncpt
);
848 __skb_queue_tail(inputq
, skb
);
849 m
->state
= MBR_JOINED
;
851 m
->usr_pending
= false;
853 m
->instance
= instance
;
854 TIPC_SKB_CB(skb
)->orig_member
= m
->instance
;
855 tipc_group_proto_xmit(grp
, m
, GRP_JOIN_MSG
, xmitq
);
856 tipc_group_update_member(m
, 0);
857 } else if (event
== TIPC_WITHDRAWN
) {
861 TIPC_SKB_CB(skb
)->orig_member
= m
->instance
;
864 m
->usr_pending
= false;
865 node_up
= tipc_node_is_up(net
, node
);
869 /* Hold back event if a LEAVE msg should be expected */
870 if (m
->state
!= MBR_LEAVING
) {
872 tipc_group_decr_active(grp
, m
);
873 m
->state
= MBR_LEAVING
;
875 msg_set_grp_bc_seqno(hdr
, m
->bc_syncpt
);
876 __skb_queue_tail(inputq
, skb
);
879 if (m
->state
!= MBR_LEAVING
) {
880 tipc_group_decr_active(grp
, m
);
881 m
->state
= MBR_LEAVING
;
882 msg_set_grp_bc_seqno(hdr
, m
->bc_rcv_nxt
);
884 msg_set_grp_bc_seqno(hdr
, m
->bc_syncpt
);
886 __skb_queue_tail(inputq
, skb
);
888 list_del_init(&m
->list
);
889 list_del_init(&m
->small_win
);
891 *sk_rcvbuf
= tipc_group_rcvbuf_limit(grp
);