net/tipc/socket.c

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