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