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