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

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

Version: ~ [ linux-5.1.2 ] ~ [ linux-5.0.16 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.43 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.119 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.176 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.179 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.139 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.67 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-3.9.11 ] ~ [ linux-3.8.13 ] ~ [ linux-3.7.10 ] ~ [ linux-3.6.11 ] ~ [ linux-3.5.7 ] ~ [ linux-3.4.113 ] ~ [ linux-3.3.8 ] ~ [ linux-3.2.102 ] ~ [ linux-3.1.10 ] ~ [ linux-3.0.101 ] ~ [ linux-2.6.39.4 ] ~ [ linux-2.6.38.8 ] ~ [ linux-2.6.37.6 ] ~ [ linux-2.6.36.4 ] ~ [ linux-2.6.35.14 ] ~ [ linux-2.6.34.15 ] ~ [ linux-2.6.33.20 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

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

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

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

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

osdn.jp