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