2 * net/tipc/server.c: TIPC server infrastructure
4 * Copyright (c) 2012-2013, Wind River Systems
5 * Copyright (c) 2017-2018, Ericsson AB
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are met:
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.
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.
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.
45 #include <linux/module.h>
47 /* Number of messages to send before rescheduling */
48 #define MAX_SEND_MSG_COUNT 25
49 #define MAX_RECV_MSG_COUNT 25
50 #define CF_CONNECTED 1
53 #define TIPC_SERVER_NAME_LEN 32
56 * struct tipc_topsrv - TIPC server structure
57 * @conn_idr: identifier set of connection
58 * @idr_lock: protect the connection identifier set
59 * @idr_in_use: amount of allocated identifier entry
60 * @net: network namspace instance
61 * @awork: accept work item
62 * @rcv_wq: receive workqueue
63 * @send_wq: send workqueue
64 * @listener: topsrv listener socket
69 spinlock_t idr_lock
; /* for idr list */
72 struct work_struct awork
;
73 struct workqueue_struct
*rcv_wq
;
74 struct workqueue_struct
*send_wq
;
75 struct socket
*listener
;
76 char name
[TIPC_SERVER_NAME_LEN
];
80 * struct tipc_conn - TIPC connection structure
81 * @kref: reference counter to connection object
82 * @conid: connection identifier
83 * @sock: socket handler associated with connection
84 * @flags: indicates connection state
85 * @server: pointer to connected server
86 * @sub_list: lsit to all pertaing subscriptions
87 * @sub_lock: lock protecting the subscription list
88 * @rwork: receive work item
89 * @outqueue: pointer to first outbound message in queue
90 * @outqueue_lock: control access to the outqueue
91 * @swork: send work item
98 struct tipc_topsrv
*server
;
99 struct list_head sub_list
;
100 spinlock_t sub_lock
; /* for subscription list */
101 struct work_struct rwork
;
102 struct list_head outqueue
;
103 spinlock_t outqueue_lock
; /* for outqueue */
104 struct work_struct swork
;
107 /* An entry waiting to be sent */
108 struct outqueue_entry
{
110 struct tipc_event evt
;
111 struct list_head list
;
114 static void tipc_conn_recv_work(struct work_struct
*work
);
115 static void tipc_conn_send_work(struct work_struct
*work
);
116 static void tipc_topsrv_kern_evt(struct net
*net
, struct tipc_event
*evt
);
117 static void tipc_conn_delete_sub(struct tipc_conn
*con
, struct tipc_subscr
*s
);
119 static bool connected(struct tipc_conn
*con
)
121 return con
&& test_bit(CF_CONNECTED
, &con
->flags
);
124 static void tipc_conn_kref_release(struct kref
*kref
)
126 struct tipc_conn
*con
= container_of(kref
, struct tipc_conn
, kref
);
127 struct tipc_topsrv
*s
= con
->server
;
128 struct outqueue_entry
*e
, *safe
;
130 spin_lock_bh(&s
->idr_lock
);
131 idr_remove(&s
->conn_idr
, con
->conid
);
133 spin_unlock_bh(&s
->idr_lock
);
135 sock_release(con
->sock
);
137 spin_lock_bh(&con
->outqueue_lock
);
138 list_for_each_entry_safe(e
, safe
, &con
->outqueue
, list
) {
142 spin_unlock_bh(&con
->outqueue_lock
);
146 static void conn_put(struct tipc_conn
*con
)
148 kref_put(&con
->kref
, tipc_conn_kref_release
);
151 static void conn_get(struct tipc_conn
*con
)
153 kref_get(&con
->kref
);
156 static void tipc_conn_close(struct tipc_conn
*con
)
158 struct sock
*sk
= con
->sock
->sk
;
159 bool disconnect
= false;
161 write_lock_bh(&sk
->sk_callback_lock
);
162 disconnect
= test_and_clear_bit(CF_CONNECTED
, &con
->flags
);
165 sk
->sk_user_data
= NULL
;
166 tipc_conn_delete_sub(con
, NULL
);
168 write_unlock_bh(&sk
->sk_callback_lock
);
170 /* Handle concurrent calls from sending and receiving threads */
174 /* Don't flush pending works, -just let them expire */
175 kernel_sock_shutdown(con
->sock
, SHUT_RDWR
);
180 static struct tipc_conn
*tipc_conn_alloc(struct tipc_topsrv
*s
)
182 struct tipc_conn
*con
;
185 con
= kzalloc(sizeof(*con
), GFP_ATOMIC
);
187 return ERR_PTR(-ENOMEM
);
189 kref_init(&con
->kref
);
190 INIT_LIST_HEAD(&con
->outqueue
);
191 INIT_LIST_HEAD(&con
->sub_list
);
192 spin_lock_init(&con
->outqueue_lock
);
193 spin_lock_init(&con
->sub_lock
);
194 INIT_WORK(&con
->swork
, tipc_conn_send_work
);
195 INIT_WORK(&con
->rwork
, tipc_conn_recv_work
);
197 spin_lock_bh(&s
->idr_lock
);
198 ret
= idr_alloc(&s
->conn_idr
, con
, 0, 0, GFP_ATOMIC
);
201 spin_unlock_bh(&s
->idr_lock
);
202 return ERR_PTR(-ENOMEM
);
206 spin_unlock_bh(&s
->idr_lock
);
208 set_bit(CF_CONNECTED
, &con
->flags
);
214 static struct tipc_conn
*tipc_conn_lookup(struct tipc_topsrv
*s
, int conid
)
216 struct tipc_conn
*con
;
218 spin_lock_bh(&s
->idr_lock
);
219 con
= idr_find(&s
->conn_idr
, conid
);
220 if (!connected(con
) || !kref_get_unless_zero(&con
->kref
))
222 spin_unlock_bh(&s
->idr_lock
);
226 /* tipc_conn_delete_sub - delete a specific or all subscriptions
227 * for a given subscriber
229 static void tipc_conn_delete_sub(struct tipc_conn
*con
, struct tipc_subscr
*s
)
231 struct tipc_net
*tn
= tipc_net(con
->server
->net
);
232 struct list_head
*sub_list
= &con
->sub_list
;
233 struct tipc_subscription
*sub
, *tmp
;
235 spin_lock_bh(&con
->sub_lock
);
236 list_for_each_entry_safe(sub
, tmp
, sub_list
, sub_list
) {
237 if (!s
|| !memcmp(s
, &sub
->evt
.s
, sizeof(*s
))) {
238 tipc_sub_unsubscribe(sub
);
239 atomic_dec(&tn
->subscription_count
);
244 spin_unlock_bh(&con
->sub_lock
);
247 static void tipc_conn_send_to_sock(struct tipc_conn
*con
)
249 struct list_head
*queue
= &con
->outqueue
;
250 struct tipc_topsrv
*srv
= con
->server
;
251 struct outqueue_entry
*e
;
252 struct tipc_event
*evt
;
258 spin_lock_bh(&con
->outqueue_lock
);
260 while (!list_empty(queue
)) {
261 e
= list_first_entry(queue
, struct outqueue_entry
, list
);
263 spin_unlock_bh(&con
->outqueue_lock
);
266 tipc_conn_delete_sub(con
, &evt
->s
);
268 memset(&msg
, 0, sizeof(msg
));
269 msg
.msg_flags
= MSG_DONTWAIT
;
271 iov
.iov_len
= sizeof(*evt
);
275 ret
= kernel_sendmsg(con
->sock
, &msg
, &iov
,
277 if (ret
== -EWOULDBLOCK
|| ret
== 0) {
280 } else if (ret
< 0) {
281 return tipc_conn_close(con
);
284 tipc_topsrv_kern_evt(srv
->net
, evt
);
287 /* Don't starve users filling buffers */
288 if (++count
>= MAX_SEND_MSG_COUNT
) {
292 spin_lock_bh(&con
->outqueue_lock
);
296 spin_unlock_bh(&con
->outqueue_lock
);
299 static void tipc_conn_send_work(struct work_struct
*work
)
301 struct tipc_conn
*con
= container_of(work
, struct tipc_conn
, swork
);
304 tipc_conn_send_to_sock(con
);
309 /* tipc_topsrv_queue_evt() - interrupt level call from a subscription instance
310 * The queued work is launched into tipc_conn_send_work()->tipc_conn_send_to_sock()
312 void tipc_topsrv_queue_evt(struct net
*net
, int conid
,
313 u32 event
, struct tipc_event
*evt
)
315 struct tipc_topsrv
*srv
= tipc_topsrv(net
);
316 struct outqueue_entry
*e
;
317 struct tipc_conn
*con
;
319 con
= tipc_conn_lookup(srv
, conid
);
326 e
= kmalloc(sizeof(*e
), GFP_ATOMIC
);
329 e
->inactive
= (event
== TIPC_SUBSCR_TIMEOUT
);
330 memcpy(&e
->evt
, evt
, sizeof(*evt
));
331 spin_lock_bh(&con
->outqueue_lock
);
332 list_add_tail(&e
->list
, &con
->outqueue
);
333 spin_unlock_bh(&con
->outqueue_lock
);
335 if (queue_work(srv
->send_wq
, &con
->swork
))
341 /* tipc_conn_write_space - interrupt callback after a sendmsg EAGAIN
342 * Indicates that there now is more space in the send buffer
343 * The queued work is launched into tipc_send_work()->tipc_conn_send_to_sock()
345 static void tipc_conn_write_space(struct sock
*sk
)
347 struct tipc_conn
*con
;
349 read_lock_bh(&sk
->sk_callback_lock
);
350 con
= sk
->sk_user_data
;
351 if (connected(con
)) {
353 if (!queue_work(con
->server
->send_wq
, &con
->swork
))
356 read_unlock_bh(&sk
->sk_callback_lock
);
359 static int tipc_conn_rcv_sub(struct tipc_topsrv
*srv
,
360 struct tipc_conn
*con
,
361 struct tipc_subscr
*s
)
363 struct tipc_net
*tn
= tipc_net(srv
->net
);
364 struct tipc_subscription
*sub
;
365 u32 s_filter
= tipc_sub_read(s
, filter
);
367 if (s_filter
& TIPC_SUB_CANCEL
) {
368 tipc_sub_write(s
, filter
, s_filter
& ~TIPC_SUB_CANCEL
);
369 tipc_conn_delete_sub(con
, s
);
372 if (atomic_read(&tn
->subscription_count
) >= TIPC_MAX_SUBSCR
) {
373 pr_warn("Subscription rejected, max (%u)\n", TIPC_MAX_SUBSCR
);
376 sub
= tipc_sub_subscribe(srv
->net
, s
, con
->conid
);
379 atomic_inc(&tn
->subscription_count
);
380 spin_lock_bh(&con
->sub_lock
);
381 list_add(&sub
->sub_list
, &con
->sub_list
);
382 spin_unlock_bh(&con
->sub_lock
);
386 static int tipc_conn_rcv_from_sock(struct tipc_conn
*con
)
388 struct tipc_topsrv
*srv
= con
->server
;
389 struct sock
*sk
= con
->sock
->sk
;
390 struct msghdr msg
= {};
391 struct tipc_subscr s
;
396 iov
.iov_len
= sizeof(s
);
398 iov_iter_kvec(&msg
.msg_iter
, READ
, &iov
, 1, iov
.iov_len
);
399 ret
= sock_recvmsg(con
->sock
, &msg
, MSG_DONTWAIT
);
400 if (ret
== -EWOULDBLOCK
)
402 if (ret
== sizeof(s
)) {
403 read_lock_bh(&sk
->sk_callback_lock
);
404 /* RACE: the connection can be closed in the meantime */
405 if (likely(connected(con
)))
406 ret
= tipc_conn_rcv_sub(srv
, con
, &s
);
407 read_unlock_bh(&sk
->sk_callback_lock
);
412 tipc_conn_close(con
);
416 static void tipc_conn_recv_work(struct work_struct
*work
)
418 struct tipc_conn
*con
= container_of(work
, struct tipc_conn
, rwork
);
421 while (connected(con
)) {
422 if (tipc_conn_rcv_from_sock(con
))
425 /* Don't flood Rx machine */
426 if (++count
>= MAX_RECV_MSG_COUNT
) {
434 /* tipc_conn_data_ready - interrupt callback indicating the socket has data
435 * The queued work is launched into tipc_recv_work()->tipc_conn_rcv_from_sock()
437 static void tipc_conn_data_ready(struct sock
*sk
)
439 struct tipc_conn
*con
;
441 read_lock_bh(&sk
->sk_callback_lock
);
442 con
= sk
->sk_user_data
;
443 if (connected(con
)) {
445 if (!queue_work(con
->server
->rcv_wq
, &con
->rwork
))
448 read_unlock_bh(&sk
->sk_callback_lock
);
451 static void tipc_topsrv_accept(struct work_struct
*work
)
453 struct tipc_topsrv
*srv
= container_of(work
, struct tipc_topsrv
, awork
);
454 struct socket
*lsock
= srv
->listener
;
455 struct socket
*newsock
;
456 struct tipc_conn
*con
;
461 ret
= kernel_accept(lsock
, &newsock
, O_NONBLOCK
);
464 con
= tipc_conn_alloc(srv
);
467 sock_release(newsock
);
470 /* Register callbacks */
472 write_lock_bh(&newsk
->sk_callback_lock
);
473 newsk
->sk_data_ready
= tipc_conn_data_ready
;
474 newsk
->sk_write_space
= tipc_conn_write_space
;
475 newsk
->sk_user_data
= con
;
477 write_unlock_bh(&newsk
->sk_callback_lock
);
479 /* Wake up receive process in case of 'SYN+' message */
480 newsk
->sk_data_ready(newsk
);
484 /* tipc_topsrv_listener_data_ready - interrupt callback with connection request
485 * The queued job is launched into tipc_topsrv_accept()
487 static void tipc_topsrv_listener_data_ready(struct sock
*sk
)
489 struct tipc_topsrv
*srv
;
491 read_lock_bh(&sk
->sk_callback_lock
);
492 srv
= sk
->sk_user_data
;
494 queue_work(srv
->rcv_wq
, &srv
->awork
);
495 read_unlock_bh(&sk
->sk_callback_lock
);
498 static int tipc_topsrv_create_listener(struct tipc_topsrv
*srv
)
500 int imp
= TIPC_CRITICAL_IMPORTANCE
;
501 struct socket
*lsock
= NULL
;
502 struct sockaddr_tipc saddr
;
506 rc
= sock_create_kern(srv
->net
, AF_TIPC
, SOCK_SEQPACKET
, 0, &lsock
);
510 srv
->listener
= lsock
;
512 write_lock_bh(&sk
->sk_callback_lock
);
513 sk
->sk_data_ready
= tipc_topsrv_listener_data_ready
;
514 sk
->sk_user_data
= srv
;
515 write_unlock_bh(&sk
->sk_callback_lock
);
517 rc
= kernel_setsockopt(lsock
, SOL_TIPC
, TIPC_IMPORTANCE
,
518 (char *)&imp
, sizeof(imp
));
522 saddr
.family
= AF_TIPC
;
523 saddr
.addrtype
= TIPC_ADDR_NAMESEQ
;
524 saddr
.addr
.nameseq
.type
= TIPC_TOP_SRV
;
525 saddr
.addr
.nameseq
.lower
= TIPC_TOP_SRV
;
526 saddr
.addr
.nameseq
.upper
= TIPC_TOP_SRV
;
527 saddr
.scope
= TIPC_NODE_SCOPE
;
529 rc
= kernel_bind(lsock
, (struct sockaddr
*)&saddr
, sizeof(saddr
));
532 rc
= kernel_listen(lsock
, 0);
536 /* As server's listening socket owner and creator is the same module,
537 * we have to decrease TIPC module reference count to guarantee that
538 * it remains zero after the server socket is created, otherwise,
539 * executing "rmmod" command is unable to make TIPC module deleted
540 * after TIPC module is inserted successfully.
542 * However, the reference count is ever increased twice in
543 * sock_create_kern(): one is to increase the reference count of owner
544 * of TIPC socket's proto_ops struct; another is to increment the
545 * reference count of owner of TIPC proto struct. Therefore, we must
546 * decrement the module reference count twice to ensure that it keeps
547 * zero after server's listening socket is created. Of course, we
548 * must bump the module reference count twice as well before the socket
551 module_put(lsock
->ops
->owner
);
552 module_put(sk
->sk_prot_creator
->owner
);
560 bool tipc_topsrv_kern_subscr(struct net
*net
, u32 port
, u32 type
, u32 lower
,
561 u32 upper
, u32 filter
, int *conid
)
563 struct tipc_subscr sub
;
564 struct tipc_conn
*con
;
568 sub
.seq
.lower
= lower
;
569 sub
.seq
.upper
= upper
;
570 sub
.timeout
= TIPC_WAIT_FOREVER
;
572 *(u32
*)&sub
.usr_handle
= port
;
574 con
= tipc_conn_alloc(tipc_topsrv(net
));
580 rc
= tipc_conn_rcv_sub(tipc_topsrv(net
), con
, &sub
);
587 void tipc_topsrv_kern_unsubscr(struct net
*net
, int conid
)
589 struct tipc_conn
*con
;
591 con
= tipc_conn_lookup(tipc_topsrv(net
), conid
);
595 test_and_clear_bit(CF_CONNECTED
, &con
->flags
);
596 tipc_conn_delete_sub(con
, NULL
);
601 static void tipc_topsrv_kern_evt(struct net
*net
, struct tipc_event
*evt
)
603 u32 port
= *(u32
*)&evt
->s
.usr_handle
;
604 u32 self
= tipc_own_addr(net
);
605 struct sk_buff_head evtq
;
608 skb
= tipc_msg_create(TOP_SRV
, 0, INT_H_SIZE
, sizeof(*evt
),
609 self
, self
, port
, port
, 0);
612 msg_set_dest_droppable(buf_msg(skb
), true);
613 memcpy(msg_data(buf_msg(skb
)), evt
, sizeof(*evt
));
614 skb_queue_head_init(&evtq
);
615 __skb_queue_tail(&evtq
, skb
);
616 tipc_loopback_trace(net
, &evtq
);
617 tipc_sk_rcv(net
, &evtq
);
620 static int tipc_topsrv_work_start(struct tipc_topsrv
*s
)
622 s
->rcv_wq
= alloc_ordered_workqueue("tipc_rcv", 0);
624 pr_err("can't start tipc receive workqueue\n");
628 s
->send_wq
= alloc_ordered_workqueue("tipc_send", 0);
630 pr_err("can't start tipc send workqueue\n");
631 destroy_workqueue(s
->rcv_wq
);
638 static void tipc_topsrv_work_stop(struct tipc_topsrv
*s
)
640 destroy_workqueue(s
->rcv_wq
);
641 destroy_workqueue(s
->send_wq
);
644 static int tipc_topsrv_start(struct net
*net
)
646 struct tipc_net
*tn
= tipc_net(net
);
647 const char name
[] = "topology_server";
648 struct tipc_topsrv
*srv
;
651 srv
= kzalloc(sizeof(*srv
), GFP_ATOMIC
);
656 INIT_WORK(&srv
->awork
, tipc_topsrv_accept
);
658 strscpy(srv
->name
, name
, sizeof(srv
->name
));
660 atomic_set(&tn
->subscription_count
, 0);
662 spin_lock_init(&srv
->idr_lock
);
663 idr_init(&srv
->conn_idr
);
666 ret
= tipc_topsrv_work_start(srv
);
670 ret
= tipc_topsrv_create_listener(srv
);
672 tipc_topsrv_work_stop(srv
);
677 static void tipc_topsrv_stop(struct net
*net
)
679 struct tipc_topsrv
*srv
= tipc_topsrv(net
);
680 struct socket
*lsock
= srv
->listener
;
681 struct tipc_conn
*con
;
684 spin_lock_bh(&srv
->idr_lock
);
685 for (id
= 0; srv
->idr_in_use
; id
++) {
686 con
= idr_find(&srv
->conn_idr
, id
);
688 spin_unlock_bh(&srv
->idr_lock
);
689 tipc_conn_close(con
);
690 spin_lock_bh(&srv
->idr_lock
);
693 __module_get(lsock
->ops
->owner
);
694 __module_get(lsock
->sk
->sk_prot_creator
->owner
);
695 srv
->listener
= NULL
;
696 spin_unlock_bh(&srv
->idr_lock
);
698 tipc_topsrv_work_stop(srv
);
699 idr_destroy(&srv
->conn_idr
);
703 int __net_init
tipc_topsrv_init_net(struct net
*net
)
705 return tipc_topsrv_start(net
);
708 void __net_exit
tipc_topsrv_exit_net(struct net
*net
)
710 tipc_topsrv_stop(net
);