tipc: use destination length for copy string
[muen/linux.git] / net / tipc / topsrv.c
1 /*
2  * net/tipc/server.c: TIPC server infrastructure
3  *
4  * Copyright (c) 2012-2013, Wind River Systems
5  * Copyright (c) 2017-2018, Ericsson AB
6  * All rights reserved.
7  *
8  * Redistribution and use in source and binary forms, with or without
9  * modification, are permitted provided that the following conditions are met:
10  *
11  * 1. Redistributions of source code must retain the above copyright
12  *    notice, this list of conditions and the following disclaimer.
13  * 2. Redistributions in binary form must reproduce the above copyright
14  *    notice, this list of conditions and the following disclaimer in the
15  *    documentation and/or other materials provided with the distribution.
16  * 3. Neither the names of the copyright holders nor the names of its
17  *    contributors may be used to endorse or promote products derived from
18  *    this software without specific prior written permission.
19  *
20  * Alternatively, this software may be distributed under the terms of the
21  * GNU General Public License ("GPL") version 2 as published by the Free
22  * Software Foundation.
23  *
24  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
25  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
26  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
27  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
28  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
29  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
30  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
31  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
32  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
33  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
34  * POSSIBILITY OF SUCH DAMAGE.
35  */
36
37 #include "subscr.h"
38 #include "topsrv.h"
39 #include "core.h"
40 #include "socket.h"
41 #include "addr.h"
42 #include "msg.h"
43 #include <net/sock.h>
44 #include <linux/module.h>
45
46 /* Number of messages to send before rescheduling */
47 #define MAX_SEND_MSG_COUNT      25
48 #define MAX_RECV_MSG_COUNT      25
49 #define CF_CONNECTED            1
50 #define CF_SERVER               2
51
52 #define TIPC_SERVER_NAME_LEN    32
53
54 /**
55  * struct tipc_topsrv - TIPC server structure
56  * @conn_idr: identifier set of connection
57  * @idr_lock: protect the connection identifier set
58  * @idr_in_use: amount of allocated identifier entry
59  * @net: network namspace instance
60  * @awork: accept work item
61  * @rcv_wq: receive workqueue
62  * @send_wq: send workqueue
63  * @max_rcvbuf_size: maximum permitted receive message length
64  * @listener: topsrv listener socket
65  * @name: server name
66  */
67 struct tipc_topsrv {
68         struct idr conn_idr;
69         spinlock_t idr_lock; /* for idr list */
70         int idr_in_use;
71         struct net *net;
72         struct work_struct awork;
73         struct workqueue_struct *rcv_wq;
74         struct workqueue_struct *send_wq;
75         int max_rcvbuf_size;
76         struct socket *listener;
77         char name[TIPC_SERVER_NAME_LEN];
78 };
79
80 /**
81  * struct tipc_conn - TIPC connection structure
82  * @kref: reference counter to connection object
83  * @conid: connection identifier
84  * @sock: socket handler associated with connection
85  * @flags: indicates connection state
86  * @server: pointer to connected server
87  * @sub_list: lsit to all pertaing subscriptions
88  * @sub_lock: lock protecting the subscription list
89  * @rwork: receive work item
90  * @outqueue: pointer to first outbound message in queue
91  * @outqueue_lock: control access to the outqueue
92  * @swork: send work item
93  */
94 struct tipc_conn {
95         struct kref kref;
96         int conid;
97         struct socket *sock;
98         unsigned long flags;
99         struct tipc_topsrv *server;
100         struct list_head sub_list;
101         spinlock_t sub_lock; /* for subscription list */
102         struct work_struct rwork;
103         struct list_head outqueue;
104         spinlock_t outqueue_lock; /* for outqueue */
105         struct work_struct swork;
106 };
107
108 /* An entry waiting to be sent */
109 struct outqueue_entry {
110         bool inactive;
111         struct tipc_event evt;
112         struct list_head list;
113 };
114
115 static void tipc_conn_recv_work(struct work_struct *work);
116 static void tipc_conn_send_work(struct work_struct *work);
117 static void tipc_topsrv_kern_evt(struct net *net, struct tipc_event *evt);
118 static void tipc_conn_delete_sub(struct tipc_conn *con, struct tipc_subscr *s);
119
120 static bool connected(struct tipc_conn *con)
121 {
122         return con && test_bit(CF_CONNECTED, &con->flags);
123 }
124
125 static void tipc_conn_kref_release(struct kref *kref)
126 {
127         struct tipc_conn *con = container_of(kref, struct tipc_conn, kref);
128         struct tipc_topsrv *s = con->server;
129         struct outqueue_entry *e, *safe;
130
131         spin_lock_bh(&s->idr_lock);
132         idr_remove(&s->conn_idr, con->conid);
133         s->idr_in_use--;
134         spin_unlock_bh(&s->idr_lock);
135         if (con->sock)
136                 sock_release(con->sock);
137
138         spin_lock_bh(&con->outqueue_lock);
139         list_for_each_entry_safe(e, safe, &con->outqueue, list) {
140                 list_del(&e->list);
141                 kfree(e);
142         }
143         spin_unlock_bh(&con->outqueue_lock);
144         kfree(con);
145 }
146
147 static void conn_put(struct tipc_conn *con)
148 {
149         kref_put(&con->kref, tipc_conn_kref_release);
150 }
151
152 static void conn_get(struct tipc_conn *con)
153 {
154         kref_get(&con->kref);
155 }
156
157 static void tipc_conn_close(struct tipc_conn *con)
158 {
159         struct sock *sk = con->sock->sk;
160         bool disconnect = false;
161
162         write_lock_bh(&sk->sk_callback_lock);
163         disconnect = test_and_clear_bit(CF_CONNECTED, &con->flags);
164
165         if (disconnect) {
166                 sk->sk_user_data = NULL;
167                 tipc_conn_delete_sub(con, NULL);
168         }
169         write_unlock_bh(&sk->sk_callback_lock);
170
171         /* Handle concurrent calls from sending and receiving threads */
172         if (!disconnect)
173                 return;
174
175         /* Don't flush pending works, -just let them expire */
176         kernel_sock_shutdown(con->sock, SHUT_RDWR);
177
178         conn_put(con);
179 }
180
181 static struct tipc_conn *tipc_conn_alloc(struct tipc_topsrv *s)
182 {
183         struct tipc_conn *con;
184         int ret;
185
186         con = kzalloc(sizeof(*con), GFP_ATOMIC);
187         if (!con)
188                 return ERR_PTR(-ENOMEM);
189
190         kref_init(&con->kref);
191         INIT_LIST_HEAD(&con->outqueue);
192         INIT_LIST_HEAD(&con->sub_list);
193         spin_lock_init(&con->outqueue_lock);
194         spin_lock_init(&con->sub_lock);
195         INIT_WORK(&con->swork, tipc_conn_send_work);
196         INIT_WORK(&con->rwork, tipc_conn_recv_work);
197
198         spin_lock_bh(&s->idr_lock);
199         ret = idr_alloc(&s->conn_idr, con, 0, 0, GFP_ATOMIC);
200         if (ret < 0) {
201                 kfree(con);
202                 spin_unlock_bh(&s->idr_lock);
203                 return ERR_PTR(-ENOMEM);
204         }
205         con->conid = ret;
206         s->idr_in_use++;
207         spin_unlock_bh(&s->idr_lock);
208
209         set_bit(CF_CONNECTED, &con->flags);
210         con->server = s;
211
212         return con;
213 }
214
215 static struct tipc_conn *tipc_conn_lookup(struct tipc_topsrv *s, int conid)
216 {
217         struct tipc_conn *con;
218
219         spin_lock_bh(&s->idr_lock);
220         con = idr_find(&s->conn_idr, conid);
221         if (!connected(con) || !kref_get_unless_zero(&con->kref))
222                 con = NULL;
223         spin_unlock_bh(&s->idr_lock);
224         return con;
225 }
226
227 /* tipc_conn_delete_sub - delete a specific or all subscriptions
228  * for a given subscriber
229  */
230 static void tipc_conn_delete_sub(struct tipc_conn *con, struct tipc_subscr *s)
231 {
232         struct tipc_net *tn = tipc_net(con->server->net);
233         struct list_head *sub_list = &con->sub_list;
234         struct tipc_subscription *sub, *tmp;
235
236         spin_lock_bh(&con->sub_lock);
237         list_for_each_entry_safe(sub, tmp, sub_list, sub_list) {
238                 if (!s || !memcmp(s, &sub->evt.s, sizeof(*s))) {
239                         tipc_sub_unsubscribe(sub);
240                         atomic_dec(&tn->subscription_count);
241                 } else if (s) {
242                         break;
243                 }
244         }
245         spin_unlock_bh(&con->sub_lock);
246 }
247
248 static void tipc_conn_send_to_sock(struct tipc_conn *con)
249 {
250         struct list_head *queue = &con->outqueue;
251         struct tipc_topsrv *srv = con->server;
252         struct outqueue_entry *e;
253         struct tipc_event *evt;
254         struct msghdr msg;
255         struct kvec iov;
256         int count = 0;
257         int ret;
258
259         spin_lock_bh(&con->outqueue_lock);
260
261         while (!list_empty(queue)) {
262                 e = list_first_entry(queue, struct outqueue_entry, list);
263                 evt = &e->evt;
264                 spin_unlock_bh(&con->outqueue_lock);
265
266                 if (e->inactive)
267                         tipc_conn_delete_sub(con, &evt->s);
268
269                 memset(&msg, 0, sizeof(msg));
270                 msg.msg_flags = MSG_DONTWAIT;
271                 iov.iov_base = evt;
272                 iov.iov_len = sizeof(*evt);
273                 msg.msg_name = NULL;
274
275                 if (con->sock) {
276                         ret = kernel_sendmsg(con->sock, &msg, &iov,
277                                              1, sizeof(*evt));
278                         if (ret == -EWOULDBLOCK || ret == 0) {
279                                 cond_resched();
280                                 return;
281                         } else if (ret < 0) {
282                                 return tipc_conn_close(con);
283                         }
284                 } else {
285                         tipc_topsrv_kern_evt(srv->net, evt);
286                 }
287
288                 /* Don't starve users filling buffers */
289                 if (++count >= MAX_SEND_MSG_COUNT) {
290                         cond_resched();
291                         count = 0;
292                 }
293                 spin_lock_bh(&con->outqueue_lock);
294                 list_del(&e->list);
295                 kfree(e);
296         }
297         spin_unlock_bh(&con->outqueue_lock);
298 }
299
300 static void tipc_conn_send_work(struct work_struct *work)
301 {
302         struct tipc_conn *con = container_of(work, struct tipc_conn, swork);
303
304         if (connected(con))
305                 tipc_conn_send_to_sock(con);
306
307         conn_put(con);
308 }
309
310 /* tipc_topsrv_queue_evt() - interrupt level call from a subscription instance
311  * The queued work is launched into tipc_conn_send_work()->tipc_conn_send_to_sock()
312  */
313 void tipc_topsrv_queue_evt(struct net *net, int conid,
314                            u32 event, struct tipc_event *evt)
315 {
316         struct tipc_topsrv *srv = tipc_topsrv(net);
317         struct outqueue_entry *e;
318         struct tipc_conn *con;
319
320         con = tipc_conn_lookup(srv, conid);
321         if (!con)
322                 return;
323
324         if (!connected(con))
325                 goto err;
326
327         e = kmalloc(sizeof(*e), GFP_ATOMIC);
328         if (!e)
329                 goto err;
330         e->inactive = (event == TIPC_SUBSCR_TIMEOUT);
331         memcpy(&e->evt, evt, sizeof(*evt));
332         spin_lock_bh(&con->outqueue_lock);
333         list_add_tail(&e->list, &con->outqueue);
334         spin_unlock_bh(&con->outqueue_lock);
335
336         if (queue_work(srv->send_wq, &con->swork))
337                 return;
338 err:
339         conn_put(con);
340 }
341
342 /* tipc_conn_write_space - interrupt callback after a sendmsg EAGAIN
343  * Indicates that there now is more space in the send buffer
344  * The queued work is launched into tipc_send_work()->tipc_conn_send_to_sock()
345  */
346 static void tipc_conn_write_space(struct sock *sk)
347 {
348         struct tipc_conn *con;
349
350         read_lock_bh(&sk->sk_callback_lock);
351         con = sk->sk_user_data;
352         if (connected(con)) {
353                 conn_get(con);
354                 if (!queue_work(con->server->send_wq, &con->swork))
355                         conn_put(con);
356         }
357         read_unlock_bh(&sk->sk_callback_lock);
358 }
359
360 static int tipc_conn_rcv_sub(struct tipc_topsrv *srv,
361                              struct tipc_conn *con,
362                              struct tipc_subscr *s)
363 {
364         struct tipc_net *tn = tipc_net(srv->net);
365         struct tipc_subscription *sub;
366
367         if (tipc_sub_read(s, filter) & TIPC_SUB_CANCEL) {
368                 tipc_conn_delete_sub(con, s);
369                 return 0;
370         }
371         if (atomic_read(&tn->subscription_count) >= TIPC_MAX_SUBSCR) {
372                 pr_warn("Subscription rejected, max (%u)\n", TIPC_MAX_SUBSCR);
373                 return -1;
374         }
375         sub = tipc_sub_subscribe(srv->net, s, con->conid);
376         if (!sub)
377                 return -1;
378         atomic_inc(&tn->subscription_count);
379         spin_lock_bh(&con->sub_lock);
380         list_add(&sub->sub_list, &con->sub_list);
381         spin_unlock_bh(&con->sub_lock);
382         return 0;
383 }
384
385 static int tipc_conn_rcv_from_sock(struct tipc_conn *con)
386 {
387         struct tipc_topsrv *srv = con->server;
388         struct sock *sk = con->sock->sk;
389         struct msghdr msg = {};
390         struct tipc_subscr s;
391         struct kvec iov;
392         int ret;
393
394         iov.iov_base = &s;
395         iov.iov_len = sizeof(s);
396         msg.msg_name = NULL;
397         iov_iter_kvec(&msg.msg_iter, READ | ITER_KVEC, &iov, 1, iov.iov_len);
398         ret = sock_recvmsg(con->sock, &msg, MSG_DONTWAIT);
399         if (ret == -EWOULDBLOCK)
400                 return -EWOULDBLOCK;
401         if (ret > 0) {
402                 read_lock_bh(&sk->sk_callback_lock);
403                 ret = tipc_conn_rcv_sub(srv, con, &s);
404                 read_unlock_bh(&sk->sk_callback_lock);
405         }
406         if (ret < 0)
407                 tipc_conn_close(con);
408
409         return ret;
410 }
411
412 static void tipc_conn_recv_work(struct work_struct *work)
413 {
414         struct tipc_conn *con = container_of(work, struct tipc_conn, rwork);
415         int count = 0;
416
417         while (connected(con)) {
418                 if (tipc_conn_rcv_from_sock(con))
419                         break;
420
421                 /* Don't flood Rx machine */
422                 if (++count >= MAX_RECV_MSG_COUNT) {
423                         cond_resched();
424                         count = 0;
425                 }
426         }
427         conn_put(con);
428 }
429
430 /* tipc_conn_data_ready - interrupt callback indicating the socket has data
431  * The queued work is launched into tipc_recv_work()->tipc_conn_rcv_from_sock()
432  */
433 static void tipc_conn_data_ready(struct sock *sk)
434 {
435         struct tipc_conn *con;
436
437         read_lock_bh(&sk->sk_callback_lock);
438         con = sk->sk_user_data;
439         if (connected(con)) {
440                 conn_get(con);
441                 if (!queue_work(con->server->rcv_wq, &con->rwork))
442                         conn_put(con);
443         }
444         read_unlock_bh(&sk->sk_callback_lock);
445 }
446
447 static void tipc_topsrv_accept(struct work_struct *work)
448 {
449         struct tipc_topsrv *srv = container_of(work, struct tipc_topsrv, awork);
450         struct socket *lsock = srv->listener;
451         struct socket *newsock;
452         struct tipc_conn *con;
453         struct sock *newsk;
454         int ret;
455
456         while (1) {
457                 ret = kernel_accept(lsock, &newsock, O_NONBLOCK);
458                 if (ret < 0)
459                         return;
460                 con = tipc_conn_alloc(srv);
461                 if (IS_ERR(con)) {
462                         ret = PTR_ERR(con);
463                         sock_release(newsock);
464                         return;
465                 }
466                 /* Register callbacks */
467                 newsk = newsock->sk;
468                 write_lock_bh(&newsk->sk_callback_lock);
469                 newsk->sk_data_ready = tipc_conn_data_ready;
470                 newsk->sk_write_space = tipc_conn_write_space;
471                 newsk->sk_user_data = con;
472                 con->sock = newsock;
473                 write_unlock_bh(&newsk->sk_callback_lock);
474
475                 /* Wake up receive process in case of 'SYN+' message */
476                 newsk->sk_data_ready(newsk);
477         }
478 }
479
480 /* tipc_toprsv_listener_data_ready - interrupt callback with connection request
481  * The queued job is launched into tipc_topsrv_accept()
482  */
483 static void tipc_topsrv_listener_data_ready(struct sock *sk)
484 {
485         struct tipc_topsrv *srv;
486
487         read_lock_bh(&sk->sk_callback_lock);
488         srv = sk->sk_user_data;
489         if (srv->listener)
490                 queue_work(srv->rcv_wq, &srv->awork);
491         read_unlock_bh(&sk->sk_callback_lock);
492 }
493
494 static int tipc_topsrv_create_listener(struct tipc_topsrv *srv)
495 {
496         int imp = TIPC_CRITICAL_IMPORTANCE;
497         struct socket *lsock = NULL;
498         struct sockaddr_tipc saddr;
499         struct sock *sk;
500         int rc;
501
502         rc = sock_create_kern(srv->net, AF_TIPC, SOCK_SEQPACKET, 0, &lsock);
503         if (rc < 0)
504                 return rc;
505
506         srv->listener = lsock;
507         sk = lsock->sk;
508         write_lock_bh(&sk->sk_callback_lock);
509         sk->sk_data_ready = tipc_topsrv_listener_data_ready;
510         sk->sk_user_data = srv;
511         write_unlock_bh(&sk->sk_callback_lock);
512
513         rc = kernel_setsockopt(lsock, SOL_TIPC, TIPC_IMPORTANCE,
514                                (char *)&imp, sizeof(imp));
515         if (rc < 0)
516                 goto err;
517
518         saddr.family                    = AF_TIPC;
519         saddr.addrtype                  = TIPC_ADDR_NAMESEQ;
520         saddr.addr.nameseq.type         = TIPC_TOP_SRV;
521         saddr.addr.nameseq.lower        = TIPC_TOP_SRV;
522         saddr.addr.nameseq.upper        = TIPC_TOP_SRV;
523         saddr.scope                     = TIPC_NODE_SCOPE;
524
525         rc = kernel_bind(lsock, (struct sockaddr *)&saddr, sizeof(saddr));
526         if (rc < 0)
527                 goto err;
528         rc = kernel_listen(lsock, 0);
529         if (rc < 0)
530                 goto err;
531
532         /* As server's listening socket owner and creator is the same module,
533          * we have to decrease TIPC module reference count to guarantee that
534          * it remains zero after the server socket is created, otherwise,
535          * executing "rmmod" command is unable to make TIPC module deleted
536          * after TIPC module is inserted successfully.
537          *
538          * However, the reference count is ever increased twice in
539          * sock_create_kern(): one is to increase the reference count of owner
540          * of TIPC socket's proto_ops struct; another is to increment the
541          * reference count of owner of TIPC proto struct. Therefore, we must
542          * decrement the module reference count twice to ensure that it keeps
543          * zero after server's listening socket is created. Of course, we
544          * must bump the module reference count twice as well before the socket
545          * is closed.
546          */
547         module_put(lsock->ops->owner);
548         module_put(sk->sk_prot_creator->owner);
549
550         return 0;
551 err:
552         sock_release(lsock);
553         return -EINVAL;
554 }
555
556 bool tipc_topsrv_kern_subscr(struct net *net, u32 port, u32 type, u32 lower,
557                              u32 upper, u32 filter, int *conid)
558 {
559         struct tipc_subscr sub;
560         struct tipc_conn *con;
561         int rc;
562
563         sub.seq.type = type;
564         sub.seq.lower = lower;
565         sub.seq.upper = upper;
566         sub.timeout = TIPC_WAIT_FOREVER;
567         sub.filter = filter;
568         *(u32 *)&sub.usr_handle = port;
569
570         con = tipc_conn_alloc(tipc_topsrv(net));
571         if (IS_ERR(con))
572                 return false;
573
574         *conid = con->conid;
575         con->sock = NULL;
576         rc = tipc_conn_rcv_sub(tipc_topsrv(net), con, &sub);
577         if (rc >= 0)
578                 return true;
579         conn_put(con);
580         return false;
581 }
582
583 void tipc_topsrv_kern_unsubscr(struct net *net, int conid)
584 {
585         struct tipc_conn *con;
586
587         con = tipc_conn_lookup(tipc_topsrv(net), conid);
588         if (!con)
589                 return;
590
591         test_and_clear_bit(CF_CONNECTED, &con->flags);
592         tipc_conn_delete_sub(con, NULL);
593         conn_put(con);
594         conn_put(con);
595 }
596
597 static void tipc_topsrv_kern_evt(struct net *net, struct tipc_event *evt)
598 {
599         u32 port = *(u32 *)&evt->s.usr_handle;
600         u32 self = tipc_own_addr(net);
601         struct sk_buff_head evtq;
602         struct sk_buff *skb;
603
604         skb = tipc_msg_create(TOP_SRV, 0, INT_H_SIZE, sizeof(*evt),
605                               self, self, port, port, 0);
606         if (!skb)
607                 return;
608         msg_set_dest_droppable(buf_msg(skb), true);
609         memcpy(msg_data(buf_msg(skb)), evt, sizeof(*evt));
610         skb_queue_head_init(&evtq);
611         __skb_queue_tail(&evtq, skb);
612         tipc_sk_rcv(net, &evtq);
613 }
614
615 static int tipc_topsrv_work_start(struct tipc_topsrv *s)
616 {
617         s->rcv_wq = alloc_ordered_workqueue("tipc_rcv", 0);
618         if (!s->rcv_wq) {
619                 pr_err("can't start tipc receive workqueue\n");
620                 return -ENOMEM;
621         }
622
623         s->send_wq = alloc_ordered_workqueue("tipc_send", 0);
624         if (!s->send_wq) {
625                 pr_err("can't start tipc send workqueue\n");
626                 destroy_workqueue(s->rcv_wq);
627                 return -ENOMEM;
628         }
629
630         return 0;
631 }
632
633 static void tipc_topsrv_work_stop(struct tipc_topsrv *s)
634 {
635         destroy_workqueue(s->rcv_wq);
636         destroy_workqueue(s->send_wq);
637 }
638
639 int tipc_topsrv_start(struct net *net)
640 {
641         struct tipc_net *tn = tipc_net(net);
642         const char name[] = "topology_server";
643         struct tipc_topsrv *srv;
644         int ret;
645
646         srv = kzalloc(sizeof(*srv), GFP_ATOMIC);
647         if (!srv)
648                 return -ENOMEM;
649
650         srv->net = net;
651         srv->max_rcvbuf_size = sizeof(struct tipc_subscr);
652         INIT_WORK(&srv->awork, tipc_topsrv_accept);
653
654         strscpy(srv->name, name, sizeof(srv->name));
655         tn->topsrv = srv;
656         atomic_set(&tn->subscription_count, 0);
657
658         spin_lock_init(&srv->idr_lock);
659         idr_init(&srv->conn_idr);
660         srv->idr_in_use = 0;
661
662         ret = tipc_topsrv_work_start(srv);
663         if (ret < 0)
664                 return ret;
665
666         ret = tipc_topsrv_create_listener(srv);
667         if (ret < 0)
668                 tipc_topsrv_work_stop(srv);
669
670         return ret;
671 }
672
673 void tipc_topsrv_stop(struct net *net)
674 {
675         struct tipc_topsrv *srv = tipc_topsrv(net);
676         struct socket *lsock = srv->listener;
677         struct tipc_conn *con;
678         int id;
679
680         spin_lock_bh(&srv->idr_lock);
681         for (id = 0; srv->idr_in_use; id++) {
682                 con = idr_find(&srv->conn_idr, id);
683                 if (con) {
684                         spin_unlock_bh(&srv->idr_lock);
685                         tipc_conn_close(con);
686                         spin_lock_bh(&srv->idr_lock);
687                 }
688         }
689         __module_get(lsock->ops->owner);
690         __module_get(lsock->sk->sk_prot_creator->owner);
691         srv->listener = NULL;
692         spin_unlock_bh(&srv->idr_lock);
693         sock_release(lsock);
694         tipc_topsrv_work_stop(srv);
695         idr_destroy(&srv->conn_idr);
696         kfree(srv);
697 }