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