Pileus Git - ~andy/linux/blob - 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-2007, 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 "dbg.h"
  40 #include "port.h"
  41 #include "addr.h"
  42 #include "link.h"
  43 #include "node.h"
  44 #include "name_table.h"
  45 #include "user_reg.h"
  46 #include "msg.h"
  47 #include "bcast.h"
  48
  49 /* Connection management: */
  50 #define PROBING_INTERVAL 3600000        /* [ms] => 1 h */
  51 #define CONFIRMED 0
  52 #define PROBING 1
  53
  54 #define MAX_REJECT_SIZE 1024
  55
  56 static struct sk_buff *msg_queue_head = NULL;
  57 static struct sk_buff *msg_queue_tail = NULL;
  58
  59 DEFINE_SPINLOCK(tipc_port_list_lock);
  60 static DEFINE_SPINLOCK(queue_lock);
  61
  62 static LIST_HEAD(ports);
  63 static void port_handle_node_down(unsigned long ref);
  64 static struct sk_buff* port_build_self_abort_msg(struct port *,u32 err);
  65 static struct sk_buff* port_build_peer_abort_msg(struct port *,u32 err);
  66 static void port_timeout(unsigned long ref);
  67
  68
  69 static u32 port_peernode(struct port *p_ptr)
  70 {
  71         return msg_destnode(&p_ptr->publ.phdr);
  72 }
  73
  74 static u32 port_peerport(struct port *p_ptr)
  75 {
  76         return msg_destport(&p_ptr->publ.phdr);
  77 }
  78
  79 static u32 port_out_seqno(struct port *p_ptr)
  80 {
  81         return msg_transp_seqno(&p_ptr->publ.phdr);
  82 }
  83
  84 static void port_incr_out_seqno(struct port *p_ptr)
  85 {
  86         struct tipc_msg *m = &p_ptr->publ.phdr;
  87
  88         if (likely(!msg_routed(m)))
  89                 return;
  90         msg_set_transp_seqno(m, (msg_transp_seqno(m) + 1));
  91 }
  92
  93 /**
  94  * tipc_multicast - send a multicast message to local and remote destinations
  95  */
  96
  97 int tipc_multicast(u32 ref, struct tipc_name_seq const *seq, u32 domain,
  98                    u32 num_sect, struct iovec const *msg_sect)
  99 {
 100         struct tipc_msg *hdr;
 101         struct sk_buff *buf;
 102         struct sk_buff *ibuf = NULL;
 103         struct port_list dports = {0, NULL, };
 104         struct port *oport = tipc_port_deref(ref);
 105         int ext_targets;
 106         int res;
 107
 108         if (unlikely(!oport))
 109                 return -EINVAL;
 110
 111         /* Create multicast message */
 112
 113         hdr = &oport->publ.phdr;
 114         msg_set_type(hdr, TIPC_MCAST_MSG);
 115         msg_set_nametype(hdr, seq->type);
 116         msg_set_namelower(hdr, seq->lower);
 117         msg_set_nameupper(hdr, seq->upper);
 118         msg_set_hdr_sz(hdr, MCAST_H_SIZE);
 119         res = msg_build(hdr, msg_sect, num_sect, MAX_MSG_SIZE,
 120                         !oport->user_port, &buf);
 121         if (unlikely(!buf))
 122                 return res;
 123
 124         /* Figure out where to send multicast message */
 125
 126         ext_targets = tipc_nametbl_mc_translate(seq->type, seq->lower, seq->upper,
 127                                                 TIPC_NODE_SCOPE, &dports);
 128
 129         /* Send message to destinations (duplicate it only if necessary) */
 130
 131         if (ext_targets) {
 132                 if (dports.count != 0) {
 133                         ibuf = skb_copy(buf, GFP_ATOMIC);
 134                         if (ibuf == NULL) {
 135                                 tipc_port_list_free(&dports);
 136                                 buf_discard(buf);
 137                                 return -ENOMEM;
 138                         }
 139                 }
 140                 res = tipc_bclink_send_msg(buf);
 141                 if ((res < 0) && (dports.count != 0)) {
 142                         buf_discard(ibuf);
 143                 }
 144         } else {
 145                 ibuf = buf;
 146         }
 147
 148         if (res >= 0) {
 149                 if (ibuf)
 150                         tipc_port_recv_mcast(ibuf, &dports);
 151         } else {
 152                 tipc_port_list_free(&dports);
 153         }
 154         return res;
 155 }
 156
 157 /**
 158  * tipc_port_recv_mcast - deliver multicast message to all destination ports
 159  *
 160  * If there is no port list, perform a lookup to create one
 161  */
 162
 163 void tipc_port_recv_mcast(struct sk_buff *buf, struct port_list *dp)
 164 {
 165         struct tipc_msg* msg;
 166         struct port_list dports = {0, NULL, };
 167         struct port_list *item = dp;
 168         int cnt = 0;
 169
 170         msg = buf_msg(buf);
 171
 172         /* Create destination port list, if one wasn't supplied */
 173
 174         if (dp == NULL) {
 175                 tipc_nametbl_mc_translate(msg_nametype(msg),
 176                                      msg_namelower(msg),
 177                                      msg_nameupper(msg),
 178                                      TIPC_CLUSTER_SCOPE,
 179                                      &dports);
 180                 item = dp = &dports;
 181         }
 182
 183         /* Deliver a copy of message to each destination port */
 184
 185         if (dp->count != 0) {
 186                 if (dp->count == 1) {
 187                         msg_set_destport(msg, dp->ports[0]);
 188                         tipc_port_recv_msg(buf);
 189                         tipc_port_list_free(dp);
 190                         return;
 191                 }
 192                 for (; cnt < dp->count; cnt++) {
 193                         int index = cnt % PLSIZE;
 194                         struct sk_buff *b = skb_clone(buf, GFP_ATOMIC);
 195
 196                         if (b == NULL) {
 197                                 warn("Unable to deliver multicast message(s)\n");
 198                                 msg_dbg(msg, "LOST:");
 199                                 goto exit;
 200                         }
 201                         if ((index == 0) && (cnt != 0)) {
 202                                 item = item->next;
 203                         }
 204                         msg_set_destport(buf_msg(b),item->ports[index]);
 205                         tipc_port_recv_msg(b);
 206                 }
 207         }
 208 exit:
 209         buf_discard(buf);
 210         tipc_port_list_free(dp);
 211 }
 212
 213 /**
 214  * tipc_createport_raw - create a generic TIPC port
 215  *
 216  * Returns port reference, or 0 if unable to create it
 217  *
 218  * Note: The newly created port is returned in the locked state.
 219  */
 220
 221 u32 tipc_createport_raw(void *usr_handle,
 222                         u32 (*dispatcher)(struct tipc_port *, struct sk_buff *),
 223                         void (*wakeup)(struct tipc_port *),
 224                         const u32 importance,
 225                         struct tipc_port **tp_ptr)
 226 {
 227         struct port *p_ptr;
 228         struct tipc_msg *msg;
 229         u32 ref;
 230
 231         p_ptr = kzalloc(sizeof(*p_ptr), GFP_ATOMIC);
 232         if (!p_ptr) {
 233                 warn("Port creation failed, no memory\n");
 234                 return 0;
 235         }
 236         ref = tipc_ref_acquire(p_ptr, &p_ptr->publ.lock);
 237         if (!ref) {
 238                 warn("Port creation failed, reference table exhausted\n");
 239                 kfree(p_ptr);
 240                 return 0;
 241         }
 242
 243         p_ptr->publ.usr_handle = usr_handle;
 244         p_ptr->publ.max_pkt = MAX_PKT_DEFAULT;
 245         p_ptr->publ.ref = ref;
 246         msg = &p_ptr->publ.phdr;
 247         msg_init(msg, importance, TIPC_NAMED_MSG, LONG_H_SIZE, 0);
 248         msg_set_origport(msg, ref);
 249         p_ptr->last_in_seqno = 41;
 250         p_ptr->sent = 1;
 251         INIT_LIST_HEAD(&p_ptr->wait_list);
 252         INIT_LIST_HEAD(&p_ptr->subscription.nodesub_list);
 253         p_ptr->congested_link = NULL;
 254         p_ptr->dispatcher = dispatcher;
 255         p_ptr->wakeup = wakeup;
 256         p_ptr->user_port = NULL;
 257         k_init_timer(&p_ptr->timer, (Handler)port_timeout, ref);
 258         spin_lock_bh(&tipc_port_list_lock);
 259         INIT_LIST_HEAD(&p_ptr->publications);
 260         INIT_LIST_HEAD(&p_ptr->port_list);
 261         list_add_tail(&p_ptr->port_list, &ports);
 262         spin_unlock_bh(&tipc_port_list_lock);
 263         *tp_ptr = &p_ptr->publ;
 264         return ref;
 265 }
 266
 267 int tipc_deleteport(u32 ref)
 268 {
 269         struct port *p_ptr;
 270         struct sk_buff *buf = NULL;
 271
 272         tipc_withdraw(ref, 0, NULL);
 273         p_ptr = tipc_port_lock(ref);
 274         if (!p_ptr)
 275                 return -EINVAL;
 276
 277         tipc_ref_discard(ref);
 278         tipc_port_unlock(p_ptr);
 279
 280         k_cancel_timer(&p_ptr->timer);
 281         if (p_ptr->publ.connected) {
 282                 buf = port_build_peer_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
 283                 tipc_nodesub_unsubscribe(&p_ptr->subscription);
 284         }
 285         if (p_ptr->user_port) {
 286                 tipc_reg_remove_port(p_ptr->user_port);
 287                 kfree(p_ptr->user_port);
 288         }
 289
 290         spin_lock_bh(&tipc_port_list_lock);
 291         list_del(&p_ptr->port_list);
 292         list_del(&p_ptr->wait_list);
 293         spin_unlock_bh(&tipc_port_list_lock);
 294         k_term_timer(&p_ptr->timer);
 295         kfree(p_ptr);
 296         dbg("Deleted port %u\n", ref);
 297         tipc_net_route_msg(buf);
 298         return TIPC_OK;
 299 }
 300
 301 /**
 302  * tipc_get_port() - return port associated with 'ref'
 303  *
 304  * Note: Port is not locked.
 305  */
 306
 307 struct tipc_port *tipc_get_port(const u32 ref)
 308 {
 309         return (struct tipc_port *)tipc_ref_deref(ref);
 310 }
 311
 312 /**
 313  * tipc_get_handle - return user handle associated to port 'ref'
 314  */
 315
 316 void *tipc_get_handle(const u32 ref)
 317 {
 318         struct port *p_ptr;
 319         void * handle;
 320
 321         p_ptr = tipc_port_lock(ref);
 322         if (!p_ptr)
 323                 return NULL;
 324         handle = p_ptr->publ.usr_handle;
 325         tipc_port_unlock(p_ptr);
 326         return handle;
 327 }
 328
 329 static int port_unreliable(struct port *p_ptr)
 330 {
 331         return msg_src_droppable(&p_ptr->publ.phdr);
 332 }
 333
 334 int tipc_portunreliable(u32 ref, unsigned int *isunreliable)
 335 {
 336         struct port *p_ptr;
 337
 338         p_ptr = tipc_port_lock(ref);
 339         if (!p_ptr)
 340                 return -EINVAL;
 341         *isunreliable = port_unreliable(p_ptr);
 342         tipc_port_unlock(p_ptr);
 343         return TIPC_OK;
 344 }
 345
 346 int tipc_set_portunreliable(u32 ref, unsigned int isunreliable)
 347 {
 348         struct port *p_ptr;
 349
 350         p_ptr = tipc_port_lock(ref);
 351         if (!p_ptr)
 352                 return -EINVAL;
 353         msg_set_src_droppable(&p_ptr->publ.phdr, (isunreliable != 0));
 354         tipc_port_unlock(p_ptr);
 355         return TIPC_OK;
 356 }
 357
 358 static int port_unreturnable(struct port *p_ptr)
 359 {
 360         return msg_dest_droppable(&p_ptr->publ.phdr);
 361 }
 362
 363 int tipc_portunreturnable(u32 ref, unsigned int *isunrejectable)
 364 {
 365         struct port *p_ptr;
 366
 367         p_ptr = tipc_port_lock(ref);
 368         if (!p_ptr)
 369                 return -EINVAL;
 370         *isunrejectable = port_unreturnable(p_ptr);
 371         tipc_port_unlock(p_ptr);
 372         return TIPC_OK;
 373 }
 374
 375 int tipc_set_portunreturnable(u32 ref, unsigned int isunrejectable)
 376 {
 377         struct port *p_ptr;
 378
 379         p_ptr = tipc_port_lock(ref);
 380         if (!p_ptr)
 381                 return -EINVAL;
 382         msg_set_dest_droppable(&p_ptr->publ.phdr, (isunrejectable != 0));
 383         tipc_port_unlock(p_ptr);
 384         return TIPC_OK;
 385 }
 386
 387 /*
 388  * port_build_proto_msg(): build a port level protocol
 389  * or a connection abortion message. Called with
 390  * tipc_port lock on.
 391  */
 392 static struct sk_buff *port_build_proto_msg(u32 destport, u32 destnode,
 393                                             u32 origport, u32 orignode,
 394                                             u32 usr, u32 type, u32 err,
 395                                             u32 seqno, u32 ack)
 396 {
 397         struct sk_buff *buf;
 398         struct tipc_msg *msg;
 399
 400         buf = buf_acquire(LONG_H_SIZE);
 401         if (buf) {
 402                 msg = buf_msg(buf);
 403                 msg_init(msg, usr, type, LONG_H_SIZE, destnode);
 404                 msg_set_errcode(msg, err);
 405                 msg_set_destport(msg, destport);
 406                 msg_set_origport(msg, origport);
 407                 msg_set_orignode(msg, orignode);
 408                 msg_set_transp_seqno(msg, seqno);
 409                 msg_set_msgcnt(msg, ack);
 410                 msg_dbg(msg, "PORT>SEND>:");
 411         }
 412         return buf;
 413 }
 414
 415 int tipc_reject_msg(struct sk_buff *buf, u32 err)
 416 {
 417         struct tipc_msg *msg = buf_msg(buf);
 418         struct sk_buff *rbuf;
 419         struct tipc_msg *rmsg;
 420         int hdr_sz;
 421         u32 imp = msg_importance(msg);
 422         u32 data_sz = msg_data_sz(msg);
 423
 424         if (data_sz > MAX_REJECT_SIZE)
 425                 data_sz = MAX_REJECT_SIZE;
 426         if (msg_connected(msg) && (imp < TIPC_CRITICAL_IMPORTANCE))
 427                 imp++;
 428         msg_dbg(msg, "port->rej: ");
 429
 430         /* discard rejected message if it shouldn't be returned to sender */
 431         if (msg_errcode(msg) || msg_dest_droppable(msg)) {
 432                 buf_discard(buf);
 433                 return data_sz;
 434         }
 435
 436         /* construct rejected message */
 437         if (msg_mcast(msg))
 438                 hdr_sz = MCAST_H_SIZE;
 439         else
 440                 hdr_sz = LONG_H_SIZE;
 441         rbuf = buf_acquire(data_sz + hdr_sz);
 442         if (rbuf == NULL) {
 443                 buf_discard(buf);
 444                 return data_sz;
 445         }
 446         rmsg = buf_msg(rbuf);
 447         msg_init(rmsg, imp, msg_type(msg), hdr_sz, msg_orignode(msg));
 448         msg_set_errcode(rmsg, err);
 449         msg_set_destport(rmsg, msg_origport(msg));
 450         msg_set_origport(rmsg, msg_destport(msg));
 451         if (msg_short(msg))
 452                 msg_set_orignode(rmsg, tipc_own_addr);
 453         else
 454                 msg_set_orignode(rmsg, msg_destnode(msg));
 455         msg_set_size(rmsg, data_sz + hdr_sz);
 456         msg_set_nametype(rmsg, msg_nametype(msg));
 457         msg_set_nameinst(rmsg, msg_nameinst(msg));
 458         skb_copy_to_linear_data_offset(rbuf, hdr_sz, msg_data(msg), data_sz);
 459
 460         /* send self-abort message when rejecting on a connected port */
 461         if (msg_connected(msg)) {
 462                 struct sk_buff *abuf = NULL;
 463                 struct port *p_ptr = tipc_port_lock(msg_destport(msg));
 464
 465                 if (p_ptr) {
 466                         if (p_ptr->publ.connected)
 467                                 abuf = port_build_self_abort_msg(p_ptr, err);
 468                         tipc_port_unlock(p_ptr);
 469                 }
 470                 tipc_net_route_msg(abuf);
 471         }
 472
 473         /* send rejected message */
 474         buf_discard(buf);
 475         tipc_net_route_msg(rbuf);
 476         return data_sz;
 477 }
 478
 479 int tipc_port_reject_sections(struct port *p_ptr, struct tipc_msg *hdr,
 480                               struct iovec const *msg_sect, u32 num_sect,
 481                               int err)
 482 {
 483         struct sk_buff *buf;
 484         int res;
 485
 486         res = msg_build(hdr, msg_sect, num_sect, MAX_MSG_SIZE,
 487                         !p_ptr->user_port, &buf);
 488         if (!buf)
 489                 return res;
 490
 491         return tipc_reject_msg(buf, err);
 492 }
 493
 494 static void port_timeout(unsigned long ref)
 495 {
 496         struct port *p_ptr = tipc_port_lock(ref);
 497         struct sk_buff *buf = NULL;
 498
 499         if (!p_ptr)
 500                 return;
 501
 502         if (!p_ptr->publ.connected) {
 503                 tipc_port_unlock(p_ptr);
 504                 return;
 505         }
 506
 507         /* Last probe answered ? */
 508         if (p_ptr->probing_state == PROBING) {
 509                 buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_PORT);
 510         } else {
 511                 buf = port_build_proto_msg(port_peerport(p_ptr),
 512                                            port_peernode(p_ptr),
 513                                            p_ptr->publ.ref,
 514                                            tipc_own_addr,
 515                                            CONN_MANAGER,
 516                                            CONN_PROBE,
 517                                            TIPC_OK,
 518                                            port_out_seqno(p_ptr),
 519                                            0);
 520                 port_incr_out_seqno(p_ptr);
 521                 p_ptr->probing_state = PROBING;
 522                 k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
 523         }
 524         tipc_port_unlock(p_ptr);
 525         tipc_net_route_msg(buf);
 526 }
 527
 528
 529 static void port_handle_node_down(unsigned long ref)
 530 {
 531         struct port *p_ptr = tipc_port_lock(ref);
 532         struct sk_buff* buf = NULL;
 533
 534         if (!p_ptr)
 535                 return;
 536         buf = port_build_self_abort_msg(p_ptr, TIPC_ERR_NO_NODE);
 537         tipc_port_unlock(p_ptr);
 538         tipc_net_route_msg(buf);
 539 }
 540
 541
 542 static struct sk_buff *port_build_self_abort_msg(struct port *p_ptr, u32 err)
 543 {
 544         u32 imp = msg_importance(&p_ptr->publ.phdr);
 545
 546         if (!p_ptr->publ.connected)
 547                 return NULL;
 548         if (imp < TIPC_CRITICAL_IMPORTANCE)
 549                 imp++;
 550         return port_build_proto_msg(p_ptr->publ.ref,
 551                                     tipc_own_addr,
 552                                     port_peerport(p_ptr),
 553                                     port_peernode(p_ptr),
 554                                     imp,
 555                                     TIPC_CONN_MSG,
 556                                     err,
 557                                     p_ptr->last_in_seqno + 1,
 558                                     0);
 559 }
 560
 561
 562 static struct sk_buff *port_build_peer_abort_msg(struct port *p_ptr, u32 err)
 563 {
 564         u32 imp = msg_importance(&p_ptr->publ.phdr);
 565
 566         if (!p_ptr->publ.connected)
 567                 return NULL;
 568         if (imp < TIPC_CRITICAL_IMPORTANCE)
 569                 imp++;
 570         return port_build_proto_msg(port_peerport(p_ptr),
 571                                     port_peernode(p_ptr),
 572                                     p_ptr->publ.ref,
 573                                     tipc_own_addr,
 574                                     imp,
 575                                     TIPC_CONN_MSG,
 576                                     err,
 577                                     port_out_seqno(p_ptr),
 578                                     0);
 579 }
 580
 581 void tipc_port_recv_proto_msg(struct sk_buff *buf)
 582 {
 583         struct tipc_msg *msg = buf_msg(buf);
 584         struct port *p_ptr = tipc_port_lock(msg_destport(msg));
 585         u32 err = TIPC_OK;
 586         struct sk_buff *r_buf = NULL;
 587         struct sk_buff *abort_buf = NULL;
 588
 589         msg_dbg(msg, "PORT<RECV<:");
 590
 591         if (!p_ptr) {
 592                 err = TIPC_ERR_NO_PORT;
 593         } else if (p_ptr->publ.connected) {
 594                 if (port_peernode(p_ptr) != msg_orignode(msg))
 595                         err = TIPC_ERR_NO_PORT;
 596                 if (port_peerport(p_ptr) != msg_origport(msg))
 597                         err = TIPC_ERR_NO_PORT;
 598                 if (!err && msg_routed(msg)) {
 599                         u32 seqno = msg_transp_seqno(msg);
 600                         u32 myno =  ++p_ptr->last_in_seqno;
 601                         if (seqno != myno) {
 602                                 err = TIPC_ERR_NO_PORT;
 603                                 abort_buf = port_build_self_abort_msg(p_ptr, err);
 604                         }
 605                 }
 606                 if (msg_type(msg) == CONN_ACK) {
 607                         int wakeup = tipc_port_congested(p_ptr) &&
 608                                      p_ptr->publ.congested &&
 609                                      p_ptr->wakeup;
 610                         p_ptr->acked += msg_msgcnt(msg);
 611                         if (tipc_port_congested(p_ptr))
 612                                 goto exit;
 613                         p_ptr->publ.congested = 0;
 614                         if (!wakeup)
 615                                 goto exit;
 616                         p_ptr->wakeup(&p_ptr->publ);
 617                         goto exit;
 618                 }
 619         } else if (p_ptr->publ.published) {
 620                 err = TIPC_ERR_NO_PORT;
 621         }
 622         if (err) {
 623                 r_buf = port_build_proto_msg(msg_origport(msg),
 624                                              msg_orignode(msg),
 625                                              msg_destport(msg),
 626                                              tipc_own_addr,
 627                                              TIPC_HIGH_IMPORTANCE,
 628                                              TIPC_CONN_MSG,
 629                                              err,
 630                                              0,
 631                                              0);
 632                 goto exit;
 633         }
 634
 635         /* All is fine */
 636         if (msg_type(msg) == CONN_PROBE) {
 637                 r_buf = port_build_proto_msg(msg_origport(msg),
 638                                              msg_orignode(msg),
 639                                              msg_destport(msg),
 640                                              tipc_own_addr,
 641                                              CONN_MANAGER,
 642                                              CONN_PROBE_REPLY,
 643                                              TIPC_OK,
 644                                              port_out_seqno(p_ptr),
 645                                              0);
 646         }
 647         p_ptr->probing_state = CONFIRMED;
 648         port_incr_out_seqno(p_ptr);
 649 exit:
 650         if (p_ptr)
 651                 tipc_port_unlock(p_ptr);
 652         tipc_net_route_msg(r_buf);
 653         tipc_net_route_msg(abort_buf);
 654         buf_discard(buf);
 655 }
 656
 657 static void port_print(struct port *p_ptr, struct print_buf *buf, int full_id)
 658 {
 659         struct publication *publ;
 660
 661         if (full_id)
 662                 tipc_printf(buf, "<%u.%u.%u:%u>:",
 663                             tipc_zone(tipc_own_addr), tipc_cluster(tipc_own_addr),
 664                             tipc_node(tipc_own_addr), p_ptr->publ.ref);
 665         else
 666                 tipc_printf(buf, "%-10u:", p_ptr->publ.ref);
 667
 668         if (p_ptr->publ.connected) {
 669                 u32 dport = port_peerport(p_ptr);
 670                 u32 destnode = port_peernode(p_ptr);
 671
 672                 tipc_printf(buf, " connected to <%u.%u.%u:%u>",
 673                             tipc_zone(destnode), tipc_cluster(destnode),
 674                             tipc_node(destnode), dport);
 675                 if (p_ptr->publ.conn_type != 0)
 676                         tipc_printf(buf, " via {%u,%u}",
 677                                     p_ptr->publ.conn_type,
 678                                     p_ptr->publ.conn_instance);
 679         }
 680         else if (p_ptr->publ.published) {
 681                 tipc_printf(buf, " bound to");
 682                 list_for_each_entry(publ, &p_ptr->publications, pport_list) {
 683                         if (publ->lower == publ->upper)
 684                                 tipc_printf(buf, " {%u,%u}", publ->type,
 685                                             publ->lower);
 686                         else
 687                                 tipc_printf(buf, " {%u,%u,%u}", publ->type,
 688                                             publ->lower, publ->upper);
 689                 }
 690         }
 691         tipc_printf(buf, "\n");
 692 }
 693
 694 #define MAX_PORT_QUERY 32768
 695
 696 struct sk_buff *tipc_port_get_ports(void)
 697 {
 698         struct sk_buff *buf;
 699         struct tlv_desc *rep_tlv;
 700         struct print_buf pb;
 701         struct port *p_ptr;
 702         int str_len;
 703
 704         buf = tipc_cfg_reply_alloc(TLV_SPACE(MAX_PORT_QUERY));
 705         if (!buf)
 706                 return NULL;
 707         rep_tlv = (struct tlv_desc *)buf->data;
 708
 709         tipc_printbuf_init(&pb, TLV_DATA(rep_tlv), MAX_PORT_QUERY);
 710         spin_lock_bh(&tipc_port_list_lock);
 711         list_for_each_entry(p_ptr, &ports, port_list) {
 712                 spin_lock_bh(p_ptr->publ.lock);
 713                 port_print(p_ptr, &pb, 0);
 714                 spin_unlock_bh(p_ptr->publ.lock);
 715         }
 716         spin_unlock_bh(&tipc_port_list_lock);
 717         str_len = tipc_printbuf_validate(&pb);
 718
 719         skb_put(buf, TLV_SPACE(str_len));
 720         TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
 721
 722         return buf;
 723 }
 724
 725 #if 0
 726
 727 #define MAX_PORT_STATS 2000
 728
 729 struct sk_buff *port_show_stats(const void *req_tlv_area, int req_tlv_space)
 730 {
 731         u32 ref;
 732         struct port *p_ptr;
 733         struct sk_buff *buf;
 734         struct tlv_desc *rep_tlv;
 735         struct print_buf pb;
 736         int str_len;
 737
 738         if (!TLV_CHECK(req_tlv_area, req_tlv_space, TIPC_TLV_PORT_REF))
 739                 return cfg_reply_error_string(TIPC_CFG_TLV_ERROR);
 740
 741         ref = *(u32 *)TLV_DATA(req_tlv_area);
 742         ref = ntohl(ref);
 743
 744         p_ptr = tipc_port_lock(ref);
 745         if (!p_ptr)
 746                 return cfg_reply_error_string("port not found");
 747
 748         buf = tipc_cfg_reply_alloc(TLV_SPACE(MAX_PORT_STATS));
 749         if (!buf) {
 750                 tipc_port_unlock(p_ptr);
 751                 return NULL;
 752         }
 753         rep_tlv = (struct tlv_desc *)buf->data;
 754
 755         tipc_printbuf_init(&pb, TLV_DATA(rep_tlv), MAX_PORT_STATS);
 756         port_print(p_ptr, &pb, 1);
 757         /* NEED TO FILL IN ADDITIONAL PORT STATISTICS HERE */
 758         tipc_port_unlock(p_ptr);
 759         str_len = tipc_printbuf_validate(&pb);
 760
 761         skb_put(buf, TLV_SPACE(str_len));
 762         TLV_SET(rep_tlv, TIPC_TLV_ULTRA_STRING, NULL, str_len);
 763
 764         return buf;
 765 }
 766
 767 #endif
 768
 769 void tipc_port_reinit(void)
 770 {
 771         struct port *p_ptr;
 772         struct tipc_msg *msg;
 773
 774         spin_lock_bh(&tipc_port_list_lock);
 775         list_for_each_entry(p_ptr, &ports, port_list) {
 776                 msg = &p_ptr->publ.phdr;
 777                 if (msg_orignode(msg) == tipc_own_addr)
 778                         break;
 779                 msg_set_prevnode(msg, tipc_own_addr);
 780                 msg_set_orignode(msg, tipc_own_addr);
 781         }
 782         spin_unlock_bh(&tipc_port_list_lock);
 783 }
 784
 785
 786 /*
 787  *  port_dispatcher_sigh(): Signal handler for messages destinated
 788  *                          to the tipc_port interface.
 789  */
 790
 791 static void port_dispatcher_sigh(void *dummy)
 792 {
 793         struct sk_buff *buf;
 794
 795         spin_lock_bh(&queue_lock);
 796         buf = msg_queue_head;
 797         msg_queue_head = NULL;
 798         spin_unlock_bh(&queue_lock);
 799
 800         while (buf) {
 801                 struct port *p_ptr;
 802                 struct user_port *up_ptr;
 803                 struct tipc_portid orig;
 804                 struct tipc_name_seq dseq;
 805                 void *usr_handle;
 806                 int connected;
 807                 int published;
 808                 u32 message_type;
 809
 810                 struct sk_buff *next = buf->next;
 811                 struct tipc_msg *msg = buf_msg(buf);
 812                 u32 dref = msg_destport(msg);
 813
 814                 message_type = msg_type(msg);
 815                 if (message_type > TIPC_DIRECT_MSG)
 816                         goto reject;    /* Unsupported message type */
 817
 818                 p_ptr = tipc_port_lock(dref);
 819                 if (!p_ptr)
 820                         goto reject;    /* Port deleted while msg in queue */
 821
 822                 orig.ref = msg_origport(msg);
 823                 orig.node = msg_orignode(msg);
 824                 up_ptr = p_ptr->user_port;
 825                 usr_handle = up_ptr->usr_handle;
 826                 connected = p_ptr->publ.connected;
 827                 published = p_ptr->publ.published;
 828
 829                 if (unlikely(msg_errcode(msg)))
 830                         goto err;
 831
 832                 switch (message_type) {
 833
 834                 case TIPC_CONN_MSG:{
 835                                 tipc_conn_msg_event cb = up_ptr->conn_msg_cb;
 836                                 u32 peer_port = port_peerport(p_ptr);
 837                                 u32 peer_node = port_peernode(p_ptr);
 838
 839                                 tipc_port_unlock(p_ptr);
 840                                 if (unlikely(!cb))
 841                                         goto reject;
 842                                 if (unlikely(!connected)) {
 843                                         if (tipc_connect2port(dref, &orig))
 844                                                 goto reject;
 845                                 } else if ((msg_origport(msg) != peer_port) ||
 846                                            (msg_orignode(msg) != peer_node))
 847                                         goto reject;
 848                                 if (unlikely(++p_ptr->publ.conn_unacked >=
 849                                              TIPC_FLOW_CONTROL_WIN))
 850                                         tipc_acknowledge(dref,
 851                                                          p_ptr->publ.conn_unacked);
 852                                 skb_pull(buf, msg_hdr_sz(msg));
 853                                 cb(usr_handle, dref, &buf, msg_data(msg),
 854                                    msg_data_sz(msg));
 855                                 break;
 856                         }
 857                 case TIPC_DIRECT_MSG:{
 858                                 tipc_msg_event cb = up_ptr->msg_cb;
 859
 860                                 tipc_port_unlock(p_ptr);
 861                                 if (unlikely(!cb || connected))
 862                                         goto reject;
 863                                 skb_pull(buf, msg_hdr_sz(msg));
 864                                 cb(usr_handle, dref, &buf, msg_data(msg),
 865                                    msg_data_sz(msg), msg_importance(msg),
 866                                    &orig);
 867                                 break;
 868                         }
 869                 case TIPC_MCAST_MSG:
 870                 case TIPC_NAMED_MSG:{
 871                                 tipc_named_msg_event cb = up_ptr->named_msg_cb;
 872
 873                                 tipc_port_unlock(p_ptr);
 874                                 if (unlikely(!cb || connected || !published))
 875                                         goto reject;
 876                                 dseq.type =  msg_nametype(msg);
 877                                 dseq.lower = msg_nameinst(msg);
 878                                 dseq.upper = (message_type == TIPC_NAMED_MSG)
 879                                         ? dseq.lower : msg_nameupper(msg);
 880                                 skb_pull(buf, msg_hdr_sz(msg));
 881                                 cb(usr_handle, dref, &buf, msg_data(msg),
 882                                    msg_data_sz(msg), msg_importance(msg),
 883                                    &orig, &dseq);
 884                                 break;
 885                         }
 886                 }
 887                 if (buf)
 888                         buf_discard(buf);
 889                 buf = next;
 890                 continue;
 891 err:
 892                 switch (message_type) {
 893
 894                 case TIPC_CONN_MSG:{
 895                                 tipc_conn_shutdown_event cb =
 896                                         up_ptr->conn_err_cb;
 897                                 u32 peer_port = port_peerport(p_ptr);
 898                                 u32 peer_node = port_peernode(p_ptr);
 899
 900                                 tipc_port_unlock(p_ptr);
 901                                 if (!cb || !connected)
 902                                         break;
 903                                 if ((msg_origport(msg) != peer_port) ||
 904                                     (msg_orignode(msg) != peer_node))
 905                                         break;
 906                                 tipc_disconnect(dref);
 907                                 skb_pull(buf, msg_hdr_sz(msg));
 908                                 cb(usr_handle, dref, &buf, msg_data(msg),
 909                                    msg_data_sz(msg), msg_errcode(msg));
 910                                 break;
 911                         }
 912                 case TIPC_DIRECT_MSG:{
 913                                 tipc_msg_err_event cb = up_ptr->err_cb;
 914
 915                                 tipc_port_unlock(p_ptr);
 916                                 if (!cb || connected)
 917                                         break;
 918                                 skb_pull(buf, msg_hdr_sz(msg));
 919                                 cb(usr_handle, dref, &buf, msg_data(msg),
 920                                    msg_data_sz(msg), msg_errcode(msg), &orig);
 921                                 break;
 922                         }
 923                 case TIPC_MCAST_MSG:
 924                 case TIPC_NAMED_MSG:{
 925                                 tipc_named_msg_err_event cb =
 926                                         up_ptr->named_err_cb;
 927
 928                                 tipc_port_unlock(p_ptr);
 929                                 if (!cb || connected)
 930                                         break;
 931                                 dseq.type =  msg_nametype(msg);
 932                                 dseq.lower = msg_nameinst(msg);
 933                                 dseq.upper = (message_type == TIPC_NAMED_MSG)
 934                                         ? dseq.lower : msg_nameupper(msg);
 935                                 skb_pull(buf, msg_hdr_sz(msg));
 936                                 cb(usr_handle, dref, &buf, msg_data(msg),
 937                                    msg_data_sz(msg), msg_errcode(msg), &dseq);
 938                                 break;
 939                         }
 940                 }
 941                 if (buf)
 942                         buf_discard(buf);
 943                 buf = next;
 944                 continue;
 945 reject:
 946                 tipc_reject_msg(buf, TIPC_ERR_NO_PORT);
 947                 buf = next;
 948         }
 949 }
 950
 951 /*
 952  *  port_dispatcher(): Dispatcher for messages destinated
 953  *  to the tipc_port interface. Called with port locked.
 954  */
 955
 956 static u32 port_dispatcher(struct tipc_port *dummy, struct sk_buff *buf)
 957 {
 958         buf->next = NULL;
 959         spin_lock_bh(&queue_lock);
 960         if (msg_queue_head) {
 961                 msg_queue_tail->next = buf;
 962                 msg_queue_tail = buf;
 963         } else {
 964                 msg_queue_tail = msg_queue_head = buf;
 965                 tipc_k_signal((Handler)port_dispatcher_sigh, 0);
 966         }
 967         spin_unlock_bh(&queue_lock);
 968         return TIPC_OK;
 969 }
 970
 971 /*
 972  * Wake up port after congestion: Called with port locked,
 973  *
 974  */
 975
 976 static void port_wakeup_sh(unsigned long ref)
 977 {
 978         struct port *p_ptr;
 979         struct user_port *up_ptr;
 980         tipc_continue_event cb = NULL;
 981         void *uh = NULL;
 982
 983         p_ptr = tipc_port_lock(ref);
 984         if (p_ptr) {
 985                 up_ptr = p_ptr->user_port;
 986                 if (up_ptr) {
 987                         cb = up_ptr->continue_event_cb;
 988                         uh = up_ptr->usr_handle;
 989                 }
 990                 tipc_port_unlock(p_ptr);
 991         }
 992         if (cb)
 993                 cb(uh, ref);
 994 }
 995
 996
 997 static void port_wakeup(struct tipc_port *p_ptr)
 998 {
 999         tipc_k_signal((Handler)port_wakeup_sh, p_ptr->ref);
1000 }
1001
1002 void tipc_acknowledge(u32 ref, u32 ack)
1003 {
1004         struct port *p_ptr;
1005         struct sk_buff *buf = NULL;
1006
1007         p_ptr = tipc_port_lock(ref);
1008         if (!p_ptr)
1009                 return;
1010         if (p_ptr->publ.connected) {
1011                 p_ptr->publ.conn_unacked -= ack;
1012                 buf = port_build_proto_msg(port_peerport(p_ptr),
1013                                            port_peernode(p_ptr),
1014                                            ref,
1015                                            tipc_own_addr,
1016                                            CONN_MANAGER,
1017                                            CONN_ACK,
1018                                            TIPC_OK,
1019                                            port_out_seqno(p_ptr),
1020                                            ack);
1021         }
1022         tipc_port_unlock(p_ptr);
1023         tipc_net_route_msg(buf);
1024 }
1025
1026 /*
1027  * tipc_createport(): user level call. Will add port to
1028  *                    registry if non-zero user_ref.
1029  */
1030
1031 int tipc_createport(u32 user_ref,
1032                     void *usr_handle,
1033                     unsigned int importance,
1034                     tipc_msg_err_event error_cb,
1035                     tipc_named_msg_err_event named_error_cb,
1036                     tipc_conn_shutdown_event conn_error_cb,
1037                     tipc_msg_event msg_cb,
1038                     tipc_named_msg_event named_msg_cb,
1039                     tipc_conn_msg_event conn_msg_cb,
1040                     tipc_continue_event continue_event_cb,/* May be zero */
1041                     u32 *portref)
1042 {
1043         struct user_port *up_ptr;
1044         struct port *p_ptr;
1045         struct tipc_port *tp_ptr;
1046         u32 ref;
1047
1048         up_ptr = kmalloc(sizeof(*up_ptr), GFP_ATOMIC);
1049         if (!up_ptr) {
1050                 warn("Port creation failed, no memory\n");
1051                 return -ENOMEM;
1052         }
1053         ref = tipc_createport_raw(NULL, port_dispatcher, port_wakeup,
1054                                   importance, &tp_ptr);
1055         if (ref == 0) {
1056                 kfree(up_ptr);
1057                 return -ENOMEM;
1058         }
1059         p_ptr = (struct port *)tp_ptr;
1060
1061         p_ptr->user_port = up_ptr;
1062         up_ptr->user_ref = user_ref;
1063         up_ptr->usr_handle = usr_handle;
1064         up_ptr->ref = p_ptr->publ.ref;
1065         up_ptr->err_cb = error_cb;
1066         up_ptr->named_err_cb = named_error_cb;
1067         up_ptr->conn_err_cb = conn_error_cb;
1068         up_ptr->msg_cb = msg_cb;
1069         up_ptr->named_msg_cb = named_msg_cb;
1070         up_ptr->conn_msg_cb = conn_msg_cb;
1071         up_ptr->continue_event_cb = continue_event_cb;
1072         INIT_LIST_HEAD(&up_ptr->uport_list);
1073         tipc_reg_add_port(up_ptr);
1074         *portref = p_ptr->publ.ref;
1075         dbg(" tipc_createport: %x with ref %u\n", p_ptr, p_ptr->publ.ref);
1076         tipc_port_unlock(p_ptr);
1077         return TIPC_OK;
1078 }
1079
1080 int tipc_ownidentity(u32 ref, struct tipc_portid *id)
1081 {
1082         id->ref = ref;
1083         id->node = tipc_own_addr;
1084         return TIPC_OK;
1085 }
1086
1087 int tipc_portimportance(u32 ref, unsigned int *importance)
1088 {
1089         struct port *p_ptr;
1090
1091         p_ptr = tipc_port_lock(ref);
1092         if (!p_ptr)
1093                 return -EINVAL;
1094         *importance = (unsigned int)msg_importance(&p_ptr->publ.phdr);
1095         tipc_port_unlock(p_ptr);
1096         return TIPC_OK;
1097 }
1098
1099 int tipc_set_portimportance(u32 ref, unsigned int imp)
1100 {
1101         struct port *p_ptr;
1102
1103         if (imp > TIPC_CRITICAL_IMPORTANCE)
1104                 return -EINVAL;
1105
1106         p_ptr = tipc_port_lock(ref);
1107         if (!p_ptr)
1108                 return -EINVAL;
1109         msg_set_importance(&p_ptr->publ.phdr, (u32)imp);
1110         tipc_port_unlock(p_ptr);
1111         return TIPC_OK;
1112 }
1113
1114
1115 int tipc_publish(u32 ref, unsigned int scope, struct tipc_name_seq const *seq)
1116 {
1117         struct port *p_ptr;
1118         struct publication *publ;
1119         u32 key;
1120         int res = -EINVAL;
1121
1122         p_ptr = tipc_port_lock(ref);
1123         if (!p_ptr)
1124                 return -EINVAL;
1125
1126         dbg("tipc_publ %u, p_ptr = %x, conn = %x, scope = %x, "
1127             "lower = %u, upper = %u\n",
1128             ref, p_ptr, p_ptr->publ.connected, scope, seq->lower, seq->upper);
1129         if (p_ptr->publ.connected)
1130                 goto exit;
1131         if (seq->lower > seq->upper)
1132                 goto exit;
1133         if ((scope < TIPC_ZONE_SCOPE) || (scope > TIPC_NODE_SCOPE))
1134                 goto exit;
1135         key = ref + p_ptr->pub_count + 1;
1136         if (key == ref) {
1137                 res = -EADDRINUSE;
1138                 goto exit;
1139         }
1140         publ = tipc_nametbl_publish(seq->type, seq->lower, seq->upper,
1141                                     scope, p_ptr->publ.ref, key);
1142         if (publ) {
1143                 list_add(&publ->pport_list, &p_ptr->publications);
1144                 p_ptr->pub_count++;
1145                 p_ptr->publ.published = 1;
1146                 res = TIPC_OK;
1147         }
1148 exit:
1149         tipc_port_unlock(p_ptr);
1150         return res;
1151 }
1152
1153 int tipc_withdraw(u32 ref, unsigned int scope, struct tipc_name_seq const *seq)
1154 {
1155         struct port *p_ptr;
1156         struct publication *publ;
1157         struct publication *tpubl;
1158         int res = -EINVAL;
1159
1160         p_ptr = tipc_port_lock(ref);
1161         if (!p_ptr)
1162                 return -EINVAL;
1163         if (!seq) {
1164                 list_for_each_entry_safe(publ, tpubl,
1165                                          &p_ptr->publications, pport_list) {
1166                         tipc_nametbl_withdraw(publ->type, publ->lower,
1167                                               publ->ref, publ->key);
1168                 }
1169                 res = TIPC_OK;
1170         } else {
1171                 list_for_each_entry_safe(publ, tpubl,
1172                                          &p_ptr->publications, pport_list) {
1173                         if (publ->scope != scope)
1174                                 continue;
1175                         if (publ->type != seq->type)
1176                                 continue;
1177                         if (publ->lower != seq->lower)
1178                                 continue;
1179                         if (publ->upper != seq->upper)
1180                                 break;
1181                         tipc_nametbl_withdraw(publ->type, publ->lower,
1182                                               publ->ref, publ->key);
1183                         res = TIPC_OK;
1184                         break;
1185                 }
1186         }
1187         if (list_empty(&p_ptr->publications))
1188                 p_ptr->publ.published = 0;
1189         tipc_port_unlock(p_ptr);
1190         return res;
1191 }
1192
1193 int tipc_connect2port(u32 ref, struct tipc_portid const *peer)
1194 {
1195         struct port *p_ptr;
1196         struct tipc_msg *msg;
1197         int res = -EINVAL;
1198
1199         p_ptr = tipc_port_lock(ref);
1200         if (!p_ptr)
1201                 return -EINVAL;
1202         if (p_ptr->publ.published || p_ptr->publ.connected)
1203                 goto exit;
1204         if (!peer->ref)
1205                 goto exit;
1206
1207         msg = &p_ptr->publ.phdr;
1208         msg_set_destnode(msg, peer->node);
1209         msg_set_destport(msg, peer->ref);
1210         msg_set_orignode(msg, tipc_own_addr);
1211         msg_set_origport(msg, p_ptr->publ.ref);
1212         msg_set_transp_seqno(msg, 42);
1213         msg_set_type(msg, TIPC_CONN_MSG);
1214         if (!may_route(peer->node))
1215                 msg_set_hdr_sz(msg, SHORT_H_SIZE);
1216         else
1217                 msg_set_hdr_sz(msg, LONG_H_SIZE);
1218
1219         p_ptr->probing_interval = PROBING_INTERVAL;
1220         p_ptr->probing_state = CONFIRMED;
1221         p_ptr->publ.connected = 1;
1222         k_start_timer(&p_ptr->timer, p_ptr->probing_interval);
1223
1224         tipc_nodesub_subscribe(&p_ptr->subscription,peer->node,
1225                           (void *)(unsigned long)ref,
1226                           (net_ev_handler)port_handle_node_down);
1227         res = TIPC_OK;
1228 exit:
1229         tipc_port_unlock(p_ptr);
1230         p_ptr->publ.max_pkt = tipc_link_get_max_pkt(peer->node, ref);
1231         return res;
1232 }
1233
1234 /**
1235  * tipc_disconnect_port - disconnect port from peer
1236  *
1237  * Port must be locked.
1238  */
1239
1240 int tipc_disconnect_port(struct tipc_port *tp_ptr)
1241 {
1242         int res;
1243
1244         if (tp_ptr->connected) {
1245                 tp_ptr->connected = 0;
1246                 /* let timer expire on it's own to avoid deadlock! */
1247                 tipc_nodesub_unsubscribe(
1248                         &((struct port *)tp_ptr)->subscription);
1249                 res = TIPC_OK;
1250         } else {
1251                 res = -ENOTCONN;
1252         }
1253         return res;
1254 }
1255
1256 /*
1257  * tipc_disconnect(): Disconnect port form peer.
1258  *                    This is a node local operation.
1259  */
1260
1261 int tipc_disconnect(u32 ref)
1262 {
1263         struct port *p_ptr;
1264         int res;
1265
1266         p_ptr = tipc_port_lock(ref);
1267         if (!p_ptr)
1268                 return -EINVAL;
1269         res = tipc_disconnect_port((struct tipc_port *)p_ptr);
1270         tipc_port_unlock(p_ptr);
1271         return res;
1272 }
1273
1274 /*
1275  * tipc_shutdown(): Send a SHUTDOWN msg to peer and disconnect
1276  */
1277 int tipc_shutdown(u32 ref)
1278 {
1279         struct port *p_ptr;
1280         struct sk_buff *buf = NULL;
1281
1282         p_ptr = tipc_port_lock(ref);
1283         if (!p_ptr)
1284                 return -EINVAL;
1285
1286         if (p_ptr->publ.connected) {
1287                 u32 imp = msg_importance(&p_ptr->publ.phdr);
1288                 if (imp < TIPC_CRITICAL_IMPORTANCE)
1289                         imp++;
1290                 buf = port_build_proto_msg(port_peerport(p_ptr),
1291                                            port_peernode(p_ptr),
1292                                            ref,
1293                                            tipc_own_addr,
1294                                            imp,
1295                                            TIPC_CONN_MSG,
1296                                            TIPC_CONN_SHUTDOWN,
1297                                            port_out_seqno(p_ptr),
1298                                            0);
1299         }
1300         tipc_port_unlock(p_ptr);
1301         tipc_net_route_msg(buf);
1302         return tipc_disconnect(ref);
1303 }
1304
1305 int tipc_isconnected(u32 ref, int *isconnected)
1306 {
1307         struct port *p_ptr;
1308
1309         p_ptr = tipc_port_lock(ref);
1310         if (!p_ptr)
1311                 return -EINVAL;
1312         *isconnected = p_ptr->publ.connected;
1313         tipc_port_unlock(p_ptr);
1314         return TIPC_OK;
1315 }
1316
1317 int tipc_peer(u32 ref, struct tipc_portid *peer)
1318 {
1319         struct port *p_ptr;
1320         int res;
1321
1322         p_ptr = tipc_port_lock(ref);
1323         if (!p_ptr)
1324                 return -EINVAL;
1325         if (p_ptr->publ.connected) {
1326                 peer->ref = port_peerport(p_ptr);
1327                 peer->node = port_peernode(p_ptr);
1328                 res = TIPC_OK;
1329         } else
1330                 res = -ENOTCONN;
1331         tipc_port_unlock(p_ptr);
1332         return res;
1333 }
1334
1335 int tipc_ref_valid(u32 ref)
1336 {
1337         /* Works irrespective of type */
1338         return !!tipc_ref_deref(ref);
1339 }
1340
1341
1342 /*
1343  *  tipc_port_recv_sections(): Concatenate and deliver sectioned
1344  *                        message for this node.
1345  */
1346
1347 int tipc_port_recv_sections(struct port *sender, unsigned int num_sect,
1348                        struct iovec const *msg_sect)
1349 {
1350         struct sk_buff *buf;
1351         int res;
1352
1353         res = msg_build(&sender->publ.phdr, msg_sect, num_sect,
1354                         MAX_MSG_SIZE, !sender->user_port, &buf);
1355         if (likely(buf))
1356                 tipc_port_recv_msg(buf);
1357         return res;
1358 }
1359
1360 /**
1361  * tipc_send - send message sections on connection
1362  */
1363
1364 int tipc_send(u32 ref, unsigned int num_sect, struct iovec const *msg_sect)
1365 {
1366         struct port *p_ptr;
1367         u32 destnode;
1368         int res;
1369
1370         p_ptr = tipc_port_deref(ref);
1371         if (!p_ptr || !p_ptr->publ.connected)
1372                 return -EINVAL;
1373
1374         p_ptr->publ.congested = 1;
1375         if (!tipc_port_congested(p_ptr)) {
1376                 destnode = port_peernode(p_ptr);
1377                 if (likely(destnode != tipc_own_addr))
1378                         res = tipc_link_send_sections_fast(p_ptr, msg_sect, num_sect,
1379                                                            destnode);
1380                 else
1381                         res = tipc_port_recv_sections(p_ptr, num_sect, msg_sect);
1382
1383                 if (likely(res != -ELINKCONG)) {
1384                         port_incr_out_seqno(p_ptr);
1385                         p_ptr->publ.congested = 0;
1386                         p_ptr->sent++;
1387                         return res;
1388                 }
1389         }
1390         if (port_unreliable(p_ptr)) {
1391                 p_ptr->publ.congested = 0;
1392                 /* Just calculate msg length and return */
1393                 return msg_calc_data_size(msg_sect, num_sect);
1394         }
1395         return -ELINKCONG;
1396 }
1397
1398 /**
1399  * tipc_send_buf - send message buffer on connection
1400  */
1401
1402 int tipc_send_buf(u32 ref, struct sk_buff *buf, unsigned int dsz)
1403 {
1404         struct port *p_ptr;
1405         struct tipc_msg *msg;
1406         u32 destnode;
1407         u32 hsz;
1408         u32 sz;
1409         u32 res;
1410
1411         p_ptr = tipc_port_deref(ref);
1412         if (!p_ptr || !p_ptr->publ.connected)
1413                 return -EINVAL;
1414
1415         msg = &p_ptr->publ.phdr;
1416         hsz = msg_hdr_sz(msg);
1417         sz = hsz + dsz;
1418         msg_set_size(msg, sz);
1419         if (skb_cow(buf, hsz))
1420                 return -ENOMEM;
1421
1422         skb_push(buf, hsz);
1423         skb_copy_to_linear_data(buf, msg, hsz);
1424         destnode = msg_destnode(msg);
1425         p_ptr->publ.congested = 1;
1426         if (!tipc_port_congested(p_ptr)) {
1427                 if (likely(destnode != tipc_own_addr))
1428                         res = tipc_send_buf_fast(buf, destnode);
1429                 else {
1430                         tipc_port_recv_msg(buf);
1431                         res = sz;
1432                 }
1433                 if (likely(res != -ELINKCONG)) {
1434                         port_incr_out_seqno(p_ptr);
1435                         p_ptr->sent++;
1436                         p_ptr->publ.congested = 0;
1437                         return res;
1438                 }
1439         }
1440         if (port_unreliable(p_ptr)) {
1441                 p_ptr->publ.congested = 0;
1442                 return dsz;
1443         }
1444         return -ELINKCONG;
1445 }
1446
1447 /**
1448  * tipc_forward2name - forward message sections to port name
1449  */
1450
1451 int tipc_forward2name(u32 ref,
1452                       struct tipc_name const *name,
1453                       u32 domain,
1454                       u32 num_sect,
1455                       struct iovec const *msg_sect,
1456                       struct tipc_portid const *orig,
1457                       unsigned int importance)
1458 {
1459         struct port *p_ptr;
1460         struct tipc_msg *msg;
1461         u32 destnode = domain;
1462         u32 destport = 0;
1463         int res;
1464
1465         p_ptr = tipc_port_deref(ref);
1466         if (!p_ptr || p_ptr->publ.connected)
1467                 return -EINVAL;
1468
1469         msg = &p_ptr->publ.phdr;
1470         msg_set_type(msg, TIPC_NAMED_MSG);
1471         msg_set_orignode(msg, orig->node);
1472         msg_set_origport(msg, orig->ref);
1473         msg_set_hdr_sz(msg, LONG_H_SIZE);
1474         msg_set_nametype(msg, name->type);
1475         msg_set_nameinst(msg, name->instance);
1476         msg_set_lookup_scope(msg, addr_scope(domain));
1477         if (importance <= TIPC_CRITICAL_IMPORTANCE)
1478                 msg_set_importance(msg,importance);
1479         destport = tipc_nametbl_translate(name->type, name->instance, &destnode);
1480         msg_set_destnode(msg, destnode);
1481         msg_set_destport(msg, destport);
1482
1483         if (likely(destport || destnode)) {
1484                 p_ptr->sent++;
1485                 if (likely(destnode == tipc_own_addr))
1486                         return tipc_port_recv_sections(p_ptr, num_sect, msg_sect);
1487                 res = tipc_link_send_sections_fast(p_ptr, msg_sect, num_sect,
1488                                                    destnode);
1489                 if (likely(res != -ELINKCONG))
1490                         return res;
1491                 if (port_unreliable(p_ptr)) {
1492                         /* Just calculate msg length and return */
1493                         return msg_calc_data_size(msg_sect, num_sect);
1494                 }
1495                 return -ELINKCONG;
1496         }
1497         return tipc_port_reject_sections(p_ptr, msg, msg_sect, num_sect,
1498                                          TIPC_ERR_NO_NAME);
1499 }
1500
1501 /**
1502  * tipc_send2name - send message sections to port name
1503  */
1504
1505 int tipc_send2name(u32 ref,
1506                    struct tipc_name const *name,
1507                    unsigned int domain,
1508                    unsigned int num_sect,
1509                    struct iovec const *msg_sect)
1510 {
1511         struct tipc_portid orig;
1512
1513         orig.ref = ref;
1514         orig.node = tipc_own_addr;
1515         return tipc_forward2name(ref, name, domain, num_sect, msg_sect, &orig,
1516                                  TIPC_PORT_IMPORTANCE);
1517 }
1518
1519 /**
1520  * tipc_forward_buf2name - forward message buffer to port name
1521  */
1522
1523 int tipc_forward_buf2name(u32 ref,
1524                           struct tipc_name const *name,
1525                           u32 domain,
1526                           struct sk_buff *buf,
1527                           unsigned int dsz,
1528                           struct tipc_portid const *orig,
1529                           unsigned int importance)
1530 {
1531         struct port *p_ptr;
1532         struct tipc_msg *msg;
1533         u32 destnode = domain;
1534         u32 destport = 0;
1535         int res;
1536
1537         p_ptr = (struct port *)tipc_ref_deref(ref);
1538         if (!p_ptr || p_ptr->publ.connected)
1539                 return -EINVAL;
1540
1541         msg = &p_ptr->publ.phdr;
1542         if (importance <= TIPC_CRITICAL_IMPORTANCE)
1543                 msg_set_importance(msg, importance);
1544         msg_set_type(msg, TIPC_NAMED_MSG);
1545         msg_set_orignode(msg, orig->node);
1546         msg_set_origport(msg, orig->ref);
1547         msg_set_nametype(msg, name->type);
1548         msg_set_nameinst(msg, name->instance);
1549         msg_set_lookup_scope(msg, addr_scope(domain));
1550         msg_set_hdr_sz(msg, LONG_H_SIZE);
1551         msg_set_size(msg, LONG_H_SIZE + dsz);
1552         destport = tipc_nametbl_translate(name->type, name->instance, &destnode);
1553         msg_set_destnode(msg, destnode);
1554         msg_set_destport(msg, destport);
1555         msg_dbg(msg, "forw2name ==> ");
1556         if (skb_cow(buf, LONG_H_SIZE))
1557                 return -ENOMEM;
1558         skb_push(buf, LONG_H_SIZE);
1559         skb_copy_to_linear_data(buf, msg, LONG_H_SIZE);
1560         msg_dbg(buf_msg(buf),"PREP:");
1561         if (likely(destport || destnode)) {
1562                 p_ptr->sent++;
1563                 if (destnode == tipc_own_addr)
1564                         return tipc_port_recv_msg(buf);
1565                 res = tipc_send_buf_fast(buf, destnode);
1566                 if (likely(res != -ELINKCONG))
1567                         return res;
1568                 if (port_unreliable(p_ptr))
1569                         return dsz;
1570                 return -ELINKCONG;
1571         }
1572         return tipc_reject_msg(buf, TIPC_ERR_NO_NAME);
1573 }
1574
1575 /**
1576  * tipc_send_buf2name - send message buffer to port name
1577  */
1578
1579 int tipc_send_buf2name(u32 ref,
1580                        struct tipc_name const *dest,
1581                        u32 domain,
1582                        struct sk_buff *buf,
1583                        unsigned int dsz)
1584 {
1585         struct tipc_portid orig;
1586
1587         orig.ref = ref;
1588         orig.node = tipc_own_addr;
1589         return tipc_forward_buf2name(ref, dest, domain, buf, dsz, &orig,
1590                                      TIPC_PORT_IMPORTANCE);
1591 }
1592
1593 /**
1594  * tipc_forward2port - forward message sections to port identity
1595  */
1596
1597 int tipc_forward2port(u32 ref,
1598                       struct tipc_portid const *dest,
1599                       unsigned int num_sect,
1600                       struct iovec const *msg_sect,
1601                       struct tipc_portid const *orig,
1602                       unsigned int importance)
1603 {
1604         struct port *p_ptr;
1605         struct tipc_msg *msg;
1606         int res;
1607
1608         p_ptr = tipc_port_deref(ref);
1609         if (!p_ptr || p_ptr->publ.connected)
1610                 return -EINVAL;
1611
1612         msg = &p_ptr->publ.phdr;
1613         msg_set_type(msg, TIPC_DIRECT_MSG);
1614         msg_set_orignode(msg, orig->node);
1615         msg_set_origport(msg, orig->ref);
1616         msg_set_destnode(msg, dest->node);
1617         msg_set_destport(msg, dest->ref);
1618         msg_set_hdr_sz(msg, DIR_MSG_H_SIZE);
1619         if (importance <= TIPC_CRITICAL_IMPORTANCE)
1620                 msg_set_importance(msg, importance);
1621         p_ptr->sent++;
1622         if (dest->node == tipc_own_addr)
1623                 return tipc_port_recv_sections(p_ptr, num_sect, msg_sect);
1624         res = tipc_link_send_sections_fast(p_ptr, msg_sect, num_sect, dest->node);
1625         if (likely(res != -ELINKCONG))
1626                 return res;
1627         if (port_unreliable(p_ptr)) {
1628                 /* Just calculate msg length and return */
1629                 return msg_calc_data_size(msg_sect, num_sect);
1630         }
1631         return -ELINKCONG;
1632 }
1633
1634 /**
1635  * tipc_send2port - send message sections to port identity
1636  */
1637
1638 int tipc_send2port(u32 ref,
1639                    struct tipc_portid const *dest,
1640                    unsigned int num_sect,
1641                    struct iovec const *msg_sect)
1642 {
1643         struct tipc_portid orig;
1644
1645         orig.ref = ref;
1646         orig.node = tipc_own_addr;
1647         return tipc_forward2port(ref, dest, num_sect, msg_sect, &orig,
1648                                  TIPC_PORT_IMPORTANCE);
1649 }
1650
1651 /**
1652  * tipc_forward_buf2port - forward message buffer to port identity
1653  */
1654 int tipc_forward_buf2port(u32 ref,
1655                           struct tipc_portid const *dest,
1656                           struct sk_buff *buf,
1657                           unsigned int dsz,
1658                           struct tipc_portid const *orig,
1659                           unsigned int importance)
1660 {
1661         struct port *p_ptr;
1662         struct tipc_msg *msg;
1663         int res;
1664
1665         p_ptr = (struct port *)tipc_ref_deref(ref);
1666         if (!p_ptr || p_ptr->publ.connected)
1667                 return -EINVAL;
1668
1669         msg = &p_ptr->publ.phdr;
1670         msg_set_type(msg, TIPC_DIRECT_MSG);
1671         msg_set_orignode(msg, orig->node);
1672         msg_set_origport(msg, orig->ref);
1673         msg_set_destnode(msg, dest->node);
1674         msg_set_destport(msg, dest->ref);
1675         msg_set_hdr_sz(msg, DIR_MSG_H_SIZE);
1676         if (importance <= TIPC_CRITICAL_IMPORTANCE)
1677                 msg_set_importance(msg, importance);
1678         msg_set_size(msg, DIR_MSG_H_SIZE + dsz);
1679         if (skb_cow(buf, DIR_MSG_H_SIZE))
1680                 return -ENOMEM;
1681
1682         skb_push(buf, DIR_MSG_H_SIZE);
1683         skb_copy_to_linear_data(buf, msg, DIR_MSG_H_SIZE);
1684         msg_dbg(msg, "buf2port: ");
1685         p_ptr->sent++;
1686         if (dest->node == tipc_own_addr)
1687                 return tipc_port_recv_msg(buf);
1688         res = tipc_send_buf_fast(buf, dest->node);
1689         if (likely(res != -ELINKCONG))
1690                 return res;
1691         if (port_unreliable(p_ptr))
1692                 return dsz;
1693         return -ELINKCONG;
1694 }
1695
1696 /**
1697  * tipc_send_buf2port - send message buffer to port identity
1698  */
1699
1700 int tipc_send_buf2port(u32 ref,
1701                        struct tipc_portid const *dest,
1702                        struct sk_buff *buf,
1703                        unsigned int dsz)
1704 {
1705         struct tipc_portid orig;
1706
1707         orig.ref = ref;
1708         orig.node = tipc_own_addr;
1709         return tipc_forward_buf2port(ref, dest, buf, dsz, &orig,
1710                                      TIPC_PORT_IMPORTANCE);
1711 }
1712