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
  63         iohandler_running = 1;
  64 }
  65
  66 void iohandler_set_threads(int threadcount) {
  67         iohandler_treads = threadcount;
  68 }
  69
  70 void iohandler_stop() {
  71         iohandler_treads = 0;
  72 }
  73
  74 #ifdef HAVE_PTHREAD_H
  75 static void iohandler_start_worker() {
  76         struct IOHandlerThread *thread = calloc(1, sizeof(*thread));
  77     if(!thread) {
  78         iolog_trigger(IOLOG_ERROR, "could not allocate memory for IOHandlerThread in %s:%d", __FILE__, __LINE__);
  79         return;
  80     }
  81         struct IOHandlerThread *cthread;
  82         for(cthread = threads; cthread; cthread = cthread->next) {
  83                 if(cthread->next == NULL) {
  84                         cthread->next = thread;
  85                         break;
  86                 }
  87         }
  88
  89         thread->run = 1;
  90
  91         int thread_err;
  92         thread_err = pthread_create(&thread->thread, NULL, iohandler_worker, thread);
  93         if(thread_err) {
  94                 cthread->next = NULL;
  95                 iolog_trigger(IOLOG_ERROR, "could not create pthread in %s:%d (Returned: %i)", __FILE__, __LINE__, thread_err);
  96         }
  97 }
  98 #endif
  99
 100 static void iohandler_worker(void *tptr) {
 101         struct IOHandlerThread *thread = tptr;
 102
 103         #ifdef HAVE_PTHREAD_H
 104         if(!thread->main) {
 105                 thread->id = pthread_self_tid();
 106         }
 107         #endif
 108
 109         while(!thread->shutdown) { // endless loop
 110                 if(thread->main && iohandler_treads != iohandler_running) {
 111                         IOSYNCHRONIZE(iothread_sync);
 112                         #ifdef HAVE_PTHREAD_H
 113                         int i;
 114                         if(iohandler_treads > iohandler_running) {
 115                                 for(i = 0; i < (iohandler_treads - iohandler_running); i++)
 116                                         iohandler_start_worker();
 117                         }
 118                         if(iohandler_treads < iohandler_running) {
 119                                 struct IOHandlerThread *cthread;
 120                                 for(i = 0; i < (iohandler_running - iohandler_treads); i++) {
 121                                         for(cthread = threads; cthread; cthread = cthread->next) {
 122                                                 if(cthread->main)
 123                                                         continue;
 124                                                 cthread->shutdown = 1;
 125                                                 iolog_trigger(IOLOG_ERROR, "Thread %d marked for shutdown.", cthread->id);
 126                                         }
 127                                         if(cthread)
 128                                                 iohandler_running--;
 129                                 }
 130                         }
 131                         #endif
 132                         if(iohandler_treads == 0) {
 133                                 #ifdef HAVE_PTHREAD_H
 134                                 struct IOHandlerThread *cthread;
 135                                 for(cthread = threads; cthread; cthread = cthread->next) {
 136                                         if(cthread->main)
 137                                                 continue;
 138                                         cthread->shutdown = 1;
 139                                         pthread_join(cthread->thread, NULL);
 140                                 }
 141                                 #endif
 142                                 thread->shutdown = 1;
 143                                 IODESYNCHRONIZE(iothread_sync);
 144                                 break;
 145                         }
 146                         IODESYNCHRONIZE(iothread_sync);
 147                 }
 148                 if(!thread->run) {
 149                         usleep(500000); // 500ms
 150                         continue;
 151                 }
 152
 153                 // iohandler calls
 154                 iogc_exec();
 155                 iodns_poll();
 156
 157         }
 158         IOSYNCHRONIZE(iothread_sync);
 159         if(thread == threads) {
 160                 threads = thread->next;
 161         } else {
 162                 struct IOHandlerThread *cthread;
 163                 for(cthread = threads; cthread; cthread = cthread->next) {
 164                         if(cthread->next == thread) {
 165                                 cthread->next = thread->next;
 166                                 break;
 167                         }
 168                 }
 169         }
 170         iolog_trigger(IOLOG_DEBUG, "Thread %d stopped.", thread->id);
 171         free(thread);
 172         IODESYNCHRONIZE(iothread_sync);
 173 }
 174
 175 void iohandler_run() {
 176         if(!iohandler_running)
 177                 return;
 178         iohandler_running = 1;
 179
 180         struct IOHandlerThread *mainthread = calloc(1, sizeof(*mainthread));
 181     if(!mainthread) {
 182         iolog_trigger(IOLOG_ERROR, "could not allocate memory for IOHandlerThread in %s:%d", __FILE__, __LINE__);
 183         return;
 184     }
 185         threads = mainthread;
 186
 187         mainthread->main = 1;
 188         mainthread->run = 1;
 189         mainthread->shutdown = 0;
 190
 191         iohandler_worker(1);
 192
 193         _stop_iodns(); /* possible worker thread */
 194 }
 195