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)
63 struct rb_node tree_node
;
64 struct list_head list
;
65 struct list_head small_win
;
66 struct sk_buff_head deferredq
;
67 struct tipc_group
*group
;
80 struct rb_root members
;
81 struct list_head small_win
;
82 struct list_head pending
;
83 struct list_head active
;
84 struct tipc_nlist dests
;
101 static void tipc_group_proto_xmit(struct tipc_group
*grp
, struct tipc_member
*m
,
102 int mtyp
, struct sk_buff_head
*xmitq
);
104 bool tipc_group_is_open(struct tipc_group
*grp
)
109 static void tipc_group_open(struct tipc_member
*m
, bool *wakeup
)
112 if (list_empty(&m
->small_win
))
114 list_del_init(&m
->small_win
);
115 m
->group
->open
= true;
119 static void tipc_group_decr_active(struct tipc_group
*grp
,
120 struct tipc_member
*m
)
122 if (m
->state
== MBR_ACTIVE
|| m
->state
== MBR_RECLAIMING
||
123 m
->state
== MBR_REMITTED
)
127 static int tipc_group_rcvbuf_limit(struct tipc_group
*grp
)
129 int max_active
, active_pool
, idle_pool
;
130 int mcnt
= grp
->member_cnt
+ 1;
132 /* Limit simultaneous reception from other members */
133 max_active
= min(mcnt
/ 8, 64);
134 max_active
= max(max_active
, 16);
135 grp
->max_active
= max_active
;
137 /* Reserve blocks for active and idle members */
138 active_pool
= max_active
* ADV_ACTIVE
;
139 idle_pool
= (mcnt
- max_active
) * ADV_IDLE
;
141 /* Scale to bytes, considering worst-case truesize/msgsize ratio */
142 return (active_pool
+ idle_pool
) * FLOWCTL_BLK_SZ
* 4;
145 u16
tipc_group_bc_snd_nxt(struct tipc_group
*grp
)
147 return grp
->bc_snd_nxt
;
150 static bool tipc_group_is_receiver(struct tipc_member
*m
)
152 return m
&& m
->state
!= MBR_JOINING
&& m
->state
!= MBR_LEAVING
;
155 static bool tipc_group_is_sender(struct tipc_member
*m
)
157 return m
&& m
->state
!= MBR_JOINING
&& m
->state
!= MBR_PUBLISHED
;
160 u32
tipc_group_exclude(struct tipc_group
*grp
)
167 int tipc_group_size(struct tipc_group
*grp
)
169 return grp
->member_cnt
;
172 struct tipc_group
*tipc_group_create(struct net
*net
, u32 portid
,
173 struct tipc_group_req
*mreq
)
175 u32 filter
= TIPC_SUB_PORTS
| TIPC_SUB_NO_STATUS
;
176 bool global
= mreq
->scope
!= TIPC_NODE_SCOPE
;
177 struct tipc_group
*grp
;
178 u32 type
= mreq
->type
;
180 grp
= kzalloc(sizeof(*grp
), GFP_ATOMIC
);
183 tipc_nlist_init(&grp
->dests
, tipc_own_addr(net
));
184 INIT_LIST_HEAD(&grp
->small_win
);
185 INIT_LIST_HEAD(&grp
->active
);
186 INIT_LIST_HEAD(&grp
->pending
);
187 grp
->members
= RB_ROOT
;
189 grp
->portid
= portid
;
191 grp
->instance
= mreq
->instance
;
192 grp
->scope
= mreq
->scope
;
193 grp
->loopback
= mreq
->flags
& TIPC_GROUP_LOOPBACK
;
194 grp
->events
= mreq
->flags
& TIPC_GROUP_MEMBER_EVTS
;
195 filter
|= global
? TIPC_SUB_CLUSTER_SCOPE
: TIPC_SUB_NODE_SCOPE
;
196 if (tipc_topsrv_kern_subscr(net
, portid
, type
, 0, ~0,
197 filter
, &grp
->subid
))
203 void tipc_group_join(struct net
*net
, struct tipc_group
*grp
, int *sk_rcvbuf
)
205 struct rb_root
*tree
= &grp
->members
;
206 struct tipc_member
*m
, *tmp
;
207 struct sk_buff_head xmitq
;
209 skb_queue_head_init(&xmitq
);
210 rbtree_postorder_for_each_entry_safe(m
, tmp
, tree
, tree_node
) {
211 tipc_group_proto_xmit(grp
, m
, GRP_JOIN_MSG
, &xmitq
);
212 tipc_group_update_member(m
, 0);
214 tipc_node_distr_xmit(net
, &xmitq
);
215 *sk_rcvbuf
= tipc_group_rcvbuf_limit(grp
);
218 void tipc_group_delete(struct net
*net
, struct tipc_group
*grp
)
220 struct rb_root
*tree
= &grp
->members
;
221 struct tipc_member
*m
, *tmp
;
222 struct sk_buff_head xmitq
;
224 __skb_queue_head_init(&xmitq
);
226 rbtree_postorder_for_each_entry_safe(m
, tmp
, tree
, tree_node
) {
227 tipc_group_proto_xmit(grp
, m
, GRP_LEAVE_MSG
, &xmitq
);
231 tipc_node_distr_xmit(net
, &xmitq
);
232 tipc_nlist_purge(&grp
->dests
);
233 tipc_topsrv_kern_unsubscr(net
, grp
->subid
);
237 struct tipc_member
*tipc_group_find_member(struct tipc_group
*grp
,
240 struct rb_node
*n
= grp
->members
.rb_node
;
241 u64 nkey
, key
= (u64
)node
<< 32 | port
;
242 struct tipc_member
*m
;
245 m
= container_of(n
, struct tipc_member
, tree_node
);
246 nkey
= (u64
)m
->node
<< 32 | m
->port
;
257 static struct tipc_member
*tipc_group_find_dest(struct tipc_group
*grp
,
260 struct tipc_member
*m
;
262 m
= tipc_group_find_member(grp
, node
, port
);
263 if (m
&& tipc_group_is_receiver(m
))
268 static struct tipc_member
*tipc_group_find_node(struct tipc_group
*grp
,
271 struct tipc_member
*m
;
274 for (n
= rb_first(&grp
->members
); n
; n
= rb_next(n
)) {
275 m
= container_of(n
, struct tipc_member
, tree_node
);
282 static void tipc_group_add_to_tree(struct tipc_group
*grp
,
283 struct tipc_member
*m
)
285 u64 nkey
, key
= (u64
)m
->node
<< 32 | m
->port
;
286 struct rb_node
**n
, *parent
= NULL
;
287 struct tipc_member
*tmp
;
289 n
= &grp
->members
.rb_node
;
291 tmp
= container_of(*n
, struct tipc_member
, tree_node
);
293 tmp
= container_of(parent
, struct tipc_member
, tree_node
);
294 nkey
= (u64
)tmp
->node
<< 32 | tmp
->port
;
302 rb_link_node(&m
->tree_node
, parent
, n
);
303 rb_insert_color(&m
->tree_node
, &grp
->members
);
306 static struct tipc_member
*tipc_group_create_member(struct tipc_group
*grp
,
308 u32 instance
, int state
)
310 struct tipc_member
*m
;
312 m
= kzalloc(sizeof(*m
), GFP_ATOMIC
);
315 INIT_LIST_HEAD(&m
->list
);
316 INIT_LIST_HEAD(&m
->small_win
);
317 __skb_queue_head_init(&m
->deferredq
);
321 m
->instance
= instance
;
322 m
->bc_acked
= grp
->bc_snd_nxt
- 1;
324 tipc_group_add_to_tree(grp
, m
);
325 tipc_nlist_add(&grp
->dests
, m
->node
);
330 void tipc_group_add_member(struct tipc_group
*grp
, u32 node
,
331 u32 port
, u32 instance
)
333 tipc_group_create_member(grp
, node
, port
, instance
, MBR_PUBLISHED
);
336 static void tipc_group_delete_member(struct tipc_group
*grp
,
337 struct tipc_member
*m
)
339 rb_erase(&m
->tree_node
, &grp
->members
);
342 /* Check if we were waiting for replicast ack from this member */
343 if (grp
->bc_ackers
&& less(m
->bc_acked
, grp
->bc_snd_nxt
- 1))
346 list_del_init(&m
->list
);
347 list_del_init(&m
->small_win
);
348 tipc_group_decr_active(grp
, m
);
350 /* If last member on a node, remove node from dest list */
351 if (!tipc_group_find_node(grp
, m
->node
))
352 tipc_nlist_del(&grp
->dests
, m
->node
);
357 struct tipc_nlist
*tipc_group_dests(struct tipc_group
*grp
)
362 void tipc_group_self(struct tipc_group
*grp
, struct tipc_name_seq
*seq
,
365 seq
->type
= grp
->type
;
366 seq
->lower
= grp
->instance
;
367 seq
->upper
= grp
->instance
;
371 void tipc_group_update_member(struct tipc_member
*m
, int len
)
373 struct tipc_group
*grp
= m
->group
;
374 struct tipc_member
*_m
, *tmp
;
376 if (!tipc_group_is_receiver(m
))
381 if (m
->window
>= ADV_IDLE
)
384 list_del_init(&m
->small_win
);
386 /* Sort member into small_window members' list */
387 list_for_each_entry_safe(_m
, tmp
, &grp
->small_win
, small_win
) {
388 if (_m
->window
> m
->window
)
391 list_add_tail(&m
->small_win
, &_m
->small_win
);
394 void tipc_group_update_bc_members(struct tipc_group
*grp
, int len
, bool ack
)
396 u16 prev
= grp
->bc_snd_nxt
- 1;
397 struct tipc_member
*m
;
401 for (n
= rb_first(&grp
->members
); n
; n
= rb_next(n
)) {
402 m
= container_of(n
, struct tipc_member
, tree_node
);
403 if (tipc_group_is_receiver(m
)) {
404 tipc_group_update_member(m
, len
);
410 /* Mark number of acknowledges to expect, if any */
412 grp
->bc_ackers
= ackers
;
416 bool tipc_group_cong(struct tipc_group
*grp
, u32 dnode
, u32 dport
,
417 int len
, struct tipc_member
**mbr
)
419 struct sk_buff_head xmitq
;
420 struct tipc_member
*m
;
423 m
= tipc_group_find_dest(grp
, dnode
, dport
);
424 if (!tipc_group_is_receiver(m
)) {
430 if (m
->window
>= len
)
435 /* If not fully advertised, do it now to prevent mutual blocking */
438 if (state
== MBR_JOINED
&& adv
== ADV_IDLE
)
440 if (state
== MBR_ACTIVE
&& adv
== ADV_ACTIVE
)
442 if (state
== MBR_PENDING
&& adv
== ADV_IDLE
)
444 skb_queue_head_init(&xmitq
);
445 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, &xmitq
);
446 tipc_node_distr_xmit(grp
->net
, &xmitq
);
450 bool tipc_group_bc_cong(struct tipc_group
*grp
, int len
)
452 struct tipc_member
*m
= NULL
;
454 /* If prev bcast was replicast, reject until all receivers have acked */
455 if (grp
->bc_ackers
) {
459 if (list_empty(&grp
->small_win
))
462 m
= list_first_entry(&grp
->small_win
, struct tipc_member
, small_win
);
463 if (m
->window
>= len
)
466 return tipc_group_cong(grp
, m
->node
, m
->port
, len
, &m
);
469 /* tipc_group_sort_msg() - sort msg into queue by bcast sequence number
471 static void tipc_group_sort_msg(struct sk_buff
*skb
, struct sk_buff_head
*defq
)
473 struct tipc_msg
*_hdr
, *hdr
= buf_msg(skb
);
474 u16 bc_seqno
= msg_grp_bc_seqno(hdr
);
475 struct sk_buff
*_skb
, *tmp
;
476 int mtyp
= msg_type(hdr
);
478 /* Bcast/mcast may be bypassed by ucast or other bcast, - sort it in */
479 if (mtyp
== TIPC_GRP_BCAST_MSG
|| mtyp
== TIPC_GRP_MCAST_MSG
) {
480 skb_queue_walk_safe(defq
, _skb
, tmp
) {
481 _hdr
= buf_msg(_skb
);
482 if (!less(bc_seqno
, msg_grp_bc_seqno(_hdr
)))
484 __skb_queue_before(defq
, _skb
, skb
);
487 /* Bcast was not bypassed, - add to tail */
489 /* Unicasts are never bypassed, - always add to tail */
490 __skb_queue_tail(defq
, skb
);
493 /* tipc_group_filter_msg() - determine if we should accept arriving message
495 void tipc_group_filter_msg(struct tipc_group
*grp
, struct sk_buff_head
*inputq
,
496 struct sk_buff_head
*xmitq
)
498 struct sk_buff
*skb
= __skb_dequeue(inputq
);
499 bool ack
, deliver
, update
, leave
= false;
500 struct sk_buff_head
*defq
;
501 struct tipc_member
*m
;
502 struct tipc_msg
*hdr
;
510 node
= msg_orignode(hdr
);
511 port
= msg_origport(hdr
);
513 if (!msg_in_group(hdr
))
516 m
= tipc_group_find_member(grp
, node
, port
);
517 if (!tipc_group_is_sender(m
))
520 if (less(msg_grp_bc_seqno(hdr
), m
->bc_rcv_nxt
))
523 TIPC_SKB_CB(skb
)->orig_member
= m
->instance
;
524 defq
= &m
->deferredq
;
525 tipc_group_sort_msg(skb
, defq
);
527 while ((skb
= skb_peek(defq
))) {
529 mtyp
= msg_type(hdr
);
530 blks
= msg_blocks(hdr
);
535 if (more(msg_grp_bc_seqno(hdr
), m
->bc_rcv_nxt
))
538 /* Decide what to do with message */
540 case TIPC_GRP_MCAST_MSG
:
541 if (msg_nameinst(hdr
) != grp
->instance
) {
546 case TIPC_GRP_BCAST_MSG
:
548 ack
= msg_grp_bc_ack_req(hdr
);
550 case TIPC_GRP_UCAST_MSG
:
552 case TIPC_GRP_MEMBER_EVT
:
553 if (m
->state
== MBR_LEAVING
)
562 /* Execute decisions */
565 __skb_queue_tail(inputq
, skb
);
570 tipc_group_proto_xmit(grp
, m
, GRP_ACK_MSG
, xmitq
);
573 __skb_queue_purge(defq
);
574 tipc_group_delete_member(grp
, m
);
580 tipc_group_update_rcv_win(grp
, blks
, node
, port
, xmitq
);
587 void tipc_group_update_rcv_win(struct tipc_group
*grp
, int blks
, u32 node
,
588 u32 port
, struct sk_buff_head
*xmitq
)
590 struct list_head
*active
= &grp
->active
;
591 int max_active
= grp
->max_active
;
592 int reclaim_limit
= max_active
* 3 / 4;
593 int active_cnt
= grp
->active_cnt
;
594 struct tipc_member
*m
, *rm
, *pm
;
596 m
= tipc_group_find_member(grp
, node
, port
);
600 m
->advertised
-= blks
;
604 /* First, decide if member can go active */
605 if (active_cnt
<= max_active
) {
606 m
->state
= MBR_ACTIVE
;
607 list_add_tail(&m
->list
, active
);
609 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, xmitq
);
611 m
->state
= MBR_PENDING
;
612 list_add_tail(&m
->list
, &grp
->pending
);
615 if (active_cnt
< reclaim_limit
)
618 /* Reclaim from oldest active member, if possible */
619 if (!list_empty(active
)) {
620 rm
= list_first_entry(active
, struct tipc_member
, list
);
621 rm
->state
= MBR_RECLAIMING
;
622 list_del_init(&rm
->list
);
623 tipc_group_proto_xmit(grp
, rm
, GRP_RECLAIM_MSG
, xmitq
);
626 /* Nobody to reclaim from; - revert oldest pending to JOINED */
627 pm
= list_first_entry(&grp
->pending
, struct tipc_member
, list
);
628 list_del_init(&pm
->list
);
629 pm
->state
= MBR_JOINED
;
630 tipc_group_proto_xmit(grp
, pm
, GRP_ADV_MSG
, xmitq
);
633 if (!list_is_last(&m
->list
, &grp
->active
))
634 list_move_tail(&m
->list
, &grp
->active
);
635 if (m
->advertised
> (ADV_ACTIVE
* 3 / 4))
637 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, xmitq
);
640 if (m
->advertised
> ADV_IDLE
)
642 m
->state
= MBR_JOINED
;
644 if (m
->advertised
< ADV_IDLE
) {
645 pr_warn_ratelimited("Rcv unexpected msg after REMIT\n");
646 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, xmitq
);
649 if (list_empty(&grp
->pending
))
652 /* Set oldest pending member to active and advertise */
653 pm
= list_first_entry(&grp
->pending
, struct tipc_member
, list
);
654 pm
->state
= MBR_ACTIVE
;
655 list_move_tail(&pm
->list
, &grp
->active
);
657 tipc_group_proto_xmit(grp
, pm
, GRP_ADV_MSG
, xmitq
);
667 static void tipc_group_create_event(struct tipc_group
*grp
,
668 struct tipc_member
*m
,
669 u32 event
, u16 seqno
,
670 struct sk_buff_head
*inputq
)
671 { u32 dnode
= tipc_own_addr(grp
->net
);
672 struct tipc_event evt
;
674 struct tipc_msg
*hdr
;
677 evt
.found_lower
= m
->instance
;
678 evt
.found_upper
= m
->instance
;
679 evt
.port
.ref
= m
->port
;
680 evt
.port
.node
= m
->node
;
681 evt
.s
.seq
.type
= grp
->type
;
682 evt
.s
.seq
.lower
= m
->instance
;
683 evt
.s
.seq
.upper
= m
->instance
;
685 skb
= tipc_msg_create(TIPC_CRITICAL_IMPORTANCE
, TIPC_GRP_MEMBER_EVT
,
686 GROUP_H_SIZE
, sizeof(evt
), dnode
, m
->node
,
687 grp
->portid
, m
->port
, 0);
692 msg_set_nametype(hdr
, grp
->type
);
693 msg_set_grp_evt(hdr
, event
);
694 msg_set_dest_droppable(hdr
, true);
695 msg_set_grp_bc_seqno(hdr
, seqno
);
696 memcpy(msg_data(hdr
), &evt
, sizeof(evt
));
697 TIPC_SKB_CB(skb
)->orig_member
= m
->instance
;
698 __skb_queue_tail(inputq
, skb
);
701 static void tipc_group_proto_xmit(struct tipc_group
*grp
, struct tipc_member
*m
,
702 int mtyp
, struct sk_buff_head
*xmitq
)
704 struct tipc_msg
*hdr
;
708 skb
= tipc_msg_create(GROUP_PROTOCOL
, mtyp
, INT_H_SIZE
, 0,
709 m
->node
, tipc_own_addr(grp
->net
),
710 m
->port
, grp
->portid
, 0);
714 if (m
->state
== MBR_ACTIVE
)
715 adv
= ADV_ACTIVE
- m
->advertised
;
716 else if (m
->state
== MBR_JOINED
|| m
->state
== MBR_PENDING
)
717 adv
= ADV_IDLE
- m
->advertised
;
721 if (mtyp
== GRP_JOIN_MSG
) {
722 msg_set_grp_bc_syncpt(hdr
, grp
->bc_snd_nxt
);
723 msg_set_adv_win(hdr
, adv
);
724 m
->advertised
+= adv
;
725 } else if (mtyp
== GRP_LEAVE_MSG
) {
726 msg_set_grp_bc_syncpt(hdr
, grp
->bc_snd_nxt
);
727 } else if (mtyp
== GRP_ADV_MSG
) {
728 msg_set_adv_win(hdr
, adv
);
729 m
->advertised
+= adv
;
730 } else if (mtyp
== GRP_ACK_MSG
) {
731 msg_set_grp_bc_acked(hdr
, m
->bc_rcv_nxt
);
732 } else if (mtyp
== GRP_REMIT_MSG
) {
733 msg_set_grp_remitted(hdr
, m
->window
);
735 msg_set_dest_droppable(hdr
, true);
736 __skb_queue_tail(xmitq
, skb
);
739 void tipc_group_proto_rcv(struct tipc_group
*grp
, bool *usr_wakeup
,
740 struct tipc_msg
*hdr
, struct sk_buff_head
*inputq
,
741 struct sk_buff_head
*xmitq
)
743 u32 node
= msg_orignode(hdr
);
744 u32 port
= msg_origport(hdr
);
745 struct tipc_member
*m
, *pm
;
746 u16 remitted
, in_flight
;
751 if (grp
->scope
== TIPC_NODE_SCOPE
&& node
!= tipc_own_addr(grp
->net
))
754 m
= tipc_group_find_member(grp
, node
, port
);
756 switch (msg_type(hdr
)) {
759 m
= tipc_group_create_member(grp
, node
, port
,
763 m
->bc_syncpt
= msg_grp_bc_syncpt(hdr
);
764 m
->bc_rcv_nxt
= m
->bc_syncpt
;
765 m
->window
+= msg_adv_win(hdr
);
767 /* Wait until PUBLISH event is received if necessary */
768 if (m
->state
!= MBR_PUBLISHED
)
771 /* Member can be taken into service */
772 m
->state
= MBR_JOINED
;
773 tipc_group_open(m
, usr_wakeup
);
774 tipc_group_update_member(m
, 0);
775 tipc_group_proto_xmit(grp
, m
, GRP_ADV_MSG
, xmitq
);
776 tipc_group_create_event(grp
, m
, TIPC_PUBLISHED
,
777 m
->bc_syncpt
, inputq
);
782 m
->bc_syncpt
= msg_grp_bc_syncpt(hdr
);
783 list_del_init(&m
->list
);
784 tipc_group_open(m
, usr_wakeup
);
785 tipc_group_decr_active(grp
, m
);
786 m
->state
= MBR_LEAVING
;
787 tipc_group_create_event(grp
, m
, TIPC_WITHDRAWN
,
788 m
->bc_syncpt
, inputq
);
793 m
->window
+= msg_adv_win(hdr
);
794 tipc_group_open(m
, usr_wakeup
);
799 m
->bc_acked
= msg_grp_bc_acked(hdr
);
800 if (--grp
->bc_ackers
)
802 list_del_init(&m
->small_win
);
803 m
->group
->open
= true;
805 tipc_group_update_member(m
, 0);
807 case GRP_RECLAIM_MSG
:
810 tipc_group_proto_xmit(grp
, m
, GRP_REMIT_MSG
, xmitq
);
811 m
->window
= ADV_IDLE
;
812 tipc_group_open(m
, usr_wakeup
);
815 if (!m
|| m
->state
!= MBR_RECLAIMING
)
818 remitted
= msg_grp_remitted(hdr
);
820 /* Messages preceding the REMIT still in receive queue */
821 if (m
->advertised
> remitted
) {
822 m
->state
= MBR_REMITTED
;
823 in_flight
= m
->advertised
- remitted
;
824 m
->advertised
= ADV_IDLE
+ in_flight
;
827 /* This should never happen */
828 if (m
->advertised
< remitted
)
829 pr_warn_ratelimited("Unexpected REMIT msg\n");
831 /* All messages preceding the REMIT have been read */
832 m
->state
= MBR_JOINED
;
834 m
->advertised
= ADV_IDLE
;
836 /* Set oldest pending member to active and advertise */
837 if (list_empty(&grp
->pending
))
839 pm
= list_first_entry(&grp
->pending
, struct tipc_member
, list
);
840 pm
->state
= MBR_ACTIVE
;
841 list_move_tail(&pm
->list
, &grp
->active
);
843 if (pm
->advertised
<= (ADV_ACTIVE
* 3 / 4))
844 tipc_group_proto_xmit(grp
, pm
, GRP_ADV_MSG
, xmitq
);
847 pr_warn("Received unknown GROUP_PROTO message\n");
851 /* tipc_group_member_evt() - receive and handle a member up/down event
853 void tipc_group_member_evt(struct tipc_group
*grp
,
856 struct tipc_msg
*hdr
,
857 struct sk_buff_head
*inputq
,
858 struct sk_buff_head
*xmitq
)
860 struct tipc_event
*evt
= (void *)msg_data(hdr
);
861 u32 instance
= evt
->found_lower
;
862 u32 node
= evt
->port
.node
;
863 u32 port
= evt
->port
.ref
;
864 int event
= evt
->event
;
865 struct tipc_member
*m
;
873 self
= tipc_own_addr(net
);
874 if (!grp
->loopback
&& node
== self
&& port
== grp
->portid
)
877 m
= tipc_group_find_member(grp
, node
, port
);
881 /* Send and wait for arrival of JOIN message if necessary */
883 m
= tipc_group_create_member(grp
, node
, port
, instance
,
887 tipc_group_update_member(m
, 0);
888 tipc_group_proto_xmit(grp
, m
, GRP_JOIN_MSG
, xmitq
);
892 if (m
->state
!= MBR_JOINING
)
895 /* Member can be taken into service */
896 m
->instance
= instance
;
897 m
->state
= MBR_JOINED
;
898 tipc_group_open(m
, usr_wakeup
);
899 tipc_group_update_member(m
, 0);
900 tipc_group_proto_xmit(grp
, m
, GRP_JOIN_MSG
, xmitq
);
901 tipc_group_create_event(grp
, m
, TIPC_PUBLISHED
,
902 m
->bc_syncpt
, inputq
);
908 tipc_group_decr_active(grp
, m
);
909 m
->state
= MBR_LEAVING
;
910 list_del_init(&m
->list
);
911 tipc_group_open(m
, usr_wakeup
);
913 /* Only send event if no LEAVE message can be expected */
914 if (!tipc_node_is_up(net
, node
))
915 tipc_group_create_event(grp
, m
, TIPC_WITHDRAWN
,
916 m
->bc_rcv_nxt
, inputq
);
921 *sk_rcvbuf
= tipc_group_rcvbuf_limit(grp
);