src/libstrongswan/networking/streams/stream_service.c

   1 /*
   2  * Copyright (C) 2013 Martin Willi
   3  * Copyright (C) 2013 revosec AG
   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 <library.h>
  17 #include <threading/thread.h>
  18 #include <threading/mutex.h>
  19 #include <threading/condvar.h>
  20 #include <processing/jobs/callback_job.h>
  21
  22 #include "stream_service.h"
  23
  24 #include <errno.h>
  25 #include <unistd.h>
  26 #include <sys/stat.h>
  27
  28 typedef struct private_stream_service_t private_stream_service_t;
  29
  30 /**
  31  * Private data of an stream_service_t object.
  32  */
  33 struct private_stream_service_t {
  34
  35         /**
  36          * Public stream_service_t interface.
  37          */
  38         stream_service_t public;
  39
  40         /**
  41          * Underlying socket
  42          */
  43         int fd;
  44
  45         /**
  46          * Accept callback
  47          */
  48         stream_service_cb_t cb;
  49
  50         /**
  51          * Accept callback data
  52          */
  53         void *data;
  54
  55         /**
  56          * Job priority to invoke callback with
  57          */
  58         job_priority_t prio;
  59
  60         /**
  61          * Maximum number of parallel callback invocations
  62          */
  63         u_int cncrncy;
  64
  65         /**
  66          * Currently active jobs
  67          */
  68         u_int active;
  69
  70         /**
  71          * mutex to lock active counter
  72          */
  73         mutex_t *mutex;
  74
  75         /**
  76          * Condvar to wait for callback termination
  77          */
  78         condvar_t *condvar;
  79 };
  80
  81 /**
  82  * Data to pass to async accept job
  83  */
  84 typedef struct {
  85         /** callback function */
  86         stream_service_cb_t cb;
  87         /** callback data */
  88         void *data;
  89         /** accepted connection */
  90         int fd;
  91         /** reference to stream service */
  92         private_stream_service_t *this;
  93 } async_data_t;
  94
  95 /**
  96  * Forward declaration
  97  */
  98 static bool watch(private_stream_service_t *this, int fd, watcher_event_t event);
  99
 100 /**
 101  * Clean up accept data
 102  */
 103 static void destroy_async_data(async_data_t *data)
 104 {
 105         private_stream_service_t *this = data->this;
 106
 107         this->mutex->lock(this->mutex);
 108         if (this->active-- == this->cncrncy)
 109         {
 110                 /* leaving concurrency limit, restart accept()ing. */
 111                 lib->watcher->add(lib->watcher, this->fd,
 112                                                   WATCHER_READ, (watcher_cb_t)watch, this);
 113         }
 114         this->condvar->signal(this->condvar);
 115         this->mutex->unlock(this->mutex);
 116
 117         if (data->fd != -1)
 118         {
 119                 close(data->fd);
 120         }
 121         free(data);
 122 }
 123
 124 /**
 125  * Async processing of accepted connection
 126  */
 127 static job_requeue_t accept_async(async_data_t *data)
 128 {
 129         stream_t *stream;
 130
 131         stream = stream_create_from_fd(data->fd);
 132         if (stream)
 133         {
 134                 /* FD is now owned by stream, don't close it during cleanup */
 135                 data->fd = -1;
 136                 thread_cleanup_push((void*)stream->destroy, stream);
 137                 thread_cleanup_pop(!data->cb(data->data, stream));
 138         }
 139         return JOB_REQUEUE_NONE;
 140 }
 141
 142 /**
 143  * Watcher callback function
 144  */
 145 static bool watch(private_stream_service_t *this, int fd, watcher_event_t event)
 146 {
 147         async_data_t *data;
 148         bool keep = TRUE;
 149
 150         INIT(data,
 151                 .cb = this->cb,
 152                 .data = this->data,
 153                 .fd = accept(fd, NULL, NULL),
 154                 .this = this,
 155         );
 156
 157         if (data->fd != -1)
 158         {
 159                 this->mutex->lock(this->mutex);
 160                 if (++this->active == this->cncrncy)
 161                 {
 162                         /* concurrency limit reached, stop accept()ing new connections */
 163                         keep = FALSE;
 164                 }
 165                 this->mutex->unlock(this->mutex);
 166
 167                 lib->processor->queue_job(lib->processor,
 168                         (job_t*)callback_job_create_with_prio((void*)accept_async, data,
 169                                 (void*)destroy_async_data, (callback_job_cancel_t)return_false,
 170                                 this->prio));
 171         }
 172         else
 173         {
 174                 free(data);
 175         }
 176         return keep;
 177 }
 178
 179 METHOD(stream_service_t, on_accept, void,
 180         private_stream_service_t *this, stream_service_cb_t cb, void *data,
 181         job_priority_t prio, u_int cncrncy)
 182 {
 183         this->mutex->lock(this->mutex);
 184
 185         /* wait for all callbacks to return */
 186         while (this->active)
 187         {
 188                 this->condvar->wait(this->condvar, this->mutex);
 189         }
 190
 191         if (this->cb)
 192         {
 193                 lib->watcher->remove(lib->watcher, this->fd);
 194         }
 195
 196         this->cb = cb;
 197         this->data = data;
 198         if (prio <= JOB_PRIO_MAX)
 199         {
 200                 this->prio = prio;
 201         }
 202         this->cncrncy = cncrncy;
 203
 204         if (this->cb)
 205         {
 206                 lib->watcher->add(lib->watcher, this->fd,
 207                                                   WATCHER_READ, (watcher_cb_t)watch, this);
 208         }
 209
 210         this->mutex->unlock(this->mutex);
 211 }
 212
 213 METHOD(stream_service_t, destroy, void,
 214         private_stream_service_t *this)
 215 {
 216         on_accept(this, NULL, NULL, this->prio, this->cncrncy);
 217         close(this->fd);
 218         this->mutex->destroy(this->mutex);
 219         this->condvar->destroy(this->condvar);
 220         free(this);
 221 }
 222
 223 /**
 224  * See header
 225  */
 226 stream_service_t *stream_service_create_from_fd(int fd)
 227 {
 228         private_stream_service_t *this;
 229
 230         INIT(this,
 231                 .public = {
 232                         .on_accept = _on_accept,
 233                         .destroy = _destroy,
 234                 },
 235                 .fd = fd,
 236                 .prio = JOB_PRIO_MEDIUM,
 237                 .mutex = mutex_create(MUTEX_TYPE_RECURSIVE),
 238                 .condvar = condvar_create(CONDVAR_TYPE_DEFAULT),
 239         );
 240
 241         return &this->public;
 242 }