Merge pull request #2326 from NGSegovia/keepalive/first_check_on_start
[sip-router] / 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 = &dmq_workers[id];
83         for(;;) {
84                 if(dmq_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(dmq_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(dmq_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(!dmq_num_workers) {
189                 LM_ERR("error in add_dmq_job: no workers spawned\n");
190                 goto error;
191         }
192         if(!dmq_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 = dmq_workers;
198         /* search for an available worker, or, if not possible,
199          * for the least busy one */
200         for(i = 0; i < dmq_num_workers; i++) {
201                 if(job_queue_size(dmq_workers[i].queue) == 0) {
202                         worker = &dmq_workers[i];
203                         found_available = 1;
204                         break;
205                 } else if(job_queue_size(dmq_workers[i].queue)
206                                   < job_queue_size(worker->queue)) {
207                         worker = &dmq_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(dmq_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(dmq_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 }