1 /* IOHandler.c - IOMultiplexer
2 * Copyright (C) 2012 Philipp Kreil (pk910)
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.
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.
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/>.
17 #include "IOHandler.h"
19 #include "IOHandler_SSL.h"
24 #define _WIN32_WINNT 0x501
33 #include <sys/socket.h>
34 #include <netinet/in.h>
35 #include <netinet/tcp.h>
39 iohandler_log_callback *iolog_backend = NULL;
41 struct IODescriptor *first_descriptor = NULL;
42 struct IODescriptor *timer_priority = NULL;
45 static pthread_mutex_t io_thread_sync;
46 static pthread_mutex_t io_poll_sync;
49 void iohandler_log(enum IOLogType type, char *text, ...) {
51 char logBuf[MAXLOG+1];
54 va_start(arg_list, text);
55 pos = vsnprintf(logBuf, MAXLOG - 1, text, arg_list);
57 if (pos < 0 || pos > (MAXLOG - 1)) pos = MAXLOG - 1;
62 iolog_backend(type, logBuf);
66 extern struct IOEngine engine_select; /* select system call (should always be useable) */
67 extern struct IOEngine engine_kevent;
68 extern struct IOEngine engine_epoll;
69 extern struct IOEngine engine_win32;
71 struct IOEngine *engine = NULL;
73 static void iohandler_init_engine() {
75 IOTHREAD_MUTEX_INIT(io_thread_sync);
76 IOTHREAD_MUTEX_INIT(io_poll_sync);
79 if(!engine && engine_kevent.init && engine_kevent.init())
80 engine = &engine_kevent;
81 if(!engine && engine_epoll.init && engine_epoll.init())
82 engine = &engine_epoll;
83 if(!engine && engine_win32.init && engine_win32.init())
84 engine = &engine_win32;
87 if(engine_select.init())
88 engine = &engine_select;
90 iohandler_log(IOLOG_FATAL, "found no useable IO engine");
94 iohandler_log(IOLOG_DEBUG, "using %s IO engine", engine->name);
98 static void iohandler_append(struct IODescriptor *descriptor) {
99 IOSYNCHRONIZE(io_thread_sync);
100 struct timeval *timeout = ((descriptor->timeout.tv_sec || descriptor->timeout.tv_usec) ? &descriptor->timeout : NULL);
102 struct IODescriptor *iofd;
103 int set_priority = 1;
104 descriptor->timeout = *timeout;
106 iofd = timer_priority;
108 iofd = first_descriptor;
110 for(;;iofd = iofd->next) {
111 if(timeval_is_smaler(timeout, (&iofd->timeout))) {
112 descriptor->prev = iofd->prev;
113 descriptor->next = iofd;
114 iofd->prev = descriptor;
116 iofd->prev->next = descriptor;
118 timer_priority = descriptor;
121 if(iofd == timer_priority)
123 if(iofd->next == NULL) {
124 descriptor->next = NULL;
125 descriptor->prev = iofd;
126 iofd->next = descriptor;
128 timer_priority = descriptor;
133 descriptor->prev = NULL;
134 descriptor->next = NULL;
135 first_descriptor = descriptor;
136 timer_priority = descriptor;
140 descriptor->prev = NULL;
141 descriptor->next = first_descriptor;
143 first_descriptor->prev = descriptor;
144 first_descriptor = descriptor;
146 IODESYNCHRONIZE(io_thread_sync);
149 static void iohandler_remove(struct IODescriptor *descriptor, int engine_remove) {
150 //remove IODescriptor from the list
151 IOSYNCHRONIZE(io_thread_sync);
153 descriptor->prev->next = descriptor->next;
155 first_descriptor = descriptor->next;
157 descriptor->next->prev = descriptor->prev;
158 if(descriptor == timer_priority)
159 timer_priority = descriptor->next;
162 engine->remove(descriptor);
163 if(descriptor->readbuf.buffer)
164 free(descriptor->readbuf.buffer);
165 if(descriptor->writebuf.buffer)
166 free(descriptor->writebuf.buffer);
167 iohandler_log(IOLOG_DEBUG, "removed IODescriptor (%d) of type `%s`", descriptor->fd, iohandler_iotype_name(descriptor->type));
169 IODESYNCHRONIZE(io_thread_sync);
172 struct IODescriptor *iohandler_add(int sockfd, enum IOType type, struct timeval *timeout, iohandler_callback *callback) {
173 //just add a new IODescriptor
174 struct IODescriptor *descriptor = calloc(1, sizeof(*descriptor));
176 iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
179 descriptor->fd = (type == IOTYPE_STDIN ? fileno(stdin) : sockfd);
180 descriptor->type = type;
181 descriptor->state = (type == IOTYPE_STDIN ? IO_CONNECTED : IO_CLOSED);
182 descriptor->callback = callback;
184 descriptor->timeout = *timeout;
185 if(type != IOTYPE_TIMER) {
186 descriptor->readbuf.buffer = malloc(IO_READ_BUFLEN + 2);
187 descriptor->readbuf.bufpos = 0;
188 descriptor->readbuf.buflen = IO_READ_BUFLEN;
189 descriptor->writebuf.buffer = malloc(IO_READ_BUFLEN + 2);
190 descriptor->writebuf.bufpos = 0;
191 descriptor->writebuf.buflen = IO_READ_BUFLEN;
195 iohandler_init_engine();
200 engine->add(descriptor);
202 //add IODescriptor to the list
203 iohandler_append(descriptor);
205 iohandler_log(IOLOG_DEBUG, "added custom socket descriptor (%d) as type `%s`", sockfd, iohandler_iotype_name(type));
209 void iohandler_set_timeout(struct IODescriptor *descriptor, struct timeval *timeout) {
211 descriptor->prev->next = descriptor->next;
213 first_descriptor = descriptor->next;
215 descriptor->next->prev = descriptor->prev;
216 if(descriptor == timer_priority)
217 timer_priority = descriptor->next;
219 descriptor->timeout = *timeout;
221 descriptor->timeout.tv_sec = 0;
222 descriptor->timeout.tv_usec = 0;
224 iohandler_append(descriptor);
227 static void iohandler_increase_iobuf(struct IOBuffer *iobuf, size_t required) {
228 if(iobuf->buflen >= required) return;
229 char *new_buf = realloc(iobuf->buffer, required + 2);
231 iobuf->buffer = new_buf;
232 iobuf->buflen = required;
236 struct IODescriptor *iohandler_timer(struct timeval timeout, iohandler_callback *callback) {
237 struct IODescriptor *descriptor;
238 descriptor = iohandler_add(-1, IOTYPE_TIMER, &timeout, callback);
240 iohandler_log(IOLOG_ERROR, "could not allocate memory for IODescriptor in %s:%d", __FILE__, __LINE__);
243 iohandler_log(IOLOG_DEBUG, "added timer descriptor (sec: %d; usec: %d)", timeout.tv_sec, timeout.tv_usec);
247 struct IODescriptor *iohandler_connect(const char *hostname, unsigned int port, int ssl, const char *bindhost, iohandler_callback *callback) {
248 //non-blocking connect
250 struct addrinfo hints, *res;
251 struct sockaddr_in *ip4 = NULL;
252 struct sockaddr_in6 *ip6 = NULL;
254 struct sockaddr *dstaddr = NULL;
255 struct IODescriptor *descriptor;
258 iohandler_init_engine();
259 if(!engine) return NULL;
261 memset (&hints, 0, sizeof (hints));
262 hints.ai_family = PF_UNSPEC;
263 hints.ai_socktype = SOCK_STREAM;
264 hints.ai_flags |= AI_CANONNAME;
265 if ((result = getaddrinfo (hostname, NULL, &hints, &res))) {
266 iohandler_log(IOLOG_ERROR, "could not resolve %s to an IP address (%d)", hostname, result);
270 switch (res->ai_family) {
272 ip4 = (struct sockaddr_in *) res->ai_addr;
275 ip6 = (struct sockaddr_in6 *) res->ai_addr;
283 sockfd = socket(AF_INET6, SOCK_STREAM, 0);
285 iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
289 ip6->sin6_family = AF_INET6;
290 ip6->sin6_port = htons(port);
292 struct sockaddr_in6 *ip6vhost = NULL;
293 if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
295 switch (res->ai_family) {
297 ip6vhost = (struct sockaddr_in6 *) res->ai_addr;
305 ip6vhost->sin6_family = AF_INET6;
306 ip6vhost->sin6_port = htons(0);
307 bind(sockfd, (struct sockaddr*)ip6vhost, sizeof(*ip6vhost));
309 dstaddr = (struct sockaddr*)ip6;
310 dstaddrlen = sizeof(*ip6);
312 sockfd = socket(AF_INET, SOCK_STREAM, 0);
314 iohandler_log(IOLOG_ERROR, "could not create socket in %s:%d", __FILE__, __LINE__);
318 ip4->sin_family = AF_INET;
319 ip4->sin_port = htons(port);
321 struct sockaddr_in *ip4vhost = NULL;
322 if (bindhost && !getaddrinfo(bindhost, NULL, &hints, &res)) {
324 switch (res->ai_family) {
326 ip4vhost = (struct sockaddr_in *) res->ai_addr;
334 ip4vhost->sin_family = AF_INET;
335 ip4vhost->sin_port = htons(0);
336 bind(sockfd, (struct sockaddr*)ip4vhost, sizeof(*ip4vhost));
338 dstaddr = (struct sockaddr*)ip4;
339 dstaddrlen = sizeof(*ip4);
342 //make sockfd unblocking
346 flags = fcntl(sockfd, F_GETFL);
347 fcntl(sockfd, F_SETFL, flags|O_NONBLOCK);
348 flags = fcntl(sockfd, F_GETFD);
349 fcntl(sockfd, F_SETFD, flags|FD_CLOEXEC);
352 /* I hope you're using the Win32 backend or something else that
353 * automatically marks the file descriptor non-blocking...
356 descriptor = iohandler_add(sockfd, IOTYPE_CLIENT, NULL, callback);
361 connect(sockfd, dstaddr, dstaddrlen); //returns EINPROGRESS here (nonblocking)
362 descriptor->state = IO_CONNECTING;
363 descriptor->ssl = (ssl ? 1 : 0);
364 descriptor->read_lines = 1;
365 engine->update(descriptor);
366 iohandler_log(IOLOG_DEBUG, "added client socket (%d) connecting to %s:%d", sockfd, hostname, port);
370 struct IODescriptor *iohandler_listen(const char *hostname, unsigned int port, iohandler_callback *callback) {
372 struct addrinfo hints, *res;
373 struct sockaddr_in *ip4 = NULL;
374 struct sockaddr_in6 *ip6 = NULL;
375 struct IODescriptor *descriptor;
379 iohandler_init_engine();
380 if(!engine) return NULL;
382 memset (&hints, 0, sizeof (hints));
383 hints.ai_family = PF_UNSPEC;
384 hints.ai_socktype = SOCK_STREAM;
385 hints.ai_flags |= AI_CANONNAME;
386 if (getaddrinfo (hostname, NULL, &hints, &res)) {
390 switch (res->ai_family) {
392 ip4 = (struct sockaddr_in *) res->ai_addr;
395 ip6 = (struct sockaddr_in6 *) res->ai_addr;
403 sockfd = socket(AF_INET6, SOCK_STREAM, 0);
404 if(sockfd == -1) return NULL;
407 setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
409 ip6->sin6_family = AF_INET6;
410 ip6->sin6_port = htons(port);
412 bind(sockfd, (struct sockaddr*)ip6, sizeof(*ip6));
414 sockfd = socket(AF_INET, SOCK_STREAM, 0);
415 if(sockfd == -1) return NULL;
418 setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
420 ip4->sin_family = AF_INET;
421 ip4->sin_port = htons(port);
423 bind(sockfd, (struct sockaddr*)ip4, sizeof(*ip4));
426 //make sockfd unblocking
430 flags = fcntl(sockfd, F_GETFL);
431 fcntl(sockfd, F_SETFL, flags|O_NONBLOCK);
432 flags = fcntl(sockfd, F_GETFD);
433 fcntl(sockfd, F_SETFD, flags|FD_CLOEXEC);
436 /* I hope you're using the Win32 backend or something else that
437 * automatically marks the file descriptor non-blocking...
440 descriptor = iohandler_add(sockfd, IOTYPE_SERVER, NULL, callback);
446 descriptor->state = IO_LISTENING;
447 engine->update(descriptor);
448 iohandler_log(IOLOG_DEBUG, "added server socket (%d) listening on %s:%d", sockfd, hostname, port);
452 void iohandler_write(struct IODescriptor *iofd, const char *line) {
453 size_t linelen = strlen(line);
454 iohandler_send(iofd, line, linelen);
457 void iohandler_send(struct IODescriptor *iofd, const char *data, size_t datalen) {
458 if(iofd->type == IOTYPE_TIMER || iofd->state == IO_CLOSED) {
459 iohandler_log(IOLOG_ERROR, "could not write to socket (%s)", (iofd->type == IOTYPE_TIMER ? "IOTYPE_TIMER" : "IO_CLOSED"));
462 iohandler_log(IOLOG_DEBUG, "add %d to writebuf (fd: %d): %s", datalen, iofd->fd, data);
463 if(iofd->writebuf.buflen < iofd->writebuf.bufpos + datalen) {
464 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));
465 iohandler_increase_iobuf(&iofd->writebuf, iofd->writebuf.bufpos + datalen);
466 if(iofd->writebuf.buflen < iofd->writebuf.bufpos + datalen) {
467 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));
471 memcpy(iofd->writebuf.buffer + iofd->writebuf.bufpos, data, datalen);
472 iofd->writebuf.bufpos += datalen;
473 engine->update(iofd);
476 void iohandler_printf(struct IODescriptor *iofd, const char *text, ...) {
478 char sendBuf[IO_LINE_LEN];
481 va_start(arg_list, text);
482 pos = vsnprintf(sendBuf, IO_LINE_LEN - 2, text, arg_list);
484 if (pos < 0 || pos > (IO_LINE_LEN - 2)) pos = IO_LINE_LEN - 2;
486 sendBuf[pos+1] = '\0';
487 iohandler_send(iofd, sendBuf, pos+1);
490 void iohandler_try_write(struct IODescriptor *iofd) {
491 if(!iofd->writebuf.bufpos) return;
492 iohandler_log(IOLOG_DEBUG, "write writebuf (%d bytes) to socket (fd: %d)", iofd->writebuf.bufpos, iofd->fd);
495 res = iohandler_ssl_write(iofd, iofd->writebuf.buffer, iofd->writebuf.bufpos);
497 res = send(iofd->fd, iofd->writebuf.buffer, iofd->writebuf.bufpos, 0);
499 if (errno != EAGAIN) {
500 iohandler_log(IOLOG_ERROR, "could not write to socket (fd: %d): %d - %s", iofd->fd, errno, strerror(errno));
503 iofd->writebuf.bufpos -= res;
504 if(iofd->state != IO_CLOSED)
505 engine->update(iofd);
509 void iohandler_close(struct IODescriptor *iofd) {
510 int engine_remove = 1;
511 iofd->state = IO_CLOSED;
512 if(iofd->writebuf.bufpos) {
513 //try to send everything before closing
517 flags = fcntl(iofd->fd, F_GETFL);
518 fcntl(iofd->fd, F_SETFL, flags & ~O_NONBLOCK);
519 flags = fcntl(iofd->fd, F_GETFD);
520 fcntl(iofd->fd, F_SETFD, flags|FD_CLOEXEC);
524 engine->remove(iofd);
526 iohandler_try_write(iofd);
530 iohandler_ssl_disconnect(iofd);
531 if(iofd->type == IOTYPE_SERVER || iofd->type == IOTYPE_CLIENT || iofd->type == IOTYPE_STDIN)
533 iohandler_remove(iofd, engine_remove);
536 void iohandler_update(struct IODescriptor *iofd) {
537 iohandler_log(IOLOG_DEBUG, "external call to iohandler_update (fd: %d)", iofd->fd);
538 engine->update(iofd);
541 static void iohandler_trigger_event(struct IOEvent *event) {
542 if(!event->iofd->callback) return;
543 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);
544 event->iofd->callback(event);
547 void iohandler_events(struct IODescriptor *iofd, int readable, int writeable) {
548 struct IOEvent callback_event;
549 callback_event.type = IOEVENT_IGNORE;
550 callback_event.iofd = iofd;
551 switch(iofd->state) {
553 if(!readable && !writeable) {
554 callback_event.type = IOEVENT_SSLFAILED;
555 iofd->state = IO_CLOSED;
557 iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_client_handshake for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
558 iohandler_ssl_client_handshake(iofd);
562 if(iofd->type == IOTYPE_TIMER)
563 callback_event.type = IOEVENT_TIMEOUT;
567 callback_event.data.accept_fd = accept(iofd->fd, NULL, 0);
568 if(callback_event.data.accept_fd < 0) {
569 iohandler_log(IOLOG_ERROR, "could not accept client (server fd: %d): %d - %s", iofd->fd, errno, strerror(errno));
571 callback_event.type = IOEVENT_ACCEPT;
575 if(readable) { //could not connect
576 callback_event.type = IOEVENT_NOTCONNECTED;
578 //arglen = sizeof(callback_event.data.errid);
579 //if (getsockopt(iofd->fd, SOL_SOCKET, SO_ERROR, &callback_event.data.errid, &arglen) < 0)
580 // callback_event.data.errid = errno;
581 iofd->state = IO_CLOSED;
582 engine->update(iofd);
583 } else if(writeable) {
584 if(iofd->ssl && !iofd->ssl_active) {
585 iohandler_log(IOLOG_DEBUG, "triggering iohandler_ssl_connect for %s (fd: %d)", iohandler_iotype_name(iofd->type), iofd->fd);
586 iohandler_ssl_connect(iofd);
589 callback_event.type = IOEVENT_CONNECTED;
590 iofd->state = IO_CONNECTED;
591 engine->update(iofd);
596 if(iofd->read_lines) {
600 bytes = iohandler_ssl_read(iofd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos);
602 if(iofd->type == IOTYPE_STDIN)
606 bytes = read(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos);
609 bytes = recv(iofd->fd, iofd->readbuf.buffer + iofd->readbuf.bufpos, iofd->readbuf.buflen - iofd->readbuf.bufpos, 0);
612 if (errno != EAGAIN) {
613 iofd->state = IO_CLOSED;
614 engine->update(iofd);
615 callback_event.type = IOEVENT_CLOSED;
616 callback_event.data.errid = errno;
619 int i, used_bytes = 0;
620 iohandler_log(IOLOG_DEBUG, "received %d bytes (fd: %d). readbuf position: %d", bytes, iofd->fd, iofd->readbuf.bufpos);
621 iofd->readbuf.bufpos += bytes;
622 callback_event.type = IOEVENT_RECV;
623 for(i = 0; i < iofd->readbuf.bufpos; i++) {
624 if(iofd->readbuf.buffer[i] == '\r' && iofd->readbuf.buffer[i+1] == '\n')
625 iofd->readbuf.buffer[i] = 0;
626 else if(iofd->readbuf.buffer[i] == '\n' || iofd->readbuf.buffer[i] == '\r') {
627 iofd->readbuf.buffer[i] = 0;
628 callback_event.data.recv_str = iofd->readbuf.buffer + used_bytes;
629 iohandler_log(IOLOG_DEBUG, "parsed line (%d bytes): %s", i - used_bytes, iofd->readbuf.buffer + used_bytes);
631 iohandler_trigger_event(&callback_event);
632 } else if(i + 1 - used_bytes >= IO_LINE_LEN) { //512 max
633 iofd->readbuf.buffer[i] = 0;
634 callback_event.data.recv_str = iofd->readbuf.buffer + used_bytes;
635 iohandler_log(IOLOG_DEBUG, "parsed and stripped line (%d bytes): %s", i - used_bytes, iofd->readbuf.buffer + used_bytes);
636 for(; i < iofd->readbuf.bufpos; i++) { //skip the rest of the line
637 if(iofd->readbuf.buffer[i] == '\n' || (iofd->readbuf.buffer[i] == '\r' && iofd->readbuf.buffer[i+1] != '\n')) {
642 iohandler_trigger_event(&callback_event);
646 if(used_bytes == iofd->readbuf.bufpos) {
647 iofd->readbuf.bufpos = 0;
648 iohandler_log(IOLOG_DEBUG, "readbuf fully processed (set buffer position to 0)");
650 iohandler_log(IOLOG_DEBUG, "readbuf rest: %d bytes (used %d bytes)", iofd->readbuf.bufpos - used_bytes, used_bytes);
651 memmove(iofd->readbuf.buffer, iofd->readbuf.buffer + used_bytes, iofd->readbuf.bufpos - used_bytes);
652 iofd->readbuf.bufpos -= used_bytes;
655 callback_event.type = IOEVENT_IGNORE;
658 callback_event.type = IOEVENT_READABLE;
661 iohandler_try_write(iofd);
665 if(callback_event.type == IOEVENT_IGNORE && !readable && !writeable)
666 callback_event.type = IOEVENT_TIMEOUT;
667 if(callback_event.type != IOEVENT_IGNORE)
668 iohandler_trigger_event(&callback_event);
671 void iohandler_poll() {
673 IOSYNCHRONIZE(io_poll_sync); //quite senceless multithread support... better support will follow
674 struct timeval timeout;
675 timeout.tv_sec = IO_MAX_TIMEOUT;
677 engine->loop(&timeout);
678 IODESYNCHRONIZE(io_poll_sync);
682 //debugging functions
683 char *iohandler_iotype_name(enum IOType type) {
686 return "IOTYPE_UNKNOWN";
688 return "IOTYPE_SERVER";
690 return "IOTYPE_CLIENT";
692 return "IOTYPE_STDIN";
694 return "IOTYPE_TIMER";
696 return "(UNDEFINED)";
700 char *iohandler_iostatus_name(enum IOStatus status) {
705 return "IO_LISTENING";
707 return "IO_CONNECTING";
709 return "IO_CONNECTED";
713 return "(UNDEFINED)";
717 char *iohandler_ioeventtype_name(enum IOEventType type) {
720 return "IOEVENT_IGNORE";
721 case IOEVENT_READABLE:
722 return "IOEVENT_READABLE";
724 return "IOEVENT_RECV";
725 case IOEVENT_CONNECTED:
726 return "IOEVENT_CONNECTED";
727 case IOEVENT_NOTCONNECTED:
728 return "IOEVENT_NOTCONNECTED";
730 return "IOEVENT_CLOSED";
732 return "IOEVENT_ACCEPT";
733 case IOEVENT_TIMEOUT:
734 return "IOEVENT_TIMEOUT";
736 return "(UNDEFINED)";