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