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 return -1;
438 } /* get_type */
440 static void
441 listener_destroy(listener_t *listener)
442 {
443 if (! listener)
444 return;
446 listener_close(listener);
447 sdb_ssl_free_options(&listener->ssl_opts);
449 if (listener->address)
450 free(listener->address);
451 listener->address = NULL;
452 } /* listener_destroy */
454 static listener_t *
455 listener_create(sdb_fe_socket_t *sock, const char *address)
456 {
457 listener_t *listener;
458 size_t len;
459 int type;
461 type = get_type(address);
462 if (type < 0) {
463 sdb_log(SDB_LOG_ERR, "frontend: Unsupported address type specified "
464 "in listen address '%s'", address);
465 return NULL;
466 }
468 listener = realloc(sock->listeners,
469 (sock->listeners_num + 1) * sizeof(*sock->listeners));
470 if (! listener) {
471 char buf[1024];
472 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s",
473 sdb_strerror(errno, buf, sizeof(buf)));
474 return NULL;
475 }
477 sock->listeners = listener;
478 listener = sock->listeners + sock->listeners_num;
480 len = strlen(listener_impls[type].prefix);
481 if ((! strncmp(address, listener_impls[type].prefix, len))
482 && (address[len] == ':'))
483 address += strlen(listener_impls[type].prefix) + 1;
484 memset(listener, 0, sizeof(*listener));
486 listener->sock_fd = -1;
487 listener->address = strdup(address);
488 if (! listener->address) {
489 char buf[1024];
490 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s",
491 sdb_strerror(errno, buf, sizeof(buf)));
492 listener_destroy(listener);
493 return NULL;
494 }
495 listener->type = type;
496 listener->setup = NULL;
497 listener->ssl = NULL;
499 ++sock->listeners_num;
500 return listener;
501 } /* listener_create */
503 static void
504 socket_clear(sdb_fe_socket_t *sock)
505 {
506 size_t i;
508 assert(sock);
509 for (i = 0; i < sock->listeners_num; ++i)
510 listener_destroy(sock->listeners + i);
511 if (sock->listeners)
512 free(sock->listeners);
513 sock->listeners = NULL;
514 sock->listeners_num = 0;
515 } /* socket_clear */
517 static void
518 socket_close(sdb_fe_socket_t *sock)
519 {
520 size_t i;
522 assert(sock);
523 for (i = 0; i < sock->listeners_num; ++i)
524 listener_close(sock->listeners + i);
525 } /* socket_close */
527 /*
528 * connection handler functions
529 */
531 static void *
532 connection_handler(void *data)
533 {
534 sdb_fe_socket_t *sock = data;
536 assert(sock);
538 while (42) {
539 struct timespec timeout = { 0, 500000000 }; /* .5 seconds */
540 sdb_conn_t *conn;
541 int status;
543 errno = 0;
544 status = sdb_channel_select(sock->chan, /* read */ NULL, &conn,
545 /* write */ NULL, NULL, &timeout);
546 if (status) {
547 char buf[1024];
549 if (errno == ETIMEDOUT)
550 continue;
551 if (errno == EBADF) /* channel shut down */
552 break;
554 sdb_log(SDB_LOG_ERR, "frontend: Failed to read from channel: %s",
555 sdb_strerror(errno, buf, sizeof(buf)));
556 continue;
557 }
559 status = (int)sdb_connection_handle(conn);
560 if (status <= 0) {
561 /* error or EOF -> close connection */
562 sdb_object_deref(SDB_OBJ(conn));
563 continue;
564 }
566 /* return the connection to the main loop */
567 if (sdb_llist_append(sock->open_connections, SDB_OBJ(conn))) {
568 sdb_log(SDB_LOG_ERR, "frontend: Failed to re-append "
569 "connection %s to list of open connections",
570 SDB_OBJ(conn)->name);
571 }
572 write(sock->trigger[TRIGGER_WRITE], "", 1);
574 /* pass ownership back to list; or destroy in case of an error */
575 sdb_object_deref(SDB_OBJ(conn));
576 }
577 return NULL;
578 } /* connection_handler */
580 static int
581 connection_accept(sdb_fe_socket_t *sock, listener_t *listener)
582 {
583 sdb_object_t *obj;
584 int status;
586 obj = SDB_OBJ(sdb_connection_accept(listener->sock_fd,
587 listener->setup, listener));
588 if (! obj)
589 return -1;
591 status = sdb_llist_append(sock->open_connections, obj);
592 if (status)
593 sdb_log(SDB_LOG_ERR, "frontend: Failed to append "
594 "connection %s to list of open connections",
595 obj->name);
597 /* hand ownership over to the list; or destroy in case of an error */
598 sdb_object_deref(obj);
599 return status;
600 } /* connection_accept */
602 static int
603 socket_handle_incoming(sdb_fe_socket_t *sock,
604 fd_set *ready, fd_set *exceptions)
605 {
606 sdb_llist_iter_t *iter;
607 size_t i;
609 for (i = 0; i < sock->listeners_num; ++i) {
610 listener_t *listener = sock->listeners + i;
611 if (FD_ISSET(listener->sock_fd, ready))
612 if (connection_accept(sock, listener))
613 continue;
614 }
616 iter = sdb_llist_get_iter(sock->open_connections);
617 if (! iter) {
618 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
619 "for open connections");
620 return -1;
621 }
623 while (sdb_llist_iter_has_next(iter)) {
624 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
626 if (FD_ISSET(CONN(obj)->fd, exceptions)) {
627 sdb_log(SDB_LOG_INFO, "Exception on fd %d",
628 CONN(obj)->fd);
629 /* close the connection */
630 sdb_llist_iter_remove_current(iter);
631 sdb_object_deref(obj);
632 continue;
633 }
635 if (FD_ISSET(CONN(obj)->fd, ready)) {
636 sdb_llist_iter_remove_current(iter);
637 sdb_channel_write(sock->chan, &obj);
638 }
639 }
640 sdb_llist_iter_destroy(iter);
641 return 0;
642 } /* socket_handle_incoming */
644 /*
645 * public API
646 */
648 sdb_fe_socket_t *
649 sdb_fe_sock_create(void)
650 {
651 sdb_fe_socket_t *sock;
652 int flags;
654 sock = calloc(1, sizeof(*sock));
655 if (! sock)
656 return NULL;
657 sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
659 sock->open_connections = sdb_llist_create();
660 if (! sock->open_connections) {
661 sdb_fe_sock_destroy(sock);
662 return NULL;
663 }
665 if (pipe(sock->trigger)) {
666 char errbuf[1024];
667 sdb_log(SDB_LOG_ERR, "frontend: Failed to create pipe: %s",
668 sdb_strerror(errno, errbuf, sizeof(errbuf)));
669 sdb_fe_sock_destroy(sock);
670 return NULL;
671 }
672 /* TODO: Can we do a zero-byte write as well to trigger select()?
673 * Linux does not seem to support the I_SWROPT ioctl on a pipe. */
674 flags = fcntl(sock->trigger[TRIGGER_WRITE], F_GETFL);
675 if (fcntl(sock->trigger[TRIGGER_WRITE], F_SETFL, flags | O_NONBLOCK)) {
676 char errbuf[1024];
677 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking "
678 "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
679 sdb_fe_sock_destroy(sock);
680 return NULL;
681 }
682 flags = fcntl(sock->trigger[TRIGGER_READ], F_GETFL);
683 if (fcntl(sock->trigger[TRIGGER_READ], F_SETFL, flags | O_NONBLOCK)) {
684 char errbuf[1024];
685 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking "
686 "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
687 sdb_fe_sock_destroy(sock);
688 return NULL;
689 }
690 return sock;
691 } /* sdb_fe_sock_create */
693 void
694 sdb_fe_sock_destroy(sdb_fe_socket_t *sock)
695 {
696 if (! sock)
697 return;
699 socket_clear(sock);
701 if (sock->trigger[TRIGGER_WRITE] >= 0)
702 close(sock->trigger[TRIGGER_WRITE]);
703 if (sock->trigger[TRIGGER_READ] >= 0)
704 close(sock->trigger[TRIGGER_READ]);
705 sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
707 sdb_llist_destroy(sock->open_connections);
708 sock->open_connections = NULL;
709 free(sock);
710 } /* sdb_fe_sock_destroy */
712 int
713 sdb_fe_sock_add_listener(sdb_fe_socket_t *sock, const char *address,
714 const sdb_ssl_options_t *opts)
715 {
716 listener_t *listener;
718 if ((! sock) || (! address))
719 return -1;
721 listener = listener_create(sock, address);
722 if (! listener)
723 return -1;
725 if (opts) {
726 int ret = 0;
728 if (opts->ca_file) {
729 listener->ssl_opts.ca_file = strdup(opts->ca_file);
730 if (! listener->ssl_opts.ca_file)
731 ret = -1;
732 }
733 if (opts->key_file) {
734 listener->ssl_opts.key_file = strdup(opts->key_file);
735 if (! listener->ssl_opts.key_file)
736 ret = -1;
737 }
738 if (opts->cert_file) {
739 listener->ssl_opts.cert_file = strdup(opts->cert_file);
740 if (! listener->ssl_opts.cert_file)
741 ret = -1;
742 }
743 if (opts->crl_file) {
744 listener->ssl_opts.crl_file = strdup(opts->crl_file);
745 if (! listener->ssl_opts.crl_file)
746 ret = -1;
747 }
749 if (ret) {
750 listener_destroy(listener);
751 --sock->listeners_num;
752 return ret;
753 }
754 }
756 if (listener_impls[listener->type].open(listener)) {
757 /* prints error */
758 listener_destroy(listener);
759 --sock->listeners_num;
760 return -1;
761 }
762 return 0;
763 } /* sdb_fe_sock_add_listener */
765 void
766 sdb_fe_sock_clear_listeners(sdb_fe_socket_t *sock)
767 {
768 if (! sock)
769 return;
771 socket_clear(sock);
772 } /* sdb_fe_sock_clear_listeners */
774 int
775 sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop)
776 {
777 fd_set sockets;
778 int max_listen_fd = 0;
779 size_t i;
781 pthread_t handler_threads[loop->num_threads];
782 size_t num_threads;
784 if ((! sock) || (! sock->listeners_num) || sock->chan
785 || (! loop) || (loop->num_threads <= 0))
786 return -1;
788 if (! loop->do_loop)
789 return 0;
791 FD_ZERO(&sockets);
792 for (i = 0; i < sock->listeners_num; ++i) {
793 listener_t *listener = sock->listeners + i;
795 if (listener_listen(listener)) {
796 socket_close(sock);
797 return -1;
798 }
800 FD_SET(listener->sock_fd, &sockets);
801 if (listener->sock_fd > max_listen_fd)
802 max_listen_fd = listener->sock_fd;
803 }
805 sock->chan = sdb_channel_create(1024, sizeof(sdb_conn_t *));
806 if (! sock->chan) {
807 socket_close(sock);
808 return -1;
809 }
811 sdb_log(SDB_LOG_INFO, "frontend: Starting %zu connection "
812 "handler thread%s managing %zu listener%s",
813 loop->num_threads, loop->num_threads == 1 ? "" : "s",
814 sock->listeners_num, sock->listeners_num == 1 ? "" : "s");
816 num_threads = loop->num_threads;
817 memset(&handler_threads, 0, sizeof(handler_threads));
818 for (i = 0; i < num_threads; ++i) {
819 errno = 0;
820 if (pthread_create(&handler_threads[i], /* attr = */ NULL,
821 connection_handler, /* arg = */ sock)) {
822 char errbuf[1024];
823 sdb_log(SDB_LOG_ERR, "frontend: Failed to create "
824 "connection handler thread: %s",
825 sdb_strerror(errno, errbuf, sizeof(errbuf)));
826 num_threads = i;
827 break;
828 }
829 }
831 while (loop->do_loop && num_threads) {
832 struct timeval timeout = { 1, 0 }; /* one second */
833 sdb_llist_iter_t *iter;
835 int max_fd = max_listen_fd;
836 fd_set ready;
837 fd_set exceptions;
838 int n;
840 FD_ZERO(&ready);
841 FD_ZERO(&exceptions);
843 ready = sockets;
844 FD_SET(sock->trigger[TRIGGER_READ], &ready);
846 iter = sdb_llist_get_iter(sock->open_connections);
847 if (! iter) {
848 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
849 "for open connections");
850 break;
851 }
853 while (sdb_llist_iter_has_next(iter)) {
854 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
856 if (CONN(obj)->fd < 0) {
857 sdb_llist_iter_remove_current(iter);
858 sdb_object_deref(obj);
859 continue;
860 }
862 FD_SET(CONN(obj)->fd, &ready);
863 FD_SET(CONN(obj)->fd, &exceptions);
865 if (CONN(obj)->fd > max_fd)
866 max_fd = CONN(obj)->fd;
867 }
868 sdb_llist_iter_destroy(iter);
870 errno = 0;
871 n = select(max_fd + 1, &ready, NULL, &exceptions, &timeout);
872 if (n < 0) {
873 char buf[1024];
875 if (errno == EINTR)
876 continue;
878 sdb_log(SDB_LOG_ERR, "frontend: Failed to monitor sockets: %s",
879 sdb_strerror(errno, buf, sizeof(buf)));
880 break;
881 }
882 else if (! n)
883 continue;
885 if (FD_ISSET(sock->trigger[TRIGGER_READ], &ready)) {
886 char buf[1024];
887 while (read(sock->trigger[TRIGGER_READ], buf, sizeof(buf)) > 0)
888 /* do nothing */;
889 }
891 /* handle new and open connections */
892 if (socket_handle_incoming(sock, &ready, &exceptions))
893 break;
894 }
896 socket_close(sock);
898 sdb_log(SDB_LOG_INFO, "frontend: Waiting for connection handler threads "
899 "to terminate");
900 if (! sdb_channel_shutdown(sock->chan))
901 for (i = 0; i < num_threads; ++i)
902 pthread_join(handler_threads[i], NULL);
903 /* else: we tried our best; let the operating system clean up */
905 sdb_channel_destroy(sock->chan);
906 sock->chan = NULL;
908 if (! num_threads)
909 return -1;
910 return 0;
911 } /* sdb_fe_sock_listen_and_server */
913 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */