net/tipc/port.c

   1 /*
   2  * net/tipc/port.c: TIPC port code
   3  *
   4  * Copyright (c) 1992-2007, Ericsson AB
   5  * Copyright (c) 2004-2008, 2010-2013, Wind River Systems
   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 "config.h"
  39 #include "port.h"
  40 #include "name_table.h"
  41
  42 /* Connection management: */
  43 #define PROBING_INTERVAL 3600000        /* [ms] => 1 h */
  44 #define CONFIRMED 0
  45 #define PROBING 1
  46
  47 #define MAX_REJECT_SIZE 1024
  48
  49 DEFINE_SPINLOCK(tipc_port_list_lock);
  50
  51 static LIST_HEAD(ports);
  52 static void port_handle_node_down(unsigned long ref);
  53 static struct sk_buff *port_build_self_abort_msg(struct tipc_port *, u32 err);
  54 static struct sk_buff *port_build_peer_abort_msg(struct tipc_port *, u32 err);
  55 static void port_timeout(unsigned long ref);
  56
  57
  58 static u32 port_peernode(struct tipc_port *p_ptr)
  59 {
  60         return msg_destnode(&p_ptr->phdr);
  61 }
  62
  63 static u32 port_peerport(struct tipc_port *p_ptr)
  64 {
  65         return msg_destport(&p_ptr->phdr);
  66 }
  67
  68 /**
  69  * tipc_port_peer_msg - verify message was sent by connected port's peer
  70  *
  71  * Handles cases where the node's network address has changed from
  72  * the default of <0.0.0> to its configured setting.
  73  */
  74 int tipc_port_peer_msg(struct tipc_port *p_ptr, struct tipc_msg *msg)
  75 {
  76         u32 peernode;
  77         u32 orignode;
  78
  79         if (msg_origport(msg) != port_peerport(p_ptr))
  80                 return 0;
  81
  82         orignode = msg_orignode(msg);
  83         peernode = port_peernode(p_ptr);
  84         return (orignode == peernode) ||
  85                 (!orignode && (peernode == tipc_own_addr)) ||
  86                 (!peernode && (orignode == tipc_own_addr));
  87 }
  88
  89 /**
  90  * tipc_multicast - send a multicast message to local and remote destinations
  91  */
  92 int tipc_multicast(u32 ref, struct tipc_name_seq const *seq,
  93                    struct iovec const *msg_sect, unsigned int len)
  94 {
  95         struct tipc_msg *hdr;
  96         struct sk_buff *buf;
  97         struct sk_buff *ibuf = NULL;
  98         struct tipc_port_list dports = {0, NULL, };
  99         struct tipc_port *oport = tipc_port_deref(ref);
 100         int ext_targets;
 101         int res;
 102
 103         if (unlikely(!oport))
 104                 return -EINVAL;
 105
 106         /* Create multicast message */
 107         hdr = &oport->phdr;
 108         msg_set_type(hdr, TIPC_MCAST_MSG);
 109         msg_set_lookup_scope(hdr, TIPC_CLUSTER_SCOPE);
 110         msg_set_destport(hdr, 0);
 111         msg_set_destnode(hdr, 0);
 112         msg_set_nametype(hdr, seq->type);
 113         msg_set_namelower(hdr, seq->lower);
 114         msg_set_nameupper(hdr, seq->upper);
 115         msg_set_hdr_sz(hdr, MCAST_H_SIZE);
 116         res = tipc_msg_build(hdr, msg_sect, len, MAX_MSG_SIZE, &buf);
 117         if (unlikely(!buf))
 118                 return res;
 119
 120         /* Figure out where to send multicast message */
 121         ext_targets = tipc_nametbl_mc_translate(seq->type, seq->lower, seq->upper,
 122                                                 TIPC_NODE_SCOPE, &dports);
 123
 124         /* Send message to destinations (duplicate it only if necessary) */
 125         if (ext_targets) {
 126                 if (dports.count != 0) {
 127                         ibuf = skb_copy(buf, GFP_ATOMIC);
 128                         if (ibuf == NULL) {
 129                                 tipc_port_list_free(&dports);
 130                                 kfree_skb(buf);
 131                                 return -ENOMEM;
 132                         }
 133                 }
 134                 res = tipc_bclink_send_msg(buf);
 135                 if ((res < 0) && (dports.count != 0))
 136                         kfree_skb(ibuf);
 137         } else {
 138                 ibuf = buf;
 139         }
 140
 141         if (res >= 0) {
 142                 if (ibuf)
 143                         tipc_port_recv_mcast(ibuf, &dports);
 144         } else {
 145                 tipc_port_list_free(&dports);
 146         }
 147         return res;
 148 }
 149
 150 /**
 151  * tipc_port_recv_mcast - deliver multicast message to all destination ports
 152  *
 153  * If there is no port list, perform a lookup to create one
 154  */
 155 void tipc_port_recv_mcast(struct sk_buff *buf, struct tipc_port_list *dp)
 156 {
 157         struct tipc_msg *msg;
 158         struct tipc_port_list dports = {0, NULL, };
 159         struct tipc_port_list *item = dp;
 160         int cnt = 0;
 161
 162         msg = buf_msg(buf);
 163
 164         /* Create destination port list, if one wasn't supplied */
 165         if (dp == NULL) {
 166                 tipc_nametbl_mc_translate(msg_nametype(msg),
 167                                      msg_namelower(msg),
 168                                      msg_nameupper(msg),
 169                                      TIPC_CLUSTER_SCOPE,
 170                                      &dports);
 171                 item = dp = &dports;
 172         }
 173
 174         /* Deliver a copy of message to each destination port */
 175         if (dp->count != 0) {
 176                 msg_set_destnode(msg, tipc_own_addr);
 177                 if (dp->count == 1) {
 178                         msg_set_destport(msg, dp->ports[0]);
 179                         tipc_port_recv_msg(buf);
 180                         tipc_port_list_free(dp);
 181                         return;
 182                 }
 183                 for (; cnt < dp->count; cnt++) {
 184                         int index = cnt % PLSIZE;
 185                         struct sk_buff *b = skb_clone(buf, GFP_ATOMIC);
 186
 187                         if (b == NULL) {
 188                                 pr_warn("Unable to deliver multicast message(s)\n");
 189                                 goto exit;
 190                         }
 191                         if ((index == 0) && (cnt != 0))
 192                                 item = item->next;
 193                         msg_set_destport(buf_msg(b), item->ports[index]);
 194                         tipc_port_recv_msg(b);
 195                 }
 196         }
 197 exit:
 198         kfree_skb(buf);
 199         tipc_port_list_free(dp);
 200 }
 201
 202 /**
 203  * tipc_createport - create a generic TIPC port
 204  *
 205  * Returns pointer to (locked) TIPC port, or NULL if unable to create it
 206  */
 207 struct tipc_port *tipc_createport(struct sock *sk,
 208                                   u32 (*dispatcher)(struct tipc_port *,
 209                                   struct sk_buff *),
 210                                   void (*wakeup)(struct tipc_port *),
 211                                   const u32 importance)
 212 {
 213         struct tipc_port *p_ptr;
 214         struct tipc_msg *msg;
 215         u32 ref;
 216
 217         p_ptr = kzalloc(sizeof(*p_ptr), GFP_ATOMIC);
 218         if (!p_ptr) {
 219                 pr_warn("Port creation failed, no memory\n");
 220                 return NULL;
 221         }
 222         ref = tipc_ref_acquire(p_ptr, &p_ptr->lock);
 223         if (!ref) {
 224                 pr_warn("Port creation failed, ref. table exhausted\n");
 225                 kfree(p_ptr);
 226                 return NULL;
 227         }
 228
 229         p_ptr->sk = sk;
 230         p_ptr->max_pkt = MAX_PKT_DEFAULT;
 231         p_ptr->ref = ref;
 232         INIT_LIST_HEAD(&p_ptr->wait_list);
 233         INIT_LIST_HEAD(&p_ptr->subscription.nodesub_list);
 234         p_ptr->dispatcher = dispatcher;
 235         p_ptr->wakeup = wakeup;
 236         k_init_timer(&p_ptr->timer, (Handler)port_timeout, ref);
 237         INIT_LIST_HEAD(&p_ptr->publications);
 238         INIT_LIST_HEAD(&p_ptr->port_list);
 239
 240         /*
 241          * Must hold port list lock while initializing message header template
 242          * to ensure a change to node's own network address doesn't result
 243          * in template containing out-dated network address information
 244          */
 245         spin_lock_bh(&tipc_port_list_lock);
 246         msg = &p_ptr->phdr;
 247         tipc_msg_init(msg, importance, TIPC_NAMED_MSG, NAMED_H_SIZE, 0);
 248         msg_set_origport(msg, ref);
 249         list_add_tail(&p_ptr->port_list, &ports);
 250         spin_unlock_bh(&tipc_port_list_lock);
 251         return p_ptr;
 252 }
 253
 254 int tipc_deleteport(u32 ref)
 255 {
 256         struct tipc_port *p_ptr;
 257         struct sk_buff *buf = NULL;
 258
 259         tipc_withdraw(ref, 0, NULL);
 260         p_ptr = tipc_port_lock(ref);
 261         if (!p_ptr)
 262                 return -EINVAL;
 263
 264         tipc_ref_discard(ref);
 265         tipc_port_unlock(p_ptr);
 266
 267         k_cancel_timer(&p_ptr->timer);
 268         if (p_ptr->connected) {
 269                 buf = port_build_peer_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
 270                 tipc_nodesub_unsubscribe(&p_ptr->subscription);
 271         }
 272
 273         spin_lock_bh(&tipc_port_list_lock);
 274         list_del(&p_ptr->port_list);
 275         list_del(&p_ptr->wait_list);
 276         spin_unlock_bh(&tipc_port_list_lock);
 277         k_term_timer(&p_ptr->timer);
 278         kfree(p_ptr);
 279         tipc_net_route_msg(buf);
 280         return 0;
 281 }
 282
 283 static int port_unreliable(struct tipc_port *p_ptr)
 284 {
 285         return msg_src_droppable(&p_ptr->phdr);
 286 }
 287
 288 int tipc_portunreliable(u32 ref, unsigned int *isunreliable)
 289 {
 290         struct tipc_port *p_ptr;
 291
 292         p_ptr = tipc_port_lock(ref);
 293         if (!p_ptr)
 294                 return -EINVAL;
 295         *isunreliable = port_unreliable(p_ptr);
 296         tipc_port_unlock(p_ptr);
 297         return 0;
 298 }
 299
 300 int tipc_set_portunreliable(u32 ref, unsigned int isunreliable)
 301 {
 302         struct tipc_port *p_ptr;
 303
 304         p_ptr = tipc_port_lock(ref);
 305         if (!p_ptr)
 306                 return -EINVAL;
 307         msg_set_src_droppable(&p_ptr->phdr, (isunreliable != 0));
 308         tipc_port_unlock(p_ptr);
 309         return 0;
 310 }
 311
 312 static int port_unreturnable(struct tipc_port *p_ptr)
 313 {
 314         return msg_dest_droppable(&p_ptr->phdr);
 315 }
 316
 317 int tipc_portunreturnable(u32 ref, unsigned int *isunrejectable)
 318 {
 319         struct tipc_port *p_ptr;
 320
 321         p_ptr = tipc_port_lock(ref);
 322         if (!p_ptr)
 323                 return -EINVAL;
 324         *isunrejectable = port_unreturnable(p_ptr);
 325         tipc_port_unlock(p_ptr);
 326         return 0;
 327 }
 328
 329 int tipc_set_portunreturnable(u32 ref, unsigned int isunrejectable)
 330 {
 331         struct tipc_port *p_ptr;
 332
 333         p_ptr = tipc_port_lock(ref);
 334         if (!p_ptr)
 335                 return -EINVAL;
 336         msg_set_dest_droppable(&p_ptr->phdr, (isunrejectable != 0));
 337         tipc_port_unlock(p_ptr);
 338         return 0;
 339 }
 340
 341 /*
 342  * port_build_proto_msg(): create connection protocol message for port
 343  *
 344  * On entry the port must be locked and connected.
 345  */
 346 static struct sk_buff *port_build_proto_msg(struct tipc_port *p_ptr,
 347                                             u32 type, u32 ack)
 348 {
 349         struct sk_buff *buf;
 350         struct tipc_msg *msg;
 351
 352         buf = tipc_buf_acquire(INT_H_SIZE);
 353         if (buf) {
 354                 msg = buf_msg(buf);
 355                 tipc_msg_init(msg, CONN_MANAGER, type, INT_H_SIZE,
 356                               port_peernode(p_ptr));
 357                 msg_set_destport(msg, port_peerport(p_ptr));
 358                 msg_set_origport(msg, p_ptr->ref);
 359                 msg_set_msgcnt(msg, ack);
 360         }
 361         return buf;
 362 }
 363
 364 int tipc_reject_msg(struct sk_buff *buf, u32 err)
 365 {
 366         struct tipc_msg *msg = buf_msg(buf);
 367         struct sk_buff *rbuf;
 368         struct tipc_msg *rmsg;
 369         int hdr_sz;
 370         u32 imp;
 371         u32 data_sz = msg_data_sz(msg);
 372         u32 src_node;
 373         u32 rmsg_sz;
 374
 375         /* discard rejected message if it shouldn't be returned to sender */
 376         if (WARN(!msg_isdata(msg),
 377                  "attempt to reject message with user=%u", msg_user(msg))) {
 378                 dump_stack();
 379                 goto exit;
 380         }
 381         if (msg_errcode(msg) || msg_dest_droppable(msg))
 382                 goto exit;
 383
 384         /*
 385          * construct returned message by copying rejected message header and
 386          * data (or subset), then updating header fields that need adjusting
 387          */
 388         hdr_sz = msg_hdr_sz(msg);
 389         rmsg_sz = hdr_sz + min_t(u32, data_sz, MAX_REJECT_SIZE);
 390
 391         rbuf = tipc_buf_acquire(rmsg_sz);
 392         if (rbuf == NULL)
 393                 goto exit;
 394
 395         rmsg = buf_msg(rbuf);
 396         skb_copy_to_linear_data(rbuf, msg, rmsg_sz);
 397
 398         if (msg_connected(rmsg)) {
 399                 imp = msg_importance(rmsg);
 400                 if (imp < TIPC_CRITICAL_IMPORTANCE)
 401                         msg_set_importance(rmsg, ++imp);
 402         }
 403         msg_set_non_seq(rmsg, 0);
 404         msg_set_size(rmsg, rmsg_sz);
 405         msg_set_errcode(rmsg, err);
 406         msg_set_prevnode(rmsg, tipc_own_addr);
 407         msg_swap_words(rmsg, 4, 5);
 408         if (!msg_short(rmsg))
 409                 msg_swap_words(rmsg, 6, 7);
 410
 411         /* send self-abort message when rejecting on a connected port */
 412         if (msg_connected(msg)) {
 413                 struct tipc_port *p_ptr = tipc_port_lock(msg_destport(msg));
 414
 415                 if (p_ptr) {
 416                         struct sk_buff *abuf = NULL;
 417
 418                         if (p_ptr->connected)
 419                                 abuf = port_build_self_abort_msg(p_ptr, err);
 420                         tipc_port_unlock(p_ptr);
 421                         tipc_net_route_msg(abuf);
 422                 }
 423         }
 424
 425         /* send returned message & dispose of rejected message */
 426         src_node = msg_prevnode(msg);
 427         if (in_own_node(src_node))
 428                 tipc_port_recv_msg(rbuf);
 429         else
 430                 tipc_link_send(rbuf, src_node, msg_link_selector(rmsg));
 431 exit:
 432         kfree_skb(buf);
 433         return data_sz;
 434 }
 435
 436 int tipc_port_reject_sections(struct tipc_port *p_ptr, struct tipc_msg *hdr,
 437                               struct iovec const *msg_sect, unsigned int len,
 438                               int err)
 439 {
 440         struct sk_buff *buf;
 441         int res;
 442
 443         res = tipc_msg_build(hdr, msg_sect, len, MAX_MSG_SIZE, &buf);
 444         if (!buf)
 445                 return res;
 446
 447         return tipc_reject_msg(buf, err);
 448 }
 449
 450 static void port_timeout(unsigned long ref)
 451 {
 452         struct tipc_port *p_ptr = tipc_port_lock(ref);
 453         struct sk_buff *buf = NULL;
 454
 455         if (!p_ptr)
 456                 return;
 457
 458         if (!p_ptr->connected) {
 459                 tipc_port_unlock(p_ptr);
 460                 return;
 461         }
 462
 463         /* Last probe answered ? */
 464         if (p_ptr->probing_state == PROBING) {
 465                 buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
 466         } else {
 467                 buf = port_build_proto_msg(p_ptr, CONN_PROBE, 0);
 468                 p_ptr->probing_state = PROBING;
 469                 k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
 470         }
 471         tipc_port_unlock(p_ptr);
 472         tipc_net_route_msg(buf);
 473 }
 474
 475
 476 static void port_handle_node_down(unsigned long ref)
 477 {
 478         struct tipc_port *p_ptr = tipc_port_lock(ref);
 479         struct sk_buff *buf = NULL;
 480
 481         if (!p_ptr)
 482                 return;
 483         buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_NODE);
 484         tipc_port_unlock(p_ptr);
 485         tipc_net_route_msg(buf);
 486 }
 487
 488
 489 static struct sk_buff *port_build_self_abort_msg(struct tipc_port *p_ptr, u32 err)
 490 {
 491         struct sk_buff *buf = port_build_peer_abort_msg(p_ptr, err);
 492
 493         if (buf) {
 494                 struct tipc_msg *msg = buf_msg(buf);
 495                 msg_swap_words(msg, 4, 5);
 496                 msg_swap_words(msg, 6, 7);
 497         }
 498         return buf;
 499 }
 500
 501
 502 static struct sk_buff *port_build_peer_abort_msg(struct tipc_port *p_ptr, u32 err)
 503 {
 504         struct sk_buff *buf;
 505         struct tipc_msg *msg;
 506         u32 imp;
 507
 508         if (!p_ptr->connected)
 509                 return NULL;
 510
 511         buf = tipc_buf_acquire(BASIC_H_SIZE);
 512         if (buf) {
 513                 msg = buf_msg(buf);
 514                 memcpy(msg, &p_ptr->phdr, BASIC_H_SIZE);
 515                 msg_set_hdr_sz(msg, BASIC_H_SIZE);
 516                 msg_set_size(msg, BASIC_H_SIZE);
 517                 imp = msg_importance(msg);
 518                 if (imp < TIPC_CRITICAL_IMPORTANCE)
 519                         msg_set_importance(msg, ++imp);
 520                 msg_set_errcode(msg, err);
 521         }
 522         return buf;
 523 }
 524
 525 void tipc_port_recv_proto_msg(struct sk_buff *buf)
 526 {
 527         struct tipc_msg *msg = buf_msg(buf);
 528         struct tipc_port *p_ptr;
 529         struct sk_buff *r_buf = NULL;
 530         u32 destport = msg_destport(msg);
 531         int wakeable;
 532
 533         /* Validate connection */
 534         p_ptr = tipc_port_lock(destport);
 535         if (!p_ptr || !p_ptr->connected || !tipc_port_peer_msg(p_ptr, msg)) {
 536                 r_buf = tipc_buf_acquire(BASIC_H_SIZE);
 537                 if (r_buf) {
 538                         msg = buf_msg(r_buf);
 539                         tipc_msg_init(msg, TIPC_HIGH_IMPORTANCE, TIPC_CONN_MSG,
 540                                       BASIC_H_SIZE, msg_orignode(msg));
 541                         msg_set_errcode(msg, TIPC_ERR_NO_PORT);
 542                         msg_set_origport(msg, destport);
 543                         msg_set_destport(msg, msg_origport(msg));
 544                 }
 545                 if (p_ptr)
 546                         tipc_port_unlock(p_ptr);
 547                 goto exit;
 548         }
 549
 550         /* Process protocol message sent by peer */
 551         switch (msg_type(msg)) {
 552         case CONN_ACK:
 553                 wakeable = tipc_port_congested(p_ptr) && p_ptr->congested &&
 554                         p_ptr->wakeup;
 555                 p_ptr->acked += msg_msgcnt(msg);
 556                 if (!tipc_port_congested(p_ptr)) {
 557                         p_ptr->congested = 0;
 558                         if (wakeable)
 559                                 p_ptr->wakeup(p_ptr);
 560                 }
 561                 break;
 562         case CONN_PROBE:
 563                 r_buf = port_build_proto_msg(p_ptr, CONN_PROBE_REPLY, 0);
 564                 break;
 565         default:
 566                 /* CONN_PROBE_REPLY or unrecognized - no action required */
 567                 break;
 568         }
 569         p_ptr->probing_state = CONFIRMED;
 570         tipc_port_unlock(p_ptr);
 571 exit:
 572         tipc_net_route_msg(r_buf);
 573         kfree_skb(buf);
 574 }
 575
 576 static int port_print(struct tipc_port *p_ptr, char *buf, int len, int full_id)
 577 {
 578         struct publication *publ;
 579         int ret;
 580
 581         if (full_id)
 582                 ret = tipc_snprintf(buf, len, "<%u.%u.%u:%u>:",
 583                                     tipc_zone(tipc_own_addr),
 584                                     tipc_cluster(tipc_own_addr),
 585                                     tipc_node(tipc_own_addr), p_ptr->ref);
 586         else
 587                 ret = tipc_snprintf(buf, len, "%-10u:", p_ptr->ref);
 588
 589         if (p_ptr->connected) {
 590                 u32 dport = port_peerport(p_ptr);
 591                 u32 destnode = port_peernode(p_ptr);
 592
 593                 ret += tipc_snprintf(buf + ret, len - ret,
 594                                      " connected to <%u.%u.%u:%u>",
 595                                      tipc_zone(destnode),
 596                                      tipc_cluster(destnode),
 597                                      tipc_node(destnode), dport);
 598                 if (p_ptr->conn_type != 0)
 599                         ret += tipc_snprintf(buf + ret, len - ret,
 600                                              " via {%u,%u}", p_ptr->conn_type,
 601                                              p_ptr->conn_instance);
 602         } else if (p_ptr->published) {
 603                 ret += tipc_snprintf(buf + ret, len - ret, " bound to");
 604                 list_for_each_entry(publ, &p_ptr->publications, pport_list) {
 605                         if (publ->lower == publ->upper)
 606                                 ret += tipc_snprintf(buf + ret, len - ret,
 607                                                      " {%u,%u}", publ->type,
 608                                                      publ->lower);
 609                         else
 610                                 ret += tipc_snprintf(buf + ret, len - ret,
 611                                                      " {%u,%u,%u}", publ->type,
 612                                                      publ->lower, publ->upper);
 613                 }
 614         }
 615         ret += tipc_snprintf(buf + ret, len - ret, "\n");
 616         return ret;
 617 }
 618
 619 struct sk_buff *tipc_port_get_ports(void)
 620 {
 621         struct sk_buff *buf;
 622         struct tlv_desc *rep_tlv;
 623         char *pb;
 624         int pb_len;
 625         struct tipc_port *p_ptr;
 626         int str_len = 0;
 627
 628         buf = tipc_cfg_reply_alloc(TLV_SPACE(ULTRA_STRING_MAX_LEN));
 629         if (!buf)
 630                 return NULL;
 631         rep_tlv = (struct tlv_desc *)buf->data;
 632         pb = TLV_DATA(rep_tlv);
 633         pb_len = ULTRA_STRING_MAX_LEN;
 634
 635         spin_lock_bh(&tipc_port_list_lock);
 636         list_for_each_entry(p_ptr, &ports, port_list) {
 637                 spin_lock_bh(p_ptr->lock);
 638                 str_len += port_print(p_ptr, pb, pb_len, 0);
 639                 spin_unlock_bh(p_ptr->lock);
 640         }
 641         spin_unlock_bh(&tipc_port_list_lock);
 642         str_len += 1;   /* for "\0" */
 643         skb_put(buf, TLV_SPACE(str_len));
 644         TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
 645
 646         return buf;
 647 }
 648
 649 void tipc_port_reinit(void)
 650 {
 651         struct tipc_port *p_ptr;
 652         struct tipc_msg *msg;
 653
 654         spin_lock_bh(&tipc_port_list_lock);
 655         list_for_each_entry(p_ptr, &ports, port_list) {
 656                 msg = &p_ptr->phdr;
 657                 msg_set_prevnode(msg, tipc_own_addr);
 658                 msg_set_orignode(msg, tipc_own_addr);
 659         }
 660         spin_unlock_bh(&tipc_port_list_lock);
 661 }
 662
 663 void tipc_acknowledge(u32 ref, u32 ack)
 664 {
 665         struct tipc_port *p_ptr;
 666         struct sk_buff *buf = NULL;
 667
 668         p_ptr = tipc_port_lock(ref);
 669         if (!p_ptr)
 670                 return;
 671         if (p_ptr->connected) {
 672                 p_ptr->conn_unacked -= ack;
 673                 buf = port_build_proto_msg(p_ptr, CONN_ACK, ack);
 674         }
 675         tipc_port_unlock(p_ptr);
 676         tipc_net_route_msg(buf);
 677 }
 678
 679 int tipc_portimportance(u32 ref, unsigned int *importance)
 680 {
 681         struct tipc_port *p_ptr;
 682
 683         p_ptr = tipc_port_lock(ref);
 684         if (!p_ptr)
 685                 return -EINVAL;
 686         *importance = (unsigned int)msg_importance(&p_ptr->phdr);
 687         tipc_port_unlock(p_ptr);
 688         return 0;
 689 }
 690
 691 int tipc_set_portimportance(u32 ref, unsigned int imp)
 692 {
 693         struct tipc_port *p_ptr;
 694
 695         if (imp > TIPC_CRITICAL_IMPORTANCE)
 696                 return -EINVAL;
 697
 698         p_ptr = tipc_port_lock(ref);
 699         if (!p_ptr)
 700                 return -EINVAL;
 701         msg_set_importance(&p_ptr->phdr, (u32)imp);
 702         tipc_port_unlock(p_ptr);
 703         return 0;
 704 }
 705
 706
 707 int tipc_publish(u32 ref, unsigned int scope, struct tipc_name_seq const *seq)
 708 {
 709         struct tipc_port *p_ptr;
 710         struct publication *publ;
 711         u32 key;
 712         int res = -EINVAL;
 713
 714         p_ptr = tipc_port_lock(ref);
 715         if (!p_ptr)
 716                 return -EINVAL;
 717
 718         if (p_ptr->connected)
 719                 goto exit;
 720         key = ref + p_ptr->pub_count + 1;
 721         if (key == ref) {
 722                 res = -EADDRINUSE;
 723                 goto exit;
 724         }
 725         publ = tipc_nametbl_publish(seq->type, seq->lower, seq->upper,
 726                                     scope, p_ptr->ref, key);
 727         if (publ) {
 728                 list_add(&publ->pport_list, &p_ptr->publications);
 729                 p_ptr->pub_count++;
 730                 p_ptr->published = 1;
 731                 res = 0;
 732         }
 733 exit:
 734         tipc_port_unlock(p_ptr);
 735         return res;
 736 }
 737
 738 int tipc_withdraw(u32 ref, unsigned int scope, struct tipc_name_seq const *seq)
 739 {
 740         struct tipc_port *p_ptr;
 741         struct publication *publ;
 742         struct publication *tpubl;
 743         int res = -EINVAL;
 744
 745         p_ptr = tipc_port_lock(ref);
 746         if (!p_ptr)
 747                 return -EINVAL;
 748         if (!seq) {
 749                 list_for_each_entry_safe(publ, tpubl,
 750                                          &p_ptr->publications, pport_list) {
 751                         tipc_nametbl_withdraw(publ->type, publ->lower,
 752                                               publ->ref, publ->key);
 753                 }
 754                 res = 0;
 755         } else {
 756                 list_for_each_entry_safe(publ, tpubl,
 757                                          &p_ptr->publications, pport_list) {
 758                         if (publ->scope != scope)
 759                                 continue;
 760                         if (publ->type != seq->type)
 761                                 continue;
 762                         if (publ->lower != seq->lower)
 763                                 continue;
 764                         if (publ->upper != seq->upper)
 765                                 break;
 766                         tipc_nametbl_withdraw(publ->type, publ->lower,
 767                                               publ->ref, publ->key);
 768                         res = 0;
 769                         break;
 770                 }
 771         }
 772         if (list_empty(&p_ptr->publications))
 773                 p_ptr->published = 0;
 774         tipc_port_unlock(p_ptr);
 775         return res;
 776 }
 777
 778 int tipc_connect(u32 ref, struct tipc_portid const *peer)
 779 {
 780         struct tipc_port *p_ptr;
 781         int res;
 782
 783         p_ptr = tipc_port_lock(ref);
 784         if (!p_ptr)
 785                 return -EINVAL;
 786         res = __tipc_connect(ref, p_ptr, peer);
 787         tipc_port_unlock(p_ptr);
 788         return res;
 789 }
 790
 791 /*
 792  * __tipc_connect - connect to a remote peer
 793  *
 794  * Port must be locked.
 795  */
 796 int __tipc_connect(u32 ref, struct tipc_port *p_ptr,
 797                         struct tipc_portid const *peer)
 798 {
 799         struct tipc_msg *msg;
 800         int res = -EINVAL;
 801
 802         if (p_ptr->published || p_ptr->connected)
 803                 goto exit;
 804         if (!peer->ref)
 805                 goto exit;
 806
 807         msg = &p_ptr->phdr;
 808         msg_set_destnode(msg, peer->node);
 809         msg_set_destport(msg, peer->ref);
 810         msg_set_type(msg, TIPC_CONN_MSG);
 811         msg_set_lookup_scope(msg, 0);
 812         msg_set_hdr_sz(msg, SHORT_H_SIZE);
 813
 814         p_ptr->probing_interval = PROBING_INTERVAL;
 815         p_ptr->probing_state = CONFIRMED;
 816         p_ptr->connected = 1;
 817         k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
 818
 819         tipc_nodesub_subscribe(&p_ptr->subscription, peer->node,
 820                           (void *)(unsigned long)ref,
 821                           (net_ev_handler)port_handle_node_down);
 822         res = 0;
 823 exit:
 824         p_ptr->max_pkt = tipc_link_get_max_pkt(peer->node, ref);
 825         return res;
 826 }
 827
 828 /*
 829  * __tipc_disconnect - disconnect port from peer
 830  *
 831  * Port must be locked.
 832  */
 833 int __tipc_disconnect(struct tipc_port *tp_ptr)
 834 {
 835         int res;
 836
 837         if (tp_ptr->connected) {
 838                 tp_ptr->connected = 0;
 839                 /* let timer expire on it's own to avoid deadlock! */
 840                 tipc_nodesub_unsubscribe(&tp_ptr->subscription);
 841                 res = 0;
 842         } else {
 843                 res = -ENOTCONN;
 844         }
 845         return res;
 846 }
 847
 848 /*
 849  * tipc_disconnect(): Disconnect port form peer.
 850  *                    This is a node local operation.
 851  */
 852 int tipc_disconnect(u32 ref)
 853 {
 854         struct tipc_port *p_ptr;
 855         int res;
 856
 857         p_ptr = tipc_port_lock(ref);
 858         if (!p_ptr)
 859                 return -EINVAL;
 860         res = __tipc_disconnect(p_ptr);
 861         tipc_port_unlock(p_ptr);
 862         return res;
 863 }
 864
 865 /*
 866  * tipc_shutdown(): Send a SHUTDOWN msg to peer and disconnect
 867  */
 868 int tipc_shutdown(u32 ref)
 869 {
 870         struct tipc_port *p_ptr;
 871         struct sk_buff *buf = NULL;
 872
 873         p_ptr = tipc_port_lock(ref);
 874         if (!p_ptr)
 875                 return -EINVAL;
 876
 877         buf = port_build_peer_abort_msg(p_ptr, TIPC_CONN_SHUTDOWN);
 878         tipc_port_unlock(p_ptr);
 879         tipc_net_route_msg(buf);
 880         return tipc_disconnect(ref);
 881 }
 882
 883 /**
 884  * tipc_port_recv_msg - receive message from lower layer and deliver to port user
 885  */
 886 int tipc_port_recv_msg(struct sk_buff *buf)
 887 {
 888         struct tipc_port *p_ptr;
 889         struct tipc_msg *msg = buf_msg(buf);
 890         u32 destport = msg_destport(msg);
 891         u32 dsz = msg_data_sz(msg);
 892         u32 err;
 893
 894         /* forward unresolved named message */
 895         if (unlikely(!destport)) {
 896                 tipc_net_route_msg(buf);
 897                 return dsz;
 898         }
 899
 900         /* validate destination & pass to port, otherwise reject message */
 901         p_ptr = tipc_port_lock(destport);
 902         if (likely(p_ptr)) {
 903                 err = p_ptr->dispatcher(p_ptr, buf);
 904                 tipc_port_unlock(p_ptr);
 905                 if (likely(!err))
 906                         return dsz;
 907         } else {
 908                 err = TIPC_ERR_NO_PORT;
 909         }
 910
 911         return tipc_reject_msg(buf, err);
 912 }
 913
 914 /*
 915  *  tipc_port_recv_sections(): Concatenate and deliver sectioned
 916  *                        message for this node.
 917  */
 918 static int tipc_port_recv_sections(struct tipc_port *sender,
 919                                    struct iovec const *msg_sect,
 920                                    unsigned int len)
 921 {
 922         struct sk_buff *buf;
 923         int res;
 924
 925         res = tipc_msg_build(&sender->phdr, msg_sect, len, MAX_MSG_SIZE, &buf);
 926         if (likely(buf))
 927                 tipc_port_recv_msg(buf);
 928         return res;
 929 }
 930
 931 /**
 932  * tipc_send - send message sections on connection
 933  */
 934 int tipc_send(u32 ref, struct iovec const *msg_sect, unsigned int len)
 935 {
 936         struct tipc_port *p_ptr;
 937         u32 destnode;
 938         int res;
 939
 940         p_ptr = tipc_port_deref(ref);
 941         if (!p_ptr || !p_ptr->connected)
 942                 return -EINVAL;
 943
 944         p_ptr->congested = 1;
 945         if (!tipc_port_congested(p_ptr)) {
 946                 destnode = port_peernode(p_ptr);
 947                 if (likely(!in_own_node(destnode)))
 948                         res = tipc_link_send_sections_fast(p_ptr, msg_sect,
 949                                                            len, destnode);
 950                 else
 951                         res = tipc_port_recv_sections(p_ptr, msg_sect, len);
 952
 953                 if (likely(res != -ELINKCONG)) {
 954                         p_ptr->congested = 0;
 955                         if (res > 0)
 956                                 p_ptr->sent++;
 957                         return res;
 958                 }
 959         }
 960         if (port_unreliable(p_ptr)) {
 961                 p_ptr->congested = 0;
 962                 return len;
 963         }
 964         return -ELINKCONG;
 965 }
 966
 967 /**
 968  * tipc_send2name - send message sections to port name
 969  */
 970 int tipc_send2name(u32 ref, struct tipc_name const *name, unsigned int domain,
 971                    struct iovec const *msg_sect, unsigned int len)
 972 {
 973         struct tipc_port *p_ptr;
 974         struct tipc_msg *msg;
 975         u32 destnode = domain;
 976         u32 destport;
 977         int res;
 978
 979         p_ptr = tipc_port_deref(ref);
 980         if (!p_ptr || p_ptr->connected)
 981                 return -EINVAL;
 982
 983         msg = &p_ptr->phdr;
 984         msg_set_type(msg, TIPC_NAMED_MSG);
 985         msg_set_hdr_sz(msg, NAMED_H_SIZE);
 986         msg_set_nametype(msg, name->type);
 987         msg_set_nameinst(msg, name->instance);
 988         msg_set_lookup_scope(msg, tipc_addr_scope(domain));
 989         destport = tipc_nametbl_translate(name->type, name->instance, &destnode);
 990         msg_set_destnode(msg, destnode);
 991         msg_set_destport(msg, destport);
 992
 993         if (likely(destport || destnode)) {
 994                 if (likely(in_own_node(destnode)))
 995                         res = tipc_port_recv_sections(p_ptr, msg_sect, len);
 996                 else if (tipc_own_addr)
 997                         res = tipc_link_send_sections_fast(p_ptr, msg_sect,
 998                                                            len, destnode);
 999                 else
1000                         res = tipc_port_reject_sections(p_ptr, msg, msg_sect,
1001                                                         len, TIPC_ERR_NO_NODE);
1002                 if (likely(res != -ELINKCONG)) {
1003                         if (res > 0)
1004                                 p_ptr->sent++;
1005                         return res;
1006                 }
1007                 if (port_unreliable(p_ptr)) {
1008                         return len;
1009                 }
1010                 return -ELINKCONG;
1011         }
1012         return tipc_port_reject_sections(p_ptr, msg, msg_sect, len,
1013                                          TIPC_ERR_NO_NAME);
1014 }
1015
1016 /**
1017  * tipc_send2port - send message sections to port identity
1018  */
1019 int tipc_send2port(u32 ref, struct tipc_portid const *dest,
1020                    struct iovec const *msg_sect, unsigned int len)
1021 {
1022         struct tipc_port *p_ptr;
1023         struct tipc_msg *msg;
1024         int res;
1025
1026         p_ptr = tipc_port_deref(ref);
1027         if (!p_ptr || p_ptr->connected)
1028                 return -EINVAL;
1029
1030         msg = &p_ptr->phdr;
1031         msg_set_type(msg, TIPC_DIRECT_MSG);
1032         msg_set_lookup_scope(msg, 0);
1033         msg_set_destnode(msg, dest->node);
1034         msg_set_destport(msg, dest->ref);
1035         msg_set_hdr_sz(msg, BASIC_H_SIZE);
1036
1037         if (in_own_node(dest->node))
1038                 res =  tipc_port_recv_sections(p_ptr, msg_sect, len);
1039         else if (tipc_own_addr)
1040                 res = tipc_link_send_sections_fast(p_ptr, msg_sect, len,
1041                                                    dest->node);
1042         else
1043                 res = tipc_port_reject_sections(p_ptr, msg, msg_sect, len,
1044                                                 TIPC_ERR_NO_NODE);
1045         if (likely(res != -ELINKCONG)) {
1046                 if (res > 0)
1047                         p_ptr->sent++;
1048                 return res;
1049         }
1050         if (port_unreliable(p_ptr)) {
1051                 return len;
1052         }
1053         return -ELINKCONG;
1054 }