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

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

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

~ [ 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