Linux Audio

Check our new training course

Embedded Linux training

Mar 31-Apr 8, 2025
Register
Loading...
v5.9
  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
 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
 51#define CF_SERVER		2
 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)
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	spin_unlock_bh(&s->idr_lock);
207
208	set_bit(CF_CONNECTED, &con->flags);
209	con->server = s;
 
 
 
210
211	return con;
212}
213
214static struct tipc_conn *tipc_conn_lookup(struct tipc_topsrv *s, int conid)
215{
216	struct tipc_conn *con;
217
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))
221		con = NULL;
222	spin_unlock_bh(&s->idr_lock);
223	return con;
224}
225
226/* tipc_conn_delete_sub - delete a specific or all subscriptions
227 * for a given subscriber
228 */
229static void tipc_conn_delete_sub(struct tipc_conn *con, struct tipc_subscr *s)
230{
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;
234
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);
240			if (s)
241				break;
242		}
243	}
244	spin_unlock_bh(&con->sub_lock);
245}
246
247static void tipc_conn_send_to_sock(struct tipc_conn *con)
248{
249	struct list_head *queue = &con->outqueue;
250	struct tipc_topsrv *srv = con->server;
251	struct outqueue_entry *e;
252	struct tipc_event *evt;
253	struct msghdr msg;
254	struct kvec iov;
255	int count = 0;
256	int ret;
257
258	spin_lock_bh(&con->outqueue_lock);
259
260	while (!list_empty(queue)) {
261		e = list_first_entry(queue, struct outqueue_entry, list);
262		evt = &e->evt;
263		spin_unlock_bh(&con->outqueue_lock);
264
265		if (e->inactive)
266			tipc_conn_delete_sub(con, &evt->s);
267
268		memset(&msg, 0, sizeof(msg));
269		msg.msg_flags = MSG_DONTWAIT;
270		iov.iov_base = evt;
271		iov.iov_len = sizeof(*evt);
272		msg.msg_name = NULL;
273
274		if (con->sock) {
275			ret = kernel_sendmsg(con->sock, &msg, &iov,
276					     1, sizeof(*evt));
277			if (ret == -EWOULDBLOCK || ret == 0) {
278				cond_resched();
279				return;
280			} else if (ret < 0) {
281				return tipc_conn_close(con);
282			}
283		} else {
284			tipc_topsrv_kern_evt(srv->net, evt);
285		}
286
287		/* Don't starve users filling buffers */
288		if (++count >= MAX_SEND_MSG_COUNT) {
289			cond_resched();
290			count = 0;
291		}
292		spin_lock_bh(&con->outqueue_lock);
293		list_del(&e->list);
294		kfree(e);
295	}
296	spin_unlock_bh(&con->outqueue_lock);
297}
298
299static void tipc_conn_send_work(struct work_struct *work)
300{
301	struct tipc_conn *con = container_of(work, struct tipc_conn, swork);
302
303	if (connected(con))
304		tipc_conn_send_to_sock(con);
305
306	conn_put(con);
307}
308
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()
311 */
312void tipc_topsrv_queue_evt(struct net *net, int conid,
313			   u32 event, struct tipc_event *evt)
314{
315	struct tipc_topsrv *srv = tipc_topsrv(net);
316	struct outqueue_entry *e;
317	struct tipc_conn *con;
318
319	con = tipc_conn_lookup(srv, conid);
320	if (!con)
321		return;
322
323	if (!connected(con))
324		goto err;
325
326	e = kmalloc(sizeof(*e), GFP_ATOMIC);
327	if (!e)
328		goto err;
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);
334
335	if (queue_work(srv->send_wq, &con->swork))
336		return;
337err:
338	conn_put(con);
339}
340
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()
344 */
345static void tipc_conn_write_space(struct sock *sk)
346{
347	struct tipc_conn *con;
348
349	read_lock_bh(&sk->sk_callback_lock);
350	con = sk->sk_user_data;
351	if (connected(con)) {
352		conn_get(con);
353		if (!queue_work(con->server->send_wq, &con->swork))
354			conn_put(con);
355	}
356	read_unlock_bh(&sk->sk_callback_lock);
357}
358
359static int tipc_conn_rcv_sub(struct tipc_topsrv *srv,
360			     struct tipc_conn *con,
361			     struct tipc_subscr *s)
362{
363	struct tipc_net *tn = tipc_net(srv->net);
364	struct tipc_subscription *sub;
365	u32 s_filter = tipc_sub_read(s, filter);
366
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);
370		return 0;
371	}
372	if (atomic_read(&tn->subscription_count) >= TIPC_MAX_SUBSCR) {
373		pr_warn("Subscription rejected, max (%u)\n", TIPC_MAX_SUBSCR);
374		return -1;
375	}
376	sub = tipc_sub_subscribe(srv->net, s, con->conid);
377	if (!sub)
378		return -1;
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);
383	return 0;
384}
385
386static int tipc_conn_rcv_from_sock(struct tipc_conn *con)
387{
388	struct tipc_topsrv *srv = con->server;
389	struct sock *sk = con->sock->sk;
390	struct msghdr msg = {};
391	struct tipc_subscr s;
392	struct kvec iov;
393	int ret;
394
395	iov.iov_base = &s;
396	iov.iov_len = sizeof(s);
397	msg.msg_name = NULL;
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)
401		return -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);
408		if (!ret)
409			return 0;
410	}
411
412	tipc_conn_close(con);
413	return ret;
414}
415
416static void tipc_conn_recv_work(struct work_struct *work)
417{
418	struct tipc_conn *con = container_of(work, struct tipc_conn, rwork);
419	int count = 0;
420
421	while (connected(con)) {
422		if (tipc_conn_rcv_from_sock(con))
423			break;
424
425		/* Don't flood Rx machine */
426		if (++count >= MAX_RECV_MSG_COUNT) {
427			cond_resched();
428			count = 0;
429		}
430	}
431	conn_put(con);
432}
433
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()
436 */
437static void tipc_conn_data_ready(struct sock *sk)
438{
439	struct tipc_conn *con;
440
 
 
441	read_lock_bh(&sk->sk_callback_lock);
442	con = sk->sk_user_data;
443	if (connected(con)) {
444		conn_get(con);
445		if (!queue_work(con->server->rcv_wq, &con->rwork))
446			conn_put(con);
447	}
448	read_unlock_bh(&sk->sk_callback_lock);
449}
450
451static void tipc_topsrv_accept(struct work_struct *work)
452{
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;
457	struct sock *newsk;
458	int ret;
459
 
 
 
 
 
 
 
 
460	while (1) {
461		ret = kernel_accept(lsock, &newsock, O_NONBLOCK);
462		if (ret < 0)
463			return;
464		con = tipc_conn_alloc(srv);
465		if (IS_ERR(con)) {
466			ret = PTR_ERR(con);
467			sock_release(newsock);
468			return;
469		}
470		/* Register callbacks */
471		newsk = newsock->sk;
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;
476		con->sock = newsock;
477		write_unlock_bh(&newsk->sk_callback_lock);
478
479		/* Wake up receive process in case of 'SYN+' message */
480		newsk->sk_data_ready(newsk);
 
481	}
482}
483
484/* tipc_topsrv_listener_data_ready - interrupt callback with connection request
485 * The queued job is launched into tipc_topsrv_accept()
486 */
487static void tipc_topsrv_listener_data_ready(struct sock *sk)
488{
489	struct tipc_topsrv *srv;
490
 
 
491	read_lock_bh(&sk->sk_callback_lock);
492	srv = sk->sk_user_data;
493	if (srv->listener)
494		queue_work(srv->rcv_wq, &srv->awork);
495	read_unlock_bh(&sk->sk_callback_lock);
496}
497
498static int tipc_topsrv_create_listener(struct tipc_topsrv *srv)
499{
500	struct socket *lsock = NULL;
501	struct sockaddr_tipc saddr;
502	struct sock *sk;
503	int rc;
504
505	rc = sock_create_kern(srv->net, AF_TIPC, SOCK_SEQPACKET, 0, &lsock);
506	if (rc < 0)
507		return rc;
508
509	srv->listener = lsock;
510	sk = lsock->sk;
511	write_lock_bh(&sk->sk_callback_lock);
512	sk->sk_data_ready = tipc_topsrv_listener_data_ready;
513	sk->sk_user_data = srv;
514	write_unlock_bh(&sk->sk_callback_lock);
515
516	lock_sock(sk);
517	rc = tsk_set_importance(sk, TIPC_CRITICAL_IMPORTANCE);
518	release_sock(sk);
519	if (rc < 0)
520		goto err;
521
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;
528
529	rc = kernel_bind(lsock, (struct sockaddr *)&saddr, sizeof(saddr));
530	if (rc < 0)
531		goto err;
532	rc = kernel_listen(lsock, 0);
533	if (rc < 0)
534		goto err;
535
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.
541	 *
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
549	 * is closed.
550	 */
551	module_put(lsock->ops->owner);
552	module_put(sk->sk_prot_creator->owner);
553
554	return 0;
555err:
556	sock_release(lsock);
557	return -EINVAL;
558}
559
560bool tipc_topsrv_kern_subscr(struct net *net, u32 port, u32 type, u32 lower,
561			     u32 upper, u32 filter, int *conid)
562{
563	struct tipc_subscr sub;
564	struct tipc_conn *con;
565	int rc;
566
567	sub.seq.type = type;
568	sub.seq.lower = lower;
569	sub.seq.upper = upper;
570	sub.timeout = TIPC_WAIT_FOREVER;
571	sub.filter = filter;
572	*(u32 *)&sub.usr_handle = port;
573
574	con = tipc_conn_alloc(tipc_topsrv(net));
575	if (IS_ERR(con))
576		return false;
577
578	*conid = con->conid;
579	con->sock = NULL;
580	rc = tipc_conn_rcv_sub(tipc_topsrv(net), con, &sub);
581	if (rc >= 0)
582		return true;
 
583	conn_put(con);
584	return false;
585}
586
587void tipc_topsrv_kern_unsubscr(struct net *net, int conid)
588{
589	struct tipc_conn *con;
590
591	con = tipc_conn_lookup(tipc_topsrv(net), conid);
592	if (!con)
593		return;
594
595	test_and_clear_bit(CF_CONNECTED, &con->flags);
596	tipc_conn_delete_sub(con, NULL);
597	conn_put(con);
598	conn_put(con);
599}
600
601static void tipc_topsrv_kern_evt(struct net *net, struct tipc_event *evt)
602{
603	u32 port = *(u32 *)&evt->s.usr_handle;
604	u32 self = tipc_own_addr(net);
605	struct sk_buff_head evtq;
606	struct sk_buff *skb;
607
608	skb = tipc_msg_create(TOP_SRV, 0, INT_H_SIZE, sizeof(*evt),
609			      self, self, port, port, 0);
610	if (!skb)
611		return;
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);
618}
619
620static int tipc_topsrv_work_start(struct tipc_topsrv *s)
621{
622	s->rcv_wq = alloc_ordered_workqueue("tipc_rcv", 0);
623	if (!s->rcv_wq) {
624		pr_err("can't start tipc receive workqueue\n");
625		return -ENOMEM;
626	}
627
628	s->send_wq = alloc_ordered_workqueue("tipc_send", 0);
629	if (!s->send_wq) {
630		pr_err("can't start tipc send workqueue\n");
631		destroy_workqueue(s->rcv_wq);
632		return -ENOMEM;
633	}
634
635	return 0;
636}
637
638static void tipc_topsrv_work_stop(struct tipc_topsrv *s)
639{
640	destroy_workqueue(s->rcv_wq);
641	destroy_workqueue(s->send_wq);
642}
643
644static int tipc_topsrv_start(struct net *net)
645{
646	struct tipc_net *tn = tipc_net(net);
647	const char name[] = "topology_server";
648	struct tipc_topsrv *srv;
649	int ret;
650
651	srv = kzalloc(sizeof(*srv), GFP_ATOMIC);
652	if (!srv)
653		return -ENOMEM;
654
655	srv->net = net;
656	INIT_WORK(&srv->awork, tipc_topsrv_accept);
657
658	strscpy(srv->name, name, sizeof(srv->name));
659	tn->topsrv = srv;
660	atomic_set(&tn->subscription_count, 0);
661
662	spin_lock_init(&srv->idr_lock);
663	idr_init(&srv->conn_idr);
664	srv->idr_in_use = 0;
665
666	ret = tipc_topsrv_work_start(srv);
667	if (ret < 0)
668		return ret;
669
670	ret = tipc_topsrv_create_listener(srv);
671	if (ret < 0)
672		tipc_topsrv_work_stop(srv);
 
 
673
 
 
 
 
674	return ret;
675}
676
677static void tipc_topsrv_stop(struct net *net)
678{
679	struct tipc_topsrv *srv = tipc_topsrv(net);
680	struct socket *lsock = srv->listener;
681	struct tipc_conn *con;
682	int id;
683
684	spin_lock_bh(&srv->idr_lock);
685	for (id = 0; srv->idr_in_use; id++) {
686		con = idr_find(&srv->conn_idr, id);
687		if (con) {
688			spin_unlock_bh(&srv->idr_lock);
689			tipc_conn_close(con);
690			spin_lock_bh(&srv->idr_lock);
691		}
692	}
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);
697	sock_release(lsock);
698	tipc_topsrv_work_stop(srv);
 
699	idr_destroy(&srv->conn_idr);
700	kfree(srv);
701}
702
703int __net_init tipc_topsrv_init_net(struct net *net)
704{
705	return tipc_topsrv_start(net);
706}
707
708void __net_exit tipc_topsrv_exit_net(struct net *net)
709{
710	tipc_topsrv_stop(net);
711}
v6.9.4
  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}