src/modules/dmq/worker.c
1977645c
 /*
  * dmq module - distributed message queue
  *
  * Copyright (C) 2011 Bucur Marius - Ovidiu
  *
  * This file is part of Kamailio, a free SIP server.
  *
  * Kamailio is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
  * (at your option) any later version
  *
  * Kamailio is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU General Public License for more details.
  *
  * You should have received a copy of the GNU General Public License 
  * along with this program; if not, write to the Free Software 
9e1ff448
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
1977645c
  *
  */
 
ba31d863
 #include "dmq.h"
a3918ba7
 #include "peer.h"
ba31d863
 #include "worker.h"
cf83221d
 #include "../../core/data_lump_rpl.h"
 #include "../../core/mod_fix.h"
 #include "../../core/sip_msg_clone.h"
 #include "../../core/parser/parse_from.h"
 #include "../../core/parser/parse_to.h"
f403eea5
 
1977645c
 /**
  * @brief set the body of a response
  */
2b7ae0f0
 static int set_reply_body(struct sip_msg *msg, str *body, str *content_type)
f403eea5
 {
2b7ae0f0
 	char *buf;
f403eea5
 	int len;
 
 	/* add content-type */
2b7ae0f0
 	len = sizeof("Content-Type: ") - 1 + content_type->len + CRLF_LEN;
 	buf = pkg_malloc(sizeof(char) * (len));
f403eea5
 
2b7ae0f0
 	if(buf == 0) {
f403eea5
 		LM_ERR("out of pkg memory\n");
 		return -1;
 	}
 	memcpy(buf, "Content-Type: ", sizeof("Content-Type: ") - 1);
2b7ae0f0
 	memcpy(buf + sizeof("Content-Type: ") - 1, content_type->s,
 			content_type->len);
 	memcpy(buf + sizeof("Content-Type: ") - 1 + content_type->len, CRLF,
 			CRLF_LEN);
 	if(add_lump_rpl(msg, buf, len, LUMP_RPL_HDR) == 0) {
f403eea5
 		LM_ERR("failed to insert content-type lump\n");
 		pkg_free(buf);
 		return -1;
 	}
 	pkg_free(buf);
 
 	/* add body */
2b7ae0f0
 	if(add_lump_rpl(msg, body->s, body->len, LUMP_RPL_BODY) == 0) {
f403eea5
 		LM_ERR("cannot add body lump\n");
 		return -1;
 	}
2b7ae0f0
 
f403eea5
 	return 1;
 }
ba31d863
 
1977645c
 /**
  * @brief dmq worker loop
  */
 void worker_loop(int id)
 {
2b7ae0f0
 	dmq_worker_t *worker;
 	dmq_job_t *current_job;
a3918ba7
 	peer_reponse_t peer_response;
ba31d863
 	int ret_value;
7bcf8732
 	int not_parsed;
c2dcf4db
 	dmq_node_t *dmq_node = NULL;
1977645c
 
d500589d
 	worker = &dmq_workers[id];
ba31d863
 	for(;;) {
d500589d
 		if(dmq_worker_usleep <= 0) {
71a88212
 			LM_DBG("dmq_worker [%d %d] getting lock\n", id, my_pid());
 			lock_get(&worker->lock);
 			LM_DBG("dmq_worker [%d %d] lock acquired\n", id, my_pid());
 		} else {
d500589d
 			sleep_us(dmq_worker_usleep);
71a88212
 		}
 
 		/* remove from queue until empty */
 		while(job_queue_size(worker->queue) > 0) {
f403eea5
 			/* fill the response with 0's */
 			memset(&peer_response, 0, sizeof(peer_response));
ba31d863
 			current_job = job_queue_pop(worker->queue);
 			/* job_queue_pop might return NULL if queue is empty */
 			if(current_job) {
c2dcf4db
 				/* extract the from uri */
2b7ae0f0
 				if(current_job->msg->from->parsed) {
7bcf8732
 					not_parsed = 0;
 				} else {
 					not_parsed = 1;
 				}
2b7ae0f0
 				if(parse_from_header(current_job->msg) < 0) {
c2dcf4db
 					LM_ERR("bad sip message or missing From hdr\n");
 				} else {
d500589d
 					dmq_node = find_dmq_node_uri(dmq_node_list,
2b7ae0f0
 							&((struct to_body *)current_job->msg->from->parsed)
 									 ->uri);
c2dcf4db
 				}
 
2b7ae0f0
 				ret_value = current_job->f(
 						current_job->msg, &peer_response, dmq_node);
ba31d863
 				if(ret_value < 0) {
 					LM_ERR("running job failed\n");
a1f5fbe2
 					goto nextjob;
ba31d863
 				}
f403eea5
 				/* add the body to the reply */
 				if(peer_response.body.s) {
1977645c
 					if(set_reply_body(current_job->msg, &peer_response.body,
2b7ae0f0
 							   &peer_response.content_type)
 							< 0) {
f403eea5
 						LM_ERR("error adding lumps\n");
a1f5fbe2
 						goto nextjob;
f403eea5
 					}
 				}
 				/* send the reply */
a675ab88
 				if(peer_response.resp_code>0 && peer_response.reason.s!=NULL
 						&& peer_response.reason.len>0) {
 					if(slb.freply(current_job->msg, peer_response.resp_code,
 							   &peer_response.reason)
 							< 0) {
 						LM_ERR("error sending reply\n");
 					} else {
 						LM_DBG("done sending reply\n");
 					}
a1f5fbe2
 				} else {
a675ab88
 					LM_WARN("no reply sent\n");
f403eea5
 				}
a1f5fbe2
 				worker->jobs_processed++;
2b7ae0f0
 
a1f5fbe2
 nextjob:
f403eea5
 				/* if body given, free the lumps and free the body */
 				if(peer_response.body.s) {
 					del_nonshm_lump_rpl(&current_job->msg->reply_lump);
 					pkg_free(peer_response.body.s);
 				}
2b7ae0f0
 				if((current_job->msg->from->parsed) && (not_parsed)) {
9b0ce4d0
 					free_to(current_job->msg->from->parsed);
 				}
 
f403eea5
 				shm_free(current_job->msg);
ba31d863
 				shm_free(current_job);
 			}
71a88212
 		}
ba31d863
 	}
 }
 
1977645c
 /**
  * @brief add a dmq job
  */
2b7ae0f0
 int add_dmq_job(struct sip_msg *msg, dmq_peer_t *peer)
1977645c
 {
ba31d863
 	int i, found_available = 0;
2b7ae0f0
 	dmq_job_t new_job = {0};
 	dmq_worker_t *worker;
 	struct sip_msg *cloned_msg = NULL;
3b028d30
 	int cloned_msg_len;
 
 	/* Pre-parse headers so they are included in our clone. Parsing later
 	 * will result in linking pkg structures to shm msg, eventually leading 
 	 * to memory errors. */
2b7ae0f0
 	if(parse_headers(msg, HDR_EOH_F, 0) == -1) {
3b028d30
 		LM_ERR("failed to parse headers\n");
 		return -1;
 	}
 
 	cloned_msg = sip_msg_shm_clone(msg, &cloned_msg_len, 1);
 	if(!cloned_msg) {
 		LM_ERR("error cloning sip message\n");
 		return -1;
 	}
1977645c
 
ba31d863
 	new_job.f = peer->callback;
3b028d30
 	new_job.msg = cloned_msg;
ba31d863
 	new_job.orig_peer = peer;
d500589d
 	if(!dmq_num_workers) {
f403eea5
 		LM_ERR("error in add_dmq_job: no workers spawned\n");
3b028d30
 		goto error;
ba31d863
 	}
d500589d
 	if(!dmq_workers[0].queue) {
63bbaabb
 		LM_ERR("workers not (yet) initialized\n");
 		goto error;
 	}
ba31d863
 	/* initialize the worker with the first one */
d500589d
 	worker = dmq_workers;
1977645c
 	/* search for an available worker, or, if not possible,
 	 * for the least busy one */
d500589d
 	for(i = 0; i < dmq_num_workers; i++) {
 		if(job_queue_size(dmq_workers[i].queue) == 0) {
 			worker = &dmq_workers[i];
ba31d863
 			found_available = 1;
 			break;
d500589d
 		} else if(job_queue_size(dmq_workers[i].queue)
2b7ae0f0
 				  < job_queue_size(worker->queue)) {
d500589d
 			worker = &dmq_workers[i];
ba31d863
 		}
 	}
 	if(!found_available) {
1977645c
 		LM_DBG("no available worker found, passing job"
2b7ae0f0
 			   " to the least busy one [%d %d]\n",
1977645c
 				worker->pid, job_queue_size(worker->queue));
ba31d863
 	}
2b7ae0f0
 	if(job_queue_push(worker->queue, &new_job) < 0) {
3b028d30
 		goto error;
 	}
d500589d
 	if(dmq_worker_usleep <= 0) {
71a88212
 		lock_release(&worker->lock);
 	}
3b028d30
 	return 0;
 error:
2b7ae0f0
 	if(cloned_msg != NULL) {
3b028d30
 		shm_free(cloned_msg);
 	}
 	return -1;
ba31d863
 }
 
1977645c
 /**
  * @brief init dmq worker
  */
21bdbff0
 int init_worker(dmq_worker_t *worker)
1977645c
 {
ba31d863
 	memset(worker, 0, sizeof(*worker));
d500589d
 	if(dmq_worker_usleep <= 0) {
71a88212
 		lock_init(&worker->lock);
 		// acquire the lock for the first time - so that dmq_worker_loop blocks
 		lock_get(&worker->lock);
 	}
ba31d863
 	worker->queue = alloc_job_queue();
21bdbff0
 	if(worker->queue==NULL) {
 		LM_ERR("queue could not be initialized\n");
 		return -1;
 	}
 	return 0;
ba31d863
 }
 
1977645c
 /**
  * @brief allog dmq job queue
  */
2b7ae0f0
 job_queue_t *alloc_job_queue()
1977645c
 {
2b7ae0f0
 	job_queue_t *queue;
 
1977645c
 	queue = shm_malloc(sizeof(job_queue_t));
2b7ae0f0
 	if(queue == NULL) {
1977645c
 		LM_ERR("no more shm\n");
 		return NULL;
 	}
 	memset(queue, 0, sizeof(job_queue_t));
ba31d863
 	atomic_set(&queue->count, 0);
 	lock_init(&queue->lock);
 	return queue;
 }
 
1977645c
 /**
  * @ brief destroy job queue
  */
2b7ae0f0
 void destroy_job_queue(job_queue_t *queue)
1977645c
 {
2b7ae0f0
 	if(queue != NULL)
1977645c
 		shm_free(queue);
ba31d863
 }
 
1977645c
 /**
  * @brief return job queue size
  */
2b7ae0f0
 int job_queue_size(job_queue_t *queue)
1977645c
 {
ba31d863
 	return atomic_get(&queue->count);
 }
 
1977645c
 /**
  * @brief push to job queue
  */
2b7ae0f0
 int job_queue_push(job_queue_t *queue, dmq_job_t *job)
1977645c
 {
ba31d863
 	/* we need to copy the dmq_job into a newly created dmq_job in shm */
2b7ae0f0
 	dmq_job_t *newjob;
 
1977645c
 	newjob = shm_malloc(sizeof(dmq_job_t));
2b7ae0f0
 	if(newjob == NULL) {
1977645c
 		LM_ERR("no more shm\n");
 		return -1;
 	}
 
ba31d863
 	*newjob = *job;
2b7ae0f0
 
ba31d863
 	lock_get(&queue->lock);
 	newjob->prev = NULL;
 	newjob->next = queue->back;
 	if(queue->back) {
 		queue->back->prev = newjob;
 	}
 	queue->back = newjob;
 	if(!queue->front) {
 		queue->front = newjob;
 	}
 	atomic_inc(&queue->count);
 	lock_release(&queue->lock);
1977645c
 	return 0;
ba31d863
 }
1977645c
 
 /**
  * @brief pop from job queue
  */
2b7ae0f0
 dmq_job_t *job_queue_pop(job_queue_t *queue)
1977645c
 {
2b7ae0f0
 	dmq_job_t *front;
ba31d863
 	lock_get(&queue->lock);
 	if(!queue->front) {
 		lock_release(&queue->lock);
 		return NULL;
 	}
 	front = queue->front;
 	if(front->prev) {
 		queue->front = front->prev;
 		front->prev->next = NULL;
 	} else {
 		queue->front = NULL;
 		queue->back = NULL;
 	}
 	atomic_dec(&queue->count);
 	lock_release(&queue->lock);
 	return front;
e91db504
 }