[TIPC] Moved configuration interface into tipc_config.h
[linux-2.6.git] / net / tipc / socket.c
1 /*
2  * net/tipc/socket.c: TIPC socket API
3  * 
4  * Copyright (c) 2003-2005, Ericsson Research Canada
5  * Copyright (c) 2004-2005, Wind River Systems
6  * Copyright (c) 2005-2006, Ericsson AB
7  * All rights reserved.
8  *
9  * Redistribution and use in source and binary forms, with or without 
10  * modification, are permitted provided that the following conditions are met:
11  *
12  * Redistributions of source code must retain the above copyright notice, this 
13  * list of conditions and the following disclaimer.
14  * Redistributions in binary form must reproduce the above copyright notice, 
15  * this list of conditions and the following disclaimer in the documentation 
16  * and/or other materials provided with the distribution.
17  * Neither the names of the copyright holders nor the names of its 
18  * contributors may be used to endorse or promote products derived from this 
19  * software without specific prior written permission.
20  *
21  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
22  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
23  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
24  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
25  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
26  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
27  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
28  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
29  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 
30  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
31  * POSSIBILITY OF SUCH DAMAGE.
32  */
33
34 #include <linux/module.h>
35 #include <linux/types.h>
36 #include <linux/net.h>
37 #include <linux/socket.h>
38 #include <linux/errno.h>
39 #include <linux/mm.h>
40 #include <linux/slab.h>
41 #include <linux/poll.h>
42 #include <linux/version.h>
43 #include <linux/fcntl.h>
44 #include <linux/version.h>
45 #include <asm/semaphore.h>
46 #include <asm/string.h>
47 #include <asm/atomic.h>
48 #include <net/sock.h>
49
50 #include <linux/tipc.h>
51 #include <linux/tipc_config.h>
52 #include <net/tipc/tipc_msg.h>
53 #include <net/tipc/tipc_port.h>
54
55 #include "core.h"
56
57 #define SS_LISTENING    -1      /* socket is listening */
58 #define SS_READY        -2      /* socket is connectionless */
59
60 #define OVERLOAD_LIMIT_BASE    5000
61
62 struct tipc_sock {
63         struct sock sk;
64         struct tipc_port *p;
65         struct semaphore sem;
66 };
67
68 #define tipc_sk(sk) ((struct tipc_sock*)sk)
69
70 static u32 dispatch(struct tipc_port *tport, struct sk_buff *buf);
71 static void wakeupdispatch(struct tipc_port *tport);
72
73 static struct proto_ops packet_ops;
74 static struct proto_ops stream_ops;
75 static struct proto_ops msg_ops;
76
77 static struct proto tipc_proto;
78
79 static int sockets_enabled = 0;
80
81 static atomic_t tipc_queue_size = ATOMIC_INIT(0);
82
83
84 /* 
85  * sock_lock(): Lock a port/socket pair. lock_sock() can 
86  * not be used here, since the same lock must protect ports 
87  * with non-socket interfaces.
88  * See net.c for description of locking policy.
89  */
90 static inline void sock_lock(struct tipc_sock* tsock)
91 {
92         spin_lock_bh(tsock->p->lock);       
93 }
94
95 /* 
96  * sock_unlock(): Unlock a port/socket pair
97  */
98 static inline void sock_unlock(struct tipc_sock* tsock)
99 {
100         spin_unlock_bh(tsock->p->lock);
101 }
102
103 /**
104  * pollmask - determine the current set of poll() events for a socket
105  * @sock: socket structure
106  * 
107  * TIPC sets the returned events as follows:
108  * a) POLLRDNORM and POLLIN are set if the socket's receive queue is non-empty
109  *    or if a connection-oriented socket is does not have an active connection
110  *    (i.e. a read operation will not block).
111  * b) POLLOUT is set except when a socket's connection has been terminated
112  *    (i.e. a write operation will not block).
113  * c) POLLHUP is set when a socket's connection has been terminated.
114  *
115  * IMPORTANT: The fact that a read or write operation will not block does NOT
116  * imply that the operation will succeed!
117  * 
118  * Returns pollmask value
119  */
120
121 static inline u32 pollmask(struct socket *sock)
122 {
123         u32 mask;
124
125         if ((skb_queue_len(&sock->sk->sk_receive_queue) != 0) ||
126             (sock->state == SS_UNCONNECTED) ||
127             (sock->state == SS_DISCONNECTING))
128                 mask = (POLLRDNORM | POLLIN);
129         else
130                 mask = 0;
131
132         if (sock->state == SS_DISCONNECTING) 
133                 mask |= POLLHUP;
134         else
135                 mask |= POLLOUT;
136
137         return mask;
138 }
139
140
141 /**
142  * advance_queue - discard first buffer in queue
143  * @tsock: TIPC socket
144  */
145
146 static inline void advance_queue(struct tipc_sock *tsock)
147 {
148         sock_lock(tsock);
149         buf_discard(skb_dequeue(&tsock->sk.sk_receive_queue));
150         sock_unlock(tsock);
151         atomic_dec(&tipc_queue_size);
152 }
153
154 /**
155  * tipc_create - create a TIPC socket
156  * @sock: pre-allocated socket structure
157  * @protocol: protocol indicator (must be 0)
158  * 
159  * This routine creates and attaches a 'struct sock' to the 'struct socket',
160  * then create and attaches a TIPC port to the 'struct sock' part.
161  *
162  * Returns 0 on success, errno otherwise
163  */
164 static int tipc_create(struct socket *sock, int protocol)
165 {
166         struct tipc_sock *tsock;
167         struct tipc_port *port;
168         struct sock *sk;
169         u32 ref;
170
171         if ((sock->type != SOCK_STREAM) && 
172             (sock->type != SOCK_SEQPACKET) &&
173             (sock->type != SOCK_DGRAM) &&
174             (sock->type != SOCK_RDM))
175                 return -EPROTOTYPE;
176
177         if (unlikely(protocol != 0))
178                 return -EPROTONOSUPPORT;
179
180         ref = tipc_createport_raw(0, &dispatch, &wakeupdispatch, TIPC_LOW_IMPORTANCE);
181         if (unlikely(!ref))
182                 return -ENOMEM;
183
184         sock->state = SS_UNCONNECTED;
185
186         switch (sock->type) {
187         case SOCK_STREAM:
188                 sock->ops = &stream_ops;
189                 break;
190         case SOCK_SEQPACKET:
191                 sock->ops = &packet_ops;
192                 break;
193         case SOCK_DGRAM:
194                 tipc_set_portunreliable(ref, 1);
195                 /* fall through */
196         case SOCK_RDM:
197                 tipc_set_portunreturnable(ref, 1);
198                 sock->ops = &msg_ops;
199                 sock->state = SS_READY;
200                 break;
201         }
202
203         sk = sk_alloc(AF_TIPC, GFP_KERNEL, &tipc_proto, 1);
204         if (!sk) {
205                 tipc_deleteport(ref);
206                 return -ENOMEM;
207         }
208
209         sock_init_data(sock, sk);
210         init_waitqueue_head(sk->sk_sleep);
211         sk->sk_rcvtimeo = 8 * HZ;   /* default connect timeout = 8s */
212
213         tsock = tipc_sk(sk);
214         port = tipc_get_port(ref);
215
216         tsock->p = port;
217         port->usr_handle = tsock;
218
219         init_MUTEX(&tsock->sem);
220
221         dbg("sock_create: %x\n",tsock);
222
223         atomic_inc(&tipc_user_count);
224
225         return 0;
226 }
227
228 /**
229  * release - destroy a TIPC socket
230  * @sock: socket to destroy
231  *
232  * This routine cleans up any messages that are still queued on the socket.
233  * For DGRAM and RDM socket types, all queued messages are rejected.
234  * For SEQPACKET and STREAM socket types, the first message is rejected
235  * and any others are discarded.  (If the first message on a STREAM socket
236  * is partially-read, it is discarded and the next one is rejected instead.)
237  * 
238  * NOTE: Rejected messages are not necessarily returned to the sender!  They
239  * are returned or discarded according to the "destination droppable" setting
240  * specified for the message by the sender.
241  *
242  * Returns 0 on success, errno otherwise
243  */
244
245 static int release(struct socket *sock)
246 {
247         struct tipc_sock *tsock = tipc_sk(sock->sk);
248         struct sock *sk = sock->sk;
249         int res = TIPC_OK;
250         struct sk_buff *buf;
251
252         dbg("sock_delete: %x\n",tsock);
253         if (!tsock)
254                 return 0;
255         down_interruptible(&tsock->sem);
256         if (!sock->sk) {
257                 up(&tsock->sem);
258                 return 0;
259         }
260         
261         /* Reject unreceived messages, unless no longer connected */
262
263         while (sock->state != SS_DISCONNECTING) {
264                 sock_lock(tsock);
265                 buf = skb_dequeue(&sk->sk_receive_queue);
266                 if (!buf)
267                         tsock->p->usr_handle = 0;
268                 sock_unlock(tsock);
269                 if (!buf)
270                         break;
271                 if (TIPC_SKB_CB(buf)->handle != msg_data(buf_msg(buf)))
272                         buf_discard(buf);
273                 else
274                         tipc_reject_msg(buf, TIPC_ERR_NO_PORT);
275                 atomic_dec(&tipc_queue_size);
276         }
277
278         /* Delete TIPC port */
279
280         res = tipc_deleteport(tsock->p->ref);
281         sock->sk = NULL;
282
283         /* Discard any remaining messages */
284
285         while ((buf = skb_dequeue(&sk->sk_receive_queue))) {
286                 buf_discard(buf);
287                 atomic_dec(&tipc_queue_size);
288         }
289
290         up(&tsock->sem);
291
292         sock_put(sk);
293
294         atomic_dec(&tipc_user_count);
295         return res;
296 }
297
298 /**
299  * bind - associate or disassocate TIPC name(s) with a socket
300  * @sock: socket structure
301  * @uaddr: socket address describing name(s) and desired operation
302  * @uaddr_len: size of socket address data structure
303  * 
304  * Name and name sequence binding is indicated using a positive scope value;
305  * a negative scope value unbinds the specified name.  Specifying no name
306  * (i.e. a socket address length of 0) unbinds all names from the socket.
307  * 
308  * Returns 0 on success, errno otherwise
309  */
310
311 static int bind(struct socket *sock, struct sockaddr *uaddr, int uaddr_len)
312 {
313         struct tipc_sock *tsock = tipc_sk(sock->sk);
314         struct sockaddr_tipc *addr = (struct sockaddr_tipc *)uaddr;
315         int res;
316
317         if (down_interruptible(&tsock->sem))
318                 return -ERESTARTSYS;
319         
320         if (unlikely(!uaddr_len)) {
321                 res = tipc_withdraw(tsock->p->ref, 0, 0);
322                 goto exit;
323         }
324
325         if (uaddr_len < sizeof(struct sockaddr_tipc)) {
326                 res = -EINVAL;
327                 goto exit;
328         }
329
330         if (addr->family != AF_TIPC) {
331                 res = -EAFNOSUPPORT;
332                 goto exit;
333         }
334         if (addr->addrtype == TIPC_ADDR_NAME)
335                 addr->addr.nameseq.upper = addr->addr.nameseq.lower;
336         else if (addr->addrtype != TIPC_ADDR_NAMESEQ) {
337                 res = -EAFNOSUPPORT;
338                 goto exit;
339         }
340         
341         if (addr->scope > 0)
342                 res = tipc_publish(tsock->p->ref, addr->scope,
343                                    &addr->addr.nameseq);
344         else
345                 res = tipc_withdraw(tsock->p->ref, -addr->scope,
346                                     &addr->addr.nameseq);
347 exit:
348         up(&tsock->sem);
349         return res;
350 }
351
352 /** 
353  * get_name - get port ID of socket or peer socket
354  * @sock: socket structure
355  * @uaddr: area for returned socket address
356  * @uaddr_len: area for returned length of socket address
357  * @peer: 0 to obtain socket name, 1 to obtain peer socket name
358  * 
359  * Returns 0 on success, errno otherwise
360  */
361
362 static int get_name(struct socket *sock, struct sockaddr *uaddr, 
363                     int *uaddr_len, int peer)
364 {
365         struct tipc_sock *tsock = tipc_sk(sock->sk);
366         struct sockaddr_tipc *addr = (struct sockaddr_tipc *)uaddr;
367         u32 res;
368
369         if (down_interruptible(&tsock->sem))
370                 return -ERESTARTSYS;
371
372         *uaddr_len = sizeof(*addr);
373         addr->addrtype = TIPC_ADDR_ID;
374         addr->family = AF_TIPC;
375         addr->scope = 0;
376         if (peer)
377                 res = tipc_peer(tsock->p->ref, &addr->addr.id);
378         else
379                 res = tipc_ownidentity(tsock->p->ref, &addr->addr.id);
380         addr->addr.name.domain = 0;
381
382         up(&tsock->sem);
383         return res;
384 }
385
386 /**
387  * poll - read and possibly block on pollmask
388  * @file: file structure associated with the socket
389  * @sock: socket for which to calculate the poll bits
390  * @wait: ???
391  *
392  * Returns the pollmask
393  */
394
395 static unsigned int poll(struct file *file, struct socket *sock, 
396                          poll_table *wait)
397 {
398         poll_wait(file, sock->sk->sk_sleep, wait);
399         /* NEED LOCK HERE? */
400         return pollmask(sock);
401 }
402
403 /** 
404  * dest_name_check - verify user is permitted to send to specified port name
405  * @dest: destination address
406  * @m: descriptor for message to be sent
407  * 
408  * Prevents restricted configuration commands from being issued by
409  * unauthorized users.
410  * 
411  * Returns 0 if permission is granted, otherwise errno
412  */
413
414 static inline int dest_name_check(struct sockaddr_tipc *dest, struct msghdr *m)
415 {
416         struct tipc_cfg_msg_hdr hdr;
417
418         if (likely(dest->addr.name.name.type >= TIPC_RESERVED_TYPES))
419                 return 0;
420         if (likely(dest->addr.name.name.type == TIPC_TOP_SRV))
421                 return 0;
422
423         if (likely(dest->addr.name.name.type != TIPC_CFG_SRV))
424                 return -EACCES;
425
426         if (copy_from_user(&hdr, m->msg_iov[0].iov_base, sizeof(hdr)))
427                 return -EFAULT;
428         if ((ntohs(hdr.tcm_type) & 0xC000) & (!capable(CAP_NET_ADMIN)))
429                 return -EACCES;
430         
431         return 0;
432 }
433
434 /**
435  * send_msg - send message in connectionless manner
436  * @iocb: (unused)
437  * @sock: socket structure
438  * @m: message to send
439  * @total_len: (unused)
440  * 
441  * Message must have an destination specified explicitly.
442  * Used for SOCK_RDM and SOCK_DGRAM messages, 
443  * and for 'SYN' messages on SOCK_SEQPACKET and SOCK_STREAM connections.
444  * (Note: 'SYN+' is prohibited on SOCK_STREAM.)
445  * 
446  * Returns the number of bytes sent on success, or errno otherwise
447  */
448
449 static int send_msg(struct kiocb *iocb, struct socket *sock,
450                     struct msghdr *m, size_t total_len)
451 {
452         struct tipc_sock *tsock = tipc_sk(sock->sk);
453         struct sockaddr_tipc *dest = (struct sockaddr_tipc *)m->msg_name;
454         struct sk_buff *buf;
455         int needs_conn;
456         int res = -EINVAL;
457
458         if (unlikely(!dest))
459                 return -EDESTADDRREQ;
460         if (unlikely(dest->family != AF_TIPC))
461                 return -EINVAL;
462
463         needs_conn = (sock->state != SS_READY);
464         if (unlikely(needs_conn)) {
465                 if (sock->state == SS_LISTENING)
466                         return -EPIPE;
467                 if (sock->state != SS_UNCONNECTED)
468                         return -EISCONN;
469                 if ((tsock->p->published) ||
470                     ((sock->type == SOCK_STREAM) && (total_len != 0)))
471                         return -EOPNOTSUPP;
472         }
473
474         if (down_interruptible(&tsock->sem))
475                 return -ERESTARTSYS;
476
477         if (needs_conn) {
478
479                 /* Abort any pending connection attempts (very unlikely) */
480
481                 while ((buf = skb_dequeue(&sock->sk->sk_receive_queue))) {
482                         tipc_reject_msg(buf, TIPC_ERR_NO_PORT);
483                         atomic_dec(&tipc_queue_size);
484                 }
485
486                 sock->state = SS_CONNECTING;
487         }
488
489         do {
490                 if (dest->addrtype == TIPC_ADDR_NAME) {
491                         if ((res = dest_name_check(dest, m)))
492                                 goto exit;
493                         res = tipc_send2name(tsock->p->ref,
494                                              &dest->addr.name.name,
495                                              dest->addr.name.domain, 
496                                              m->msg_iovlen,
497                                              m->msg_iov);
498                 }
499                 else if (dest->addrtype == TIPC_ADDR_ID) {
500                         res = tipc_send2port(tsock->p->ref,
501                                              &dest->addr.id,
502                                              m->msg_iovlen,
503                                              m->msg_iov);
504                 }
505                 else if (dest->addrtype == TIPC_ADDR_MCAST) {
506                         if (needs_conn) {
507                                 res = -EOPNOTSUPP;
508                                 goto exit;
509                         }
510                         if ((res = dest_name_check(dest, m)))
511                                 goto exit;
512                         res = tipc_multicast(tsock->p->ref,
513                                              &dest->addr.nameseq,
514                                              0,
515                                              m->msg_iovlen,
516                                              m->msg_iov);
517                 }
518                 if (likely(res != -ELINKCONG)) {
519 exit:                                
520                         up(&tsock->sem);
521                         return res;
522                 }
523                 if (m->msg_flags & MSG_DONTWAIT) {
524                         res = -EWOULDBLOCK;
525                         goto exit;
526                 }
527                 if (wait_event_interruptible(*sock->sk->sk_sleep,
528                                              !tsock->p->congested)) {
529                     res = -ERESTARTSYS;
530                     goto exit;
531                 }
532         } while (1);
533 }
534
535 /** 
536  * send_packet - send a connection-oriented message
537  * @iocb: (unused)
538  * @sock: socket structure
539  * @m: message to send
540  * @total_len: (unused)
541  * 
542  * Used for SOCK_SEQPACKET messages and SOCK_STREAM data.
543  * 
544  * Returns the number of bytes sent on success, or errno otherwise
545  */
546
547 static int send_packet(struct kiocb *iocb, struct socket *sock,
548                        struct msghdr *m, size_t total_len)
549 {
550         struct tipc_sock *tsock = tipc_sk(sock->sk);
551         struct sockaddr_tipc *dest = (struct sockaddr_tipc *)m->msg_name;
552         int res;
553
554         /* Handle implied connection establishment */
555
556         if (unlikely(dest))
557                 return send_msg(iocb, sock, m, total_len);
558
559         if (down_interruptible(&tsock->sem)) {
560                 return -ERESTARTSYS;
561         }
562
563         if (unlikely(sock->state != SS_CONNECTED)) {
564                 if (sock->state == SS_DISCONNECTING)
565                         res = -EPIPE;   
566                 else
567                         res = -ENOTCONN;
568                 goto exit;
569         }
570
571         do {
572                 res = tipc_send(tsock->p->ref, m->msg_iovlen, m->msg_iov);
573                 if (likely(res != -ELINKCONG)) {
574 exit:
575                         up(&tsock->sem);
576                         return res;
577                 }
578                 if (m->msg_flags & MSG_DONTWAIT) {
579                         res = -EWOULDBLOCK;
580                         goto exit;
581                 }
582                 if (wait_event_interruptible(*sock->sk->sk_sleep,
583                                              !tsock->p->congested)) {
584                     res = -ERESTARTSYS;
585                     goto exit;
586                 }
587         } while (1);
588 }
589
590 /** 
591  * send_stream - send stream-oriented data
592  * @iocb: (unused)
593  * @sock: socket structure
594  * @m: data to send
595  * @total_len: total length of data to be sent
596  * 
597  * Used for SOCK_STREAM data.
598  * 
599  * Returns the number of bytes sent on success, or errno otherwise
600  */
601
602
603 static int send_stream(struct kiocb *iocb, struct socket *sock,
604                        struct msghdr *m, size_t total_len)
605 {
606         struct msghdr my_msg;
607         struct iovec my_iov;
608         struct iovec *curr_iov;
609         int curr_iovlen;
610         char __user *curr_start;
611         int curr_left;
612         int bytes_to_send;
613         int res;
614         
615         if (likely(total_len <= TIPC_MAX_USER_MSG_SIZE))
616                 return send_packet(iocb, sock, m, total_len);
617
618         /* Can only send large data streams if already connected */
619
620         if (unlikely(sock->state != SS_CONNECTED)) {
621                 if (sock->state == SS_DISCONNECTING)
622                         return -EPIPE;   
623                 else
624                         return -ENOTCONN;
625         }
626
627         /* 
628          * Send each iovec entry using one or more messages
629          *
630          * Note: This algorithm is good for the most likely case 
631          * (i.e. one large iovec entry), but could be improved to pass sets
632          * of small iovec entries into send_packet().
633          */
634
635         my_msg = *m;
636         curr_iov = my_msg.msg_iov;
637         curr_iovlen = my_msg.msg_iovlen;
638         my_msg.msg_iov = &my_iov;
639         my_msg.msg_iovlen = 1;
640
641         while (curr_iovlen--) {
642                 curr_start = curr_iov->iov_base;
643                 curr_left = curr_iov->iov_len;
644
645                 while (curr_left) {
646                         bytes_to_send = (curr_left < TIPC_MAX_USER_MSG_SIZE)
647                                 ? curr_left : TIPC_MAX_USER_MSG_SIZE;
648                         my_iov.iov_base = curr_start;
649                         my_iov.iov_len = bytes_to_send;
650                         if ((res = send_packet(iocb, sock, &my_msg, 0)) < 0)
651                                 return res;
652                         curr_left -= bytes_to_send;
653                         curr_start += bytes_to_send;
654                 }
655
656                 curr_iov++;
657         }
658
659         return total_len;
660 }
661
662 /**
663  * auto_connect - complete connection setup to a remote port
664  * @sock: socket structure
665  * @tsock: TIPC-specific socket structure
666  * @msg: peer's response message
667  * 
668  * Returns 0 on success, errno otherwise
669  */
670
671 static int auto_connect(struct socket *sock, struct tipc_sock *tsock, 
672                         struct tipc_msg *msg)
673 {
674         struct tipc_portid peer;
675
676         if (msg_errcode(msg)) {
677                 sock->state = SS_DISCONNECTING;
678                 return -ECONNREFUSED;
679         }
680
681         peer.ref = msg_origport(msg);
682         peer.node = msg_orignode(msg);
683         tipc_connect2port(tsock->p->ref, &peer);
684         tipc_set_portimportance(tsock->p->ref, msg_importance(msg));
685         sock->state = SS_CONNECTED;
686         return 0;
687 }
688
689 /**
690  * set_orig_addr - capture sender's address for received message
691  * @m: descriptor for message info
692  * @msg: received message header
693  * 
694  * Note: Address is not captured if not requested by receiver.
695  */
696
697 static inline void set_orig_addr(struct msghdr *m, struct tipc_msg *msg)
698 {
699         struct sockaddr_tipc *addr = (struct sockaddr_tipc *)m->msg_name;
700
701         if (addr) {
702                 addr->family = AF_TIPC;
703                 addr->addrtype = TIPC_ADDR_ID;
704                 addr->addr.id.ref = msg_origport(msg);
705                 addr->addr.id.node = msg_orignode(msg);
706                 addr->addr.name.domain = 0;     /* could leave uninitialized */
707                 addr->scope = 0;                /* could leave uninitialized */
708                 m->msg_namelen = sizeof(struct sockaddr_tipc);
709         }
710 }
711
712 /**
713  * anc_data_recv - optionally capture ancillary data for received message 
714  * @m: descriptor for message info
715  * @msg: received message header
716  * @tport: TIPC port associated with message
717  * 
718  * Note: Ancillary data is not captured if not requested by receiver.
719  * 
720  * Returns 0 if successful, otherwise errno
721  */
722
723 static inline int anc_data_recv(struct msghdr *m, struct tipc_msg *msg, 
724                                 struct tipc_port *tport)
725 {
726         u32 anc_data[3];
727         u32 err;
728         u32 dest_type;
729         int res;
730
731         if (likely(m->msg_controllen == 0))
732                 return 0;
733
734         /* Optionally capture errored message object(s) */
735
736         err = msg ? msg_errcode(msg) : 0;
737         if (unlikely(err)) {
738                 anc_data[0] = err;
739                 anc_data[1] = msg_data_sz(msg);
740                 if ((res = put_cmsg(m, SOL_SOCKET, TIPC_ERRINFO, 8, anc_data)))
741                         return res;
742                 if (anc_data[1] &&
743                     (res = put_cmsg(m, SOL_SOCKET, TIPC_RETDATA, anc_data[1], 
744                                     msg_data(msg))))
745                         return res;
746         }
747
748         /* Optionally capture message destination object */
749
750         dest_type = msg ? msg_type(msg) : TIPC_DIRECT_MSG;
751         switch (dest_type) {
752         case TIPC_NAMED_MSG:
753                 anc_data[0] = msg_nametype(msg);
754                 anc_data[1] = msg_namelower(msg);
755                 anc_data[2] = msg_namelower(msg);
756                 break;
757         case TIPC_MCAST_MSG:
758                 anc_data[0] = msg_nametype(msg);
759                 anc_data[1] = msg_namelower(msg);
760                 anc_data[2] = msg_nameupper(msg);
761                 break;
762         case TIPC_CONN_MSG:
763                 anc_data[0] = tport->conn_type;
764                 anc_data[1] = tport->conn_instance;
765                 anc_data[2] = tport->conn_instance;
766                 break;
767         default:
768                 anc_data[0] = 0;
769         }
770         if (anc_data[0] &&
771             (res = put_cmsg(m, SOL_SOCKET, TIPC_DESTNAME, 12, anc_data)))
772                 return res;
773
774         return 0;
775 }
776
777 /** 
778  * recv_msg - receive packet-oriented message
779  * @iocb: (unused)
780  * @m: descriptor for message info
781  * @buf_len: total size of user buffer area
782  * @flags: receive flags
783  * 
784  * Used for SOCK_DGRAM, SOCK_RDM, and SOCK_SEQPACKET messages.
785  * If the complete message doesn't fit in user area, truncate it.
786  *
787  * Returns size of returned message data, errno otherwise
788  */
789
790 static int recv_msg(struct kiocb *iocb, struct socket *sock,
791                     struct msghdr *m, size_t buf_len, int flags)
792 {
793         struct tipc_sock *tsock = tipc_sk(sock->sk);
794         struct sk_buff *buf;
795         struct tipc_msg *msg;
796         unsigned int q_len;
797         unsigned int sz;
798         u32 err;
799         int res;
800
801         /* Currently doesn't support receiving into multiple iovec entries */
802
803         if (m->msg_iovlen != 1)
804                 return -EOPNOTSUPP;
805
806         /* Catch invalid receive attempts */
807
808         if (unlikely(!buf_len))
809                 return -EINVAL;
810
811         if (sock->type == SOCK_SEQPACKET) {
812                 if (unlikely(sock->state == SS_UNCONNECTED))
813                         return -ENOTCONN;
814                 if (unlikely((sock->state == SS_DISCONNECTING) && 
815                              (skb_queue_len(&sock->sk->sk_receive_queue) == 0)))
816                         return -ENOTCONN;
817         }
818
819         /* Look for a message in receive queue; wait if necessary */
820
821         if (unlikely(down_interruptible(&tsock->sem)))
822                 return -ERESTARTSYS;
823
824 restart:
825         if (unlikely((skb_queue_len(&sock->sk->sk_receive_queue) == 0) &&
826                      (flags & MSG_DONTWAIT))) {
827                 res = -EWOULDBLOCK;
828                 goto exit;
829         }
830
831         if ((res = wait_event_interruptible(
832                 *sock->sk->sk_sleep, 
833                 ((q_len = skb_queue_len(&sock->sk->sk_receive_queue)) ||
834                  (sock->state == SS_DISCONNECTING))) )) {
835                 goto exit;
836         }
837
838         /* Catch attempt to receive on an already terminated connection */
839         /* [THIS CHECK MAY OVERLAP WITH AN EARLIER CHECK] */
840
841         if (!q_len) {
842                 res = -ENOTCONN;
843                 goto exit;
844         }
845
846         /* Get access to first message in receive queue */
847
848         buf = skb_peek(&sock->sk->sk_receive_queue);
849         msg = buf_msg(buf);
850         sz = msg_data_sz(msg);
851         err = msg_errcode(msg);
852
853         /* Complete connection setup for an implied connect */
854
855         if (unlikely(sock->state == SS_CONNECTING)) {
856                 if ((res = auto_connect(sock, tsock, msg)))
857                         goto exit;
858         }
859
860         /* Discard an empty non-errored message & try again */
861
862         if ((!sz) && (!err)) {
863                 advance_queue(tsock);
864                 goto restart;
865         }
866
867         /* Capture sender's address (optional) */
868
869         set_orig_addr(m, msg);
870
871         /* Capture ancillary data (optional) */
872
873         if ((res = anc_data_recv(m, msg, tsock->p)))
874                 goto exit;
875
876         /* Capture message data (if valid) & compute return value (always) */
877         
878         if (!err) {
879                 if (unlikely(buf_len < sz)) {
880                         sz = buf_len;
881                         m->msg_flags |= MSG_TRUNC;
882                 }
883                 if (unlikely(copy_to_user(m->msg_iov->iov_base, msg_data(msg),
884                                           sz))) {
885                         res = -EFAULT;
886                         goto exit;
887                 }
888                 res = sz;
889         } else {
890                 if ((sock->state == SS_READY) ||
891                     ((err == TIPC_CONN_SHUTDOWN) || m->msg_control))
892                         res = 0;
893                 else
894                         res = -ECONNRESET;
895         }
896
897         /* Consume received message (optional) */
898
899         if (likely(!(flags & MSG_PEEK))) {
900                 if (unlikely(++tsock->p->conn_unacked >= TIPC_FLOW_CONTROL_WIN))
901                         tipc_acknowledge(tsock->p->ref, tsock->p->conn_unacked);
902                 advance_queue(tsock);
903         }
904 exit:
905         up(&tsock->sem);
906         return res;
907 }
908
909 /** 
910  * recv_stream - receive stream-oriented data
911  * @iocb: (unused)
912  * @m: descriptor for message info
913  * @buf_len: total size of user buffer area
914  * @flags: receive flags
915  * 
916  * Used for SOCK_STREAM messages only.  If not enough data is available 
917  * will optionally wait for more; never truncates data.
918  *
919  * Returns size of returned message data, errno otherwise
920  */
921
922 static int recv_stream(struct kiocb *iocb, struct socket *sock,
923                        struct msghdr *m, size_t buf_len, int flags)
924 {
925         struct tipc_sock *tsock = tipc_sk(sock->sk);
926         struct sk_buff *buf;
927         struct tipc_msg *msg;
928         unsigned int q_len;
929         unsigned int sz;
930         int sz_to_copy;
931         int sz_copied = 0;
932         int needed;
933         char *crs = m->msg_iov->iov_base;
934         unsigned char *buf_crs;
935         u32 err;
936         int res;
937
938         /* Currently doesn't support receiving into multiple iovec entries */
939
940         if (m->msg_iovlen != 1)
941                 return -EOPNOTSUPP;
942
943         /* Catch invalid receive attempts */
944
945         if (unlikely(!buf_len))
946                 return -EINVAL;
947
948         if (unlikely(sock->state == SS_DISCONNECTING)) {
949                 if (skb_queue_len(&sock->sk->sk_receive_queue) == 0)
950                         return -ENOTCONN;
951         } else if (unlikely(sock->state != SS_CONNECTED))
952                 return -ENOTCONN;
953
954         /* Look for a message in receive queue; wait if necessary */
955
956         if (unlikely(down_interruptible(&tsock->sem)))
957                 return -ERESTARTSYS;
958
959 restart:
960         if (unlikely((skb_queue_len(&sock->sk->sk_receive_queue) == 0) &&
961                      (flags & MSG_DONTWAIT))) {
962                 res = (sz_copied == 0) ? -EWOULDBLOCK : 0;
963                 goto exit;
964         }
965
966         if ((res = wait_event_interruptible(
967                 *sock->sk->sk_sleep, 
968                 ((q_len = skb_queue_len(&sock->sk->sk_receive_queue)) ||
969                  (sock->state == SS_DISCONNECTING))) )) {
970                 goto exit;
971         }
972
973         /* Catch attempt to receive on an already terminated connection */
974         /* [THIS CHECK MAY OVERLAP WITH AN EARLIER CHECK] */
975
976         if (!q_len) {
977                 res = -ENOTCONN;
978                 goto exit;
979         }
980
981         /* Get access to first message in receive queue */
982
983         buf = skb_peek(&sock->sk->sk_receive_queue);
984         msg = buf_msg(buf);
985         sz = msg_data_sz(msg);
986         err = msg_errcode(msg);
987
988         /* Discard an empty non-errored message & try again */
989
990         if ((!sz) && (!err)) {
991                 advance_queue(tsock);
992                 goto restart;
993         }
994
995         /* Optionally capture sender's address & ancillary data of first msg */
996
997         if (sz_copied == 0) {
998                 set_orig_addr(m, msg);
999                 if ((res = anc_data_recv(m, msg, tsock->p)))
1000                         goto exit;
1001         }
1002
1003         /* Capture message data (if valid) & compute return value (always) */
1004         
1005         if (!err) {
1006                 buf_crs = (unsigned char *)(TIPC_SKB_CB(buf)->handle);
1007                 sz = buf->tail - buf_crs;
1008
1009                 needed = (buf_len - sz_copied);
1010                 sz_to_copy = (sz <= needed) ? sz : needed;
1011                 if (unlikely(copy_to_user(crs, buf_crs, sz_to_copy))) {
1012                         res = -EFAULT;
1013                         goto exit;
1014                 }
1015                 sz_copied += sz_to_copy;
1016
1017                 if (sz_to_copy < sz) {
1018                         if (!(flags & MSG_PEEK))
1019                                 TIPC_SKB_CB(buf)->handle = buf_crs + sz_to_copy;
1020                         goto exit;
1021                 }
1022
1023                 crs += sz_to_copy;
1024         } else {
1025                 if (sz_copied != 0)
1026                         goto exit; /* can't add error msg to valid data */
1027
1028                 if ((err == TIPC_CONN_SHUTDOWN) || m->msg_control)
1029                         res = 0;
1030                 else
1031                         res = -ECONNRESET;
1032         }
1033
1034         /* Consume received message (optional) */
1035
1036         if (likely(!(flags & MSG_PEEK))) {
1037                 if (unlikely(++tsock->p->conn_unacked >= TIPC_FLOW_CONTROL_WIN))
1038                         tipc_acknowledge(tsock->p->ref, tsock->p->conn_unacked);
1039                 advance_queue(tsock);
1040         }
1041
1042         /* Loop around if more data is required */
1043
1044         if ((sz_copied < buf_len)    /* didn't get all requested data */ 
1045             && (flags & MSG_WAITALL) /* ... and need to wait for more */
1046             && (!(flags & MSG_PEEK)) /* ... and aren't just peeking at data */
1047             && (!err)                /* ... and haven't reached a FIN */
1048             )
1049                 goto restart;
1050
1051 exit:
1052         up(&tsock->sem);
1053         return res ? res : sz_copied;
1054 }
1055
1056 /**
1057  * queue_overloaded - test if queue overload condition exists
1058  * @queue_size: current size of queue
1059  * @base: nominal maximum size of queue
1060  * @msg: message to be added to queue
1061  * 
1062  * Returns 1 if queue is currently overloaded, 0 otherwise
1063  */
1064
1065 static int queue_overloaded(u32 queue_size, u32 base, struct tipc_msg *msg)
1066 {
1067         u32 threshold;
1068         u32 imp = msg_importance(msg);
1069
1070         if (imp == TIPC_LOW_IMPORTANCE)
1071                 threshold = base;
1072         else if (imp == TIPC_MEDIUM_IMPORTANCE)
1073                 threshold = base * 2;
1074         else if (imp == TIPC_HIGH_IMPORTANCE)
1075                 threshold = base * 100;
1076         else
1077                 return 0;
1078
1079         if (msg_connected(msg))
1080                 threshold *= 4;
1081
1082         return (queue_size > threshold);
1083 }
1084
1085 /** 
1086  * async_disconnect - wrapper function used to disconnect port
1087  * @portref: TIPC port reference (passed as pointer-sized value)
1088  */
1089
1090 static void async_disconnect(unsigned long portref)
1091 {
1092         tipc_disconnect((u32)portref);
1093 }
1094
1095 /** 
1096  * dispatch - handle arriving message
1097  * @tport: TIPC port that received message
1098  * @buf: message
1099  * 
1100  * Called with port locked.  Must not take socket lock to avoid deadlock risk.
1101  * 
1102  * Returns TIPC error status code (TIPC_OK if message is not to be rejected)
1103  */
1104
1105 static u32 dispatch(struct tipc_port *tport, struct sk_buff *buf)
1106 {
1107         struct tipc_msg *msg = buf_msg(buf);
1108         struct tipc_sock *tsock = (struct tipc_sock *)tport->usr_handle;
1109         struct socket *sock;
1110         u32 recv_q_len;
1111
1112         /* Reject message if socket is closing */
1113
1114         if (!tsock)
1115                 return TIPC_ERR_NO_PORT;
1116
1117         /* Reject message if it is wrong sort of message for socket */
1118
1119         /*
1120          * WOULD IT BE BETTER TO JUST DISCARD THESE MESSAGES INSTEAD?
1121          * "NO PORT" ISN'T REALLY THE RIGHT ERROR CODE, AND THERE MAY
1122          * BE SECURITY IMPLICATIONS INHERENT IN REJECTING INVALID TRAFFIC
1123          */
1124         sock = tsock->sk.sk_socket;
1125         if (sock->state == SS_READY) {
1126                 if (msg_connected(msg)) {
1127                         msg_dbg(msg, "dispatch filter 1\n");
1128                         return TIPC_ERR_NO_PORT;
1129                 }
1130         } else {
1131                 if (msg_mcast(msg)) {
1132                         msg_dbg(msg, "dispatch filter 2\n");
1133                         return TIPC_ERR_NO_PORT;
1134                 }
1135                 if (sock->state == SS_CONNECTED) {
1136                         if (!msg_connected(msg)) {
1137                                 msg_dbg(msg, "dispatch filter 3\n");
1138                                 return TIPC_ERR_NO_PORT;
1139                         }
1140                 }
1141                 else if (sock->state == SS_CONNECTING) {
1142                         if (!msg_connected(msg) && (msg_errcode(msg) == 0)) {
1143                                 msg_dbg(msg, "dispatch filter 4\n");
1144                                 return TIPC_ERR_NO_PORT;
1145                         }
1146                 } 
1147                 else if (sock->state == SS_LISTENING) {
1148                         if (msg_connected(msg) || msg_errcode(msg)) {
1149                                 msg_dbg(msg, "dispatch filter 5\n");
1150                                 return TIPC_ERR_NO_PORT;
1151                         }
1152                 } 
1153                 else if (sock->state == SS_DISCONNECTING) {
1154                         msg_dbg(msg, "dispatch filter 6\n");
1155                         return TIPC_ERR_NO_PORT;
1156                 }
1157                 else /* (sock->state == SS_UNCONNECTED) */ {
1158                         if (msg_connected(msg) || msg_errcode(msg)) {
1159                                 msg_dbg(msg, "dispatch filter 7\n");
1160                                 return TIPC_ERR_NO_PORT;
1161                         }
1162                 }
1163         }
1164
1165         /* Reject message if there isn't room to queue it */
1166
1167         if (unlikely((u32)atomic_read(&tipc_queue_size) > 
1168                      OVERLOAD_LIMIT_BASE)) {
1169                 if (queue_overloaded(atomic_read(&tipc_queue_size), 
1170                                      OVERLOAD_LIMIT_BASE, msg))
1171                         return TIPC_ERR_OVERLOAD;
1172         }
1173         recv_q_len = skb_queue_len(&tsock->sk.sk_receive_queue);
1174         if (unlikely(recv_q_len > (OVERLOAD_LIMIT_BASE / 2))) {
1175                 if (queue_overloaded(recv_q_len, 
1176                                      OVERLOAD_LIMIT_BASE / 2, msg)) 
1177                         return TIPC_ERR_OVERLOAD;
1178         }
1179
1180         /* Initiate connection termination for an incoming 'FIN' */
1181
1182         if (unlikely(msg_errcode(msg) && (sock->state == SS_CONNECTED))) {
1183                 sock->state = SS_DISCONNECTING;
1184                 /* Note: Use signal since port lock is already taken! */
1185                 k_signal((Handler)async_disconnect, tport->ref);
1186         }
1187
1188         /* Enqueue message (finally!) */
1189
1190         msg_dbg(msg,"<DISP<: ");
1191         TIPC_SKB_CB(buf)->handle = msg_data(msg);
1192         atomic_inc(&tipc_queue_size);
1193         skb_queue_tail(&sock->sk->sk_receive_queue, buf);
1194
1195         wake_up_interruptible(sock->sk->sk_sleep);
1196         return TIPC_OK;
1197 }
1198
1199 /** 
1200  * wakeupdispatch - wake up port after congestion
1201  * @tport: port to wakeup
1202  * 
1203  * Called with port lock on.
1204  */
1205
1206 static void wakeupdispatch(struct tipc_port *tport)
1207 {
1208         struct tipc_sock *tsock = (struct tipc_sock *)tport->usr_handle;
1209
1210         wake_up_interruptible(tsock->sk.sk_sleep);
1211 }
1212
1213 /**
1214  * connect - establish a connection to another TIPC port
1215  * @sock: socket structure
1216  * @dest: socket address for destination port
1217  * @destlen: size of socket address data structure
1218  * @flags: (unused)
1219  *
1220  * Returns 0 on success, errno otherwise
1221  */
1222
1223 static int connect(struct socket *sock, struct sockaddr *dest, int destlen, 
1224                    int flags)
1225 {
1226    struct tipc_sock *tsock = tipc_sk(sock->sk);
1227    struct sockaddr_tipc *dst = (struct sockaddr_tipc *)dest;
1228    struct msghdr m = {0,};
1229    struct sk_buff *buf;
1230    struct tipc_msg *msg;
1231    int res;
1232
1233    /* For now, TIPC does not allow use of connect() with DGRAM or RDM types */
1234
1235    if (sock->state == SS_READY)
1236            return -EOPNOTSUPP;
1237
1238    /* MOVE THE REST OF THIS ERROR CHECKING TO send_msg()? */
1239    if (sock->state == SS_LISTENING)
1240            return -EOPNOTSUPP;
1241    if (sock->state == SS_CONNECTING)
1242            return -EALREADY;
1243    if (sock->state != SS_UNCONNECTED)
1244            return -EISCONN;
1245
1246    if ((dst->family != AF_TIPC) ||
1247        ((dst->addrtype != TIPC_ADDR_NAME) && (dst->addrtype != TIPC_ADDR_ID)))
1248            return -EINVAL;
1249
1250    /* Send a 'SYN-' to destination */
1251
1252    m.msg_name = dest;
1253    if ((res = send_msg(0, sock, &m, 0)) < 0) {
1254            sock->state = SS_DISCONNECTING;
1255            return res;
1256    }
1257
1258    if (down_interruptible(&tsock->sem)) 
1259            return -ERESTARTSYS;
1260         
1261    /* Wait for destination's 'ACK' response */
1262
1263    res = wait_event_interruptible_timeout(*sock->sk->sk_sleep,
1264                                           skb_queue_len(&sock->sk->sk_receive_queue),
1265                                           sock->sk->sk_rcvtimeo);
1266    buf = skb_peek(&sock->sk->sk_receive_queue);
1267    if (res > 0) {
1268            msg = buf_msg(buf);
1269            res = auto_connect(sock, tsock, msg);
1270            if (!res) {
1271                    if (dst->addrtype == TIPC_ADDR_NAME) {
1272                            tsock->p->conn_type = dst->addr.name.name.type;
1273                            tsock->p->conn_instance = dst->addr.name.name.instance;
1274                    }
1275                    if (!msg_data_sz(msg))
1276                            advance_queue(tsock);
1277            }
1278    } else {
1279            if (res == 0) {
1280                    res = -ETIMEDOUT;
1281            } else
1282                    { /* leave "res" unchanged */ }
1283            sock->state = SS_DISCONNECTING;
1284    }
1285
1286    up(&tsock->sem);
1287    return res;
1288 }
1289
1290 /** 
1291  * listen - allow socket to listen for incoming connections
1292  * @sock: socket structure
1293  * @len: (unused)
1294  * 
1295  * Returns 0 on success, errno otherwise
1296  */
1297
1298 static int listen(struct socket *sock, int len)
1299 {
1300         /* REQUIRES SOCKET LOCKING OF SOME SORT? */
1301
1302         if (sock->state == SS_READY)
1303                 return -EOPNOTSUPP;
1304         if (sock->state != SS_UNCONNECTED)
1305                 return -EINVAL;
1306         sock->state = SS_LISTENING;
1307         return 0;
1308 }
1309
1310 /** 
1311  * accept - wait for connection request
1312  * @sock: listening socket
1313  * @newsock: new socket that is to be connected
1314  * @flags: file-related flags associated with socket
1315  * 
1316  * Returns 0 on success, errno otherwise
1317  */
1318
1319 static int accept(struct socket *sock, struct socket *newsock, int flags)
1320 {
1321         struct tipc_sock *tsock = tipc_sk(sock->sk);
1322         struct sk_buff *buf;
1323         int res = -EFAULT;
1324
1325         if (sock->state == SS_READY)
1326                 return -EOPNOTSUPP;
1327         if (sock->state != SS_LISTENING)
1328                 return -EINVAL;
1329         
1330         if (unlikely((skb_queue_len(&sock->sk->sk_receive_queue) == 0) && 
1331                      (flags & O_NONBLOCK)))
1332                 return -EWOULDBLOCK;
1333
1334         if (down_interruptible(&tsock->sem))
1335                 return -ERESTARTSYS;
1336
1337         if (wait_event_interruptible(*sock->sk->sk_sleep, 
1338                                      skb_queue_len(&sock->sk->sk_receive_queue))) {
1339                 res = -ERESTARTSYS;
1340                 goto exit;
1341         }
1342         buf = skb_peek(&sock->sk->sk_receive_queue);
1343
1344         res = tipc_create(newsock, 0);
1345         if (!res) {
1346                 struct tipc_sock *new_tsock = tipc_sk(newsock->sk);
1347                 struct tipc_portid id;
1348                 struct tipc_msg *msg = buf_msg(buf);
1349                 u32 new_ref = new_tsock->p->ref;
1350
1351                 id.ref = msg_origport(msg);
1352                 id.node = msg_orignode(msg);
1353                 tipc_connect2port(new_ref, &id);
1354                 newsock->state = SS_CONNECTED;
1355
1356                 tipc_set_portimportance(new_ref, msg_importance(msg));
1357                 if (msg_named(msg)) {
1358                         new_tsock->p->conn_type = msg_nametype(msg);
1359                         new_tsock->p->conn_instance = msg_nameinst(msg);
1360                 }
1361
1362                /* 
1363                  * Respond to 'SYN-' by discarding it & returning 'ACK'-.
1364                  * Respond to 'SYN+' by queuing it on new socket.
1365                  */
1366
1367                 msg_dbg(msg,"<ACC<: ");
1368                 if (!msg_data_sz(msg)) {
1369                         struct msghdr m = {0,};
1370
1371                         send_packet(0, newsock, &m, 0);      
1372                         advance_queue(tsock);
1373                 } else {
1374                         sock_lock(tsock);
1375                         skb_dequeue(&sock->sk->sk_receive_queue);
1376                         sock_unlock(tsock);
1377                         skb_queue_head(&newsock->sk->sk_receive_queue, buf);
1378                 }
1379         }
1380 exit:
1381         up(&tsock->sem);
1382         return res;
1383 }
1384
1385 /**
1386  * shutdown - shutdown socket connection
1387  * @sock: socket structure
1388  * @how: direction to close (always treated as read + write)
1389  *
1390  * Terminates connection (if necessary), then purges socket's receive queue.
1391  * 
1392  * Returns 0 on success, errno otherwise
1393  */
1394
1395 static int shutdown(struct socket *sock, int how)
1396 {
1397         struct tipc_sock* tsock = tipc_sk(sock->sk);
1398         struct sk_buff *buf;
1399         int res;
1400
1401         /* Could return -EINVAL for an invalid "how", but why bother? */
1402
1403         if (down_interruptible(&tsock->sem))
1404                 return -ERESTARTSYS;
1405
1406         sock_lock(tsock);
1407
1408         switch (sock->state) {
1409         case SS_CONNECTED:
1410
1411                 /* Send 'FIN+' or 'FIN-' message to peer */
1412
1413                 sock_unlock(tsock);
1414 restart:
1415                 if ((buf = skb_dequeue(&sock->sk->sk_receive_queue))) {
1416                         atomic_dec(&tipc_queue_size);
1417                         if (TIPC_SKB_CB(buf)->handle != msg_data(buf_msg(buf))) {
1418                                 buf_discard(buf);
1419                                 goto restart;
1420                         }
1421                         tipc_reject_msg(buf, TIPC_CONN_SHUTDOWN);
1422                 }
1423                 else {
1424                         tipc_shutdown(tsock->p->ref);
1425                 }
1426                 sock_lock(tsock);
1427
1428                 /* fall through */
1429
1430         case SS_DISCONNECTING:
1431
1432                 /* Discard any unreceived messages */
1433
1434                 while ((buf = skb_dequeue(&sock->sk->sk_receive_queue))) {
1435                         atomic_dec(&tipc_queue_size);
1436                         buf_discard(buf);
1437                 }
1438                 tsock->p->conn_unacked = 0;
1439
1440                 /* fall through */
1441
1442         case SS_CONNECTING:
1443                 sock->state = SS_DISCONNECTING;
1444                 res = 0;
1445                 break;
1446
1447         default:
1448                 res = -ENOTCONN;
1449         }
1450
1451         sock_unlock(tsock);
1452
1453         up(&tsock->sem);
1454         return res;
1455 }
1456
1457 /**
1458  * setsockopt - set socket option
1459  * @sock: socket structure
1460  * @lvl: option level
1461  * @opt: option identifier
1462  * @ov: pointer to new option value
1463  * @ol: length of option value
1464  * 
1465  * For stream sockets only, accepts and ignores all IPPROTO_TCP options 
1466  * (to ease compatibility).
1467  * 
1468  * Returns 0 on success, errno otherwise
1469  */
1470
1471 static int setsockopt(struct socket *sock, int lvl, int opt, char *ov, int ol)
1472 {
1473         struct tipc_sock *tsock = tipc_sk(sock->sk);
1474         u32 value;
1475         int res;
1476
1477         if ((lvl == IPPROTO_TCP) && (sock->type == SOCK_STREAM))
1478                 return 0;
1479         if (lvl != SOL_TIPC)
1480                 return -ENOPROTOOPT;
1481         if (ol < sizeof(value))
1482                 return -EINVAL;
1483         if ((res = get_user(value, (u32 *)ov)))
1484                 return res;
1485
1486         if (down_interruptible(&tsock->sem)) 
1487                 return -ERESTARTSYS;
1488         
1489         switch (opt) {
1490         case TIPC_IMPORTANCE:
1491                 res = tipc_set_portimportance(tsock->p->ref, value);
1492                 break;
1493         case TIPC_SRC_DROPPABLE:
1494                 if (sock->type != SOCK_STREAM)
1495                         res = tipc_set_portunreliable(tsock->p->ref, value);
1496                 else 
1497                         res = -ENOPROTOOPT;
1498                 break;
1499         case TIPC_DEST_DROPPABLE:
1500                 res = tipc_set_portunreturnable(tsock->p->ref, value);
1501                 break;
1502         case TIPC_CONN_TIMEOUT:
1503                 sock->sk->sk_rcvtimeo = (value * HZ / 1000);
1504                 break;
1505         default:
1506                 res = -EINVAL;
1507         }
1508
1509         up(&tsock->sem);
1510         return res;
1511 }
1512
1513 /**
1514  * getsockopt - get socket option
1515  * @sock: socket structure
1516  * @lvl: option level
1517  * @opt: option identifier
1518  * @ov: receptacle for option value
1519  * @ol: receptacle for length of option value
1520  * 
1521  * For stream sockets only, returns 0 length result for all IPPROTO_TCP options 
1522  * (to ease compatibility).
1523  * 
1524  * Returns 0 on success, errno otherwise
1525  */
1526
1527 static int getsockopt(struct socket *sock, int lvl, int opt, char *ov, int *ol)
1528 {
1529         struct tipc_sock *tsock = tipc_sk(sock->sk);
1530         int len;
1531         u32 value;
1532         int res;
1533
1534         if ((lvl == IPPROTO_TCP) && (sock->type == SOCK_STREAM))
1535                 return put_user(0, ol);
1536         if (lvl != SOL_TIPC)
1537                 return -ENOPROTOOPT;
1538         if ((res = get_user(len, ol)))
1539                 return res;
1540
1541         if (down_interruptible(&tsock->sem)) 
1542                 return -ERESTARTSYS;
1543
1544         switch (opt) {
1545         case TIPC_IMPORTANCE:
1546                 res = tipc_portimportance(tsock->p->ref, &value);
1547                 break;
1548         case TIPC_SRC_DROPPABLE:
1549                 res = tipc_portunreliable(tsock->p->ref, &value);
1550                 break;
1551         case TIPC_DEST_DROPPABLE:
1552                 res = tipc_portunreturnable(tsock->p->ref, &value);
1553                 break;
1554         case TIPC_CONN_TIMEOUT:
1555                 value = (sock->sk->sk_rcvtimeo * 1000) / HZ;
1556                 break;
1557         default:
1558                 res = -EINVAL;
1559         }
1560
1561         if (res) {
1562                 /* "get" failed */
1563         }
1564         else if (len < sizeof(value)) {
1565                 res = -EINVAL;
1566         }
1567         else if ((res = copy_to_user(ov, &value, sizeof(value)))) {
1568                 /* couldn't return value */
1569         }
1570         else {
1571                 res = put_user(sizeof(value), ol);
1572         }
1573
1574         up(&tsock->sem);
1575         return res;
1576 }
1577
1578 /**
1579  * Placeholders for non-implemented functionality
1580  * 
1581  * Returns error code (POSIX-compliant where defined)
1582  */
1583
1584 static int ioctl(struct socket *s, u32 cmd, unsigned long arg)
1585 {
1586         return -EINVAL;
1587 }
1588
1589 static int no_mmap(struct file *file, struct socket *sock,
1590                    struct vm_area_struct *vma)
1591 {
1592         return -EINVAL;
1593 }
1594 static ssize_t no_sendpage(struct socket *sock, struct page *page,
1595                            int offset, size_t size, int flags)
1596 {
1597         return -EINVAL;
1598 }
1599
1600 static int no_skpair(struct socket *s1, struct socket *s2)
1601 {
1602         return -EOPNOTSUPP;
1603 }
1604
1605 /**
1606  * Protocol switches for the various types of TIPC sockets
1607  */
1608
1609 static struct proto_ops msg_ops = {
1610         .owner          = THIS_MODULE,
1611         .family         = AF_TIPC,
1612         .release        = release,
1613         .bind           = bind,
1614         .connect        = connect,
1615         .socketpair     = no_skpair,
1616         .accept         = accept,
1617         .getname        = get_name,
1618         .poll           = poll,
1619         .ioctl          = ioctl,
1620         .listen         = listen,
1621         .shutdown       = shutdown,
1622         .setsockopt     = setsockopt,
1623         .getsockopt     = getsockopt,
1624         .sendmsg        = send_msg,
1625         .recvmsg        = recv_msg,
1626         .mmap           = no_mmap,
1627         .sendpage       = no_sendpage
1628 };
1629
1630 static struct proto_ops packet_ops = {
1631         .owner          = THIS_MODULE,
1632         .family         = AF_TIPC,
1633         .release        = release,
1634         .bind           = bind,
1635         .connect        = connect,
1636         .socketpair     = no_skpair,
1637         .accept         = accept,
1638         .getname        = get_name,
1639         .poll           = poll,
1640         .ioctl          = ioctl,
1641         .listen         = listen,
1642         .shutdown       = shutdown,
1643         .setsockopt     = setsockopt,
1644         .getsockopt     = getsockopt,
1645         .sendmsg        = send_packet,
1646         .recvmsg        = recv_msg,
1647         .mmap           = no_mmap,
1648         .sendpage       = no_sendpage
1649 };
1650
1651 static struct proto_ops stream_ops = {
1652         .owner          = THIS_MODULE,
1653         .family         = AF_TIPC,
1654         .release        = release,
1655         .bind           = bind,
1656         .connect        = connect,
1657         .socketpair     = no_skpair,
1658         .accept         = accept,
1659         .getname        = get_name,
1660         .poll           = poll,
1661         .ioctl          = ioctl,
1662         .listen         = listen,
1663         .shutdown       = shutdown,
1664         .setsockopt     = setsockopt,
1665         .getsockopt     = getsockopt,
1666         .sendmsg        = send_stream,
1667         .recvmsg        = recv_stream,
1668         .mmap           = no_mmap,
1669         .sendpage       = no_sendpage
1670 };
1671
1672 static struct net_proto_family tipc_family_ops = {
1673         .owner          = THIS_MODULE,
1674         .family         = AF_TIPC,
1675         .create         = tipc_create
1676 };
1677
1678 static struct proto tipc_proto = {
1679         .name           = "TIPC",
1680         .owner          = THIS_MODULE,
1681         .obj_size       = sizeof(struct tipc_sock)
1682 };
1683
1684 /**
1685  * socket_init - initialize TIPC socket interface
1686  * 
1687  * Returns 0 on success, errno otherwise
1688  */
1689 int socket_init(void)
1690 {
1691         int res;
1692
1693         res = proto_register(&tipc_proto, 1);
1694         if (res) {
1695                 err("Unable to register TIPC protocol type\n");
1696                 goto out;
1697         }
1698
1699         res = sock_register(&tipc_family_ops);
1700         if (res) {
1701                 err("Unable to register TIPC socket type\n");
1702                 proto_unregister(&tipc_proto);
1703                 goto out;
1704         }
1705
1706         sockets_enabled = 1;
1707  out:
1708         return res;
1709 }
1710
1711 /**
1712  * sock_stop - stop TIPC socket interface
1713  */
1714 void socket_stop(void)
1715 {
1716         if (!sockets_enabled)
1717                 return;
1718
1719         sockets_enabled = 0;
1720         sock_unregister(tipc_family_ops.family);
1721         proto_unregister(&tipc_proto);
1722 }
1723