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