net/tipc/socket.c

   1 /*
   2  * net/tipc/socket.c: TIPC socket API
   3  *
   4  * Copyright (c) 2001-2007, 2012-2016, 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 <linux/rhashtable.h>
  38 #include "core.h"
  39 #include "name_table.h"
  40 #include "node.h"
  41 #include "link.h"
  42 #include "name_distr.h"
  43 #include "socket.h"
  44 #include "bcast.h"
  45 #include "netlink.h"
  46
  47 #define CONN_TIMEOUT_DEFAULT    8000    /* default connect timeout = 8s */
  48 #define CONN_PROBING_INTERVAL   msecs_to_jiffies(3600000)  /* [ms] => 1 h */
  49 #define TIPC_FWD_MSG            1
  50 #define TIPC_MAX_PORT           0xffffffff
  51 #define TIPC_MIN_PORT           1
  52
  53 enum {
  54         TIPC_LISTEN = TCP_LISTEN,
  55         TIPC_ESTABLISHED = TCP_ESTABLISHED,
  56         TIPC_OPEN = TCP_CLOSE,
  57         TIPC_DISCONNECTING = TCP_CLOSE_WAIT,
  58         TIPC_CONNECTING = TCP_SYN_SENT,
  59 };
  60
  61 /**
  62  * struct tipc_sock - TIPC socket structure
  63  * @sk: socket - interacts with 'port' and with user via the socket API
  64  * @conn_type: TIPC type used when connection was established
  65  * @conn_instance: TIPC instance used when connection was established
  66  * @published: non-zero if port has one or more associated names
  67  * @max_pkt: maximum packet size "hint" used when building messages sent by port
  68  * @portid: unique port identity in TIPC socket hash table
  69  * @phdr: preformatted message header used when sending messages
  70  * #cong_links: list of congested links
  71  * @publications: list of publications for port
  72  * @blocking_link: address of the congested link we are currently sleeping on
  73  * @pub_count: total # of publications port has made during its lifetime
  74  * @probing_state:
  75  * @conn_timeout: the time we can wait for an unresponded setup request
  76  * @dupl_rcvcnt: number of bytes counted twice, in both backlog and rcv queue
  77  * @cong_link_cnt: number of congested links
  78  * @sent_unacked: # messages sent by socket, and not yet acked by peer
  79  * @rcv_unacked: # messages read by user, but not yet acked back to peer
  80  * @peer: 'connected' peer for dgram/rdm
  81  * @node: hash table node
  82  * @mc_method: cookie for use between socket and broadcast layer
  83  * @rcu: rcu struct for tipc_sock
  84  */
  85 struct tipc_sock {
  86         struct sock sk;
  87         u32 conn_type;
  88         u32 conn_instance;
  89         int published;
  90         u32 max_pkt;
  91         u32 portid;
  92         struct tipc_msg phdr;
  93         struct list_head cong_links;
  94         struct list_head publications;
  95         u32 pub_count;
  96         uint conn_timeout;
  97         atomic_t dupl_rcvcnt;
  98         bool probe_unacked;
  99         u16 cong_link_cnt;
 100         u16 snt_unacked;
 101         u16 snd_win;
 102         u16 peer_caps;
 103         u16 rcv_unacked;
 104         u16 rcv_win;
 105         struct sockaddr_tipc peer;
 106         struct rhash_head node;
 107         struct tipc_mc_method mc_method;
 108         struct rcu_head rcu;
 109 };
 110
 111 static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb);
 112 static void tipc_data_ready(struct sock *sk);
 113 static void tipc_write_space(struct sock *sk);
 114 static void tipc_sock_destruct(struct sock *sk);
 115 static int tipc_release(struct socket *sock);
 116 static int tipc_accept(struct socket *sock, struct socket *new_sock, int flags);
 117 static void tipc_sk_timeout(unsigned long data);
 118 static int tipc_sk_publish(struct tipc_sock *tsk, uint scope,
 119                            struct tipc_name_seq const *seq);
 120 static int tipc_sk_withdraw(struct tipc_sock *tsk, uint scope,
 121                             struct tipc_name_seq const *seq);
 122 static struct tipc_sock *tipc_sk_lookup(struct net *net, u32 portid);
 123 static int tipc_sk_insert(struct tipc_sock *tsk);
 124 static void tipc_sk_remove(struct tipc_sock *tsk);
 125 static int __tipc_sendstream(struct socket *sock, struct msghdr *m, size_t dsz);
 126 static int __tipc_sendmsg(struct socket *sock, struct msghdr *m, size_t dsz);
 127
 128 static const struct proto_ops packet_ops;
 129 static const struct proto_ops stream_ops;
 130 static const struct proto_ops msg_ops;
 131 static struct proto tipc_proto;
 132 static const struct rhashtable_params tsk_rht_params;
 133
 134 static u32 tsk_own_node(struct tipc_sock *tsk)
 135 {
 136         return msg_prevnode(&tsk->phdr);
 137 }
 138
 139 static u32 tsk_peer_node(struct tipc_sock *tsk)
 140 {
 141         return msg_destnode(&tsk->phdr);
 142 }
 143
 144 static u32 tsk_peer_port(struct tipc_sock *tsk)
 145 {
 146         return msg_destport(&tsk->phdr);
 147 }
 148
 149 static  bool tsk_unreliable(struct tipc_sock *tsk)
 150 {
 151         return msg_src_droppable(&tsk->phdr) != 0;
 152 }
 153
 154 static void tsk_set_unreliable(struct tipc_sock *tsk, bool unreliable)
 155 {
 156         msg_set_src_droppable(&tsk->phdr, unreliable ? 1 : 0);
 157 }
 158
 159 static bool tsk_unreturnable(struct tipc_sock *tsk)
 160 {
 161         return msg_dest_droppable(&tsk->phdr) != 0;
 162 }
 163
 164 static void tsk_set_unreturnable(struct tipc_sock *tsk, bool unreturnable)
 165 {
 166         msg_set_dest_droppable(&tsk->phdr, unreturnable ? 1 : 0);
 167 }
 168
 169 static int tsk_importance(struct tipc_sock *tsk)
 170 {
 171         return msg_importance(&tsk->phdr);
 172 }
 173
 174 static int tsk_set_importance(struct tipc_sock *tsk, int imp)
 175 {
 176         if (imp > TIPC_CRITICAL_IMPORTANCE)
 177                 return -EINVAL;
 178         msg_set_importance(&tsk->phdr, (u32)imp);
 179         return 0;
 180 }
 181
 182 static struct tipc_sock *tipc_sk(const struct sock *sk)
 183 {
 184         return container_of(sk, struct tipc_sock, sk);
 185 }
 186
 187 static bool tsk_conn_cong(struct tipc_sock *tsk)
 188 {
 189         return tsk->snt_unacked > tsk->snd_win;
 190 }
 191
 192 /* tsk_blocks(): translate a buffer size in bytes to number of
 193  * advertisable blocks, taking into account the ratio truesize(len)/len
 194  * We can trust that this ratio is always < 4 for len >= FLOWCTL_BLK_SZ
 195  */
 196 static u16 tsk_adv_blocks(int len)
 197 {
 198         return len / FLOWCTL_BLK_SZ / 4;
 199 }
 200
 201 /* tsk_inc(): increment counter for sent or received data
 202  * - If block based flow control is not supported by peer we
 203  *   fall back to message based ditto, incrementing the counter
 204  */
 205 static u16 tsk_inc(struct tipc_sock *tsk, int msglen)
 206 {
 207         if (likely(tsk->peer_caps & TIPC_BLOCK_FLOWCTL))
 208                 return ((msglen / FLOWCTL_BLK_SZ) + 1);
 209         return 1;
 210 }
 211
 212 /**
 213  * tsk_advance_rx_queue - discard first buffer in socket receive queue
 214  *
 215  * Caller must hold socket lock
 216  */
 217 static void tsk_advance_rx_queue(struct sock *sk)
 218 {
 219         kfree_skb(__skb_dequeue(&sk->sk_receive_queue));
 220 }
 221
 222 /* tipc_sk_respond() : send response message back to sender
 223  */
 224 static void tipc_sk_respond(struct sock *sk, struct sk_buff *skb, int err)
 225 {
 226         u32 selector;
 227         u32 dnode;
 228         u32 onode = tipc_own_addr(sock_net(sk));
 229
 230         if (!tipc_msg_reverse(onode, &skb, err))
 231                 return;
 232
 233         dnode = msg_destnode(buf_msg(skb));
 234         selector = msg_origport(buf_msg(skb));
 235         tipc_node_xmit_skb(sock_net(sk), skb, dnode, selector);
 236 }
 237
 238 /**
 239  * tsk_rej_rx_queue - reject all buffers in socket receive queue
 240  *
 241  * Caller must hold socket lock
 242  */
 243 static void tsk_rej_rx_queue(struct sock *sk)
 244 {
 245         struct sk_buff *skb;
 246
 247         while ((skb = __skb_dequeue(&sk->sk_receive_queue)))
 248                 tipc_sk_respond(sk, skb, TIPC_ERR_NO_PORT);
 249 }
 250
 251 static bool tipc_sk_connected(struct sock *sk)
 252 {
 253         return sk->sk_state == TIPC_ESTABLISHED;
 254 }
 255
 256 /* tipc_sk_type_connectionless - check if the socket is datagram socket
 257  * @sk: socket
 258  *
 259  * Returns true if connection less, false otherwise
 260  */
 261 static bool tipc_sk_type_connectionless(struct sock *sk)
 262 {
 263         return sk->sk_type == SOCK_RDM || sk->sk_type == SOCK_DGRAM;
 264 }
 265
 266 /* tsk_peer_msg - verify if message was sent by connected port's peer
 267  *
 268  * Handles cases where the node's network address has changed from
 269  * the default of <0.0.0> to its configured setting.
 270  */
 271 static bool tsk_peer_msg(struct tipc_sock *tsk, struct tipc_msg *msg)
 272 {
 273         struct sock *sk = &tsk->sk;
 274         struct tipc_net *tn = net_generic(sock_net(sk), tipc_net_id);
 275         u32 peer_port = tsk_peer_port(tsk);
 276         u32 orig_node;
 277         u32 peer_node;
 278
 279         if (unlikely(!tipc_sk_connected(sk)))
 280                 return false;
 281
 282         if (unlikely(msg_origport(msg) != peer_port))
 283                 return false;
 284
 285         orig_node = msg_orignode(msg);
 286         peer_node = tsk_peer_node(tsk);
 287
 288         if (likely(orig_node == peer_node))
 289                 return true;
 290
 291         if (!orig_node && (peer_node == tn->own_addr))
 292                 return true;
 293
 294         if (!peer_node && (orig_node == tn->own_addr))
 295                 return true;
 296
 297         return false;
 298 }
 299
 300 /* tipc_set_sk_state - set the sk_state of the socket
 301  * @sk: socket
 302  *
 303  * Caller must hold socket lock
 304  *
 305  * Returns 0 on success, errno otherwise
 306  */
 307 static int tipc_set_sk_state(struct sock *sk, int state)
 308 {
 309         int oldsk_state = sk->sk_state;
 310         int res = -EINVAL;
 311
 312         switch (state) {
 313         case TIPC_OPEN:
 314                 res = 0;
 315                 break;
 316         case TIPC_LISTEN:
 317         case TIPC_CONNECTING:
 318                 if (oldsk_state == TIPC_OPEN)
 319                         res = 0;
 320                 break;
 321         case TIPC_ESTABLISHED:
 322                 if (oldsk_state == TIPC_CONNECTING ||
 323                     oldsk_state == TIPC_OPEN)
 324                         res = 0;
 325                 break;
 326         case TIPC_DISCONNECTING:
 327                 if (oldsk_state == TIPC_CONNECTING ||
 328                     oldsk_state == TIPC_ESTABLISHED)
 329                         res = 0;
 330                 break;
 331         }
 332
 333         if (!res)
 334                 sk->sk_state = state;
 335
 336         return res;
 337 }
 338
 339 static int tipc_sk_sock_err(struct socket *sock, long *timeout)
 340 {
 341         struct sock *sk = sock->sk;
 342         int err = sock_error(sk);
 343         int typ = sock->type;
 344
 345         if (err)
 346                 return err;
 347         if (typ == SOCK_STREAM || typ == SOCK_SEQPACKET) {
 348                 if (sk->sk_state == TIPC_DISCONNECTING)
 349                         return -EPIPE;
 350                 else if (!tipc_sk_connected(sk))
 351                         return -ENOTCONN;
 352         }
 353         if (!*timeout)
 354                 return -EAGAIN;
 355         if (signal_pending(current))
 356                 return sock_intr_errno(*timeout);
 357
 358         return 0;
 359 }
 360
 361 #define tipc_wait_for_cond(sock_, timeout_, condition_)                 \
 362 ({                                                                      \
 363         int rc_ = 0;                                                    \
 364         int done_ = 0;                                                  \
 365                                                                         \
 366         while (!(condition_) && !done_) {                               \
 367                 struct sock *sk_ = sock->sk;                            \
 368                 DEFINE_WAIT_FUNC(wait_, woken_wake_function);           \
 369                                                                         \
 370                 rc_ = tipc_sk_sock_err(sock_, timeout_);                \
 371                 if (rc_)                                                \
 372                         break;                                          \
 373                 prepare_to_wait(sk_sleep(sk_), &wait_,                  \
 374                                 TASK_INTERRUPTIBLE);                    \
 375                 done_ = sk_wait_event(sk_, timeout_,                    \
 376                                       (condition_), &wait_);            \
 377                 remove_wait_queue(sk_sleep(sk_), &wait_);               \
 378         }                                                               \
 379         rc_;                                                            \
 380 })
 381
 382 /**
 383  * tipc_sk_create - create a TIPC socket
 384  * @net: network namespace (must be default network)
 385  * @sock: pre-allocated socket structure
 386  * @protocol: protocol indicator (must be 0)
 387  * @kern: caused by kernel or by userspace?
 388  *
 389  * This routine creates additional data structures used by the TIPC socket,
 390  * initializes them, and links them together.
 391  *
 392  * Returns 0 on success, errno otherwise
 393  */
 394 static int tipc_sk_create(struct net *net, struct socket *sock,
 395                           int protocol, int kern)
 396 {
 397         struct tipc_net *tn;
 398         const struct proto_ops *ops;
 399         struct sock *sk;
 400         struct tipc_sock *tsk;
 401         struct tipc_msg *msg;
 402
 403         /* Validate arguments */
 404         if (unlikely(protocol != 0))
 405                 return -EPROTONOSUPPORT;
 406
 407         switch (sock->type) {
 408         case SOCK_STREAM:
 409                 ops = &stream_ops;
 410                 break;
 411         case SOCK_SEQPACKET:
 412                 ops = &packet_ops;
 413                 break;
 414         case SOCK_DGRAM:
 415         case SOCK_RDM:
 416                 ops = &msg_ops;
 417                 break;
 418         default:
 419                 return -EPROTOTYPE;
 420         }
 421
 422         /* Allocate socket's protocol area */
 423         sk = sk_alloc(net, AF_TIPC, GFP_KERNEL, &tipc_proto, kern);
 424         if (sk == NULL)
 425                 return -ENOMEM;
 426
 427         tsk = tipc_sk(sk);
 428         tsk->max_pkt = MAX_PKT_DEFAULT;
 429         INIT_LIST_HEAD(&tsk->publications);
 430         INIT_LIST_HEAD(&tsk->cong_links);
 431         msg = &tsk->phdr;
 432         tn = net_generic(sock_net(sk), tipc_net_id);
 433
 434         /* Finish initializing socket data structures */
 435         sock->ops = ops;
 436         sock_init_data(sock, sk);
 437         tipc_set_sk_state(sk, TIPC_OPEN);
 438         if (tipc_sk_insert(tsk)) {
 439                 pr_warn("Socket create failed; port number exhausted\n");
 440                 return -EINVAL;
 441         }
 442
 443         /* Ensure tsk is visible before we read own_addr. */
 444         smp_mb();
 445
 446         tipc_msg_init(tn->own_addr, msg, TIPC_LOW_IMPORTANCE, TIPC_NAMED_MSG,
 447                       NAMED_H_SIZE, 0);
 448
 449         msg_set_origport(msg, tsk->portid);
 450         setup_timer(&sk->sk_timer, tipc_sk_timeout, (unsigned long)tsk);
 451         sk->sk_shutdown = 0;
 452         sk->sk_backlog_rcv = tipc_backlog_rcv;
 453         sk->sk_rcvbuf = sysctl_tipc_rmem[1];
 454         sk->sk_data_ready = tipc_data_ready;
 455         sk->sk_write_space = tipc_write_space;
 456         sk->sk_destruct = tipc_sock_destruct;
 457         tsk->conn_timeout = CONN_TIMEOUT_DEFAULT;
 458         atomic_set(&tsk->dupl_rcvcnt, 0);
 459
 460         /* Start out with safe limits until we receive an advertised window */
 461         tsk->snd_win = tsk_adv_blocks(RCVBUF_MIN);
 462         tsk->rcv_win = tsk->snd_win;
 463
 464         if (tipc_sk_type_connectionless(sk)) {
 465                 tsk_set_unreturnable(tsk, true);
 466                 if (sock->type == SOCK_DGRAM)
 467                         tsk_set_unreliable(tsk, true);
 468         }
 469
 470         return 0;
 471 }
 472
 473 static void tipc_sk_callback(struct rcu_head *head)
 474 {
 475         struct tipc_sock *tsk = container_of(head, struct tipc_sock, rcu);
 476
 477         sock_put(&tsk->sk);
 478 }
 479
 480 /* Caller should hold socket lock for the socket. */
 481 static void __tipc_shutdown(struct socket *sock, int error)
 482 {
 483         struct sock *sk = sock->sk;
 484         struct tipc_sock *tsk = tipc_sk(sk);
 485         struct net *net = sock_net(sk);
 486         long timeout = CONN_TIMEOUT_DEFAULT;
 487         u32 dnode = tsk_peer_node(tsk);
 488         struct sk_buff *skb;
 489
 490         /* Avoid that hi-prio shutdown msgs bypass msgs in link wakeup queue */
 491         tipc_wait_for_cond(sock, &timeout, (!tsk->cong_link_cnt &&
 492                                             !tsk_conn_cong(tsk)));
 493
 494         /* Reject all unreceived messages, except on an active connection
 495          * (which disconnects locally & sends a 'FIN+' to peer).
 496          */
 497         while ((skb = __skb_dequeue(&sk->sk_receive_queue)) != NULL) {
 498                 if (TIPC_SKB_CB(skb)->bytes_read) {
 499                         kfree_skb(skb);
 500                         continue;
 501                 }
 502                 if (!tipc_sk_type_connectionless(sk) &&
 503                     sk->sk_state != TIPC_DISCONNECTING) {
 504                         tipc_set_sk_state(sk, TIPC_DISCONNECTING);
 505                         tipc_node_remove_conn(net, dnode, tsk->portid);
 506                 }
 507                 tipc_sk_respond(sk, skb, error);
 508         }
 509
 510         if (tipc_sk_type_connectionless(sk))
 511                 return;
 512
 513         if (sk->sk_state != TIPC_DISCONNECTING) {
 514                 skb = tipc_msg_create(TIPC_CRITICAL_IMPORTANCE,
 515                                       TIPC_CONN_MSG, SHORT_H_SIZE, 0, dnode,
 516                                       tsk_own_node(tsk), tsk_peer_port(tsk),
 517                                       tsk->portid, error);
 518                 if (skb)
 519                         tipc_node_xmit_skb(net, skb, dnode, tsk->portid);
 520                 tipc_node_remove_conn(net, dnode, tsk->portid);
 521                 tipc_set_sk_state(sk, TIPC_DISCONNECTING);
 522         }
 523 }
 524
 525 /**
 526  * tipc_release - destroy a TIPC socket
 527  * @sock: socket to destroy
 528  *
 529  * This routine cleans up any messages that are still queued on the socket.
 530  * For DGRAM and RDM socket types, all queued messages are rejected.
 531  * For SEQPACKET and STREAM socket types, the first message is rejected
 532  * and any others are discarded.  (If the first message on a STREAM socket
 533  * is partially-read, it is discarded and the next one is rejected instead.)
 534  *
 535  * NOTE: Rejected messages are not necessarily returned to the sender!  They
 536  * are returned or discarded according to the "destination droppable" setting
 537  * specified for the message by the sender.
 538  *
 539  * Returns 0 on success, errno otherwise
 540  */
 541 static int tipc_release(struct socket *sock)
 542 {
 543         struct sock *sk = sock->sk;
 544         struct tipc_sock *tsk;
 545
 546         /*
 547          * Exit if socket isn't fully initialized (occurs when a failed accept()
 548          * releases a pre-allocated child socket that was never used)
 549          */
 550         if (sk == NULL)
 551                 return 0;
 552
 553         tsk = tipc_sk(sk);
 554         lock_sock(sk);
 555
 556         __tipc_shutdown(sock, TIPC_ERR_NO_PORT);
 557         sk->sk_shutdown = SHUTDOWN_MASK;
 558         tipc_sk_withdraw(tsk, 0, NULL);
 559         sk_stop_timer(sk, &sk->sk_timer);
 560         tipc_sk_remove(tsk);
 561
 562         /* Reject any messages that accumulated in backlog queue */
 563         release_sock(sk);
 564         u32_list_purge(&tsk->cong_links);
 565         tsk->cong_link_cnt = 0;
 566         call_rcu(&tsk->rcu, tipc_sk_callback);
 567         sock->sk = NULL;
 568
 569         return 0;
 570 }
 571
 572 /**
 573  * tipc_bind - associate or disassocate TIPC name(s) with a socket
 574  * @sock: socket structure
 575  * @uaddr: socket address describing name(s) and desired operation
 576  * @uaddr_len: size of socket address data structure
 577  *
 578  * Name and name sequence binding is indicated using a positive scope value;
 579  * a negative scope value unbinds the specified name.  Specifying no name
 580  * (i.e. a socket address length of 0) unbinds all names from the socket.
 581  *
 582  * Returns 0 on success, errno otherwise
 583  *
 584  * NOTE: This routine doesn't need to take the socket lock since it doesn't
 585  *       access any non-constant socket information.
 586  */
 587 static int tipc_bind(struct socket *sock, struct sockaddr *uaddr,
 588                      int uaddr_len)
 589 {
 590         struct sock *sk = sock->sk;
 591         struct sockaddr_tipc *addr = (struct sockaddr_tipc *)uaddr;
 592         struct tipc_sock *tsk = tipc_sk(sk);
 593         int res = -EINVAL;
 594
 595         lock_sock(sk);
 596         if (unlikely(!uaddr_len)) {
 597                 res = tipc_sk_withdraw(tsk, 0, NULL);
 598                 goto exit;
 599         }
 600
 601         if (uaddr_len < sizeof(struct sockaddr_tipc)) {
 602                 res = -EINVAL;
 603                 goto exit;
 604         }
 605         if (addr->family != AF_TIPC) {
 606                 res = -EAFNOSUPPORT;
 607                 goto exit;
 608         }
 609
 610         if (addr->addrtype == TIPC_ADDR_NAME)
 611                 addr->addr.nameseq.upper = addr->addr.nameseq.lower;
 612         else if (addr->addrtype != TIPC_ADDR_NAMESEQ) {
 613                 res = -EAFNOSUPPORT;
 614                 goto exit;
 615         }
 616
 617         if ((addr->addr.nameseq.type < TIPC_RESERVED_TYPES) &&
 618             (addr->addr.nameseq.type != TIPC_TOP_SRV) &&
 619             (addr->addr.nameseq.type != TIPC_CFG_SRV)) {
 620                 res = -EACCES;
 621                 goto exit;
 622         }
 623
 624         res = (addr->scope > 0) ?
 625                 tipc_sk_publish(tsk, addr->scope, &addr->addr.nameseq) :
 626                 tipc_sk_withdraw(tsk, -addr->scope, &addr->addr.nameseq);
 627 exit:
 628         release_sock(sk);
 629         return res;
 630 }
 631
 632 /**
 633  * tipc_getname - get port ID of socket or peer socket
 634  * @sock: socket structure
 635  * @uaddr: area for returned socket address
 636  * @uaddr_len: area for returned length of socket address
 637  * @peer: 0 = own ID, 1 = current peer ID, 2 = current/former peer ID
 638  *
 639  * Returns 0 on success, errno otherwise
 640  *
 641  * NOTE: This routine doesn't need to take the socket lock since it only
 642  *       accesses socket information that is unchanging (or which changes in
 643  *       a completely predictable manner).
 644  */
 645 static int tipc_getname(struct socket *sock, struct sockaddr *uaddr,
 646                         int *uaddr_len, int peer)
 647 {
 648         struct sockaddr_tipc *addr = (struct sockaddr_tipc *)uaddr;
 649         struct sock *sk = sock->sk;
 650         struct tipc_sock *tsk = tipc_sk(sk);
 651         struct tipc_net *tn = net_generic(sock_net(sock->sk), tipc_net_id);
 652
 653         memset(addr, 0, sizeof(*addr));
 654         if (peer) {
 655                 if ((!tipc_sk_connected(sk)) &&
 656                     ((peer != 2) || (sk->sk_state != TIPC_DISCONNECTING)))
 657                         return -ENOTCONN;
 658                 addr->addr.id.ref = tsk_peer_port(tsk);
 659                 addr->addr.id.node = tsk_peer_node(tsk);
 660         } else {
 661                 addr->addr.id.ref = tsk->portid;
 662                 addr->addr.id.node = tn->own_addr;
 663         }
 664
 665         *uaddr_len = sizeof(*addr);
 666         addr->addrtype = TIPC_ADDR_ID;
 667         addr->family = AF_TIPC;
 668         addr->scope = 0;
 669         addr->addr.name.domain = 0;
 670
 671         return 0;
 672 }
 673
 674 /**
 675  * tipc_poll - read and possibly block on pollmask
 676  * @file: file structure associated with the socket
 677  * @sock: socket for which to calculate the poll bits
 678  * @wait: ???
 679  *
 680  * Returns pollmask value
 681  *
 682  * COMMENTARY:
 683  * It appears that the usual socket locking mechanisms are not useful here
 684  * since the pollmask info is potentially out-of-date the moment this routine
 685  * exits.  TCP and other protocols seem to rely on higher level poll routines
 686  * to handle any preventable race conditions, so TIPC will do the same ...
 687  *
 688  * IMPORTANT: The fact that a read or write operation is indicated does NOT
 689  * imply that the operation will succeed, merely that it should be performed
 690  * and will not block.
 691  */
 692 static unsigned int tipc_poll(struct file *file, struct socket *sock,
 693                               poll_table *wait)
 694 {
 695         struct sock *sk = sock->sk;
 696         struct tipc_sock *tsk = tipc_sk(sk);
 697         u32 mask = 0;
 698
 699         sock_poll_wait(file, sk_sleep(sk), wait);
 700
 701         if (sk->sk_shutdown & RCV_SHUTDOWN)
 702                 mask |= POLLRDHUP | POLLIN | POLLRDNORM;
 703         if (sk->sk_shutdown == SHUTDOWN_MASK)
 704                 mask |= POLLHUP;
 705
 706         switch (sk->sk_state) {
 707         case TIPC_ESTABLISHED:
 708                 if (!tsk->cong_link_cnt && !tsk_conn_cong(tsk))
 709                         mask |= POLLOUT;
 710                 /* fall thru' */
 711         case TIPC_LISTEN:
 712         case TIPC_CONNECTING:
 713                 if (!skb_queue_empty(&sk->sk_receive_queue))
 714                         mask |= (POLLIN | POLLRDNORM);
 715                 break;
 716         case TIPC_OPEN:
 717                 if (!tsk->cong_link_cnt)
 718                         mask |= POLLOUT;
 719                 if (tipc_sk_type_connectionless(sk) &&
 720                     (!skb_queue_empty(&sk->sk_receive_queue)))
 721                         mask |= (POLLIN | POLLRDNORM);
 722                 break;
 723         case TIPC_DISCONNECTING:
 724                 mask = (POLLIN | POLLRDNORM | POLLHUP);
 725                 break;
 726         }
 727
 728         return mask;
 729 }
 730
 731 /**
 732  * tipc_sendmcast - send multicast message
 733  * @sock: socket structure
 734  * @seq: destination address
 735  * @msg: message to send
 736  * @dlen: length of data to send
 737  * @timeout: timeout to wait for wakeup
 738  *
 739  * Called from function tipc_sendmsg(), which has done all sanity checks
 740  * Returns the number of bytes sent on success, or errno
 741  */
 742 static int tipc_sendmcast(struct  socket *sock, struct tipc_name_seq *seq,
 743                           struct msghdr *msg, size_t dlen, long timeout)
 744 {
 745         struct sock *sk = sock->sk;
 746         struct tipc_sock *tsk = tipc_sk(sk);
 747         struct tipc_msg *hdr = &tsk->phdr;
 748         struct net *net = sock_net(sk);
 749         int mtu = tipc_bcast_get_mtu(net);
 750         struct tipc_mc_method *method = &tsk->mc_method;
 751         u32 domain = addr_domain(net, TIPC_CLUSTER_SCOPE);
 752         struct sk_buff_head pkts;
 753         struct tipc_nlist dsts;
 754         int rc;
 755
 756         /* Block or return if any destination link is congested */
 757         rc = tipc_wait_for_cond(sock, &timeout, !tsk->cong_link_cnt);
 758         if (unlikely(rc))
 759                 return rc;
 760
 761         /* Lookup destination nodes */
 762         tipc_nlist_init(&dsts, tipc_own_addr(net));
 763         tipc_nametbl_lookup_dst_nodes(net, seq->type, seq->lower,
 764                                       seq->upper, domain, &dsts);
 765         if (!dsts.local && !dsts.remote)
 766                 return -EHOSTUNREACH;
 767
 768         /* Build message header */
 769         msg_set_type(hdr, TIPC_MCAST_MSG);
 770         msg_set_hdr_sz(hdr, MCAST_H_SIZE);
 771         msg_set_lookup_scope(hdr, TIPC_CLUSTER_SCOPE);
 772         msg_set_destport(hdr, 0);
 773         msg_set_destnode(hdr, 0);
 774         msg_set_nametype(hdr, seq->type);
 775         msg_set_namelower(hdr, seq->lower);
 776         msg_set_nameupper(hdr, seq->upper);
 777
 778         /* Build message as chain of buffers */
 779         skb_queue_head_init(&pkts);
 780         rc = tipc_msg_build(hdr, msg, 0, dlen, mtu, &pkts);
 781
 782         /* Send message if build was successful */
 783         if (unlikely(rc == dlen))
 784                 rc = tipc_mcast_xmit(net, &pkts, method, &dsts,
 785                                      &tsk->cong_link_cnt);
 786
 787         tipc_nlist_purge(&dsts);
 788
 789         return rc ? rc : dlen;
 790 }
 791
 792 /**
 793  * tipc_sk_mcast_rcv - Deliver multicast messages to all destination sockets
 794  * @arrvq: queue with arriving messages, to be cloned after destination lookup
 795  * @inputq: queue with cloned messages, delivered to socket after dest lookup
 796  *
 797  * Multi-threaded: parallel calls with reference to same queues may occur
 798  */
 799 void tipc_sk_mcast_rcv(struct net *net, struct sk_buff_head *arrvq,
 800                        struct sk_buff_head *inputq)
 801 {
 802         struct tipc_msg *msg;
 803         struct list_head dports;
 804         u32 portid;
 805         u32 scope = TIPC_CLUSTER_SCOPE;
 806         struct sk_buff_head tmpq;
 807         uint hsz;
 808         struct sk_buff *skb, *_skb;
 809
 810         __skb_queue_head_init(&tmpq);
 811         INIT_LIST_HEAD(&dports);
 812
 813         skb = tipc_skb_peek(arrvq, &inputq->lock);
 814         for (; skb; skb = tipc_skb_peek(arrvq, &inputq->lock)) {
 815                 msg = buf_msg(skb);
 816                 hsz = skb_headroom(skb) + msg_hdr_sz(msg);
 817
 818                 if (in_own_node(net, msg_orignode(msg)))
 819                         scope = TIPC_NODE_SCOPE;
 820
 821                 /* Create destination port list and message clones: */
 822                 tipc_nametbl_mc_translate(net,
 823                                           msg_nametype(msg), msg_namelower(msg),
 824                                           msg_nameupper(msg), scope, &dports);
 825                 portid = u32_pop(&dports);
 826                 for (; portid; portid = u32_pop(&dports)) {
 827                         _skb = __pskb_copy(skb, hsz, GFP_ATOMIC);
 828                         if (_skb) {
 829                                 msg_set_destport(buf_msg(_skb), portid);
 830                                 __skb_queue_tail(&tmpq, _skb);
 831                                 continue;
 832                         }
 833                         pr_warn("Failed to clone mcast rcv buffer\n");
 834                 }
 835                 /* Append to inputq if not already done by other thread */
 836                 spin_lock_bh(&inputq->lock);
 837                 if (skb_peek(arrvq) == skb) {
 838                         skb_queue_splice_tail_init(&tmpq, inputq);
 839                         kfree_skb(__skb_dequeue(arrvq));
 840                 }
 841                 spin_unlock_bh(&inputq->lock);
 842                 __skb_queue_purge(&tmpq);
 843                 kfree_skb(skb);
 844         }
 845         tipc_sk_rcv(net, inputq);
 846 }
 847
 848 /**
 849  * tipc_sk_proto_rcv - receive a connection mng protocol message
 850  * @tsk: receiving socket
 851  * @skb: pointer to message buffer.
 852  */
 853 static void tipc_sk_proto_rcv(struct tipc_sock *tsk, struct sk_buff *skb,
 854                               struct sk_buff_head *xmitq)
 855 {
 856         struct sock *sk = &tsk->sk;
 857         u32 onode = tsk_own_node(tsk);
 858         struct tipc_msg *hdr = buf_msg(skb);
 859         int mtyp = msg_type(hdr);
 860         bool conn_cong;
 861
 862         /* Ignore if connection cannot be validated: */
 863         if (!tsk_peer_msg(tsk, hdr))
 864                 goto exit;
 865
 866         tsk->probe_unacked = false;
 867
 868         if (mtyp == CONN_PROBE) {
 869                 msg_set_type(hdr, CONN_PROBE_REPLY);
 870                 if (tipc_msg_reverse(onode, &skb, TIPC_OK))
 871                         __skb_queue_tail(xmitq, skb);
 872                 return;
 873         } else if (mtyp == CONN_ACK) {
 874                 conn_cong = tsk_conn_cong(tsk);
 875                 tsk->snt_unacked -= msg_conn_ack(hdr);
 876                 if (tsk->peer_caps & TIPC_BLOCK_FLOWCTL)
 877                         tsk->snd_win = msg_adv_win(hdr);
 878                 if (conn_cong)
 879                         sk->sk_write_space(sk);
 880         } else if (mtyp != CONN_PROBE_REPLY) {
 881                 pr_warn("Received unknown CONN_PROTO msg\n");
 882         }
 883 exit:
 884         kfree_skb(skb);
 885 }
 886
 887 /**
 888  * tipc_sendmsg - send message in connectionless manner
 889  * @sock: socket structure
 890  * @m: message to send
 891  * @dsz: amount of user data to be sent
 892  *
 893  * Message must have an destination specified explicitly.
 894  * Used for SOCK_RDM and SOCK_DGRAM messages,
 895  * and for 'SYN' messages on SOCK_SEQPACKET and SOCK_STREAM connections.
 896  * (Note: 'SYN+' is prohibited on SOCK_STREAM.)
 897  *
 898  * Returns the number of bytes sent on success, or errno otherwise
 899  */
 900 static int tipc_sendmsg(struct socket *sock,
 901                         struct msghdr *m, size_t dsz)
 902 {
 903         struct sock *sk = sock->sk;
 904         int ret;
 905
 906         lock_sock(sk);
 907         ret = __tipc_sendmsg(sock, m, dsz);
 908         release_sock(sk);
 909
 910         return ret;
 911 }
 912
 913 static int __tipc_sendmsg(struct socket *sock, struct msghdr *m, size_t dlen)
 914 {
 915         struct sock *sk = sock->sk;
 916         struct net *net = sock_net(sk);
 917         struct tipc_sock *tsk = tipc_sk(sk);
 918         DECLARE_SOCKADDR(struct sockaddr_tipc *, dest, m->msg_name);
 919         long timeout = sock_sndtimeo(sk, m->msg_flags & MSG_DONTWAIT);
 920         struct list_head *clinks = &tsk->cong_links;
 921         bool syn = !tipc_sk_type_connectionless(sk);
 922         struct tipc_msg *hdr = &tsk->phdr;
 923         struct tipc_name_seq *seq;
 924         struct sk_buff_head pkts;
 925         u32 type, inst, domain;
 926         u32 dnode, dport;
 927         int mtu, rc;
 928
 929         if (unlikely(dlen > TIPC_MAX_USER_MSG_SIZE))
 930                 return -EMSGSIZE;
 931
 932         if (unlikely(!dest)) {
 933                 dest = &tsk->peer;
 934                 if (!syn || dest->family != AF_TIPC)
 935                         return -EDESTADDRREQ;
 936         }
 937
 938         if (unlikely(m->msg_namelen < sizeof(*dest)))
 939                 return -EINVAL;
 940
 941         if (unlikely(dest->family != AF_TIPC))
 942                 return -EINVAL;
 943
 944         if (unlikely(syn)) {
 945                 if (sk->sk_state == TIPC_LISTEN)
 946                         return -EPIPE;
 947                 if (sk->sk_state != TIPC_OPEN)
 948                         return -EISCONN;
 949                 if (tsk->published)
 950                         return -EOPNOTSUPP;
 951                 if (dest->addrtype == TIPC_ADDR_NAME) {
 952                         tsk->conn_type = dest->addr.name.name.type;
 953                         tsk->conn_instance = dest->addr.name.name.instance;
 954                 }
 955         }
 956
 957         seq = &dest->addr.nameseq;
 958         if (dest->addrtype == TIPC_ADDR_MCAST)
 959                 return tipc_sendmcast(sock, seq, m, dlen, timeout);
 960
 961         if (dest->addrtype == TIPC_ADDR_NAME) {
 962                 type = dest->addr.name.name.type;
 963                 inst = dest->addr.name.name.instance;
 964                 domain = dest->addr.name.domain;
 965                 dnode = domain;
 966                 msg_set_type(hdr, TIPC_NAMED_MSG);
 967                 msg_set_hdr_sz(hdr, NAMED_H_SIZE);
 968                 msg_set_nametype(hdr, type);
 969                 msg_set_nameinst(hdr, inst);
 970                 msg_set_lookup_scope(hdr, tipc_addr_scope(domain));
 971                 dport = tipc_nametbl_translate(net, type, inst, &dnode);
 972                 msg_set_destnode(hdr, dnode);
 973                 msg_set_destport(hdr, dport);
 974                 if (unlikely(!dport && !dnode))
 975                         return -EHOSTUNREACH;
 976
 977         } else if (dest->addrtype == TIPC_ADDR_ID) {
 978                 dnode = dest->addr.id.node;
 979                 msg_set_type(hdr, TIPC_DIRECT_MSG);
 980                 msg_set_lookup_scope(hdr, 0);
 981                 msg_set_destnode(hdr, dnode);
 982                 msg_set_destport(hdr, dest->addr.id.ref);
 983                 msg_set_hdr_sz(hdr, BASIC_H_SIZE);
 984         }
 985
 986         /* Block or return if destination link is congested */
 987         rc = tipc_wait_for_cond(sock, &timeout, !u32_find(clinks, dnode));
 988         if (unlikely(rc))
 989                 return rc;
 990
 991         skb_queue_head_init(&pkts);
 992         mtu = tipc_node_get_mtu(net, dnode, tsk->portid);
 993         rc = tipc_msg_build(hdr, m, 0, dlen, mtu, &pkts);
 994         if (unlikely(rc != dlen))
 995                 return rc;
 996
 997         rc = tipc_node_xmit(net, &pkts, dnode, tsk->portid);
 998         if (unlikely(rc == -ELINKCONG)) {
 999                 u32_push(clinks, dnode);
1000                 tsk->cong_link_cnt++;
1001                 rc = 0;
1002         }
1003
1004         if (unlikely(syn && !rc))
1005                 tipc_set_sk_state(sk, TIPC_CONNECTING);
1006
1007         return rc ? rc : dlen;
1008 }
1009
1010 /**
1011  * tipc_sendstream - send stream-oriented data
1012  * @sock: socket structure
1013  * @m: data to send
1014  * @dsz: total length of data to be transmitted
1015  *
1016  * Used for SOCK_STREAM data.
1017  *
1018  * Returns the number of bytes sent on success (or partial success),
1019  * or errno if no data sent
1020  */
1021 static int tipc_sendstream(struct socket *sock, struct msghdr *m, size_t dsz)
1022 {
1023         struct sock *sk = sock->sk;
1024         int ret;
1025
1026         lock_sock(sk);
1027         ret = __tipc_sendstream(sock, m, dsz);
1028         release_sock(sk);
1029
1030         return ret;
1031 }
1032
1033 static int __tipc_sendstream(struct socket *sock, struct msghdr *m, size_t dlen)
1034 {
1035         struct sock *sk = sock->sk;
1036         DECLARE_SOCKADDR(struct sockaddr_tipc *, dest, m->msg_name);
1037         long timeout = sock_sndtimeo(sk, m->msg_flags & MSG_DONTWAIT);
1038         struct tipc_sock *tsk = tipc_sk(sk);
1039         struct tipc_msg *hdr = &tsk->phdr;
1040         struct net *net = sock_net(sk);
1041         struct sk_buff_head pkts;
1042         u32 dnode = tsk_peer_node(tsk);
1043         int send, sent = 0;
1044         int rc = 0;
1045
1046         skb_queue_head_init(&pkts);
1047
1048         if (unlikely(dlen > INT_MAX))
1049                 return -EMSGSIZE;
1050
1051         /* Handle implicit connection setup */
1052         if (unlikely(dest)) {
1053                 rc = __tipc_sendmsg(sock, m, dlen);
1054                 if (dlen && (dlen == rc))
1055                         tsk->snt_unacked = tsk_inc(tsk, dlen + msg_hdr_sz(hdr));
1056                 return rc;
1057         }
1058
1059         do {
1060                 rc = tipc_wait_for_cond(sock, &timeout,
1061                                         (!tsk->cong_link_cnt &&
1062                                          !tsk_conn_cong(tsk) &&
1063                                          tipc_sk_connected(sk)));
1064                 if (unlikely(rc))
1065                         break;
1066
1067                 send = min_t(size_t, dlen - sent, TIPC_MAX_USER_MSG_SIZE);
1068                 rc = tipc_msg_build(hdr, m, sent, send, tsk->max_pkt, &pkts);
1069                 if (unlikely(rc != send))
1070                         break;
1071
1072                 rc = tipc_node_xmit(net, &pkts, dnode, tsk->portid);
1073                 if (unlikely(rc == -ELINKCONG)) {
1074                         tsk->cong_link_cnt = 1;
1075                         rc = 0;
1076                 }
1077                 if (likely(!rc)) {
1078                         tsk->snt_unacked += tsk_inc(tsk, send + MIN_H_SIZE);
1079                         sent += send;
1080                 }
1081         } while (sent < dlen && !rc);
1082
1083         return rc ? rc : sent;
1084 }
1085
1086 /**
1087  * tipc_send_packet - send a connection-oriented message
1088  * @sock: socket structure
1089  * @m: message to send
1090  * @dsz: length of data to be transmitted
1091  *
1092  * Used for SOCK_SEQPACKET messages.
1093  *
1094  * Returns the number of bytes sent on success, or errno otherwise
1095  */
1096 static int tipc_send_packet(struct socket *sock, struct msghdr *m, size_t dsz)
1097 {
1098         if (dsz > TIPC_MAX_USER_MSG_SIZE)
1099                 return -EMSGSIZE;
1100
1101         return tipc_sendstream(sock, m, dsz);
1102 }
1103
1104 /* tipc_sk_finish_conn - complete the setup of a connection
1105  */
1106 static void tipc_sk_finish_conn(struct tipc_sock *tsk, u32 peer_port,
1107                                 u32 peer_node)
1108 {
1109         struct sock *sk = &tsk->sk;
1110         struct net *net = sock_net(sk);
1111         struct tipc_msg *msg = &tsk->phdr;
1112
1113         msg_set_destnode(msg, peer_node);
1114         msg_set_destport(msg, peer_port);
1115         msg_set_type(msg, TIPC_CONN_MSG);
1116         msg_set_lookup_scope(msg, 0);
1117         msg_set_hdr_sz(msg, SHORT_H_SIZE);
1118
1119         sk_reset_timer(sk, &sk->sk_timer, jiffies + CONN_PROBING_INTERVAL);
1120         tipc_set_sk_state(sk, TIPC_ESTABLISHED);
1121         tipc_node_add_conn(net, peer_node, tsk->portid, peer_port);
1122         tsk->max_pkt = tipc_node_get_mtu(net, peer_node, tsk->portid);
1123         tsk->peer_caps = tipc_node_get_capabilities(net, peer_node);
1124         if (tsk->peer_caps & TIPC_BLOCK_FLOWCTL)
1125                 return;
1126
1127         /* Fall back to message based flow control */
1128         tsk->rcv_win = FLOWCTL_MSG_WIN;
1129         tsk->snd_win = FLOWCTL_MSG_WIN;
1130 }
1131
1132 /**
1133  * set_orig_addr - capture sender's address for received message
1134  * @m: descriptor for message info
1135  * @msg: received message header
1136  *
1137  * Note: Address is not captured if not requested by receiver.
1138  */
1139 static void set_orig_addr(struct msghdr *m, struct tipc_msg *msg)
1140 {
1141         DECLARE_SOCKADDR(struct sockaddr_tipc *, addr, m->msg_name);
1142
1143         if (addr) {
1144                 addr->family = AF_TIPC;
1145                 addr->addrtype = TIPC_ADDR_ID;
1146                 memset(&addr->addr, 0, sizeof(addr->addr));
1147                 addr->addr.id.ref = msg_origport(msg);
1148                 addr->addr.id.node = msg_orignode(msg);
1149                 addr->addr.name.domain = 0;     /* could leave uninitialized */
1150                 addr->scope = 0;                /* could leave uninitialized */
1151                 m->msg_namelen = sizeof(struct sockaddr_tipc);
1152         }
1153 }
1154
1155 /**
1156  * tipc_sk_anc_data_recv - optionally capture ancillary data for received message
1157  * @m: descriptor for message info
1158  * @msg: received message header
1159  * @tsk: TIPC port associated with message
1160  *
1161  * Note: Ancillary data is not captured if not requested by receiver.
1162  *
1163  * Returns 0 if successful, otherwise errno
1164  */
1165 static int tipc_sk_anc_data_recv(struct msghdr *m, struct tipc_msg *msg,
1166                                  struct tipc_sock *tsk)
1167 {
1168         u32 anc_data[3];
1169         u32 err;
1170         u32 dest_type;
1171         int has_name;
1172         int res;
1173
1174         if (likely(m->msg_controllen == 0))
1175                 return 0;
1176
1177         /* Optionally capture errored message object(s) */
1178         err = msg ? msg_errcode(msg) : 0;
1179         if (unlikely(err)) {
1180                 anc_data[0] = err;
1181                 anc_data[1] = msg_data_sz(msg);
1182                 res = put_cmsg(m, SOL_TIPC, TIPC_ERRINFO, 8, anc_data);
1183                 if (res)
1184                         return res;
1185                 if (anc_data[1]) {
1186                         res = put_cmsg(m, SOL_TIPC, TIPC_RETDATA, anc_data[1],
1187                                        msg_data(msg));
1188                         if (res)
1189                                 return res;
1190                 }
1191         }
1192
1193         /* Optionally capture message destination object */
1194         dest_type = msg ? msg_type(msg) : TIPC_DIRECT_MSG;
1195         switch (dest_type) {
1196         case TIPC_NAMED_MSG:
1197                 has_name = 1;
1198                 anc_data[0] = msg_nametype(msg);
1199                 anc_data[1] = msg_namelower(msg);
1200                 anc_data[2] = msg_namelower(msg);
1201                 break;
1202         case TIPC_MCAST_MSG:
1203                 has_name = 1;
1204                 anc_data[0] = msg_nametype(msg);
1205                 anc_data[1] = msg_namelower(msg);
1206                 anc_data[2] = msg_nameupper(msg);
1207                 break;
1208         case TIPC_CONN_MSG:
1209                 has_name = (tsk->conn_type != 0);
1210                 anc_data[0] = tsk->conn_type;
1211                 anc_data[1] = tsk->conn_instance;
1212                 anc_data[2] = tsk->conn_instance;
1213                 break;
1214         default:
1215                 has_name = 0;
1216         }
1217         if (has_name) {
1218                 res = put_cmsg(m, SOL_TIPC, TIPC_DESTNAME, 12, anc_data);
1219                 if (res)
1220                         return res;
1221         }
1222
1223         return 0;
1224 }
1225
1226 static void tipc_sk_send_ack(struct tipc_sock *tsk)
1227 {
1228         struct sock *sk = &tsk->sk;
1229         struct net *net = sock_net(sk);
1230         struct sk_buff *skb = NULL;
1231         struct tipc_msg *msg;
1232         u32 peer_port = tsk_peer_port(tsk);
1233         u32 dnode = tsk_peer_node(tsk);
1234
1235         if (!tipc_sk_connected(sk))
1236                 return;
1237         skb = tipc_msg_create(CONN_MANAGER, CONN_ACK, INT_H_SIZE, 0,
1238                               dnode, tsk_own_node(tsk), peer_port,
1239                               tsk->portid, TIPC_OK);
1240         if (!skb)
1241                 return;
1242         msg = buf_msg(skb);
1243         msg_set_conn_ack(msg, tsk->rcv_unacked);
1244         tsk->rcv_unacked = 0;
1245
1246         /* Adjust to and advertize the correct window limit */
1247         if (tsk->peer_caps & TIPC_BLOCK_FLOWCTL) {
1248                 tsk->rcv_win = tsk_adv_blocks(tsk->sk.sk_rcvbuf);
1249                 msg_set_adv_win(msg, tsk->rcv_win);
1250         }
1251         tipc_node_xmit_skb(net, skb, dnode, msg_link_selector(msg));
1252 }
1253
1254 static int tipc_wait_for_rcvmsg(struct socket *sock, long *timeop)
1255 {
1256         struct sock *sk = sock->sk;
1257         DEFINE_WAIT(wait);
1258         long timeo = *timeop;
1259         int err;
1260
1261         for (;;) {
1262                 prepare_to_wait(sk_sleep(sk), &wait, TASK_INTERRUPTIBLE);
1263                 if (timeo && skb_queue_empty(&sk->sk_receive_queue)) {
1264                         if (sk->sk_shutdown & RCV_SHUTDOWN) {
1265                                 err = -ENOTCONN;
1266                                 break;
1267                         }
1268                         release_sock(sk);
1269                         timeo = schedule_timeout(timeo);
1270                         lock_sock(sk);
1271                 }
1272                 err = 0;
1273                 if (!skb_queue_empty(&sk->sk_receive_queue))
1274                         break;
1275                 err = -EAGAIN;
1276                 if (!timeo)
1277                         break;
1278                 err = sock_intr_errno(timeo);
1279                 if (signal_pending(current))
1280                         break;
1281         }
1282         finish_wait(sk_sleep(sk), &wait);
1283         *timeop = timeo;
1284         return err;
1285 }
1286
1287 /**
1288  * tipc_recvmsg - receive packet-oriented message
1289  * @m: descriptor for message info
1290  * @buf_len: total size of user buffer area
1291  * @flags: receive flags
1292  *
1293  * Used for SOCK_DGRAM, SOCK_RDM, and SOCK_SEQPACKET messages.
1294  * If the complete message doesn't fit in user area, truncate it.
1295  *
1296  * Returns size of returned message data, errno otherwise
1297  */
1298 static int tipc_recvmsg(struct socket *sock, struct msghdr *m, size_t buf_len,
1299                         int flags)
1300 {
1301         struct sock *sk = sock->sk;
1302         struct tipc_sock *tsk = tipc_sk(sk);
1303         struct sk_buff *buf;
1304         struct tipc_msg *msg;
1305         bool is_connectionless = tipc_sk_type_connectionless(sk);
1306         long timeo;
1307         unsigned int sz;
1308         u32 err;
1309         int res, hlen;
1310
1311         /* Catch invalid receive requests */
1312         if (unlikely(!buf_len))
1313                 return -EINVAL;
1314
1315         lock_sock(sk);
1316
1317         if (!is_connectionless && unlikely(sk->sk_state == TIPC_OPEN)) {
1318                 res = -ENOTCONN;
1319                 goto exit;
1320         }
1321
1322         timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
1323 restart:
1324
1325         /* Look for a message in receive queue; wait if necessary */
1326         res = tipc_wait_for_rcvmsg(sock, &timeo);
1327         if (res)
1328                 goto exit;
1329
1330         /* Look at first message in receive queue */
1331         buf = skb_peek(&sk->sk_receive_queue);
1332         msg = buf_msg(buf);
1333         sz = msg_data_sz(msg);
1334         hlen = msg_hdr_sz(msg);
1335         err = msg_errcode(msg);
1336
1337         /* Discard an empty non-errored message & try again */
1338         if ((!sz) && (!err)) {
1339                 tsk_advance_rx_queue(sk);
1340                 goto restart;
1341         }
1342
1343         /* Capture sender's address (optional) */
1344         set_orig_addr(m, msg);
1345
1346         /* Capture ancillary data (optional) */
1347         res = tipc_sk_anc_data_recv(m, msg, tsk);
1348         if (res)
1349                 goto exit;
1350
1351         /* Capture message data (if valid) & compute return value (always) */
1352         if (!err) {
1353                 if (unlikely(buf_len < sz)) {
1354                         sz = buf_len;
1355                         m->msg_flags |= MSG_TRUNC;
1356                 }
1357                 res = skb_copy_datagram_msg(buf, hlen, m, sz);
1358                 if (res)
1359                         goto exit;
1360                 res = sz;
1361         } else {
1362                 if (is_connectionless || err == TIPC_CONN_SHUTDOWN ||
1363                     m->msg_control)
1364                         res = 0;
1365                 else
1366                         res = -ECONNRESET;
1367         }
1368
1369         if (unlikely(flags & MSG_PEEK))
1370                 goto exit;
1371
1372         if (likely(!is_connectionless)) {
1373                 tsk->rcv_unacked += tsk_inc(tsk, hlen + sz);
1374                 if (unlikely(tsk->rcv_unacked >= (tsk->rcv_win / 4)))
1375                         tipc_sk_send_ack(tsk);
1376         }
1377         tsk_advance_rx_queue(sk);
1378 exit:
1379         release_sock(sk);
1380         return res;
1381 }
1382
1383 /**
1384  * tipc_recv_stream - receive stream-oriented data
1385  * @m: descriptor for message info
1386  * @buf_len: total size of user buffer area
1387  * @flags: receive flags
1388  *
1389  * Used for SOCK_STREAM messages only.  If not enough data is available
1390  * will optionally wait for more; never truncates data.
1391  *
1392  * Returns size of returned message data, errno otherwise
1393  */
1394 static int tipc_recv_stream(struct socket *sock, struct msghdr *m,
1395                             size_t buf_len, int flags)
1396 {
1397         struct sock *sk = sock->sk;
1398         struct tipc_sock *tsk = tipc_sk(sk);
1399         struct sk_buff *buf;
1400         struct tipc_msg *msg;
1401         long timeo;
1402         unsigned int sz;
1403         int target;
1404         int sz_copied = 0;
1405         u32 err;
1406         int res = 0, hlen;
1407
1408         /* Catch invalid receive attempts */
1409         if (unlikely(!buf_len))
1410                 return -EINVAL;
1411
1412         lock_sock(sk);
1413
1414         if (unlikely(sk->sk_state == TIPC_OPEN)) {
1415                 res = -ENOTCONN;
1416                 goto exit;
1417         }
1418
1419         target = sock_rcvlowat(sk, flags & MSG_WAITALL, buf_len);
1420         timeo = sock_rcvtimeo(sk, flags & MSG_DONTWAIT);
1421
1422 restart:
1423         /* Look for a message in receive queue; wait if necessary */
1424         res = tipc_wait_for_rcvmsg(sock, &timeo);
1425         if (res)
1426                 goto exit;
1427
1428         /* Look at first message in receive queue */
1429         buf = skb_peek(&sk->sk_receive_queue);
1430         msg = buf_msg(buf);
1431         sz = msg_data_sz(msg);
1432         hlen = msg_hdr_sz(msg);
1433         err = msg_errcode(msg);
1434
1435         /* Discard an empty non-errored message & try again */
1436         if ((!sz) && (!err)) {
1437                 tsk_advance_rx_queue(sk);
1438                 goto restart;
1439         }
1440
1441         /* Optionally capture sender's address & ancillary data of first msg */
1442         if (sz_copied == 0) {
1443                 set_orig_addr(m, msg);
1444                 res = tipc_sk_anc_data_recv(m, msg, tsk);
1445                 if (res)
1446                         goto exit;
1447         }
1448
1449         /* Capture message data (if valid) & compute return value (always) */
1450         if (!err) {
1451                 u32 offset = TIPC_SKB_CB(buf)->bytes_read;
1452                 u32 needed;
1453                 int sz_to_copy;
1454
1455                 sz -= offset;
1456                 needed = (buf_len - sz_copied);
1457                 sz_to_copy = min(sz, needed);
1458
1459                 res = skb_copy_datagram_msg(buf, hlen + offset, m, sz_to_copy);
1460                 if (res)
1461                         goto exit;
1462
1463                 sz_copied += sz_to_copy;
1464
1465                 if (sz_to_copy < sz) {
1466                         if (!(flags & MSG_PEEK))
1467                                 TIPC_SKB_CB(buf)->bytes_read =
1468                                         offset + sz_to_copy;
1469                         goto exit;
1470                 }
1471         } else {
1472                 if (sz_copied != 0)
1473                         goto exit; /* can't add error msg to valid data */
1474
1475                 if ((err == TIPC_CONN_SHUTDOWN) || m->msg_control)
1476                         res = 0;
1477                 else
1478                         res = -ECONNRESET;
1479         }
1480
1481         if (unlikely(flags & MSG_PEEK))
1482                 goto exit;
1483
1484         tsk->rcv_unacked += tsk_inc(tsk, hlen + sz);
1485         if (unlikely(tsk->rcv_unacked >= (tsk->rcv_win / 4)))
1486                 tipc_sk_send_ack(tsk);
1487         tsk_advance_rx_queue(sk);
1488
1489         /* Loop around if more data is required */
1490         if ((sz_copied < buf_len) &&    /* didn't get all requested data */
1491             (!skb_queue_empty(&sk->sk_receive_queue) ||
1492             (sz_copied < target)) &&    /* and more is ready or required */
1493             (!err))                     /* and haven't reached a FIN */
1494                 goto restart;
1495
1496 exit:
1497         release_sock(sk);
1498         return sz_copied ? sz_copied : res;
1499 }
1500
1501 /**
1502  * tipc_write_space - wake up thread if port congestion is released
1503  * @sk: socket
1504  */
1505 static void tipc_write_space(struct sock *sk)
1506 {
1507         struct socket_wq *wq;
1508
1509         rcu_read_lock();
1510         wq = rcu_dereference(sk->sk_wq);
1511         if (skwq_has_sleeper(wq))
1512                 wake_up_interruptible_sync_poll(&wq->wait, POLLOUT |
1513                                                 POLLWRNORM | POLLWRBAND);
1514         rcu_read_unlock();
1515 }
1516
1517 /**
1518  * tipc_data_ready - wake up threads to indicate messages have been received
1519  * @sk: socket
1520  * @len: the length of messages
1521  */
1522 static void tipc_data_ready(struct sock *sk)
1523 {
1524         struct socket_wq *wq;
1525
1526         rcu_read_lock();
1527         wq = rcu_dereference(sk->sk_wq);
1528         if (skwq_has_sleeper(wq))
1529                 wake_up_interruptible_sync_poll(&wq->wait, POLLIN |
1530                                                 POLLRDNORM | POLLRDBAND);
1531         rcu_read_unlock();
1532 }
1533
1534 static void tipc_sock_destruct(struct sock *sk)
1535 {
1536         __skb_queue_purge(&sk->sk_receive_queue);
1537 }
1538
1539 /**
1540  * filter_connect - Handle all incoming messages for a connection-based socket
1541  * @tsk: TIPC socket
1542  * @skb: pointer to message buffer. Set to NULL if buffer is consumed
1543  *
1544  * Returns true if everything ok, false otherwise
1545  */
1546 static bool filter_connect(struct tipc_sock *tsk, struct sk_buff *skb)
1547 {
1548         struct sock *sk = &tsk->sk;
1549         struct net *net = sock_net(sk);
1550         struct tipc_msg *hdr = buf_msg(skb);
1551
1552         if (unlikely(msg_mcast(hdr)))
1553                 return false;
1554
1555         switch (sk->sk_state) {
1556         case TIPC_CONNECTING:
1557                 /* Accept only ACK or NACK message */
1558                 if (unlikely(!msg_connected(hdr)))
1559                         return false;
1560
1561                 if (unlikely(msg_errcode(hdr))) {
1562                         tipc_set_sk_state(sk, TIPC_DISCONNECTING);
1563                         sk->sk_err = ECONNREFUSED;
1564                         return true;
1565                 }
1566
1567                 if (unlikely(!msg_isdata(hdr))) {
1568                         tipc_set_sk_state(sk, TIPC_DISCONNECTING);
1569                         sk->sk_err = EINVAL;
1570                         return true;
1571                 }
1572
1573                 tipc_sk_finish_conn(tsk, msg_origport(hdr), msg_orignode(hdr));
1574                 msg_set_importance(&tsk->phdr, msg_importance(hdr));
1575
1576                 /* If 'ACK+' message, add to socket receive queue */
1577                 if (msg_data_sz(hdr))
1578                         return true;
1579
1580                 /* If empty 'ACK-' message, wake up sleeping connect() */
1581                 if (waitqueue_active(sk_sleep(sk)))
1582                         wake_up_interruptible(sk_sleep(sk));
1583
1584                 /* 'ACK-' message is neither accepted nor rejected: */
1585                 msg_set_dest_droppable(hdr, 1);
1586                 return false;
1587
1588         case TIPC_OPEN:
1589         case TIPC_DISCONNECTING:
1590                 break;
1591         case TIPC_LISTEN:
1592                 /* Accept only SYN message */
1593                 if (!msg_connected(hdr) && !(msg_errcode(hdr)))
1594                         return true;
1595                 break;
1596         case TIPC_ESTABLISHED:
1597                 /* Accept only connection-based messages sent by peer */
1598                 if (unlikely(!tsk_peer_msg(tsk, hdr)))
1599                         return false;
1600
1601                 if (unlikely(msg_errcode(hdr))) {
1602                         tipc_set_sk_state(sk, TIPC_DISCONNECTING);
1603                         /* Let timer expire on it's own */
1604                         tipc_node_remove_conn(net, tsk_peer_node(tsk),
1605                                               tsk->portid);
1606                         sk->sk_state_change(sk);
1607                 }
1608                 return true;
1609         default:
1610                 pr_err("Unknown sk_state %u\n", sk->sk_state);
1611         }
1612
1613         return false;
1614 }
1615
1616 /**
1617  * rcvbuf_limit - get proper overload limit of socket receive queue
1618  * @sk: socket
1619  * @skb: message
1620  *
1621  * For connection oriented messages, irrespective of importance,
1622  * default queue limit is 2 MB.
1623  *
1624  * For connectionless messages, queue limits are based on message
1625  * importance as follows:
1626  *
1627  * TIPC_LOW_IMPORTANCE       (2 MB)
1628  * TIPC_MEDIUM_IMPORTANCE    (4 MB)
1629  * TIPC_HIGH_IMPORTANCE      (8 MB)
1630  * TIPC_CRITICAL_IMPORTANCE  (16 MB)
1631  *
1632  * Returns overload limit according to corresponding message importance
1633  */
1634 static unsigned int rcvbuf_limit(struct sock *sk, struct sk_buff *skb)
1635 {
1636         struct tipc_sock *tsk = tipc_sk(sk);
1637         struct tipc_msg *hdr = buf_msg(skb);
1638
1639         if (unlikely(!msg_connected(hdr)))
1640                 return sk->sk_rcvbuf << msg_importance(hdr);
1641
1642         if (likely(tsk->peer_caps & TIPC_BLOCK_FLOWCTL))
1643                 return sk->sk_rcvbuf;
1644
1645         return FLOWCTL_MSG_LIM;
1646 }
1647
1648 /**
1649  * filter_rcv - validate incoming message
1650  * @sk: socket
1651  * @skb: pointer to message.
1652  *
1653  * Enqueues message on receive queue if acceptable; optionally handles
1654  * disconnect indication for a connected socket.
1655  *
1656  * Called with socket lock already taken
1657  *
1658  * Returns true if message was added to socket receive queue, otherwise false
1659  */
1660 static bool filter_rcv(struct sock *sk, struct sk_buff *skb,
1661                        struct sk_buff_head *xmitq)
1662 {
1663         struct tipc_sock *tsk = tipc_sk(sk);
1664         struct tipc_msg *hdr = buf_msg(skb);
1665         unsigned int limit = rcvbuf_limit(sk, skb);
1666         int err = TIPC_OK;
1667         int usr = msg_user(hdr);
1668         u32 onode;
1669
1670         if (unlikely(msg_user(hdr) == CONN_MANAGER)) {
1671                 tipc_sk_proto_rcv(tsk, skb, xmitq);
1672                 return false;
1673         }
1674
1675         if (unlikely(usr == SOCK_WAKEUP)) {
1676                 onode = msg_orignode(hdr);
1677                 kfree_skb(skb);
1678                 u32_del(&tsk->cong_links, onode);
1679                 tsk->cong_link_cnt--;
1680                 sk->sk_write_space(sk);
1681                 return false;
1682         }
1683
1684         /* Drop if illegal message type */
1685         if (unlikely(msg_type(hdr) > TIPC_DIRECT_MSG)) {
1686                 kfree_skb(skb);
1687                 return false;
1688         }
1689
1690         /* Reject if wrong message type for current socket state */
1691         if (tipc_sk_type_connectionless(sk)) {
1692                 if (msg_connected(hdr)) {
1693                         err = TIPC_ERR_NO_PORT;
1694                         goto reject;
1695                 }
1696         } else if (unlikely(!filter_connect(tsk, skb))) {
1697                 err = TIPC_ERR_NO_PORT;
1698                 goto reject;
1699         }
1700
1701         /* Reject message if there isn't room to queue it */
1702         if (unlikely(sk_rmem_alloc_get(sk) + skb->truesize >= limit)) {
1703                 err = TIPC_ERR_OVERLOAD;
1704                 goto reject;
1705         }
1706
1707         /* Enqueue message */
1708         TIPC_SKB_CB(skb)->bytes_read = 0;
1709         __skb_queue_tail(&sk->sk_receive_queue, skb);
1710         skb_set_owner_r(skb, sk);
1711
1712         sk->sk_data_ready(sk);
1713         return true;
1714
1715 reject:
1716         if (tipc_msg_reverse(tsk_own_node(tsk), &skb, err))
1717                 __skb_queue_tail(xmitq, skb);
1718         return false;
1719 }
1720
1721 /**
1722  * tipc_backlog_rcv - handle incoming message from backlog queue
1723  * @sk: socket
1724  * @skb: message
1725  *
1726  * Caller must hold socket lock
1727  *
1728  * Returns 0
1729  */
1730 static int tipc_backlog_rcv(struct sock *sk, struct sk_buff *skb)
1731 {
1732         unsigned int truesize = skb->truesize;
1733         struct sk_buff_head xmitq;
1734         u32 dnode, selector;
1735
1736         __skb_queue_head_init(&xmitq);
1737
1738         if (likely(filter_rcv(sk, skb, &xmitq))) {
1739                 atomic_add(truesize, &tipc_sk(sk)->dupl_rcvcnt);
1740                 return 0;
1741         }
1742
1743         if (skb_queue_empty(&xmitq))
1744                 return 0;
1745
1746         /* Send response/rejected message */
1747         skb = __skb_dequeue(&xmitq);
1748         dnode = msg_destnode(buf_msg(skb));
1749         selector = msg_origport(buf_msg(skb));
1750         tipc_node_xmit_skb(sock_net(sk), skb, dnode, selector);
1751         return 0;
1752 }
1753
1754 /**
1755  * tipc_sk_enqueue - extract all buffers with destination 'dport' from
1756  *                   inputq and try adding them to socket or backlog queue
1757  * @inputq: list of incoming buffers with potentially different destinations
1758  * @sk: socket where the buffers should be enqueued
1759  * @dport: port number for the socket
1760  *
1761  * Caller must hold socket lock
1762  */
1763 static void tipc_sk_enqueue(struct sk_buff_head *inputq, struct sock *sk,
1764                             u32 dport, struct sk_buff_head *xmitq)
1765 {
1766         unsigned long time_limit = jiffies + 2;
1767         struct sk_buff *skb;
1768         unsigned int lim;
1769         atomic_t *dcnt;
1770         u32 onode;
1771
1772         while (skb_queue_len(inputq)) {
1773                 if (unlikely(time_after_eq(jiffies, time_limit)))
1774                         return;
1775
1776                 skb = tipc_skb_dequeue(inputq, dport);
1777                 if (unlikely(!skb))
1778                         return;
1779
1780                 /* Add message directly to receive queue if possible */
1781                 if (!sock_owned_by_user(sk)) {
1782                         filter_rcv(sk, skb, xmitq);
1783                         continue;
1784                 }
1785
1786                 /* Try backlog, compensating for double-counted bytes */
1787                 dcnt = &tipc_sk(sk)->dupl_rcvcnt;
1788                 if (!sk->sk_backlog.len)
1789                         atomic_set(dcnt, 0);
1790                 lim = rcvbuf_limit(sk, skb) + atomic_read(dcnt);
1791                 if (likely(!sk_add_backlog(sk, skb, lim)))
1792                         continue;
1793
1794                 /* Overload => reject message back to sender */
1795                 onode = tipc_own_addr(sock_net(sk));
1796                 if (tipc_msg_reverse(onode, &skb, TIPC_ERR_OVERLOAD))
1797                         __skb_queue_tail(xmitq, skb);
1798                 break;
1799         }
1800 }
1801
1802 /**
1803  * tipc_sk_rcv - handle a chain of incoming buffers
1804  * @inputq: buffer list containing the buffers
1805  * Consumes all buffers in list until inputq is empty
1806  * Note: may be called in multiple threads referring to the same queue
1807  */
1808 void tipc_sk_rcv(struct net *net, struct sk_buff_head *inputq)
1809 {
1810         struct sk_buff_head xmitq;
1811         u32 dnode, dport = 0;
1812         int err;
1813         struct tipc_sock *tsk;
1814         struct sock *sk;
1815         struct sk_buff *skb;
1816
1817         __skb_queue_head_init(&xmitq);
1818         while (skb_queue_len(inputq)) {
1819                 dport = tipc_skb_peek_port(inputq, dport);
1820                 tsk = tipc_sk_lookup(net, dport);
1821
1822                 if (likely(tsk)) {
1823                         sk = &tsk->sk;
1824                         if (likely(spin_trylock_bh(&sk->sk_lock.slock))) {
1825                                 tipc_sk_enqueue(inputq, sk, dport, &xmitq);
1826                                 spin_unlock_bh(&sk->sk_lock.slock);
1827                         }
1828                         /* Send pending response/rejected messages, if any */
1829                         while ((skb = __skb_dequeue(&xmitq))) {
1830                                 dnode = msg_destnode(buf_msg(skb));
1831                                 tipc_node_xmit_skb(net, skb, dnode, dport);
1832                         }
1833                         sock_put(sk);
1834                         continue;
1835                 }
1836
1837                 /* No destination socket => dequeue skb if still there */
1838                 skb = tipc_skb_dequeue(inputq, dport);
1839                 if (!skb)
1840                         return;
1841
1842                 /* Try secondary lookup if unresolved named message */
1843                 err = TIPC_ERR_NO_PORT;
1844                 if (tipc_msg_lookup_dest(net, skb, &err))
1845                         goto xmit;
1846
1847                 /* Prepare for message rejection */
1848                 if (!tipc_msg_reverse(tipc_own_addr(net), &skb, err))
1849                         continue;
1850 xmit:
1851                 dnode = msg_destnode(buf_msg(skb));
1852                 tipc_node_xmit_skb(net, skb, dnode, dport);
1853         }
1854 }
1855
1856 static int tipc_wait_for_connect(struct socket *sock, long *timeo_p)
1857 {
1858         DEFINE_WAIT_FUNC(wait, woken_wake_function);
1859         struct sock *sk = sock->sk;
1860         int done;
1861
1862         do {
1863                 int err = sock_error(sk);
1864                 if (err)
1865                         return err;
1866                 if (!*timeo_p)
1867                         return -ETIMEDOUT;
1868                 if (signal_pending(current))
1869                         return sock_intr_errno(*timeo_p);
1870
1871                 add_wait_queue(sk_sleep(sk), &wait);
1872                 done = sk_wait_event(sk, timeo_p,
1873                                      sk->sk_state != TIPC_CONNECTING, &wait);
1874                 remove_wait_queue(sk_sleep(sk), &wait);
1875         } while (!done);
1876         return 0;
1877 }
1878
1879 /**
1880  * tipc_connect - establish a connection to another TIPC port
1881  * @sock: socket structure
1882  * @dest: socket address for destination port
1883  * @destlen: size of socket address data structure
1884  * @flags: file-related flags associated with socket
1885  *
1886  * Returns 0 on success, errno otherwise
1887  */
1888 static int tipc_connect(struct socket *sock, struct sockaddr *dest,
1889                         int destlen, int flags)
1890 {
1891         struct sock *sk = sock->sk;
1892         struct tipc_sock *tsk = tipc_sk(sk);
1893         struct sockaddr_tipc *dst = (struct sockaddr_tipc *)dest;
1894         struct msghdr m = {NULL,};
1895         long timeout = (flags & O_NONBLOCK) ? 0 : tsk->conn_timeout;
1896         int previous;
1897         int res = 0;
1898
1899         lock_sock(sk);
1900
1901         /* DGRAM/RDM connect(), just save the destaddr */
1902         if (tipc_sk_type_connectionless(sk)) {
1903                 if (dst->family == AF_UNSPEC) {
1904                         memset(&tsk->peer, 0, sizeof(struct sockaddr_tipc));
1905                 } else if (destlen != sizeof(struct sockaddr_tipc)) {
1906                         res = -EINVAL;
1907                 } else {
1908                         memcpy(&tsk->peer, dest, destlen);
1909                 }
1910                 goto exit;
1911         }
1912
1913         /*
1914          * Reject connection attempt using multicast address
1915          *
1916          * Note: send_msg() validates the rest of the address fields,
1917          *       so there's no need to do it here
1918          */
1919         if (dst->addrtype == TIPC_ADDR_MCAST) {
1920                 res = -EINVAL;
1921                 goto exit;
1922         }
1923
1924         previous = sk->sk_state;
1925
1926         switch (sk->sk_state) {
1927         case TIPC_OPEN:
1928                 /* Send a 'SYN-' to destination */
1929                 m.msg_name = dest;
1930                 m.msg_namelen = destlen;
1931
1932                 /* If connect is in non-blocking case, set MSG_DONTWAIT to
1933                  * indicate send_msg() is never blocked.
1934                  */
1935                 if (!timeout)
1936                         m.msg_flags = MSG_DONTWAIT;
1937
1938                 res = __tipc_sendmsg(sock, &m, 0);
1939                 if ((res < 0) && (res != -EWOULDBLOCK))
1940                         goto exit;
1941
1942                 /* Just entered TIPC_CONNECTING state; the only
1943                  * difference is that return value in non-blocking
1944                  * case is EINPROGRESS, rather than EALREADY.
1945                  */
1946                 res = -EINPROGRESS;
1947                 /* fall thru' */
1948         case TIPC_CONNECTING:
1949                 if (!timeout) {
1950                         if (previous == TIPC_CONNECTING)
1951                                 res = -EALREADY;
1952                         goto exit;
1953                 }
1954                 timeout = msecs_to_jiffies(timeout);
1955                 /* Wait until an 'ACK' or 'RST' arrives, or a timeout occurs */
1956                 res = tipc_wait_for_connect(sock, &timeout);
1957                 break;
1958         case TIPC_ESTABLISHED:
1959                 res = -EISCONN;
1960                 break;
1961         default:
1962                 res = -EINVAL;
1963         }
1964
1965 exit:
1966         release_sock(sk);
1967         return res;
1968 }
1969
1970 /**
1971  * tipc_listen - allow socket to listen for incoming connections
1972  * @sock: socket structure
1973  * @len: (unused)
1974  *
1975  * Returns 0 on success, errno otherwise
1976  */
1977 static int tipc_listen(struct socket *sock, int len)
1978 {
1979         struct sock *sk = sock->sk;
1980         int res;
1981
1982         lock_sock(sk);
1983         res = tipc_set_sk_state(sk, TIPC_LISTEN);
1984         release_sock(sk);
1985
1986         return res;
1987 }
1988
1989 static int tipc_wait_for_accept(struct socket *sock, long timeo)
1990 {
1991         struct sock *sk = sock->sk;
1992         DEFINE_WAIT(wait);
1993         int err;
1994
1995         /* True wake-one mechanism for incoming connections: only
1996          * one process gets woken up, not the 'whole herd'.
1997          * Since we do not 'race & poll' for established sockets
1998          * anymore, the common case will execute the loop only once.
1999         */
2000         for (;;) {
2001                 prepare_to_wait_exclusive(sk_sleep(sk), &wait,
2002                                           TASK_INTERRUPTIBLE);
2003                 if (timeo && skb_queue_empty(&sk->sk_receive_queue)) {
2004                         release_sock(sk);
2005                         timeo = schedule_timeout(timeo);
2006                         lock_sock(sk);
2007                 }
2008                 err = 0;
2009                 if (!skb_queue_empty(&sk->sk_receive_queue))
2010                         break;
2011                 err = -EAGAIN;
2012                 if (!timeo)
2013                         break;
2014                 err = sock_intr_errno(timeo);
2015                 if (signal_pending(current))
2016                         break;
2017         }
2018         finish_wait(sk_sleep(sk), &wait);
2019         return err;
2020 }
2021
2022 /**
2023  * tipc_accept - wait for connection request
2024  * @sock: listening socket
2025  * @newsock: new socket that is to be connected
2026  * @flags: file-related flags associated with socket
2027  *
2028  * Returns 0 on success, errno otherwise
2029  */
2030 static int tipc_accept(struct socket *sock, struct socket *new_sock, int flags)
2031 {
2032         struct sock *new_sk, *sk = sock->sk;
2033         struct sk_buff *buf;
2034         struct tipc_sock *new_tsock;
2035         struct tipc_msg *msg;
2036         long timeo;
2037         int res;
2038
2039         lock_sock(sk);
2040
2041         if (sk->sk_state != TIPC_LISTEN) {
2042                 res = -EINVAL;
2043                 goto exit;
2044         }
2045         timeo = sock_rcvtimeo(sk, flags & O_NONBLOCK);
2046         res = tipc_wait_for_accept(sock, timeo);
2047         if (res)
2048                 goto exit;
2049
2050         buf = skb_peek(&sk->sk_receive_queue);
2051
2052         res = tipc_sk_create(sock_net(sock->sk), new_sock, 0, 0);
2053         if (res)
2054                 goto exit;
2055         security_sk_clone(sock->sk, new_sock->sk);
2056
2057         new_sk = new_sock->sk;
2058         new_tsock = tipc_sk(new_sk);
2059         msg = buf_msg(buf);
2060
2061         /* we lock on new_sk; but lockdep sees the lock on sk */
2062         lock_sock_nested(new_sk, SINGLE_DEPTH_NESTING);
2063
2064         /*
2065          * Reject any stray messages received by new socket
2066          * before the socket lock was taken (very, very unlikely)
2067          */
2068         tsk_rej_rx_queue(new_sk);
2069
2070         /* Connect new socket to it's peer */
2071         tipc_sk_finish_conn(new_tsock, msg_origport(msg), msg_orignode(msg));
2072
2073         tsk_set_importance(new_tsock, msg_importance(msg));
2074         if (msg_named(msg)) {
2075                 new_tsock->conn_type = msg_nametype(msg);
2076                 new_tsock->conn_instance = msg_nameinst(msg);
2077         }
2078
2079         /*
2080          * Respond to 'SYN-' by discarding it & returning 'ACK'-.
2081          * Respond to 'SYN+' by queuing it on new socket.
2082          */
2083         if (!msg_data_sz(msg)) {
2084                 struct msghdr m = {NULL,};
2085
2086                 tsk_advance_rx_queue(sk);
2087                 __tipc_sendstream(new_sock, &m, 0);
2088         } else {
2089                 __skb_dequeue(&sk->sk_receive_queue);
2090                 __skb_queue_head(&new_sk->sk_receive_queue, buf);
2091                 skb_set_owner_r(buf, new_sk);
2092         }
2093         release_sock(new_sk);
2094 exit:
2095         release_sock(sk);
2096         return res;
2097 }
2098
2099 /**
2100  * tipc_shutdown - shutdown socket connection
2101  * @sock: socket structure
2102  * @how: direction to close (must be SHUT_RDWR)
2103  *
2104  * Terminates connection (if necessary), then purges socket's receive queue.
2105  *
2106  * Returns 0 on success, errno otherwise
2107  */
2108 static int tipc_shutdown(struct socket *sock, int how)
2109 {
2110         struct sock *sk = sock->sk;
2111         int res;
2112
2113         if (how != SHUT_RDWR)
2114                 return -EINVAL;
2115
2116         lock_sock(sk);
2117
2118         __tipc_shutdown(sock, TIPC_CONN_SHUTDOWN);
2119         sk->sk_shutdown = SEND_SHUTDOWN;
2120
2121         if (sk->sk_state == TIPC_DISCONNECTING) {
2122                 /* Discard any unreceived messages */
2123                 __skb_queue_purge(&sk->sk_receive_queue);
2124
2125                 /* Wake up anyone sleeping in poll */
2126                 sk->sk_state_change(sk);
2127                 res = 0;
2128         } else {
2129                 res = -ENOTCONN;
2130         }
2131
2132         release_sock(sk);
2133         return res;
2134 }
2135
2136 static void tipc_sk_timeout(unsigned long data)
2137 {
2138         struct tipc_sock *tsk = (struct tipc_sock *)data;
2139         struct sock *sk = &tsk->sk;
2140         struct sk_buff *skb = NULL;
2141         u32 peer_port, peer_node;
2142         u32 own_node = tsk_own_node(tsk);
2143
2144         bh_lock_sock(sk);
2145         if (!tipc_sk_connected(sk)) {
2146                 bh_unlock_sock(sk);
2147                 goto exit;
2148         }
2149         peer_port = tsk_peer_port(tsk);
2150         peer_node = tsk_peer_node(tsk);
2151
2152         if (tsk->probe_unacked) {
2153                 if (!sock_owned_by_user(sk)) {
2154                         tipc_set_sk_state(sk, TIPC_DISCONNECTING);
2155                         tipc_node_remove_conn(sock_net(sk), tsk_peer_node(tsk),
2156                                               tsk_peer_port(tsk));
2157                         sk->sk_state_change(sk);
2158                 } else {
2159                         /* Try again later */
2160                         sk_reset_timer(sk, &sk->sk_timer, (HZ / 20));
2161                 }
2162
2163                 bh_unlock_sock(sk);
2164                 goto exit;
2165         }
2166
2167         skb = tipc_msg_create(CONN_MANAGER, CONN_PROBE,
2168                               INT_H_SIZE, 0, peer_node, own_node,
2169                               peer_port, tsk->portid, TIPC_OK);
2170         tsk->probe_unacked = true;
2171         sk_reset_timer(sk, &sk->sk_timer, jiffies + CONN_PROBING_INTERVAL);
2172         bh_unlock_sock(sk);
2173         if (skb)
2174                 tipc_node_xmit_skb(sock_net(sk), skb, peer_node, tsk->portid);
2175 exit:
2176         sock_put(sk);
2177 }
2178
2179 static int tipc_sk_publish(struct tipc_sock *tsk, uint scope,
2180                            struct tipc_name_seq const *seq)
2181 {
2182         struct sock *sk = &tsk->sk;
2183         struct net *net = sock_net(sk);
2184         struct publication *publ;
2185         u32 key;
2186
2187         if (tipc_sk_connected(sk))
2188                 return -EINVAL;
2189         key = tsk->portid + tsk->pub_count + 1;
2190         if (key == tsk->portid)
2191                 return -EADDRINUSE;
2192
2193         publ = tipc_nametbl_publish(net, seq->type, seq->lower, seq->upper,
2194                                     scope, tsk->portid, key);
2195         if (unlikely(!publ))
2196                 return -EINVAL;
2197
2198         list_add(&publ->pport_list, &tsk->publications);
2199         tsk->pub_count++;
2200         tsk->published = 1;
2201         return 0;
2202 }
2203
2204 static int tipc_sk_withdraw(struct tipc_sock *tsk, uint scope,
2205                             struct tipc_name_seq const *seq)
2206 {
2207         struct net *net = sock_net(&tsk->sk);
2208         struct publication *publ;
2209         struct publication *safe;
2210         int rc = -EINVAL;
2211
2212         list_for_each_entry_safe(publ, safe, &tsk->publications, pport_list) {
2213                 if (seq) {
2214                         if (publ->scope != scope)
2215                                 continue;
2216                         if (publ->type != seq->type)
2217                                 continue;
2218                         if (publ->lower != seq->lower)
2219                                 continue;
2220                         if (publ->upper != seq->upper)
2221                                 break;
2222                         tipc_nametbl_withdraw(net, publ->type, publ->lower,
2223                                               publ->ref, publ->key);
2224                         rc = 0;
2225                         break;
2226                 }
2227                 tipc_nametbl_withdraw(net, publ->type, publ->lower,
2228                                       publ->ref, publ->key);
2229                 rc = 0;
2230         }
2231         if (list_empty(&tsk->publications))
2232                 tsk->published = 0;
2233         return rc;
2234 }
2235
2236 /* tipc_sk_reinit: set non-zero address in all existing sockets
2237  *                 when we go from standalone to network mode.
2238  */
2239 void tipc_sk_reinit(struct net *net)
2240 {
2241         struct tipc_net *tn = net_generic(net, tipc_net_id);
2242         struct rhashtable_iter iter;
2243         struct tipc_sock *tsk;
2244         struct tipc_msg *msg;
2245
2246         rhashtable_walk_enter(&tn->sk_rht, &iter);
2247
2248         do {
2249                 tsk = ERR_PTR(rhashtable_walk_start(&iter));
2250                 if (tsk)
2251                         continue;
2252
2253                 while ((tsk = rhashtable_walk_next(&iter)) && !IS_ERR(tsk)) {
2254                         spin_lock_bh(&tsk->sk.sk_lock.slock);
2255                         msg = &tsk->phdr;
2256                         msg_set_prevnode(msg, tn->own_addr);
2257                         msg_set_orignode(msg, tn->own_addr);
2258                         spin_unlock_bh(&tsk->sk.sk_lock.slock);
2259                 }
2260
2261                 rhashtable_walk_stop(&iter);
2262         } while (tsk == ERR_PTR(-EAGAIN));
2263 }
2264
2265 static struct tipc_sock *tipc_sk_lookup(struct net *net, u32 portid)
2266 {
2267         struct tipc_net *tn = net_generic(net, tipc_net_id);
2268         struct tipc_sock *tsk;
2269
2270         rcu_read_lock();
2271         tsk = rhashtable_lookup_fast(&tn->sk_rht, &portid, tsk_rht_params);
2272         if (tsk)
2273                 sock_hold(&tsk->sk);
2274         rcu_read_unlock();
2275
2276         return tsk;
2277 }
2278
2279 static int tipc_sk_insert(struct tipc_sock *tsk)
2280 {
2281         struct sock *sk = &tsk->sk;
2282         struct net *net = sock_net(sk);
2283         struct tipc_net *tn = net_generic(net, tipc_net_id);
2284         u32 remaining = (TIPC_MAX_PORT - TIPC_MIN_PORT) + 1;
2285         u32 portid = prandom_u32() % remaining + TIPC_MIN_PORT;
2286
2287         while (remaining--) {
2288                 portid++;
2289                 if ((portid < TIPC_MIN_PORT) || (portid > TIPC_MAX_PORT))
2290                         portid = TIPC_MIN_PORT;
2291                 tsk->portid = portid;
2292                 sock_hold(&tsk->sk);
2293                 if (!rhashtable_lookup_insert_fast(&tn->sk_rht, &tsk->node,
2294                                                    tsk_rht_params))
2295                         return 0;
2296                 sock_put(&tsk->sk);
2297         }
2298
2299         return -1;
2300 }
2301
2302 static void tipc_sk_remove(struct tipc_sock *tsk)
2303 {
2304         struct sock *sk = &tsk->sk;
2305         struct tipc_net *tn = net_generic(sock_net(sk), tipc_net_id);
2306
2307         if (!rhashtable_remove_fast(&tn->sk_rht, &tsk->node, tsk_rht_params)) {
2308                 WARN_ON(atomic_read(&sk->sk_refcnt) == 1);
2309                 __sock_put(sk);
2310         }
2311 }
2312
2313 static const struct rhashtable_params tsk_rht_params = {
2314         .nelem_hint = 192,
2315         .head_offset = offsetof(struct tipc_sock, node),
2316         .key_offset = offsetof(struct tipc_sock, portid),
2317         .key_len = sizeof(u32), /* portid */
2318         .max_size = 1048576,
2319         .min_size = 256,
2320         .automatic_shrinking = true,
2321 };
2322
2323 int tipc_sk_rht_init(struct net *net)
2324 {
2325         struct tipc_net *tn = net_generic(net, tipc_net_id);
2326
2327         return rhashtable_init(&tn->sk_rht, &tsk_rht_params);
2328 }
2329
2330 void tipc_sk_rht_destroy(struct net *net)
2331 {
2332         struct tipc_net *tn = net_generic(net, tipc_net_id);
2333
2334         /* Wait for socket readers to complete */
2335         synchronize_net();
2336
2337         rhashtable_destroy(&tn->sk_rht);
2338 }
2339
2340 /**
2341  * tipc_setsockopt - set socket option
2342  * @sock: socket structure
2343  * @lvl: option level
2344  * @opt: option identifier
2345  * @ov: pointer to new option value
2346  * @ol: length of option value
2347  *
2348  * For stream sockets only, accepts and ignores all IPPROTO_TCP options
2349  * (to ease compatibility).
2350  *
2351  * Returns 0 on success, errno otherwise
2352  */
2353 static int tipc_setsockopt(struct socket *sock, int lvl, int opt,
2354                            char __user *ov, unsigned int ol)
2355 {
2356         struct sock *sk = sock->sk;
2357         struct tipc_sock *tsk = tipc_sk(sk);
2358         u32 value = 0;
2359         int res = 0;
2360
2361         if ((lvl == IPPROTO_TCP) && (sock->type == SOCK_STREAM))
2362                 return 0;
2363         if (lvl != SOL_TIPC)
2364                 return -ENOPROTOOPT;
2365
2366         switch (opt) {
2367         case TIPC_IMPORTANCE:
2368         case TIPC_SRC_DROPPABLE:
2369         case TIPC_DEST_DROPPABLE:
2370         case TIPC_CONN_TIMEOUT:
2371                 if (ol < sizeof(value))
2372                         return -EINVAL;
2373                 res = get_user(value, (u32 __user *)ov);
2374                 if (res)
2375                         return res;
2376                 break;
2377         default:
2378                 if (ov || ol)
2379                         return -EINVAL;
2380         }
2381
2382         lock_sock(sk);
2383
2384         switch (opt) {
2385         case TIPC_IMPORTANCE:
2386                 res = tsk_set_importance(tsk, value);
2387                 break;
2388         case TIPC_SRC_DROPPABLE:
2389                 if (sock->type != SOCK_STREAM)
2390                         tsk_set_unreliable(tsk, value);
2391                 else
2392                         res = -ENOPROTOOPT;
2393                 break;
2394         case TIPC_DEST_DROPPABLE:
2395                 tsk_set_unreturnable(tsk, value);
2396                 break;
2397         case TIPC_CONN_TIMEOUT:
2398                 tipc_sk(sk)->conn_timeout = value;
2399                 break;
2400         case TIPC_MCAST_BROADCAST:
2401                 tsk->mc_method.rcast = false;
2402                 tsk->mc_method.mandatory = true;
2403                 break;
2404         case TIPC_MCAST_REPLICAST:
2405                 tsk->mc_method.rcast = true;
2406                 tsk->mc_method.mandatory = true;
2407                 break;
2408         default:
2409                 res = -EINVAL;
2410         }
2411
2412         release_sock(sk);
2413
2414         return res;
2415 }
2416
2417 /**
2418  * tipc_getsockopt - get socket option
2419  * @sock: socket structure
2420  * @lvl: option level
2421  * @opt: option identifier
2422  * @ov: receptacle for option value
2423  * @ol: receptacle for length of option value
2424  *
2425  * For stream sockets only, returns 0 length result for all IPPROTO_TCP options
2426  * (to ease compatibility).
2427  *
2428  * Returns 0 on success, errno otherwise
2429  */
2430 static int tipc_getsockopt(struct socket *sock, int lvl, int opt,
2431                            char __user *ov, int __user *ol)
2432 {
2433         struct sock *sk = sock->sk;
2434         struct tipc_sock *tsk = tipc_sk(sk);
2435         int len;
2436         u32 value;
2437         int res;
2438
2439         if ((lvl == IPPROTO_TCP) && (sock->type == SOCK_STREAM))
2440                 return put_user(0, ol);
2441         if (lvl != SOL_TIPC)
2442                 return -ENOPROTOOPT;
2443         res = get_user(len, ol);
2444         if (res)
2445                 return res;
2446
2447         lock_sock(sk);
2448
2449         switch (opt) {
2450         case TIPC_IMPORTANCE:
2451                 value = tsk_importance(tsk);
2452                 break;
2453         case TIPC_SRC_DROPPABLE:
2454                 value = tsk_unreliable(tsk);
2455                 break;
2456         case TIPC_DEST_DROPPABLE:
2457                 value = tsk_unreturnable(tsk);
2458                 break;
2459         case TIPC_CONN_TIMEOUT:
2460                 value = tsk->conn_timeout;
2461                 /* no need to set "res", since already 0 at this point */
2462                 break;
2463         case TIPC_NODE_RECVQ_DEPTH:
2464                 value = 0; /* was tipc_queue_size, now obsolete */
2465                 break;
2466         case TIPC_SOCK_RECVQ_DEPTH:
2467                 value = skb_queue_len(&sk->sk_receive_queue);
2468                 break;
2469         default:
2470                 res = -EINVAL;
2471         }
2472
2473         release_sock(sk);
2474
2475         if (res)
2476                 return res;     /* "get" failed */
2477
2478         if (len < sizeof(value))
2479                 return -EINVAL;
2480
2481         if (copy_to_user(ov, &value, sizeof(value)))
2482                 return -EFAULT;
2483
2484         return put_user(sizeof(value), ol);
2485 }
2486
2487 static int tipc_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg)
2488 {
2489         struct sock *sk = sock->sk;
2490         struct tipc_sioc_ln_req lnr;
2491         void __user *argp = (void __user *)arg;
2492
2493         switch (cmd) {
2494         case SIOCGETLINKNAME:
2495                 if (copy_from_user(&lnr, argp, sizeof(lnr)))
2496                         return -EFAULT;
2497                 if (!tipc_node_get_linkname(sock_net(sk),
2498                                             lnr.bearer_id & 0xffff, lnr.peer,
2499                                             lnr.linkname, TIPC_MAX_LINK_NAME)) {
2500                         if (copy_to_user(argp, &lnr, sizeof(lnr)))
2501                                 return -EFAULT;
2502                         return 0;
2503                 }
2504                 return -EADDRNOTAVAIL;
2505         default:
2506                 return -ENOIOCTLCMD;
2507         }
2508 }
2509
2510 /* Protocol switches for the various types of TIPC sockets */
2511
2512 static const struct proto_ops msg_ops = {
2513         .owner          = THIS_MODULE,
2514         .family         = AF_TIPC,
2515         .release        = tipc_release,
2516         .bind           = tipc_bind,
2517         .connect        = tipc_connect,
2518         .socketpair     = sock_no_socketpair,
2519         .accept         = sock_no_accept,
2520         .getname        = tipc_getname,
2521         .poll           = tipc_poll,
2522         .ioctl          = tipc_ioctl,
2523         .listen         = sock_no_listen,
2524         .shutdown       = tipc_shutdown,
2525         .setsockopt     = tipc_setsockopt,
2526         .getsockopt     = tipc_getsockopt,
2527         .sendmsg        = tipc_sendmsg,
2528         .recvmsg        = tipc_recvmsg,
2529         .mmap           = sock_no_mmap,
2530         .sendpage       = sock_no_sendpage
2531 };
2532
2533 static const struct proto_ops packet_ops = {
2534         .owner          = THIS_MODULE,
2535         .family         = AF_TIPC,
2536         .release        = tipc_release,
2537         .bind           = tipc_bind,
2538         .connect        = tipc_connect,
2539         .socketpair     = sock_no_socketpair,
2540         .accept         = tipc_accept,
2541         .getname        = tipc_getname,
2542         .poll           = tipc_poll,
2543         .ioctl          = tipc_ioctl,
2544         .listen         = tipc_listen,
2545         .shutdown       = tipc_shutdown,
2546         .setsockopt     = tipc_setsockopt,
2547         .getsockopt     = tipc_getsockopt,
2548         .sendmsg        = tipc_send_packet,
2549         .recvmsg        = tipc_recvmsg,
2550         .mmap           = sock_no_mmap,
2551         .sendpage       = sock_no_sendpage
2552 };
2553
2554 static const struct proto_ops stream_ops = {
2555         .owner          = THIS_MODULE,
2556         .family         = AF_TIPC,
2557         .release        = tipc_release,
2558         .bind           = tipc_bind,
2559         .connect        = tipc_connect,
2560         .socketpair     = sock_no_socketpair,
2561         .accept         = tipc_accept,
2562         .getname        = tipc_getname,
2563         .poll           = tipc_poll,
2564         .ioctl          = tipc_ioctl,
2565         .listen         = tipc_listen,
2566         .shutdown       = tipc_shutdown,
2567         .setsockopt     = tipc_setsockopt,
2568         .getsockopt     = tipc_getsockopt,
2569         .sendmsg        = tipc_sendstream,
2570         .recvmsg        = tipc_recv_stream,
2571         .mmap           = sock_no_mmap,
2572         .sendpage       = sock_no_sendpage
2573 };
2574
2575 static const struct net_proto_family tipc_family_ops = {
2576         .owner          = THIS_MODULE,
2577         .family         = AF_TIPC,
2578         .create         = tipc_sk_create
2579 };
2580
2581 static struct proto tipc_proto = {
2582         .name           = "TIPC",
2583         .owner          = THIS_MODULE,
2584         .obj_size       = sizeof(struct tipc_sock),
2585         .sysctl_rmem    = sysctl_tipc_rmem
2586 };
2587
2588 /**
2589  * tipc_socket_init - initialize TIPC socket interface
2590  *
2591  * Returns 0 on success, errno otherwise
2592  */
2593 int tipc_socket_init(void)
2594 {
2595         int res;
2596
2597         res = proto_register(&tipc_proto, 1);
2598         if (res) {
2599                 pr_err("Failed to register TIPC protocol type\n");
2600                 goto out;
2601         }
2602
2603         res = sock_register(&tipc_family_ops);
2604         if (res) {
2605                 pr_err("Failed to register TIPC socket type\n");
2606                 proto_unregister(&tipc_proto);
2607                 goto out;
2608         }
2609  out:
2610         return res;
2611 }
2612
2613 /**
2614  * tipc_socket_stop - stop TIPC socket interface
2615  */
2616 void tipc_socket_stop(void)
2617 {
2618         sock_unregister(tipc_family_ops.family);
2619         proto_unregister(&tipc_proto);
2620 }
2621
2622 /* Caller should hold socket lock for the passed tipc socket. */
2623 static int __tipc_nl_add_sk_con(struct sk_buff *skb, struct tipc_sock *tsk)
2624 {
2625         u32 peer_node;
2626         u32 peer_port;
2627         struct nlattr *nest;
2628
2629         peer_node = tsk_peer_node(tsk);
2630         peer_port = tsk_peer_port(tsk);
2631
2632         nest = nla_nest_start(skb, TIPC_NLA_SOCK_CON);
2633
2634         if (nla_put_u32(skb, TIPC_NLA_CON_NODE, peer_node))
2635                 goto msg_full;
2636         if (nla_put_u32(skb, TIPC_NLA_CON_SOCK, peer_port))
2637                 goto msg_full;
2638
2639         if (tsk->conn_type != 0) {
2640                 if (nla_put_flag(skb, TIPC_NLA_CON_FLAG))
2641                         goto msg_full;
2642                 if (nla_put_u32(skb, TIPC_NLA_CON_TYPE, tsk->conn_type))
2643                         goto msg_full;
2644                 if (nla_put_u32(skb, TIPC_NLA_CON_INST, tsk->conn_instance))
2645                         goto msg_full;
2646         }
2647         nla_nest_end(skb, nest);
2648
2649         return 0;
2650
2651 msg_full:
2652         nla_nest_cancel(skb, nest);
2653
2654         return -EMSGSIZE;
2655 }
2656
2657 /* Caller should hold socket lock for the passed tipc socket. */
2658 static int __tipc_nl_add_sk(struct sk_buff *skb, struct netlink_callback *cb,
2659                             struct tipc_sock *tsk)
2660 {
2661         int err;
2662         void *hdr;
2663         struct nlattr *attrs;
2664         struct net *net = sock_net(skb->sk);
2665         struct tipc_net *tn = net_generic(net, tipc_net_id);
2666         struct sock *sk = &tsk->sk;
2667
2668         hdr = genlmsg_put(skb, NETLINK_CB(cb->skb).portid, cb->nlh->nlmsg_seq,
2669                           &tipc_genl_family, NLM_F_MULTI, TIPC_NL_SOCK_GET);
2670         if (!hdr)
2671                 goto msg_cancel;
2672
2673         attrs = nla_nest_start(skb, TIPC_NLA_SOCK);
2674         if (!attrs)
2675                 goto genlmsg_cancel;
2676         if (nla_put_u32(skb, TIPC_NLA_SOCK_REF, tsk->portid))
2677                 goto attr_msg_cancel;
2678         if (nla_put_u32(skb, TIPC_NLA_SOCK_ADDR, tn->own_addr))
2679                 goto attr_msg_cancel;
2680
2681         if (tipc_sk_connected(sk)) {
2682                 err = __tipc_nl_add_sk_con(skb, tsk);
2683                 if (err)
2684                         goto attr_msg_cancel;
2685         } else if (!list_empty(&tsk->publications)) {
2686                 if (nla_put_flag(skb, TIPC_NLA_SOCK_HAS_PUBL))
2687                         goto attr_msg_cancel;
2688         }
2689         nla_nest_end(skb, attrs);
2690         genlmsg_end(skb, hdr);
2691
2692         return 0;
2693
2694 attr_msg_cancel:
2695         nla_nest_cancel(skb, attrs);
2696 genlmsg_cancel:
2697         genlmsg_cancel(skb, hdr);
2698 msg_cancel:
2699         return -EMSGSIZE;
2700 }
2701
2702 int tipc_nl_sk_dump(struct sk_buff *skb, struct netlink_callback *cb)
2703 {
2704         int err;
2705         struct tipc_sock *tsk;
2706         const struct bucket_table *tbl;
2707         struct rhash_head *pos;
2708         struct net *net = sock_net(skb->sk);
2709         struct tipc_net *tn = net_generic(net, tipc_net_id);
2710         u32 tbl_id = cb->args[0];
2711         u32 prev_portid = cb->args[1];
2712
2713         rcu_read_lock();
2714         tbl = rht_dereference_rcu((&tn->sk_rht)->tbl, &tn->sk_rht);
2715         for (; tbl_id < tbl->size; tbl_id++) {
2716                 rht_for_each_entry_rcu(tsk, pos, tbl, tbl_id, node) {
2717                         spin_lock_bh(&tsk->sk.sk_lock.slock);
2718                         if (prev_portid && prev_portid != tsk->portid) {
2719                                 spin_unlock_bh(&tsk->sk.sk_lock.slock);
2720                                 continue;
2721                         }
2722
2723                         err = __tipc_nl_add_sk(skb, cb, tsk);
2724                         if (err) {
2725                                 prev_portid = tsk->portid;
2726                                 spin_unlock_bh(&tsk->sk.sk_lock.slock);
2727                                 goto out;
2728                         }
2729                         prev_portid = 0;
2730                         spin_unlock_bh(&tsk->sk.sk_lock.slock);
2731                 }
2732         }
2733 out:
2734         rcu_read_unlock();
2735         cb->args[0] = tbl_id;
2736         cb->args[1] = prev_portid;
2737
2738         return skb->len;
2739 }
2740
2741 /* Caller should hold socket lock for the passed tipc socket. */
2742 static int __tipc_nl_add_sk_publ(struct sk_buff *skb,
2743                                  struct netlink_callback *cb,
2744                                  struct publication *publ)
2745 {
2746         void *hdr;
2747         struct nlattr *attrs;
2748
2749         hdr = genlmsg_put(skb, NETLINK_CB(cb->skb).portid, cb->nlh->nlmsg_seq,
2750                           &tipc_genl_family, NLM_F_MULTI, TIPC_NL_PUBL_GET);
2751         if (!hdr)
2752                 goto msg_cancel;
2753
2754         attrs = nla_nest_start(skb, TIPC_NLA_PUBL);
2755         if (!attrs)
2756                 goto genlmsg_cancel;
2757
2758         if (nla_put_u32(skb, TIPC_NLA_PUBL_KEY, publ->key))
2759                 goto attr_msg_cancel;
2760         if (nla_put_u32(skb, TIPC_NLA_PUBL_TYPE, publ->type))
2761                 goto attr_msg_cancel;
2762         if (nla_put_u32(skb, TIPC_NLA_PUBL_LOWER, publ->lower))
2763                 goto attr_msg_cancel;
2764         if (nla_put_u32(skb, TIPC_NLA_PUBL_UPPER, publ->upper))
2765                 goto attr_msg_cancel;
2766
2767         nla_nest_end(skb, attrs);
2768         genlmsg_end(skb, hdr);
2769
2770         return 0;
2771
2772 attr_msg_cancel:
2773         nla_nest_cancel(skb, attrs);
2774 genlmsg_cancel:
2775         genlmsg_cancel(skb, hdr);
2776 msg_cancel:
2777         return -EMSGSIZE;
2778 }
2779
2780 /* Caller should hold socket lock for the passed tipc socket. */
2781 static int __tipc_nl_list_sk_publ(struct sk_buff *skb,
2782                                   struct netlink_callback *cb,
2783                                   struct tipc_sock *tsk, u32 *last_publ)
2784 {
2785         int err;
2786         struct publication *p;
2787
2788         if (*last_publ) {
2789                 list_for_each_entry(p, &tsk->publications, pport_list) {
2790                         if (p->key == *last_publ)
2791                                 break;
2792                 }
2793                 if (p->key != *last_publ) {
2794                         /* We never set seq or call nl_dump_check_consistent()
2795                          * this means that setting prev_seq here will cause the
2796                          * consistence check to fail in the netlink callback
2797                          * handler. Resulting in the last NLMSG_DONE message
2798                          * having the NLM_F_DUMP_INTR flag set.
2799                          */
2800                         cb->prev_seq = 1;
2801                         *last_publ = 0;
2802                         return -EPIPE;
2803                 }
2804         } else {
2805                 p = list_first_entry(&tsk->publications, struct publication,
2806                                      pport_list);
2807         }
2808
2809         list_for_each_entry_from(p, &tsk->publications, pport_list) {
2810                 err = __tipc_nl_add_sk_publ(skb, cb, p);
2811                 if (err) {
2812                         *last_publ = p->key;
2813                         return err;
2814                 }
2815         }
2816         *last_publ = 0;
2817
2818         return 0;
2819 }
2820
2821 int tipc_nl_publ_dump(struct sk_buff *skb, struct netlink_callback *cb)
2822 {
2823         int err;
2824         u32 tsk_portid = cb->args[0];
2825         u32 last_publ = cb->args[1];
2826         u32 done = cb->args[2];
2827         struct net *net = sock_net(skb->sk);
2828         struct tipc_sock *tsk;
2829
2830         if (!tsk_portid) {
2831                 struct nlattr **attrs;
2832                 struct nlattr *sock[TIPC_NLA_SOCK_MAX + 1];
2833
2834                 err = tipc_nlmsg_parse(cb->nlh, &attrs);
2835                 if (err)
2836                         return err;
2837
2838                 if (!attrs[TIPC_NLA_SOCK])
2839                         return -EINVAL;
2840
2841                 err = nla_parse_nested(sock, TIPC_NLA_SOCK_MAX,
2842                                        attrs[TIPC_NLA_SOCK],
2843                                        tipc_nl_sock_policy);
2844                 if (err)
2845                         return err;
2846
2847                 if (!sock[TIPC_NLA_SOCK_REF])
2848                         return -EINVAL;
2849
2850                 tsk_portid = nla_get_u32(sock[TIPC_NLA_SOCK_REF]);
2851         }
2852
2853         if (done)
2854                 return 0;
2855
2856         tsk = tipc_sk_lookup(net, tsk_portid);
2857         if (!tsk)
2858                 return -EINVAL;
2859
2860         lock_sock(&tsk->sk);
2861         err = __tipc_nl_list_sk_publ(skb, cb, tsk, &last_publ);
2862         if (!err)
2863                 done = 1;
2864         release_sock(&tsk->sk);
2865         sock_put(&tsk->sk);
2866
2867         cb->args[0] = tsk_portid;
2868         cb->args[1] = last_publ;
2869         cb->args[2] = done;
2870
2871         return skb->len;
2872 }