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