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