src/libfast/dispatcher.c

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