c9d28b1bf16365fa4c86853fc4b74a54a9da0bc4
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_server_t *ssl;
85 /* listener configuration */
86 int sock_fd;
87 int (*setup)(sdb_conn_t *, void *);
88 } listener_t;
90 typedef struct {
91 int type;
92 const char *prefix;
94 int (*open)(listener_t *);
95 void (*close)(listener_t *);
96 } fe_listener_impl_t;
98 struct sdb_fe_socket {
99 listener_t *listeners;
100 size_t listeners_num;
102 /* list of open, idle connections; active connections will be passed on to
103 * the connection handler threads which place them back after handling
104 * pending actions; the trigger pipe is used to notify the main thread
105 * after adding a connection (back) to the list */
106 sdb_llist_t *open_connections;
107 int trigger[2];
108 #define TRIGGER_READ 0
109 #define TRIGGER_WRITE 1
111 /* channel used for communication between main
112 * and connection handler threads */
113 sdb_channel_t *chan;
114 };
116 /*
117 * SSL helper functions
118 */
120 static ssize_t
121 ssl_read(sdb_conn_t *conn, size_t n)
122 {
123 char buf[n];
124 ssize_t ret;
126 ret = sdb_ssl_session_read(conn->ssl_session, buf, n);
127 if (ret <= 0)
128 return ret;
130 sdb_strbuf_memappend(conn->buf, buf, ret);
131 return ret;
132 } /* ssl_read */
134 static ssize_t
135 ssl_write(sdb_conn_t *conn, const void *buf, size_t n)
136 {
137 return sdb_ssl_session_write(conn->ssl_session, buf, n);
138 } /* ssl_write */
140 /*
141 * connection management functions
142 */
144 static int
145 setup_unixsock(sdb_conn_t *conn, void __attribute__((unused)) *user_data)
146 {
147 uid_t uid;
149 struct passwd pw_entry;
150 struct passwd *result = NULL;
151 char buf[1024];
153 #ifdef SO_PEERCRED
154 struct ucred cred;
155 socklen_t len = sizeof(cred);
157 if (getsockopt(conn->fd, SOL_SOCKET, SO_PEERCRED, &cred, &len)
158 || (len != sizeof(cred))) {
159 char errbuf[1024];
160 sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for "
161 "connection conn#%i: %s", conn->fd,
162 sdb_strerror(errno, errbuf, sizeof(errbuf)));
163 return -1;
164 }
165 uid = cred.uid;
166 #else /* SO_PEERCRED */
167 sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for "
168 "connection conn#%i: operation not supported", conn->fd);
169 return -1;
170 #endif
172 memset(&pw_entry, 0, sizeof(pw_entry));
173 if (getpwuid_r(uid, &pw_entry, buf, sizeof(buf), &result) || (! result)
174 || (! (conn->username = strdup(result->pw_name)))) {
175 char errbuf[1024];
176 sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for "
177 "connection conn#%i: %s", conn->fd,
178 sdb_strerror(errno, errbuf, sizeof(errbuf)));
179 return -1;
180 }
181 return 0;
182 } /* setup_unixsock */
184 static int
185 open_unixsock(listener_t *listener)
186 {
187 char *addr_copy;
188 char *base_dir;
189 struct sockaddr_un sa;
190 int status;
192 listener->sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
193 if (listener->sock_fd < 0) {
194 char buf[1024];
195 sdb_log(SDB_LOG_ERR, "frontend: Failed to open UNIX socket %s: %s",
196 listener->address, sdb_strerror(errno, buf, sizeof(buf)));
197 return -1;
198 }
200 memset(&sa, 0, sizeof(sa));
201 sa.sun_family = AF_UNIX;
202 strncpy(sa.sun_path, listener->address, sizeof(sa.sun_path));
204 addr_copy = strdup(listener->address);
205 if (! addr_copy) {
206 char errbuf[1024];
207 sdb_log(SDB_LOG_ERR, "frontend: strdup failed: %s",
208 sdb_strerror(errno, errbuf, sizeof(errbuf)));
209 return -1;
210 }
211 base_dir = dirname(addr_copy);
213 /* ensure that the directory exists */
214 if (sdb_mkdir_all(base_dir, 0777)) {
215 char errbuf[1024];
216 sdb_log(SDB_LOG_ERR, "frontend: Failed to create directory '%s': %s",
217 base_dir, sdb_strerror(errno, errbuf, sizeof(errbuf)));
218 free(addr_copy);
219 return -1;
220 }
221 free(addr_copy);
223 if (unlink(listener->address) && (errno != ENOENT)) {
224 char errbuf[1024];
225 sdb_log(SDB_LOG_WARNING, "frontend: Failed to remove stale UNIX "
226 "socket %s: %s", listener->address,
227 sdb_strerror(errno, errbuf, sizeof(errbuf)));
228 }
230 status = bind(listener->sock_fd, (struct sockaddr *)&sa, sizeof(sa));
231 if (status) {
232 char buf[1024];
233 sdb_log(SDB_LOG_ERR, "frontend: Failed to bind to UNIX socket %s: %s",
234 listener->address, sdb_strerror(errno, buf, sizeof(buf)));
235 return -1;
236 }
238 listener->setup = setup_unixsock;
239 return 0;
240 } /* open_unixsock */
242 static void
243 close_unixsock(listener_t *listener)
244 {
245 assert(listener);
247 if (! listener->address)
248 return;
250 if (listener->sock_fd >= 0)
251 close(listener->sock_fd);
252 listener->sock_fd = -1;
254 unlink(listener->address);
255 } /* close_unixsock */
257 static int
258 finish_tcp(sdb_conn_t *conn)
259 {
260 if (! conn->ssl_session)
261 return 0;
263 sdb_ssl_session_destroy(conn->ssl_session);
264 conn->ssl_session = NULL;
265 return 0;
266 } /* finish_tcp */
268 static int
269 setup_tcp(sdb_conn_t *conn, void *user_data)
270 {
271 listener_t *listener = user_data;
273 conn->ssl_session = sdb_ssl_server_accept(listener->ssl, conn->fd);
274 if (! conn->ssl_session)
275 return -1;
277 conn->username = sdb_ssl_session_peer(conn->ssl_session);
279 conn->finish = finish_tcp;
280 conn->read = ssl_read;
281 conn->write = ssl_write;
282 return 0;
283 } /* setup_tcp */
285 static int
286 open_tcp(listener_t *listener)
287 {
288 struct addrinfo *ai, *ai_list = NULL;
289 int status;
291 assert(listener);
293 /* TODO: make options configurable */
294 listener->ssl = sdb_ssl_server_create(NULL);
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 return 0;
397 } /* listener_listen */
399 static void
400 listener_close(listener_t *listener)
401 {
402 assert(listener);
404 if (listener_impls[listener->type].close)
405 listener_impls[listener->type].close(listener);
407 if (listener->sock_fd >= 0)
408 close(listener->sock_fd);
409 listener->sock_fd = -1;
410 } /* listener_close */
412 static int
413 get_type(const char *address)
414 {
415 char *sep;
416 size_t len;
417 size_t i;
419 if (*address == '/')
420 return LISTENER_UNIXSOCK;
421 sep = strchr(address, (int)':');
422 if (! sep)
423 return listener_impls[0].type;
425 assert(sep > address);
426 len = (size_t)(sep - address);
428 for (i = 0; i < SDB_STATIC_ARRAY_LEN(listener_impls); ++i) {
429 fe_listener_impl_t *impl = listener_impls + i;
431 if (!strncmp(address, impl->prefix, len)) {
432 assert(impl->type == (int)i);
433 return impl->type;
434 }
435 }
436 return -1;
437 } /* get_type */
439 static void
440 listener_destroy(listener_t *listener)
441 {
442 if (! listener)
443 return;
445 listener_close(listener);
447 if (listener->address)
448 free(listener->address);
449 listener->address = NULL;
450 } /* listener_destroy */
452 static listener_t *
453 listener_create(sdb_fe_socket_t *sock, const char *address)
454 {
455 listener_t *listener;
456 size_t len;
457 int type;
459 type = get_type(address);
460 if (type < 0) {
461 sdb_log(SDB_LOG_ERR, "frontend: Unsupported address type specified "
462 "in listen address '%s'", address);
463 return NULL;
464 }
466 listener = realloc(sock->listeners,
467 (sock->listeners_num + 1) * sizeof(*sock->listeners));
468 if (! listener) {
469 char buf[1024];
470 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s",
471 sdb_strerror(errno, buf, sizeof(buf)));
472 return NULL;
473 }
475 sock->listeners = listener;
476 listener = sock->listeners + sock->listeners_num;
478 len = strlen(listener_impls[type].prefix);
479 if ((! strncmp(address, listener_impls[type].prefix, len))
480 && (address[len] == ':'))
481 address += strlen(listener_impls[type].prefix) + 1;
483 listener->sock_fd = -1;
484 listener->address = strdup(address);
485 if (! listener->address) {
486 char buf[1024];
487 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s",
488 sdb_strerror(errno, buf, sizeof(buf)));
489 listener_destroy(listener);
490 return NULL;
491 }
492 listener->type = type;
493 listener->setup = NULL;
494 listener->ssl = NULL;
496 if (listener_impls[type].open(listener)) {
497 /* prints error */
498 listener_destroy(listener);
499 return NULL;
500 }
502 ++sock->listeners_num;
503 return listener;
504 } /* listener_create */
506 static void
507 socket_clear(sdb_fe_socket_t *sock)
508 {
509 size_t i;
511 assert(sock);
512 for (i = 0; i < sock->listeners_num; ++i)
513 listener_destroy(sock->listeners + i);
514 if (sock->listeners)
515 free(sock->listeners);
516 sock->listeners = NULL;
517 sock->listeners_num = 0;
518 } /* socket_clear */
520 static void
521 socket_close(sdb_fe_socket_t *sock)
522 {
523 size_t i;
525 assert(sock);
526 for (i = 0; i < sock->listeners_num; ++i)
527 listener_close(sock->listeners + i);
528 } /* socket_close */
530 /*
531 * connection handler functions
532 */
534 static void *
535 connection_handler(void *data)
536 {
537 sdb_fe_socket_t *sock = data;
539 assert(sock);
541 while (42) {
542 struct timespec timeout = { 0, 500000000 }; /* .5 seconds */
543 sdb_conn_t *conn;
544 int status;
546 errno = 0;
547 status = sdb_channel_select(sock->chan, /* read */ NULL, &conn,
548 /* write */ NULL, NULL, &timeout);
549 if (status) {
550 char buf[1024];
552 if (errno == ETIMEDOUT)
553 continue;
554 if (errno == EBADF) /* channel shut down */
555 break;
557 sdb_log(SDB_LOG_ERR, "frontend: Failed to read from channel: %s",
558 sdb_strerror(errno, buf, sizeof(buf)));
559 continue;
560 }
562 status = (int)sdb_connection_handle(conn);
563 if (status <= 0) {
564 /* error or EOF -> close connection */
565 sdb_object_deref(SDB_OBJ(conn));
566 continue;
567 }
569 /* return the connection to the main loop */
570 if (sdb_llist_append(sock->open_connections, SDB_OBJ(conn))) {
571 sdb_log(SDB_LOG_ERR, "frontend: Failed to re-append "
572 "connection %s to list of open connections",
573 SDB_OBJ(conn)->name);
574 }
575 write(sock->trigger[TRIGGER_WRITE], "", 1);
577 /* pass ownership back to list; or destroy in case of an error */
578 sdb_object_deref(SDB_OBJ(conn));
579 }
580 return NULL;
581 } /* connection_handler */
583 static int
584 connection_accept(sdb_fe_socket_t *sock, listener_t *listener)
585 {
586 sdb_object_t *obj;
587 int status;
589 obj = SDB_OBJ(sdb_connection_accept(listener->sock_fd,
590 listener->setup, listener));
591 if (! obj)
592 return -1;
594 status = sdb_llist_append(sock->open_connections, obj);
595 if (status)
596 sdb_log(SDB_LOG_ERR, "frontend: Failed to append "
597 "connection %s to list of open connections",
598 obj->name);
600 /* hand ownership over to the list; or destroy in case of an error */
601 sdb_object_deref(obj);
602 return status;
603 } /* connection_accept */
605 static int
606 socket_handle_incoming(sdb_fe_socket_t *sock,
607 fd_set *ready, fd_set *exceptions)
608 {
609 sdb_llist_iter_t *iter;
610 size_t i;
612 for (i = 0; i < sock->listeners_num; ++i) {
613 listener_t *listener = sock->listeners + i;
614 if (FD_ISSET(listener->sock_fd, ready))
615 if (connection_accept(sock, listener))
616 continue;
617 }
619 iter = sdb_llist_get_iter(sock->open_connections);
620 if (! iter) {
621 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
622 "for open connections");
623 return -1;
624 }
626 while (sdb_llist_iter_has_next(iter)) {
627 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
629 if (FD_ISSET(CONN(obj)->fd, exceptions)) {
630 sdb_log(SDB_LOG_INFO, "Exception on fd %d",
631 CONN(obj)->fd);
632 /* close the connection */
633 sdb_llist_iter_remove_current(iter);
634 sdb_object_deref(obj);
635 continue;
636 }
638 if (FD_ISSET(CONN(obj)->fd, ready)) {
639 sdb_llist_iter_remove_current(iter);
640 sdb_channel_write(sock->chan, &obj);
641 }
642 }
643 sdb_llist_iter_destroy(iter);
644 return 0;
645 } /* socket_handle_incoming */
647 /*
648 * public API
649 */
651 sdb_fe_socket_t *
652 sdb_fe_sock_create(void)
653 {
654 sdb_fe_socket_t *sock;
655 int flags;
657 sock = calloc(1, sizeof(*sock));
658 if (! sock)
659 return NULL;
660 sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
662 sock->open_connections = sdb_llist_create();
663 if (! sock->open_connections) {
664 sdb_fe_sock_destroy(sock);
665 return NULL;
666 }
668 if (pipe(sock->trigger)) {
669 char errbuf[1024];
670 sdb_log(SDB_LOG_ERR, "frontend: Failed to create pipe: %s",
671 sdb_strerror(errno, errbuf, sizeof(errbuf)));
672 sdb_fe_sock_destroy(sock);
673 return NULL;
674 }
675 /* TODO: Can we do a zero-byte write as well to trigger select()?
676 * Linux does not seem to support the I_SWROPT ioctl on a pipe. */
677 flags = fcntl(sock->trigger[TRIGGER_WRITE], F_GETFL);
678 if (fcntl(sock->trigger[TRIGGER_WRITE], F_SETFL, flags | O_NONBLOCK)) {
679 char errbuf[1024];
680 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking "
681 "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
682 sdb_fe_sock_destroy(sock);
683 return NULL;
684 }
685 flags = fcntl(sock->trigger[TRIGGER_READ], F_GETFL);
686 if (fcntl(sock->trigger[TRIGGER_READ], F_SETFL, flags | O_NONBLOCK)) {
687 char errbuf[1024];
688 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking "
689 "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf)));
690 sdb_fe_sock_destroy(sock);
691 return NULL;
692 }
693 return sock;
694 } /* sdb_fe_sock_create */
696 void
697 sdb_fe_sock_destroy(sdb_fe_socket_t *sock)
698 {
699 if (! sock)
700 return;
702 socket_clear(sock);
704 if (sock->trigger[TRIGGER_WRITE] >= 0)
705 close(sock->trigger[TRIGGER_WRITE]);
706 if (sock->trigger[TRIGGER_READ] >= 0)
707 close(sock->trigger[TRIGGER_READ]);
708 sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
710 sdb_llist_destroy(sock->open_connections);
711 sock->open_connections = NULL;
712 free(sock);
713 } /* sdb_fe_sock_destroy */
715 int
716 sdb_fe_sock_add_listener(sdb_fe_socket_t *sock, const char *address)
717 {
718 listener_t *listener;
720 if ((! sock) || (! address))
721 return -1;
723 listener = listener_create(sock, address);
724 if (! listener)
725 return -1;
726 return 0;
727 } /* sdb_fe_sock_add_listener */
729 void
730 sdb_fe_sock_clear_listeners(sdb_fe_socket_t *sock)
731 {
732 if (! sock)
733 return;
735 socket_clear(sock);
736 } /* sdb_fe_sock_clear_listeners */
738 int
739 sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop)
740 {
741 fd_set sockets;
742 int max_listen_fd = 0;
743 size_t i;
745 pthread_t handler_threads[loop->num_threads];
746 size_t num_threads;
748 if ((! sock) || (! sock->listeners_num) || sock->chan
749 || (! loop) || (loop->num_threads <= 0))
750 return -1;
752 if (! loop->do_loop)
753 return 0;
755 FD_ZERO(&sockets);
756 for (i = 0; i < sock->listeners_num; ++i) {
757 listener_t *listener = sock->listeners + i;
759 if (listener_listen(listener)) {
760 socket_close(sock);
761 return -1;
762 }
764 FD_SET(listener->sock_fd, &sockets);
765 if (listener->sock_fd > max_listen_fd)
766 max_listen_fd = listener->sock_fd;
767 }
769 sock->chan = sdb_channel_create(1024, sizeof(sdb_conn_t *));
770 if (! sock->chan) {
771 socket_close(sock);
772 return -1;
773 }
775 sdb_log(SDB_LOG_INFO, "frontend: Starting %zu connection "
776 "handler thread%s managing %zu listener%s",
777 loop->num_threads, loop->num_threads == 1 ? "" : "s",
778 sock->listeners_num, sock->listeners_num == 1 ? "" : "s");
780 num_threads = loop->num_threads;
781 memset(&handler_threads, 0, sizeof(handler_threads));
782 for (i = 0; i < num_threads; ++i) {
783 errno = 0;
784 if (pthread_create(&handler_threads[i], /* attr = */ NULL,
785 connection_handler, /* arg = */ sock)) {
786 char errbuf[1024];
787 sdb_log(SDB_LOG_ERR, "frontend: Failed to create "
788 "connection handler thread: %s",
789 sdb_strerror(errno, errbuf, sizeof(errbuf)));
790 num_threads = i;
791 break;
792 }
793 }
795 while (loop->do_loop && num_threads) {
796 struct timeval timeout = { 1, 0 }; /* one second */
797 sdb_llist_iter_t *iter;
799 int max_fd = max_listen_fd;
800 fd_set ready;
801 fd_set exceptions;
802 int n;
804 FD_ZERO(&ready);
805 FD_ZERO(&exceptions);
807 ready = sockets;
808 FD_SET(sock->trigger[TRIGGER_READ], &ready);
810 iter = sdb_llist_get_iter(sock->open_connections);
811 if (! iter) {
812 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
813 "for open connections");
814 break;
815 }
817 while (sdb_llist_iter_has_next(iter)) {
818 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
820 if (CONN(obj)->fd < 0) {
821 sdb_llist_iter_remove_current(iter);
822 sdb_object_deref(obj);
823 continue;
824 }
826 FD_SET(CONN(obj)->fd, &ready);
827 FD_SET(CONN(obj)->fd, &exceptions);
829 if (CONN(obj)->fd > max_fd)
830 max_fd = CONN(obj)->fd;
831 }
832 sdb_llist_iter_destroy(iter);
834 errno = 0;
835 n = select(max_fd + 1, &ready, NULL, &exceptions, &timeout);
836 if (n < 0) {
837 char buf[1024];
839 if (errno == EINTR)
840 continue;
842 sdb_log(SDB_LOG_ERR, "frontend: Failed to monitor sockets: %s",
843 sdb_strerror(errno, buf, sizeof(buf)));
844 break;
845 }
846 else if (! n)
847 continue;
849 if (FD_ISSET(sock->trigger[TRIGGER_READ], &ready)) {
850 char buf[1024];
851 while (read(sock->trigger[TRIGGER_READ], buf, sizeof(buf)) > 0)
852 /* do nothing */;
853 }
855 /* handle new and open connections */
856 if (socket_handle_incoming(sock, &ready, &exceptions))
857 break;
858 }
860 socket_close(sock);
862 sdb_log(SDB_LOG_INFO, "frontend: Waiting for connection handler threads "
863 "to terminate");
864 if (! sdb_channel_shutdown(sock->chan))
865 for (i = 0; i < num_threads; ++i)
866 pthread_join(handler_threads[i], NULL);
867 /* else: we tried our best; let the operating system clean up */
869 sdb_channel_destroy(sock->chan);
870 sock->chan = NULL;
872 if (! num_threads)
873 return -1;
874 return 0;
875 } /* sdb_fe_sock_listen_and_server */
877 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */