src/IOHandler.c

   1 /* IOHandler.c - IOMultiplexer v2
   2  * Copyright (C) 2014  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 #define _IOHandler_internals
  18 #include "IOInternal.h"
  19 #include "IOHandler.h"
  20
  21 #include "IOLog.h"
  22 #include "IOGarbageCollector.h"
  23 #include "IOTimer.h"
  24 #include "IODNSLookup.h"
  25
  26 /* compat */
  27 #include "compat/usleep.c"
  28
  29 #ifdef HAVE_PTHREAD_H
  30 static pthread_mutex_t iothread_sync;
  31 #ifdef WIN32
  32 #define pthread_self_tid() pthread_self().p
  33 #else
  34 #define pthread_self_tid() pthread_self()
  35 #endif
  36
  37 #endif
  38
  39 static struct IOHandlerThread {
  40         unsigned int id;
  41         unsigned int main : 1;
  42         unsigned int run : 1;
  43         unsigned int shutdown : 1;
  44         #ifdef HAVE_PTHREAD_H
  45         static pthread_t *thread;
  46         #endif
  47         struct IOHandlerThread *next;
  48 }
  49
  50 static int iohandler_running = 0;
  51 static int iohandler_treads = 1;
  52 static struct IOHandlerThread *threads;
  53
  54 void iohandler_init() {
  55         IOTHREAD_MUTEX_INIT(iothread_sync);
  56
  57         iolog_init();
  58         iogc_init();
  59
  60         _init_timers();
  61         _init_iodns();
  62         _init_sockets();
  63
  64         iohandler_running = 1;
  65 }
  66
  67 void iohandler_set_threads(int threadcount) {
  68         iohandler_treads = threadcount;
  69 }
  70
  71 void iohandler_stop() {
  72         iohandler_treads = 0;
  73 }
  74
  75 #ifdef HAVE_PTHREAD_H
  76 static void iohandler_start_worker() {
  77         struct IOHandlerThread *thread = calloc(1, sizeof(*thread));
  78     if(!thread) {
  79         iolog_trigger(IOLOG_ERROR, "could not allocate memory for IOHandlerThread in %s:%d", __FILE__, __LINE__);
  80         return;
  81     }
  82         struct IOHandlerThread *cthread;
  83         for(cthread = threads; cthread; cthread = cthread->next) {
  84                 if(cthread->next == NULL) {
  85                         cthread->next = thread;
  86                         break;
  87                 }
  88         }
  89
  90         thread->run = 1;
  91
  92         int thread_err;
  93         thread_err = pthread_create(&thread->thread, NULL, iohandler_worker, thread);
  94         if(thread_err) {
  95                 cthread->next = NULL;
  96                 iolog_trigger(IOLOG_ERROR, "could not create pthread in %s:%d (Returned: %i)", __FILE__, __LINE__, thread_err);
  97         }
  98 }
  99 #endif
 100
 101 static void iohandler_worker(void *tptr) {
 102         struct IOHandlerThread *thread = tptr;
 103
 104         #ifdef HAVE_PTHREAD_H
 105         if(!thread->main) {
 106                 thread->id = pthread_self_tid();
 107         }
 108         #endif
 109
 110         while(!thread->shutdown) { // endless loop
 111                 if(thread->main && iohandler_treads != iohandler_running) {
 112                         IOSYNCHRONIZE(iothread_sync);
 113                         #ifdef HAVE_PTHREAD_H
 114                         int i;
 115                         if(iohandler_treads > iohandler_running) {
 116                                 for(i = 0; i < (iohandler_treads - iohandler_running); i++)
 117                                         iohandler_start_worker();
 118                         }
 119                         if(iohandler_treads < iohandler_running) {
 120                                 struct IOHandlerThread *cthread;
 121                                 for(i = 0; i < (iohandler_running - iohandler_treads); i++) {
 122                                         for(cthread = threads; cthread; cthread = cthread->next) {
 123                                                 if(cthread->main)
 124                                                         continue;
 125                                                 cthread->shutdown = 1;
 126                                                 iolog_trigger(IOLOG_ERROR, "Thread %d marked for shutdown.", cthread->id);
 127                                         }
 128                                         if(cthread)
 129                                                 iohandler_running--;
 130                                 }
 131                         }
 132                         #endif
 133                         if(iohandler_treads == 0) {
 134                                 #ifdef HAVE_PTHREAD_H
 135                                 struct IOHandlerThread *cthread;
 136                                 for(cthread = threads; cthread; cthread = cthread->next) {
 137                                         if(cthread->main)
 138                                                 continue;
 139                                         cthread->shutdown = 1;
 140                                         pthread_join(cthread->thread, NULL);
 141                                 }
 142                                 #endif
 143                                 thread->shutdown = 1;
 144                                 IODESYNCHRONIZE(iothread_sync);
 145                                 break;
 146                         }
 147                         IODESYNCHRONIZE(iothread_sync);
 148                 }
 149                 if(!thread->run) {
 150                         usleep(500000); // 500ms
 151                         continue;
 152                 }
 153
 154                 // iohandler calls
 155                 iogc_exec();
 156                 iodns_poll();
 157
 158         }
 159         IOSYNCHRONIZE(iothread_sync);
 160         if(thread == threads) {
 161                 threads = thread->next;
 162         } else {
 163                 struct IOHandlerThread *cthread;
 164                 for(cthread = threads; cthread; cthread = cthread->next) {
 165                         if(cthread->next == thread) {
 166                                 cthread->next = thread->next;
 167                                 break;
 168                         }
 169                 }
 170         }
 171         iolog_trigger(IOLOG_DEBUG, "Thread %d stopped.", thread->id);
 172         free(thread);
 173         IODESYNCHRONIZE(iothread_sync);
 174 }
 175
 176 void iohandler_run() {
 177         if(!iohandler_running)
 178                 return;
 179         iohandler_running = 1;
 180
 181         struct IOHandlerThread *mainthread = calloc(1, sizeof(*mainthread));
 182     if(!mainthread) {
 183         iolog_trigger(IOLOG_ERROR, "could not allocate memory for IOHandlerThread in %s:%d", __FILE__, __LINE__);
 184         return;
 185     }
 186         threads = mainthread;
 187
 188         mainthread->main = 1;
 189         mainthread->run = 1;
 190         mainthread->shutdown = 0;
 191
 192         iohandler_worker(1);
 193
 194         _stop_iodns(); /* possible worker thread */
 195 }
 196