]> Git Repo - linux.git/blob - net/tipc/group.c
Merge branch 'x86/urgent' of git://git.kernel.org/pub/scm/linux/kernel/git/tip/tip
[linux.git] / net / tipc / group.c
1 /*
2  * net/tipc/group.c: TIPC group messaging code
3  *
4  * Copyright (c) 2017, Ericsson AB
5  * All rights reserved.
6  *
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted provided that the following conditions are met:
9  *
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.
18  *
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.
22  *
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.
34  */
35
36 #include "core.h"
37 #include "addr.h"
38 #include "group.h"
39 #include "bcast.h"
40 #include "server.h"
41 #include "msg.h"
42 #include "socket.h"
43 #include "node.h"
44 #include "name_table.h"
45 #include "subscr.h"
46
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)
50
51 enum mbr_state {
52         MBR_QUARANTINED,
53         MBR_DISCOVERED,
54         MBR_JOINING,
55         MBR_PUBLISHED,
56         MBR_JOINED,
57         MBR_PENDING,
58         MBR_ACTIVE,
59         MBR_RECLAIMING,
60         MBR_REMITTED,
61         MBR_LEAVING
62 };
63
64 struct tipc_member {
65         struct rb_node tree_node;
66         struct list_head list;
67         struct list_head congested;
68         struct sk_buff *event_msg;
69         struct sk_buff_head deferredq;
70         struct tipc_group *group;
71         u32 node;
72         u32 port;
73         u32 instance;
74         enum mbr_state state;
75         u16 advertised;
76         u16 window;
77         u16 bc_rcv_nxt;
78         u16 bc_syncpt;
79         u16 bc_acked;
80         bool usr_pending;
81 };
82
83 struct tipc_group {
84         struct rb_root members;
85         struct list_head congested;
86         struct list_head pending;
87         struct list_head active;
88         struct list_head reclaiming;
89         struct tipc_nlist dests;
90         struct net *net;
91         int subid;
92         u32 type;
93         u32 instance;
94         u32 domain;
95         u32 scope;
96         u32 portid;
97         u16 member_cnt;
98         u16 active_cnt;
99         u16 max_active;
100         u16 bc_snd_nxt;
101         u16 bc_ackers;
102         bool loopback;
103         bool events;
104 };
105
106 static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,
107                                   int mtyp, struct sk_buff_head *xmitq);
108
109 static void tipc_group_decr_active(struct tipc_group *grp,
110                                    struct tipc_member *m)
111 {
112         if (m->state == MBR_ACTIVE || m->state == MBR_RECLAIMING)
113                 grp->active_cnt--;
114 }
115
116 static int tipc_group_rcvbuf_limit(struct tipc_group *grp)
117 {
118         int max_active, active_pool, idle_pool;
119         int mcnt = grp->member_cnt + 1;
120
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;
125
126         /* Reserve blocks for active and idle members */
127         active_pool = max_active * ADV_ACTIVE;
128         idle_pool = (mcnt - max_active) * ADV_IDLE;
129
130         /* Scale to bytes, considering worst-case truesize/msgsize ratio */
131         return (active_pool + idle_pool) * FLOWCTL_BLK_SZ * 4;
132 }
133
134 u16 tipc_group_bc_snd_nxt(struct tipc_group *grp)
135 {
136         return grp->bc_snd_nxt;
137 }
138
139 static bool tipc_group_is_enabled(struct tipc_member *m)
140 {
141         return m->state != MBR_QUARANTINED && m->state != MBR_LEAVING;
142 }
143
144 static bool tipc_group_is_receiver(struct tipc_member *m)
145 {
146         return m && m->state >= MBR_JOINED;
147 }
148
149 u32 tipc_group_exclude(struct tipc_group *grp)
150 {
151         if (!grp->loopback)
152                 return grp->portid;
153         return 0;
154 }
155
156 int tipc_group_size(struct tipc_group *grp)
157 {
158         return grp->member_cnt;
159 }
160
161 struct tipc_group *tipc_group_create(struct net *net, u32 portid,
162                                      struct tipc_group_req *mreq)
163 {
164         struct tipc_group *grp;
165         u32 type = mreq->type;
166
167         grp = kzalloc(sizeof(*grp), GFP_ATOMIC);
168         if (!grp)
169                 return NULL;
170         tipc_nlist_init(&grp->dests, tipc_own_addr(net));
171         INIT_LIST_HEAD(&grp->congested);
172         INIT_LIST_HEAD(&grp->active);
173         INIT_LIST_HEAD(&grp->pending);
174         INIT_LIST_HEAD(&grp->reclaiming);
175         grp->members = RB_ROOT;
176         grp->net = net;
177         grp->portid = portid;
178         grp->domain = addr_domain(net, mreq->scope);
179         grp->type = type;
180         grp->instance = mreq->instance;
181         grp->scope = mreq->scope;
182         grp->loopback = mreq->flags & TIPC_GROUP_LOOPBACK;
183         grp->events = mreq->flags & TIPC_GROUP_MEMBER_EVTS;
184         if (tipc_topsrv_kern_subscr(net, portid, type, 0, ~0, &grp->subid))
185                 return grp;
186         kfree(grp);
187         return NULL;
188 }
189
190 void tipc_group_delete(struct net *net, struct tipc_group *grp)
191 {
192         struct rb_root *tree = &grp->members;
193         struct tipc_member *m, *tmp;
194         struct sk_buff_head xmitq;
195
196         __skb_queue_head_init(&xmitq);
197
198         rbtree_postorder_for_each_entry_safe(m, tmp, tree, tree_node) {
199                 tipc_group_proto_xmit(grp, m, GRP_LEAVE_MSG, &xmitq);
200                 list_del(&m->list);
201                 kfree(m);
202         }
203         tipc_node_distr_xmit(net, &xmitq);
204         tipc_nlist_purge(&grp->dests);
205         tipc_topsrv_kern_unsubscr(net, grp->subid);
206         kfree(grp);
207 }
208
209 struct tipc_member *tipc_group_find_member(struct tipc_group *grp,
210                                            u32 node, u32 port)
211 {
212         struct rb_node *n = grp->members.rb_node;
213         u64 nkey, key = (u64)node << 32 | port;
214         struct tipc_member *m;
215
216         while (n) {
217                 m = container_of(n, struct tipc_member, tree_node);
218                 nkey = (u64)m->node << 32 | m->port;
219                 if (key < nkey)
220                         n = n->rb_left;
221                 else if (key > nkey)
222                         n = n->rb_right;
223                 else
224                         return m;
225         }
226         return NULL;
227 }
228
229 static struct tipc_member *tipc_group_find_dest(struct tipc_group *grp,
230                                                 u32 node, u32 port)
231 {
232         struct tipc_member *m;
233
234         m = tipc_group_find_member(grp, node, port);
235         if (m && tipc_group_is_enabled(m))
236                 return m;
237         return NULL;
238 }
239
240 static struct tipc_member *tipc_group_find_node(struct tipc_group *grp,
241                                                 u32 node)
242 {
243         struct tipc_member *m;
244         struct rb_node *n;
245
246         for (n = rb_first(&grp->members); n; n = rb_next(n)) {
247                 m = container_of(n, struct tipc_member, tree_node);
248                 if (m->node == node)
249                         return m;
250         }
251         return NULL;
252 }
253
254 static void tipc_group_add_to_tree(struct tipc_group *grp,
255                                    struct tipc_member *m)
256 {
257         u64 nkey, key = (u64)m->node << 32 | m->port;
258         struct rb_node **n, *parent = NULL;
259         struct tipc_member *tmp;
260
261         n = &grp->members.rb_node;
262         while (*n) {
263                 tmp = container_of(*n, struct tipc_member, tree_node);
264                 parent = *n;
265                 tmp = container_of(parent, struct tipc_member, tree_node);
266                 nkey = (u64)tmp->node << 32 | tmp->port;
267                 if (key < nkey)
268                         n = &(*n)->rb_left;
269                 else if (key > nkey)
270                         n = &(*n)->rb_right;
271                 else
272                         return;
273         }
274         rb_link_node(&m->tree_node, parent, n);
275         rb_insert_color(&m->tree_node, &grp->members);
276 }
277
278 static struct tipc_member *tipc_group_create_member(struct tipc_group *grp,
279                                                     u32 node, u32 port,
280                                                     int state)
281 {
282         struct tipc_member *m;
283
284         m = kzalloc(sizeof(*m), GFP_ATOMIC);
285         if (!m)
286                 return NULL;
287         INIT_LIST_HEAD(&m->list);
288         INIT_LIST_HEAD(&m->congested);
289         __skb_queue_head_init(&m->deferredq);
290         m->group = grp;
291         m->node = node;
292         m->port = port;
293         m->bc_acked = grp->bc_snd_nxt - 1;
294         grp->member_cnt++;
295         tipc_group_add_to_tree(grp, m);
296         tipc_nlist_add(&grp->dests, m->node);
297         m->state = state;
298         return m;
299 }
300
301 void tipc_group_add_member(struct tipc_group *grp, u32 node, u32 port)
302 {
303         tipc_group_create_member(grp, node, port, MBR_DISCOVERED);
304 }
305
306 static void tipc_group_delete_member(struct tipc_group *grp,
307                                      struct tipc_member *m)
308 {
309         rb_erase(&m->tree_node, &grp->members);
310         grp->member_cnt--;
311
312         /* Check if we were waiting for replicast ack from this member */
313         if (grp->bc_ackers && less(m->bc_acked, grp->bc_snd_nxt - 1))
314                 grp->bc_ackers--;
315
316         list_del_init(&m->list);
317         list_del_init(&m->congested);
318         tipc_group_decr_active(grp, m);
319
320         /* If last member on a node, remove node from dest list */
321         if (!tipc_group_find_node(grp, m->node))
322                 tipc_nlist_del(&grp->dests, m->node);
323
324         kfree(m);
325 }
326
327 struct tipc_nlist *tipc_group_dests(struct tipc_group *grp)
328 {
329         return &grp->dests;
330 }
331
332 void tipc_group_self(struct tipc_group *grp, struct tipc_name_seq *seq,
333                      int *scope)
334 {
335         seq->type = grp->type;
336         seq->lower = grp->instance;
337         seq->upper = grp->instance;
338         *scope = grp->scope;
339 }
340
341 void tipc_group_update_member(struct tipc_member *m, int len)
342 {
343         struct tipc_group *grp = m->group;
344         struct tipc_member *_m, *tmp;
345
346         if (!tipc_group_is_enabled(m))
347                 return;
348
349         m->window -= len;
350
351         if (m->window >= ADV_IDLE)
352                 return;
353
354         list_del_init(&m->congested);
355
356         /* Sort member into congested members' list */
357         list_for_each_entry_safe(_m, tmp, &grp->congested, congested) {
358                 if (m->window > _m->window)
359                         continue;
360                 list_add_tail(&m->congested, &_m->congested);
361                 return;
362         }
363         list_add_tail(&m->congested, &grp->congested);
364 }
365
366 void tipc_group_update_bc_members(struct tipc_group *grp, int len, bool ack)
367 {
368         u16 prev = grp->bc_snd_nxt - 1;
369         struct tipc_member *m;
370         struct rb_node *n;
371         u16 ackers = 0;
372
373         for (n = rb_first(&grp->members); n; n = rb_next(n)) {
374                 m = container_of(n, struct tipc_member, tree_node);
375                 if (tipc_group_is_enabled(m)) {
376                         tipc_group_update_member(m, len);
377                         m->bc_acked = prev;
378                         ackers++;
379                 }
380         }
381
382         /* Mark number of acknowledges to expect, if any */
383         if (ack)
384                 grp->bc_ackers = ackers;
385         grp->bc_snd_nxt++;
386 }
387
388 bool tipc_group_cong(struct tipc_group *grp, u32 dnode, u32 dport,
389                      int len, struct tipc_member **mbr)
390 {
391         struct sk_buff_head xmitq;
392         struct tipc_member *m;
393         int adv, state;
394
395         m = tipc_group_find_dest(grp, dnode, dport);
396         *mbr = m;
397         if (!m)
398                 return false;
399         if (m->usr_pending)
400                 return true;
401         if (m->window >= len)
402                 return false;
403         m->usr_pending = true;
404
405         /* If not fully advertised, do it now to prevent mutual blocking */
406         adv = m->advertised;
407         state = m->state;
408         if (state < MBR_JOINED)
409                 return true;
410         if (state == MBR_JOINED && adv == ADV_IDLE)
411                 return true;
412         if (state == MBR_ACTIVE && adv == ADV_ACTIVE)
413                 return true;
414         if (state == MBR_PENDING && adv == ADV_IDLE)
415                 return true;
416         skb_queue_head_init(&xmitq);
417         tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, &xmitq);
418         tipc_node_distr_xmit(grp->net, &xmitq);
419         return true;
420 }
421
422 bool tipc_group_bc_cong(struct tipc_group *grp, int len)
423 {
424         struct tipc_member *m = NULL;
425
426         /* If prev bcast was replicast, reject until all receivers have acked */
427         if (grp->bc_ackers)
428                 return true;
429
430         if (list_empty(&grp->congested))
431                 return false;
432
433         m = list_first_entry(&grp->congested, struct tipc_member, congested);
434         if (m->window >= len)
435                 return false;
436
437         return tipc_group_cong(grp, m->node, m->port, len, &m);
438 }
439
440 /* tipc_group_sort_msg() - sort msg into queue by bcast sequence number
441  */
442 static void tipc_group_sort_msg(struct sk_buff *skb, struct sk_buff_head *defq)
443 {
444         struct tipc_msg *_hdr, *hdr = buf_msg(skb);
445         u16 bc_seqno = msg_grp_bc_seqno(hdr);
446         struct sk_buff *_skb, *tmp;
447         int mtyp = msg_type(hdr);
448
449         /* Bcast/mcast may be bypassed by ucast or other bcast, - sort it in */
450         if (mtyp == TIPC_GRP_BCAST_MSG || mtyp == TIPC_GRP_MCAST_MSG) {
451                 skb_queue_walk_safe(defq, _skb, tmp) {
452                         _hdr = buf_msg(_skb);
453                         if (!less(bc_seqno, msg_grp_bc_seqno(_hdr)))
454                                 continue;
455                         __skb_queue_before(defq, _skb, skb);
456                         return;
457                 }
458                 /* Bcast was not bypassed, - add to tail */
459         }
460         /* Unicasts are never bypassed, - always add to tail */
461         __skb_queue_tail(defq, skb);
462 }
463
464 /* tipc_group_filter_msg() - determine if we should accept arriving message
465  */
466 void tipc_group_filter_msg(struct tipc_group *grp, struct sk_buff_head *inputq,
467                            struct sk_buff_head *xmitq)
468 {
469         struct sk_buff *skb = __skb_dequeue(inputq);
470         bool ack, deliver, update, leave = false;
471         struct sk_buff_head *defq;
472         struct tipc_member *m;
473         struct tipc_msg *hdr;
474         u32 node, port;
475         int mtyp, blks;
476
477         if (!skb)
478                 return;
479
480         hdr = buf_msg(skb);
481         node =  msg_orignode(hdr);
482         port = msg_origport(hdr);
483
484         if (!msg_in_group(hdr))
485                 goto drop;
486
487         m = tipc_group_find_member(grp, node, port);
488         if (!tipc_group_is_receiver(m))
489                 goto drop;
490
491         if (less(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt))
492                 goto drop;
493
494         TIPC_SKB_CB(skb)->orig_member = m->instance;
495         defq = &m->deferredq;
496         tipc_group_sort_msg(skb, defq);
497
498         while ((skb = skb_peek(defq))) {
499                 hdr = buf_msg(skb);
500                 mtyp = msg_type(hdr);
501                 blks = msg_blocks(hdr);
502                 deliver = true;
503                 ack = false;
504                 update = false;
505
506                 if (more(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt))
507                         break;
508
509                 /* Decide what to do with message */
510                 switch (mtyp) {
511                 case TIPC_GRP_MCAST_MSG:
512                         if (msg_nameinst(hdr) != grp->instance) {
513                                 update = true;
514                                 deliver = false;
515                         }
516                         /* Fall thru */
517                 case TIPC_GRP_BCAST_MSG:
518                         m->bc_rcv_nxt++;
519                         ack = msg_grp_bc_ack_req(hdr);
520                         break;
521                 case TIPC_GRP_UCAST_MSG:
522                         break;
523                 case TIPC_GRP_MEMBER_EVT:
524                         if (m->state == MBR_LEAVING)
525                                 leave = true;
526                         if (!grp->events)
527                                 deliver = false;
528                         break;
529                 default:
530                         break;
531                 }
532
533                 /* Execute decisions */
534                 __skb_dequeue(defq);
535                 if (deliver)
536                         __skb_queue_tail(inputq, skb);
537                 else
538                         kfree_skb(skb);
539
540                 if (ack)
541                         tipc_group_proto_xmit(grp, m, GRP_ACK_MSG, xmitq);
542
543                 if (leave) {
544                         __skb_queue_purge(defq);
545                         tipc_group_delete_member(grp, m);
546                         break;
547                 }
548                 if (!update)
549                         continue;
550
551                 tipc_group_update_rcv_win(grp, blks, node, port, xmitq);
552         }
553         return;
554 drop:
555         kfree_skb(skb);
556 }
557
558 void tipc_group_update_rcv_win(struct tipc_group *grp, int blks, u32 node,
559                                u32 port, struct sk_buff_head *xmitq)
560 {
561         struct list_head *active = &grp->active;
562         int max_active = grp->max_active;
563         int reclaim_limit = max_active * 3 / 4;
564         int active_cnt = grp->active_cnt;
565         struct tipc_member *m, *rm;
566
567         m = tipc_group_find_member(grp, node, port);
568         if (!m)
569                 return;
570
571         m->advertised -= blks;
572
573         switch (m->state) {
574         case MBR_JOINED:
575                 /* Reclaim advertised space from least active member */
576                 if (!list_empty(active) && active_cnt >= reclaim_limit) {
577                         rm = list_first_entry(active, struct tipc_member, list);
578                         rm->state = MBR_RECLAIMING;
579                         list_move_tail(&rm->list, &grp->reclaiming);
580                         tipc_group_proto_xmit(grp, rm, GRP_RECLAIM_MSG, xmitq);
581                 }
582                 /* If max active, become pending and wait for reclaimed space */
583                 if (active_cnt >= max_active) {
584                         m->state = MBR_PENDING;
585                         list_add_tail(&m->list, &grp->pending);
586                         break;
587                 }
588                 /* Otherwise become active */
589                 m->state = MBR_ACTIVE;
590                 list_add_tail(&m->list, &grp->active);
591                 grp->active_cnt++;
592                 /* Fall through */
593         case MBR_ACTIVE:
594                 if (!list_is_last(&m->list, &grp->active))
595                         list_move_tail(&m->list, &grp->active);
596                 if (m->advertised > (ADV_ACTIVE * 3 / 4))
597                         break;
598                 tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
599                 break;
600         case MBR_REMITTED:
601                 if (m->advertised > ADV_IDLE)
602                         break;
603                 m->state = MBR_JOINED;
604                 if (m->advertised < ADV_IDLE) {
605                         pr_warn_ratelimited("Rcv unexpected msg after REMIT\n");
606                         tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
607                 }
608                 break;
609         case MBR_RECLAIMING:
610         case MBR_DISCOVERED:
611         case MBR_JOINING:
612         case MBR_LEAVING:
613         default:
614                 break;
615         }
616 }
617
618 static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,
619                                   int mtyp, struct sk_buff_head *xmitq)
620 {
621         struct tipc_msg *hdr;
622         struct sk_buff *skb;
623         int adv = 0;
624
625         skb = tipc_msg_create(GROUP_PROTOCOL, mtyp, INT_H_SIZE, 0,
626                               m->node, tipc_own_addr(grp->net),
627                               m->port, grp->portid, 0);
628         if (!skb)
629                 return;
630
631         if (m->state == MBR_ACTIVE)
632                 adv = ADV_ACTIVE - m->advertised;
633         else if (m->state == MBR_JOINED || m->state == MBR_PENDING)
634                 adv = ADV_IDLE - m->advertised;
635
636         hdr = buf_msg(skb);
637
638         if (mtyp == GRP_JOIN_MSG) {
639                 msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
640                 msg_set_adv_win(hdr, adv);
641                 m->advertised += adv;
642         } else if (mtyp == GRP_LEAVE_MSG) {
643                 msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
644         } else if (mtyp == GRP_ADV_MSG) {
645                 msg_set_adv_win(hdr, adv);
646                 m->advertised += adv;
647         } else if (mtyp == GRP_ACK_MSG) {
648                 msg_set_grp_bc_acked(hdr, m->bc_rcv_nxt);
649         } else if (mtyp == GRP_REMIT_MSG) {
650                 msg_set_grp_remitted(hdr, m->window);
651         }
652         msg_set_dest_droppable(hdr, true);
653         __skb_queue_tail(xmitq, skb);
654 }
655
656 void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup,
657                           struct tipc_msg *hdr, struct sk_buff_head *inputq,
658                           struct sk_buff_head *xmitq)
659 {
660         u32 node = msg_orignode(hdr);
661         u32 port = msg_origport(hdr);
662         struct tipc_member *m, *pm;
663         struct tipc_msg *ehdr;
664         u16 remitted, in_flight;
665
666         if (!grp)
667                 return;
668
669         m = tipc_group_find_member(grp, node, port);
670
671         switch (msg_type(hdr)) {
672         case GRP_JOIN_MSG:
673                 if (!m)
674                         m = tipc_group_create_member(grp, node, port,
675                                                      MBR_QUARANTINED);
676                 if (!m)
677                         return;
678                 m->bc_syncpt = msg_grp_bc_syncpt(hdr);
679                 m->bc_rcv_nxt = m->bc_syncpt;
680                 m->window += msg_adv_win(hdr);
681
682                 /* Wait until PUBLISH event is received */
683                 if (m->state == MBR_DISCOVERED) {
684                         m->state = MBR_JOINING;
685                 } else if (m->state == MBR_PUBLISHED) {
686                         m->state = MBR_JOINED;
687                         *usr_wakeup = true;
688                         m->usr_pending = false;
689                         tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
690                         ehdr = buf_msg(m->event_msg);
691                         msg_set_grp_bc_seqno(ehdr, m->bc_syncpt);
692                         __skb_queue_tail(inputq, m->event_msg);
693                 }
694                 list_del_init(&m->congested);
695                 tipc_group_update_member(m, 0);
696                 return;
697         case GRP_LEAVE_MSG:
698                 if (!m)
699                         return;
700                 m->bc_syncpt = msg_grp_bc_syncpt(hdr);
701                 list_del_init(&m->list);
702                 list_del_init(&m->congested);
703                 *usr_wakeup = true;
704
705                 /* Wait until WITHDRAW event is received */
706                 if (m->state != MBR_LEAVING) {
707                         tipc_group_decr_active(grp, m);
708                         m->state = MBR_LEAVING;
709                         return;
710                 }
711                 /* Otherwise deliver already received WITHDRAW event */
712                 ehdr = buf_msg(m->event_msg);
713                 msg_set_grp_bc_seqno(ehdr, m->bc_syncpt);
714                 __skb_queue_tail(inputq, m->event_msg);
715                 return;
716         case GRP_ADV_MSG:
717                 if (!m)
718                         return;
719                 m->window += msg_adv_win(hdr);
720                 *usr_wakeup = m->usr_pending;
721                 m->usr_pending = false;
722                 list_del_init(&m->congested);
723                 return;
724         case GRP_ACK_MSG:
725                 if (!m)
726                         return;
727                 m->bc_acked = msg_grp_bc_acked(hdr);
728                 if (--grp->bc_ackers)
729                         break;
730                 *usr_wakeup = true;
731                 m->usr_pending = false;
732                 return;
733         case GRP_RECLAIM_MSG:
734                 if (!m)
735                         return;
736                 *usr_wakeup = m->usr_pending;
737                 m->usr_pending = false;
738                 tipc_group_proto_xmit(grp, m, GRP_REMIT_MSG, xmitq);
739                 m->window = ADV_IDLE;
740                 return;
741         case GRP_REMIT_MSG:
742                 if (!m || m->state != MBR_RECLAIMING)
743                         return;
744
745                 list_del_init(&m->list);
746                 grp->active_cnt--;
747                 remitted = msg_grp_remitted(hdr);
748
749                 /* Messages preceding the REMIT still in receive queue */
750                 if (m->advertised > remitted) {
751                         m->state = MBR_REMITTED;
752                         in_flight = m->advertised - remitted;
753                 }
754                 /* All messages preceding the REMIT have been read */
755                 if (m->advertised <= remitted) {
756                         m->state = MBR_JOINED;
757                         in_flight = 0;
758                 }
759                 /* ..and the REMIT overtaken by more messages => re-advertise */
760                 if (m->advertised < remitted)
761                         tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
762
763                 m->advertised = ADV_IDLE + in_flight;
764
765                 /* Set oldest pending member to active and advertise */
766                 if (list_empty(&grp->pending))
767                         return;
768                 pm = list_first_entry(&grp->pending, struct tipc_member, list);
769                 pm->state = MBR_ACTIVE;
770                 list_move_tail(&pm->list, &grp->active);
771                 grp->active_cnt++;
772                 if (pm->advertised <= (ADV_ACTIVE * 3 / 4))
773                         tipc_group_proto_xmit(grp, pm, GRP_ADV_MSG, xmitq);
774                 return;
775         default:
776                 pr_warn("Received unknown GROUP_PROTO message\n");
777         }
778 }
779
780 /* tipc_group_member_evt() - receive and handle a member up/down event
781  */
782 void tipc_group_member_evt(struct tipc_group *grp,
783                            bool *usr_wakeup,
784                            int *sk_rcvbuf,
785                            struct sk_buff *skb,
786                            struct sk_buff_head *inputq,
787                            struct sk_buff_head *xmitq)
788 {
789         struct tipc_msg *hdr = buf_msg(skb);
790         struct tipc_event *evt = (void *)msg_data(hdr);
791         u32 instance = evt->found_lower;
792         u32 node = evt->port.node;
793         u32 port = evt->port.ref;
794         int event = evt->event;
795         struct tipc_member *m;
796         struct net *net;
797         bool node_up;
798         u32 self;
799
800         if (!grp)
801                 goto drop;
802
803         net = grp->net;
804         self = tipc_own_addr(net);
805         if (!grp->loopback && node == self && port == grp->portid)
806                 goto drop;
807
808         /* Convert message before delivery to user */
809         msg_set_hdr_sz(hdr, GROUP_H_SIZE);
810         msg_set_user(hdr, TIPC_CRITICAL_IMPORTANCE);
811         msg_set_type(hdr, TIPC_GRP_MEMBER_EVT);
812         msg_set_origport(hdr, port);
813         msg_set_orignode(hdr, node);
814         msg_set_nametype(hdr, grp->type);
815         msg_set_grp_evt(hdr, event);
816
817         m = tipc_group_find_member(grp, node, port);
818
819         if (event == TIPC_PUBLISHED) {
820                 if (!m)
821                         m = tipc_group_create_member(grp, node, port,
822                                                      MBR_DISCOVERED);
823                 if (!m)
824                         goto drop;
825
826                 /* Hold back event if JOIN message not yet received */
827                 if (m->state == MBR_DISCOVERED) {
828                         m->event_msg = skb;
829                         m->state = MBR_PUBLISHED;
830                 } else {
831                         msg_set_grp_bc_seqno(hdr, m->bc_syncpt);
832                         __skb_queue_tail(inputq, skb);
833                         m->state = MBR_JOINED;
834                         *usr_wakeup = true;
835                         m->usr_pending = false;
836                 }
837                 m->instance = instance;
838                 TIPC_SKB_CB(skb)->orig_member = m->instance;
839                 tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
840                 if (m->window < ADV_IDLE)
841                         tipc_group_update_member(m, 0);
842                 else
843                         list_del_init(&m->congested);
844         } else if (event == TIPC_WITHDRAWN) {
845                 if (!m)
846                         goto drop;
847
848                 TIPC_SKB_CB(skb)->orig_member = m->instance;
849
850                 *usr_wakeup = true;
851                 m->usr_pending = false;
852                 node_up = tipc_node_is_up(net, node);
853                 m->event_msg = NULL;
854
855                 if (node_up) {
856                         /* Hold back event if a LEAVE msg should be expected */
857                         if (m->state != MBR_LEAVING) {
858                                 m->event_msg = skb;
859                                 tipc_group_decr_active(grp, m);
860                                 m->state = MBR_LEAVING;
861                         } else {
862                                 msg_set_grp_bc_seqno(hdr, m->bc_syncpt);
863                                 __skb_queue_tail(inputq, skb);
864                         }
865                 } else {
866                         if (m->state != MBR_LEAVING) {
867                                 tipc_group_decr_active(grp, m);
868                                 m->state = MBR_LEAVING;
869                                 msg_set_grp_bc_seqno(hdr, m->bc_rcv_nxt);
870                         } else {
871                                 msg_set_grp_bc_seqno(hdr, m->bc_syncpt);
872                         }
873                         __skb_queue_tail(inputq, skb);
874                 }
875                 list_del_init(&m->list);
876                 list_del_init(&m->congested);
877         }
878         *sk_rcvbuf = tipc_group_rcvbuf_limit(grp);
879         return;
880 drop:
881         kfree_skb(skb);
882 }
This page took 0.088019 seconds and 4 git commands to generate.