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 #include <utils/hashtable.h>
  30
  31 /** Intervall to check for expired sessions, in seconds */
  32 #define CLEANUP_INTERVAL 30
  33
  34 typedef struct private_dispatcher_t private_dispatcher_t;
  35
  36 /**
  37  * private data of the task manager
  38  */
  39 struct private_dispatcher_t {
  40
  41         /**
  42          * public functions
  43          */
  44         dispatcher_t public;
  45
  46         /**
  47          * fcgi socket fd
  48          */
  49         int fd;
  50
  51         /**
  52          * thread list
  53          */
  54         pthread_t *threads;
  55
  56         /**
  57          * number of threads in "threads"
  58          */
  59         int thread_count;
  60
  61         /**
  62          * session locking mutex
  63          */
  64         mutex_t *mutex;
  65
  66         /**
  67          * Hahstable with active sessions
  68          */
  69         hashtable_t *sessions;
  70
  71         /**
  72          * session timeout
  73          */
  74         time_t timeout;
  75
  76         /**
  77          * timestamp of last session cleanup round
  78          */
  79         time_t last_cleanup;
  80
  81         /**
  82          * running in debug mode?
  83          */
  84         bool debug;
  85
  86         /**
  87          * List of controllers controller_constructor_t
  88          */
  89         linked_list_t *controllers;
  90
  91         /**
  92          * List of filters filter_constructor_t
  93          */
  94         linked_list_t *filters;
  95
  96         /**
  97          * constructor function to create session context (in controller_entry_t)
  98          */
  99         context_constructor_t context_constructor;
 100
 101         /**
 102          * user param to context constructor
 103          */
 104         void *param;
 105 };
 106
 107 typedef struct {
 108         /** constructor function */
 109         controller_constructor_t constructor;
 110         /** parameter to constructor */
 111         void *param;
 112 } controller_entry_t;
 113
 114 typedef struct {
 115         /** constructor function */
 116         filter_constructor_t constructor;
 117         /** parameter to constructor */
 118         void *param;
 119 } filter_entry_t;
 120
 121 typedef struct {
 122         /** session instance */
 123         session_t *session;
 124         /** condvar to wait for session */
 125         condvar_t *cond;
 126         /** client host address, to prevent session hijacking */
 127         char *host;
 128         /** TRUE if session is in use */
 129         bool in_use;
 130         /** last use of the session */
 131         time_t used;
 132         /** has the session been closed by the handler? */
 133         bool closed;
 134 } session_entry_t;
 135
 136 /**
 137  * create a session and instanciate controllers
 138  */
 139 static session_t* load_session(private_dispatcher_t *this)
 140 {
 141         enumerator_t *enumerator;
 142         controller_entry_t *centry;
 143         filter_entry_t *fentry;
 144         session_t *session;
 145         context_t *context = NULL;
 146         controller_t *controller;
 147         filter_t *filter;
 148
 149         if (this->context_constructor)
 150         {
 151                 context = this->context_constructor(this->param);
 152         }
 153         session = session_create(context);
 154
 155         enumerator = this->controllers->create_enumerator(this->controllers);
 156         while (enumerator->enumerate(enumerator, &centry))
 157         {
 158                 controller = centry->constructor(context, centry->param);
 159                 session->add_controller(session, controller);
 160         }
 161         enumerator->destroy(enumerator);
 162
 163         enumerator = this->filters->create_enumerator(this->filters);
 164         while (enumerator->enumerate(enumerator, &fentry))
 165         {
 166                 filter = fentry->constructor(context, fentry->param);
 167                 session->add_filter(session, filter);
 168         }
 169         enumerator->destroy(enumerator);
 170
 171         return session;
 172 }
 173
 174 /**
 175  * create a new session entry
 176  */
 177 static session_entry_t *session_entry_create(private_dispatcher_t *this,
 178                                                                                          char *host)
 179 {
 180         session_entry_t *entry;
 181
 182         entry = malloc_thing(session_entry_t);
 183         entry->in_use = FALSE;
 184         entry->closed = FALSE;
 185         entry->cond = condvar_create(CONDVAR_TYPE_DEFAULT);
 186         entry->session = load_session(this);
 187         entry->used = time_monotonic(NULL);
 188         entry->host = strdup(host);
 189
 190         return entry;
 191 }
 192
 193 /**
 194  * destroy a session
 195  */
 196 static void session_entry_destroy(session_entry_t *entry)
 197 {
 198         entry->session->destroy(entry->session);
 199         entry->cond->destroy(entry->cond);
 200         free(entry->host);
 201         free(entry);
 202 }
 203
 204 /**
 205  * Implementation of dispatcher_t.add_controller.
 206  */
 207 static void add_controller(private_dispatcher_t *this,
 208                                                    controller_constructor_t constructor, void *param)
 209 {
 210         controller_entry_t *entry = malloc_thing(controller_entry_t);
 211
 212         entry->constructor = constructor;
 213         entry->param = param;
 214         this->controllers->insert_last(this->controllers, entry);
 215 }
 216
 217 /**
 218  * Implementation of dispatcher_t.add_filter.
 219  */
 220 static void add_filter(private_dispatcher_t *this,
 221                                            filter_constructor_t constructor, void *param)
 222 {
 223         filter_entry_t *entry = malloc_thing(filter_entry_t);
 224
 225         entry->constructor = constructor;
 226         entry->param = param;
 227         this->filters->insert_last(this->filters, entry);
 228 }
 229
 230 /**
 231  * Hashtable hash function
 232  */
 233 static u_int session_hash(char *sid)
 234 {
 235         return chunk_hash(chunk_create(sid, strlen(sid)));
 236 }
 237
 238 /**
 239  * Hashtable equals function
 240  */
 241 static bool session_equals(char *sid1, char *sid2)
 242 {
 243         return streq(sid1, sid2);
 244 }
 245
 246 /**
 247  * Cleanup unused sessions
 248  */
 249 static void cleanup_sessions(private_dispatcher_t *this, time_t now)
 250 {
 251         if (this->last_cleanup < now - CLEANUP_INTERVAL)
 252         {
 253                 char *sid;
 254                 session_entry_t *entry;
 255                 enumerator_t *enumerator;
 256                 linked_list_t *remove;
 257
 258                 this->last_cleanup = now;
 259                 remove = linked_list_create();
 260                 enumerator = this->sessions->create_enumerator(this->sessions);
 261                 while (enumerator->enumerate(enumerator, &sid, &entry))
 262                 {
 263                         /* check all sessions for timeout or close flag */
 264                         if (!entry->in_use &&
 265                                 (entry->used < now - this->timeout || entry->closed))
 266                         {
 267                                 remove->insert_last(remove, sid);
 268                         }
 269                 }
 270                 enumerator->destroy(enumerator);
 271
 272                 while (remove->remove_last(remove, (void**)&sid) == SUCCESS)
 273                 {
 274                         entry = this->sessions->remove(this->sessions, sid);
 275                         if (entry)
 276                         {
 277                                 session_entry_destroy(entry);
 278                         }
 279                 }
 280                 remove->destroy(remove);
 281         }
 282 }
 283
 284 /**
 285  * Actual dispatching code
 286  */
 287 static void dispatch(private_dispatcher_t *this)
 288 {
 289         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 290
 291         while (TRUE)
 292         {
 293                 request_t *request;
 294                 session_entry_t *found = NULL;
 295                 time_t now;
 296                 char *sid;
 297
 298                 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
 299                 request = request_create(this->fd, this->debug);
 300                 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 301
 302                 if (request == NULL)
 303                 {
 304                         continue;
 305                 }
 306                 now = time_monotonic(NULL);
 307                 sid = request->get_cookie(request, "SID");
 308
 309                 this->mutex->lock(this->mutex);
 310                 if (sid)
 311                 {
 312                         found = this->sessions->get(this->sessions, sid);
 313                 }
 314                 if (found && !streq(found->host, request->get_host(request)))
 315                 {
 316                         found = NULL;
 317                 }
 318                 if (found)
 319                 {
 320                         /* wait until session is unused */
 321                         while (found->in_use)
 322                         {
 323                                 found->cond->wait(found->cond, this->mutex);
 324                         }
 325                 }
 326                 else
 327                 {       /* create a new session if not found */
 328                         found = session_entry_create(this, request->get_host(request));
 329                         sid = found->session->get_sid(found->session);
 330                         this->sessions->put(this->sessions, sid, found);
 331                 }
 332                 found->in_use = TRUE;
 333                 this->mutex->unlock(this->mutex);
 334
 335                 /* start processing */
 336                 found->session->process(found->session, request);
 337                 found->used = time_monotonic(NULL);
 338
 339                 /* release session */
 340                 this->mutex->lock(this->mutex);
 341                 found->in_use = FALSE;
 342                 found->closed = request->session_closed(request);
 343                 found->cond->signal(found->cond);
 344                 cleanup_sessions(this, now);
 345                 this->mutex->unlock(this->mutex);
 346
 347                 request->destroy(request);
 348         }
 349 }
 350
 351 /**
 352  * Implementation of dispatcher_t.run.
 353  */
 354 static void run(private_dispatcher_t *this, int threads)
 355 {
 356         this->thread_count = threads;
 357         this->threads = malloc(sizeof(pthread_t) * threads);
 358         while (threads)
 359         {
 360                 if (pthread_create(&this->threads[threads - 1],
 361                                                    NULL, (void*)dispatch, this) == 0)
 362                 {
 363                         threads--;
 364                 }
 365         }
 366 }
 367
 368 /**
 369  * Implementation of dispatcher_t.waitsignal.
 370  */
 371 static void waitsignal(private_dispatcher_t *this)
 372 {
 373         sigset_t set;
 374         int sig;
 375
 376         sigemptyset(&set);
 377         sigaddset(&set, SIGINT);
 378         sigaddset(&set, SIGTERM);
 379         sigaddset(&set, SIGHUP);
 380         sigprocmask(SIG_BLOCK, &set, NULL);
 381         sigwait(&set, &sig);
 382 }
 383
 384 /**
 385  * Implementation of dispatcher_t.destroy
 386  */
 387 static void destroy(private_dispatcher_t *this)
 388 {
 389         char *sid;
 390         session_entry_t *entry;
 391         enumerator_t *enumerator;
 392
 393         FCGX_ShutdownPending();
 394         while (this->thread_count--)
 395         {
 396                 pthread_cancel(this->threads[this->thread_count]);
 397                 pthread_join(this->threads[this->thread_count], NULL);
 398         }
 399         enumerator = this->sessions->create_enumerator(this->sessions);
 400         while (enumerator->enumerate(enumerator, &sid, &entry))
 401         {
 402                 session_entry_destroy(entry);
 403         }
 404         enumerator->destroy(enumerator);
 405         this->sessions->destroy(this->sessions);
 406         this->controllers->destroy_function(this->controllers, free);
 407         this->filters->destroy_function(this->filters, free);
 408         this->mutex->destroy(this->mutex);
 409         free(this->threads);
 410         free(this);
 411 }
 412
 413 /*
 414  * see header file
 415  */
 416 dispatcher_t *dispatcher_create(char *socket, bool debug, int timeout,
 417                                                                 context_constructor_t constructor, void *param)
 418 {
 419         private_dispatcher_t *this = malloc_thing(private_dispatcher_t);
 420
 421         this->public.add_controller = (void(*)(dispatcher_t*, controller_constructor_t, void*))add_controller;
 422         this->public.add_filter = (void(*)(dispatcher_t*,filter_constructor_t constructor, void *param))add_filter;
 423         this->public.run = (void(*)(dispatcher_t*, int threads))run;
 424         this->public.waitsignal = (void(*)(dispatcher_t*))waitsignal;
 425         this->public.destroy = (void(*)(dispatcher_t*))destroy;
 426
 427         this->sessions = hashtable_create((void*)session_hash,
 428                                                                           (void*)session_equals, 4096);
 429         this->controllers = linked_list_create();
 430         this->filters = linked_list_create();
 431         this->context_constructor = constructor;
 432         this->mutex = mutex_create(MUTEX_TYPE_DEFAULT);
 433         this->param = param;
 434         this->fd = 0;
 435         this->timeout = timeout;
 436         this->last_cleanup = time_monotonic(NULL);
 437         this->debug = debug;
 438         this->threads = NULL;
 439
 440         FCGX_Init();
 441
 442         if (socket)
 443         {
 444                 unlink(socket);
 445                 this->fd = FCGX_OpenSocket(socket, 10);
 446         }
 447         return &this->public;
 448 }
 449