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