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