]>
git.proxmox.com Git - mirror_ubuntu-disco-kernel.git/blob - net/tipc/server.c
2 * net/tipc/server.c: TIPC server infrastructure
4 * Copyright (c) 2012-2013, Wind River Systems
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions are met:
10 * 1. Redistributions of source code must retain the above copyright
11 * notice, this list of conditions and the following disclaimer.
12 * 2. Redistributions in binary form must reproduce the above copyright
13 * notice, this list of conditions and the following disclaimer in the
14 * documentation and/or other materials provided with the distribution.
15 * 3. Neither the names of the copyright holders nor the names of its
16 * contributors may be used to endorse or promote products derived from
17 * this software without specific prior written permission.
19 * Alternatively, this software may be distributed under the terms of the
20 * GNU General Public License ("GPL") version 2 as published by the Free
21 * Software Foundation.
23 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
24 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
25 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
26 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
27 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
28 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
29 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
30 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
31 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
32 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
33 * POSSIBILITY OF SUCH DAMAGE.
42 #include <linux/module.h>
44 /* Number of messages to send before rescheduling */
45 #define MAX_SEND_MSG_COUNT 25
46 #define MAX_RECV_MSG_COUNT 25
47 #define CF_CONNECTED 1
50 #define sock2con(x) ((struct tipc_conn *)(x)->sk_user_data)
53 * struct tipc_conn - TIPC connection structure
54 * @kref: reference counter to connection object
55 * @conid: connection identifier
56 * @sock: socket handler associated with connection
57 * @flags: indicates connection state
58 * @server: pointer to connected server
59 * @rwork: receive work item
60 * @usr_data: user-specified field
61 * @rx_action: what to do when connection socket is active
62 * @outqueue: pointer to first outbound message in queue
63 * @outqueue_lock: control access to the outqueue
64 * @outqueue: list of connection objects for its server
65 * @swork: send work item
72 struct tipc_server
*server
;
73 struct work_struct rwork
;
74 int (*rx_action
) (struct tipc_conn
*con
);
76 struct list_head outqueue
;
77 spinlock_t outqueue_lock
;
78 struct work_struct swork
;
81 /* An entry waiting to be sent */
82 struct outqueue_entry
{
83 struct list_head list
;
85 struct sockaddr_tipc dest
;
88 static void tipc_recv_work(struct work_struct
*work
);
89 static void tipc_send_work(struct work_struct
*work
);
90 static void tipc_clean_outqueues(struct tipc_conn
*con
);
92 static void tipc_conn_kref_release(struct kref
*kref
)
94 struct tipc_conn
*con
= container_of(kref
, struct tipc_conn
, kref
);
95 struct tipc_server
*s
= con
->server
;
96 struct sockaddr_tipc
*saddr
= s
->saddr
;
97 struct socket
*sock
= con
->sock
;
102 if (test_bit(CF_SERVER
, &con
->flags
)) {
103 __module_get(sock
->ops
->owner
);
104 __module_get(sk
->sk_prot_creator
->owner
);
106 saddr
->scope
= -TIPC_NODE_SCOPE
;
107 kernel_bind(sock
, (struct sockaddr
*)saddr
, sizeof(*saddr
));
111 spin_lock_bh(&s
->idr_lock
);
112 idr_remove(&s
->conn_idr
, con
->conid
);
114 spin_unlock_bh(&s
->idr_lock
);
115 tipc_clean_outqueues(con
);
119 static void conn_put(struct tipc_conn
*con
)
121 kref_put(&con
->kref
, tipc_conn_kref_release
);
124 static void conn_get(struct tipc_conn
*con
)
126 kref_get(&con
->kref
);
129 static struct tipc_conn
*tipc_conn_lookup(struct tipc_server
*s
, int conid
)
131 struct tipc_conn
*con
;
133 spin_lock_bh(&s
->idr_lock
);
134 con
= idr_find(&s
->conn_idr
, conid
);
135 if (con
&& test_bit(CF_CONNECTED
, &con
->flags
))
139 spin_unlock_bh(&s
->idr_lock
);
143 static void sock_data_ready(struct sock
*sk
)
145 struct tipc_conn
*con
;
147 read_lock_bh(&sk
->sk_callback_lock
);
149 if (con
&& test_bit(CF_CONNECTED
, &con
->flags
)) {
151 if (!queue_work(con
->server
->rcv_wq
, &con
->rwork
))
154 read_unlock_bh(&sk
->sk_callback_lock
);
157 static void sock_write_space(struct sock
*sk
)
159 struct tipc_conn
*con
;
161 read_lock_bh(&sk
->sk_callback_lock
);
163 if (con
&& test_bit(CF_CONNECTED
, &con
->flags
)) {
165 if (!queue_work(con
->server
->send_wq
, &con
->swork
))
168 read_unlock_bh(&sk
->sk_callback_lock
);
171 static void tipc_register_callbacks(struct socket
*sock
, struct tipc_conn
*con
)
173 struct sock
*sk
= sock
->sk
;
175 write_lock_bh(&sk
->sk_callback_lock
);
177 sk
->sk_data_ready
= sock_data_ready
;
178 sk
->sk_write_space
= sock_write_space
;
179 sk
->sk_user_data
= con
;
183 write_unlock_bh(&sk
->sk_callback_lock
);
186 static void tipc_unregister_callbacks(struct tipc_conn
*con
)
188 struct sock
*sk
= con
->sock
->sk
;
190 write_lock_bh(&sk
->sk_callback_lock
);
191 sk
->sk_user_data
= NULL
;
192 write_unlock_bh(&sk
->sk_callback_lock
);
195 static void tipc_close_conn(struct tipc_conn
*con
)
197 struct tipc_server
*s
= con
->server
;
199 if (test_and_clear_bit(CF_CONNECTED
, &con
->flags
)) {
201 tipc_unregister_callbacks(con
);
204 s
->tipc_conn_release(con
->conid
, con
->usr_data
);
206 /* We shouldn't flush pending works as we may be in the
207 * thread. In fact the races with pending rx/tx work structs
208 * are harmless for us here as we have already deleted this
209 * connection from server connection list.
212 kernel_sock_shutdown(con
->sock
, SHUT_RDWR
);
217 static struct tipc_conn
*tipc_alloc_conn(struct tipc_server
*s
)
219 struct tipc_conn
*con
;
222 con
= kzalloc(sizeof(struct tipc_conn
), GFP_ATOMIC
);
224 return ERR_PTR(-ENOMEM
);
226 kref_init(&con
->kref
);
227 INIT_LIST_HEAD(&con
->outqueue
);
228 spin_lock_init(&con
->outqueue_lock
);
229 INIT_WORK(&con
->swork
, tipc_send_work
);
230 INIT_WORK(&con
->rwork
, tipc_recv_work
);
232 spin_lock_bh(&s
->idr_lock
);
233 ret
= idr_alloc(&s
->conn_idr
, con
, 0, 0, GFP_ATOMIC
);
236 spin_unlock_bh(&s
->idr_lock
);
237 return ERR_PTR(-ENOMEM
);
241 spin_unlock_bh(&s
->idr_lock
);
243 set_bit(CF_CONNECTED
, &con
->flags
);
249 static int tipc_receive_from_sock(struct tipc_conn
*con
)
251 struct msghdr msg
= {};
252 struct tipc_server
*s
= con
->server
;
253 struct sockaddr_tipc addr
;
258 buf
= kmem_cache_alloc(s
->rcvbuf_cache
, GFP_ATOMIC
);
265 iov
.iov_len
= s
->max_rcvbuf_size
;
266 msg
.msg_name
= &addr
;
267 ret
= kernel_recvmsg(con
->sock
, &msg
, &iov
, 1, iov
.iov_len
,
270 kmem_cache_free(s
->rcvbuf_cache
, buf
);
274 s
->tipc_conn_recvmsg(sock_net(con
->sock
->sk
), con
->conid
, &addr
,
275 con
->usr_data
, buf
, ret
);
277 kmem_cache_free(s
->rcvbuf_cache
, buf
);
282 if (ret
!= -EWOULDBLOCK
)
283 tipc_close_conn(con
);
285 /* Don't return success if we really got EOF */
291 static int tipc_accept_from_sock(struct tipc_conn
*con
)
293 struct tipc_server
*s
= con
->server
;
294 struct socket
*sock
= con
->sock
;
295 struct socket
*newsock
;
296 struct tipc_conn
*newcon
;
299 ret
= kernel_accept(sock
, &newsock
, O_NONBLOCK
);
303 newcon
= tipc_alloc_conn(con
->server
);
304 if (IS_ERR(newcon
)) {
305 ret
= PTR_ERR(newcon
);
306 sock_release(newsock
);
310 newcon
->rx_action
= tipc_receive_from_sock
;
311 tipc_register_callbacks(newsock
, newcon
);
313 /* Notify that new connection is incoming */
314 newcon
->usr_data
= s
->tipc_conn_new(newcon
->conid
);
315 if (!newcon
->usr_data
) {
316 sock_release(newsock
);
321 /* Wake up receive process in case of 'SYN+' message */
322 newsock
->sk
->sk_data_ready(newsock
->sk
);
326 static struct socket
*tipc_create_listen_sock(struct tipc_conn
*con
)
328 struct tipc_server
*s
= con
->server
;
329 struct socket
*sock
= NULL
;
332 ret
= sock_create_kern(s
->net
, AF_TIPC
, SOCK_SEQPACKET
, 0, &sock
);
335 ret
= kernel_setsockopt(sock
, SOL_TIPC
, TIPC_IMPORTANCE
,
336 (char *)&s
->imp
, sizeof(s
->imp
));
339 ret
= kernel_bind(sock
, (struct sockaddr
*)s
->saddr
, sizeof(*s
->saddr
));
346 con
->rx_action
= tipc_accept_from_sock
;
348 ret
= kernel_listen(sock
, 0);
354 con
->rx_action
= tipc_receive_from_sock
;
357 pr_err("Unknown socket type %d\n", s
->type
);
361 /* As server's listening socket owner and creator is the same module,
362 * we have to decrease TIPC module reference count to guarantee that
363 * it remains zero after the server socket is created, otherwise,
364 * executing "rmmod" command is unable to make TIPC module deleted
365 * after TIPC module is inserted successfully.
367 * However, the reference count is ever increased twice in
368 * sock_create_kern(): one is to increase the reference count of owner
369 * of TIPC socket's proto_ops struct; another is to increment the
370 * reference count of owner of TIPC proto struct. Therefore, we must
371 * decrement the module reference count twice to ensure that it keeps
372 * zero after server's listening socket is created. Of course, we
373 * must bump the module reference count twice as well before the socket
376 module_put(sock
->ops
->owner
);
377 module_put(sock
->sk
->sk_prot_creator
->owner
);
378 set_bit(CF_SERVER
, &con
->flags
);
383 kernel_sock_shutdown(sock
, SHUT_RDWR
);
388 static int tipc_open_listening_sock(struct tipc_server
*s
)
391 struct tipc_conn
*con
;
393 con
= tipc_alloc_conn(s
);
397 sock
= tipc_create_listen_sock(con
);
399 idr_remove(&s
->conn_idr
, con
->conid
);
405 tipc_register_callbacks(sock
, con
);
409 static struct outqueue_entry
*tipc_alloc_entry(void *data
, int len
)
411 struct outqueue_entry
*entry
;
414 entry
= kmalloc(sizeof(struct outqueue_entry
), GFP_ATOMIC
);
418 buf
= kmemdup(data
, len
, GFP_ATOMIC
);
424 entry
->iov
.iov_base
= buf
;
425 entry
->iov
.iov_len
= len
;
430 static void tipc_free_entry(struct outqueue_entry
*e
)
432 kfree(e
->iov
.iov_base
);
436 static void tipc_clean_outqueues(struct tipc_conn
*con
)
438 struct outqueue_entry
*e
, *safe
;
440 spin_lock_bh(&con
->outqueue_lock
);
441 list_for_each_entry_safe(e
, safe
, &con
->outqueue
, list
) {
445 spin_unlock_bh(&con
->outqueue_lock
);
448 int tipc_conn_sendmsg(struct tipc_server
*s
, int conid
,
449 struct sockaddr_tipc
*addr
, void *data
, size_t len
)
451 struct outqueue_entry
*e
;
452 struct tipc_conn
*con
;
454 con
= tipc_conn_lookup(s
, conid
);
458 if (!test_bit(CF_CONNECTED
, &con
->flags
)) {
463 e
= tipc_alloc_entry(data
, len
);
470 memcpy(&e
->dest
, addr
, sizeof(struct sockaddr_tipc
));
472 spin_lock_bh(&con
->outqueue_lock
);
473 list_add_tail(&e
->list
, &con
->outqueue
);
474 spin_unlock_bh(&con
->outqueue_lock
);
476 if (!queue_work(s
->send_wq
, &con
->swork
))
481 void tipc_conn_terminate(struct tipc_server
*s
, int conid
)
483 struct tipc_conn
*con
;
485 con
= tipc_conn_lookup(s
, conid
);
487 tipc_close_conn(con
);
492 bool tipc_topsrv_kern_subscr(struct net
*net
, u32 port
, u32 type
,
493 u32 lower
, u32 upper
, int *conid
)
495 struct tipc_subscriber
*scbr
;
496 struct tipc_subscr sub
;
497 struct tipc_server
*s
;
498 struct tipc_conn
*con
;
501 sub
.seq
.lower
= lower
;
502 sub
.seq
.upper
= upper
;
503 sub
.timeout
= TIPC_WAIT_FOREVER
;
504 sub
.filter
= TIPC_SUB_PORTS
;
505 *(u32
*)&sub
.usr_handle
= port
;
507 con
= tipc_alloc_conn(tipc_topsrv(net
));
513 scbr
= s
->tipc_conn_new(*conid
);
519 con
->usr_data
= scbr
;
521 s
->tipc_conn_recvmsg(net
, *conid
, NULL
, scbr
, &sub
, sizeof(sub
));
525 void tipc_topsrv_kern_unsubscr(struct net
*net
, int conid
)
527 struct tipc_conn
*con
;
529 con
= tipc_conn_lookup(tipc_topsrv(net
), conid
);
532 tipc_close_conn(con
);
536 static void tipc_send_kern_top_evt(struct net
*net
, struct tipc_event
*evt
)
538 u32 port
= *(u32
*)&evt
->s
.usr_handle
;
539 u32 self
= tipc_own_addr(net
);
540 struct sk_buff_head evtq
;
543 skb
= tipc_msg_create(TOP_SRV
, 0, INT_H_SIZE
, sizeof(*evt
),
544 self
, self
, port
, port
, 0);
547 msg_set_dest_droppable(buf_msg(skb
), true);
548 memcpy(msg_data(buf_msg(skb
)), evt
, sizeof(*evt
));
549 skb_queue_head_init(&evtq
);
550 __skb_queue_tail(&evtq
, skb
);
551 tipc_sk_rcv(net
, &evtq
);
554 static void tipc_send_to_sock(struct tipc_conn
*con
)
556 struct tipc_server
*s
= con
->server
;
557 struct outqueue_entry
*e
;
558 struct tipc_event
*evt
;
563 spin_lock_bh(&con
->outqueue_lock
);
564 while (test_bit(CF_CONNECTED
, &con
->flags
)) {
565 e
= list_entry(con
->outqueue
.next
, struct outqueue_entry
, list
);
566 if ((struct list_head
*) e
== &con
->outqueue
)
569 spin_unlock_bh(&con
->outqueue_lock
);
572 memset(&msg
, 0, sizeof(msg
));
573 msg
.msg_flags
= MSG_DONTWAIT
;
574 if (s
->type
== SOCK_DGRAM
|| s
->type
== SOCK_RDM
) {
575 msg
.msg_name
= &e
->dest
;
576 msg
.msg_namelen
= sizeof(struct sockaddr_tipc
);
578 ret
= kernel_sendmsg(con
->sock
, &msg
, &e
->iov
, 1,
580 if (ret
== -EWOULDBLOCK
|| ret
== 0) {
583 } else if (ret
< 0) {
587 evt
= e
->iov
.iov_base
;
588 tipc_send_kern_top_evt(s
->net
, evt
);
590 /* Don't starve users filling buffers */
591 if (++count
>= MAX_SEND_MSG_COUNT
) {
596 spin_lock_bh(&con
->outqueue_lock
);
600 spin_unlock_bh(&con
->outqueue_lock
);
605 tipc_close_conn(con
);
608 static void tipc_recv_work(struct work_struct
*work
)
610 struct tipc_conn
*con
= container_of(work
, struct tipc_conn
, rwork
);
613 while (test_bit(CF_CONNECTED
, &con
->flags
)) {
614 if (con
->rx_action(con
))
617 /* Don't flood Rx machine */
618 if (++count
>= MAX_RECV_MSG_COUNT
) {
626 static void tipc_send_work(struct work_struct
*work
)
628 struct tipc_conn
*con
= container_of(work
, struct tipc_conn
, swork
);
630 if (test_bit(CF_CONNECTED
, &con
->flags
))
631 tipc_send_to_sock(con
);
636 static void tipc_work_stop(struct tipc_server
*s
)
638 destroy_workqueue(s
->rcv_wq
);
639 destroy_workqueue(s
->send_wq
);
642 static int tipc_work_start(struct tipc_server
*s
)
644 s
->rcv_wq
= alloc_ordered_workqueue("tipc_rcv", 0);
646 pr_err("can't start tipc receive workqueue\n");
650 s
->send_wq
= alloc_ordered_workqueue("tipc_send", 0);
652 pr_err("can't start tipc send workqueue\n");
653 destroy_workqueue(s
->rcv_wq
);
660 int tipc_server_start(struct tipc_server
*s
)
664 spin_lock_init(&s
->idr_lock
);
665 idr_init(&s
->conn_idr
);
668 s
->rcvbuf_cache
= kmem_cache_create(s
->name
, s
->max_rcvbuf_size
,
669 0, SLAB_HWCACHE_ALIGN
, NULL
);
670 if (!s
->rcvbuf_cache
)
673 ret
= tipc_work_start(s
);
675 kmem_cache_destroy(s
->rcvbuf_cache
);
678 ret
= tipc_open_listening_sock(s
);
681 kmem_cache_destroy(s
->rcvbuf_cache
);
687 void tipc_server_stop(struct tipc_server
*s
)
689 struct tipc_conn
*con
;
692 spin_lock_bh(&s
->idr_lock
);
693 for (id
= 0; s
->idr_in_use
; id
++) {
694 con
= idr_find(&s
->conn_idr
, id
);
696 spin_unlock_bh(&s
->idr_lock
);
697 tipc_close_conn(con
);
698 spin_lock_bh(&s
->idr_lock
);
701 spin_unlock_bh(&s
->idr_lock
);
704 kmem_cache_destroy(s
->rcvbuf_cache
);
705 idr_destroy(&s
->conn_idr
);