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 if (!list_empty(&m->congested)) 355 return; 356 357 /* Sort member into congested members' list */ 358 list_for_each_entry_safe(_m, tmp, &grp->congested, congested) { 359 if (m->window > _m->window) 360 continue; 361 list_add_tail(&m->congested, &_m->congested); 362 return; 363 } 364 list_add_tail(&m->congested, &grp->congested); 365 } 366 367 void tipc_group_update_bc_members(struct tipc_group *grp, int len, bool ack) 368 { 369 u16 prev = grp->bc_snd_nxt - 1; 370 struct tipc_member *m; 371 struct rb_node *n; 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 } 379 } 380 381 /* Mark number of acknowledges to expect, if any */ 382 if (ack) 383 grp->bc_ackers = grp->member_cnt; 384 grp->bc_snd_nxt++; 385 } 386 387 bool tipc_group_cong(struct tipc_group *grp, u32 dnode, u32 dport, 388 int len, struct tipc_member **mbr) 389 { 390 struct sk_buff_head xmitq; 391 struct tipc_member *m; 392 int adv, state; 393 394 m = tipc_group_find_dest(grp, dnode, dport); 395 *mbr = m; 396 if (!m) 397 return false; 398 if (m->usr_pending) 399 return true; 400 if (m->window >= len) 401 return false; 402 m->usr_pending = true; 403 404 /* If not fully advertised, do it now to prevent mutual blocking */ 405 adv = m->advertised; 406 state = m->state; 407 if (state < MBR_JOINED) 408 return true; 409 if (state == MBR_JOINED && adv == ADV_IDLE) 410 return true; 411 if (state == MBR_ACTIVE && adv == ADV_ACTIVE) 412 return true; 413 if (state == MBR_PENDING && adv == ADV_IDLE) 414 return true; 415 skb_queue_head_init(&xmitq); 416 tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, &xmitq); 417 tipc_node_distr_xmit(grp->net, &xmitq); 418 return true; 419 } 420 421 bool tipc_group_bc_cong(struct tipc_group *grp, int len) 422 { 423 struct tipc_member *m = NULL; 424 425 /* If prev bcast was replicast, reject until all receivers have acked */ 426 if (grp->bc_ackers) 427 return true; 428 429 if (list_empty(&grp->congested)) 430 return false; 431 432 m = list_first_entry(&grp->congested, struct tipc_member, congested); 433 if (m->window >= len) 434 return false; 435 436 return tipc_group_cong(grp, m->node, m->port, len, &m); 437 } 438 439 /* tipc_group_sort_msg() - sort msg into queue by bcast sequence number 440 */ 441 static void tipc_group_sort_msg(struct sk_buff *skb, struct sk_buff_head *defq) 442 { 443 struct tipc_msg *_hdr, *hdr = buf_msg(skb); 444 u16 bc_seqno = msg_grp_bc_seqno(hdr); 445 struct sk_buff *_skb, *tmp; 446 int mtyp = msg_type(hdr); 447 448 /* Bcast/mcast may be bypassed by ucast or other bcast, - sort it in */ 449 if (mtyp == TIPC_GRP_BCAST_MSG || mtyp == TIPC_GRP_MCAST_MSG) { 450 skb_queue_walk_safe(defq, _skb, tmp) { 451 _hdr = buf_msg(_skb); 452 if (!less(bc_seqno, msg_grp_bc_seqno(_hdr))) 453 continue; 454 __skb_queue_before(defq, _skb, skb); 455 return; 456 } 457 /* Bcast was not bypassed, - add to tail */ 458 } 459 /* Unicasts are never bypassed, - always add to tail */ 460 __skb_queue_tail(defq, skb); 461 } 462 463 /* tipc_group_filter_msg() - determine if we should accept arriving message 464 */ 465 void tipc_group_filter_msg(struct tipc_group *grp, struct sk_buff_head *inputq, 466 struct sk_buff_head *xmitq) 467 { 468 struct sk_buff *skb = __skb_dequeue(inputq); 469 bool ack, deliver, update, leave = false; 470 struct sk_buff_head *defq; 471 struct tipc_member *m; 472 struct tipc_msg *hdr; 473 u32 node, port; 474 int mtyp, blks; 475 476 if (!skb) 477 return; 478 479 hdr = buf_msg(skb); 480 node = msg_orignode(hdr); 481 port = msg_origport(hdr); 482 483 if (!msg_in_group(hdr)) 484 goto drop; 485 486 m = tipc_group_find_member(grp, node, port); 487 if (!tipc_group_is_receiver(m)) 488 goto drop; 489 490 if (less(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt)) 491 goto drop; 492 493 TIPC_SKB_CB(skb)->orig_member = m->instance; 494 defq = &m->deferredq; 495 tipc_group_sort_msg(skb, defq); 496 497 while ((skb = skb_peek(defq))) { 498 hdr = buf_msg(skb); 499 mtyp = msg_type(hdr); 500 blks = msg_blocks(hdr); 501 deliver = true; 502 ack = false; 503 update = false; 504 505 if (more(msg_grp_bc_seqno(hdr), m->bc_rcv_nxt)) 506 break; 507 508 /* Decide what to do with message */ 509 switch (mtyp) { 510 case TIPC_GRP_MCAST_MSG: 511 if (msg_nameinst(hdr) != grp->instance) { 512 update = true; 513 deliver = false; 514 } 515 /* Fall thru */ 516 case TIPC_GRP_BCAST_MSG: 517 m->bc_rcv_nxt++; 518 ack = msg_grp_bc_ack_req(hdr); 519 break; 520 case TIPC_GRP_UCAST_MSG: 521 break; 522 case TIPC_GRP_MEMBER_EVT: 523 if (m->state == MBR_LEAVING) 524 leave = true; 525 if (!grp->events) 526 deliver = false; 527 break; 528 default: 529 break; 530 } 531 532 /* Execute decisions */ 533 __skb_dequeue(defq); 534 if (deliver) 535 __skb_queue_tail(inputq, skb); 536 else 537 kfree_skb(skb); 538 539 if (ack) 540 tipc_group_proto_xmit(grp, m, GRP_ACK_MSG, xmitq); 541 542 if (leave) { 543 __skb_queue_purge(defq); 544 tipc_group_delete_member(grp, m); 545 break; 546 } 547 if (!update) 548 continue; 549 550 tipc_group_update_rcv_win(grp, blks, node, port, xmitq); 551 } 552 return; 553 drop: 554 kfree_skb(skb); 555 } 556 557 void tipc_group_update_rcv_win(struct tipc_group *grp, int blks, u32 node, 558 u32 port, struct sk_buff_head *xmitq) 559 { 560 struct list_head *active = &grp->active; 561 int max_active = grp->max_active; 562 int reclaim_limit = max_active * 3 / 4; 563 int active_cnt = grp->active_cnt; 564 struct tipc_member *m, *rm; 565 566 m = tipc_group_find_member(grp, node, port); 567 if (!m) 568 return; 569 570 m->advertised -= blks; 571 572 switch (m->state) { 573 case MBR_JOINED: 574 /* Reclaim advertised space from least active member */ 575 if (!list_empty(active) && active_cnt >= reclaim_limit) { 576 rm = list_first_entry(active, struct tipc_member, list); 577 rm->state = MBR_RECLAIMING; 578 list_move_tail(&rm->list, &grp->reclaiming); 579 tipc_group_proto_xmit(grp, rm, GRP_RECLAIM_MSG, xmitq); 580 } 581 /* If max active, become pending and wait for reclaimed space */ 582 if (active_cnt >= max_active) { 583 m->state = MBR_PENDING; 584 list_add_tail(&m->list, &grp->pending); 585 break; 586 } 587 /* Otherwise become active */ 588 m->state = MBR_ACTIVE; 589 list_add_tail(&m->list, &grp->active); 590 grp->active_cnt++; 591 /* Fall through */ 592 case MBR_ACTIVE: 593 if (!list_is_last(&m->list, &grp->active)) 594 list_move_tail(&m->list, &grp->active); 595 if (m->advertised > (ADV_ACTIVE * 3 / 4)) 596 break; 597 tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq); 598 break; 599 case MBR_REMITTED: 600 if (m->advertised > ADV_IDLE) 601 break; 602 m->state = MBR_JOINED; 603 if (m->advertised < ADV_IDLE) { 604 pr_warn_ratelimited("Rcv unexpected msg after REMIT\n"); 605 tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq); 606 } 607 break; 608 case MBR_RECLAIMING: 609 case MBR_DISCOVERED: 610 case MBR_JOINING: 611 case MBR_LEAVING: 612 default: 613 break; 614 } 615 } 616 617 static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m, 618 int mtyp, struct sk_buff_head *xmitq) 619 { 620 struct tipc_msg *hdr; 621 struct sk_buff *skb; 622 int adv = 0; 623 624 skb = tipc_msg_create(GROUP_PROTOCOL, mtyp, INT_H_SIZE, 0, 625 m->node, tipc_own_addr(grp->net), 626 m->port, grp->portid, 0); 627 if (!skb) 628 return; 629 630 if (m->state == MBR_ACTIVE) 631 adv = ADV_ACTIVE - m->advertised; 632 else if (m->state == MBR_JOINED || m->state == MBR_PENDING) 633 adv = ADV_IDLE - m->advertised; 634 635 hdr = buf_msg(skb); 636 637 if (mtyp == GRP_JOIN_MSG) { 638 msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt); 639 msg_set_adv_win(hdr, adv); 640 m->advertised += adv; 641 } else if (mtyp == GRP_LEAVE_MSG) { 642 msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt); 643 } else if (mtyp == GRP_ADV_MSG) { 644 msg_set_adv_win(hdr, adv); 645 m->advertised += adv; 646 } else if (mtyp == GRP_ACK_MSG) { 647 msg_set_grp_bc_acked(hdr, m->bc_rcv_nxt); 648 } else if (mtyp == GRP_REMIT_MSG) { 649 msg_set_grp_remitted(hdr, m->window); 650 } 651 __skb_queue_tail(xmitq, skb); 652 } 653 654 void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup, 655 struct tipc_msg *hdr, struct sk_buff_head *inputq, 656 struct sk_buff_head *xmitq) 657 { 658 u32 node = msg_orignode(hdr); 659 u32 port = msg_origport(hdr); 660 struct tipc_member *m, *pm; 661 struct tipc_msg *ehdr; 662 u16 remitted, in_flight; 663 664 if (!grp) 665 return; 666 667 m = tipc_group_find_member(grp, node, port); 668 669 switch (msg_type(hdr)) { 670 case GRP_JOIN_MSG: 671 if (!m) 672 m = tipc_group_create_member(grp, node, port, 673 MBR_QUARANTINED); 674 if (!m) 675 return; 676 m->bc_syncpt = msg_grp_bc_syncpt(hdr); 677 m->bc_rcv_nxt = m->bc_syncpt; 678 m->window += msg_adv_win(hdr); 679 680 /* Wait until PUBLISH event is received */ 681 if (m->state == MBR_DISCOVERED) { 682 m->state = MBR_JOINING; 683 } else if (m->state == MBR_PUBLISHED) { 684 m->state = MBR_JOINED; 685 *usr_wakeup = true; 686 m->usr_pending = false; 687 tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq); 688 ehdr = buf_msg(m->event_msg); 689 msg_set_grp_bc_seqno(ehdr, m->bc_syncpt); 690 __skb_queue_tail(inputq, m->event_msg); 691 } 692 if (m->window < ADV_IDLE) 693 tipc_group_update_member(m, 0); 694 else 695 list_del_init(&m->congested); 696 return; 697 case GRP_LEAVE_MSG: 698 if (!m) 699 return; 700 m->bc_syncpt = msg_grp_bc_syncpt(hdr); 701 702 /* Wait until WITHDRAW event is received */ 703 if (m->state != MBR_LEAVING) { 704 tipc_group_decr_active(grp, m); 705 m->state = MBR_LEAVING; 706 return; 707 } 708 /* Otherwise deliver already received WITHDRAW event */ 709 ehdr = buf_msg(m->event_msg); 710 msg_set_grp_bc_seqno(ehdr, m->bc_syncpt); 711 __skb_queue_tail(inputq, m->event_msg); 712 *usr_wakeup = true; 713 list_del_init(&m->congested); 714 return; 715 case GRP_ADV_MSG: 716 if (!m) 717 return; 718 m->window += msg_adv_win(hdr); 719 *usr_wakeup = m->usr_pending; 720 m->usr_pending = false; 721 list_del_init(&m->congested); 722 return; 723 case GRP_ACK_MSG: 724 if (!m) 725 return; 726 m->bc_acked = msg_grp_bc_acked(hdr); 727 if (--grp->bc_ackers) 728 break; 729 *usr_wakeup = true; 730 m->usr_pending = false; 731 return; 732 case GRP_RECLAIM_MSG: 733 if (!m) 734 return; 735 *usr_wakeup = m->usr_pending; 736 m->usr_pending = false; 737 tipc_group_proto_xmit(grp, m, GRP_REMIT_MSG, xmitq); 738 m->window = ADV_IDLE; 739 return; 740 case GRP_REMIT_MSG: 741 if (!m || m->state != MBR_RECLAIMING) 742 return; 743 744 list_del_init(&m->list); 745 grp->active_cnt--; 746 remitted = msg_grp_remitted(hdr); 747 748 /* Messages preceding the REMIT still in receive queue */ 749 if (m->advertised > remitted) { 750 m->state = MBR_REMITTED; 751 in_flight = m->advertised - remitted; 752 } 753 /* All messages preceding the REMIT have been read */ 754 if (m->advertised <= remitted) { 755 m->state = MBR_JOINED; 756 in_flight = 0; 757 } 758 /* ..and the REMIT overtaken by more messages => re-advertise */ 759 if (m->advertised < remitted) 760 tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq); 761 762 m->advertised = ADV_IDLE + in_flight; 763 764 /* Set oldest pending member to active and advertise */ 765 if (list_empty(&grp->pending)) 766 return; 767 pm = list_first_entry(&grp->pending, struct tipc_member, list); 768 pm->state = MBR_ACTIVE; 769 list_move_tail(&pm->list, &grp->active); 770 grp->active_cnt++; 771 if (pm->advertised <= (ADV_ACTIVE * 3 / 4)) 772 tipc_group_proto_xmit(grp, pm, GRP_ADV_MSG, xmitq); 773 return; 774 default: 775 pr_warn("Received unknown GROUP_PROTO message\n"); 776 } 777 } 778 779 /* tipc_group_member_evt() - receive and handle a member up/down event 780 */ 781 void tipc_group_member_evt(struct tipc_group *grp, 782 bool *usr_wakeup, 783 int *sk_rcvbuf, 784 struct sk_buff *skb, 785 struct sk_buff_head *inputq, 786 struct sk_buff_head *xmitq) 787 { 788 struct tipc_msg *hdr = buf_msg(skb); 789 struct tipc_event *evt = (void *)msg_data(hdr); 790 u32 instance = evt->found_lower; 791 u32 node = evt->port.node; 792 u32 port = evt->port.ref; 793 int event = evt->event; 794 struct tipc_member *m; 795 struct net *net; 796 bool node_up; 797 u32 self; 798 799 if (!grp) 800 goto drop; 801 802 net = grp->net; 803 self = tipc_own_addr(net); 804 if (!grp->loopback && node == self && port == grp->portid) 805 goto drop; 806 807 /* Convert message before delivery to user */ 808 msg_set_hdr_sz(hdr, GROUP_H_SIZE); 809 msg_set_user(hdr, TIPC_CRITICAL_IMPORTANCE); 810 msg_set_type(hdr, TIPC_GRP_MEMBER_EVT); 811 msg_set_origport(hdr, port); 812 msg_set_orignode(hdr, node); 813 msg_set_nametype(hdr, grp->type); 814 msg_set_grp_evt(hdr, event); 815 816 m = tipc_group_find_member(grp, node, port); 817 818 if (event == TIPC_PUBLISHED) { 819 if (!m) 820 m = tipc_group_create_member(grp, node, port, 821 MBR_DISCOVERED); 822 if (!m) 823 goto drop; 824 825 /* Hold back event if JOIN message not yet received */ 826 if (m->state == MBR_DISCOVERED) { 827 m->event_msg = skb; 828 m->state = MBR_PUBLISHED; 829 } else { 830 msg_set_grp_bc_seqno(hdr, m->bc_syncpt); 831 __skb_queue_tail(inputq, skb); 832 m->state = MBR_JOINED; 833 *usr_wakeup = true; 834 m->usr_pending = false; 835 } 836 m->instance = instance; 837 TIPC_SKB_CB(skb)->orig_member = m->instance; 838 tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq); 839 if (m->window < ADV_IDLE) 840 tipc_group_update_member(m, 0); 841 else 842 list_del_init(&m->congested); 843 } else if (event == TIPC_WITHDRAWN) { 844 if (!m) 845 goto drop; 846 847 TIPC_SKB_CB(skb)->orig_member = m->instance; 848 849 *usr_wakeup = true; 850 m->usr_pending = false; 851 node_up = tipc_node_is_up(net, node); 852 853 /* Hold back event if more messages might be expected */ 854 if (m->state != MBR_LEAVING && node_up) { 855 m->event_msg = skb; 856 tipc_group_decr_active(grp, m); 857 m->state = MBR_LEAVING; 858 } else { 859 if (node_up) 860 msg_set_grp_bc_seqno(hdr, m->bc_syncpt); 861 else 862 msg_set_grp_bc_seqno(hdr, m->bc_rcv_nxt); 863 __skb_queue_tail(inputq, skb); 864 } 865 list_del_init(&m->congested); 866 } 867 *sk_rcvbuf = tipc_group_rcvbuf_limit(grp); 868 return; 869 drop: 870 kfree_skb(skb); 871 } 872