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