~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/net/tipc/socket.c

Version: ~ [ linux-5.10-rc6 ] ~ [ linux-5.9.12 ] ~ [ linux-5.8.18 ] ~ [ linux-5.7.19 ] ~ [ linux-5.6.19 ] ~ [ linux-5.5.19 ] ~ [ linux-5.4.81 ] ~ [ linux-5.3.18 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.161 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.210 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.247 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.247 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.85 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

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

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp