]> git.proxmox.com Git - mirror_ubuntu-artful-kernel.git/blob - net/tipc/server.c
tipc: fix connection refcount error
[mirror_ubuntu-artful-kernel.git] / net / tipc / server.c
1 /*
2 * net/tipc/server.c: TIPC server infrastructure
3 *
4 * Copyright (c) 2012-2013, Wind River Systems
5 * All rights reserved.
6 *
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions are met:
9 *
10 * 1. Redistributions of source code must retain the above copyright
11 * notice, this list of conditions and the following disclaimer.
12 * 2. Redistributions in binary form must reproduce the above copyright
13 * notice, this list of conditions and the following disclaimer in the
14 * documentation and/or other materials provided with the distribution.
15 * 3. Neither the names of the copyright holders nor the names of its
16 * contributors may be used to endorse or promote products derived from
17 * this software without specific prior written permission.
18 *
19 * Alternatively, this software may be distributed under the terms of the
20 * GNU General Public License ("GPL") version 2 as published by the Free
21 * Software Foundation.
22 *
23 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
24 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
25 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
26 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
27 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
28 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
29 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
30 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
31 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
32 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
33 * POSSIBILITY OF SUCH DAMAGE.
34 */
35
36 #include "server.h"
37 #include "core.h"
38 #include "socket.h"
39 #include <net/sock.h>
40 #include <linux/module.h>
41
42 /* Number of messages to send before rescheduling */
43 #define MAX_SEND_MSG_COUNT 25
44 #define MAX_RECV_MSG_COUNT 25
45 #define CF_CONNECTED 1
46 #define CF_SERVER 2
47
48 #define sock2con(x) ((struct tipc_conn *)(x)->sk_user_data)
49
50 /**
51 * struct tipc_conn - TIPC connection structure
52 * @kref: reference counter to connection object
53 * @conid: connection identifier
54 * @sock: socket handler associated with connection
55 * @flags: indicates connection state
56 * @server: pointer to connected server
57 * @rwork: receive work item
58 * @usr_data: user-specified field
59 * @rx_action: what to do when connection socket is active
60 * @outqueue: pointer to first outbound message in queue
61 * @outqueue_lock: control access to the outqueue
62 * @outqueue: list of connection objects for its server
63 * @swork: send work item
64 */
65 struct tipc_conn {
66 struct kref kref;
67 int conid;
68 struct socket *sock;
69 unsigned long flags;
70 struct tipc_server *server;
71 struct work_struct rwork;
72 int (*rx_action) (struct tipc_conn *con);
73 void *usr_data;
74 struct list_head outqueue;
75 spinlock_t outqueue_lock;
76 struct work_struct swork;
77 };
78
79 /* An entry waiting to be sent */
80 struct outqueue_entry {
81 struct list_head list;
82 struct kvec iov;
83 struct sockaddr_tipc dest;
84 };
85
86 static void tipc_recv_work(struct work_struct *work);
87 static void tipc_send_work(struct work_struct *work);
88 static void tipc_clean_outqueues(struct tipc_conn *con);
89 static void tipc_sock_release(struct tipc_conn *con);
90
91 static void tipc_conn_kref_release(struct kref *kref)
92 {
93 struct tipc_conn *con = container_of(kref, struct tipc_conn, kref);
94 struct tipc_server *s = con->server;
95 struct sockaddr_tipc *saddr = s->saddr;
96 struct socket *sock = con->sock;
97 struct sock *sk;
98
99 if (sock) {
100 sk = sock->sk;
101 if (test_bit(CF_SERVER, &con->flags)) {
102 __module_get(sock->ops->owner);
103 __module_get(sk->sk_prot_creator->owner);
104 }
105 saddr->scope = -TIPC_NODE_SCOPE;
106 kernel_bind(sock, (struct sockaddr *)saddr, sizeof(*saddr));
107 tipc_sock_release(con);
108 sock_release(sock);
109 con->sock = NULL;
110
111 spin_lock_bh(&s->idr_lock);
112 idr_remove(&s->conn_idr, con->conid);
113 s->idr_in_use--;
114 spin_unlock_bh(&s->idr_lock);
115 }
116
117 tipc_clean_outqueues(con);
118 kfree(con);
119 }
120
121 static void conn_put(struct tipc_conn *con)
122 {
123 kref_put(&con->kref, tipc_conn_kref_release);
124 }
125
126 static void conn_get(struct tipc_conn *con)
127 {
128 kref_get(&con->kref);
129 }
130
131 static struct tipc_conn *tipc_conn_lookup(struct tipc_server *s, int conid)
132 {
133 struct tipc_conn *con;
134
135 spin_lock_bh(&s->idr_lock);
136 con = idr_find(&s->conn_idr, conid);
137 if (con && test_bit(CF_CONNECTED, &con->flags))
138 conn_get(con);
139 else
140 con = NULL;
141 spin_unlock_bh(&s->idr_lock);
142 return con;
143 }
144
145 static void sock_data_ready(struct sock *sk)
146 {
147 struct tipc_conn *con;
148
149 read_lock_bh(&sk->sk_callback_lock);
150 con = sock2con(sk);
151 if (con && test_bit(CF_CONNECTED, &con->flags)) {
152 conn_get(con);
153 if (!queue_work(con->server->rcv_wq, &con->rwork))
154 conn_put(con);
155 }
156 read_unlock_bh(&sk->sk_callback_lock);
157 }
158
159 static void sock_write_space(struct sock *sk)
160 {
161 struct tipc_conn *con;
162
163 read_lock_bh(&sk->sk_callback_lock);
164 con = sock2con(sk);
165 if (con && test_bit(CF_CONNECTED, &con->flags)) {
166 conn_get(con);
167 if (!queue_work(con->server->send_wq, &con->swork))
168 conn_put(con);
169 }
170 read_unlock_bh(&sk->sk_callback_lock);
171 }
172
173 static void tipc_register_callbacks(struct socket *sock, struct tipc_conn *con)
174 {
175 struct sock *sk = sock->sk;
176
177 write_lock_bh(&sk->sk_callback_lock);
178
179 sk->sk_data_ready = sock_data_ready;
180 sk->sk_write_space = sock_write_space;
181 sk->sk_user_data = con;
182
183 con->sock = sock;
184
185 write_unlock_bh(&sk->sk_callback_lock);
186 }
187
188 static void tipc_unregister_callbacks(struct tipc_conn *con)
189 {
190 struct sock *sk = con->sock->sk;
191
192 write_lock_bh(&sk->sk_callback_lock);
193 sk->sk_user_data = NULL;
194 write_unlock_bh(&sk->sk_callback_lock);
195 }
196
197 static void tipc_sock_release(struct tipc_conn *con)
198 {
199 struct tipc_server *s = con->server;
200
201 if (con->conid)
202 s->tipc_conn_release(con->conid, con->usr_data);
203
204 tipc_unregister_callbacks(con);
205 }
206
207 static void tipc_close_conn(struct tipc_conn *con)
208 {
209 if (test_and_clear_bit(CF_CONNECTED, &con->flags)) {
210
211 /* We shouldn't flush pending works as we may be in the
212 * thread. In fact the races with pending rx/tx work structs
213 * are harmless for us here as we have already deleted this
214 * connection from server connection list.
215 */
216 kernel_sock_shutdown(con->sock, SHUT_RDWR);
217
218 conn_put(con);
219 }
220 }
221
222 static struct tipc_conn *tipc_alloc_conn(struct tipc_server *s)
223 {
224 struct tipc_conn *con;
225 int ret;
226
227 con = kzalloc(sizeof(struct tipc_conn), GFP_ATOMIC);
228 if (!con)
229 return ERR_PTR(-ENOMEM);
230
231 kref_init(&con->kref);
232 INIT_LIST_HEAD(&con->outqueue);
233 spin_lock_init(&con->outqueue_lock);
234 INIT_WORK(&con->swork, tipc_send_work);
235 INIT_WORK(&con->rwork, tipc_recv_work);
236
237 spin_lock_bh(&s->idr_lock);
238 ret = idr_alloc(&s->conn_idr, con, 0, 0, GFP_ATOMIC);
239 if (ret < 0) {
240 kfree(con);
241 spin_unlock_bh(&s->idr_lock);
242 return ERR_PTR(-ENOMEM);
243 }
244 con->conid = ret;
245 s->idr_in_use++;
246 spin_unlock_bh(&s->idr_lock);
247
248 set_bit(CF_CONNECTED, &con->flags);
249 con->server = s;
250
251 return con;
252 }
253
254 static int tipc_receive_from_sock(struct tipc_conn *con)
255 {
256 struct msghdr msg = {};
257 struct tipc_server *s = con->server;
258 struct sockaddr_tipc addr;
259 struct kvec iov;
260 void *buf;
261 int ret;
262
263 buf = kmem_cache_alloc(s->rcvbuf_cache, GFP_ATOMIC);
264 if (!buf) {
265 ret = -ENOMEM;
266 goto out_close;
267 }
268
269 iov.iov_base = buf;
270 iov.iov_len = s->max_rcvbuf_size;
271 msg.msg_name = &addr;
272 ret = kernel_recvmsg(con->sock, &msg, &iov, 1, iov.iov_len,
273 MSG_DONTWAIT);
274 if (ret <= 0) {
275 kmem_cache_free(s->rcvbuf_cache, buf);
276 goto out_close;
277 }
278
279 s->tipc_conn_recvmsg(sock_net(con->sock->sk), con->conid, &addr,
280 con->usr_data, buf, ret);
281
282 kmem_cache_free(s->rcvbuf_cache, buf);
283
284 return 0;
285
286 out_close:
287 if (ret != -EWOULDBLOCK)
288 tipc_close_conn(con);
289 else if (ret == 0)
290 /* Don't return success if we really got EOF */
291 ret = -EAGAIN;
292
293 return ret;
294 }
295
296 static int tipc_accept_from_sock(struct tipc_conn *con)
297 {
298 struct tipc_server *s = con->server;
299 struct socket *sock = con->sock;
300 struct socket *newsock;
301 struct tipc_conn *newcon;
302 int ret;
303
304 ret = kernel_accept(sock, &newsock, O_NONBLOCK);
305 if (ret < 0)
306 return ret;
307
308 newcon = tipc_alloc_conn(con->server);
309 if (IS_ERR(newcon)) {
310 ret = PTR_ERR(newcon);
311 sock_release(newsock);
312 return ret;
313 }
314
315 newcon->rx_action = tipc_receive_from_sock;
316 tipc_register_callbacks(newsock, newcon);
317
318 /* Notify that new connection is incoming */
319 newcon->usr_data = s->tipc_conn_new(newcon->conid);
320 if (!newcon->usr_data) {
321 sock_release(newsock);
322 return -ENOMEM;
323 }
324
325 /* Wake up receive process in case of 'SYN+' message */
326 newsock->sk->sk_data_ready(newsock->sk);
327 return ret;
328 }
329
330 static struct socket *tipc_create_listen_sock(struct tipc_conn *con)
331 {
332 struct tipc_server *s = con->server;
333 struct socket *sock = NULL;
334 int ret;
335
336 ret = sock_create_kern(s->net, AF_TIPC, SOCK_SEQPACKET, 0, &sock);
337 if (ret < 0)
338 return NULL;
339 ret = kernel_setsockopt(sock, SOL_TIPC, TIPC_IMPORTANCE,
340 (char *)&s->imp, sizeof(s->imp));
341 if (ret < 0)
342 goto create_err;
343 ret = kernel_bind(sock, (struct sockaddr *)s->saddr, sizeof(*s->saddr));
344 if (ret < 0)
345 goto create_err;
346
347 switch (s->type) {
348 case SOCK_STREAM:
349 case SOCK_SEQPACKET:
350 con->rx_action = tipc_accept_from_sock;
351
352 ret = kernel_listen(sock, 0);
353 if (ret < 0)
354 goto create_err;
355 break;
356 case SOCK_DGRAM:
357 case SOCK_RDM:
358 con->rx_action = tipc_receive_from_sock;
359 break;
360 default:
361 pr_err("Unknown socket type %d\n", s->type);
362 goto create_err;
363 }
364
365 /* As server's listening socket owner and creator is the same module,
366 * we have to decrease TIPC module reference count to guarantee that
367 * it remains zero after the server socket is created, otherwise,
368 * executing "rmmod" command is unable to make TIPC module deleted
369 * after TIPC module is inserted successfully.
370 *
371 * However, the reference count is ever increased twice in
372 * sock_create_kern(): one is to increase the reference count of owner
373 * of TIPC socket's proto_ops struct; another is to increment the
374 * reference count of owner of TIPC proto struct. Therefore, we must
375 * decrement the module reference count twice to ensure that it keeps
376 * zero after server's listening socket is created. Of course, we
377 * must bump the module reference count twice as well before the socket
378 * is closed.
379 */
380 module_put(sock->ops->owner);
381 module_put(sock->sk->sk_prot_creator->owner);
382 set_bit(CF_SERVER, &con->flags);
383
384 return sock;
385
386 create_err:
387 kernel_sock_shutdown(sock, SHUT_RDWR);
388 sock_release(sock);
389 return NULL;
390 }
391
392 static int tipc_open_listening_sock(struct tipc_server *s)
393 {
394 struct socket *sock;
395 struct tipc_conn *con;
396
397 con = tipc_alloc_conn(s);
398 if (IS_ERR(con))
399 return PTR_ERR(con);
400
401 sock = tipc_create_listen_sock(con);
402 if (!sock) {
403 idr_remove(&s->conn_idr, con->conid);
404 s->idr_in_use--;
405 kfree(con);
406 return -EINVAL;
407 }
408
409 tipc_register_callbacks(sock, con);
410 return 0;
411 }
412
413 static struct outqueue_entry *tipc_alloc_entry(void *data, int len)
414 {
415 struct outqueue_entry *entry;
416 void *buf;
417
418 entry = kmalloc(sizeof(struct outqueue_entry), GFP_ATOMIC);
419 if (!entry)
420 return NULL;
421
422 buf = kmemdup(data, len, GFP_ATOMIC);
423 if (!buf) {
424 kfree(entry);
425 return NULL;
426 }
427
428 entry->iov.iov_base = buf;
429 entry->iov.iov_len = len;
430
431 return entry;
432 }
433
434 static void tipc_free_entry(struct outqueue_entry *e)
435 {
436 kfree(e->iov.iov_base);
437 kfree(e);
438 }
439
440 static void tipc_clean_outqueues(struct tipc_conn *con)
441 {
442 struct outqueue_entry *e, *safe;
443
444 spin_lock_bh(&con->outqueue_lock);
445 list_for_each_entry_safe(e, safe, &con->outqueue, list) {
446 list_del(&e->list);
447 tipc_free_entry(e);
448 }
449 spin_unlock_bh(&con->outqueue_lock);
450 }
451
452 int tipc_conn_sendmsg(struct tipc_server *s, int conid,
453 struct sockaddr_tipc *addr, void *data, size_t len)
454 {
455 struct outqueue_entry *e;
456 struct tipc_conn *con;
457
458 con = tipc_conn_lookup(s, conid);
459 if (!con)
460 return -EINVAL;
461
462 e = tipc_alloc_entry(data, len);
463 if (!e) {
464 conn_put(con);
465 return -ENOMEM;
466 }
467
468 if (addr)
469 memcpy(&e->dest, addr, sizeof(struct sockaddr_tipc));
470
471 spin_lock_bh(&con->outqueue_lock);
472 list_add_tail(&e->list, &con->outqueue);
473 spin_unlock_bh(&con->outqueue_lock);
474
475 if (test_bit(CF_CONNECTED, &con->flags)) {
476 if (!queue_work(s->send_wq, &con->swork))
477 conn_put(con);
478 } else {
479 conn_put(con);
480 }
481 return 0;
482 }
483
484 void tipc_conn_terminate(struct tipc_server *s, int conid)
485 {
486 struct tipc_conn *con;
487
488 con = tipc_conn_lookup(s, conid);
489 if (con) {
490 tipc_close_conn(con);
491 conn_put(con);
492 }
493 }
494
495 static void tipc_send_to_sock(struct tipc_conn *con)
496 {
497 int count = 0;
498 struct tipc_server *s = con->server;
499 struct outqueue_entry *e;
500 struct msghdr msg;
501 int ret;
502
503 spin_lock_bh(&con->outqueue_lock);
504 while (1) {
505 e = list_entry(con->outqueue.next, struct outqueue_entry,
506 list);
507 if ((struct list_head *) e == &con->outqueue)
508 break;
509 spin_unlock_bh(&con->outqueue_lock);
510
511 memset(&msg, 0, sizeof(msg));
512 msg.msg_flags = MSG_DONTWAIT;
513
514 if (s->type == SOCK_DGRAM || s->type == SOCK_RDM) {
515 msg.msg_name = &e->dest;
516 msg.msg_namelen = sizeof(struct sockaddr_tipc);
517 }
518 ret = kernel_sendmsg(con->sock, &msg, &e->iov, 1,
519 e->iov.iov_len);
520 if (ret == -EWOULDBLOCK || ret == 0) {
521 cond_resched();
522 goto out;
523 } else if (ret < 0) {
524 goto send_err;
525 }
526
527 /* Don't starve users filling buffers */
528 if (++count >= MAX_SEND_MSG_COUNT) {
529 cond_resched();
530 count = 0;
531 }
532
533 spin_lock_bh(&con->outqueue_lock);
534 list_del(&e->list);
535 tipc_free_entry(e);
536 }
537 spin_unlock_bh(&con->outqueue_lock);
538 out:
539 return;
540
541 send_err:
542 tipc_close_conn(con);
543 }
544
545 static void tipc_recv_work(struct work_struct *work)
546 {
547 struct tipc_conn *con = container_of(work, struct tipc_conn, rwork);
548 int count = 0;
549
550 while (test_bit(CF_CONNECTED, &con->flags)) {
551 if (con->rx_action(con))
552 break;
553
554 /* Don't flood Rx machine */
555 if (++count >= MAX_RECV_MSG_COUNT) {
556 cond_resched();
557 count = 0;
558 }
559 }
560 conn_put(con);
561 }
562
563 static void tipc_send_work(struct work_struct *work)
564 {
565 struct tipc_conn *con = container_of(work, struct tipc_conn, swork);
566
567 if (test_bit(CF_CONNECTED, &con->flags))
568 tipc_send_to_sock(con);
569
570 conn_put(con);
571 }
572
573 static void tipc_work_stop(struct tipc_server *s)
574 {
575 destroy_workqueue(s->rcv_wq);
576 destroy_workqueue(s->send_wq);
577 }
578
579 static int tipc_work_start(struct tipc_server *s)
580 {
581 s->rcv_wq = alloc_ordered_workqueue("tipc_rcv", 0);
582 if (!s->rcv_wq) {
583 pr_err("can't start tipc receive workqueue\n");
584 return -ENOMEM;
585 }
586
587 s->send_wq = alloc_ordered_workqueue("tipc_send", 0);
588 if (!s->send_wq) {
589 pr_err("can't start tipc send workqueue\n");
590 destroy_workqueue(s->rcv_wq);
591 return -ENOMEM;
592 }
593
594 return 0;
595 }
596
597 int tipc_server_start(struct tipc_server *s)
598 {
599 int ret;
600
601 spin_lock_init(&s->idr_lock);
602 idr_init(&s->conn_idr);
603 s->idr_in_use = 0;
604
605 s->rcvbuf_cache = kmem_cache_create(s->name, s->max_rcvbuf_size,
606 0, SLAB_HWCACHE_ALIGN, NULL);
607 if (!s->rcvbuf_cache)
608 return -ENOMEM;
609
610 ret = tipc_work_start(s);
611 if (ret < 0) {
612 kmem_cache_destroy(s->rcvbuf_cache);
613 return ret;
614 }
615 ret = tipc_open_listening_sock(s);
616 if (ret < 0) {
617 tipc_work_stop(s);
618 kmem_cache_destroy(s->rcvbuf_cache);
619 return ret;
620 }
621 return ret;
622 }
623
624 void tipc_server_stop(struct tipc_server *s)
625 {
626 struct tipc_conn *con;
627 int total = 0;
628 int id;
629
630 spin_lock_bh(&s->idr_lock);
631 for (id = 0; total < s->idr_in_use; id++) {
632 con = idr_find(&s->conn_idr, id);
633 if (con) {
634 total++;
635 spin_unlock_bh(&s->idr_lock);
636 tipc_close_conn(con);
637 spin_lock_bh(&s->idr_lock);
638 }
639 }
640 spin_unlock_bh(&s->idr_lock);
641
642 tipc_work_stop(s);
643 kmem_cache_destroy(s->rcvbuf_cache);
644 idr_destroy(&s->conn_idr);
645 }