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(struct tipc_port *p_ptr)
 255 {
 256         struct sk_buff *buf = NULL;
 257
 258         tipc_withdraw(p_ptr, 0, NULL);
 259
 260         spin_lock_bh(p_ptr->lock);
 261         tipc_ref_discard(p_ptr->ref);
 262         spin_unlock_bh(p_ptr->lock);
 263
 264         k_cancel_timer(&p_ptr->timer);
 265         if (p_ptr->connected) {
 266                 buf = port_build_peer_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
 267                 tipc_nodesub_unsubscribe(&p_ptr->subscription);
 268         }
 269
 270         spin_lock_bh(&tipc_port_list_lock);
 271         list_del(&p_ptr->port_list);
 272         list_del(&p_ptr->wait_list);
 273         spin_unlock_bh(&tipc_port_list_lock);
 274         k_term_timer(&p_ptr->timer);
 275         kfree(p_ptr);
 276         tipc_net_route_msg(buf);
 277         return 0;
 278 }
 279
 280 static int port_unreliable(struct tipc_port *p_ptr)
 281 {
 282         return msg_src_droppable(&p_ptr->phdr);
 283 }
 284
 285 int tipc_portunreliable(u32 ref, unsigned int *isunreliable)
 286 {
 287         struct tipc_port *p_ptr;
 288
 289         p_ptr = tipc_port_lock(ref);
 290         if (!p_ptr)
 291                 return -EINVAL;
 292         *isunreliable = port_unreliable(p_ptr);
 293         tipc_port_unlock(p_ptr);
 294         return 0;
 295 }
 296
 297 int tipc_set_portunreliable(u32 ref, unsigned int isunreliable)
 298 {
 299         struct tipc_port *p_ptr;
 300
 301         p_ptr = tipc_port_lock(ref);
 302         if (!p_ptr)
 303                 return -EINVAL;
 304         msg_set_src_droppable(&p_ptr->phdr, (isunreliable != 0));
 305         tipc_port_unlock(p_ptr);
 306         return 0;
 307 }
 308
 309 static int port_unreturnable(struct tipc_port *p_ptr)
 310 {
 311         return msg_dest_droppable(&p_ptr->phdr);
 312 }
 313
 314 int tipc_portunreturnable(u32 ref, unsigned int *isunrejectable)
 315 {
 316         struct tipc_port *p_ptr;
 317
 318         p_ptr = tipc_port_lock(ref);
 319         if (!p_ptr)
 320                 return -EINVAL;
 321         *isunrejectable = port_unreturnable(p_ptr);
 322         tipc_port_unlock(p_ptr);
 323         return 0;
 324 }
 325
 326 int tipc_set_portunreturnable(u32 ref, unsigned int isunrejectable)
 327 {
 328         struct tipc_port *p_ptr;
 329
 330         p_ptr = tipc_port_lock(ref);
 331         if (!p_ptr)
 332                 return -EINVAL;
 333         msg_set_dest_droppable(&p_ptr->phdr, (isunrejectable != 0));
 334         tipc_port_unlock(p_ptr);
 335         return 0;
 336 }
 337
 338 /*
 339  * port_build_proto_msg(): create connection protocol message for port
 340  *
 341  * On entry the port must be locked and connected.
 342  */
 343 static struct sk_buff *port_build_proto_msg(struct tipc_port *p_ptr,
 344                                             u32 type, u32 ack)
 345 {
 346         struct sk_buff *buf;
 347         struct tipc_msg *msg;
 348
 349         buf = tipc_buf_acquire(INT_H_SIZE);
 350         if (buf) {
 351                 msg = buf_msg(buf);
 352                 tipc_msg_init(msg, CONN_MANAGER, type, INT_H_SIZE,
 353                               port_peernode(p_ptr));
 354                 msg_set_destport(msg, port_peerport(p_ptr));
 355                 msg_set_origport(msg, p_ptr->ref);
 356                 msg_set_msgcnt(msg, ack);
 357         }
 358         return buf;
 359 }
 360
 361 int tipc_reject_msg(struct sk_buff *buf, u32 err)
 362 {
 363         struct tipc_msg *msg = buf_msg(buf);
 364         struct sk_buff *rbuf;
 365         struct tipc_msg *rmsg;
 366         int hdr_sz;
 367         u32 imp;
 368         u32 data_sz = msg_data_sz(msg);
 369         u32 src_node;
 370         u32 rmsg_sz;
 371
 372         /* discard rejected message if it shouldn't be returned to sender */
 373         if (WARN(!msg_isdata(msg),
 374                  "attempt to reject message with user=%u", msg_user(msg))) {
 375                 dump_stack();
 376                 goto exit;
 377         }
 378         if (msg_errcode(msg) || msg_dest_droppable(msg))
 379                 goto exit;
 380
 381         /*
 382          * construct returned message by copying rejected message header and
 383          * data (or subset), then updating header fields that need adjusting
 384          */
 385         hdr_sz = msg_hdr_sz(msg);
 386         rmsg_sz = hdr_sz + min_t(u32, data_sz, MAX_REJECT_SIZE);
 387
 388         rbuf = tipc_buf_acquire(rmsg_sz);
 389         if (rbuf == NULL)
 390                 goto exit;
 391
 392         rmsg = buf_msg(rbuf);
 393         skb_copy_to_linear_data(rbuf, msg, rmsg_sz);
 394
 395         if (msg_connected(rmsg)) {
 396                 imp = msg_importance(rmsg);
 397                 if (imp < TIPC_CRITICAL_IMPORTANCE)
 398                         msg_set_importance(rmsg, ++imp);
 399         }
 400         msg_set_non_seq(rmsg, 0);
 401         msg_set_size(rmsg, rmsg_sz);
 402         msg_set_errcode(rmsg, err);
 403         msg_set_prevnode(rmsg, tipc_own_addr);
 404         msg_swap_words(rmsg, 4, 5);
 405         if (!msg_short(rmsg))
 406                 msg_swap_words(rmsg, 6, 7);
 407
 408         /* send self-abort message when rejecting on a connected port */
 409         if (msg_connected(msg)) {
 410                 struct tipc_port *p_ptr = tipc_port_lock(msg_destport(msg));
 411
 412                 if (p_ptr) {
 413                         struct sk_buff *abuf = NULL;
 414
 415                         if (p_ptr->connected)
 416                                 abuf = port_build_self_abort_msg(p_ptr, err);
 417                         tipc_port_unlock(p_ptr);
 418                         tipc_net_route_msg(abuf);
 419                 }
 420         }
 421
 422         /* send returned message & dispose of rejected message */
 423         src_node = msg_prevnode(msg);
 424         if (in_own_node(src_node))
 425                 tipc_port_recv_msg(rbuf);
 426         else
 427                 tipc_link_send(rbuf, src_node, msg_link_selector(rmsg));
 428 exit:
 429         kfree_skb(buf);
 430         return data_sz;
 431 }
 432
 433 int tipc_port_reject_sections(struct tipc_port *p_ptr, struct tipc_msg *hdr,
 434                               struct iovec const *msg_sect, unsigned int len,
 435                               int err)
 436 {
 437         struct sk_buff *buf;
 438         int res;
 439
 440         res = tipc_msg_build(hdr, msg_sect, len, MAX_MSG_SIZE, &buf);
 441         if (!buf)
 442                 return res;
 443
 444         return tipc_reject_msg(buf, err);
 445 }
 446
 447 static void port_timeout(unsigned long ref)
 448 {
 449         struct tipc_port *p_ptr = tipc_port_lock(ref);
 450         struct sk_buff *buf = NULL;
 451
 452         if (!p_ptr)
 453                 return;
 454
 455         if (!p_ptr->connected) {
 456                 tipc_port_unlock(p_ptr);
 457                 return;
 458         }
 459
 460         /* Last probe answered ? */
 461         if (p_ptr->probing_state == PROBING) {
 462                 buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
 463         } else {
 464                 buf = port_build_proto_msg(p_ptr, CONN_PROBE, 0);
 465                 p_ptr->probing_state = PROBING;
 466                 k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
 467         }
 468         tipc_port_unlock(p_ptr);
 469         tipc_net_route_msg(buf);
 470 }
 471
 472
 473 static void port_handle_node_down(unsigned long ref)
 474 {
 475         struct tipc_port *p_ptr = tipc_port_lock(ref);
 476         struct sk_buff *buf = NULL;
 477
 478         if (!p_ptr)
 479                 return;
 480         buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_NODE);
 481         tipc_port_unlock(p_ptr);
 482         tipc_net_route_msg(buf);
 483 }
 484
 485
 486 static struct sk_buff *port_build_self_abort_msg(struct tipc_port *p_ptr, u32 err)
 487 {
 488         struct sk_buff *buf = port_build_peer_abort_msg(p_ptr, err);
 489
 490         if (buf) {
 491                 struct tipc_msg *msg = buf_msg(buf);
 492                 msg_swap_words(msg, 4, 5);
 493                 msg_swap_words(msg, 6, 7);
 494         }
 495         return buf;
 496 }
 497
 498
 499 static struct sk_buff *port_build_peer_abort_msg(struct tipc_port *p_ptr, u32 err)
 500 {
 501         struct sk_buff *buf;
 502         struct tipc_msg *msg;
 503         u32 imp;
 504
 505         if (!p_ptr->connected)
 506                 return NULL;
 507
 508         buf = tipc_buf_acquire(BASIC_H_SIZE);
 509         if (buf) {
 510                 msg = buf_msg(buf);
 511                 memcpy(msg, &p_ptr->phdr, BASIC_H_SIZE);
 512                 msg_set_hdr_sz(msg, BASIC_H_SIZE);
 513                 msg_set_size(msg, BASIC_H_SIZE);
 514                 imp = msg_importance(msg);
 515                 if (imp < TIPC_CRITICAL_IMPORTANCE)
 516                         msg_set_importance(msg, ++imp);
 517                 msg_set_errcode(msg, err);
 518         }
 519         return buf;
 520 }
 521
 522 void tipc_port_recv_proto_msg(struct sk_buff *buf)
 523 {
 524         struct tipc_msg *msg = buf_msg(buf);
 525         struct tipc_port *p_ptr;
 526         struct sk_buff *r_buf = NULL;
 527         u32 destport = msg_destport(msg);
 528         int wakeable;
 529
 530         /* Validate connection */
 531         p_ptr = tipc_port_lock(destport);
 532         if (!p_ptr || !p_ptr->connected || !tipc_port_peer_msg(p_ptr, msg)) {
 533                 r_buf = tipc_buf_acquire(BASIC_H_SIZE);
 534                 if (r_buf) {
 535                         msg = buf_msg(r_buf);
 536                         tipc_msg_init(msg, TIPC_HIGH_IMPORTANCE, TIPC_CONN_MSG,
 537                                       BASIC_H_SIZE, msg_orignode(msg));
 538                         msg_set_errcode(msg, TIPC_ERR_NO_PORT);
 539                         msg_set_origport(msg, destport);
 540                         msg_set_destport(msg, msg_origport(msg));
 541                 }
 542                 if (p_ptr)
 543                         tipc_port_unlock(p_ptr);
 544                 goto exit;
 545         }
 546
 547         /* Process protocol message sent by peer */
 548         switch (msg_type(msg)) {
 549         case CONN_ACK:
 550                 wakeable = tipc_port_congested(p_ptr) && p_ptr->congested &&
 551                         p_ptr->wakeup;
 552                 p_ptr->acked += msg_msgcnt(msg);
 553                 if (!tipc_port_congested(p_ptr)) {
 554                         p_ptr->congested = 0;
 555                         if (wakeable)
 556                                 p_ptr->wakeup(p_ptr);
 557                 }
 558                 break;
 559         case CONN_PROBE:
 560                 r_buf = port_build_proto_msg(p_ptr, CONN_PROBE_REPLY, 0);
 561                 break;
 562         default:
 563                 /* CONN_PROBE_REPLY or unrecognized - no action required */
 564                 break;
 565         }
 566         p_ptr->probing_state = CONFIRMED;
 567         tipc_port_unlock(p_ptr);
 568 exit:
 569         tipc_net_route_msg(r_buf);
 570         kfree_skb(buf);
 571 }
 572
 573 static int port_print(struct tipc_port *p_ptr, char *buf, int len, int full_id)
 574 {
 575         struct publication *publ;
 576         int ret;
 577
 578         if (full_id)
 579                 ret = tipc_snprintf(buf, len, "<%u.%u.%u:%u>:",
 580                                     tipc_zone(tipc_own_addr),
 581                                     tipc_cluster(tipc_own_addr),
 582                                     tipc_node(tipc_own_addr), p_ptr->ref);
 583         else
 584                 ret = tipc_snprintf(buf, len, "%-10u:", p_ptr->ref);
 585
 586         if (p_ptr->connected) {
 587                 u32 dport = port_peerport(p_ptr);
 588                 u32 destnode = port_peernode(p_ptr);
 589
 590                 ret += tipc_snprintf(buf + ret, len - ret,
 591                                      " connected to <%u.%u.%u:%u>",
 592                                      tipc_zone(destnode),
 593                                      tipc_cluster(destnode),
 594                                      tipc_node(destnode), dport);
 595                 if (p_ptr->conn_type != 0)
 596                         ret += tipc_snprintf(buf + ret, len - ret,
 597                                              " via {%u,%u}", p_ptr->conn_type,
 598                                              p_ptr->conn_instance);
 599         } else if (p_ptr->published) {
 600                 ret += tipc_snprintf(buf + ret, len - ret, " bound to");
 601                 list_for_each_entry(publ, &p_ptr->publications, pport_list) {
 602                         if (publ->lower == publ->upper)
 603                                 ret += tipc_snprintf(buf + ret, len - ret,
 604                                                      " {%u,%u}", publ->type,
 605                                                      publ->lower);
 606                         else
 607                                 ret += tipc_snprintf(buf + ret, len - ret,
 608                                                      " {%u,%u,%u}", publ->type,
 609                                                      publ->lower, publ->upper);
 610                 }
 611         }
 612         ret += tipc_snprintf(buf + ret, len - ret, "\n");
 613         return ret;
 614 }
 615
 616 struct sk_buff *tipc_port_get_ports(void)
 617 {
 618         struct sk_buff *buf;
 619         struct tlv_desc *rep_tlv;
 620         char *pb;
 621         int pb_len;
 622         struct tipc_port *p_ptr;
 623         int str_len = 0;
 624
 625         buf = tipc_cfg_reply_alloc(TLV_SPACE(ULTRA_STRING_MAX_LEN));
 626         if (!buf)
 627                 return NULL;
 628         rep_tlv = (struct tlv_desc *)buf->data;
 629         pb = TLV_DATA(rep_tlv);
 630         pb_len = ULTRA_STRING_MAX_LEN;
 631
 632         spin_lock_bh(&tipc_port_list_lock);
 633         list_for_each_entry(p_ptr, &ports, port_list) {
 634                 spin_lock_bh(p_ptr->lock);
 635                 str_len += port_print(p_ptr, pb, pb_len, 0);
 636                 spin_unlock_bh(p_ptr->lock);
 637         }
 638         spin_unlock_bh(&tipc_port_list_lock);
 639         str_len += 1;   /* for "\0" */
 640         skb_put(buf, TLV_SPACE(str_len));
 641         TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
 642
 643         return buf;
 644 }
 645
 646 void tipc_port_reinit(void)
 647 {
 648         struct tipc_port *p_ptr;
 649         struct tipc_msg *msg;
 650
 651         spin_lock_bh(&tipc_port_list_lock);
 652         list_for_each_entry(p_ptr, &ports, port_list) {
 653                 msg = &p_ptr->phdr;
 654                 msg_set_prevnode(msg, tipc_own_addr);
 655                 msg_set_orignode(msg, tipc_own_addr);
 656         }
 657         spin_unlock_bh(&tipc_port_list_lock);
 658 }
 659
 660 void tipc_acknowledge(u32 ref, u32 ack)
 661 {
 662         struct tipc_port *p_ptr;
 663         struct sk_buff *buf = NULL;
 664
 665         p_ptr = tipc_port_lock(ref);
 666         if (!p_ptr)
 667                 return;
 668         if (p_ptr->connected) {
 669                 p_ptr->conn_unacked -= ack;
 670                 buf = port_build_proto_msg(p_ptr, CONN_ACK, ack);
 671         }
 672         tipc_port_unlock(p_ptr);
 673         tipc_net_route_msg(buf);
 674 }
 675
 676 int tipc_portimportance(u32 ref, unsigned int *importance)
 677 {
 678         struct tipc_port *p_ptr;
 679
 680         p_ptr = tipc_port_lock(ref);
 681         if (!p_ptr)
 682                 return -EINVAL;
 683         *importance = (unsigned int)msg_importance(&p_ptr->phdr);
 684         tipc_port_unlock(p_ptr);
 685         return 0;
 686 }
 687
 688 int tipc_set_portimportance(u32 ref, unsigned int imp)
 689 {
 690         struct tipc_port *p_ptr;
 691
 692         if (imp > TIPC_CRITICAL_IMPORTANCE)
 693                 return -EINVAL;
 694
 695         p_ptr = tipc_port_lock(ref);
 696         if (!p_ptr)
 697                 return -EINVAL;
 698         msg_set_importance(&p_ptr->phdr, (u32)imp);
 699         tipc_port_unlock(p_ptr);
 700         return 0;
 701 }
 702
 703
 704 int tipc_publish(struct tipc_port *p_ptr, unsigned int scope,
 705                  struct tipc_name_seq const *seq)
 706 {
 707         struct publication *publ;
 708         u32 key;
 709
 710         if (p_ptr->connected)
 711                 return -EINVAL;
 712         key = p_ptr->ref + p_ptr->pub_count + 1;
 713         if (key == p_ptr->ref)
 714                 return -EADDRINUSE;
 715
 716         publ = tipc_nametbl_publish(seq->type, seq->lower, seq->upper,
 717                                     scope, p_ptr->ref, key);
 718         if (publ) {
 719                 list_add(&publ->pport_list, &p_ptr->publications);
 720                 p_ptr->pub_count++;
 721                 p_ptr->published = 1;
 722                 return 0;
 723         }
 724         return -EINVAL;
 725 }
 726
 727 int tipc_withdraw(struct tipc_port *p_ptr, unsigned int scope,
 728                   struct tipc_name_seq const *seq)
 729 {
 730         struct publication *publ;
 731         struct publication *tpubl;
 732         int res = -EINVAL;
 733
 734         if (!seq) {
 735                 list_for_each_entry_safe(publ, tpubl,
 736                                          &p_ptr->publications, pport_list) {
 737                         tipc_nametbl_withdraw(publ->type, publ->lower,
 738                                               publ->ref, publ->key);
 739                 }
 740                 res = 0;
 741         } else {
 742                 list_for_each_entry_safe(publ, tpubl,
 743                                          &p_ptr->publications, pport_list) {
 744                         if (publ->scope != scope)
 745                                 continue;
 746                         if (publ->type != seq->type)
 747                                 continue;
 748                         if (publ->lower != seq->lower)
 749                                 continue;
 750                         if (publ->upper != seq->upper)
 751                                 break;
 752                         tipc_nametbl_withdraw(publ->type, publ->lower,
 753                                               publ->ref, publ->key);
 754                         res = 0;
 755                         break;
 756                 }
 757         }
 758         if (list_empty(&p_ptr->publications))
 759                 p_ptr->published = 0;
 760         return res;
 761 }
 762
 763 int tipc_connect(u32 ref, struct tipc_portid const *peer)
 764 {
 765         struct tipc_port *p_ptr;
 766         int res;
 767
 768         p_ptr = tipc_port_lock(ref);
 769         if (!p_ptr)
 770                 return -EINVAL;
 771         res = __tipc_connect(ref, p_ptr, peer);
 772         tipc_port_unlock(p_ptr);
 773         return res;
 774 }
 775
 776 /*
 777  * __tipc_connect - connect to a remote peer
 778  *
 779  * Port must be locked.
 780  */
 781 int __tipc_connect(u32 ref, struct tipc_port *p_ptr,
 782                         struct tipc_portid const *peer)
 783 {
 784         struct tipc_msg *msg;
 785         int res = -EINVAL;
 786
 787         if (p_ptr->published || p_ptr->connected)
 788                 goto exit;
 789         if (!peer->ref)
 790                 goto exit;
 791
 792         msg = &p_ptr->phdr;
 793         msg_set_destnode(msg, peer->node);
 794         msg_set_destport(msg, peer->ref);
 795         msg_set_type(msg, TIPC_CONN_MSG);
 796         msg_set_lookup_scope(msg, 0);
 797         msg_set_hdr_sz(msg, SHORT_H_SIZE);
 798
 799         p_ptr->probing_interval = PROBING_INTERVAL;
 800         p_ptr->probing_state = CONFIRMED;
 801         p_ptr->connected = 1;
 802         k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
 803
 804         tipc_nodesub_subscribe(&p_ptr->subscription, peer->node,
 805                           (void *)(unsigned long)ref,
 806                           (net_ev_handler)port_handle_node_down);
 807         res = 0;
 808 exit:
 809         p_ptr->max_pkt = tipc_link_get_max_pkt(peer->node, ref);
 810         return res;
 811 }
 812
 813 /*
 814  * __tipc_disconnect - disconnect port from peer
 815  *
 816  * Port must be locked.
 817  */
 818 int __tipc_disconnect(struct tipc_port *tp_ptr)
 819 {
 820         if (tp_ptr->connected) {
 821                 tp_ptr->connected = 0;
 822                 /* let timer expire on it's own to avoid deadlock! */
 823                 tipc_nodesub_unsubscribe(&tp_ptr->subscription);
 824                 return 0;
 825         }
 826
 827         return -ENOTCONN;
 828 }
 829
 830 /*
 831  * tipc_disconnect(): Disconnect port form peer.
 832  *                    This is a node local operation.
 833  */
 834 int tipc_disconnect(u32 ref)
 835 {
 836         struct tipc_port *p_ptr;
 837         int res;
 838
 839         p_ptr = tipc_port_lock(ref);
 840         if (!p_ptr)
 841                 return -EINVAL;
 842         res = __tipc_disconnect(p_ptr);
 843         tipc_port_unlock(p_ptr);
 844         return res;
 845 }
 846
 847 /*
 848  * tipc_shutdown(): Send a SHUTDOWN msg to peer and disconnect
 849  */
 850 int tipc_shutdown(u32 ref)
 851 {
 852         struct tipc_port *p_ptr;
 853         struct sk_buff *buf = NULL;
 854
 855         p_ptr = tipc_port_lock(ref);
 856         if (!p_ptr)
 857                 return -EINVAL;
 858
 859         buf = port_build_peer_abort_msg(p_ptr, TIPC_CONN_SHUTDOWN);
 860         tipc_port_unlock(p_ptr);
 861         tipc_net_route_msg(buf);
 862         return tipc_disconnect(ref);
 863 }
 864
 865 /**
 866  * tipc_port_recv_msg - receive message from lower layer and deliver to port user
 867  */
 868 int tipc_port_recv_msg(struct sk_buff *buf)
 869 {
 870         struct tipc_port *p_ptr;
 871         struct tipc_msg *msg = buf_msg(buf);
 872         u32 destport = msg_destport(msg);
 873         u32 dsz = msg_data_sz(msg);
 874         u32 err;
 875
 876         /* forward unresolved named message */
 877         if (unlikely(!destport)) {
 878                 tipc_net_route_msg(buf);
 879                 return dsz;
 880         }
 881
 882         /* validate destination & pass to port, otherwise reject message */
 883         p_ptr = tipc_port_lock(destport);
 884         if (likely(p_ptr)) {
 885                 err = p_ptr->dispatcher(p_ptr, buf);
 886                 tipc_port_unlock(p_ptr);
 887                 if (likely(!err))
 888                         return dsz;
 889         } else {
 890                 err = TIPC_ERR_NO_PORT;
 891         }
 892
 893         return tipc_reject_msg(buf, err);
 894 }
 895
 896 /*
 897  *  tipc_port_recv_sections(): Concatenate and deliver sectioned
 898  *                        message for this node.
 899  */
 900 static int tipc_port_recv_sections(struct tipc_port *sender,
 901                                    struct iovec const *msg_sect,
 902                                    unsigned int len)
 903 {
 904         struct sk_buff *buf;
 905         int res;
 906
 907         res = tipc_msg_build(&sender->phdr, msg_sect, len, MAX_MSG_SIZE, &buf);
 908         if (likely(buf))
 909                 tipc_port_recv_msg(buf);
 910         return res;
 911 }
 912
 913 /**
 914  * tipc_send - send message sections on connection
 915  */
 916 int tipc_send(u32 ref, struct iovec const *msg_sect, unsigned int len)
 917 {
 918         struct tipc_port *p_ptr;
 919         u32 destnode;
 920         int res;
 921
 922         p_ptr = tipc_port_deref(ref);
 923         if (!p_ptr || !p_ptr->connected)
 924                 return -EINVAL;
 925
 926         p_ptr->congested = 1;
 927         if (!tipc_port_congested(p_ptr)) {
 928                 destnode = port_peernode(p_ptr);
 929                 if (likely(!in_own_node(destnode)))
 930                         res = tipc_link_send_sections_fast(p_ptr, msg_sect,
 931                                                            len, destnode);
 932                 else
 933                         res = tipc_port_recv_sections(p_ptr, msg_sect, len);
 934
 935                 if (likely(res != -ELINKCONG)) {
 936                         p_ptr->congested = 0;
 937                         if (res > 0)
 938                                 p_ptr->sent++;
 939                         return res;
 940                 }
 941         }
 942         if (port_unreliable(p_ptr)) {
 943                 p_ptr->congested = 0;
 944                 return len;
 945         }
 946         return -ELINKCONG;
 947 }
 948
 949 /**
 950  * tipc_send2name - send message sections to port name
 951  */
 952 int tipc_send2name(u32 ref, struct tipc_name const *name, unsigned int domain,
 953                    struct iovec const *msg_sect, unsigned int len)
 954 {
 955         struct tipc_port *p_ptr;
 956         struct tipc_msg *msg;
 957         u32 destnode = domain;
 958         u32 destport;
 959         int res;
 960
 961         p_ptr = tipc_port_deref(ref);
 962         if (!p_ptr || p_ptr->connected)
 963                 return -EINVAL;
 964
 965         msg = &p_ptr->phdr;
 966         msg_set_type(msg, TIPC_NAMED_MSG);
 967         msg_set_hdr_sz(msg, NAMED_H_SIZE);
 968         msg_set_nametype(msg, name->type);
 969         msg_set_nameinst(msg, name->instance);
 970         msg_set_lookup_scope(msg, tipc_addr_scope(domain));
 971         destport = tipc_nametbl_translate(name->type, name->instance, &destnode);
 972         msg_set_destnode(msg, destnode);
 973         msg_set_destport(msg, destport);
 974
 975         if (likely(destport || destnode)) {
 976                 if (likely(in_own_node(destnode)))
 977                         res = tipc_port_recv_sections(p_ptr, msg_sect, len);
 978                 else if (tipc_own_addr)
 979                         res = tipc_link_send_sections_fast(p_ptr, msg_sect,
 980                                                            len, destnode);
 981                 else
 982                         res = tipc_port_reject_sections(p_ptr, msg, msg_sect,
 983                                                         len, TIPC_ERR_NO_NODE);
 984                 if (likely(res != -ELINKCONG)) {
 985                         if (res > 0)
 986                                 p_ptr->sent++;
 987                         return res;
 988                 }
 989                 if (port_unreliable(p_ptr)) {
 990                         return len;
 991                 }
 992                 return -ELINKCONG;
 993         }
 994         return tipc_port_reject_sections(p_ptr, msg, msg_sect, len,
 995                                          TIPC_ERR_NO_NAME);
 996 }
 997
 998 /**
 999  * tipc_send2port - send message sections to port identity
1000  */
1001 int tipc_send2port(u32 ref, struct tipc_portid const *dest,
1002                    struct iovec const *msg_sect, unsigned int len)
1003 {
1004         struct tipc_port *p_ptr;
1005         struct tipc_msg *msg;
1006         int res;
1007
1008         p_ptr = tipc_port_deref(ref);
1009         if (!p_ptr || p_ptr->connected)
1010                 return -EINVAL;
1011
1012         msg = &p_ptr->phdr;
1013         msg_set_type(msg, TIPC_DIRECT_MSG);
1014         msg_set_lookup_scope(msg, 0);
1015         msg_set_destnode(msg, dest->node);
1016         msg_set_destport(msg, dest->ref);
1017         msg_set_hdr_sz(msg, BASIC_H_SIZE);
1018
1019         if (in_own_node(dest->node))
1020                 res =  tipc_port_recv_sections(p_ptr, msg_sect, len);
1021         else if (tipc_own_addr)
1022                 res = tipc_link_send_sections_fast(p_ptr, msg_sect, len,
1023                                                    dest->node);
1024         else
1025                 res = tipc_port_reject_sections(p_ptr, msg, msg_sect, len,
1026                                                 TIPC_ERR_NO_NODE);
1027         if (likely(res != -ELINKCONG)) {
1028                 if (res > 0)
1029                         p_ptr->sent++;
1030                 return res;
1031         }
1032         if (port_unreliable(p_ptr)) {
1033                 return len;
1034         }
1035         return -ELINKCONG;
1036 }