src/frontend/sock.c

   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  */
  27
  28 #if HAVE_CONFIG_H
  29 #       include "config.h"
  30 #endif /* HAVE_CONFIG_H */
  31
  32 #include "sysdb.h"
  33 #include "core/object.h"
  34 #include "frontend/connection-private.h"
  35 #include "frontend/sock.h"
  36
  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"
  43
  44 #include <assert.h>
  45 #include <errno.h>
  46
  47 #include <stdio.h>
  48 #include <stdlib.h>
  49 #include <string.h>
  50
  51 #include <unistd.h>
  52
  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>
  59
  60 #ifdef HAVE_UCRED_H
  61 #       include <ucred.h>
  62 #endif
  63 #ifdef HAVE_SYS_UCRED_H
  64 #       include <sys/ucred.h>
  65 #endif
  66
  67 #include <pwd.h>
  68
  69 #include <fcntl.h>
  70 #include <netdb.h>
  71 #include <libgen.h>
  72 #include <pthread.h>
  73
  74 /*
  75  * private data types
  76  */
  77
  78 typedef struct {
  79         char *address;
  80         int   type;
  81
  82         /* optional SSL settings */
  83         sdb_ssl_server_t *ssl;
  84
  85         /* listener configuration */
  86         int sock_fd;
  87         int (*setup)(sdb_conn_t *, void *);
  88 } listener_t;
  89
  90 typedef struct {
  91         int type;
  92         const char *prefix;
  93
  94         int (*open)(listener_t *);
  95         void (*close)(listener_t *);
  96 } fe_listener_impl_t;
  97
  98 struct sdb_fe_socket {
  99         listener_t *listeners;
 100         size_t listeners_num;
 101
 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
 110
 111         /* channel used for communication between main
 112          * and connection handler threads */
 113         sdb_channel_t *chan;
 114 };
 115
 116 /*
 117  * SSL helper functions
 118  */
 119
 120 static ssize_t
 121 ssl_read(sdb_conn_t *conn, size_t n)
 122 {
 123         char buf[n];
 124         ssize_t ret;
 125
 126         ret = sdb_ssl_session_read(conn->ssl_session, buf, n);
 127         if (ret <= 0)
 128                 return ret;
 129
 130         sdb_strbuf_memappend(conn->buf, buf, ret);
 131         return ret;
 132 } /* ssl_read */
 133
 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 */
 139
 140 /*
 141  * connection management functions
 142  */
 143
 144 static int
 145 setup_unixsock(sdb_conn_t *conn, void __attribute__((unused)) *user_data)
 146 {
 147         uid_t uid;
 148
 149         struct passwd pw_entry;
 150         struct passwd *result = NULL;
 151         char buf[1024];
 152
 153 #ifdef SO_PEERCRED
 154         struct ucred cred;
 155         socklen_t len = sizeof(cred);
 156
 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
 171
 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 */
 183
 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;
 191
 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         }
 199
 200         memset(&sa, 0, sizeof(sa));
 201         sa.sun_family = AF_UNIX;
 202         strncpy(sa.sun_path, listener->address, sizeof(sa.sun_path));
 203
 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);
 212
 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);
 222
 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         }
 229
 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         }
 237
 238         listener->setup = setup_unixsock;
 239         return 0;
 240 } /* open_unixsock */
 241
 242 static void
 243 close_unixsock(listener_t *listener)
 244 {
 245         assert(listener);
 246
 247         if (! listener->address)
 248                 return;
 249
 250         if (listener->sock_fd >= 0)
 251                 close(listener->sock_fd);
 252         listener->sock_fd = -1;
 253
 254         unlink(listener->address);
 255 } /* close_unixsock */
 256
 257 static int
 258 finish_tcp(sdb_conn_t *conn)
 259 {
 260         if (! conn->ssl_session)
 261                 return 0;
 262
 263         sdb_ssl_session_destroy(conn->ssl_session);
 264         conn->ssl_session = NULL;
 265         return 0;
 266 } /* finish_tcp */
 267
 268 static int
 269 setup_tcp(sdb_conn_t *conn, void *user_data)
 270 {
 271         listener_t *listener = user_data;
 272
 273         conn->ssl_session = sdb_ssl_server_accept(listener->ssl, conn->fd);
 274         if (! conn->ssl_session)
 275                 return -1;
 276
 277         conn->username = sdb_ssl_session_peer(conn->ssl_session);
 278
 279         conn->finish = finish_tcp;
 280         conn->read = ssl_read;
 281         conn->write = ssl_write;
 282         return 0;
 283 } /* setup_tcp */
 284
 285 static int
 286 open_tcp(listener_t *listener)
 287 {
 288         struct addrinfo *ai, *ai_list = NULL;
 289         int status;
 290
 291         assert(listener);
 292
 293         /* TODO: make options configurable */
 294         listener->ssl = sdb_ssl_server_create(NULL);
 295         if (! listener->ssl)
 296                 return -1;
 297
 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         }
 303
 304         for (ai = ai_list; ai != NULL; ai = ai->ai_next) {
 305                 char errbuf[1024];
 306                 int reuse = 1;
 307
 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                 }
 316
 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                 }
 325
 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);
 339
 340         if (listener->sock_fd < 0)
 341                 return -1;
 342
 343         listener->setup = setup_tcp;
 344         return 0;
 345 } /* open_tcp */
 346
 347 static void
 348 close_tcp(listener_t *listener)
 349 {
 350         assert(listener);
 351
 352         sdb_ssl_server_destroy(listener->ssl);
 353         listener->ssl = NULL;
 354
 355         if (listener->sock_fd >= 0)
 356                 close(listener->sock_fd);
 357         listener->sock_fd = -1;
 358 } /* close_tcp */
 359
 360 /*
 361  * private variables
 362  */
 363
 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 };
 374
 375 /*
 376  * private helper functions
 377  */
 378
 379 static int
 380 listener_listen(listener_t *listener)
 381 {
 382         assert(listener);
 383
 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);
 389
 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 */
 398
 399 static void
 400 listener_close(listener_t *listener)
 401 {
 402         assert(listener);
 403
 404         if (listener_impls[listener->type].close)
 405                 listener_impls[listener->type].close(listener);
 406
 407         if (listener->sock_fd >= 0)
 408                 close(listener->sock_fd);
 409         listener->sock_fd = -1;
 410 } /* listener_close */
 411
 412 static int
 413 get_type(const char *address)
 414 {
 415         char *sep;
 416         size_t len;
 417         size_t i;
 418
 419         if (*address == '/')
 420                 return LISTENER_UNIXSOCK;
 421         sep = strchr(address, (int)':');
 422         if (! sep)
 423                 return listener_impls[0].type;
 424
 425         assert(sep > address);
 426         len = (size_t)(sep - address);
 427
 428         for (i = 0; i < SDB_STATIC_ARRAY_LEN(listener_impls); ++i) {
 429                 fe_listener_impl_t *impl = listener_impls + i;
 430
 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 */
 438
 439 static void
 440 listener_destroy(listener_t *listener)
 441 {
 442         if (! listener)
 443                 return;
 444
 445         listener_close(listener);
 446
 447         if (listener->address)
 448                 free(listener->address);
 449         listener->address = NULL;
 450 } /* listener_destroy */
 451
 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;
 458
 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         }
 465
 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         }
 474
 475         sock->listeners = listener;
 476         listener = sock->listeners + sock->listeners_num;
 477
 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;
 482
 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;
 495
 496         if (listener_impls[type].open(listener)) {
 497                 /* prints error */
 498                 listener_destroy(listener);
 499                 return NULL;
 500         }
 501
 502         ++sock->listeners_num;
 503         return listener;
 504 } /* listener_create */
 505
 506 static void
 507 socket_clear(sdb_fe_socket_t *sock)
 508 {
 509         size_t i;
 510
 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 */
 519
 520 static void
 521 socket_close(sdb_fe_socket_t *sock)
 522 {
 523         size_t i;
 524
 525         assert(sock);
 526         for (i = 0; i < sock->listeners_num; ++i)
 527                 listener_close(sock->listeners + i);
 528 } /* socket_close */
 529
 530 /*
 531  * connection handler functions
 532  */
 533
 534 static void *
 535 connection_handler(void *data)
 536 {
 537         sdb_fe_socket_t *sock = data;
 538
 539         assert(sock);
 540
 541         while (42) {
 542                 struct timespec timeout = { 0, 500000000 }; /* .5 seconds */
 543                 sdb_conn_t *conn;
 544                 int status;
 545
 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];
 551
 552                         if (errno == ETIMEDOUT)
 553                                 continue;
 554                         if (errno == EBADF) /* channel shut down */
 555                                 break;
 556
 557                         sdb_log(SDB_LOG_ERR, "frontend: Failed to read from channel: %s",
 558                                         sdb_strerror(errno, buf, sizeof(buf)));
 559                         continue;
 560                 }
 561
 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                 }
 568
 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);
 576
 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 */
 582
 583 static int
 584 connection_accept(sdb_fe_socket_t *sock, listener_t *listener)
 585 {
 586         sdb_object_t *obj;
 587         int status;
 588
 589         obj = SDB_OBJ(sdb_connection_accept(listener->sock_fd,
 590                                 listener->setup, listener));
 591         if (! obj)
 592                 return -1;
 593
 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);
 599
 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 */
 604
 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;
 611
 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         }
 618
 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         }
 625
 626         while (sdb_llist_iter_has_next(iter)) {
 627                 sdb_object_t *obj = sdb_llist_iter_get_next(iter);
 628
 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                 }
 637
 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 */
 646
 647 /*
 648  * public API
 649  */
 650
 651 sdb_fe_socket_t *
 652 sdb_fe_sock_create(void)
 653 {
 654         sdb_fe_socket_t *sock;
 655         int flags;
 656
 657         sock = calloc(1, sizeof(*sock));
 658         if (! sock)
 659                 return NULL;
 660         sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1;
 661
 662         sock->open_connections = sdb_llist_create();
 663         if (! sock->open_connections) {
 664                 sdb_fe_sock_destroy(sock);
 665                 return NULL;
 666         }
 667
 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 */
 695
 696 void
 697 sdb_fe_sock_destroy(sdb_fe_socket_t *sock)
 698 {
 699         if (! sock)
 700                 return;
 701
 702         socket_clear(sock);
 703
 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;
 709
 710         sdb_llist_destroy(sock->open_connections);
 711         sock->open_connections = NULL;
 712         free(sock);
 713 } /* sdb_fe_sock_destroy */
 714
 715 int
 716 sdb_fe_sock_add_listener(sdb_fe_socket_t *sock, const char *address)
 717 {
 718         listener_t *listener;
 719
 720         if ((! sock) || (! address))
 721                 return -1;
 722
 723         listener = listener_create(sock, address);
 724         if (! listener)
 725                 return -1;
 726         return 0;
 727 } /* sdb_fe_sock_add_listener */
 728
 729 void
 730 sdb_fe_sock_clear_listeners(sdb_fe_socket_t *sock)
 731 {
 732         if (! sock)
 733                 return;
 734
 735         socket_clear(sock);
 736 } /* sdb_fe_sock_clear_listeners */
 737
 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;
 744
 745         pthread_t handler_threads[loop->num_threads];
 746         size_t num_threads;
 747
 748         if ((! sock) || (! sock->listeners_num) || sock->chan
 749                         || (! loop) || (loop->num_threads <= 0))
 750                 return -1;
 751
 752         if (! loop->do_loop)
 753                 return 0;
 754
 755         FD_ZERO(&sockets);
 756         for (i = 0; i < sock->listeners_num; ++i) {
 757                 listener_t *listener = sock->listeners + i;
 758
 759                 if (listener_listen(listener)) {
 760                         socket_close(sock);
 761                         return -1;
 762                 }
 763
 764                 FD_SET(listener->sock_fd, &sockets);
 765                 if (listener->sock_fd > max_listen_fd)
 766                         max_listen_fd = listener->sock_fd;
 767         }
 768
 769         sock->chan = sdb_channel_create(1024, sizeof(sdb_conn_t *));
 770         if (! sock->chan) {
 771                 socket_close(sock);
 772                 return -1;
 773         }
 774
 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");
 779
 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         }
 794
 795         while (loop->do_loop && num_threads) {
 796                 struct timeval timeout = { 1, 0 }; /* one second */
 797                 sdb_llist_iter_t *iter;
 798
 799                 int max_fd = max_listen_fd;
 800                 fd_set ready;
 801                 fd_set exceptions;
 802                 int n;
 803
 804                 FD_ZERO(&ready);
 805                 FD_ZERO(&exceptions);
 806
 807                 ready = sockets;
 808                 FD_SET(sock->trigger[TRIGGER_READ], &ready);
 809
 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                 }
 816
 817                 while (sdb_llist_iter_has_next(iter)) {
 818                         sdb_object_t *obj = sdb_llist_iter_get_next(iter);
 819
 820                         if (CONN(obj)->fd < 0) {
 821                                 sdb_llist_iter_remove_current(iter);
 822                                 sdb_object_deref(obj);
 823                                 continue;
 824                         }
 825
 826                         FD_SET(CONN(obj)->fd, &ready);
 827                         FD_SET(CONN(obj)->fd, &exceptions);
 828
 829                         if (CONN(obj)->fd > max_fd)
 830                                 max_fd = CONN(obj)->fd;
 831                 }
 832                 sdb_llist_iter_destroy(iter);
 833
 834                 errno = 0;
 835                 n = select(max_fd + 1, &ready, NULL, &exceptions, &timeout);
 836                 if (n < 0) {
 837                         char buf[1024];
 838
 839                         if (errno == EINTR)
 840                                 continue;
 841
 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;
 848
 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                 }
 854
 855                 /* handle new and open connections */
 856                 if (socket_handle_incoming(sock, &ready, &exceptions))
 857                         break;
 858         }
 859
 860         socket_close(sock);
 861
 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 */
 868
 869         sdb_channel_destroy(sock->chan);
 870         sock->chan = NULL;
 871
 872         if (! num_threads)
 873                 return -1;
 874         return 0;
 875 } /* sdb_fe_sock_listen_and_server */
 876
 877 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */
 878