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_LEAVING
  58 };
  59
  60 struct tipc_member {
  61         struct rb_node tree_node;
  62         struct list_head list;
  63         struct list_head congested;
  64         struct sk_buff *event_msg;
  65         struct tipc_group *group;
  66         u32 node;
  67         u32 port;
  68         u32 instance;
  69         enum mbr_state state;
  70         u16 advertised;
  71         u16 window;
  72         u16 bc_rcv_nxt;
  73         bool usr_pending;
  74 };
  75
  76 struct tipc_group {
  77         struct rb_root members;
  78         struct list_head congested;
  79         struct tipc_nlist dests;
  80         struct net *net;
  81         int subid;
  82         u32 type;
  83         u32 instance;
  84         u32 domain;
  85         u32 scope;
  86         u32 portid;
  87         u16 member_cnt;
  88         u16 bc_snd_nxt;
  89         bool loopback;
  90         bool events;
  91 };
  92
  93 static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,
  94                                   int mtyp, struct sk_buff_head *xmitq);
  95
  96 static int tipc_group_rcvbuf_limit(struct tipc_group *grp)
  97 {
  98         int mcnt = grp->member_cnt + 1;
  99
 100         /* Scale to bytes, considering worst-case truesize/msgsize ratio */
 101         return mcnt * ADV_ACTIVE * FLOWCTL_BLK_SZ * 4;
 102 }
 103
 104 u16 tipc_group_bc_snd_nxt(struct tipc_group *grp)
 105 {
 106         return grp->bc_snd_nxt;
 107 }
 108
 109 static bool tipc_group_is_enabled(struct tipc_member *m)
 110 {
 111         return m->state != MBR_QUARANTINED && m->state != MBR_LEAVING;
 112 }
 113
 114 static bool tipc_group_is_receiver(struct tipc_member *m)
 115 {
 116         return m && m->state >= MBR_JOINED;
 117 }
 118
 119 u32 tipc_group_exclude(struct tipc_group *grp)
 120 {
 121         if (!grp->loopback)
 122                 return grp->portid;
 123         return 0;
 124 }
 125
 126 int tipc_group_size(struct tipc_group *grp)
 127 {
 128         return grp->member_cnt;
 129 }
 130
 131 struct tipc_group *tipc_group_create(struct net *net, u32 portid,
 132                                      struct tipc_group_req *mreq)
 133 {
 134         struct tipc_group *grp;
 135         u32 type = mreq->type;
 136
 137         grp = kzalloc(sizeof(*grp), GFP_ATOMIC);
 138         if (!grp)
 139                 return NULL;
 140         tipc_nlist_init(&grp->dests, tipc_own_addr(net));
 141         INIT_LIST_HEAD(&grp->congested);
 142         grp->members = RB_ROOT;
 143         grp->net = net;
 144         grp->portid = portid;
 145         grp->domain = addr_domain(net, mreq->scope);
 146         grp->type = type;
 147         grp->instance = mreq->instance;
 148         grp->scope = mreq->scope;
 149         grp->loopback = mreq->flags & TIPC_GROUP_LOOPBACK;
 150         grp->events = mreq->flags & TIPC_GROUP_MEMBER_EVTS;
 151         if (tipc_topsrv_kern_subscr(net, portid, type, 0, ~0, &grp->subid))
 152                 return grp;
 153         kfree(grp);
 154         return NULL;
 155 }
 156
 157 void tipc_group_delete(struct net *net, struct tipc_group *grp)
 158 {
 159         struct rb_root *tree = &grp->members;
 160         struct tipc_member *m, *tmp;
 161         struct sk_buff_head xmitq;
 162
 163         __skb_queue_head_init(&xmitq);
 164
 165         rbtree_postorder_for_each_entry_safe(m, tmp, tree, tree_node) {
 166                 tipc_group_proto_xmit(grp, m, GRP_LEAVE_MSG, &xmitq);
 167                 list_del(&m->list);
 168                 kfree(m);
 169         }
 170         tipc_node_distr_xmit(net, &xmitq);
 171         tipc_nlist_purge(&grp->dests);
 172         tipc_topsrv_kern_unsubscr(net, grp->subid);
 173         kfree(grp);
 174 }
 175
 176 struct tipc_member *tipc_group_find_member(struct tipc_group *grp,
 177                                            u32 node, u32 port)
 178 {
 179         struct rb_node *n = grp->members.rb_node;
 180         u64 nkey, key = (u64)node << 32 | port;
 181         struct tipc_member *m;
 182
 183         while (n) {
 184                 m = container_of(n, struct tipc_member, tree_node);
 185                 nkey = (u64)m->node << 32 | m->port;
 186                 if (key < nkey)
 187                         n = n->rb_left;
 188                 else if (key > nkey)
 189                         n = n->rb_right;
 190                 else
 191                         return m;
 192         }
 193         return NULL;
 194 }
 195
 196 static struct tipc_member *tipc_group_find_dest(struct tipc_group *grp,
 197                                                 u32 node, u32 port)
 198 {
 199         struct tipc_member *m;
 200
 201         m = tipc_group_find_member(grp, node, port);
 202         if (m && tipc_group_is_enabled(m))
 203                 return m;
 204         return NULL;
 205 }
 206
 207 static struct tipc_member *tipc_group_find_node(struct tipc_group *grp,
 208                                                 u32 node)
 209 {
 210         struct tipc_member *m;
 211         struct rb_node *n;
 212
 213         for (n = rb_first(&grp->members); n; n = rb_next(n)) {
 214                 m = container_of(n, struct tipc_member, tree_node);
 215                 if (m->node == node)
 216                         return m;
 217         }
 218         return NULL;
 219 }
 220
 221 static void tipc_group_add_to_tree(struct tipc_group *grp,
 222                                    struct tipc_member *m)
 223 {
 224         u64 nkey, key = (u64)m->node << 32 | m->port;
 225         struct rb_node **n, *parent = NULL;
 226         struct tipc_member *tmp;
 227
 228         n = &grp->members.rb_node;
 229         while (*n) {
 230                 tmp = container_of(*n, struct tipc_member, tree_node);
 231                 parent = *n;
 232                 tmp = container_of(parent, struct tipc_member, tree_node);
 233                 nkey = (u64)tmp->node << 32 | tmp->port;
 234                 if (key < nkey)
 235                         n = &(*n)->rb_left;
 236                 else if (key > nkey)
 237                         n = &(*n)->rb_right;
 238                 else
 239                         return;
 240         }
 241         rb_link_node(&m->tree_node, parent, n);
 242         rb_insert_color(&m->tree_node, &grp->members);
 243 }
 244
 245 static struct tipc_member *tipc_group_create_member(struct tipc_group *grp,
 246                                                     u32 node, u32 port,
 247                                                     int state)
 248 {
 249         struct tipc_member *m;
 250
 251         m = kzalloc(sizeof(*m), GFP_ATOMIC);
 252         if (!m)
 253                 return NULL;
 254         INIT_LIST_HEAD(&m->list);
 255         INIT_LIST_HEAD(&m->congested);
 256         m->group = grp;
 257         m->node = node;
 258         m->port = port;
 259         grp->member_cnt++;
 260         tipc_group_add_to_tree(grp, m);
 261         tipc_nlist_add(&grp->dests, m->node);
 262         m->state = state;
 263         return m;
 264 }
 265
 266 void tipc_group_add_member(struct tipc_group *grp, u32 node, u32 port)
 267 {
 268         tipc_group_create_member(grp, node, port, MBR_DISCOVERED);
 269 }
 270
 271 static void tipc_group_delete_member(struct tipc_group *grp,
 272                                      struct tipc_member *m)
 273 {
 274         rb_erase(&m->tree_node, &grp->members);
 275         grp->member_cnt--;
 276         list_del_init(&m->list);
 277         list_del_init(&m->congested);
 278
 279         /* If last member on a node, remove node from dest list */
 280         if (!tipc_group_find_node(grp, m->node))
 281                 tipc_nlist_del(&grp->dests, m->node);
 282
 283         kfree(m);
 284 }
 285
 286 struct tipc_nlist *tipc_group_dests(struct tipc_group *grp)
 287 {
 288         return &grp->dests;
 289 }
 290
 291 void tipc_group_self(struct tipc_group *grp, struct tipc_name_seq *seq,
 292                      int *scope)
 293 {
 294         seq->type = grp->type;
 295         seq->lower = grp->instance;
 296         seq->upper = grp->instance;
 297         *scope = grp->scope;
 298 }
 299
 300 void tipc_group_update_member(struct tipc_member *m, int len)
 301 {
 302         struct tipc_group *grp = m->group;
 303         struct tipc_member *_m, *tmp;
 304
 305         if (!tipc_group_is_enabled(m))
 306                 return;
 307
 308         m->window -= len;
 309
 310         if (m->window >= ADV_IDLE)
 311                 return;
 312
 313         if (!list_empty(&m->congested))
 314                 return;
 315
 316         /* Sort member into congested members' list */
 317         list_for_each_entry_safe(_m, tmp, &grp->congested, congested) {
 318                 if (m->window > _m->window)
 319                         continue;
 320                 list_add_tail(&m->congested, &_m->congested);
 321                 return;
 322         }
 323         list_add_tail(&m->congested, &grp->congested);
 324 }
 325
 326 void tipc_group_update_bc_members(struct tipc_group *grp, int len)
 327 {
 328         struct tipc_member *m;
 329         struct rb_node *n;
 330
 331         for (n = rb_first(&grp->members); n; n = rb_next(n)) {
 332                 m = container_of(n, struct tipc_member, tree_node);
 333                 if (tipc_group_is_enabled(m))
 334                         tipc_group_update_member(m, len);
 335         }
 336         grp->bc_snd_nxt++;
 337 }
 338
 339 bool tipc_group_cong(struct tipc_group *grp, u32 dnode, u32 dport,
 340                      int len, struct tipc_member **mbr)
 341 {
 342         struct sk_buff_head xmitq;
 343         struct tipc_member *m;
 344         int adv, state;
 345
 346         m = tipc_group_find_dest(grp, dnode, dport);
 347         *mbr = m;
 348         if (!m)
 349                 return false;
 350         if (m->usr_pending)
 351                 return true;
 352         if (m->window >= len)
 353                 return false;
 354         m->usr_pending = true;
 355
 356         /* If not fully advertised, do it now to prevent mutual blocking */
 357         adv = m->advertised;
 358         state = m->state;
 359         if (state < MBR_JOINED)
 360                 return true;
 361         if (state == MBR_JOINED && adv == ADV_IDLE)
 362                 return true;
 363         skb_queue_head_init(&xmitq);
 364         tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, &xmitq);
 365         tipc_node_distr_xmit(grp->net, &xmitq);
 366         return true;
 367 }
 368
 369 bool tipc_group_bc_cong(struct tipc_group *grp, int len)
 370 {
 371         struct tipc_member *m = NULL;
 372
 373         if (list_empty(&grp->congested))
 374                 return false;
 375
 376         m = list_first_entry(&grp->congested, struct tipc_member, congested);
 377         if (m->window >= len)
 378                 return false;
 379
 380         return tipc_group_cong(grp, m->node, m->port, len, &m);
 381 }
 382
 383 /* tipc_group_filter_msg() - determine if we should accept arriving message
 384  */
 385 void tipc_group_filter_msg(struct tipc_group *grp, struct sk_buff_head *inputq,
 386                            struct sk_buff_head *xmitq)
 387 {
 388         struct sk_buff *skb = __skb_dequeue(inputq);
 389         struct tipc_member *m;
 390         struct tipc_msg *hdr;
 391         u32 node, port;
 392         int mtyp;
 393
 394         if (!skb)
 395                 return;
 396
 397         hdr = buf_msg(skb);
 398         mtyp = msg_type(hdr);
 399         node =  msg_orignode(hdr);
 400         port = msg_origport(hdr);
 401
 402         if (!msg_in_group(hdr))
 403                 goto drop;
 404
 405         if (mtyp == TIPC_GRP_MEMBER_EVT) {
 406                 if (!grp->events)
 407                         goto drop;
 408                 __skb_queue_tail(inputq, skb);
 409                 return;
 410         }
 411
 412         m = tipc_group_find_member(grp, node, port);
 413         if (!tipc_group_is_receiver(m))
 414                 goto drop;
 415
 416         TIPC_SKB_CB(skb)->orig_member = m->instance;
 417         __skb_queue_tail(inputq, skb);
 418
 419         m->bc_rcv_nxt = msg_grp_bc_seqno(hdr) + 1;
 420         return;
 421 drop:
 422         kfree_skb(skb);
 423 }
 424
 425 void tipc_group_update_rcv_win(struct tipc_group *grp, int blks, u32 node,
 426                                u32 port, struct sk_buff_head *xmitq)
 427 {
 428         struct tipc_member *m;
 429
 430         m = tipc_group_find_member(grp, node, port);
 431         if (!m)
 432                 return;
 433
 434         m->advertised -= blks;
 435
 436         switch (m->state) {
 437         case MBR_JOINED:
 438                 if (m->advertised <= (ADV_ACTIVE - ADV_UNIT))
 439                         tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
 440                 break;
 441         case MBR_DISCOVERED:
 442         case MBR_JOINING:
 443         case MBR_LEAVING:
 444         default:
 445                 break;
 446         }
 447 }
 448
 449 static void tipc_group_proto_xmit(struct tipc_group *grp, struct tipc_member *m,
 450                                   int mtyp, struct sk_buff_head *xmitq)
 451 {
 452         struct tipc_msg *hdr;
 453         struct sk_buff *skb;
 454         int adv = 0;
 455
 456         skb = tipc_msg_create(GROUP_PROTOCOL, mtyp, INT_H_SIZE, 0,
 457                               m->node, tipc_own_addr(grp->net),
 458                               m->port, grp->portid, 0);
 459         if (!skb)
 460                 return;
 461
 462         if (m->state == MBR_JOINED)
 463                 adv = ADV_ACTIVE - m->advertised;
 464
 465         hdr = buf_msg(skb);
 466
 467         if (mtyp == GRP_JOIN_MSG) {
 468                 msg_set_grp_bc_syncpt(hdr, grp->bc_snd_nxt);
 469                 msg_set_adv_win(hdr, adv);
 470                 m->advertised += adv;
 471         } else if (mtyp == GRP_ADV_MSG) {
 472                 msg_set_adv_win(hdr, adv);
 473                 m->advertised += adv;
 474         }
 475         __skb_queue_tail(xmitq, skb);
 476 }
 477
 478 void tipc_group_proto_rcv(struct tipc_group *grp, bool *usr_wakeup,
 479                           struct tipc_msg *hdr, struct sk_buff_head *inputq,
 480                           struct sk_buff_head *xmitq)
 481 {
 482         u32 node = msg_orignode(hdr);
 483         u32 port = msg_origport(hdr);
 484         struct tipc_member *m;
 485
 486         if (!grp)
 487                 return;
 488
 489         m = tipc_group_find_member(grp, node, port);
 490
 491         switch (msg_type(hdr)) {
 492         case GRP_JOIN_MSG:
 493                 if (!m)
 494                         m = tipc_group_create_member(grp, node, port,
 495                                                      MBR_QUARANTINED);
 496                 if (!m)
 497                         return;
 498                 m->bc_rcv_nxt = msg_grp_bc_syncpt(hdr);
 499                 m->window += msg_adv_win(hdr);
 500
 501                 /* Wait until PUBLISH event is received */
 502                 if (m->state == MBR_DISCOVERED) {
 503                         m->state = MBR_JOINING;
 504                 } else if (m->state == MBR_PUBLISHED) {
 505                         m->state = MBR_JOINED;
 506                         *usr_wakeup = true;
 507                         m->usr_pending = false;
 508                         tipc_group_proto_xmit(grp, m, GRP_ADV_MSG, xmitq);
 509                         __skb_queue_tail(inputq, m->event_msg);
 510                 }
 511                 if (m->window < ADV_IDLE)
 512                         tipc_group_update_member(m, 0);
 513                 else
 514                         list_del_init(&m->congested);
 515                 return;
 516         case GRP_LEAVE_MSG:
 517                 if (!m)
 518                         return;
 519
 520                 /* Wait until WITHDRAW event is received */
 521                 if (m->state != MBR_LEAVING) {
 522                         m->state = MBR_LEAVING;
 523                         return;
 524                 }
 525                 /* Otherwise deliver already received WITHDRAW event */
 526                 __skb_queue_tail(inputq, m->event_msg);
 527                 *usr_wakeup = m->usr_pending;
 528                 tipc_group_delete_member(grp, m);
 529                 list_del_init(&m->congested);
 530                 return;
 531         case GRP_ADV_MSG:
 532                 if (!m)
 533                         return;
 534                 m->window += msg_adv_win(hdr);
 535                 *usr_wakeup = m->usr_pending;
 536                 m->usr_pending = false;
 537                 list_del_init(&m->congested);
 538                 return;
 539         default:
 540                 pr_warn("Received unknown GROUP_PROTO message\n");
 541         }
 542 }
 543
 544 /* tipc_group_member_evt() - receive and handle a member up/down event
 545  */
 546 void tipc_group_member_evt(struct tipc_group *grp,
 547                            bool *usr_wakeup,
 548                            int *sk_rcvbuf,
 549                            struct sk_buff *skb,
 550                            struct sk_buff_head *inputq,
 551                            struct sk_buff_head *xmitq)
 552 {
 553         struct tipc_msg *hdr = buf_msg(skb);
 554         struct tipc_event *evt = (void *)msg_data(hdr);
 555         u32 instance = evt->found_lower;
 556         u32 node = evt->port.node;
 557         u32 port = evt->port.ref;
 558         int event = evt->event;
 559         struct tipc_member *m;
 560         struct net *net;
 561         u32 self;
 562
 563         if (!grp)
 564                 goto drop;
 565
 566         net = grp->net;
 567         self = tipc_own_addr(net);
 568         if (!grp->loopback && node == self && port == grp->portid)
 569                 goto drop;
 570
 571         /* Convert message before delivery to user */
 572         msg_set_hdr_sz(hdr, GROUP_H_SIZE);
 573         msg_set_user(hdr, TIPC_CRITICAL_IMPORTANCE);
 574         msg_set_type(hdr, TIPC_GRP_MEMBER_EVT);
 575         msg_set_origport(hdr, port);
 576         msg_set_orignode(hdr, node);
 577         msg_set_nametype(hdr, grp->type);
 578         msg_set_grp_evt(hdr, event);
 579
 580         m = tipc_group_find_member(grp, node, port);
 581
 582         if (event == TIPC_PUBLISHED) {
 583                 if (!m)
 584                         m = tipc_group_create_member(grp, node, port,
 585                                                      MBR_DISCOVERED);
 586                 if (!m)
 587                         goto drop;
 588
 589                 /* Hold back event if JOIN message not yet received */
 590                 if (m->state == MBR_DISCOVERED) {
 591                         m->event_msg = skb;
 592                         m->state = MBR_PUBLISHED;
 593                 } else {
 594                         __skb_queue_tail(inputq, skb);
 595                         m->state = MBR_JOINED;
 596                         *usr_wakeup = true;
 597                         m->usr_pending = false;
 598                 }
 599                 m->instance = instance;
 600                 TIPC_SKB_CB(skb)->orig_member = m->instance;
 601                 tipc_group_proto_xmit(grp, m, GRP_JOIN_MSG, xmitq);
 602                 if (m->window < ADV_IDLE)
 603                         tipc_group_update_member(m, 0);
 604                 else
 605                         list_del_init(&m->congested);
 606         } else if (event == TIPC_WITHDRAWN) {
 607                 if (!m)
 608                         goto drop;
 609
 610                 TIPC_SKB_CB(skb)->orig_member = m->instance;
 611
 612                 *usr_wakeup = m->usr_pending;
 613                 m->usr_pending = false;
 614
 615                 /* Hold back event if more messages might be expected */
 616                 if (m->state != MBR_LEAVING && tipc_node_is_up(net, node)) {
 617                         m->event_msg = skb;
 618                         m->state = MBR_LEAVING;
 619                 } else {
 620                         __skb_queue_tail(inputq, skb);
 621                         tipc_group_delete_member(grp, m);
 622                 }
 623                 list_del_init(&m->congested);
 624         }
 625         *sk_rcvbuf = tipc_group_rcvbuf_limit(grp);
 626         return;
 627 drop:
 628         kfree_skb(skb);
 629 }