1 /*
2 * SysDB - src/frontend/sock.c
3 * Copyright (C) 2013 Sebastian 'tokkee' Harl <sh@tokkee.org>
4 * All rights reserved.
5 *
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions
8 * are met:
9 * 1. Redistributions of source code must retain the above copyright
10 * notice, this list of conditions and the following disclaimer.
11 * 2. Redistributions in binary form must reproduce the above copyright
12 * notice, this list of conditions and the following disclaimer in the
13 * documentation and/or other materials provided with the distribution.
14 *
15 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
16 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
17 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
18 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR
19 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
20 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
21 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
22 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
23 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
24 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
25 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
26 */
28 #if HAVE_CONFIG_H
29 # include "config.h"
30 #endif /* HAVE_CONFIG_H */
32 #include "sysdb.h"
33 #include "core/object.h"
34 #include "frontend/connection-private.h"
35 #include "frontend/sock.h"
37 #include "utils/channel.h"
38 #include "utils/error.h"
39 #include "utils/llist.h"
40 #include "utils/os.h"
41 #include "utils/ssl.h"
42 #include "utils/strbuf.h"
44 #include <assert.h>
45 #include <errno.h>
47 #include <stdio.h>
48 #include <stdlib.h>
49 #include <string.h>
51 #include <unistd.h>
53 #include <sys/time.h>
54 #include <sys/types.h>
55 #include <sys/select.h>
56 #include <sys/socket.h>
57 #include <sys/param.h>
58 #include <sys/un.h>
60 #ifdef HAVE_UCRED_H
61 # include <ucred.h>
62 #endif
63 #ifdef HAVE_SYS_UCRED_H
64 # include <sys/ucred.h>
65 #endif
67 #include <pwd.h>
69 #include <fcntl.h>
70 #include <netdb.h>
71 #include <libgen.h>
72 #include <pthread.h>
74 /*
75 * private data types
76 */
78 typedef struct {
79 char *address;
80 int type;
82 /* optional SSL settings */
83 sdb_ssl_options_t ssl_opts;
84 sdb_ssl_server_t *ssl;
86 /* listener configuration */
87 int sock_fd;
88 int (*setup)(sdb_conn_t *, void *);
89 } listener_t;
91 typedef struct {
92 int type;
93 const char *prefix;
95 int (*open)(listener_t *);
96 void (*close)(listener_t *);
97 } fe_listener_impl_t;
99 struct sdb_fe_socket {
100 listener_t *listeners;
101 size_t listeners_num;
103 /* list of open, idle connections; active connections will be passed on to
104 * the connection handler threads which place them back after handling
105 * pending actions; the trigger pipe is used to notify the main thread
106 * after adding a connection (back) to the list */
107 sdb_llist_t *open_connections;
108 int trigger[2];
109 #define TRIGGER_READ 0
110 #define TRIGGER_WRITE 1
112 /* channel used for communication between main
113 * and connection handler threads */
114 sdb_channel_t *chan;
115 };
117 /*
118 * SSL helper functions
119 */
121 static ssize_t
122 ssl_read(sdb_conn_t *conn, size_t n)
123 {
124 char buf[n];
125 ssize_t ret;
127 ret = sdb_ssl_session_read(conn->ssl_session, buf, n);
128 if (ret <= 0)
129 return ret;
131 sdb_strbuf_memappend(conn->buf, buf, ret);
132 return ret;
133 } /* ssl_read */
135 static ssize_t
136 ssl_write(sdb_conn_t *conn, const void *buf, size_t n)
137 {
138 return sdb_ssl_session_write(conn->ssl_session, buf, n);
139 } /* ssl_write */
141 /*
142 * connection management functions
143 */
145 static int
146 setup_unixsock(sdb_conn_t *conn, void __attribute__((unused)) *user_data)
147 {
148 uid_t uid;
150 struct passwd pw_entry;
151 struct passwd *result = NULL;
152 char buf[1024];
154 #ifdef SO_PEERCRED
155 struct ucred cred;
156 socklen_t len = sizeof(cred);
158 if (getsockopt(conn->fd, SOL_SOCKET, SO_PEERCRED, &cred, &len)
159 || (len != sizeof(cred))) {
160 char errbuf[1024];
161 sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for "
162 "connection conn#%i: %s", conn->fd,
163 sdb_strerror(errno, errbuf, sizeof(errbuf)));
164 return -1;
165 }
166 uid = cred.uid;
167 #else /* SO_PEERCRED */
168 sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for "
169 "connection conn#%i: operation not supported", conn->fd);
170 return -1;
171 #endif
173 memset(&pw_entry, 0, sizeof(pw_entry));
174 if (getpwuid_r(uid, &pw_entry, buf, sizeof(buf), &result) || (! result)
175 || (! (conn->username = strdup(result->pw_name)))) {
176 char errbuf[1024];
177 sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for "
178 "connection conn#%i: %s", conn->fd,
179 sdb_strerror(errno, errbuf, sizeof(errbuf)));
180 return -1;
181 }
182 return 0;
183 } /* setup_unixsock */
185 static int
186 open_unixsock(listener_t *listener)
187 {
188 char *addr_copy;
189 char *base_dir;
190 struct sockaddr_un sa;
191 int status;
193 listener->sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
194 if (listener->sock_fd < 0) {
195 char buf[1024];
196 sdb_log(SDB_LOG_ERR, "frontend: Failed to open UNIX socket %s: %s",
197 listener->address, sdb_strerror(errno, buf, sizeof(buf)));
198 return -1;
199 }
201 memset(&sa, 0, sizeof(sa));
202 sa.sun_family = AF_UNIX;
203 strncpy(sa.sun_path, listener->address, sizeof(sa.sun_path));
205 addr_copy = strdup(listener->address);
206 if (! addr_copy) {
207 char errbuf[1024];
208 sdb_log(SDB_LOG_ERR, "frontend: strdup failed: %s",
209 sdb_strerror(errno, errbuf, sizeof(errbuf)));
210 return -1;
211 }
212 base_dir = dirname(addr_copy);
214 /* ensure that the directory exists */
215 if (sdb_mkdir_all(base_dir, 0777)) {
216 char errbuf[1024];
217 sdb_log(SDB_LOG_ERR, "frontend: Failed to create directory '%s': %s",
218 base_dir, sdb_strerror(errno, errbuf, sizeof(errbuf)));
219 free(addr_copy);
220 return -1;
221 }
222 free(addr_copy);
224 if (unlink(listener->address) && (errno != ENOENT)) {
225 char errbuf[1024];
226 sdb_log(SDB_LOG_WARNING, "frontend: Failed to remove stale UNIX "
227 "socket %s: %s", listener->address,
228 sdb_strerror(errno, errbuf, sizeof(errbuf)));
229 }
231 status = bind(listener->sock_fd, (struct sockaddr *)&sa, sizeof(sa));
232 if (status) {
233 char buf[1024];
234 sdb_log(SDB_LOG_ERR, "frontend: Failed to bind to UNIX socket %s: %s",
235 listener->address, sdb_strerror(errno, buf, sizeof(buf)));
236 return -1;
237 }
239 listener->setup = setup_unixsock;
240 return 0;
241 } /* open_unixsock */
243 static void
244 close_unixsock(listener_t *listener)
245 {
246 assert(listener);
248 if (! listener->address)
249 return;
251 if (listener->sock_fd >= 0)
252 close(listener->sock_fd);
253 listener->sock_fd = -1;
255 unlink(listener->address);
256 } /* close_unixsock */
258 static int
259 finish_tcp(sdb_conn_t *conn)
260 {
261 if (! conn->ssl_session)
262 return 0;
264 sdb_ssl_session_destroy(conn->ssl_session);
265 conn->ssl_session = NULL;
266 return 0;
267 } /* finish_tcp */
269 static int
270 setup_tcp(sdb_conn_t *conn, void *user_data)
271 {
272 listener_t *listener = user_data;
274 conn->ssl_session = sdb_ssl_server_accept(listener->ssl, conn->fd);
275 if (! conn->ssl_session)
276 return -1;
278 conn->username = sdb_ssl_session_peer(conn->ssl_session);
280 conn->finish = finish_tcp;
281 conn->read = ssl_read;
282 conn->write = ssl_write;
283 return 0;
284 } /* setup_tcp */
286 static int
287 open_tcp(listener_t *listener)
288 {
289 struct addrinfo *ai, *ai_list = NULL;
290 int status;
292 assert(listener);
294 listener->ssl = sdb_ssl_server_create(&listener->ssl_opts);
295 if (! listener->ssl)
296 return -1;
298 if ((status = sdb_resolve(SDB_NET_TCP, listener->address, &ai_list))) {
299 sdb_log(SDB_LOG_ERR, "frontend: Failed to resolve '%s': %s",
300 listener->address, gai_strerror(status));
301 return -1;
302 }
304 for (ai = ai_list; ai != NULL; ai = ai->ai_next) {
305 char errbuf[1024];
306 int reuse = 1;
308 listener->sock_fd = socket(ai->ai_family,
309 ai->ai_socktype, ai->ai_protocol);
310 if (listener->sock_fd < 0) {
311 sdb_log(SDB_LOG_ERR, "frontend: Failed to open socket for %s: %s",
312 listener->address,
313 sdb_strerror(errno, errbuf, sizeof(errbuf)));
314 continue;
315 }
317 if (setsockopt(listener->sock_fd, SOL_SOCKET, SO_REUSEADDR,
318 &reuse, sizeof(reuse)) < 0) {
319 sdb_log(SDB_LOG_ERR, "frontend: Failed to set socket option: %s",
320 sdb_strerror(errno, errbuf, sizeof(errbuf)));
321 close(listener->sock_fd);
322 listener->sock_fd = -1;
323 continue;
324 }
326 if (bind(listener->sock_fd, ai->ai_addr, ai->ai_addrlen) < 0) {
327 char host[1024], port[32];
328 getnameinfo(ai->ai_addr, ai->ai_addrlen, host, sizeof(host),
329 port, sizeof(port), NI_NUMERICHOST | NI_NUMERICSERV);
330 sdb_log(SDB_LOG_ERR, "frontend: Failed to bind to %s:%s: %s",
331 host, port, sdb_strerror(errno, errbuf, sizeof(errbuf)));
332 close(listener->sock_fd);
333 listener->sock_fd = -1;
334 continue;
335 }
336 break;
337 }
338 freeaddrinfo(ai_list);
340 if (listener->sock_fd < 0)
341 return -1;
343 listener->setup = setup_tcp;
344 return 0;
345 } /* open_tcp */
347 static void
348 close_tcp(listener_t *listener)
349 {
350 assert(listener);
352 sdb_ssl_server_destroy(listener->ssl);
353 listener->ssl = NULL;
355 if (listener->sock_fd >= 0)
356 close(listener->sock_fd);
357 listener->sock_fd = -1;
358 } /* close_tcp */
360 /*
361 * private variables
362 */
364 /* the enum has to be sorted the same as the implementations array
365 * to ensure that the type may be used as index into the array */
366 enum {
367 LISTENER_TCP = 0, /* this is the default */
368 LISTENER_UNIXSOCK,
369 };
370 static fe_listener_impl_t listener_impls[] = {
371 { LISTENER_TCP, "tcp", open_tcp, close_tcp },
372 { LISTENER_UNIXSOCK, "unix", open_unixsock, close_unixsock },
373 };
375 /*
376 * private helper functions
377 */
379 static int
380 listener_listen(listener_t *listener)
381 {
382 assert(listener);
384 /* try to reopen */
385 if (listener->sock_fd < 0)
386 if (listener_impls[listener->type].open(listener))
387 return -1;
388 assert(listener->sock_fd >= 0);
390 if (listen(listener->sock_fd, /* backlog = */ 32)) {
391 char buf[1024];
392 sdb_log(SDB_LOG_ERR, "frontend: Failed to listen on socket %s: %s",
393 listener->address, sdb_strerror(errno, buf, sizeof(buf)));
394 return -1;
395 }
396 sdb_log(SDB_LOG_INFO, "frontend: Listening on %s", listener->address);
397 return 0;
398 } /* listener_listen */
400 static void
401 listener_close(listener_t *listener)
402 {
403 assert(listener);
405 if (listener_impls[listener->type].close)
406 listener_impls[listener->type].close(listener);
408 if (listener->sock_fd >= 0)
409 close(listener->sock_fd);
410 listener->sock_fd = -1;
411 } /* listener_close */
413 static int
414 get_type(const char *address)
415 {
416 char *sep;
417 size_t len;
418 size_t i;
420 if (*address == '/')
421 return LISTENER_UNIXSOCK;
422 sep = strchr(address, (int)':');
423 if (! sep)
424 return listener_impls[0].type;
426 assert(sep > address);
427 len = (size_t)(sep - address);
429 for (i = 0; i < SDB_STATIC_ARRAY_LEN(listener_impls); ++i) {
430 fe_listener_impl_t *impl = listener_impls + i;
432 if (!strncmp(address, impl->prefix, len)) {
433 assert(impl->type == (int)i);
434 return impl->type;
435 }
436 }
437 /* don't report an error, this could be an IPv6 address */
438 return listener_impls[0].type;
439 } /* get_type */
441 static void
442 listener_destroy(listener_t *listener)
443 {
444 if (! listener)
445 return;
447 listener_close(listener);
448 sdb_ssl_free_options(&listener->ssl_opts);
450 if (listener->address)
451 free(listener->address);
452 listener->address = NULL;
453 } /* listener_destroy */
455 static listener_t *
456 listener_create(sdb_fe_socket_t *sock, const char *address)
457 {
458 listener_t *listener;
459 size_t len;
460 int type;
462 type = get_type(address);
463 if (type < 0) {
464 sdb_log(SDB_LOG_ERR, "frontend: Unsupported address type specified "
465 "in listen address '%s'", address);
466 return NULL;
467 }
469 listener = realloc(sock->listeners,
470 (sock->listeners_num + 1) * sizeof(*sock->listeners));
471 if (! listener) {
472 char buf[1024];
473 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s",
474 sdb_strerror(errno, buf, sizeof(buf)));
475 return NULL;
476 }
478 sock->listeners = listener;
479 listener = sock->listeners + sock->listeners_num;
481 len = strlen(listener_impls[type].prefix);
482 if ((! strncmp(address, listener_impls[type].prefix, len))
483 && (address[len] == ':'))
484 address += strlen(listener_impls[type].prefix) + 1;
485 memset(listener, 0, sizeof(*listener));
487 listener->sock_fd = -1;
488 listener->address = strdup(address);
489 if (! listener->address) {
490 char buf[1024];
491 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s",
492 sdb_strerror(errno, buf, sizeof(buf)));
493 listener_destroy(listener);
494 return NULL;
495 }
496 listener->type = type;
497 listener->setup = NULL;
498 listener->ssl = NULL;
500 ++sock->listeners_num;
501 return listener;
502 } /* listener_create */
504 static void
505 socket_clear(sdb_fe_socket_t *sock)
506 {
507 size_t i;
509 assert(sock);
510 for (i = 0; i < sock->listeners_num; ++i)
511 listener_destroy(sock->listeners + i);
512 if (sock->listeners)
513 free(sock->listeners);
514 sock->listeners = NULL;
515 sock->listeners_num = 0;
516 } /* socket_clear */
518 static void
519 socket_close(sdb_fe_socket_t *sock)
520 {
521 size_t i;
523 assert(sock);
524 for (i = 0; i < sock->listeners_num; ++i)
525 listener_close(sock->listeners + i);
526 } /* socket_close */
528 /*
529 * connection handler functions
530 */
532 static void *
533 connection_handler(void *data)
534 {
535 sdb_fe_socket_t *sock = data;
537 assert(sock);
539 while (42) {
540 struct timespec timeout = { 0, 500000000 }; /* .5 seconds */
541 sdb_conn_t *conn;
542 int status;
544 errno = 0;
545 status = sdb_channel_select(sock->chan, /* read */ NULL, &conn,
546 /* write */ NULL, NULL, &timeout);
547 if (status) {
548 char buf[1024];
550 if (errno == ETIMEDOUT)
551 continue;
552 if (errno == EBADF) /* channel shut down */
553 break;
555 sdb_log(SDB_LOG_ERR, "frontend: Failed to read from channel: %s",
556 sdb_strerror(errno, buf, sizeof(buf)));
557 continue;
558 }
560 status = (int)sdb_connection_handle(conn);
561 if (status <= 0) {
562 /* error or EOF -> close connection */
563 sdb_object_deref(SDB_OBJ(conn));
564 continue;
565 }
567 /* return the connection to the main loop */
568 if (sdb_llist_append(sock->open_connections, SDB_OBJ(conn))) {
569 sdb_log(SDB_LOG_ERR, "frontend: Failed to re-append "
570 "connection %s to list of open connections",
571 SDB_OBJ(conn)->name);
572 }
573 if (write(sock->trigger[TRIGGER_WRITE], "", 1) <= 0) {
574 /* This shouldn't happen and it's not critical; in the worst cases
575 * it slows us down. */
576 sdb_log(SDB_LOG_WARNING, "frontend: Failed to trigger main loop");
577 }
579 /* pass ownership back to list; or destroy in case of an error */
580 sdb_object_deref(SDB_OBJ(conn));
581 }
582 return NULL;
583 } /* connection_handler */
585 static int
586 connection_accept(sdb_fe_socket_t *sock, listener_t *listener)
587 {
588 sdb_object_t *obj;
589 int status;
591 obj = SDB_OBJ(sdb_connection_accept(listener->sock_fd,
592 listener->setup, listener));
593 if (! obj)
594 return -1;
596 status = sdb_llist_append(sock->open_connections, obj);
597 if (status)
598 sdb_log(SDB_LOG_ERR, "frontend: Failed to append "
599 "connection %s to list of open connections",
600 obj->name);
602 /* hand ownership over to the list; or destroy in case of an error */
603 sdb_object_deref(obj);
604 return status;
605 } /* connection_accept */
607 static int
608 socket_handle_incoming(sdb_fe_socket_t *sock,
609 fd_set *ready, fd_set *exceptions)
610 {
611 sdb_llist_iter_t *iter;
612 size_t i;
614 for (i = 0; i < sock->listeners_num; ++i) {
615 listener_t *listener = sock->listeners + i;
616 if (FD_ISSET(listener->sock_fd, ready))
617 if (connection_accept(sock, listener))
618 continue;
619 }
621 iter = sdb_llist_get_iter(sock->open_connections);
622 if (! iter) {
623 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
624 "for open connections");
625 return -1;
626 }
628 while (sdb_llist_iter_has_next(iter)) {
629 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
631 if (FD_ISSET(CONN(obj)->fd, exceptions)) {
632 sdb_log(SDB_LOG_INFO, "Exception on fd %d",
633 CONN(obj)->fd);
634 /* close the connection */
635 sdb_llist_iter_remove_current(iter);
636 sdb_object_deref(obj);
637 continue;
638 }
640 if (FD_ISSET(CONN(obj)->fd, ready)) {
641 sdb_llist_iter_remove_current(iter);
642 sdb_channel_write(sock->chan, &obj);
643 }
644 }
645 sdb_llist_iter_destroy(iter);
646 return 0;
647 } /* socket_handle_incoming */
649 /*
650 * public API
651 */
653 sdb_fe_socket_t *
654 sdb_fe_sock_create(void)
655 {
656 sdb_fe_socket_t *sock;
657 int flags;
659 sock = calloc(1, sizeof(*sock));
660 if (! sock)
661 return NULL;
662 sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
664 sock->open_connections = sdb_llist_create();
665 if (! sock->open_connections) {
666 sdb_fe_sock_destroy(sock);
667 return NULL;
668 }
670 if (pipe(sock->trigger)) {
671 char errbuf[1024];
672 sdb_log(SDB_LOG_ERR, "frontend: Failed to create pipe: %s",
673 sdb_strerror(errno, errbuf, sizeof(errbuf)));
674 sdb_fe_sock_destroy(sock);
675 return NULL;
676 }
677 /* TODO: Can we do a zero-byte write as well to trigger select()?
678 * Linux does not seem to support the I_SWROPT ioctl on a pipe. */
679 flags = fcntl(sock->trigger[TRIGGER_WRITE], F_GETFL);
680 if (fcntl(sock->trigger[TRIGGER_WRITE], F_SETFL, flags | O_NONBLOCK)) {
681 char errbuf[1024];
682 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking "
683 "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
684 sdb_fe_sock_destroy(sock);
685 return NULL;
686 }
687 flags = fcntl(sock->trigger[TRIGGER_READ], F_GETFL);
688 if (fcntl(sock->trigger[TRIGGER_READ], F_SETFL, flags | O_NONBLOCK)) {
689 char errbuf[1024];
690 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking "
691 "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
692 sdb_fe_sock_destroy(sock);
693 return NULL;
694 }
695 return sock;
696 } /* sdb_fe_sock_create */
698 void
699 sdb_fe_sock_destroy(sdb_fe_socket_t *sock)
700 {
701 if (! sock)
702 return;
704 socket_clear(sock);
706 if (sock->trigger[TRIGGER_WRITE] >= 0)
707 close(sock->trigger[TRIGGER_WRITE]);
708 if (sock->trigger[TRIGGER_READ] >= 0)
709 close(sock->trigger[TRIGGER_READ]);
710 sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
712 sdb_llist_destroy(sock->open_connections);
713 sock->open_connections = NULL;
714 free(sock);
715 } /* sdb_fe_sock_destroy */
717 int
718 sdb_fe_sock_add_listener(sdb_fe_socket_t *sock, const char *address,
719 const sdb_ssl_options_t *opts)
720 {
721 listener_t *listener;
723 if ((! sock) || (! address))
724 return -1;
726 listener = listener_create(sock, address);
727 if (! listener)
728 return -1;
730 if (opts) {
731 int ret = 0;
733 if (opts->ca_file) {
734 listener->ssl_opts.ca_file = strdup(opts->ca_file);
735 if (! listener->ssl_opts.ca_file)
736 ret = -1;
737 }
738 if (opts->key_file) {
739 listener->ssl_opts.key_file = strdup(opts->key_file);
740 if (! listener->ssl_opts.key_file)
741 ret = -1;
742 }
743 if (opts->cert_file) {
744 listener->ssl_opts.cert_file = strdup(opts->cert_file);
745 if (! listener->ssl_opts.cert_file)
746 ret = -1;
747 }
748 if (opts->crl_file) {
749 listener->ssl_opts.crl_file = strdup(opts->crl_file);
750 if (! listener->ssl_opts.crl_file)
751 ret = -1;
752 }
754 if (ret) {
755 listener_destroy(listener);
756 --sock->listeners_num;
757 return ret;
758 }
759 }
761 if (listener_impls[listener->type].open(listener)) {
762 /* prints error */
763 listener_destroy(listener);
764 --sock->listeners_num;
765 return -1;
766 }
767 return 0;
768 } /* sdb_fe_sock_add_listener */
770 void
771 sdb_fe_sock_clear_listeners(sdb_fe_socket_t *sock)
772 {
773 if (! sock)
774 return;
776 socket_clear(sock);
777 } /* sdb_fe_sock_clear_listeners */
779 int
780 sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop)
781 {
782 fd_set sockets;
783 int max_listen_fd = 0;
784 size_t i;
786 pthread_t handler_threads[loop->num_threads];
787 size_t num_threads;
789 if ((! sock) || (! sock->listeners_num) || sock->chan
790 || (! loop) || (loop->num_threads <= 0))
791 return -1;
793 if (! loop->do_loop)
794 return 0;
796 FD_ZERO(&sockets);
797 for (i = 0; i < sock->listeners_num; ++i) {
798 listener_t *listener = sock->listeners + i;
800 if (listener_listen(listener)) {
801 socket_close(sock);
802 return -1;
803 }
805 FD_SET(listener->sock_fd, &sockets);
806 if (listener->sock_fd > max_listen_fd)
807 max_listen_fd = listener->sock_fd;
808 }
810 sock->chan = sdb_channel_create(1024, sizeof(sdb_conn_t *));
811 if (! sock->chan) {
812 socket_close(sock);
813 return -1;
814 }
816 sdb_log(SDB_LOG_INFO, "frontend: Starting %zu connection "
817 "handler thread%s managing %zu listener%s",
818 loop->num_threads, loop->num_threads == 1 ? "" : "s",
819 sock->listeners_num, sock->listeners_num == 1 ? "" : "s");
821 num_threads = loop->num_threads;
822 memset(&handler_threads, 0, sizeof(handler_threads));
823 for (i = 0; i < num_threads; ++i) {
824 errno = 0;
825 if (pthread_create(&handler_threads[i], /* attr = */ NULL,
826 connection_handler, /* arg = */ sock)) {
827 char errbuf[1024];
828 sdb_log(SDB_LOG_ERR, "frontend: Failed to create "
829 "connection handler thread: %s",
830 sdb_strerror(errno, errbuf, sizeof(errbuf)));
831 num_threads = i;
832 break;
833 }
834 }
836 while (loop->do_loop && num_threads) {
837 struct timeval timeout = { 1, 0 }; /* one second */
838 sdb_llist_iter_t *iter;
840 int max_fd = max_listen_fd;
841 fd_set ready;
842 fd_set exceptions;
843 int n;
845 FD_ZERO(&ready);
846 FD_ZERO(&exceptions);
848 ready = sockets;
849 FD_SET(sock->trigger[TRIGGER_READ], &ready);
851 iter = sdb_llist_get_iter(sock->open_connections);
852 if (! iter) {
853 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
854 "for open connections");
855 break;
856 }
858 while (sdb_llist_iter_has_next(iter)) {
859 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
861 if (CONN(obj)->fd < 0) {
862 sdb_llist_iter_remove_current(iter);
863 sdb_object_deref(obj);
864 continue;
865 }
867 FD_SET(CONN(obj)->fd, &ready);
868 FD_SET(CONN(obj)->fd, &exceptions);
870 if (CONN(obj)->fd > max_fd)
871 max_fd = CONN(obj)->fd;
872 }
873 sdb_llist_iter_destroy(iter);
875 errno = 0;
876 n = select(max_fd + 1, &ready, NULL, &exceptions, &timeout);
877 if (n < 0) {
878 char buf[1024];
880 if (errno == EINTR)
881 continue;
883 sdb_log(SDB_LOG_ERR, "frontend: Failed to monitor sockets: %s",
884 sdb_strerror(errno, buf, sizeof(buf)));
885 break;
886 }
887 else if (! n)
888 continue;
890 if (FD_ISSET(sock->trigger[TRIGGER_READ], &ready)) {
891 char buf[1024];
892 while (read(sock->trigger[TRIGGER_READ], buf, sizeof(buf)) > 0)
893 /* do nothing */;
894 }
896 /* handle new and open connections */
897 if (socket_handle_incoming(sock, &ready, &exceptions))
898 break;
899 }
901 socket_close(sock);
903 sdb_log(SDB_LOG_INFO, "frontend: Waiting for connection handler threads "
904 "to terminate");
905 if (! sdb_channel_shutdown(sock->chan))
906 for (i = 0; i < num_threads; ++i)
907 pthread_join(handler_threads[i], NULL);
908 /* else: we tried our best; let the operating system clean up */
910 sdb_channel_destroy(sock->chan);
911 sock->chan = NULL;
913 if (! num_threads)
914 return -1;
915 return 0;
916 } /* sdb_fe_sock_listen_and_server */
918 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */