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 #include "sysdb.h"
29 #include "core/error.h"
30 #include "core/object.h"
31 #include "frontend/sock.h"
33 #include "utils/channel.h"
34 #include "utils/llist.h"
35 #include "utils/strbuf.h"
37 #include <assert.h>
38 #include <errno.h>
40 #include <arpa/inet.h>
42 #include <stdio.h>
43 #include <stdlib.h>
44 #include <string.h>
46 #include <unistd.h>
48 #include <fcntl.h>
50 #include <sys/time.h>
51 #include <sys/types.h>
52 #include <sys/select.h>
53 #include <sys/socket.h>
54 #include <sys/un.h>
56 #include <pthread.h>
58 /* name of connection objects */
59 #define CONN_FD_PREFIX "conn#"
60 #define CONN_FD_PLACEHOLDER "XXXXXXX"
62 /*
63 * private data types
64 */
66 typedef struct {
67 sdb_object_t super;
69 /* connection and client information */
70 int fd;
71 struct sockaddr_storage client_addr;
72 socklen_t client_addr_len;
74 /* read buffer */
75 sdb_strbuf_t *buf;
77 /* state information for the currently executed command */
78 uint32_t cmd;
79 uint32_t cmd_len;
80 } connection_obj_t;
81 #define CONN(obj) ((connection_obj_t *)(obj))
83 typedef struct {
84 char *address;
85 int type;
87 int sock_fd;
88 } listener_t;
90 typedef struct {
91 int type;
92 const char *prefix;
94 int (*opener)(listener_t *);
95 } fe_listener_impl_t;
97 struct sdb_fe_socket {
98 listener_t *listeners;
99 size_t listeners_num;
101 sdb_llist_t *open_connections;
103 /* channel used for communication between main
104 * and connection handler threads */
105 sdb_channel_t *chan;
106 };
108 /*
109 * connection management functions
110 */
112 static int
113 open_unix_sock(listener_t *listener)
114 {
115 struct sockaddr_un sa;
116 int status;
118 listener->sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
119 if (listener->sock_fd < 0) {
120 char buf[1024];
121 sdb_log(SDB_LOG_ERR, "frontend: Failed to open UNIX socket %s: %s",
122 listener->address, sdb_strerror(errno, buf, sizeof(buf)));
123 return -1;
124 }
126 memset(&sa, 0, sizeof(sa));
127 sa.sun_family = AF_UNIX;
128 strncpy(sa.sun_path, listener->address + strlen("unix:"),
129 sizeof(sa.sun_path));
131 status = bind(listener->sock_fd, (struct sockaddr *)&sa, sizeof(sa));
132 if (status) {
133 char buf[1024];
134 sdb_log(SDB_LOG_ERR, "frontend: Failed to bind to UNIX socket %s: %s",
135 listener->address, sdb_strerror(errno, buf, sizeof(buf)));
136 return -1;
137 }
138 return 0;
139 } /* open_unix_sock */
141 /*
142 * private variables
143 */
145 /* the enum has to be sorted the same as the implementations array
146 * to ensure that the type may be used as index into the array */
147 enum {
148 LISTENER_UNIXSOCK = 0,
149 };
150 static fe_listener_impl_t listener_impls[] = {
151 { LISTENER_UNIXSOCK, "unix", open_unix_sock },
152 };
154 /*
155 * private helper functions
156 */
158 static int
159 get_type(const char *address)
160 {
161 char *sep;
162 size_t len;
163 size_t i;
165 sep = strchr(address, (int)':');
166 if (! sep)
167 return -1;
169 assert(sep > address);
170 len = (size_t)(sep - address);
172 for (i = 0; i < SDB_STATIC_ARRAY_LEN(listener_impls); ++i) {
173 fe_listener_impl_t *impl = listener_impls + i;
175 if (!strncmp(address, impl->prefix, len)) {
176 assert(impl->type == (int)i);
177 return impl->type;
178 }
179 }
180 return -1;
181 } /* get_type */
183 static void
184 listener_destroy(listener_t *listener)
185 {
186 if (! listener)
187 return;
189 if (listener->sock_fd >= 0)
190 close(listener->sock_fd);
191 listener->sock_fd = -1;
193 if (listener->address)
194 free(listener->address);
195 } /* listener_destroy */
197 static listener_t *
198 listener_create(sdb_fe_socket_t *sock, const char *address)
199 {
200 listener_t *listener;
201 int type;
203 type = get_type(address);
204 if (type < 0) {
205 sdb_log(SDB_LOG_ERR, "frontend: Unsupported address type specified "
206 "in listen address '%s'", address);
207 return NULL;
208 }
210 listener = realloc(sock->listeners,
211 sock->listeners_num * sizeof(*sock->listeners));
212 if (! listener) {
213 char buf[1024];
214 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s",
215 sdb_strerror(errno, buf, sizeof(buf)));
216 return NULL;
217 }
219 sock->listeners = listener;
220 listener = sock->listeners + sock->listeners_num;
222 listener->sock_fd = -1;
223 listener->address = strdup(address);
224 if (! listener->address) {
225 char buf[1024];
226 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s",
227 sdb_strerror(errno, buf, sizeof(buf)));
228 listener_destroy(listener);
229 return NULL;
230 }
231 listener->type = type;
233 if (listener_impls[type].opener(listener)) {
234 /* prints error */
235 listener_destroy(listener);
236 return NULL;
237 }
239 ++sock->listeners_num;
240 return listener;
241 } /* listener_create */
243 static int
244 listener_listen(listener_t *listener)
245 {
246 assert(listener);
248 /* try to reopen */
249 if (listener->sock_fd < 0)
250 if (listener_impls[listener->type].opener(listener))
251 return -1;
252 assert(listener->sock_fd >= 0);
254 if (listen(listener->sock_fd, /* backlog = */ 32)) {
255 char buf[1024];
256 sdb_log(SDB_LOG_ERR, "frontend: Failed to listen on socket %s: %s",
257 listener->address, sdb_strerror(errno, buf, sizeof(buf)));
258 return -1;
259 }
260 return 0;
261 } /* listener_listen */
263 static void
264 listener_close(listener_t *listener)
265 {
266 assert(listener);
268 if (listener->sock_fd < 0)
269 return;
271 close(listener->sock_fd);
272 listener->sock_fd = -1;
273 } /* listener_close */
275 static void
276 socket_close(sdb_fe_socket_t *sock)
277 {
278 size_t i;
280 assert(sock);
281 for (i = 0; i < sock->listeners_num; ++i)
282 listener_close(sock->listeners + i);
283 } /* socket_close */
285 /*
286 * private data types
287 */
289 static int
290 connection_init(sdb_object_t *obj, va_list ap)
291 {
292 connection_obj_t *conn;
293 int sock_fd;
294 int sock_fl;
296 assert(obj);
297 conn = CONN(obj);
299 sock_fd = va_arg(ap, int);
301 CONN(obj)->buf = sdb_strbuf_create(/* size = */ 128);
302 if (! CONN(obj)->buf) {
303 sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate a read buffer "
304 "for a new remote connection");
305 return -1;
306 }
308 conn->client_addr_len = sizeof(conn->client_addr);
309 conn->fd = accept(sock_fd, (struct sockaddr *)&conn->client_addr,
310 &conn->client_addr_len);
312 if (conn->fd < 0) {
313 char buf[1024];
314 sdb_log(SDB_LOG_ERR, "frontend: Failed to accept remote "
315 "connection: %s", sdb_strerror(errno,
316 buf, sizeof(buf)));
317 return -1;
318 }
320 if (conn->client_addr.ss_family != AF_UNIX) {
321 sdb_log(SDB_LOG_ERR, "frontend: Accepted connection using "
322 "unexpected family type %d", conn->client_addr.ss_family);
323 return -1;
324 }
326 sock_fl = fcntl(conn->fd, F_GETFL);
327 if (fcntl(conn->fd, F_SETFL, sock_fl | O_NONBLOCK)) {
328 char buf[1024];
329 sdb_log(SDB_LOG_ERR, "frontend: Failed to switch connection conn#%i "
330 "to non-blocking mode: %s", conn->fd,
331 sdb_strerror(errno, buf, sizeof(buf)));
332 return -1;
333 }
335 sdb_log(SDB_LOG_DEBUG, "frontend: Accepted connection on fd=%i",
336 conn->fd);
338 /* update the object name */
339 snprintf(obj->name + strlen(CONN_FD_PREFIX),
340 strlen(CONN_FD_PLACEHOLDER), "%i", conn->fd);
341 return 0;
342 } /* connection_init */
344 static void
345 connection_destroy(sdb_object_t *obj)
346 {
347 connection_obj_t *conn;
348 size_t len;
350 assert(obj);
351 conn = CONN(obj);
353 len = sdb_strbuf_len(conn->buf);
354 if (len)
355 sdb_log(SDB_LOG_INFO, "frontend: Discarding incomplete command "
356 "(%zu bytes left in buffer)", len);
358 sdb_log(SDB_LOG_DEBUG, "frontend: Closing connection on fd=%i", conn->fd);
359 close(conn->fd);
360 conn->fd = -1;
362 sdb_strbuf_destroy(CONN(obj)->buf);
363 } /* connection_destroy */
365 static sdb_type_t connection_type = {
366 /* size = */ sizeof(connection_obj_t),
367 /* init = */ connection_init,
368 /* destroy = */ connection_destroy,
369 };
371 /*
372 * connection handler functions
373 */
375 static uint32_t
376 connection_get_int32(connection_obj_t *conn, size_t offset)
377 {
378 const char *data;
379 uint32_t n;
381 assert(conn && (sdb_strbuf_len(conn->buf) >= offset + sizeof(uint32_t)));
383 data = sdb_strbuf_string(conn->buf);
384 memcpy(&n, data + offset, sizeof(n));
385 n = ntohl(n);
386 return n;
387 } /* connection_get_int32 */
389 static int
390 command_handle(connection_obj_t *conn)
391 {
392 assert(conn && conn->cmd && conn->cmd_len);
393 /* XXX */
394 sdb_strbuf_skip(conn->buf, conn->cmd_len);
395 return 0;
396 } /* command_handle */
398 /* initialize the connection state information */
399 static int
400 command_init(connection_obj_t *conn)
401 {
402 assert(conn && (! conn->cmd) && (! conn->cmd_len));
404 conn->cmd = connection_get_int32(conn, 0);
405 conn->cmd_len = connection_get_int32(conn, sizeof(uint32_t));
406 sdb_strbuf_skip(conn->buf, 2 * sizeof(uint32_t));
407 return 0;
408 } /* command_init */
410 /* returns negative value on error, 0 on EOF, number of octets else */
411 static ssize_t
412 connection_read(connection_obj_t *conn)
413 {
414 ssize_t n = 0;
416 while (42) {
417 ssize_t status;
419 errno = 0;
420 status = sdb_strbuf_read(conn->buf, conn->fd, 1024);
421 if (status < 0) {
422 if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
423 return n;
424 return (int)status;
425 }
426 else if (! status) /* EOF */
427 return n;
429 n += status;
430 }
432 return n;
433 } /* connection_read */
435 static void *
436 connection_handler(void *data)
437 {
438 sdb_fe_socket_t *sock = data;
440 assert(sock);
442 while (42) {
443 struct timespec timeout = { 0, 500000000 }; /* .5 seconds */
444 connection_obj_t *conn;
445 int status;
447 errno = 0;
448 status = sdb_channel_select(sock->chan, /* read */ NULL, &conn,
449 /* write */ NULL, NULL, &timeout);
450 if (status) {
451 char buf[1024];
453 if (errno == ETIMEDOUT)
454 continue;
455 if (errno == EBADF) /* channel shut down */
456 break;
458 sdb_log(SDB_LOG_ERR, "frontend: Failed to read from channel: %s",
459 sdb_strerror(errno, buf, sizeof(buf)));
460 continue;
461 }
463 status = (int)connection_read(conn);
464 if (status <= 0) {
465 /* error or EOF -> close connection */
466 sdb_object_deref(SDB_OBJ(conn));
467 continue;
468 }
470 if (conn->cmd_len && (sdb_strbuf_len(conn->buf) >= conn->cmd_len))
471 command_handle(conn);
472 else if (sdb_strbuf_len(conn->buf) >= 2 * sizeof(int32_t))
473 command_init(conn);
475 /* return the connection to the main loop */
476 if (sdb_llist_append(sock->open_connections, SDB_OBJ(conn))) {
477 sdb_log(SDB_LOG_ERR, "frontend: Failed to re-append "
478 "connection %s to list of open connections",
479 SDB_OBJ(conn)->name);
480 }
482 /* pass ownership back to list; or destroy in case of an error */
483 sdb_object_deref(SDB_OBJ(conn));
484 }
485 return NULL;
486 } /* connection_handler */
488 static int
489 connection_accept(sdb_fe_socket_t *sock, listener_t *listener)
490 {
491 sdb_object_t *obj;
493 /* the placeholder will be replaced with the accepted file
494 * descriptor when initializing the object */
495 obj = sdb_object_create(CONN_FD_PREFIX CONN_FD_PLACEHOLDER,
496 connection_type, listener->sock_fd);
497 if (! obj)
498 return -1;
500 if (sdb_llist_append(sock->open_connections, obj)) {
501 sdb_log(SDB_LOG_ERR, "frontend: Failed to append "
502 "connection %s to list of open connections",
503 obj->name);
504 sdb_object_deref(obj);
505 return -1;
506 }
508 /* hand ownership over to the list */
509 sdb_object_deref(obj);
510 return 0;
511 } /* connection_accept */
513 static int
514 socket_handle_incoming(sdb_fe_socket_t *sock,
515 fd_set *ready, fd_set *exceptions)
516 {
517 sdb_llist_iter_t *iter;
518 size_t i;
520 for (i = 0; i < sock->listeners_num; ++i) {
521 listener_t *listener = sock->listeners + i;
522 if (FD_ISSET(listener->sock_fd, ready))
523 if (connection_accept(sock, listener))
524 continue;
525 }
527 iter = sdb_llist_get_iter(sock->open_connections);
528 if (! iter) {
529 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
530 "for open connections");
531 return -1;
532 }
534 while (sdb_llist_iter_has_next(iter)) {
535 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
537 if (FD_ISSET(CONN(obj)->fd, exceptions))
538 sdb_log(SDB_LOG_INFO, "Exception on fd %d",
539 CONN(obj)->fd);
541 if (FD_ISSET(CONN(obj)->fd, ready)) {
542 sdb_llist_iter_remove_current(iter);
543 sdb_channel_write(sock->chan, &obj);
544 }
545 }
546 sdb_llist_iter_destroy(iter);
547 return 0;
548 } /* socket_handle_incoming */
550 /*
551 * public API
552 */
554 sdb_fe_socket_t *
555 sdb_fe_sock_create(void)
556 {
557 sdb_fe_socket_t *sock;
559 sock = calloc(1, sizeof(*sock));
560 if (! sock)
561 return NULL;
563 sock->open_connections = sdb_llist_create();
564 if (! sock->open_connections) {
565 sdb_fe_sock_destroy(sock);
566 return NULL;
567 }
568 return sock;
569 } /* sdb_fe_sock_create */
571 void
572 sdb_fe_sock_destroy(sdb_fe_socket_t *sock)
573 {
574 size_t i;
576 if (! sock)
577 return;
579 for (i = 0; i < sock->listeners_num; ++i) {
580 listener_destroy(sock->listeners + i);
581 }
582 if (sock->listeners)
583 free(sock->listeners);
584 sock->listeners = NULL;
586 sdb_llist_destroy(sock->open_connections);
587 sock->open_connections = NULL;
588 free(sock);
589 } /* sdb_fe_sock_destroy */
591 int
592 sdb_fe_sock_add_listener(sdb_fe_socket_t *sock, const char *address)
593 {
594 listener_t *listener;
596 if ((! sock) || (! address))
597 return -1;
599 listener = listener_create(sock, address);
600 if (! listener)
601 return -1;
602 return 0;
603 } /* sdb_fe_sock_add_listener */
605 int
606 sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop)
607 {
608 fd_set sockets;
609 int max_listen_fd = 0;
610 size_t i;
612 /* XXX: make the number of threads configurable */
613 pthread_t handler_threads[5];
615 if ((! sock) || (! sock->listeners_num) || (! loop) || sock->chan)
616 return -1;
618 FD_ZERO(&sockets);
619 for (i = 0; i < sock->listeners_num; ++i) {
620 listener_t *listener = sock->listeners + i;
622 if (listener_listen(listener)) {
623 socket_close(sock);
624 return -1;
625 }
627 FD_SET(listener->sock_fd, &sockets);
628 if (listener->sock_fd > max_listen_fd)
629 max_listen_fd = listener->sock_fd;
630 }
632 sock->chan = sdb_channel_create(1024, sizeof(connection_obj_t *));
633 if (! sock->chan) {
634 socket_close(sock);
635 return -1;
636 }
638 memset(&handler_threads, 0, sizeof(handler_threads));
639 /* XXX: error handling */
640 for (i = 0; i < SDB_STATIC_ARRAY_LEN(handler_threads); ++i)
641 pthread_create(&handler_threads[i], /* attr = */ NULL,
642 connection_handler, /* arg = */ sock);
644 while (loop->do_loop) {
645 struct timeval timeout = { 1, 0 }; /* one second */
646 sdb_llist_iter_t *iter;
648 int max_fd = max_listen_fd;
649 fd_set ready;
650 fd_set exceptions;
651 int n;
653 FD_ZERO(&ready);
654 FD_ZERO(&exceptions);
656 ready = sockets;
658 iter = sdb_llist_get_iter(sock->open_connections);
659 if (! iter) {
660 sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator "
661 "for open connections");
662 break;
663 }
665 while (sdb_llist_iter_has_next(iter)) {
666 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
667 FD_SET(CONN(obj)->fd, &ready);
668 FD_SET(CONN(obj)->fd, &exceptions);
670 if (CONN(obj)->fd > max_fd)
671 max_fd = CONN(obj)->fd;
672 }
673 sdb_llist_iter_destroy(iter);
675 errno = 0;
676 n = select(max_fd + 1, &ready, NULL, &exceptions, &timeout);
677 if (n < 0) {
678 char buf[1024];
680 if (errno == EINTR)
681 continue;
683 sdb_log(SDB_LOG_ERR, "frontend: Failed to monitor sockets: %s",
684 sdb_strerror(errno, buf, sizeof(buf)));
685 break;
686 }
687 else if (! n)
688 continue;
690 /* handle new and open connections */
691 if (socket_handle_incoming(sock, &ready, &exceptions))
692 break;
693 }
695 socket_close(sock);
697 sdb_log(SDB_LOG_INFO, "frontend: Waiting for connection handler threads "
698 "to terminate");
699 if (! sdb_channel_shutdown(sock->chan))
700 for (i = 0; i < SDB_STATIC_ARRAY_LEN(handler_threads); ++i)
701 pthread_join(handler_threads[i], NULL);
702 /* else: we tried our best; let the operating system clean up */
704 sdb_channel_destroy(sock->chan);
705 sock->chan = NULL;
706 return 0;
707 } /* sdb_fe_sock_listen_and_server */
709 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */