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 write(sock->trigger[TRIGGER_WRITE], "", 1);
575 /* pass ownership back to list; or destroy in case of an error */
576 sdb_object_deref(SDB_OBJ(conn));
577 }
578 return NULL;
579 } /* connection_handler */
581 static int
582 connection_accept(sdb_fe_socket_t *sock, listener_t *listener)
583 {
584 sdb_object_t *obj;
585 int status;
587 obj = SDB_OBJ(sdb_connection_accept(listener->sock_fd,
588 listener->setup, listener));
589 if (! obj)
590 return -1;
592 status = sdb_llist_append(sock->open_connections, obj);
593 if (status)
594 sdb_log(SDB_LOG_ERR, "frontend: Failed to append "
595 "connection %s to list of open connections",
596 obj->name);
598 /* hand ownership over to the list; or destroy in case of an error */
599 sdb_object_deref(obj);
600 return status;
601 } /* connection_accept */
603 static int
604 socket_handle_incoming(sdb_fe_socket_t *sock,
605 fd_set *ready, fd_set *exceptions)
606 {
607 sdb_llist_iter_t *iter;
608 size_t i;
610 for (i = 0; i < sock->listeners_num; ++i) {
611 listener_t *listener = sock->listeners + i;
612 if (FD_ISSET(listener->sock_fd, ready))
613 if (connection_accept(sock, listener))
614 continue;
615 }
617 iter = sdb_llist_get_iter(sock->open_connections);
618 if (! iter) {
619 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
620 "for open connections");
621 return -1;
622 }
624 while (sdb_llist_iter_has_next(iter)) {
625 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
627 if (FD_ISSET(CONN(obj)->fd, exceptions)) {
628 sdb_log(SDB_LOG_INFO, "Exception on fd %d",
629 CONN(obj)->fd);
630 /* close the connection */
631 sdb_llist_iter_remove_current(iter);
632 sdb_object_deref(obj);
633 continue;
634 }
636 if (FD_ISSET(CONN(obj)->fd, ready)) {
637 sdb_llist_iter_remove_current(iter);
638 sdb_channel_write(sock->chan, &obj);
639 }
640 }
641 sdb_llist_iter_destroy(iter);
642 return 0;
643 } /* socket_handle_incoming */
645 /*
646 * public API
647 */
649 sdb_fe_socket_t *
650 sdb_fe_sock_create(void)
651 {
652 sdb_fe_socket_t *sock;
653 int flags;
655 sock = calloc(1, sizeof(*sock));
656 if (! sock)
657 return NULL;
658 sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
660 sock->open_connections = sdb_llist_create();
661 if (! sock->open_connections) {
662 sdb_fe_sock_destroy(sock);
663 return NULL;
664 }
666 if (pipe(sock->trigger)) {
667 char errbuf[1024];
668 sdb_log(SDB_LOG_ERR, "frontend: Failed to create pipe: %s",
669 sdb_strerror(errno, errbuf, sizeof(errbuf)));
670 sdb_fe_sock_destroy(sock);
671 return NULL;
672 }
673 /* TODO: Can we do a zero-byte write as well to trigger select()?
674 * Linux does not seem to support the I_SWROPT ioctl on a pipe. */
675 flags = fcntl(sock->trigger[TRIGGER_WRITE], F_GETFL);
676 if (fcntl(sock->trigger[TRIGGER_WRITE], F_SETFL, flags | O_NONBLOCK)) {
677 char errbuf[1024];
678 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking "
679 "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
680 sdb_fe_sock_destroy(sock);
681 return NULL;
682 }
683 flags = fcntl(sock->trigger[TRIGGER_READ], F_GETFL);
684 if (fcntl(sock->trigger[TRIGGER_READ], F_SETFL, flags | O_NONBLOCK)) {
685 char errbuf[1024];
686 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking "
687 "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
688 sdb_fe_sock_destroy(sock);
689 return NULL;
690 }
691 return sock;
692 } /* sdb_fe_sock_create */
694 void
695 sdb_fe_sock_destroy(sdb_fe_socket_t *sock)
696 {
697 if (! sock)
698 return;
700 socket_clear(sock);
702 if (sock->trigger[TRIGGER_WRITE] >= 0)
703 close(sock->trigger[TRIGGER_WRITE]);
704 if (sock->trigger[TRIGGER_READ] >= 0)
705 close(sock->trigger[TRIGGER_READ]);
706 sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
708 sdb_llist_destroy(sock->open_connections);
709 sock->open_connections = NULL;
710 free(sock);
711 } /* sdb_fe_sock_destroy */
713 int
714 sdb_fe_sock_add_listener(sdb_fe_socket_t *sock, const char *address,
715 const sdb_ssl_options_t *opts)
716 {
717 listener_t *listener;
719 if ((! sock) || (! address))
720 return -1;
722 listener = listener_create(sock, address);
723 if (! listener)
724 return -1;
726 if (opts) {
727 int ret = 0;
729 if (opts->ca_file) {
730 listener->ssl_opts.ca_file = strdup(opts->ca_file);
731 if (! listener->ssl_opts.ca_file)
732 ret = -1;
733 }
734 if (opts->key_file) {
735 listener->ssl_opts.key_file = strdup(opts->key_file);
736 if (! listener->ssl_opts.key_file)
737 ret = -1;
738 }
739 if (opts->cert_file) {
740 listener->ssl_opts.cert_file = strdup(opts->cert_file);
741 if (! listener->ssl_opts.cert_file)
742 ret = -1;
743 }
744 if (opts->crl_file) {
745 listener->ssl_opts.crl_file = strdup(opts->crl_file);
746 if (! listener->ssl_opts.crl_file)
747 ret = -1;
748 }
750 if (ret) {
751 listener_destroy(listener);
752 --sock->listeners_num;
753 return ret;
754 }
755 }
757 if (listener_impls[listener->type].open(listener)) {
758 /* prints error */
759 listener_destroy(listener);
760 --sock->listeners_num;
761 return -1;
762 }
763 return 0;
764 } /* sdb_fe_sock_add_listener */
766 void
767 sdb_fe_sock_clear_listeners(sdb_fe_socket_t *sock)
768 {
769 if (! sock)
770 return;
772 socket_clear(sock);
773 } /* sdb_fe_sock_clear_listeners */
775 int
776 sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop)
777 {
778 fd_set sockets;
779 int max_listen_fd = 0;
780 size_t i;
782 pthread_t handler_threads[loop->num_threads];
783 size_t num_threads;
785 if ((! sock) || (! sock->listeners_num) || sock->chan
786 || (! loop) || (loop->num_threads <= 0))
787 return -1;
789 if (! loop->do_loop)
790 return 0;
792 FD_ZERO(&sockets);
793 for (i = 0; i < sock->listeners_num; ++i) {
794 listener_t *listener = sock->listeners + i;
796 if (listener_listen(listener)) {
797 socket_close(sock);
798 return -1;
799 }
801 FD_SET(listener->sock_fd, &sockets);
802 if (listener->sock_fd > max_listen_fd)
803 max_listen_fd = listener->sock_fd;
804 }
806 sock->chan = sdb_channel_create(1024, sizeof(sdb_conn_t *));
807 if (! sock->chan) {
808 socket_close(sock);
809 return -1;
810 }
812 sdb_log(SDB_LOG_INFO, "frontend: Starting %zu connection "
813 "handler thread%s managing %zu listener%s",
814 loop->num_threads, loop->num_threads == 1 ? "" : "s",
815 sock->listeners_num, sock->listeners_num == 1 ? "" : "s");
817 num_threads = loop->num_threads;
818 memset(&handler_threads, 0, sizeof(handler_threads));
819 for (i = 0; i < num_threads; ++i) {
820 errno = 0;
821 if (pthread_create(&handler_threads[i], /* attr = */ NULL,
822 connection_handler, /* arg = */ sock)) {
823 char errbuf[1024];
824 sdb_log(SDB_LOG_ERR, "frontend: Failed to create "
825 "connection handler thread: %s",
826 sdb_strerror(errno, errbuf, sizeof(errbuf)));
827 num_threads = i;
828 break;
829 }
830 }
832 while (loop->do_loop && num_threads) {
833 struct timeval timeout = { 1, 0 }; /* one second */
834 sdb_llist_iter_t *iter;
836 int max_fd = max_listen_fd;
837 fd_set ready;
838 fd_set exceptions;
839 int n;
841 FD_ZERO(&ready);
842 FD_ZERO(&exceptions);
844 ready = sockets;
845 FD_SET(sock->trigger[TRIGGER_READ], &ready);
847 iter = sdb_llist_get_iter(sock->open_connections);
848 if (! iter) {
849 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
850 "for open connections");
851 break;
852 }
854 while (sdb_llist_iter_has_next(iter)) {
855 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
857 if (CONN(obj)->fd < 0) {
858 sdb_llist_iter_remove_current(iter);
859 sdb_object_deref(obj);
860 continue;
861 }
863 FD_SET(CONN(obj)->fd, &ready);
864 FD_SET(CONN(obj)->fd, &exceptions);
866 if (CONN(obj)->fd > max_fd)
867 max_fd = CONN(obj)->fd;
868 }
869 sdb_llist_iter_destroy(iter);
871 errno = 0;
872 n = select(max_fd + 1, &ready, NULL, &exceptions, &timeout);
873 if (n < 0) {
874 char buf[1024];
876 if (errno == EINTR)
877 continue;
879 sdb_log(SDB_LOG_ERR, "frontend: Failed to monitor sockets: %s",
880 sdb_strerror(errno, buf, sizeof(buf)));
881 break;
882 }
883 else if (! n)
884 continue;
886 if (FD_ISSET(sock->trigger[TRIGGER_READ], &ready)) {
887 char buf[1024];
888 while (read(sock->trigger[TRIGGER_READ], buf, sizeof(buf)) > 0)
889 /* do nothing */;
890 }
892 /* handle new and open connections */
893 if (socket_handle_incoming(sock, &ready, &exceptions))
894 break;
895 }
897 socket_close(sock);
899 sdb_log(SDB_LOG_INFO, "frontend: Waiting for connection handler threads "
900 "to terminate");
901 if (! sdb_channel_shutdown(sock->chan))
902 for (i = 0; i < num_threads; ++i)
903 pthread_join(handler_threads[i], NULL);
904 /* else: we tried our best; let the operating system clean up */
906 sdb_channel_destroy(sock->chan);
907 sock->chan = NULL;
909 if (! num_threads)
910 return -1;
911 return 0;
912 } /* sdb_fe_sock_listen_and_server */
914 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */