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

[Marius Ovidiu Bucur]

Module: sip-router
Branch: mariusbucur/dmq
Commit: a4f6960b50fd84da1c3f2f29d96d08f525878051
URL:    http://git.sip-router.org/cgi-bin/gitweb.cgi/sip-router/?a=commit;h=a4f6960b50fd84da1c3f2f29d96d08f525878051

Author: Marius Bucur <marius.bucur at 1and1.ro>
Committer: Marius Bucur <marius.bucur at 1and1.ro>
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);