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 "bearer.h" 44#include <net/sock.h> 45#include <linux/module.h> 46#include <trace/events/sock.h> 47 48/* Number of messages to send before rescheduling */ 49#define MAX_SEND_MSG_COUNT 25 50#define MAX_RECV_MSG_COUNT 25 51#define CF_CONNECTED 1 52 53#define TIPC_SERVER_NAME_LEN 32 54 55/** 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 65 * @name: server name 66 */ 67struct 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 struct socket *listener; 76 char name[TIPC_SERVER_NAME_LEN]; 77}; 78 79/** 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 92 */ 93struct tipc_conn { 94 struct kref kref; 95 int conid; 96 struct socket *sock; 97 unsigned long flags; 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; 105}; 106 107/* An entry waiting to be sent */ 108struct outqueue_entry { 109 bool inactive; 110 struct tipc_event evt; 111 struct list_head list; 112}; 113 114static void tipc_conn_recv_work(struct work_struct *work); 115static void tipc_conn_send_work(struct work_struct *work); 116static void tipc_topsrv_kern_evt(struct net *net, struct tipc_event *evt); 117static void tipc_conn_delete_sub(struct tipc_conn *con, struct tipc_subscr *s); 118 119static bool connected(struct tipc_conn *con) 120{ 121 return con && test_bit(CF_CONNECTED, &con->flags); 122} 123 124static void tipc_conn_kref_release(struct kref *kref) 125{ 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; 129 130 spin_lock_bh(&s->idr_lock); 131 idr_remove(&s->conn_idr, con->conid); 132 s->idr_in_use--; 133 spin_unlock_bh(&s->idr_lock); 134 if (con->sock) 135 sock_release(con->sock); 136 137 spin_lock_bh(&con->outqueue_lock); 138 list_for_each_entry_safe(e, safe, &con->outqueue, list) { 139 list_del(&e->list); 140 kfree(e); 141 } 142 spin_unlock_bh(&con->outqueue_lock); 143 kfree(con); 144} 145 146static void conn_put(struct tipc_conn *con) 147{ 148 kref_put(&con->kref, tipc_conn_kref_release); 149} 150 151static void conn_get(struct tipc_conn *con) 152{ 153 kref_get(&con->kref); 154} 155 156static void tipc_conn_close(struct tipc_conn *con) 157{ 158 struct sock *sk = con->sock->sk; 159 bool disconnect = false; 160 161 write_lock_bh(&sk->sk_callback_lock); 162 disconnect = test_and_clear_bit(CF_CONNECTED, &con->flags); 163 164 if (disconnect) { 165 sk->sk_user_data = NULL; 166 tipc_conn_delete_sub(con, NULL); 167 } 168 write_unlock_bh(&sk->sk_callback_lock); 169 170 /* Handle concurrent calls from sending and receiving threads */ 171 if (!disconnect) 172 return; 173 174 /* Don't flush pending works, -just let them expire */ 175 kernel_sock_shutdown(con->sock, SHUT_RDWR); 176 177 conn_put(con); 178} 179 180static struct tipc_conn *tipc_conn_alloc(struct tipc_topsrv *s, struct socket *sock) 181{ 182 struct tipc_conn *con; 183 int ret; 184 185 con = kzalloc(sizeof(*con), GFP_ATOMIC); 186 if (!con) 187 return ERR_PTR(-ENOMEM); 188 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); 196 197 spin_lock_bh(&s->idr_lock); 198 ret = idr_alloc(&s->conn_idr, con, 0, 0, GFP_ATOMIC); 199 if (ret < 0) { 200 kfree(con); 201 spin_unlock_bh(&s->idr_lock); 202 return ERR_PTR(-ENOMEM); 203 } 204 con->conid = ret; 205 s->idr_in_use++; 206 207 set_bit(CF_CONNECTED, &con->flags); 208 con->server = s; 209 con->sock = sock; 210 conn_get(con); 211 spin_unlock_bh(&s->idr_lock); 212 213 return con; 214} 215 216static struct tipc_conn *tipc_conn_lookup(struct tipc_topsrv *s, int conid) 217{ 218 struct tipc_conn *con; 219 220 spin_lock_bh(&s->idr_lock); 221 con = idr_find(&s->conn_idr, conid); 222 if (!connected(con) || !kref_get_unless_zero(&con->kref)) 223 con = NULL; 224 spin_unlock_bh(&s->idr_lock); 225 return con; 226} 227 228/* tipc_conn_delete_sub - delete a specific or all subscriptions 229 * for a given subscriber 230 */ 231static void tipc_conn_delete_sub(struct tipc_conn *con, struct tipc_subscr *s) 232{ 233 struct tipc_net *tn = tipc_net(con->server->net); 234 struct list_head *sub_list = &con->sub_list; 235 struct tipc_subscription *sub, *tmp; 236 237 spin_lock_bh(&con->sub_lock); 238 list_for_each_entry_safe(sub, tmp, sub_list, sub_list) { 239 if (!s || !memcmp(s, &sub->evt.s, sizeof(*s))) { 240 tipc_sub_unsubscribe(sub); 241 atomic_dec(&tn->subscription_count); 242 if (s) 243 break; 244 } 245 } 246 spin_unlock_bh(&con->sub_lock); 247} 248 249static void tipc_conn_send_to_sock(struct tipc_conn *con) 250{ 251 struct list_head *queue = &con->outqueue; 252 struct tipc_topsrv *srv = con->server; 253 struct outqueue_entry *e; 254 struct tipc_event *evt; 255 struct msghdr msg; 256 struct kvec iov; 257 int count = 0; 258 int ret; 259 260 spin_lock_bh(&con->outqueue_lock); 261 262 while (!list_empty(queue)) { 263 e = list_first_entry(queue, struct outqueue_entry, list); 264 evt = &e->evt; 265 spin_unlock_bh(&con->outqueue_lock); 266 267 if (e->inactive) 268 tipc_conn_delete_sub(con, &evt->s); 269 270 memset(&msg, 0, sizeof(msg)); 271 msg.msg_flags = MSG_DONTWAIT; 272 iov.iov_base = evt; 273 iov.iov_len = sizeof(*evt); 274 msg.msg_name = NULL; 275 276 if (con->sock) { 277 ret = kernel_sendmsg(con->sock, &msg, &iov, 278 1, sizeof(*evt)); 279 if (ret == -EWOULDBLOCK || ret == 0) { 280 cond_resched(); 281 return; 282 } else if (ret < 0) { 283 return tipc_conn_close(con); 284 } 285 } else { 286 tipc_topsrv_kern_evt(srv->net, evt); 287 } 288 289 /* Don't starve users filling buffers */ 290 if (++count >= MAX_SEND_MSG_COUNT) { 291 cond_resched(); 292 count = 0; 293 } 294 spin_lock_bh(&con->outqueue_lock); 295 list_del(&e->list); 296 kfree(e); 297 } 298 spin_unlock_bh(&con->outqueue_lock); 299} 300 301static void tipc_conn_send_work(struct work_struct *work) 302{ 303 struct tipc_conn *con = container_of(work, struct tipc_conn, swork); 304 305 if (connected(con)) 306 tipc_conn_send_to_sock(con); 307 308 conn_put(con); 309} 310 311/* tipc_topsrv_queue_evt() - interrupt level call from a subscription instance 312 * The queued work is launched into tipc_conn_send_work()->tipc_conn_send_to_sock() 313 */ 314void tipc_topsrv_queue_evt(struct net *net, int conid, 315 u32 event, struct tipc_event *evt) 316{ 317 struct tipc_topsrv *srv = tipc_topsrv(net); 318 struct outqueue_entry *e; 319 struct tipc_conn *con; 320 321 con = tipc_conn_lookup(srv, conid); 322 if (!con) 323 return; 324 325 if (!connected(con)) 326 goto err; 327 328 e = kmalloc(sizeof(*e), GFP_ATOMIC); 329 if (!e) 330 goto err; 331 e->inactive = (event == TIPC_SUBSCR_TIMEOUT); 332 memcpy(&e->evt, evt, sizeof(*evt)); 333 spin_lock_bh(&con->outqueue_lock); 334 list_add_tail(&e->list, &con->outqueue); 335 spin_unlock_bh(&con->outqueue_lock); 336 337 if (queue_work(srv->send_wq, &con->swork)) 338 return; 339err: 340 conn_put(con); 341} 342 343/* tipc_conn_write_space - interrupt callback after a sendmsg EAGAIN 344 * Indicates that there now is more space in the send buffer 345 * The queued work is launched into tipc_send_work()->tipc_conn_send_to_sock() 346 */ 347static void tipc_conn_write_space(struct sock *sk) 348{ 349 struct tipc_conn *con; 350 351 read_lock_bh(&sk->sk_callback_lock); 352 con = sk->sk_user_data; 353 if (connected(con)) { 354 conn_get(con); 355 if (!queue_work(con->server->send_wq, &con->swork)) 356 conn_put(con); 357 } 358 read_unlock_bh(&sk->sk_callback_lock); 359} 360 361static int tipc_conn_rcv_sub(struct tipc_topsrv *srv, 362 struct tipc_conn *con, 363 struct tipc_subscr *s) 364{ 365 struct tipc_net *tn = tipc_net(srv->net); 366 struct tipc_subscription *sub; 367 u32 s_filter = tipc_sub_read(s, filter); 368 369 if (s_filter & TIPC_SUB_CANCEL) { 370 tipc_sub_write(s, filter, s_filter & ~TIPC_SUB_CANCEL); 371 tipc_conn_delete_sub(con, s); 372 return 0; 373 } 374 if (atomic_read(&tn->subscription_count) >= TIPC_MAX_SUBSCR) { 375 pr_warn("Subscription rejected, max (%u)\n", TIPC_MAX_SUBSCR); 376 return -1; 377 } 378 sub = tipc_sub_subscribe(srv->net, s, con->conid); 379 if (!sub) 380 return -1; 381 atomic_inc(&tn->subscription_count); 382 spin_lock_bh(&con->sub_lock); 383 list_add(&sub->sub_list, &con->sub_list); 384 spin_unlock_bh(&con->sub_lock); 385 return 0; 386} 387 388static int tipc_conn_rcv_from_sock(struct tipc_conn *con) 389{ 390 struct tipc_topsrv *srv = con->server; 391 struct sock *sk = con->sock->sk; 392 struct msghdr msg = {}; 393 struct tipc_subscr s; 394 struct kvec iov; 395 int ret; 396 397 iov.iov_base = &s; 398 iov.iov_len = sizeof(s); 399 msg.msg_name = NULL; 400 iov_iter_kvec(&msg.msg_iter, ITER_DEST, &iov, 1, iov.iov_len); 401 ret = sock_recvmsg(con->sock, &msg, MSG_DONTWAIT); 402 if (ret == -EWOULDBLOCK) 403 return -EWOULDBLOCK; 404 if (ret == sizeof(s)) { 405 read_lock_bh(&sk->sk_callback_lock); 406 /* RACE: the connection can be closed in the meantime */ 407 if (likely(connected(con))) 408 ret = tipc_conn_rcv_sub(srv, con, &s); 409 read_unlock_bh(&sk->sk_callback_lock); 410 if (!ret) 411 return 0; 412 } 413 414 tipc_conn_close(con); 415 return ret; 416} 417 418static void tipc_conn_recv_work(struct work_struct *work) 419{ 420 struct tipc_conn *con = container_of(work, struct tipc_conn, rwork); 421 int count = 0; 422 423 while (connected(con)) { 424 if (tipc_conn_rcv_from_sock(con)) 425 break; 426 427 /* Don't flood Rx machine */ 428 if (++count >= MAX_RECV_MSG_COUNT) { 429 cond_resched(); 430 count = 0; 431 } 432 } 433 conn_put(con); 434} 435 436/* tipc_conn_data_ready - interrupt callback indicating the socket has data 437 * The queued work is launched into tipc_recv_work()->tipc_conn_rcv_from_sock() 438 */ 439static void tipc_conn_data_ready(struct sock *sk) 440{ 441 struct tipc_conn *con; 442 443 trace_sk_data_ready(sk); 444 445 read_lock_bh(&sk->sk_callback_lock); 446 con = sk->sk_user_data; 447 if (connected(con)) { 448 conn_get(con); 449 if (!queue_work(con->server->rcv_wq, &con->rwork)) 450 conn_put(con); 451 } 452 read_unlock_bh(&sk->sk_callback_lock); 453} 454 455static void tipc_topsrv_accept(struct work_struct *work) 456{ 457 struct tipc_topsrv *srv = container_of(work, struct tipc_topsrv, awork); 458 struct socket *newsock, *lsock; 459 struct tipc_conn *con; 460 struct sock *newsk; 461 int ret; 462 463 spin_lock_bh(&srv->idr_lock); 464 if (!srv->listener) { 465 spin_unlock_bh(&srv->idr_lock); 466 return; 467 } 468 lsock = srv->listener; 469 spin_unlock_bh(&srv->idr_lock); 470 471 while (1) { 472 ret = kernel_accept(lsock, &newsock, O_NONBLOCK); 473 if (ret < 0) 474 return; 475 con = tipc_conn_alloc(srv, newsock); 476 if (IS_ERR(con)) { 477 ret = PTR_ERR(con); 478 sock_release(newsock); 479 return; 480 } 481 /* Register callbacks */ 482 newsk = newsock->sk; 483 write_lock_bh(&newsk->sk_callback_lock); 484 newsk->sk_data_ready = tipc_conn_data_ready; 485 newsk->sk_write_space = tipc_conn_write_space; 486 newsk->sk_user_data = con; 487 write_unlock_bh(&newsk->sk_callback_lock); 488 489 /* Wake up receive process in case of 'SYN+' message */ 490 newsk->sk_data_ready(newsk); 491 conn_put(con); 492 } 493} 494 495/* tipc_topsrv_listener_data_ready - interrupt callback with connection request 496 * The queued job is launched into tipc_topsrv_accept() 497 */ 498static void tipc_topsrv_listener_data_ready(struct sock *sk) 499{ 500 struct tipc_topsrv *srv; 501 502 trace_sk_data_ready(sk); 503 504 read_lock_bh(&sk->sk_callback_lock); 505 srv = sk->sk_user_data; 506 if (srv) 507 queue_work(srv->rcv_wq, &srv->awork); 508 read_unlock_bh(&sk->sk_callback_lock); 509} 510 511static int tipc_topsrv_create_listener(struct tipc_topsrv *srv) 512{ 513 struct socket *lsock = NULL; 514 struct sockaddr_tipc saddr; 515 struct sock *sk; 516 int rc; 517 518 rc = sock_create_kern(srv->net, AF_TIPC, SOCK_SEQPACKET, 0, &lsock); 519 if (rc < 0) 520 return rc; 521 522 srv->listener = lsock; 523 sk = lsock->sk; 524 write_lock_bh(&sk->sk_callback_lock); 525 sk->sk_data_ready = tipc_topsrv_listener_data_ready; 526 sk->sk_user_data = srv; 527 write_unlock_bh(&sk->sk_callback_lock); 528 529 lock_sock(sk); 530 rc = tsk_set_importance(sk, TIPC_CRITICAL_IMPORTANCE); 531 release_sock(sk); 532 if (rc < 0) 533 goto err; 534 535 saddr.family = AF_TIPC; 536 saddr.addrtype = TIPC_SERVICE_RANGE; 537 saddr.addr.nameseq.type = TIPC_TOP_SRV; 538 saddr.addr.nameseq.lower = TIPC_TOP_SRV; 539 saddr.addr.nameseq.upper = TIPC_TOP_SRV; 540 saddr.scope = TIPC_NODE_SCOPE; 541 542 rc = tipc_sk_bind(lsock, (struct sockaddr *)&saddr, sizeof(saddr)); 543 if (rc < 0) 544 goto err; 545 rc = kernel_listen(lsock, 0); 546 if (rc < 0) 547 goto err; 548 549 /* As server's listening socket owner and creator is the same module, 550 * we have to decrease TIPC module reference count to guarantee that 551 * it remains zero after the server socket is created, otherwise, 552 * executing "rmmod" command is unable to make TIPC module deleted 553 * after TIPC module is inserted successfully. 554 * 555 * However, the reference count is ever increased twice in 556 * sock_create_kern(): one is to increase the reference count of owner 557 * of TIPC socket's proto_ops struct; another is to increment the 558 * reference count of owner of TIPC proto struct. Therefore, we must 559 * decrement the module reference count twice to ensure that it keeps 560 * zero after server's listening socket is created. Of course, we 561 * must bump the module reference count twice as well before the socket 562 * is closed. 563 */ 564 module_put(lsock->ops->owner); 565 module_put(sk->sk_prot_creator->owner); 566 567 return 0; 568err: 569 sock_release(lsock); 570 return -EINVAL; 571} 572 573bool tipc_topsrv_kern_subscr(struct net *net, u32 port, u32 type, u32 lower, 574 u32 upper, u32 filter, int *conid) 575{ 576 struct tipc_subscr sub; 577 struct tipc_conn *con; 578 int rc; 579 580 sub.seq.type = type; 581 sub.seq.lower = lower; 582 sub.seq.upper = upper; 583 sub.timeout = TIPC_WAIT_FOREVER; 584 sub.filter = filter; 585 *(u64 *)&sub.usr_handle = (u64)port; 586 587 con = tipc_conn_alloc(tipc_topsrv(net), NULL); 588 if (IS_ERR(con)) 589 return false; 590 591 *conid = con->conid; 592 rc = tipc_conn_rcv_sub(tipc_topsrv(net), con, &sub); 593 if (rc) 594 conn_put(con); 595 596 conn_put(con); 597 return !rc; 598} 599 600void tipc_topsrv_kern_unsubscr(struct net *net, int conid) 601{ 602 struct tipc_conn *con; 603 604 con = tipc_conn_lookup(tipc_topsrv(net), conid); 605 if (!con) 606 return; 607 608 test_and_clear_bit(CF_CONNECTED, &con->flags); 609 tipc_conn_delete_sub(con, NULL); 610 conn_put(con); 611 conn_put(con); 612} 613 614static void tipc_topsrv_kern_evt(struct net *net, struct tipc_event *evt) 615{ 616 u32 port = *(u32 *)&evt->s.usr_handle; 617 u32 self = tipc_own_addr(net); 618 struct sk_buff_head evtq; 619 struct sk_buff *skb; 620 621 skb = tipc_msg_create(TOP_SRV, 0, INT_H_SIZE, sizeof(*evt), 622 self, self, port, port, 0); 623 if (!skb) 624 return; 625 msg_set_dest_droppable(buf_msg(skb), true); 626 memcpy(msg_data(buf_msg(skb)), evt, sizeof(*evt)); 627 skb_queue_head_init(&evtq); 628 __skb_queue_tail(&evtq, skb); 629 tipc_loopback_trace(net, &evtq); 630 tipc_sk_rcv(net, &evtq); 631} 632 633static int tipc_topsrv_work_start(struct tipc_topsrv *s) 634{ 635 s->rcv_wq = alloc_ordered_workqueue("tipc_rcv", 0); 636 if (!s->rcv_wq) { 637 pr_err("can't start tipc receive workqueue\n"); 638 return -ENOMEM; 639 } 640 641 s->send_wq = alloc_ordered_workqueue("tipc_send", 0); 642 if (!s->send_wq) { 643 pr_err("can't start tipc send workqueue\n"); 644 destroy_workqueue(s->rcv_wq); 645 return -ENOMEM; 646 } 647 648 return 0; 649} 650 651static void tipc_topsrv_work_stop(struct tipc_topsrv *s) 652{ 653 destroy_workqueue(s->rcv_wq); 654 destroy_workqueue(s->send_wq); 655} 656 657static int tipc_topsrv_start(struct net *net) 658{ 659 struct tipc_net *tn = tipc_net(net); 660 const char name[] = "topology_server"; 661 struct tipc_topsrv *srv; 662 int ret; 663 664 srv = kzalloc(sizeof(*srv), GFP_ATOMIC); 665 if (!srv) 666 return -ENOMEM; 667 668 srv->net = net; 669 INIT_WORK(&srv->awork, tipc_topsrv_accept); 670 671 strscpy(srv->name, name, sizeof(srv->name)); 672 tn->topsrv = srv; 673 atomic_set(&tn->subscription_count, 0); 674 675 spin_lock_init(&srv->idr_lock); 676 idr_init(&srv->conn_idr); 677 srv->idr_in_use = 0; 678 679 ret = tipc_topsrv_work_start(srv); 680 if (ret < 0) 681 goto err_start; 682 683 ret = tipc_topsrv_create_listener(srv); 684 if (ret < 0) 685 goto err_create; 686 687 return 0; 688 689err_create: 690 tipc_topsrv_work_stop(srv); 691err_start: 692 kfree(srv); 693 return ret; 694} 695 696static void tipc_topsrv_stop(struct net *net) 697{ 698 struct tipc_topsrv *srv = tipc_topsrv(net); 699 struct socket *lsock = srv->listener; 700 struct tipc_conn *con; 701 int id; 702 703 spin_lock_bh(&srv->idr_lock); 704 for (id = 0; srv->idr_in_use; id++) { 705 con = idr_find(&srv->conn_idr, id); 706 if (con) { 707 spin_unlock_bh(&srv->idr_lock); 708 tipc_conn_close(con); 709 spin_lock_bh(&srv->idr_lock); 710 } 711 } 712 __module_get(lsock->ops->owner); 713 __module_get(lsock->sk->sk_prot_creator->owner); 714 srv->listener = NULL; 715 spin_unlock_bh(&srv->idr_lock); 716 717 tipc_topsrv_work_stop(srv); 718 sock_release(lsock); 719 idr_destroy(&srv->conn_idr); 720 kfree(srv); 721} 722 723int __net_init tipc_topsrv_init_net(struct net *net) 724{ 725 return tipc_topsrv_start(net); 726} 727 728void __net_exit tipc_topsrv_exit_net(struct net *net) 729{ 730 tipc_topsrv_stop(net); 731} 732