src/manager/lib/dispatcher.c

   1 /**
   2  * @file dispatcher.c
   3  *
   4  * @brief Implementation of dispatcher_t.
   5  *
   6  */
   7
   8 /*
   9  * Copyright (C) 2007 Martin Willi
  10  * Hochschule fuer Technik Rapperswil
  11  *
  12  * This program is free software; you can redistribute it and/or modify it
  13  * under the terms of the GNU General Public License as published by the
  14  * Free Software Foundation; either version 2 of the License, or (at your
  15  * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
  16  *
  17  * This program is distributed in the hope that it will be useful, but
  18  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  19  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  20  * for more details.
  21  */
  22
  23 #include "dispatcher.h"
  24
  25 #include "request.h"
  26 #include "session.h"
  27
  28 #include <fcgiapp.h>
  29 #include <pthread.h>
  30 #include <signal.h>
  31 #include <unistd.h>
  32
  33 #include <debug.h>
  34 #include <utils/linked_list.h>
  35
  36 typedef struct private_dispatcher_t private_dispatcher_t;
  37
  38 /**
  39  * private data of the task manager
  40  */
  41 struct private_dispatcher_t {
  42
  43         /**
  44          * public functions
  45          */
  46         dispatcher_t public;
  47
  48         /**
  49          * fcgi socket fd
  50          */
  51         int fd;
  52
  53         /**
  54          * thread list
  55          */
  56         pthread_t *threads;
  57
  58         /**
  59          * number of threads in "threads"
  60          */
  61         int thread_count;
  62
  63         /**
  64          * session locking mutex
  65          */
  66         pthread_mutex_t mutex;
  67
  68         /**
  69          * List of sessions
  70          */
  71         linked_list_t *sessions;
  72
  73         /**
  74          * session timeout
  75          */
  76         time_t timeout;
  77
  78         /**
  79          * List of controllers controller_constructor_t
  80          */
  81         linked_list_t *controllers;
  82
  83         /**
  84          * constructor function to create session context (in constructor_entry_t)
  85          */
  86         context_constructor_t context_constructor;
  87
  88         /**
  89          * user param to context constructor
  90          */
  91         void *param;
  92
  93         /**
  94          * thread specific initialization handler
  95          */
  96         void (*init)(void *param);
  97
  98         /**
  99          * argument to pass to thread intiializer
 100          */
 101         void *init_param;
 102
 103         /**
 104          * thread specific deinitialization handler
 105          */
 106         void (*deinit)(void *param);
 107
 108         /**
 109          * param tho thread specific deinitialization handler
 110          */
 111         void *deinit_param;
 112 };
 113
 114 typedef struct {
 115         /** constructor function */
 116         controller_constructor_t constructor;
 117         /** parameter to constructor */
 118         void *param;
 119 } constructor_entry_t;
 120
 121 typedef struct {
 122         /** session instance */
 123         session_t *session;
 124         /** condvar to wait for session */
 125         pthread_cond_t cond;
 126         /** TRUE if session is in use */
 127         bool in_use;
 128         /** last use of the session */
 129         time_t used;
 130 } session_entry_t;
 131
 132 /**
 133  * create a session and instanciate controllers
 134  */
 135 static session_t* load_session(private_dispatcher_t *this)
 136 {
 137         iterator_t *iterator;
 138         constructor_entry_t *entry;
 139         session_t *session;
 140         context_t *context = NULL;
 141         controller_t *controller;
 142
 143         if (this->context_constructor)
 144         {
 145                 context = this->context_constructor(this->param);
 146         }
 147         session = session_create(context);
 148
 149         iterator = this->controllers->create_iterator(this->controllers, TRUE);
 150         while (iterator->iterate(iterator, (void**)&entry))
 151         {
 152                 controller = entry->constructor(context, entry->param);
 153                 session->add_controller(session, controller);
 154         }
 155         iterator->destroy(iterator);
 156
 157         return session;
 158 }
 159
 160 /**
 161  * create a new session entry
 162  */
 163 static session_entry_t *session_entry_create(private_dispatcher_t *this)
 164 {
 165         session_entry_t *entry;
 166
 167         entry = malloc_thing(session_entry_t);
 168         entry->in_use = FALSE;
 169         pthread_cond_init(&entry->cond, NULL);
 170         entry->session = load_session(this);
 171         entry->used = time(NULL);
 172
 173         return entry;
 174 }
 175
 176 static void session_entry_destroy(session_entry_t *entry)
 177 {
 178         entry->session->destroy(entry->session);
 179         free(entry);
 180 }
 181
 182 /**
 183  * Implementation of dispatcher_t.add_controller.
 184  */
 185 static void add_controller(private_dispatcher_t *this,
 186                                                    controller_constructor_t constructor, void *param)
 187 {
 188         constructor_entry_t *entry = malloc_thing(constructor_entry_t);
 189
 190         entry->constructor = constructor;
 191         entry->param = param;
 192         this->controllers->insert_last(this->controllers, entry);
 193 }
 194
 195 /**
 196  * Actual dispatching code
 197  */
 198 static void dispatch(private_dispatcher_t *this)
 199 {
 200         FCGX_Request fcgi_req;
 201
 202         if (FCGX_InitRequest(&fcgi_req, this->fd, 0) == 0)
 203         {
 204                 while (TRUE)
 205                 {
 206                         request_t *request;
 207                         session_entry_t *current, *found = NULL;
 208                         iterator_t *iterator;
 209                         time_t now;
 210                         char *sid;
 211                         int accepted;
 212
 213                         pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
 214                         accepted = FCGX_Accept_r(&fcgi_req);
 215                         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 216
 217                         if (accepted != 0)
 218                         {
 219                                 break;
 220                         }
 221
 222                         /* prepare */
 223                         request = request_create(&fcgi_req, TRUE);
 224                         if (request == NULL)
 225                         {
 226                                 continue;
 227                         }
 228                         sid = request->get_cookie(request, "SID");
 229                         now = time(NULL);
 230
 231                         /* find session */
 232                         pthread_mutex_lock(&this->mutex);
 233                         iterator = this->sessions->create_iterator(this->sessions, TRUE);
 234                         while (iterator->iterate(iterator, (void**)&current))
 235                         {
 236                                 /* check all sessions for timeout */
 237                                 if (!current->in_use &&
 238                                         current->used < now - this->timeout)
 239                                 {
 240                                         iterator->remove(iterator);
 241                                         session_entry_destroy(current);
 242                                         continue;
 243                                 }
 244                                 if (!found && sid &&
 245                                         streq(current->session->get_sid(current->session), sid))
 246                                 {
 247                                         found = current;
 248                                 }
 249                         }
 250                         iterator->destroy(iterator);
 251
 252                         if (found)
 253                         {       /* wait until session is unused */
 254                                 while (found->in_use)
 255                                 {
 256                                         pthread_cond_wait(&found->cond, &this->mutex);
 257                                 }
 258                         }
 259                         else
 260                         {       /* create a new session if not found */
 261                                 found = session_entry_create(this);
 262                                 this->sessions->insert_first(this->sessions, found);
 263                         }
 264                         found->in_use = TRUE;
 265                         pthread_mutex_unlock(&this->mutex);
 266
 267                         /* start processing */
 268                         found->session->process(found->session, request);
 269                         found->used = time(NULL);
 270
 271                         /* release session */
 272                         pthread_mutex_lock(&this->mutex);
 273                         found->in_use = FALSE;
 274                         pthread_cond_signal(&found->cond);
 275                         pthread_mutex_unlock(&this->mutex);
 276
 277                         /* cleanup */
 278                         request->destroy(request);
 279
 280                         /*
 281                     FCGX_FPrintF(fcgi_req.out, "<ul>");
 282                     char **env = fcgi_req.envp;
 283                     while (*env)
 284                     {
 285                         FCGX_FPrintF(fcgi_req.out, "<li>%s</li>", *env);
 286                         env++;
 287                     }
 288                     FCGX_FPrintF(fcgi_req.out, "</ul>");
 289                     */
 290                 }
 291         }
 292 }
 293
 294 /**
 295  * Setup thread and start dispatching
 296  */
 297 static void start_dispatching(private_dispatcher_t *this)
 298 {
 299         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 300         if (this->init)
 301         {
 302                 this->init(this->init_param);
 303         }
 304         if (this->deinit)
 305         {
 306                 pthread_cleanup_push(this->deinit, this->deinit_param);
 307                 dispatch(this);
 308                 pthread_cleanup_pop(1);
 309         }
 310         else
 311         {
 312                 dispatch(this);
 313         }
 314 }
 315
 316 /**
 317  * Implementation of dispatcher_t.run.
 318  */
 319 static void run(private_dispatcher_t *this, int threads,
 320                                 void(*init)(void *param), void *init_param,
 321                                 void(*deinit)(void *param), void *deinit_param)
 322 {
 323         this->init = init;
 324         this->init_param = init_param;
 325         this->deinit = deinit;
 326         this->deinit_param = deinit_param;
 327         this->thread_count = threads;
 328         this->threads = malloc(sizeof(pthread_t) * threads);
 329         while (threads)
 330         {
 331                 if (pthread_create(&this->threads[threads - 1],
 332                                                    NULL, (void*)start_dispatching, this) == 0)
 333                 {
 334                         threads--;
 335                 }
 336         }
 337 }
 338
 339 /**
 340  * Implementation of dispatcher_t.waitsignal.
 341  */
 342 static void waitsignal(private_dispatcher_t *this)
 343 {
 344         sigset_t set;
 345         int sig;
 346
 347         sigemptyset(&set);
 348         sigaddset(&set, SIGINT);
 349         sigaddset(&set, SIGTERM);
 350         sigaddset(&set, SIGHUP);
 351         sigprocmask(SIG_BLOCK, &set, NULL);
 352         sigwait(&set, &sig);
 353 }
 354
 355 /**
 356  * Implementation of dispatcher_t.destroy
 357  */
 358 static void destroy(private_dispatcher_t *this)
 359 {
 360         FCGX_ShutdownPending();
 361         while (this->thread_count--)
 362         {
 363                 pthread_cancel(this->threads[this->thread_count]);
 364                 pthread_join(this->threads[this->thread_count], NULL);
 365         }
 366         this->sessions->destroy_function(this->sessions, (void*)session_entry_destroy);
 367         this->controllers->destroy_function(this->controllers, free);
 368         free(this);
 369 }
 370
 371 /*
 372  * see header file
 373  */
 374 dispatcher_t *dispatcher_create(char *socket, int timeout,
 375                                                                 context_constructor_t constructor, void *param)
 376 {
 377         private_dispatcher_t *this = malloc_thing(private_dispatcher_t);
 378
 379         this->public.add_controller = (void(*)(dispatcher_t*, controller_constructor_t, void*))add_controller;
 380         this->public.run = (void(*)(dispatcher_t*, int threads,void(*)(void *),void *,void(*)(void *),void *))run;
 381         this->public.waitsignal = (void(*)(dispatcher_t*))waitsignal;
 382         this->public.destroy = (void(*)(dispatcher_t*))destroy;
 383
 384         this->sessions = linked_list_create();
 385         this->controllers = linked_list_create();
 386         this->context_constructor = constructor;
 387         pthread_mutex_init(&this->mutex, NULL);
 388         this->param = param;
 389     this->fd = 0;
 390     this->timeout = timeout;
 391
 392     FCGX_Init();
 393
 394     if (socket)
 395     {
 396                 unlink(socket);
 397                 this->fd = FCGX_OpenSocket(socket, 10);
 398         }
 399         return &this->public;
 400 }
 401