src/IOHandler.c

   1 /* IOHandler.c - IOMultiplexer
   2  * Copyright (C) 2012  Philipp Kreil (pk910)
   3  *
   4  * This program is free software: you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License as published by
   6  * the Free Software Foundation, either version 3 of the License, or
   7  * (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  * GNU General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License
  15  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  16  */
  17 #include "IOHandler.h"
  18 #include "IOEngine.h"
  19 #include "IOHandler_SSL.h"
  20 #include <errno.h>
  21 #include <stdio.h>
  22 #include <unistd.h>
  23 #ifdef WIN32
  24 #define _WIN32_WINNT 0x501
  25 #include <windows.h>
  26 #include <winsock2.h>
  27 #include <ws2tcpip.h>
  28 #else
  29 #include <signal.h>
  30 #include <stdarg.h>
  31 #include <stdlib.h>
  32 #include <string.h>
  33 #include <netdb.h>
  34 #include <sys/socket.h>
  35 #include <netinet/in.h>
  36 #include <netinet/tcp.h>
  37 #endif
  38
  39 #ifndef EWOULDBLOCK
  40 #define EWOULDBLOCK EAGAIN
  41 #endif
  42
  43 #define MAXLOG 1024
  44 iohandler_log_callback *iolog_backend = NULL;
  45
  46 struct IODescriptor *first_descriptor = NULL;
  47 struct IODescriptor *timer_priority = NULL;
  48
  49 #ifdef HAVE_PTHREAD_H
  50 static pthread_mutex_t io_thread_sync;
  51 static pthread_mutex_t io_poll_sync;
  52 #endif
  53
  54 void iohandler_log(enum IOLogType type, char *text, ...) {
  55     va_list arg_list;
  56     char logBuf[MAXLOG+1];
  57     int pos;
  58     logBuf[0] = '\0';
  59     va_start(arg_list, text);
  60     pos = vsnprintf(logBuf, MAXLOG - 1, text, arg_list);
  61     va_end(arg_list);
  62     if (pos < 0 || pos > (MAXLOG - 1)) pos = MAXLOG - 1;
  63     logBuf[pos] = '\n';
  64     logBuf[pos+1] = '\0';
  65
  66     if(iolog_backend)
  67         iolog_backend(type, logBuf);
  68 }
  69
  70 /* IO Engines */
  71 extern struct IOEngine engine_select; /* select system call (should always be useable) */
  72 extern struct IOEngine engine_kevent;
  73 extern struct IOEngine engine_epoll;
  74 extern struct IOEngine engine_win32;
  75
  76 struct IOEngine *engine = NULL;
  77
  78 static void iohandler_init_engine() {
  79     if(engine) return;
  80     IOTHREAD_MUTEX_INIT(io_thread_sync);
  81     IOTHREAD_MUTEX_INIT(io_poll_sync);
  82
  83     //try other engines
  84     if(!engine && engine_kevent.init && engine_kevent.init())
  85         engine = &engine_kevent;
  86     if(!engine && engine_epoll.init && engine_epoll.init())
  87         engine = &engine_epoll;
  88     if(!engine && engine_win32.init && engine_win32.init())
  89         engine = &engine_win32;
  90
  91     if (!engine) {
  92         if(engine_select.init())
  93             engine = &engine_select;
  94         else {
  95             iohandler_log(IOLOG_FATAL, "found no useable IO engine");
  96             return;
  97         }
  98     }
  99     iohandler_log(IOLOG_DEBUG, "using %s IO engine", engine->name);
 100     iohandler_ssl_init();
 101 }
 102
 103 static void iohandler_append(struct IODescriptor *descriptor) {
 104     IOSYNCHRONIZE(io_thread_sync);
 105     struct timeval *timeout = ((descriptor->timeout.tv_sec || descriptor->timeout.tv_usec) ? &descriptor->timeout : NULL);
 106     if(timeout) {
 107         struct IODescriptor *iofd;
 108         int set_priority = 1;
 109         descriptor->timeout = *timeout;
 110         if(timer_priority)
 111             iofd = timer_priority;
 112         else
 113             iofd = first_descriptor;
 114         if(iofd) {
 115             for(;;iofd = iofd->next) {
 116                 if(timeval_is_smaler(timeout, (&iofd->timeout))) {
 117                     descriptor->prev = iofd->prev;
 118                     descriptor->next = iofd;
 119                     if(iofd->prev)
 120                         iofd->prev->next = descriptor;
 121                     iofd->prev = descriptor;
 122                     if(set_priority)
 123                         timer_priority = descriptor;
 124                     break;
 125                 }
 126                 if(iofd == timer_priority)
 127                     set_priority = 0;
 128                 if(iofd->next == NULL) {
 129                     descriptor->next = NULL;
 130                     descriptor->prev = iofd;
 131                     iofd->next = descriptor;
 132                     if(set_priority)
 133                         timer_priority = descriptor;
 134                     break;
 135                 }
 136             }
 137         } else {
 138             descriptor->prev = NULL;
 139             descriptor->next = NULL;
 140             first_descriptor = descriptor;
 141             timer_priority = descriptor;
 142         }
 143
 144     } else {
 145         descriptor->prev = NULL;
 146         descriptor->next = first_descriptor;
 147         if(first_descriptor)
 148             first_descriptor->prev = descriptor;
 149         first_descriptor = descriptor;
 150     }
 151     IODESYNCHRONIZE(io_thread_sync);
 152 }
 153
 154 static void iohandler_remove(struct IODescriptor *descriptor, int engine_remove) {
 155     //remove IODescriptor from the list
 156     IOSYNCHRONIZE(io_thread_sync);
 157     if(descriptor->prev)
 158         descriptor->prev->next = descriptor->next;
 159     else
 160         first_descriptor = descriptor->next;
 161     if(descriptor->next)
 162         descriptor->next->prev = descriptor->prev;
 163     if(descriptor == timer_priority)
 164         timer_priority = descriptor->next;
 165
 166     if(engine_remove)
 167         engine->remove(descriptor);
 168     if(descriptor->readbuf.buffer)
 169         free(descriptor->readbuf.buffer);
 170     if(descriptor->writebuf.buffer)
 171         free(descriptor->writebuf.buffer);
 172     iohandler_log(IOLOG_DEBUG, "removed IODescriptor (%d) of type `%s`", descriptor->fd, iohandler_iotype_name(descriptor->type));
 173     free(descriptor);
 174     IODESYNCHRONIZE(io_thread_sync);
 175 }
 176
 177 struct IODescriptor *iohandler_add(int sockfd, enum IOType type, struct timeval *timeout, iohandler_callback *callback) {
 178     //just add a new IODescriptor
 179     struct IODescriptor *descriptor = calloc(1, sizeof(*descriptor));
 180     if(!descriptor) {
 181         iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
 182         return NULL;
 183     }
 184     descriptor->fd = (type == IOTYPE_STDIN ? fileno(stdin) : sockfd);
 185     descriptor->type = type;
 186     descriptor->state = (type == IOTYPE_STDIN ? IO_CONNECTED : IO_CLOSED);
 187     descriptor->callback = callback;
 188     if(timeout)
 189         descriptor->timeout = *timeout;
 190     if(type != IOTYPE_TIMER) {
 191         descriptor->readbuf.buffer = malloc(IO_READ_BUFLEN + 2);
 192         descriptor->readbuf.bufpos = 0;
 193         descriptor->readbuf.buflen = IO_READ_BUFLEN;
 194         descriptor->writebuf.buffer = malloc(IO_READ_BUFLEN + 2);
 195         descriptor->writebuf.bufpos = 0;
 196         descriptor->writebuf.buflen = IO_READ_BUFLEN;
 197     }
 198
 199     if(!engine) {
 200         iohandler_init_engine();
 201         if(!engine) {
 202             return NULL;
 203         }
 204     }
 205     engine->add(descriptor);
 206
 207     //add IODescriptor to the list
 208     iohandler_append(descriptor);
 209
 210     iohandler_log(IOLOG_DEBUG, "added custom socket descriptor (%d) as type `%s`", sockfd, iohandler_iotype_name(type));
 211     return descriptor;
 212 }
 213
 214 void iohandler_set_timeout(struct IODescriptor *descriptor, struct timeval *timeout) {
 215     if(descriptor->prev)
 216         descriptor->prev->next = descriptor->next;
 217     else
 218         first_descriptor = descriptor->next;
 219     if(descriptor->next)
 220         descriptor->next->prev = descriptor->prev;
 221     if(descriptor == timer_priority)
 222         timer_priority = descriptor->next;
 223     if(timeout)
 224         descriptor->timeout = *timeout;
 225     else {
 226         descriptor->timeout.tv_sec = 0;
 227         descriptor->timeout.tv_usec = 0;
 228     }
 229     iohandler_append(descriptor);
 230 }
 231
 232 static void iohandler_increase_iobuf(struct IOBuffer *iobuf, size_t required) {
 233     if(iobuf->buflen >= required) return;
 234     char *new_buf = realloc(iobuf->buffer, required + 2);
 235     if(new_buf) {
 236         iobuf->buffer = new_buf;
 237         iobuf->buflen = required;
 238     }
 239 }
 240
 241 struct IODescriptor *iohandler_timer(struct timeval timeout, iohandler_callback *callback) {
 242     struct IODescriptor *descriptor;
 243     descriptor = iohandler_add(-1, IOTYPE_TIMER, &timeout, callback);
 244     if(!descriptor) {
 245         iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
 246         return NULL;
 247     }
 248     iohandler_log(IOLOG_DEBUG, "added timer descriptor (sec: %d; usec: %d)", timeout.tv_sec, timeout.tv_usec);
 249     return descriptor;
 250 }
 251
 252 struct IODescriptor *iohandler_connect(const char *hostname, unsigned int port, int ssl, const char *bindhost, iohandler_callback *callback) {
 253     return iohandler_connect_flags(hostname, port, ssl, bindhost, callback, IOHANDLER_CONNECT_IPV4 | IOHANDLER_CONNECT_IPV6);
 254 }
 255
 256 struct IODescriptor *iohandler_connect_flags(const char *hostname, unsigned int port, int ssl, const char *bindhost, iohandler_callback *callback, int flags) {
 257     //non-blocking connect
 258     int sockfd, result;
 259     struct addrinfo hints, *res;
 260     struct sockaddr_in *ip4 = NULL;
 261     struct sockaddr_in6 *ip6 = NULL;
 262     size_t dstaddrlen;
 263     struct sockaddr *dstaddr = NULL;
 264     struct IODescriptor *descriptor;
 265
 266     if(!engine) {
 267         iohandler_init_engine();
 268         if(!engine) return NULL;
 269     }
 270     memset (&hints, 0, sizeof (hints));
 271     hints.ai_family = PF_UNSPEC;
 272     hints.ai_socktype = SOCK_STREAM;
 273     hints.ai_flags |= AI_CANONNAME;
 274     if ((result = getaddrinfo (hostname, NULL, &hints, &res))) {
 275         iohandler_log(IOLOG_ERROR, "could not resolve %s to an IP address (%d)", hostname, result);
 276         return NULL;
 277     }
 278     while (res) {
 279         switch (res->ai_family) {
 280         case AF_INET:
 281             ip4 = (struct sockaddr_in *) res->ai_addr;
 282             break;
 283         case AF_INET6:
 284             ip6 = (struct sockaddr_in6 *) res->ai_addr;
 285             break;
 286         }
 287         res = res->ai_next;
 288         freeaddrinfo(res);
 289     }
 290
 291     if(ip6 && (flags & IOHANDLER_CONNECT_IPV6)) {
 292         sockfd = socket(AF_INET6, SOCK_STREAM, 0);
 293         if(sockfd == -1) {
 294             iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
 295             return NULL;
 296         }
 297
 298         ip6->sin6_family = AF_INET6;
 299         ip6->sin6_port = htons(port);
 300
 301         struct sockaddr_in6 *ip6vhost = NULL;
 302         if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
 303             while (res) {
 304                 switch (res->ai_family) {
 305                 case AF_INET6:
 306                     ip6vhost = (struct sockaddr_in6 *) res->ai_addr;
 307                     break;
 308                 }
 309                 res = res->ai_next;
 310                 freeaddrinfo(res);
 311             }
 312         }
 313         if(ip6vhost) {
 314             ip6vhost->sin6_family = AF_INET6;
 315             ip6vhost->sin6_port = htons(0);
 316             bind(sockfd, (struct sockaddr*)ip6vhost, sizeof(*ip6vhost));
 317         }
 318         dstaddr = (struct sockaddr*)ip6;
 319         dstaddrlen = sizeof(*ip6);
 320     } else if(ip4 && (flags & IOHANDLER_CONNECT_IPV4)) {
 321         sockfd = socket(AF_INET, SOCK_STREAM, 0);
 322         if(sockfd == -1) {
 323             iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
 324             return NULL;
 325         }
 326
 327         ip4->sin_family = AF_INET;
 328         ip4->sin_port = htons(port);
 329
 330         struct sockaddr_in *ip4vhost = NULL;
 331         if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
 332             while (res) {
 333                 switch (res->ai_family) {
 334                 case AF_INET:
 335                     ip4vhost = (struct sockaddr_in *) res->ai_addr;
 336                     break;
 337                 }
 338                 res = res->ai_next;
 339                 freeaddrinfo(res);
 340             }
 341         }
 342         if(ip4vhost) {
 343             ip4vhost->sin_family = AF_INET;
 344             ip4vhost->sin_port = htons(0);
 345             bind(sockfd, (struct sockaddr*)ip4vhost, sizeof(*ip4vhost));
 346         }
 347         dstaddr = (struct sockaddr*)ip4;
 348         dstaddrlen = sizeof(*ip4);
 349     } else
 350         return NULL;
 351     //prevent SIGPIPE
 352     #ifndef WIN32
 353     #if defined(SO_NOSIGPIPE)
 354     {
 355         int set = 1;
 356         setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
 357     }
 358     #else
 359     signal(SIGPIPE, SIG_IGN);
 360     #endif
 361     #endif
 362     //make sockfd unblocking
 363     #if defined(F_GETFL)
 364     {
 365         int flags;
 366         flags = fcntl(sockfd, F_GETFL);
 367         fcntl(sockfd, F_SETFL, flags|O_NONBLOCK);
 368         flags = fcntl(sockfd, F_GETFD);
 369         fcntl(sockfd, F_SETFD, flags|FD_CLOEXEC);
 370     }
 371     #else
 372     /* I hope you're using the Win32 backend or something else that
 373      * automatically marks the file descriptor non-blocking...
 374      */
 375     #endif
 376     descriptor = iohandler_add(sockfd, IOTYPE_CLIENT, NULL, callback);
 377     if(!descriptor) {
 378         close(sockfd);
 379         return NULL;
 380     }
 381     connect(sockfd, dstaddr, dstaddrlen); //returns EINPROGRESS here (nonblocking)
 382     descriptor->state = IO_CONNECTING;
 383     descriptor->ssl = (ssl ? 1 : 0);
 384     descriptor->read_lines = 1;
 385     engine->update(descriptor);
 386     iohandler_log(IOLOG_DEBUG, "added client socket (%d) connecting to %s:%d", sockfd, hostname, port);
 387     return descriptor;
 388 }
 389
 390 struct IODescriptor *iohandler_listen(const char *hostname, unsigned int port, iohandler_callback *callback) {
 391     return iohandler_listen_flags(hostname, port, callback, IOHANDLER_LISTEN_IPV4 | IOHANDLER_LISTEN_IPV6);
 392 }
 393
 394 struct IODescriptor *iohandler_listen_flags(const char *hostname, unsigned int port, iohandler_callback *callback, int flags) {
 395     int sockfd;
 396     struct addrinfo hints, *res;
 397     struct sockaddr_in *ip4 = NULL;
 398     struct sockaddr_in6 *ip6 = NULL;
 399     struct IODescriptor *descriptor;
 400     unsigned int opt;
 401
 402     if(!engine) {
 403         iohandler_init_engine();
 404         if(!engine) return NULL;
 405     }
 406     memset (&hints, 0, sizeof (hints));
 407     hints.ai_family = PF_UNSPEC;
 408     hints.ai_socktype = SOCK_STREAM;
 409     hints.ai_flags |= AI_CANONNAME;
 410     if (getaddrinfo (hostname, NULL, &hints, &res)) {
 411         return NULL;
 412     }
 413     while (res) {
 414         switch (res->ai_family) {
 415         case AF_INET:
 416             ip4 = (struct sockaddr_in *) res->ai_addr;
 417             break;
 418         case AF_INET6:
 419             ip6 = (struct sockaddr_in6 *) res->ai_addr;
 420             break;
 421         }
 422         res = res->ai_next;
 423         freeaddrinfo(res);
 424     }
 425
 426     if(ip6 && (flags & IOHANDLER_LISTEN_IPV6)) {
 427         sockfd = socket(AF_INET6, SOCK_STREAM, 0);
 428         if(sockfd == -1) return NULL;
 429
 430         opt = 1;
 431         setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
 432
 433         ip6->sin6_family = AF_INET6;
 434         ip6->sin6_port = htons(port);
 435
 436         bind(sockfd, (struct sockaddr*)ip6, sizeof(*ip6));
 437     } else if(ip4 && (flags && IOHANDLER_LISTEN_IPV4)) {
 438         sockfd = socket(AF_INET, SOCK_STREAM, 0);
 439         if(sockfd == -1) return NULL;
 440
 441         opt = 1;
 442         setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
 443
 444         ip4->sin_family = AF_INET;
 445         ip4->sin_port = htons(port);
 446
 447         bind(sockfd, (struct sockaddr*)ip4, sizeof(*ip4));
 448     } else
 449         return NULL;
 450     //prevent SIGPIPE
 451     #ifndef WIN32
 452     #if defined(SO_NOSIGPIPE)
 453     {
 454         int set = 1;
 455         setsockopt(sockfd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
 456     }
 457     #else
 458     signal(SIGPIPE, SIG_IGN);
 459     #endif
 460     #endif
 461     //make sockfd unblocking
 462     #if defined(F_GETFL)
 463     {
 464         int flag;
 465         flag = fcntl(sockfd, F_GETFL);
 466         fcntl(sockfd, F_SETFL, flag|O_NONBLOCK);
 467         flag = fcntl(sockfd, F_GETFD);
 468         fcntl(sockfd, F_SETFD, flag|FD_CLOEXEC);
 469     }
 470     #else
 471     /* I hope you're using the Win32 backend or something else that
 472      * automatically marks the file descriptor non-blocking...
 473      */
 474     #endif
 475     descriptor = iohandler_add(sockfd, IOTYPE_SERVER, NULL, callback);
 476     if(!descriptor) {
 477         close(sockfd);
 478         return NULL;
 479     }
 480     listen(sockfd, 1);
 481     descriptor->state = IO_LISTENING;
 482     engine->update(descriptor);
 483     iohandler_log(IOLOG_DEBUG, "added server socket (%d) listening on %s:%d", sockfd, hostname, port);
 484     return descriptor;
 485 }
 486
 487 struct IODescriptor *iohandler_listen_ssl(const char *hostname, unsigned int port, const char *certfile, const char *keyfile, iohandler_callback *callback) {
 488     return iohandler_listen_ssl_flags(hostname, port, certfile, keyfile, callback, IOHANDLER_LISTEN_IPV4 | IOHANDLER_LISTEN_IPV6);
 489 }
 490
 491 struct IODescriptor *iohandler_listen_ssl_flags(const char *hostname, unsigned int port, const char *certfile, const char *keyfile, iohandler_callback *callback, int flags) {
 492     struct IODescriptor *descriptor = iohandler_listen_flags(hostname, port, callback, flags);
 493     if(!descriptor)
 494         return NULL;
 495     //SSL Server Socket
 496     iohandler_ssl_listen(descriptor, certfile, keyfile);
 497     if(descriptor->sslnode)
 498         descriptor->ssl = 1;
 499     return descriptor;
 500 }
 501
 502 void iohandler_write(struct IODescriptor *iofd, const char *line) {
 503     size_t linelen = strlen(line);
 504     iohandler_send(iofd, line, linelen);
 505 }
 506
 507 void iohandler_send(struct IODescriptor *iofd, const char *data, size_t datalen) {
 508     if(iofd->type == IOTYPE_TIMER || iofd->state == IO_CLOSED) {
 509         iohandler_log(IOLOG_ERROR, "could not write to socket (%s)", (iofd->type == IOTYPE_TIMER ? "IOTYPE_TIMER" : "IO_CLOSED"));
 510         return;
 511     }
 512     iohandler_log(IOLOG_DEBUG, "add %d to writebuf (fd: %d): %s", datalen, iofd->fd, data);
 513     if(iofd->writebuf.buflen < iofd->writebuf.bufpos + datalen) {
 514         iohandler_log(IOLOG_DEBUG, "increase writebuf (curr: %d) to %d (+%d bytes)", iofd->writebuf.buflen, iofd->writebuf.bufpos + datalen, (iofd->writebuf.bufpos + datalen - iofd->writebuf.buflen));
 515         iohandler_increase_iobuf(&iofd->writebuf, iofd->writebuf.bufpos + datalen);
 516         if(iofd->writebuf.buflen < iofd->writebuf.bufpos + datalen) {
 517             iohandler_log(IOLOG_ERROR, "increase writebuf (curr: %d) to %d (+%d bytes) FAILED", iofd->writebuf.buflen, iofd->writebuf.bufpos + datalen, (iofd->writebuf.bufpos + datalen - iofd->writebuf.buflen));
 518             return;
 519         }
 520     }
 521     memcpy(iofd->writebuf.buffer + iofd->writebuf.bufpos, data, datalen);
 522     iofd->writebuf.bufpos += datalen;
 523     engine->update(iofd);
 524 }
 525
 526 void iohandler_printf(struct IODescriptor *iofd, const char *text, ...) {
 527     va_list arg_list;
 528     char sendBuf[IO_LINE_LEN];
 529     int pos;
 530     sendBuf[0] = '\0';
 531     va_start(arg_list, text);
 532     pos = vsnprintf(sendBuf, IO_LINE_LEN - 2, text, arg_list);
 533     va_end(arg_list);
 534     if (pos < 0 || pos > (IO_LINE_LEN - 2)) pos = IO_LINE_LEN - 2;
 535     sendBuf[pos] = '\n';
 536     sendBuf[pos+1] = '\0';
 537     iohandler_send(iofd, sendBuf, pos+1);
 538 }
 539
 540 static int iohandler_try_write(struct IODescriptor *iofd) {
 541     if(!iofd->writebuf.bufpos) return 0;
 542     iohandler_log(IOLOG_DEBUG, "write writebuf (%d bytes) to socket (fd: %d)", iofd->writebuf.bufpos, iofd->fd);
 543     int res;
 544     if(iofd->ssl_active)
 545         res = iohandler_ssl_write(iofd, iofd->writebuf.buffer, iofd->writebuf.bufpos);
 546     else
 547         res = send(iofd->fd, iofd->writebuf.buffer, iofd->writebuf.bufpos, 0);
 548     if(res < 0) {
 549         if (errno != EAGAIN && errno != EWOULDBLOCK)
 550             iohandler_log(IOLOG_ERROR, "could not write to socket (fd: %d): %d - %s", iofd->fd, errno, strerror(errno));
 551         else
 552             res = 0;
 553     } else {
 554         iofd->writebuf.bufpos -= res;
 555         if(iofd->state != IO_CLOSED)
 556             engine->update(iofd);
 557     }
 558     return res;
 559 }
 560
 561 void iohandler_close(struct IODescriptor *iofd) {
 562     int engine_remove = 1;
 563     iofd->state = IO_CLOSED;
 564     if(iofd->writebuf.bufpos) {
 565         //try to send everything before closing
 566 #if defined(F_GETFL)
 567         {
 568             int flags;
 569             flags = fcntl(iofd->fd, F_GETFL);
 570             fcntl(iofd->fd, F_SETFL, flags & ~O_NONBLOCK);
 571             flags = fcntl(iofd->fd, F_GETFD);
 572             fcntl(iofd->fd, F_SETFD, flags|FD_CLOEXEC);
 573         }
 574 #else
 575         engine_remove = 0;
 576         engine->remove(iofd);
 577 #endif
 578         iohandler_try_write(iofd);
 579     }
 580     //close IODescriptor
 581     if(iofd->ssl)
 582         iohandler_ssl_disconnect(iofd);
 583     if(iofd->type == IOTYPE_SERVER || iofd->type == IOTYPE_CLIENT || iofd->type == IOTYPE_STDIN)
 584         close(iofd->fd);
 585     iohandler_remove(iofd, engine_remove);
 586 }
 587
 588 void iohandler_update(struct IODescriptor *iofd) {
 589     iohandler_log(IOLOG_DEBUG, "external call to iohandler_update (fd: %d)", iofd->fd);
 590     engine->update(iofd);
 591 }
 592
 593 static void iohandler_trigger_event(struct IOEvent *event) {
 594     if(!event->iofd->callback) return;
 595     iohandler_log(IOLOG_DEBUG, "triggering event (%s) for %s (fd: %d)", iohandler_ioeventtype_name(event->type), iohandler_iotype_name(event->iofd->type), event->iofd->fd);
 596     event->iofd->callback(event);
 597 }
 598
 599 void iohandler_events(struct IODescriptor *iofd, int readable, int writeable) {
 600     struct IOEvent callback_event;
 601     callback_event.type = IOEVENT_IGNORE;
 602     callback_event.iofd = iofd;
 603     switch(iofd->state) {
 604         case IO_SSLWAIT:
 605             if(!readable && !writeable) {
 606                 if(!iofd->ssl_server_hs) {
 607                     callback_event.type = IOEVENT_SSLFAILED;
 608                     iofd->state = IO_CLOSED;
 609                     engine->update(iofd);
 610                 } else
 611                     iohandler_close(iofd);
 612             } else if(iofd->ssl_server_hs) {
 613                 iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_server_handshake for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
 614                 iohandler_ssl_server_handshake(iofd);
 615             } else {
 616                 iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_client_handshake for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
 617                 iohandler_ssl_client_handshake(iofd);
 618             }
 619             break;
 620         case IO_CLOSED:
 621             if(iofd->type == IOTYPE_TIMER)
 622                 callback_event.type = IOEVENT_TIMEOUT;
 623             break;
 624         case IO_LISTENING:
 625             if(readable) {
 626                 callback_event.data.accept_fd = accept(iofd->fd, NULL, 0);
 627                 if(callback_event.data.accept_fd < 0) {
 628                     iohandler_log(IOLOG_ERROR, "could not accept client (server fd: %d): %d - %s", iofd->fd, errno, strerror(errno));
 629                 } else if(iofd->ssl) {
 630                     struct IODescriptor *client_iofd = iohandler_add(callback_event.data.accept_fd, IOTYPE_CLIENT, NULL, NULL);
 631                     iohandler_ssl_client_accepted(iofd, client_iofd);
 632                 } else
 633                     callback_event.type = IOEVENT_ACCEPT;
 634             }
 635             break;
 636         case IO_CONNECTING:
 637             if(readable) { //could not connect
 638                 callback_event.type = IOEVENT_NOTCONNECTED;
 639                 //socklen_t arglen;
 640                 //arglen = sizeof(callback_event.data.errid);
 641                 //if (getsockopt(iofd->fd, SOL_SOCKET, SO_ERROR, &callback_event.data.errid, &arglen) < 0)
 642                 //    callback_event.data.errid = errno;
 643                 iofd->state = IO_CLOSED;
 644                                 engine->update(iofd);
 645             } else if(writeable) {
 646                 if(iofd->ssl && !iofd->ssl_active) {
 647                     iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_connect for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
 648                     iohandler_ssl_connect(iofd);
 649                     return;
 650                 }
 651                 if(iofd->ssl && iofd->ssl_server_hs) {
 652                     callback_event.type = IOEVENT_SSLACCEPT;
 653                     callback_event.iofd = iofd->data;
 654                     callback_event.data.accept_iofd = iofd;
 655                     iofd->data = NULL;
 656                 }
 657                 else
 658                     callback_event.type = IOEVENT_CONNECTED;
 659                 iofd->state = IO_CONNECTED;
 660                 engine->update(iofd);
 661             }
 662             break;
 663         case IO_CONNECTED:
 664             if(readable) {
 665                 if(iofd->read_lines) {
 666                     int bytes;
 667
 668                     if(iofd->ssl_active)
 669                         bytes = iohandler_ssl_read(iofd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos);
 670                     else {
 671                         if(iofd->type == IOTYPE_STDIN)
 672                             #ifdef WIN32
 673                             bytes = readable;
 674                             #else
 675                             bytes = read(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos);
 676                             #endif
 677                         else
 678                             bytes = recv(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos, 0);
 679                     }
 680                     if(bytes <= 0) {
 681                         if (errno != EAGAIN || errno != EWOULDBLOCK) {
 682                             iofd->state = IO_CLOSED;
 683                                                         engine->update(iofd);
 684                             callback_event.type = IOEVENT_CLOSED;
 685                             callback_event.data.errid = errno;
 686                         }
 687                     } else {
 688                         int i, used_bytes = 0;
 689                         iohandler_log(IOLOG_DEBUG, "received %d bytes (fd: %d). readbuf position: %d", bytes, iofd->fd, iofd->readbuf.bufpos);
 690                         iofd->readbuf.bufpos += bytes;
 691                         callback_event.type = IOEVENT_RECV;
 692                         for(i = 0; i < iofd->readbuf.bufpos; i++) {
 693                             if(iofd->readbuf.buffer[i] == '\r' && iofd->readbuf.buffer[i+1] == '\n')
 694                                 iofd->readbuf.buffer[i] = 0;
 695                             else if(iofd->readbuf.buffer[i] == '\n' || iofd->readbuf.buffer[i] == '\r') {
 696                                 iofd->readbuf.buffer[i] = 0;
 697                                 callback_event.data.recv_str = iofd->readbuf.buffer + used_bytes;
 698                                 iohandler_log(IOLOG_DEBUG, "parsed line (%d bytes): %s", i - used_bytes, iofd->readbuf.buffer + used_bytes);
 699                                 used_bytes = i+1;
 700                                 iohandler_trigger_event(&callback_event);
 701                             } else if(i + 1 - used_bytes >= IO_LINE_LEN) { //512 max
 702                                 iofd->readbuf.buffer[i] = 0;
 703                                 callback_event.data.recv_str = iofd->readbuf.buffer + used_bytes;
 704                                 iohandler_log(IOLOG_DEBUG, "parsed and stripped line (%d bytes): %s", i - used_bytes, iofd->readbuf.buffer + used_bytes);
 705                                 for(; i < iofd->readbuf.bufpos; i++) { //skip the rest of the line
 706                                     if(iofd->readbuf.buffer[i] == '\n' || (iofd->readbuf.buffer[i] == '\r' && iofd->readbuf.buffer[i+1] != '\n')) {
 707                                         break;
 708                                     }
 709                                 }
 710                                 used_bytes = i+1;
 711                                 iohandler_trigger_event(&callback_event);
 712                             }
 713                         }
 714                         if(used_bytes) {
 715                             if(used_bytes == iofd->readbuf.bufpos) {
 716                                 iofd->readbuf.bufpos = 0;
 717                                 iohandler_log(IOLOG_DEBUG, "readbuf fully processed (set buffer position to 0)");
 718                             } else {
 719                                 iohandler_log(IOLOG_DEBUG, "readbuf rest: %d bytes (used %d bytes)", iofd->readbuf.bufpos - used_bytes, used_bytes);
 720                                 memmove(iofd->readbuf.buffer, iofd->readbuf.buffer + used_bytes, iofd->readbuf.bufpos - used_bytes);
 721                                 iofd->readbuf.bufpos -= used_bytes;
 722                             }
 723                         }
 724                         callback_event.type = IOEVENT_IGNORE;
 725                     }
 726                 } else
 727                     callback_event.type = IOEVENT_READABLE;
 728             }
 729             if(writeable) {
 730                 int bytes;
 731                 bytes = iohandler_try_write(iofd);
 732                 if(bytes < 0) {
 733                     iofd->state = IO_CLOSED;
 734                     engine->update(iofd);
 735                     callback_event.type = IOEVENT_CLOSED;
 736                     callback_event.data.errid = errno;
 737                 }
 738             }
 739             break;
 740     }
 741     if(callback_event.type == IOEVENT_IGNORE && !readable && !writeable)
 742         callback_event.type = IOEVENT_TIMEOUT;
 743     if(callback_event.type != IOEVENT_IGNORE)
 744         iohandler_trigger_event(&callback_event);
 745 }
 746
 747 void iohandler_poll() {
 748     struct timeval timeout;
 749     timeout.tv_sec = IO_MAX_TIMEOUT;
 750     timeout.tv_usec = 0;
 751     iohandler_poll_timeout(timeout);
 752 }
 753
 754 void iohandler_poll_timeout(struct timeval timeout) {
 755     if(engine) {
 756         IOSYNCHRONIZE(io_poll_sync); //quite senceless multithread support... better support will follow
 757         engine->loop(&timeout);
 758         IODESYNCHRONIZE(io_poll_sync);
 759     }
 760 }
 761
 762 //debugging functions
 763 char *iohandler_iotype_name(enum IOType type) {
 764     switch(type) {
 765         case IOTYPE_UNKNOWN:
 766             return "IOTYPE_UNKNOWN";
 767         case IOTYPE_SERVER:
 768             return "IOTYPE_SERVER";
 769         case IOTYPE_CLIENT:
 770             return "IOTYPE_CLIENT";
 771         case IOTYPE_STDIN:
 772             return "IOTYPE_STDIN";
 773         case IOTYPE_TIMER:
 774             return "IOTYPE_TIMER";
 775         default:
 776             return "(UNDEFINED)";
 777     }
 778 }
 779
 780 char *iohandler_iostatus_name(enum IOStatus status) {
 781     switch(status) {
 782         case IO_CLOSED:
 783             return "IO_CLOSED";
 784         case IO_LISTENING:
 785             return "IO_LISTENING";
 786         case IO_CONNECTING:
 787             return "IO_CONNECTING";
 788         case IO_CONNECTED:
 789             return "IO_CONNECTED";
 790         case IO_SSLWAIT:
 791             return "IO_SSLWAIT";
 792         default:
 793             return "(UNDEFINED)";
 794     }
 795 }
 796
 797 char *iohandler_ioeventtype_name(enum IOEventType type) {
 798     switch(type) {
 799         case IOEVENT_IGNORE:
 800             return "IOEVENT_IGNORE";
 801         case IOEVENT_READABLE:
 802             return "IOEVENT_READABLE";
 803         case IOEVENT_RECV:
 804             return "IOEVENT_RECV";
 805         case IOEVENT_CONNECTED:
 806             return "IOEVENT_CONNECTED";
 807         case IOEVENT_NOTCONNECTED:
 808             return "IOEVENT_NOTCONNECTED";
 809         case IOEVENT_CLOSED:
 810             return "IOEVENT_CLOSED";
 811         case IOEVENT_ACCEPT:
 812             return "IOEVENT_ACCEPT";
 813         case IOEVENT_SSLACCEPT:
 814             return "IOEVENT_SSLACCEPT";
 815         case IOEVENT_TIMEOUT:
 816             return "IOEVENT_TIMEOUT";
 817         default:
 818             return "(UNDEFINED)";
 819     }
 820 }