net/tipc/socket.c

   1 /*
   2  * net/tipc/socket.c: TIPC socket API
   3  *
   4  * Copyright (c) 2001-2007, 2012 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 "port.h"
  39
  40 #include <linux/export.h>
  41 #include <net/sock.h>
  42
  43 #define SS_LISTENING    -1      /* socket is listening */
  44 #define SS_READY        -2      /* socket is connectionless */
  45
  46 #define CONN_TIMEOUT_DEFAULT    8000    /* default connect timeout = 8s */
  47
  48 struct tipc_sock {
  49         struct sock sk;
  50         struct tipc_port *p;
  51         struct tipc_portid peer_name;
  52         unsigned int conn_timeout;
  53 };
  54
  55 #define tipc_sk(sk) ((struct tipc_sock *)(sk))
  56 #define tipc_sk_port(sk) (tipc_sk(sk)->p)
  57
  58 #define tipc_rx_ready(sock) (!skb_queue_empty(&sock->sk->sk_receive_queue) || \
  59                         (sock->state == SS_DISCONNECTING))
  60
  61 static int backlog_rcv(struct sock *sk, struct sk_buff *skb);
  62 static u32 dispatch(struct tipc_port *tport, struct sk_buff *buf);
  63 static void wakeupdispatch(struct tipc_port *tport);
  64 static void tipc_data_ready(struct sock *sk, int len);
  65 static void tipc_write_space(struct sock *sk);
  66 static int release(struct socket *sock);
  67 static int accept(struct socket *sock, struct socket *new_sock, int flags);
  68
  69 static const struct proto_ops packet_ops;
  70 static const struct proto_ops stream_ops;
  71 static const struct proto_ops msg_ops;
  72
  73 static struct proto tipc_proto;
  74 static struct proto tipc_proto_kern;
  75
  76 static int sockets_enabled;
  77
  78 /*
  79  * Revised TIPC socket locking policy:
  80  *
  81  * Most socket operations take the standard socket lock when they start
  82  * and hold it until they finish (or until they need to sleep).  Acquiring
  83  * this lock grants the owner exclusive access to the fields of the socket
  84  * data structures, with the exception of the backlog queue.  A few socket
  85  * operations can be done without taking the socket lock because they only
  86  * read socket information that never changes during the life of the socket.
  87  *
  88  * Socket operations may acquire the lock for the associated TIPC port if they
  89  * need to perform an operation on the port.  If any routine needs to acquire
  90  * both the socket lock and the port lock it must take the socket lock first
  91  * to avoid the risk of deadlock.
  92  *
  93  * The dispatcher handling incoming messages cannot grab the socket lock in
  94  * the standard fashion, since invoked it runs at the BH level and cannot block.
  95  * Instead, it checks to see if the socket lock is currently owned by someone,
  96  * and either handles the message itself or adds it to the socket's backlog
  97  * queue; in the latter case the queued message is processed once the process
  98  * owning the socket lock releases it.
  99  *
 100  * NOTE: Releasing the socket lock while an operation is sleeping overcomes
 101  * the problem of a blocked socket operation preventing any other operations
 102  * from occurring.  However, applications must be careful if they have
 103  * multiple threads trying to send (or receive) on the same socket, as these
 104  * operations might interfere with each other.  For example, doing a connect
 105  * and a receive at the same time might allow the receive to consume the
 106  * ACK message meant for the connect.  While additional work could be done
 107  * to try and overcome this, it doesn't seem to be worthwhile at the present.
 108  *
 109  * NOTE: Releasing the socket lock while an operation is sleeping also ensures
 110  * that another operation that must be performed in a non-blocking manner is
 111  * not delayed for very long because the lock has already been taken.
 112  *
 113  * NOTE: This code assumes that certain fields of a port/socket pair are
 114  * constant over its lifetime; such fields can be examined without taking
 115  * the socket lock and/or port lock, and do not need to be re-read even
 116  * after resuming processing after waiting.  These fields include:
 117  *   - socket type
 118  *   - pointer to socket sk structure (aka tipc_sock structure)
 119  *   - pointer to port structure
 120  *   - port reference
 121  */
 122
 123 /**
 124  * advance_rx_queue - discard first buffer in socket receive queue
 125  *
 126  * Caller must hold socket lock
 127  */
 128 static void advance_rx_queue(struct sock *sk)
 129 {
 130         kfree_skb(__skb_dequeue(&sk->sk_receive_queue));
 131 }
 132
 133 /**
 134  * reject_rx_queue - reject all buffers in socket receive queue
 135  *
 136  * Caller must hold socket lock
 137  */
 138 static void reject_rx_queue(struct sock *sk)
 139 {
 140         struct sk_buff *buf;
 141
 142         while ((buf = __skb_dequeue(&sk->sk_receive_queue)))
 143                 tipc_reject_msg(buf, TIPC_ERR_NO_PORT);
 144 }
 145
 146 /**
 147  * tipc_sk_create - create a TIPC socket
 148  * @net: network namespace (must be default network)
 149  * @sock: pre-allocated socket structure
 150  * @protocol: protocol indicator (must be 0)
 151  * @kern: caused by kernel or by userspace?
 152  *
 153  * This routine creates additional data structures used by the TIPC socket,
 154  * initializes them, and links them together.
 155  *
 156  * Returns 0 on success, errno otherwise
 157  */
 158 static int tipc_sk_create(struct net *net, struct socket *sock, int protocol,
 159                           int kern)
 160 {
 161         const struct proto_ops *ops;
 162         socket_state state;
 163         struct sock *sk;
 164         struct tipc_port *tp_ptr;
 165
 166         /* Validate arguments */
 167         if (unlikely(protocol != 0))
 168                 return -EPROTONOSUPPORT;
 169
 170         switch (sock->type) {
 171         case SOCK_STREAM:
 172                 ops = &stream_ops;
 173                 state = SS_UNCONNECTED;
 174                 break;
 175         case SOCK_SEQPACKET:
 176                 ops = &packet_ops;
 177                 state = SS_UNCONNECTED;
 178                 break;
 179         case SOCK_DGRAM:
 180         case SOCK_RDM:
 181                 ops = &msg_ops;
 182                 state = SS_READY;
 183                 break;
 184         default:
 185                 return -EPROTOTYPE;
 186         }
 187
 188         /* Allocate socket's protocol area */
 189         if (!kern)
 190                 sk = sk_alloc(net, AF_TIPC, GFP_KERNEL, &tipc_proto);
 191         else
 192                 sk = sk_alloc(net, AF_TIPC, GFP_KERNEL, &tipc_proto_kern);
 193
 194         if (sk == NULL)
 195                 return -ENOMEM;
 196
 197         /* Allocate TIPC port for socket to use */
 198         tp_ptr = tipc_createport(sk, &dispatch, &wakeupdispatch,
 199                                  TIPC_LOW_IMPORTANCE);
 200         if (unlikely(!tp_ptr)) {
 201                 sk_free(sk);
 202                 return -ENOMEM;
 203         }
 204
 205         /* Finish initializing socket data structures */
 206         sock->ops = ops;
 207         sock->state = state;
 208
 209         sock_init_data(sock, sk);
 210         sk->sk_backlog_rcv = backlog_rcv;
 211         sk->sk_rcvbuf = sysctl_tipc_rmem[1];
 212         sk->sk_data_ready = tipc_data_ready;
 213         sk->sk_write_space = tipc_write_space;
 214         tipc_sk(sk)->p = tp_ptr;
 215         tipc_sk(sk)->conn_timeout = CONN_TIMEOUT_DEFAULT;
 216
 217         spin_unlock_bh(tp_ptr->lock);
 218
 219         if (sock->state == SS_READY) {
 220                 tipc_set_portunreturnable(tp_ptr->ref, 1);
 221                 if (sock->type == SOCK_DGRAM)
 222                         tipc_set_portunreliable(tp_ptr->ref, 1);
 223         }
 224
 225         return 0;
 226 }
 227
 228 /**
 229  * tipc_sock_create_local - create TIPC socket from inside TIPC module
 230  * @type: socket type - SOCK_RDM or SOCK_SEQPACKET
 231  *
 232  * We cannot use sock_creat_kern here because it bumps module user count.
 233  * Since socket owner and creator is the same module we must make sure
 234  * that module count remains zero for module local sockets, otherwise
 235  * we cannot do rmmod.
 236  *
 237  * Returns 0 on success, errno otherwise
 238  */
 239 int tipc_sock_create_local(int type, struct socket **res)
 240 {
 241         int rc;
 242         struct sock *sk;
 243
 244         rc = sock_create_lite(AF_TIPC, type, 0, res);
 245         if (rc < 0) {
 246                 pr_err("Failed to create kernel socket\n");
 247                 return rc;
 248         }
 249         tipc_sk_create(&init_net, *res, 0, 1);
 250
 251         sk = (*res)->sk;
 252
 253         return 0;
 254 }
 255
 256 /**
 257  * tipc_sock_release_local - release socket created by tipc_sock_create_local
 258  * @sock: the socket to be released.
 259  *
 260  * Module reference count is not incremented when such sockets are created,
 261  * so we must keep it from being decremented when they are released.
 262  */
 263 void tipc_sock_release_local(struct socket *sock)
 264 {
 265         release(sock);
 266         sock->ops = NULL;
 267         sock_release(sock);
 268 }
 269
 270 /**
 271  * tipc_sock_accept_local - accept a connection on a socket created
 272  * with tipc_sock_create_local. Use this function to avoid that
 273  * module reference count is inadvertently incremented.
 274  *
 275  * @sock:    the accepting socket
 276  * @newsock: reference to the new socket to be created
 277  * @flags:   socket flags
 278  */
 279
 280 int tipc_sock_accept_local(struct socket *sock, struct socket **newsock,
 281                            int flags)
 282 {
 283         struct sock *sk = sock->sk;
 284         int ret;
 285
 286         ret = sock_create_lite(sk->sk_family, sk->sk_type,
 287                                sk->sk_protocol, newsock);
 288         if (ret < 0)
 289                 return ret;
 290
 291         ret = accept(sock, *newsock, flags);
 292         if (ret < 0) {
 293                 sock_release(*newsock);
 294                 return ret;
 295         }
 296         (*newsock)->ops = sock->ops;
 297         return ret;
 298 }
 299
 300 /**
 301  * release - destroy a TIPC socket
 302  * @sock: socket to destroy
 303  *
 304  * This routine cleans up any messages that are still queued on the socket.
 305  * For DGRAM and RDM socket types, all queued messages are rejected.
 306  * For SEQPACKET and STREAM socket types, the first message is rejected
 307  * and any others are discarded.  (If the first message on a STREAM socket
 308  * is partially-read, it is discarded and the next one is rejected instead.)
 309  *
 310  * NOTE: Rejected messages are not necessarily returned to the sender!  They
 311  * are returned or discarded according to the "destination droppable" setting
 312  * specified for the message by the sender.
 313  *
 314  * Returns 0 on success, errno otherwise
 315  */
 316 static int release(struct socket *sock)
 317 {
 318         struct sock *sk = sock->sk;
 319         struct tipc_port *tport;
 320         struct sk_buff *buf;
 321         int res;
 322
 323         /*
 324          * Exit if socket isn't fully initialized (occurs when a failed accept()
 325          * releases a pre-allocated child socket that was never used)
 326          */
 327         if (sk == NULL)
 328                 return 0;
 329
 330         tport = tipc_sk_port(sk);
 331         lock_sock(sk);
 332
 333         /*
 334          * Reject all unreceived messages, except on an active connection
 335          * (which disconnects locally & sends a 'FIN+' to peer)
 336          */
 337         while (sock->state != SS_DISCONNECTING) {
 338                 buf = __skb_dequeue(&sk->sk_receive_queue);
 339                 if (buf == NULL)
 340                         break;
 341                 if (TIPC_SKB_CB(buf)->handle != NULL)
 342                         kfree_skb(buf);
 343                 else {
 344                         if ((sock->state == SS_CONNECTING) ||
 345                             (sock->state == SS_CONNECTED)) {
 346                                 sock->state = SS_DISCONNECTING;
 347                                 tipc_disconnect(tport->ref);
 348                         }
 349                         tipc_reject_msg(buf, TIPC_ERR_NO_PORT);
 350                 }
 351         }
 352
 353         /*
 354          * Delete TIPC port; this ensures no more messages are queued
 355          * (also disconnects an active connection & sends a 'FIN-' to peer)
 356          */
 357         res = tipc_deleteport(tport);
 358
 359         /* Discard any remaining (connection-based) messages in receive queue */
 360         __skb_queue_purge(&sk->sk_receive_queue);
 361
 362         /* Reject any messages that accumulated in backlog queue */
 363         sock->state = SS_DISCONNECTING;
 364         release_sock(sk);
 365
 366         sock_put(sk);
 367         sock->sk = NULL;
 368
 369         return res;
 370 }
 371
 372 /**
 373  * bind - associate or disassocate TIPC name(s) with a socket
 374  * @sock: socket structure
 375  * @uaddr: socket address describing name(s) and desired operation
 376  * @uaddr_len: size of socket address data structure
 377  *
 378  * Name and name sequence binding is indicated using a positive scope value;
 379  * a negative scope value unbinds the specified name.  Specifying no name
 380  * (i.e. a socket address length of 0) unbinds all names from the socket.
 381  *
 382  * Returns 0 on success, errno otherwise
 383  *
 384  * NOTE: This routine doesn't need to take the socket lock since it doesn't
 385  *       access any non-constant socket information.
 386  */
 387 static int bind(struct socket *sock, struct sockaddr *uaddr, int uaddr_len)
 388 {
 389         struct sock *sk = sock->sk;
 390         struct sockaddr_tipc *addr = (struct sockaddr_tipc *)uaddr;
 391         struct tipc_port *tport = tipc_sk_port(sock->sk);
 392         int res = -EINVAL;
 393
 394         lock_sock(sk);
 395         if (unlikely(!uaddr_len)) {
 396                 res = tipc_withdraw(tport, 0, NULL);
 397                 goto exit;
 398         }
 399
 400         if (uaddr_len < sizeof(struct sockaddr_tipc)) {
 401                 res = -EINVAL;
 402                 goto exit;
 403         }
 404         if (addr->family != AF_TIPC) {
 405                 res = -EAFNOSUPPORT;
 406                 goto exit;
 407         }
 408
 409         if (addr->addrtype == TIPC_ADDR_NAME)
 410                 addr->addr.nameseq.upper = addr->addr.nameseq.lower;
 411         else if (addr->addrtype != TIPC_ADDR_NAMESEQ) {
 412                 res = -EAFNOSUPPORT;
 413                 goto exit;
 414         }
 415
 416         if ((addr->addr.nameseq.type < TIPC_RESERVED_TYPES) &&
 417             (addr->addr.nameseq.type != TIPC_TOP_SRV) &&
 418             (addr->addr.nameseq.type != TIPC_CFG_SRV)) {
 419                 res = -EACCES;
 420                 goto exit;
 421         }
 422
 423         res = (addr->scope > 0) ?
 424                 tipc_publish(tport, addr->scope, &addr->addr.nameseq) :
 425                 tipc_withdraw(tport, -addr->scope, &addr->addr.nameseq);
 426 exit:
 427         release_sock(sk);
 428         return res;
 429 }
 430
 431 /**
 432  * get_name - get port ID of socket or peer socket
 433  * @sock: socket structure
 434  * @uaddr: area for returned socket address
 435  * @uaddr_len: area for returned length of socket address
 436  * @peer: 0 = own ID, 1 = current peer ID, 2 = current/former peer ID
 437  *
 438  * Returns 0 on success, errno otherwise
 439  *
 440  * NOTE: This routine doesn't need to take the socket lock since it only
 441  *       accesses socket information that is unchanging (or which changes in
 442  *       a completely predictable manner).
 443  */
 444 static int get_name(struct socket *sock, struct sockaddr *uaddr,
 445                     int *uaddr_len, int peer)
 446 {
 447         struct sockaddr_tipc *addr = (struct sockaddr_tipc *)uaddr;
 448         struct tipc_sock *tsock = tipc_sk(sock->sk);
 449
 450         memset(addr, 0, sizeof(*addr));
 451         if (peer) {
 452                 if ((sock->state != SS_CONNECTED) &&
 453                         ((peer != 2) || (sock->state != SS_DISCONNECTING)))
 454                         return -ENOTCONN;
 455                 addr->addr.id.ref = tsock->peer_name.ref;
 456                 addr->addr.id.node = tsock->peer_name.node;
 457         } else {
 458                 addr->addr.id.ref = tsock->p->ref;
 459                 addr->addr.id.node = tipc_own_addr;
 460         }
 461
 462         *uaddr_len = sizeof(*addr);
 463         addr->addrtype = TIPC_ADDR_ID;
 464         addr->family = AF_TIPC;
 465         addr->scope = 0;
 466         addr->addr.name.domain = 0;
 467
 468         return 0;
 469 }
 470
 471 /**
 472  * poll - read and possibly block on pollmask
 473  * @file: file structure associated with the socket
 474  * @sock: socket for which to calculate the poll bits
 475  * @wait: ???
 476  *
 477  * Returns pollmask value
 478  *
 479  * COMMENTARY:
 480  * It appears that the usual socket locking mechanisms are not useful here
 481  * since the pollmask info is potentially out-of-date the moment this routine
 482  * exits.  TCP and other protocols seem to rely on higher level poll routines
 483  * to handle any preventable race conditions, so TIPC will do the same ...
 484  *
 485  * TIPC sets the returned events as follows:
 486  *
 487  * socket state         flags set
 488  * ------------         ---------
 489  * unconnected          no read flags
 490  *                      POLLOUT if port is not congested
 491  *
 492  * connecting           POLLIN/POLLRDNORM if ACK/NACK in rx queue
 493  *                      no write flags
 494  *
 495  * connected            POLLIN/POLLRDNORM if data in rx queue
 496  *                      POLLOUT if port is not congested
 497  *
 498  * disconnecting        POLLIN/POLLRDNORM/POLLHUP
 499  *                      no write flags
 500  *
 501  * listening            POLLIN if SYN in rx queue
 502  *                      no write flags
 503  *
 504  * ready                POLLIN/POLLRDNORM if data in rx queue
 505  * [connectionless]     POLLOUT (since port cannot be congested)
 506  *
 507  * IMPORTANT: The fact that a read or write operation is indicated does NOT
 508  * imply that the operation will succeed, merely that it should be performed
 509  * and will not block.
 510  */
 511 static unsigned int poll(struct file *file, struct socket *sock,
 512                          poll_table *wait)
 513 {
 514         struct sock *sk = sock->sk;
 515         u32 mask = 0;
 516
 517         sock_poll_wait(file, sk_sleep(sk), wait);
 518
 519         switch ((int)sock->state) {
 520         case SS_UNCONNECTED:
 521                 if (!tipc_sk_port(sk)->congested)
 522                         mask |= POLLOUT;
 523                 break;
 524         case SS_READY:
 525         case SS_CONNECTED:
 526                 if (!tipc_sk_port(sk)->congested)
 527                         mask |= POLLOUT;
 528                 /* fall thru' */
 529         case SS_CONNECTING:
 530         case SS_LISTENING:
 531                 if (!skb_queue_empty(&sk->sk_receive_queue))
 532                         mask |= (POLLIN | POLLRDNORM);
 533                 break;
 534         case SS_DISCONNECTING:
 535                 mask = (POLLIN | POLLRDNORM | POLLHUP);
 536                 break;
 537         }
 538
 539         return mask;
 540 }
 541
 542 /**
 543  * dest_name_check - verify user is permitted to send to specified port name
 544  * @dest: destination address
 545  * @m: descriptor for message to be sent
 546  *
 547  * Prevents restricted configuration commands from being issued by
 548  * unauthorized users.
 549  *
 550  * Returns 0 if permission is granted, otherwise errno
 551  */
 552 static int dest_name_check(struct sockaddr_tipc *dest, struct msghdr *m)
 553 {
 554         struct tipc_cfg_msg_hdr hdr;
 555
 556         if (likely(dest->addr.name.name.type >= TIPC_RESERVED_TYPES))
 557                 return 0;
 558         if (likely(dest->addr.name.name.type == TIPC_TOP_SRV))
 559                 return 0;
 560         if (likely(dest->addr.name.name.type != TIPC_CFG_SRV))
 561                 return -EACCES;
 562
 563         if (!m->msg_iovlen || (m->msg_iov[0].iov_len < sizeof(hdr)))
 564                 return -EMSGSIZE;
 565         if (copy_from_user(&hdr, m->msg_iov[0].iov_base, sizeof(hdr)))
 566                 return -EFAULT;
 567         if ((ntohs(hdr.tcm_type) & 0xC000) && (!capable(CAP_NET_ADMIN)))
 568                 return -EACCES;
 569
 570         return 0;
 571 }
 572
 573 /**
 574  * send_msg - send message in connectionless manner
 575  * @iocb: if NULL, indicates that socket lock is already held
 576  * @sock: socket structure
 577  * @m: message to send
 578  * @total_len: length of message
 579  *
 580  * Message must have an destination specified explicitly.
 581  * Used for SOCK_RDM and SOCK_DGRAM messages,
 582  * and for 'SYN' messages on SOCK_SEQPACKET and SOCK_STREAM connections.
 583  * (Note: 'SYN+' is prohibited on SOCK_STREAM.)
 584  *
 585  * Returns the number of bytes sent on success, or errno otherwise
 586  */
 587 static int send_msg(struct kiocb *iocb, struct socket *sock,
 588                     struct msghdr *m, size_t total_len)
 589 {
 590         struct sock *sk = sock->sk;
 591         struct tipc_port *tport = tipc_sk_port(sk);
 592         struct sockaddr_tipc *dest = (struct sockaddr_tipc *)m->msg_name;
 593         int needs_conn;
 594         long timeout_val;
 595         int res = -EINVAL;
 596
 597         if (unlikely(!dest))
 598                 return -EDESTADDRREQ;
 599         if (unlikely((m->msg_namelen < sizeof(*dest)) ||
 600                      (dest->family != AF_TIPC)))
 601                 return -EINVAL;
 602         if (total_len > TIPC_MAX_USER_MSG_SIZE)
 603                 return -EMSGSIZE;
 604
 605         if (iocb)
 606                 lock_sock(sk);
 607
 608         needs_conn = (sock->state != SS_READY);
 609         if (unlikely(needs_conn)) {
 610                 if (sock->state == SS_LISTENING) {
 611                         res = -EPIPE;
 612                         goto exit;
 613                 }
 614                 if (sock->state != SS_UNCONNECTED) {
 615                         res = -EISCONN;
 616                         goto exit;
 617                 }
 618                 if (tport->published) {
 619                         res = -EOPNOTSUPP;
 620                         goto exit;
 621                 }
 622                 if (dest->addrtype == TIPC_ADDR_NAME) {
 623                         tport->conn_type = dest->addr.name.name.type;
 624                         tport->conn_instance = dest->addr.name.name.instance;
 625                 }
 626
 627                 /* Abort any pending connection attempts (very unlikely) */
 628                 reject_rx_queue(sk);
 629         }
 630
 631         timeout_val = sock_sndtimeo(sk, m->msg_flags & MSG_DONTWAIT);
 632
 633         do {
 634                 if (dest->addrtype == TIPC_ADDR_NAME) {
 635                         res = dest_name_check(dest, m);
 636                         if (res)
 637                                 break;
 638                         res = tipc_send2name(tport->ref,
 639                                              &dest->addr.name.name,
 640                                              dest->addr.name.domain,
 641                                              m->msg_iov,
 642                                              total_len);
 643                 } else if (dest->addrtype == TIPC_ADDR_ID) {
 644                         res = tipc_send2port(tport->ref,
 645                                              &dest->addr.id,
 646                                              m->msg_iov,
 647                                              total_len);
 648                 } else if (dest->addrtype == TIPC_ADDR_MCAST) {
 649                         if (needs_conn) {
 650                                 res = -EOPNOTSUPP;
 651                                 break;
 652                         }
 653                         res = dest_name_check(dest, m);
 654                         if (res)
 655                                 break;
 656                         res = tipc_multicast(tport->ref,
 657                                              &dest->addr.nameseq,
 658                                              m->msg_iov,
 659                                              total_len);
 660                 }
 661                 if (likely(res != -ELINKCONG)) {
 662                         if (needs_conn && (res >= 0))
 663                                 sock->state = SS_CONNECTING;
 664                         break;
 665                 }
 666                 if (timeout_val <= 0L) {
 667                         res = timeout_val ? timeout_val : -EWOULDBLOCK;
 668                         break;
 669                 }
 670                 release_sock(sk);
 671                 timeout_val = wait_event_interruptible_timeout(*sk_sleep(sk),
 672                                                !tport->congested, timeout_val);
 673                 lock_sock(sk);
 674         } while (1);
 675
 676 exit:
 677         if (iocb)
 678                 release_sock(sk);
 679         return res;
 680 }
 681
 682 /**
 683  * send_packet - send a connection-oriented message
 684  * @iocb: if NULL, indicates that socket lock is already held
 685  * @sock: socket structure
 686  * @m: message to send
 687  * @total_len: length of message
 688  *
 689  * Used for SOCK_SEQPACKET messages and SOCK_STREAM data.
 690  *
 691  * Returns the number of bytes sent on success, or errno otherwise
 692  */
 693 static int send_packet(struct kiocb *iocb, struct socket *sock,
 694                        struct msghdr *m, size_t total_len)
 695 {
 696         struct sock *sk = sock->sk;
 697         struct tipc_port *tport = tipc_sk_port(sk);
 698         struct sockaddr_tipc *dest = (struct sockaddr_tipc *)m->msg_name;
 699         long timeout_val;
 700         int res;
 701
 702         /* Handle implied connection establishment */
 703         if (unlikely(dest))
 704                 return send_msg(iocb, sock, m, total_len);
 705
 706         if (total_len > TIPC_MAX_USER_MSG_SIZE)
 707                 return -EMSGSIZE;
 708
 709         if (iocb)
 710                 lock_sock(sk);
 711
 712         timeout_val = sock_sndtimeo(sk, m->msg_flags & MSG_DONTWAIT);
 713
 714         do {
 715                 if (unlikely(sock->state != SS_CONNECTED)) {
 716                         if (sock->state == SS_DISCONNECTING)
 717                                 res = -EPIPE;
 718                         else
 719                                 res = -ENOTCONN;
 720                         break;
 721                 }
 722
 723                 res = tipc_send(tport->ref, m->msg_iov, total_len);
 724                 if (likely(res != -ELINKCONG))
 725                         break;
 726                 if (timeout_val <= 0L) {
 727                         res = timeout_val ? timeout_val : -EWOULDBLOCK;
 728                         break;
 729                 }
 730                 release_sock(sk);
 731                 timeout_val = wait_event_interruptible_timeout(*sk_sleep(sk),
 732                         (!tport->congested || !tport->connected), timeout_val);
 733                 lock_sock(sk);
 734         } while (1);
 735
 736         if (iocb)
 737                 release_sock(sk);
 738         return res;
 739 }
 740
 741 /**
 742  * send_stream - send stream-oriented data
 743  * @iocb: (unused)
 744  * @sock: socket structure
 745  * @m: data to send
 746  * @total_len: total length of data to be sent
 747  *
 748  * Used for SOCK_STREAM data.
 749  *
 750  * Returns the number of bytes sent on success (or partial success),
 751  * or errno if no data sent
 752  */
 753 static int send_stream(struct kiocb *iocb, struct socket *sock,
 754                        struct msghdr *m, size_t total_len)
 755 {
 756         struct sock *sk = sock->sk;
 757         struct tipc_port *tport = tipc_sk_port(sk);
 758         struct msghdr my_msg;
 759         struct iovec my_iov;
 760         struct iovec *curr_iov;
 761         int curr_iovlen;
 762         char __user *curr_start;
 763         u32 hdr_size;
 764         int curr_left;
 765         int bytes_to_send;
 766         int bytes_sent;
 767         int res;
 768
 769         lock_sock(sk);
 770
 771         /* Handle special cases where there is no connection */
 772         if (unlikely(sock->state != SS_CONNECTED)) {
 773                 if (sock->state == SS_UNCONNECTED) {
 774                         res = send_packet(NULL, sock, m, total_len);
 775                         goto exit;
 776                 } else if (sock->state == SS_DISCONNECTING) {
 777                         res = -EPIPE;
 778                         goto exit;
 779                 } else {
 780                         res = -ENOTCONN;
 781                         goto exit;
 782                 }
 783         }
 784
 785         if (unlikely(m->msg_name)) {
 786                 res = -EISCONN;
 787                 goto exit;
 788         }
 789
 790         if (total_len > (unsigned int)INT_MAX) {
 791                 res = -EMSGSIZE;
 792                 goto exit;
 793         }
 794
 795         /*
 796          * Send each iovec entry using one or more messages
 797          *
 798          * Note: This algorithm is good for the most likely case
 799          * (i.e. one large iovec entry), but could be improved to pass sets
 800          * of small iovec entries into send_packet().
 801          */
 802         curr_iov = m->msg_iov;
 803         curr_iovlen = m->msg_iovlen;
 804         my_msg.msg_iov = &my_iov;
 805         my_msg.msg_iovlen = 1;
 806         my_msg.msg_flags = m->msg_flags;
 807         my_msg.msg_name = NULL;
 808         bytes_sent = 0;
 809
 810         hdr_size = msg_hdr_sz(&tport->phdr);
 811
 812         while (curr_iovlen--) {
 813                 curr_start = curr_iov->iov_base;
 814                 curr_left = curr_iov->iov_len;
 815
 816                 while (curr_left) {
 817                         bytes_to_send = tport->max_pkt - hdr_size;
 818                         if (bytes_to_send > TIPC_MAX_USER_MSG_SIZE)
 819                                 bytes_to_send = TIPC_MAX_USER_MSG_SIZE;
 820                         if (curr_left < bytes_to_send)
 821                                 bytes_to_send = curr_left;
 822                         my_iov.iov_base = curr_start;
 823                         my_iov.iov_len = bytes_to_send;
 824                         res = send_packet(NULL, sock, &my_msg, bytes_to_send);
 825                         if (res < 0) {
 826                                 if (bytes_sent)
 827                                         res = bytes_sent;
 828                                 goto exit;
 829                         }
 830                         curr_left -= bytes_to_send;
 831                         curr_start += bytes_to_send;
 832                         bytes_sent += bytes_to_send;
 833                 }
 834
 835                 curr_iov++;
 836         }
 837         res = bytes_sent;
 838 exit:
 839         release_sock(sk);
 840         return res;
 841 }
 842
 843 /**
 844  * auto_connect - complete connection setup to a remote port
 845  * @sock: socket structure
 846  * @msg: peer's response message
 847  *
 848  * Returns 0 on success, errno otherwise
 849  */
 850 static int auto_connect(struct socket *sock, struct tipc_msg *msg)
 851 {
 852         struct tipc_sock *tsock = tipc_sk(sock->sk);
 853         struct tipc_port *p_ptr;
 854
 855         tsock->peer_name.ref = msg_origport(msg);
 856         tsock->peer_name.node = msg_orignode(msg);
 857         p_ptr = tipc_port_deref(tsock->p->ref);
 858         if (!p_ptr)
 859                 return -EINVAL;
 860
 861         __tipc_connect(tsock->p->ref, p_ptr, &tsock->peer_name);
 862
 863         if (msg_importance(msg) > TIPC_CRITICAL_IMPORTANCE)
 864                 return -EINVAL;
 865         msg_set_importance(&p_ptr->phdr, (u32)msg_importance(msg));
 866         sock->state = SS_CONNECTED;
 867         return 0;
 868 }
 869
 870 /**
 871  * set_orig_addr - capture sender's address for received message
 872  * @m: descriptor for message info
 873  * @msg: received message header
 874  *
 875  * Note: Address is not captured if not requested by receiver.
 876  */
 877 static void set_orig_addr(struct msghdr *m, struct tipc_msg *msg)
 878 {
 879         struct sockaddr_tipc *addr = (struct sockaddr_tipc *)m->msg_name;
 880
 881         if (addr) {
 882                 addr->family = AF_TIPC;
 883                 addr->addrtype = TIPC_ADDR_ID;
 884                 memset(&addr->addr, 0, sizeof(addr->addr));
 885                 addr->addr.id.ref = msg_origport(msg);
 886                 addr->addr.id.node = msg_orignode(msg);
 887                 addr->addr.name.domain = 0;     /* could leave uninitialized */
 888                 addr->scope = 0;                /* could leave uninitialized */
 889                 m->msg_namelen = sizeof(struct sockaddr_tipc);
 890         }
 891 }
 892
 893 /**
 894  * anc_data_recv - optionally capture ancillary data for received message
 895  * @m: descriptor for message info
 896  * @msg: received message header
 897  * @tport: TIPC port associated with message
 898  *
 899  * Note: Ancillary data is not captured if not requested by receiver.
 900  *
 901  * Returns 0 if successful, otherwise errno
 902  */
 903 static int anc_data_recv(struct msghdr *m, struct tipc_msg *msg,
 904                          struct tipc_port *tport)
 905 {
 906         u32 anc_data[3];
 907         u32 err;
 908         u32 dest_type;
 909         int has_name;
 910         int res;
 911
 912         if (likely(m->msg_controllen == 0))
 913                 return 0;
 914
 915         /* Optionally capture errored message object(s) */
 916         err = msg ? msg_errcode(msg) : 0;
 917         if (unlikely(err)) {
 918                 anc_data[0] = err;
 919                 anc_data[1] = msg_data_sz(msg);
 920                 res = put_cmsg(m, SOL_TIPC, TIPC_ERRINFO, 8, anc_data);
 921                 if (res)
 922                         return res;
 923                 if (anc_data[1]) {
 924                         res = put_cmsg(m, SOL_TIPC, TIPC_RETDATA, anc_data[1],
 925                                        msg_data(msg));
 926                         if (res)
 927                                 return res;
 928                 }
 929         }
 930
 931         /* Optionally capture message destination object */
 932         dest_type = msg ? msg_type(msg) : TIPC_DIRECT_MSG;
 933         switch (dest_type) {
 934         case TIPC_NAMED_MSG:
 935                 has_name = 1;
 936                 anc_data[0] = msg_nametype(msg);
 937                 anc_data[1] = msg_namelower(msg);
 938                 anc_data[2] = msg_namelower(msg);
 939                 break;
 940         case TIPC_MCAST_MSG:
 941                 has_name = 1;
 942                 anc_data[0] = msg_nametype(msg);
 943                 anc_data[1] = msg_namelower(msg);
 944                 anc_data[2] = msg_nameupper(msg);
 945                 break;
 946         case TIPC_CONN_MSG:
 947                 has_name = (tport->conn_type != 0);
 948                 anc_data[0] = tport->conn_type;
 949                 anc_data[1] = tport->conn_instance;
 950                 anc_data[2] = tport->conn_instance;
 951                 break;
 952         default:
 953                 has_name = 0;
 954         }
 955         if (has_name) {
 956                 res = put_cmsg(m, SOL_TIPC, TIPC_DESTNAME, 12, anc_data);
 957                 if (res)
 958                         return res;
 959         }
 960
 961         return 0;
 962 }
 963
 964 /**
 965  * recv_msg - receive packet-oriented message
 966  * @iocb: (unused)
 967  * @m: descriptor for message info
 968  * @buf_len: total size of user buffer area
 969  * @flags: receive flags
 970  *
 971  * Used for SOCK_DGRAM, SOCK_RDM, and SOCK_SEQPACKET messages.
 972  * If the complete message doesn't fit in user area, truncate it.
 973  *
 974  * Returns size of returned message data, errno otherwise
 975  */
 976 static int recv_msg(struct kiocb *iocb, struct socket *sock,
 977                     struct msghdr *m, size_t buf_len, int flags)
 978 {
 979         struct sock *sk = sock->sk;
 980         struct tipc_port *tport = tipc_sk_port(sk);
 981         struct sk_buff *buf;
 982         struct tipc_msg *msg;
 983         long timeout;
 984         unsigned int sz;
 985         u32 err;
 986         int res;
 987
 988         /* Catch invalid receive requests */
 989         if (unlikely(!buf_len))
 990                 return -EINVAL;
 991
 992         lock_sock(sk);
 993
 994         if (unlikely(sock->state == SS_UNCONNECTED)) {
 995                 res = -ENOTCONN;
 996                 goto exit;
 997         }
 998
 999         timeout = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
1000 restart:
1001
1002         /* Look for a message in receive queue; wait if necessary */
1003         while (skb_queue_empty(&sk->sk_receive_queue)) {
1004                 if (sock->state == SS_DISCONNECTING) {
1005                         res = -ENOTCONN;
1006                         goto exit;
1007                 }
1008                 if (timeout <= 0L) {
1009                         res = timeout ? timeout : -EWOULDBLOCK;
1010                         goto exit;
1011                 }
1012                 release_sock(sk);
1013                 timeout = wait_event_interruptible_timeout(*sk_sleep(sk),
1014                                                            tipc_rx_ready(sock),
1015                                                            timeout);
1016                 lock_sock(sk);
1017         }
1018
1019         /* Look at first message in receive queue */
1020         buf = skb_peek(&sk->sk_receive_queue);
1021         msg = buf_msg(buf);
1022         sz = msg_data_sz(msg);
1023         err = msg_errcode(msg);
1024
1025         /* Discard an empty non-errored message & try again */
1026         if ((!sz) && (!err)) {
1027                 advance_rx_queue(sk);
1028                 goto restart;
1029         }
1030
1031         /* Capture sender's address (optional) */
1032         set_orig_addr(m, msg);
1033
1034         /* Capture ancillary data (optional) */
1035         res = anc_data_recv(m, msg, tport);
1036         if (res)
1037                 goto exit;
1038
1039         /* Capture message data (if valid) & compute return value (always) */
1040         if (!err) {
1041                 if (unlikely(buf_len < sz)) {
1042                         sz = buf_len;
1043                         m->msg_flags |= MSG_TRUNC;
1044                 }
1045                 res = skb_copy_datagram_iovec(buf, msg_hdr_sz(msg),
1046                                               m->msg_iov, sz);
1047                 if (res)
1048                         goto exit;
1049                 res = sz;
1050         } else {
1051                 if ((sock->state == SS_READY) ||
1052                     ((err == TIPC_CONN_SHUTDOWN) || m->msg_control))
1053                         res = 0;
1054                 else
1055                         res = -ECONNRESET;
1056         }
1057
1058         /* Consume received message (optional) */
1059         if (likely(!(flags & MSG_PEEK))) {
1060                 if ((sock->state != SS_READY) &&
1061                     (++tport->conn_unacked >= TIPC_FLOW_CONTROL_WIN))
1062                         tipc_acknowledge(tport->ref, tport->conn_unacked);
1063                 advance_rx_queue(sk);
1064         }
1065 exit:
1066         release_sock(sk);
1067         return res;
1068 }
1069
1070 /**
1071  * recv_stream - receive stream-oriented data
1072  * @iocb: (unused)
1073  * @m: descriptor for message info
1074  * @buf_len: total size of user buffer area
1075  * @flags: receive flags
1076  *
1077  * Used for SOCK_STREAM messages only.  If not enough data is available
1078  * will optionally wait for more; never truncates data.
1079  *
1080  * Returns size of returned message data, errno otherwise
1081  */
1082 static int recv_stream(struct kiocb *iocb, struct socket *sock,
1083                        struct msghdr *m, size_t buf_len, int flags)
1084 {
1085         struct sock *sk = sock->sk;
1086         struct tipc_port *tport = tipc_sk_port(sk);
1087         struct sk_buff *buf;
1088         struct tipc_msg *msg;
1089         long timeout;
1090         unsigned int sz;
1091         int sz_to_copy, target, needed;
1092         int sz_copied = 0;
1093         u32 err;
1094         int res = 0;
1095
1096         /* Catch invalid receive attempts */
1097         if (unlikely(!buf_len))
1098                 return -EINVAL;
1099
1100         lock_sock(sk);
1101
1102         if (unlikely((sock->state == SS_UNCONNECTED))) {
1103                 res = -ENOTCONN;
1104                 goto exit;
1105         }
1106
1107         target = sock_rcvlowat(sk, flags & MSG_WAITALL, buf_len);
1108         timeout = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
1109
1110 restart:
1111         /* Look for a message in receive queue; wait if necessary */
1112         while (skb_queue_empty(&sk->sk_receive_queue)) {
1113                 if (sock->state == SS_DISCONNECTING) {
1114                         res = -ENOTCONN;
1115                         goto exit;
1116                 }
1117                 if (timeout <= 0L) {
1118                         res = timeout ? timeout : -EWOULDBLOCK;
1119                         goto exit;
1120                 }
1121                 release_sock(sk);
1122                 timeout = wait_event_interruptible_timeout(*sk_sleep(sk),
1123                                                            tipc_rx_ready(sock),
1124                                                            timeout);
1125                 lock_sock(sk);
1126         }
1127
1128         /* Look at first message in receive queue */
1129         buf = skb_peek(&sk->sk_receive_queue);
1130         msg = buf_msg(buf);
1131         sz = msg_data_sz(msg);
1132         err = msg_errcode(msg);
1133
1134         /* Discard an empty non-errored message & try again */
1135         if ((!sz) && (!err)) {
1136                 advance_rx_queue(sk);
1137                 goto restart;
1138         }
1139
1140         /* Optionally capture sender's address & ancillary data of first msg */
1141         if (sz_copied == 0) {
1142                 set_orig_addr(m, msg);
1143                 res = anc_data_recv(m, msg, tport);
1144                 if (res)
1145                         goto exit;
1146         }
1147
1148         /* Capture message data (if valid) & compute return value (always) */
1149         if (!err) {
1150                 u32 offset = (u32)(unsigned long)(TIPC_SKB_CB(buf)->handle);
1151
1152                 sz -= offset;
1153                 needed = (buf_len - sz_copied);
1154                 sz_to_copy = (sz <= needed) ? sz : needed;
1155
1156                 res = skb_copy_datagram_iovec(buf, msg_hdr_sz(msg) + offset,
1157                                               m->msg_iov, sz_to_copy);
1158                 if (res)
1159                         goto exit;
1160
1161                 sz_copied += sz_to_copy;
1162
1163                 if (sz_to_copy < sz) {
1164                         if (!(flags & MSG_PEEK))
1165                                 TIPC_SKB_CB(buf)->handle =
1166                                 (void *)(unsigned long)(offset + sz_to_copy);
1167                         goto exit;
1168                 }
1169         } else {
1170                 if (sz_copied != 0)
1171                         goto exit; /* can't add error msg to valid data */
1172
1173                 if ((err == TIPC_CONN_SHUTDOWN) || m->msg_control)
1174                         res = 0;
1175                 else
1176                         res = -ECONNRESET;
1177         }
1178
1179         /* Consume received message (optional) */
1180         if (likely(!(flags & MSG_PEEK))) {
1181                 if (unlikely(++tport->conn_unacked >= TIPC_FLOW_CONTROL_WIN))
1182                         tipc_acknowledge(tport->ref, tport->conn_unacked);
1183                 advance_rx_queue(sk);
1184         }
1185
1186         /* Loop around if more data is required */
1187         if ((sz_copied < buf_len) &&    /* didn't get all requested data */
1188             (!skb_queue_empty(&sk->sk_receive_queue) ||
1189             (sz_copied < target)) &&    /* and more is ready or required */
1190             (!(flags & MSG_PEEK)) &&    /* and aren't just peeking at data */
1191             (!err))                     /* and haven't reached a FIN */
1192                 goto restart;
1193
1194 exit:
1195         release_sock(sk);
1196         return sz_copied ? sz_copied : res;
1197 }
1198
1199 /**
1200  * tipc_write_space - wake up thread if port congestion is released
1201  * @sk: socket
1202  */
1203 static void tipc_write_space(struct sock *sk)
1204 {
1205         struct socket_wq *wq;
1206
1207         rcu_read_lock();
1208         wq = rcu_dereference(sk->sk_wq);
1209         if (wq_has_sleeper(wq))
1210                 wake_up_interruptible_sync_poll(&wq->wait, POLLOUT |
1211                                                 POLLWRNORM | POLLWRBAND);
1212         rcu_read_unlock();
1213 }
1214
1215 /**
1216  * tipc_data_ready - wake up threads to indicate messages have been received
1217  * @sk: socket
1218  * @len: the length of messages
1219  */
1220 static void tipc_data_ready(struct sock *sk, int len)
1221 {
1222         struct socket_wq *wq;
1223
1224         rcu_read_lock();
1225         wq = rcu_dereference(sk->sk_wq);
1226         if (wq_has_sleeper(wq))
1227                 wake_up_interruptible_sync_poll(&wq->wait, POLLIN |
1228                                                 POLLRDNORM | POLLRDBAND);
1229         rcu_read_unlock();
1230 }
1231
1232 /**
1233  * filter_connect - Handle all incoming messages for a connection-based socket
1234  * @tsock: TIPC socket
1235  * @msg: message
1236  *
1237  * Returns TIPC error status code and socket error status code
1238  * once it encounters some errors
1239  */
1240 static u32 filter_connect(struct tipc_sock *tsock, struct sk_buff **buf)
1241 {
1242         struct socket *sock = tsock->sk.sk_socket;
1243         struct tipc_msg *msg = buf_msg(*buf);
1244         struct sock *sk = &tsock->sk;
1245         u32 retval = TIPC_ERR_NO_PORT;
1246         int res;
1247
1248         if (msg_mcast(msg))
1249                 return retval;
1250
1251         switch ((int)sock->state) {
1252         case SS_CONNECTED:
1253                 /* Accept only connection-based messages sent by peer */
1254                 if (msg_connected(msg) && tipc_port_peer_msg(tsock->p, msg)) {
1255                         if (unlikely(msg_errcode(msg))) {
1256                                 sock->state = SS_DISCONNECTING;
1257                                 __tipc_disconnect(tsock->p);
1258                         }
1259                         retval = TIPC_OK;
1260                 }
1261                 break;
1262         case SS_CONNECTING:
1263                 /* Accept only ACK or NACK message */
1264                 if (unlikely(msg_errcode(msg))) {
1265                         sock->state = SS_DISCONNECTING;
1266                         sk->sk_err = ECONNREFUSED;
1267                         retval = TIPC_OK;
1268                         break;
1269                 }
1270
1271                 if (unlikely(!msg_connected(msg)))
1272                         break;
1273
1274                 res = auto_connect(sock, msg);
1275                 if (res) {
1276                         sock->state = SS_DISCONNECTING;
1277                         sk->sk_err = -res;
1278                         retval = TIPC_OK;
1279                         break;
1280                 }
1281
1282                 /* If an incoming message is an 'ACK-', it should be
1283                  * discarded here because it doesn't contain useful
1284                  * data. In addition, we should try to wake up
1285                  * connect() routine if sleeping.
1286                  */
1287                 if (msg_data_sz(msg) == 0) {
1288                         kfree_skb(*buf);
1289                         *buf = NULL;
1290                         if (waitqueue_active(sk_sleep(sk)))
1291                                 wake_up_interruptible(sk_sleep(sk));
1292                 }
1293                 retval = TIPC_OK;
1294                 break;
1295         case SS_LISTENING:
1296         case SS_UNCONNECTED:
1297                 /* Accept only SYN message */
1298                 if (!msg_connected(msg) && !(msg_errcode(msg)))
1299                         retval = TIPC_OK;
1300                 break;
1301         case SS_DISCONNECTING:
1302                 break;
1303         default:
1304                 pr_err("Unknown socket state %u\n", sock->state);
1305         }
1306         return retval;
1307 }
1308
1309 /**
1310  * rcvbuf_limit - get proper overload limit of socket receive queue
1311  * @sk: socket
1312  * @buf: message
1313  *
1314  * For all connection oriented messages, irrespective of importance,
1315  * the default overload value (i.e. 67MB) is set as limit.
1316  *
1317  * For all connectionless messages, by default new queue limits are
1318  * as belows:
1319  *
1320  * TIPC_LOW_IMPORTANCE       (4 MB)
1321  * TIPC_MEDIUM_IMPORTANCE    (8 MB)
1322  * TIPC_HIGH_IMPORTANCE      (16 MB)
1323  * TIPC_CRITICAL_IMPORTANCE  (32 MB)
1324  *
1325  * Returns overload limit according to corresponding message importance
1326  */
1327 static unsigned int rcvbuf_limit(struct sock *sk, struct sk_buff *buf)
1328 {
1329         struct tipc_msg *msg = buf_msg(buf);
1330         unsigned int limit;
1331
1332         if (msg_connected(msg))
1333                 limit = sysctl_tipc_rmem[2];
1334         else
1335                 limit = sk->sk_rcvbuf >> TIPC_CRITICAL_IMPORTANCE <<
1336                         msg_importance(msg);
1337         return limit;
1338 }
1339
1340 /**
1341  * filter_rcv - validate incoming message
1342  * @sk: socket
1343  * @buf: message
1344  *
1345  * Enqueues message on receive queue if acceptable; optionally handles
1346  * disconnect indication for a connected socket.
1347  *
1348  * Called with socket lock already taken; port lock may also be taken.
1349  *
1350  * Returns TIPC error status code (TIPC_OK if message is not to be rejected)
1351  */
1352 static u32 filter_rcv(struct sock *sk, struct sk_buff *buf)
1353 {
1354         struct socket *sock = sk->sk_socket;
1355         struct tipc_msg *msg = buf_msg(buf);
1356         unsigned int limit = rcvbuf_limit(sk, buf);
1357         u32 res = TIPC_OK;
1358
1359         /* Reject message if it is wrong sort of message for socket */
1360         if (msg_type(msg) > TIPC_DIRECT_MSG)
1361                 return TIPC_ERR_NO_PORT;
1362
1363         if (sock->state == SS_READY) {
1364                 if (msg_connected(msg))
1365                         return TIPC_ERR_NO_PORT;
1366         } else {
1367                 res = filter_connect(tipc_sk(sk), &buf);
1368                 if (res != TIPC_OK || buf == NULL)
1369                         return res;
1370         }
1371
1372         /* Reject message if there isn't room to queue it */
1373         if (sk_rmem_alloc_get(sk) + buf->truesize >= limit)
1374                 return TIPC_ERR_OVERLOAD;
1375
1376         /* Enqueue message */
1377         TIPC_SKB_CB(buf)->handle = NULL;
1378         __skb_queue_tail(&sk->sk_receive_queue, buf);
1379         skb_set_owner_r(buf, sk);
1380
1381         sk->sk_data_ready(sk, 0);
1382         return TIPC_OK;
1383 }
1384
1385 /**
1386  * backlog_rcv - handle incoming message from backlog queue
1387  * @sk: socket
1388  * @buf: message
1389  *
1390  * Caller must hold socket lock, but not port lock.
1391  *
1392  * Returns 0
1393  */
1394 static int backlog_rcv(struct sock *sk, struct sk_buff *buf)
1395 {
1396         u32 res;
1397
1398         res = filter_rcv(sk, buf);
1399         if (res)
1400                 tipc_reject_msg(buf, res);
1401         return 0;
1402 }
1403
1404 /**
1405  * dispatch - handle incoming message
1406  * @tport: TIPC port that received message
1407  * @buf: message
1408  *
1409  * Called with port lock already taken.
1410  *
1411  * Returns TIPC error status code (TIPC_OK if message is not to be rejected)
1412  */
1413 static u32 dispatch(struct tipc_port *tport, struct sk_buff *buf)
1414 {
1415         struct sock *sk = tport->sk;
1416         u32 res;
1417
1418         /*
1419          * Process message if socket is unlocked; otherwise add to backlog queue
1420          *
1421          * This code is based on sk_receive_skb(), but must be distinct from it
1422          * since a TIPC-specific filter/reject mechanism is utilized
1423          */
1424         bh_lock_sock(sk);
1425         if (!sock_owned_by_user(sk)) {
1426                 res = filter_rcv(sk, buf);
1427         } else {
1428                 if (sk_add_backlog(sk, buf, rcvbuf_limit(sk, buf)))
1429                         res = TIPC_ERR_OVERLOAD;
1430                 else
1431                         res = TIPC_OK;
1432         }
1433         bh_unlock_sock(sk);
1434
1435         return res;
1436 }
1437
1438 /**
1439  * wakeupdispatch - wake up port after congestion
1440  * @tport: port to wakeup
1441  *
1442  * Called with port lock already taken.
1443  */
1444 static void wakeupdispatch(struct tipc_port *tport)
1445 {
1446         struct sock *sk = tport->sk;
1447
1448         sk->sk_write_space(sk);
1449 }
1450
1451 /**
1452  * connect - establish a connection to another TIPC port
1453  * @sock: socket structure
1454  * @dest: socket address for destination port
1455  * @destlen: size of socket address data structure
1456  * @flags: file-related flags associated with socket
1457  *
1458  * Returns 0 on success, errno otherwise
1459  */
1460 static int connect(struct socket *sock, struct sockaddr *dest, int destlen,
1461                    int flags)
1462 {
1463         struct sock *sk = sock->sk;
1464         struct sockaddr_tipc *dst = (struct sockaddr_tipc *)dest;
1465         struct msghdr m = {NULL,};
1466         unsigned int timeout;
1467         int res;
1468
1469         lock_sock(sk);
1470
1471         /* For now, TIPC does not allow use of connect() with DGRAM/RDM types */
1472         if (sock->state == SS_READY) {
1473                 res = -EOPNOTSUPP;
1474                 goto exit;
1475         }
1476
1477         /*
1478          * Reject connection attempt using multicast address
1479          *
1480          * Note: send_msg() validates the rest of the address fields,
1481          *       so there's no need to do it here
1482          */
1483         if (dst->addrtype == TIPC_ADDR_MCAST) {
1484                 res = -EINVAL;
1485                 goto exit;
1486         }
1487
1488         timeout = (flags & O_NONBLOCK) ? 0 : tipc_sk(sk)->conn_timeout;
1489
1490         switch (sock->state) {
1491         case SS_UNCONNECTED:
1492                 /* Send a 'SYN-' to destination */
1493                 m.msg_name = dest;
1494                 m.msg_namelen = destlen;
1495
1496                 /* If connect is in non-blocking case, set MSG_DONTWAIT to
1497                  * indicate send_msg() is never blocked.
1498                  */
1499                 if (!timeout)
1500                         m.msg_flags = MSG_DONTWAIT;
1501
1502                 res = send_msg(NULL, sock, &m, 0);
1503                 if ((res < 0) && (res != -EWOULDBLOCK))
1504                         goto exit;
1505
1506                 /* Just entered SS_CONNECTING state; the only
1507                  * difference is that return value in non-blocking
1508                  * case is EINPROGRESS, rather than EALREADY.
1509                  */
1510                 res = -EINPROGRESS;
1511                 break;
1512         case SS_CONNECTING:
1513                 res = -EALREADY;
1514                 break;
1515         case SS_CONNECTED:
1516                 res = -EISCONN;
1517                 break;
1518         default:
1519                 res = -EINVAL;
1520                 goto exit;
1521         }
1522
1523         if (sock->state == SS_CONNECTING) {
1524                 if (!timeout)
1525                         goto exit;
1526
1527                 /* Wait until an 'ACK' or 'RST' arrives, or a timeout occurs */
1528                 release_sock(sk);
1529                 res = wait_event_interruptible_timeout(*sk_sleep(sk),
1530                                 sock->state != SS_CONNECTING,
1531                                 timeout ? (long)msecs_to_jiffies(timeout)
1532                                         : MAX_SCHEDULE_TIMEOUT);
1533                 lock_sock(sk);
1534                 if (res <= 0) {
1535                         if (res == 0)
1536                                 res = -ETIMEDOUT;
1537                         else
1538                                 ; /* leave "res" unchanged */
1539                         goto exit;
1540                 }
1541         }
1542
1543         if (unlikely(sock->state == SS_DISCONNECTING))
1544                 res = sock_error(sk);
1545         else
1546                 res = 0;
1547
1548 exit:
1549         release_sock(sk);
1550         return res;
1551 }
1552
1553 /**
1554  * listen - allow socket to listen for incoming connections
1555  * @sock: socket structure
1556  * @len: (unused)
1557  *
1558  * Returns 0 on success, errno otherwise
1559  */
1560 static int listen(struct socket *sock, int len)
1561 {
1562         struct sock *sk = sock->sk;
1563         int res;
1564
1565         lock_sock(sk);
1566
1567         if (sock->state != SS_UNCONNECTED)
1568                 res = -EINVAL;
1569         else {
1570                 sock->state = SS_LISTENING;
1571                 res = 0;
1572         }
1573
1574         release_sock(sk);
1575         return res;
1576 }
1577
1578 /**
1579  * accept - wait for connection request
1580  * @sock: listening socket
1581  * @newsock: new socket that is to be connected
1582  * @flags: file-related flags associated with socket
1583  *
1584  * Returns 0 on success, errno otherwise
1585  */
1586 static int accept(struct socket *sock, struct socket *new_sock, int flags)
1587 {
1588         struct sock *new_sk, *sk = sock->sk;
1589         struct sk_buff *buf;
1590         struct tipc_sock *new_tsock;
1591         struct tipc_port *new_tport;
1592         struct tipc_msg *msg;
1593         u32 new_ref;
1594
1595         int res;
1596
1597         lock_sock(sk);
1598
1599         if (sock->state != SS_LISTENING) {
1600                 res = -EINVAL;
1601                 goto exit;
1602         }
1603
1604         while (skb_queue_empty(&sk->sk_receive_queue)) {
1605                 if (flags & O_NONBLOCK) {
1606                         res = -EWOULDBLOCK;
1607                         goto exit;
1608                 }
1609                 release_sock(sk);
1610                 res = wait_event_interruptible(*sk_sleep(sk),
1611                                 (!skb_queue_empty(&sk->sk_receive_queue)));
1612                 lock_sock(sk);
1613                 if (res)
1614                         goto exit;
1615         }
1616
1617         buf = skb_peek(&sk->sk_receive_queue);
1618
1619         res = tipc_sk_create(sock_net(sock->sk), new_sock, 0, 1);
1620         if (res)
1621                 goto exit;
1622
1623         new_sk = new_sock->sk;
1624         new_tsock = tipc_sk(new_sk);
1625         new_tport = new_tsock->p;
1626         new_ref = new_tport->ref;
1627         msg = buf_msg(buf);
1628
1629         /* we lock on new_sk; but lockdep sees the lock on sk */
1630         lock_sock_nested(new_sk, SINGLE_DEPTH_NESTING);
1631
1632         /*
1633          * Reject any stray messages received by new socket
1634          * before the socket lock was taken (very, very unlikely)
1635          */
1636         reject_rx_queue(new_sk);
1637
1638         /* Connect new socket to it's peer */
1639         new_tsock->peer_name.ref = msg_origport(msg);
1640         new_tsock->peer_name.node = msg_orignode(msg);
1641         tipc_connect(new_ref, &new_tsock->peer_name);
1642         new_sock->state = SS_CONNECTED;
1643
1644         tipc_set_portimportance(new_ref, msg_importance(msg));
1645         if (msg_named(msg)) {
1646                 new_tport->conn_type = msg_nametype(msg);
1647                 new_tport->conn_instance = msg_nameinst(msg);
1648         }
1649
1650         /*
1651          * Respond to 'SYN-' by discarding it & returning 'ACK'-.
1652          * Respond to 'SYN+' by queuing it on new socket.
1653          */
1654         if (!msg_data_sz(msg)) {
1655                 struct msghdr m = {NULL,};
1656
1657                 advance_rx_queue(sk);
1658                 send_packet(NULL, new_sock, &m, 0);
1659         } else {
1660                 __skb_dequeue(&sk->sk_receive_queue);
1661                 __skb_queue_head(&new_sk->sk_receive_queue, buf);
1662                 skb_set_owner_r(buf, new_sk);
1663         }
1664         release_sock(new_sk);
1665
1666 exit:
1667         release_sock(sk);
1668         return res;
1669 }
1670
1671 /**
1672  * shutdown - shutdown socket connection
1673  * @sock: socket structure
1674  * @how: direction to close (must be SHUT_RDWR)
1675  *
1676  * Terminates connection (if necessary), then purges socket's receive queue.
1677  *
1678  * Returns 0 on success, errno otherwise
1679  */
1680 static int shutdown(struct socket *sock, int how)
1681 {
1682         struct sock *sk = sock->sk;
1683         struct tipc_port *tport = tipc_sk_port(sk);
1684         struct sk_buff *buf;
1685         int res;
1686
1687         if (how != SHUT_RDWR)
1688                 return -EINVAL;
1689
1690         lock_sock(sk);
1691
1692         switch (sock->state) {
1693         case SS_CONNECTING:
1694         case SS_CONNECTED:
1695
1696 restart:
1697                 /* Disconnect and send a 'FIN+' or 'FIN-' message to peer */
1698                 buf = __skb_dequeue(&sk->sk_receive_queue);
1699                 if (buf) {
1700                         if (TIPC_SKB_CB(buf)->handle != NULL) {
1701                                 kfree_skb(buf);
1702                                 goto restart;
1703                         }
1704                         tipc_disconnect(tport->ref);
1705                         tipc_reject_msg(buf, TIPC_CONN_SHUTDOWN);
1706                 } else {
1707                         tipc_shutdown(tport->ref);
1708                 }
1709
1710                 sock->state = SS_DISCONNECTING;
1711
1712                 /* fall through */
1713
1714         case SS_DISCONNECTING:
1715
1716                 /* Discard any unreceived messages */
1717                 __skb_queue_purge(&sk->sk_receive_queue);
1718
1719                 /* Wake up anyone sleeping in poll */
1720                 sk->sk_state_change(sk);
1721                 res = 0;
1722                 break;
1723
1724         default:
1725                 res = -ENOTCONN;
1726         }
1727
1728         release_sock(sk);
1729         return res;
1730 }
1731
1732 /**
1733  * setsockopt - set socket option
1734  * @sock: socket structure
1735  * @lvl: option level
1736  * @opt: option identifier
1737  * @ov: pointer to new option value
1738  * @ol: length of option value
1739  *
1740  * For stream sockets only, accepts and ignores all IPPROTO_TCP options
1741  * (to ease compatibility).
1742  *
1743  * Returns 0 on success, errno otherwise
1744  */
1745 static int setsockopt(struct socket *sock, int lvl, int opt, char __user *ov,
1746                       unsigned int ol)
1747 {
1748         struct sock *sk = sock->sk;
1749         struct tipc_port *tport = tipc_sk_port(sk);
1750         u32 value;
1751         int res;
1752
1753         if ((lvl == IPPROTO_TCP) && (sock->type == SOCK_STREAM))
1754                 return 0;
1755         if (lvl != SOL_TIPC)
1756                 return -ENOPROTOOPT;
1757         if (ol < sizeof(value))
1758                 return -EINVAL;
1759         res = get_user(value, (u32 __user *)ov);
1760         if (res)
1761                 return res;
1762
1763         lock_sock(sk);
1764
1765         switch (opt) {
1766         case TIPC_IMPORTANCE:
1767                 res = tipc_set_portimportance(tport->ref, value);
1768                 break;
1769         case TIPC_SRC_DROPPABLE:
1770                 if (sock->type != SOCK_STREAM)
1771                         res = tipc_set_portunreliable(tport->ref, value);
1772                 else
1773                         res = -ENOPROTOOPT;
1774                 break;
1775         case TIPC_DEST_DROPPABLE:
1776                 res = tipc_set_portunreturnable(tport->ref, value);
1777                 break;
1778         case TIPC_CONN_TIMEOUT:
1779                 tipc_sk(sk)->conn_timeout = value;
1780                 /* no need to set "res", since already 0 at this point */
1781                 break;
1782         default:
1783                 res = -EINVAL;
1784         }
1785
1786         release_sock(sk);
1787
1788         return res;
1789 }
1790
1791 /**
1792  * getsockopt - get socket option
1793  * @sock: socket structure
1794  * @lvl: option level
1795  * @opt: option identifier
1796  * @ov: receptacle for option value
1797  * @ol: receptacle for length of option value
1798  *
1799  * For stream sockets only, returns 0 length result for all IPPROTO_TCP options
1800  * (to ease compatibility).
1801  *
1802  * Returns 0 on success, errno otherwise
1803  */
1804 static int getsockopt(struct socket *sock, int lvl, int opt, char __user *ov,
1805                       int __user *ol)
1806 {
1807         struct sock *sk = sock->sk;
1808         struct tipc_port *tport = tipc_sk_port(sk);
1809         int len;
1810         u32 value;
1811         int res;
1812
1813         if ((lvl == IPPROTO_TCP) && (sock->type == SOCK_STREAM))
1814                 return put_user(0, ol);
1815         if (lvl != SOL_TIPC)
1816                 return -ENOPROTOOPT;
1817         res = get_user(len, ol);
1818         if (res)
1819                 return res;
1820
1821         lock_sock(sk);
1822
1823         switch (opt) {
1824         case TIPC_IMPORTANCE:
1825                 res = tipc_portimportance(tport->ref, &value);
1826                 break;
1827         case TIPC_SRC_DROPPABLE:
1828                 res = tipc_portunreliable(tport->ref, &value);
1829                 break;
1830         case TIPC_DEST_DROPPABLE:
1831                 res = tipc_portunreturnable(tport->ref, &value);
1832                 break;
1833         case TIPC_CONN_TIMEOUT:
1834                 value = tipc_sk(sk)->conn_timeout;
1835                 /* no need to set "res", since already 0 at this point */
1836                 break;
1837         case TIPC_NODE_RECVQ_DEPTH:
1838                 value = 0; /* was tipc_queue_size, now obsolete */
1839                 break;
1840         case TIPC_SOCK_RECVQ_DEPTH:
1841                 value = skb_queue_len(&sk->sk_receive_queue);
1842                 break;
1843         default:
1844                 res = -EINVAL;
1845         }
1846
1847         release_sock(sk);
1848
1849         if (res)
1850                 return res;     /* "get" failed */
1851
1852         if (len < sizeof(value))
1853                 return -EINVAL;
1854
1855         if (copy_to_user(ov, &value, sizeof(value)))
1856                 return -EFAULT;
1857
1858         return put_user(sizeof(value), ol);
1859 }
1860
1861 /* Protocol switches for the various types of TIPC sockets */
1862
1863 static const struct proto_ops msg_ops = {
1864         .owner          = THIS_MODULE,
1865         .family         = AF_TIPC,
1866         .release        = release,
1867         .bind           = bind,
1868         .connect        = connect,
1869         .socketpair     = sock_no_socketpair,
1870         .accept         = sock_no_accept,
1871         .getname        = get_name,
1872         .poll           = poll,
1873         .ioctl          = sock_no_ioctl,
1874         .listen         = sock_no_listen,
1875         .shutdown       = shutdown,
1876         .setsockopt     = setsockopt,
1877         .getsockopt     = getsockopt,
1878         .sendmsg        = send_msg,
1879         .recvmsg        = recv_msg,
1880         .mmap           = sock_no_mmap,
1881         .sendpage       = sock_no_sendpage
1882 };
1883
1884 static const struct proto_ops packet_ops = {
1885         .owner          = THIS_MODULE,
1886         .family         = AF_TIPC,
1887         .release        = release,
1888         .bind           = bind,
1889         .connect        = connect,
1890         .socketpair     = sock_no_socketpair,
1891         .accept         = accept,
1892         .getname        = get_name,
1893         .poll           = poll,
1894         .ioctl          = sock_no_ioctl,
1895         .listen         = listen,
1896         .shutdown       = shutdown,
1897         .setsockopt     = setsockopt,
1898         .getsockopt     = getsockopt,
1899         .sendmsg        = send_packet,
1900         .recvmsg        = recv_msg,
1901         .mmap           = sock_no_mmap,
1902         .sendpage       = sock_no_sendpage
1903 };
1904
1905 static const struct proto_ops stream_ops = {
1906         .owner          = THIS_MODULE,
1907         .family         = AF_TIPC,
1908         .release        = release,
1909         .bind           = bind,
1910         .connect        = connect,
1911         .socketpair     = sock_no_socketpair,
1912         .accept         = accept,
1913         .getname        = get_name,
1914         .poll           = poll,
1915         .ioctl          = sock_no_ioctl,
1916         .listen         = listen,
1917         .shutdown       = shutdown,
1918         .setsockopt     = setsockopt,
1919         .getsockopt     = getsockopt,
1920         .sendmsg        = send_stream,
1921         .recvmsg        = recv_stream,
1922         .mmap           = sock_no_mmap,
1923         .sendpage       = sock_no_sendpage
1924 };
1925
1926 static const struct net_proto_family tipc_family_ops = {
1927         .owner          = THIS_MODULE,
1928         .family         = AF_TIPC,
1929         .create         = tipc_sk_create
1930 };
1931
1932 static struct proto tipc_proto = {
1933         .name           = "TIPC",
1934         .owner          = THIS_MODULE,
1935         .obj_size       = sizeof(struct tipc_sock),
1936         .sysctl_rmem    = sysctl_tipc_rmem
1937 };
1938
1939 static struct proto tipc_proto_kern = {
1940         .name           = "TIPC",
1941         .obj_size       = sizeof(struct tipc_sock),
1942         .sysctl_rmem    = sysctl_tipc_rmem
1943 };
1944
1945 /**
1946  * tipc_socket_init - initialize TIPC socket interface
1947  *
1948  * Returns 0 on success, errno otherwise
1949  */
1950 int tipc_socket_init(void)
1951 {
1952         int res;
1953
1954         res = proto_register(&tipc_proto, 1);
1955         if (res) {
1956                 pr_err("Failed to register TIPC protocol type\n");
1957                 goto out;
1958         }
1959
1960         res = sock_register(&tipc_family_ops);
1961         if (res) {
1962                 pr_err("Failed to register TIPC socket type\n");
1963                 proto_unregister(&tipc_proto);
1964                 goto out;
1965         }
1966
1967         sockets_enabled = 1;
1968  out:
1969         return res;
1970 }
1971
1972 /**
1973  * tipc_socket_stop - stop TIPC socket interface
1974  */
1975 void tipc_socket_stop(void)
1976 {
1977         if (!sockets_enabled)
1978                 return;
1979
1980         sockets_enabled = 0;
1981         sock_unregister(tipc_family_ops.family);
1982         proto_unregister(&tipc_proto);
1983 }