[sr-dev] git:mariusbucur/dmq: started implementation of worker proceses in dmq

Module: sip-router Branch: mariusbucur/dmq Commit: a4f6960b50fd84da1c3f2f29d96d08f525878051 URL: http://git.sip-router.org/cgi-bin/gitweb.cgi/sip-router/?a=commit;h=a4f6960… Author: Marius Bucur &lt;marius.bucur(a)1and1.ro&gt; Committer: Marius Bucur &lt;marius.bucur(a)1and1.ro&gt; Date: Tue Apr 5 19:06:42 2011 +0300 started implementation of worker proceses in dmq --- modules_k/dmq/dmq.c | 88 ++++++++++++-------------------------------------- modules_k/dmq/dmq.h | 2 + 2 files changed, 23 insertions(+), 67 deletions(-) diff --git a/modules_k/dmq/dmq.c b/modules_k/dmq/dmq.c index f2bc2a8..f5aae9a 100644 --- a/modules_k/dmq/dmq.c +++ b/modules_k/dmq/dmq.c @@ -34,15 +34,10 @@ #include <fcntl.h> #include <time.h> -#include "../../lib/srdb1/db.h" #include "../../sr_module.h" #include "../../dprint.h" #include "../../error.h" #include "../../ut.h" -#include "../../parser/parse_to.h" -#include "../../parser/parse_uri.h" -#include "../../parser/parse_content.h" -#include "../../parser/parse_from.h" #include "../../mem/mem.h" #include "../../mem/shm_mem.h" #include "../../usr_avp.h" @@ -51,9 +46,9 @@ #include "../../pt.h" #include "../../lib/kmi/mi.h" #include "../../lib/kcore/hash_func.h" -#include "../pua/hash.h" #include "dmq.h" #include "bind_dmq.h" +#include "dmq_worker.h" #include "../../mod_fix.h" static int mi_child_init(void); @@ -64,21 +59,22 @@ static void destroy(void); MODULE_VERSION /* database connection */ -db1_con_t *dmq_db = NULL; -db_func_t dmq_dbf; int library_mode = 0; str server_address = {0, 0}; int startup_time = 0; int pid = 0; /* module parameters */ -str db_url; +int num_workers = DEFAULT_NUM_WORKERS; /* TM bind */ struct tm_binds tmb; /* SL API structure */ sl_api_t slb; +/** module variables */ +dmq_worker_t* workers; + /** module functions */ static int mod_init(void); static int child_init(int); @@ -92,7 +88,7 @@ static cmd_export_t cmds[] = { }; static param_export_t params[] = { - {"db_url", STR_PARAM, &db_url.s}, + {"num_workers", INT_PARAM, &num_workers}, {0, 0, 0} }; @@ -121,17 +117,13 @@ struct module_exports exports = { * init module function */ static int mod_init(void) { + int i = 0; + if(register_mi_mod(exports.name, mi_cmds)!=0) { LM_ERR("failed to register MI commands\n"); return -1; } - db_url.len = db_url.s ? strlen(db_url.s) : 0; - LM_DBG("db_url=%s/%d/%p\n", ZSW(db_url.s), db_url.len,db_url.s); - if(db_url.s== NULL) { - library_mode= 1; - } - if(library_mode== 1) { LM_DBG("dmq module used for API library purpose only\n"); } @@ -148,27 +140,19 @@ static int mod_init(void) { return -1; } - if(db_url.s== NULL) { - LM_ERR("database url not set!\n"); - return -1; - } - - /* binding to database module */ - if (db_bind_mod(&db_url, &dmq_dbf)) { - LM_ERR("database module not found\n"); - return -1; - } - - - if (!DB_CAPABILITY(dmq_dbf, DB_CAP_ALL)) { - LM_ERR("database module does not implement all functions needed by dmq module\n"); - return -1; - } - - dmq_db = dmq_dbf.init(&db_url); - if (!dmq_db) { - LM_ERR("connection to database failed\n"); - return -1; + /* fork worker processes */ + workers = shm_malloc(num_workers * sizeof(*workers)); + for(i = 0; i < num_workers; i++) { + int newpid = fork_process(PROC_NOCHLDINIT, "DMQ WORKER", 0); + if(newpid < 0) { + LM_ERR("failed to form process\n"); + return -1; + } else if(newpid == 0) { + /* child */ + // worker loop + } else { + workers[i].pid = newpid; + } } startup_time = (int) time(NULL); @@ -189,40 +173,10 @@ static int child_init(int rank) { if(library_mode) return 0; - if (dmq_dbf.init==0) { - LM_CRIT("child_init: database not bound\n"); - return -1; - } - if (dmq_db) - return 0; - - dmq_db = dmq_dbf.init(&db_url); - if (!dmq_db) { - LM_ERR("child %d: unsuccessful connecting to database\n", rank); - return -1; - } - - LM_DBG("child %d: database connection opened successfully\n", rank); return 0; } static int mi_child_init(void) { - if(library_mode) - return 0; - - if (dmq_dbf.init==0) { - LM_CRIT("database not bound\n"); - return -1; - } - if (dmq_db) - return 0; - dmq_db = dmq_dbf.init(&db_url); - if (!dmq_db) { - LM_ERR("connecting database\n"); - return -1; - } - - LM_DBG("database connection opened successfully\n"); return 0; } diff --git a/modules_k/dmq/dmq.h b/modules_k/dmq/dmq.h index bea79d3..d4b13de 100644 --- a/modules_k/dmq/dmq.h +++ b/modules_k/dmq/dmq.h @@ -1 +1,3 @@ + +#define DEFAULT_NUM_WORKERS 2 int handle_dmq_message(struct sip_msg* msg, char* str1 ,char* str2);