src/libstrongswan/processing/jobs/callback_job.c

   1 /*
   2  * Copyright (C) 2009 Tobias Brunner
   3  * Copyright (C) 2007-2011 Martin Willi
   4  * Copyright (C) 2011 revosec AG
   5  * Hochschule fuer Technik Rapperswil
   6  *
   7  * This program is free software; you can redistribute it and/or modify it
   8  * under the terms of the GNU General Public License as published by the
   9  * Free Software Foundation; either version 2 of the License, or (at your
  10  * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  14  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  15  * for more details.
  16  */
  17
  18 #include "callback_job.h"
  19
  20 #include <semaphore.h>
  21
  22 #include <threading/thread.h>
  23 #include <threading/condvar.h>
  24 #include <threading/mutex.h>
  25 #include <utils/linked_list.h>
  26
  27 typedef struct private_callback_job_t private_callback_job_t;
  28
  29 /**
  30  * Private data of an callback_job_t Object.
  31  */
  32 struct private_callback_job_t {
  33
  34         /**
  35          * Public callback_job_t interface.
  36          */
  37         callback_job_t public;
  38
  39         /**
  40          * Callback to call on execution
  41          */
  42         callback_job_cb_t callback;
  43
  44         /**
  45          * parameter to supply to callback
  46          */
  47         void *data;
  48
  49         /**
  50          * cleanup function for data
  51          */
  52         callback_job_cleanup_t cleanup;
  53
  54         /**
  55          * thread of the job, if running
  56          */
  57         thread_t *thread;
  58
  59         /**
  60          * mutex to access jobs interna
  61          */
  62         mutex_t *mutex;
  63
  64         /**
  65          * list of asociated child jobs
  66          */
  67         linked_list_t *children;
  68
  69         /**
  70          * parent of this job, or NULL
  71          */
  72         private_callback_job_t *parent;
  73
  74         /**
  75          * TRUE if the job got cancelled
  76          */
  77         bool cancelled;
  78
  79         /**
  80          * condvar to synchronize the cancellation/destruction of the job
  81          */
  82         condvar_t *destroyable;
  83
  84         /**
  85          * semaphore to synchronize the termination of the assigned thread.
  86          *
  87          * separately allocated during cancellation, so that we can wait on it
  88          * without risking that it gets freed too early during destruction.
  89          */
  90         sem_t *terminated;
  91 };
  92
  93 /**
  94  * unregister a child from its parent, if any.
  95  * note: this->mutex has to be locked
  96  */
  97 static void unregister(private_callback_job_t *this)
  98 {
  99         if (this->parent)
 100         {
 101                 this->parent->mutex->lock(this->parent->mutex);
 102                 if (this->parent->cancelled && !this->cancelled)
 103                 {
 104                         /* if the parent has been cancelled but we have not yet, we do not
 105                          * unregister until we got cancelled by the parent. */
 106                         this->parent->mutex->unlock(this->parent->mutex);
 107                         this->destroyable->wait(this->destroyable, this->mutex);
 108                         this->parent->mutex->lock(this->parent->mutex);
 109                 }
 110                 this->parent->children->remove(this->parent->children, this, NULL);
 111                 this->parent->mutex->unlock(this->parent->mutex);
 112                 this->parent = NULL;
 113         }
 114 }
 115
 116 METHOD(job_t, destroy, void,
 117         private_callback_job_t *this)
 118 {
 119         this->mutex->lock(this->mutex);
 120         unregister(this);
 121         if (this->cleanup)
 122         {
 123                 this->cleanup(this->data);
 124         }
 125         if (this->terminated)
 126         {
 127                 sem_post(this->terminated);
 128         }
 129         this->children->destroy(this->children);
 130         this->destroyable->destroy(this->destroyable);
 131         this->mutex->unlock(this->mutex);
 132         this->mutex->destroy(this->mutex);
 133         free(this);
 134 }
 135
 136 METHOD(callback_job_t, cancel, void,
 137         private_callback_job_t *this)
 138 {
 139         callback_job_t *child;
 140         sem_t *terminated = NULL;
 141
 142         this->mutex->lock(this->mutex);
 143         this->cancelled = TRUE;
 144         /* terminate children */
 145         while (this->children->get_first(this->children, (void**)&child) == SUCCESS)
 146         {
 147                 this->mutex->unlock(this->mutex);
 148                 child->cancel(child);
 149                 this->mutex->lock(this->mutex);
 150         }
 151         if (this->thread)
 152         {
 153                 /* terminate the thread, if there is currently one executing the job.
 154                  * we wait for its termination using a semaphore */
 155                 this->thread->cancel(this->thread);
 156                 terminated = this->terminated = malloc_thing(sem_t);
 157                 sem_init(terminated, 0, 0);
 158         }
 159         else
 160         {
 161                 /* if the job is currently queued, it gets terminated later.
 162                  * we can't wait, because it might not get executed at all.
 163                  * we also unregister the queued job manually from its parent (the
 164                  * others get unregistered during destruction) */
 165                 unregister(this);
 166         }
 167         this->destroyable->signal(this->destroyable);
 168         this->mutex->unlock(this->mutex);
 169
 170         if (terminated)
 171         {
 172                 sem_wait(terminated);
 173                 sem_destroy(terminated);
 174                 free(terminated);
 175         }
 176 }
 177
 178 METHOD(job_t, execute, void,
 179         private_callback_job_t *this)
 180 {
 181         bool cleanup = FALSE, requeue = FALSE;
 182
 183         thread_cleanup_push((thread_cleanup_t)destroy, this);
 184
 185         this->mutex->lock(this->mutex);
 186         this->thread = thread_current();
 187         this->mutex->unlock(this->mutex);
 188
 189         while (TRUE)
 190         {
 191                 this->mutex->lock(this->mutex);
 192                 if (this->cancelled)
 193                 {
 194                         this->mutex->unlock(this->mutex);
 195                         cleanup = TRUE;
 196                         break;
 197                 }
 198                 this->mutex->unlock(this->mutex);
 199                 switch (this->callback(this->data))
 200                 {
 201                         case JOB_REQUEUE_DIRECT:
 202                                 continue;
 203                         case JOB_REQUEUE_FAIR:
 204                         {
 205                                 requeue = TRUE;
 206                                 break;
 207                         }
 208                         case JOB_REQUEUE_NONE:
 209                         default:
 210                         {
 211                                 cleanup = TRUE;
 212                                 break;
 213                         }
 214                 }
 215                 break;
 216         }
 217         this->mutex->lock(this->mutex);
 218         this->thread = NULL;
 219         this->mutex->unlock(this->mutex);
 220         /* manually create a cancellation point to avoid that a cancelled thread
 221          * goes back into the thread pool */
 222         thread_cancellation_point();
 223         if (requeue)
 224         {
 225                 lib->processor->queue_job(lib->processor, &this->public.job);
 226         }
 227         thread_cleanup_pop(cleanup);
 228 }
 229
 230 METHOD(job_t, get_priority, job_priority_t,
 231         private_callback_job_t *this)
 232 {
 233         return JOB_PRIO_MEDIUM;
 234 }
 235
 236 /*
 237  * Described in header.
 238  */
 239 callback_job_t *callback_job_create(callback_job_cb_t cb, void *data,
 240                                                                         callback_job_cleanup_t cleanup,
 241                                                                         callback_job_t *parent)
 242 {
 243         private_callback_job_t *this;
 244
 245         INIT(this,
 246                 .public = {
 247                         .job = {
 248                                 .execute = _execute,
 249                                 .get_priority = _get_priority,
 250                                 .destroy = _destroy,
 251                         },
 252                         .cancel = _cancel,
 253                 },
 254                 .mutex = mutex_create(MUTEX_TYPE_DEFAULT),
 255                 .callback = cb,
 256                 .data = data,
 257                 .cleanup = cleanup,
 258                 .children = linked_list_create(),
 259                 .parent = (private_callback_job_t*)parent,
 260                 .destroyable = condvar_create(CONDVAR_TYPE_DEFAULT),
 261         );
 262
 263         /* register us at parent */
 264         if (parent)
 265         {
 266                 this->parent->mutex->lock(this->parent->mutex);
 267                 this->parent->children->insert_last(this->parent->children, this);
 268                 this->parent->mutex->unlock(this->parent->mutex);
 269         }
 270
 271         return &this->public;
 272 }
 273