~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/net/tipc/topsrv.c

Version: ~ [ linux-5.5-rc7 ] ~ [ linux-5.4.13 ] ~ [ linux-5.3.18 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.97 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.166 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.210 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.210 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.81 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-3.9.11 ] ~ [ linux-3.8.13 ] ~ [ linux-3.7.10 ] ~ [ linux-3.6.11 ] ~ [ linux-3.5.7 ] ~ [ linux-3.4.113 ] ~ [ linux-3.3.8 ] ~ [ linux-3.2.102 ] ~ [ linux-3.1.10 ] ~ [ linux-3.0.101 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

  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, &iov, 1, iov.iov_len);
398         ret = sock_recvmsg(con->sock, &msg, MSG_DONTWAIT);
399         if (ret == -EWOULDBLOCK)
400                 return -EWOULDBLOCK;
401         if (ret == sizeof(s)) {
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 }
698 

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp