git.sip-router.org Git - sip-router/blob - src/modules/dmq/worker.c

   1 /*
   2  * dmq module - distributed message queue
   3  *
   4  * Copyright (C) 2011 Bucur Marius - Ovidiu
   5  *
   6  * This file is part of Kamailio, a free SIP server.
   7  *
   8  * Kamailio is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License as published by
  10  * the Free Software Foundation; either version 2 of the License, or
  11  * (at your option) any later version
  12  *
  13  * Kamailio is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
  21  *
  22  */
  23
  24 #include "dmq.h"
  25 #include "peer.h"
  26 #include "worker.h"
  27 #include "../../core/data_lump_rpl.h"
  28 #include "../../core/mod_fix.h"
  29 #include "../../core/sip_msg_clone.h"
  30 #include "../../core/parser/parse_from.h"
  31 #include "../../core/parser/parse_to.h"
  32
  33 /**
  34  * @brief set the body of a response
  35  */
  36 static int set_reply_body(struct sip_msg *msg, str *body, str *content_type)
  37 {
  38         char *buf;
  39         int len;
  40
  41         /* add content-type */
  42         len = sizeof("Content-Type: ") - 1 + content_type->len + CRLF_LEN;
  43         buf = pkg_malloc(sizeof(char) * (len));
  44
  45         if(buf == 0) {
  46                 LM_ERR("out of pkg memory\n");
  47                 return -1;
  48         }
  49         memcpy(buf, "Content-Type: ", sizeof("Content-Type: ") - 1);
  50         memcpy(buf + sizeof("Content-Type: ") - 1, content_type->s,
  51                         content_type->len);
  52         memcpy(buf + sizeof("Content-Type: ") - 1 + content_type->len, CRLF,
  53                         CRLF_LEN);
  54         if(add_lump_rpl(msg, buf, len, LUMP_RPL_HDR) == 0) {
  55                 LM_ERR("failed to insert content-type lump\n");
  56                 pkg_free(buf);
  57                 return -1;
  58         }
  59         pkg_free(buf);
  60
  61         /* add body */
  62         if(add_lump_rpl(msg, body->s, body->len, LUMP_RPL_BODY) == 0) {
  63                 LM_ERR("cannot add body lump\n");
  64                 return -1;
  65         }
  66
  67         return 1;
  68 }
  69
  70 /**
  71  * @brief dmq worker loop
  72  */
  73 void worker_loop(int id)
  74 {
  75         dmq_worker_t *worker;
  76         dmq_job_t *current_job;
  77         peer_reponse_t peer_response;
  78         int ret_value;
  79         int not_parsed;
  80         dmq_node_t *dmq_node = NULL;
  81
  82         worker = &workers[id];
  83         for(;;) {
  84                 if(worker_usleep <= 0) {
  85                         LM_DBG("dmq_worker [%d %d] getting lock\n", id, my_pid());
  86                         lock_get(&worker->lock);
  87                         LM_DBG("dmq_worker [%d %d] lock acquired\n", id, my_pid());
  88                 } else {
  89                         sleep_us(worker_usleep);
  90                 }
  91
  92                 /* remove from queue until empty */
  93                 while(job_queue_size(worker->queue) > 0) {
  94                         /* fill the response with 0's */
  95                         memset(&peer_response, 0, sizeof(peer_response));
  96                         current_job = job_queue_pop(worker->queue);
  97                         /* job_queue_pop might return NULL if queue is empty */
  98                         if(current_job) {
  99                                 /* extract the from uri */
 100                                 if(current_job->msg->from->parsed) {
 101                                         not_parsed = 0;
 102                                 } else {
 103                                         not_parsed = 1;
 104                                 }
 105                                 if(parse_from_header(current_job->msg) < 0) {
 106                                         LM_ERR("bad sip message or missing From hdr\n");
 107                                 } else {
 108                                         dmq_node = find_dmq_node_uri(node_list,
 109                                                         &((struct to_body *)current_job->msg->from->parsed)
 110                                                                          ->uri);
 111                                 }
 112
 113                                 ret_value = current_job->f(
 114                                                 current_job->msg, &peer_response, dmq_node);
 115                                 if(ret_value < 0) {
 116                                         LM_ERR("running job failed\n");
 117                                         goto nextjob;
 118                                 }
 119                                 /* add the body to the reply */
 120                                 if(peer_response.body.s) {
 121                                         if(set_reply_body(current_job->msg, &peer_response.body,
 122                                                            &peer_response.content_type)
 123                                                         < 0) {
 124                                                 LM_ERR("error adding lumps\n");
 125                                                 goto nextjob;
 126                                         }
 127                                 }
 128                                 /* send the reply */
 129                                 if(peer_response.resp_code>0 && peer_response.reason.s!=NULL
 130                                                 && peer_response.reason.len>0) {
 131                                         if(slb.freply(current_job->msg, peer_response.resp_code,
 132                                                            &peer_response.reason)
 133                                                         < 0) {
 134                                                 LM_ERR("error sending reply\n");
 135                                         } else {
 136                                                 LM_DBG("done sending reply\n");
 137                                         }
 138                                 } else {
 139                                         LM_WARN("no reply sent\n");
 140                                 }
 141                                 worker->jobs_processed++;
 142
 143 nextjob:
 144                                 /* if body given, free the lumps and free the body */
 145                                 if(peer_response.body.s) {
 146                                         del_nonshm_lump_rpl(&current_job->msg->reply_lump);
 147                                         pkg_free(peer_response.body.s);
 148                                 }
 149                                 if((current_job->msg->from->parsed) && (not_parsed)) {
 150                                         free_to(current_job->msg->from->parsed);
 151                                 }
 152
 153                                 shm_free(current_job->msg);
 154                                 shm_free(current_job);
 155                         }
 156                 }
 157         }
 158 }
 159
 160 /**
 161  * @brief add a dmq job
 162  */
 163 int add_dmq_job(struct sip_msg *msg, dmq_peer_t *peer)
 164 {
 165         int i, found_available = 0;
 166         dmq_job_t new_job = {0};
 167         dmq_worker_t *worker;
 168         struct sip_msg *cloned_msg = NULL;
 169         int cloned_msg_len;
 170
 171         /* Pre-parse headers so they are included in our clone. Parsing later
 172          * will result in linking pkg structures to shm msg, eventually leading
 173          * to memory errors. */
 174         if(parse_headers(msg, HDR_EOH_F, 0) == -1) {
 175                 LM_ERR("failed to parse headers\n");
 176                 return -1;
 177         }
 178
 179         cloned_msg = sip_msg_shm_clone(msg, &cloned_msg_len, 1);
 180         if(!cloned_msg) {
 181                 LM_ERR("error cloning sip message\n");
 182                 return -1;
 183         }
 184
 185         new_job.f = peer->callback;
 186         new_job.msg = cloned_msg;
 187         new_job.orig_peer = peer;
 188         if(!num_workers) {
 189                 LM_ERR("error in add_dmq_job: no workers spawned\n");
 190                 goto error;
 191         }
 192         if(!workers[0].queue) {
 193                 LM_ERR("workers not (yet) initialized\n");
 194                 goto error;
 195         }
 196         /* initialize the worker with the first one */
 197         worker = workers;
 198         /* search for an available worker, or, if not possible,
 199          * for the least busy one */
 200         for(i = 0; i < num_workers; i++) {
 201                 if(job_queue_size(workers[i].queue) == 0) {
 202                         worker = &workers[i];
 203                         found_available = 1;
 204                         break;
 205                 } else if(job_queue_size(workers[i].queue)
 206                                   < job_queue_size(worker->queue)) {
 207                         worker = &workers[i];
 208                 }
 209         }
 210         if(!found_available) {
 211                 LM_DBG("no available worker found, passing job"
 212                            " to the least busy one [%d %d]\n",
 213                                 worker->pid, job_queue_size(worker->queue));
 214         }
 215         if(job_queue_push(worker->queue, &new_job) < 0) {
 216                 goto error;
 217         }
 218         if(worker_usleep <= 0) {
 219                 lock_release(&worker->lock);
 220         }
 221         return 0;
 222 error:
 223         if(cloned_msg != NULL) {
 224                 shm_free(cloned_msg);
 225         }
 226         return -1;
 227 }
 228
 229 /**
 230  * @brief init dmq worker
 231  */
 232 int init_worker(dmq_worker_t *worker)
 233 {
 234         memset(worker, 0, sizeof(*worker));
 235         if(worker_usleep <= 0) {
 236                 lock_init(&worker->lock);
 237                 // acquire the lock for the first time - so that dmq_worker_loop blocks
 238                 lock_get(&worker->lock);
 239         }
 240         worker->queue = alloc_job_queue();
 241         if(worker->queue==NULL) {
 242                 LM_ERR("queue could not be initialized\n");
 243                 return -1;
 244         }
 245         return 0;
 246 }
 247
 248 /**
 249  * @brief allog dmq job queue
 250  */
 251 job_queue_t *alloc_job_queue()
 252 {
 253         job_queue_t *queue;
 254
 255         queue = shm_malloc(sizeof(job_queue_t));
 256         if(queue == NULL) {
 257                 LM_ERR("no more shm\n");
 258                 return NULL;
 259         }
 260         memset(queue, 0, sizeof(job_queue_t));
 261         atomic_set(&queue->count, 0);
 262         lock_init(&queue->lock);
 263         return queue;
 264 }
 265
 266 /**
 267  * @ brief destroy job queue
 268  */
 269 void destroy_job_queue(job_queue_t *queue)
 270 {
 271         if(queue != NULL)
 272                 shm_free(queue);
 273 }
 274
 275 /**
 276  * @brief return job queue size
 277  */
 278 int job_queue_size(job_queue_t *queue)
 279 {
 280         return atomic_get(&queue->count);
 281 }
 282
 283 /**
 284  * @brief push to job queue
 285  */
 286 int job_queue_push(job_queue_t *queue, dmq_job_t *job)
 287 {
 288         /* we need to copy the dmq_job into a newly created dmq_job in shm */
 289         dmq_job_t *newjob;
 290
 291         newjob = shm_malloc(sizeof(dmq_job_t));
 292         if(newjob == NULL) {
 293                 LM_ERR("no more shm\n");
 294                 return -1;
 295         }
 296
 297         *newjob = *job;
 298
 299         lock_get(&queue->lock);
 300         newjob->prev = NULL;
 301         newjob->next = queue->back;
 302         if(queue->back) {
 303                 queue->back->prev = newjob;
 304         }
 305         queue->back = newjob;
 306         if(!queue->front) {
 307                 queue->front = newjob;
 308         }
 309         atomic_inc(&queue->count);
 310         lock_release(&queue->lock);
 311         return 0;
 312 }
 313
 314 /**
 315  * @brief pop from job queue
 316  */
 317 dmq_job_t *job_queue_pop(job_queue_t *queue)
 318 {
 319         dmq_job_t *front;
 320         lock_get(&queue->lock);
 321         if(!queue->front) {
 322                 lock_release(&queue->lock);
 323                 return NULL;
 324         }
 325         front = queue->front;
 326         if(front->prev) {
 327                 queue->front = front->prev;
 328                 front->prev->next = NULL;
 329         } else {
 330                 queue->front = NULL;
 331                 queue->back = NULL;
 332         }
 333         atomic_dec(&queue->count);
 334         lock_release(&queue->lock);
 335         return front;
 336 }