modules_k/pua: Added (start|end|abort)_transaction calls (when in DB only mode and where supported)

- Helps with some multi-server/process race hazards

modules_k/pua/pua.h

@@ -26,6 +26,7 @@
 #define PUBLISH_H
 
 #include "../../str.h"
+#include "../../lib/srdb1/db.h"
 #include "hash.h"
 
 #define INSERT_TYPE      1<<1
@@ -54,4 +55,9 @@ int reginfo_increase_version;
 extern int update_pua(ua_pres_t* p);
 extern int clean_puadb( int update_period, int min_expires );
 
+extern db_func_t pua_dbf;
+extern db1_con_t *pua_db;
+extern int pua_fetch_rows;
+
+
 #endif

modules_k/pua/pua_db.c

@@ -37,10 +37,6 @@
 #include "pua.h"
 #include "pua_db.h"
 
-/* database connection */
-extern db1_con_t *pua_db;
-extern int pua_fetch_rows;
-
 /* database colums */
 static str str_id_col = str_init( "id" );
 static str str_pres_uri_col = str_init("pres_uri");

modules_k/pua/pua_db.h

@@ -47,8 +47,6 @@
 #define PUA_VERSION (1<<16)
 #define PUA_EXTRA_HEADERS (1<<17)
 
-extern db_func_t pua_dbf;
-
 void free_results_puadb( db1_res_t *res );
 int is_dialog_puadb(ua_pres_t *pres);
 int get_record_id_puadb(ua_pres_t *pres, str **rec_id );

modules_k/pua/send_publish.c

@@ -213,6 +213,15 @@ void publ_cback_func(struct cell *t, int type, struct tmcb_params *ps)
 	dbpres.watcher_uri = &watcher_uri;
 	dbpres.extra_headers = &extra_headers;
 
+	if (dbmode == PUA_DB_ONLY && pua_dbf.start_transaction)
+	{
+		if (pua_dbf.start_transaction(pua_db) < 0)
+		{
+			LM_ERR("in start_transaction\n");
+			goto error;
+		}
+	}
+
 	if(ps->param== NULL|| *ps->param== NULL)
 	{
 		LM_ERR("NULL callback parameter\n");
@@ -422,6 +431,16 @@ done:
 	}
 
 	if (res) free_results_puadb(res);
+
+	if (dbmode == PUA_DB_ONLY && pua_dbf.end_transaction)
+	{
+		if (pua_dbf.end_transaction(pua_db) < 0)
+		{
+			LM_ERR("in end_transaction\n");
+			goto error;
+		}
+	}
+
 	return;
 
 error:
@@ -433,6 +452,13 @@ error:
 	if(presentity) shm_free(presentity);
 
 	if (res) free_results_puadb(res);
+
+	if (dbmode == PUA_DB_ONLY && pua_dbf.abort_transaction)
+	{
+		if (pua_dbf.abort_transaction(pua_db) < 0)
+			LM_ERR("in abort_transaction\n");
+	}
+
 	return;
 }	
 
@@ -454,11 +480,20 @@ int send_publish( publ_info_t* publ )
 	db1_res_t *res=NULL;
 	ua_pres_t dbpres; 
 	str pres_uri={0,0}, watcher_uri={0,0}, extra_headers={0,0};
+	int ret = -1;
 
 	LM_DBG("pres_uri=%.*s\n", publ->pres_uri->len, publ->pres_uri->s );
 	
-	/* get event from list */
+	if (dbmode == PUA_DB_ONLY && pua_dbf.start_transaction)
+	{
+		if (pua_dbf.start_transaction(pua_db) < 0)
+		{
+			LM_ERR("in start_transaction\n");
+			goto error;
+		}
+	}
 
+	/* get event from list */
 	ev= get_event(publ->event);
 	if(ev== NULL)
 	{
@@ -495,8 +530,8 @@ int send_publish( publ_info_t* publ )
 	{
 		if (dbmode!=PUA_DB_ONLY) 
 			lock_release(&HashT->p_records[hash_code].lock);
-		free_results_puadb(res);
-		return 418;
+		ret = 418;
+		goto error;
 	}
 
 	if(publ->flag & INSERT_TYPE)
@@ -521,14 +556,14 @@ insert:
 		{
 			LM_DBG("request for a publish with expires 0 and"
 					" no record found\n");
-			free_results_puadb(res);
-			return 0;
+			goto done;
+			
 		}
 		if(publ->body== NULL)
 		{
 			LM_ERR("New PUBLISH and no body found- invalid request\n");
-			free_results_puadb(res);
-			return ERR_PUBLISH_NO_BODY;
+			ret = ERR_PUBLISH_NO_BODY;
+			goto error;
 		}
 	}
 	else
@@ -541,8 +576,7 @@ insert:
 			LM_ERR("while allocating memory\n");
 			if (dbmode!=PUA_DB_ONLY) 
 				lock_release(&HashT->p_records[hash_code].lock);
-			free_results_puadb(res);
-			return -1;
+			goto error;
 		}
 		memcpy(etag.s, presentity->etag.s, presentity->etag.len);
 		etag.len= presentity->etag.len;
@@ -650,26 +684,18 @@ send_publish:
 		goto error;
 	}
 
-	pkg_free(str_hdr);
+done:
+	ret = 0;
 
-	if( body && ret_code)
-	{
-		if(body->s)
-			xmlFree(body->s);
-		pkg_free(body);
-	}	
-	if(etag.s)
-		pkg_free(etag.s);
-	if(tuple_id)
+	if (dbmode == PUA_DB_ONLY && pua_dbf.end_transaction)
 	{
-		if(tuple_id->s)
-			pkg_free(tuple_id->s);
-		pkg_free(tuple_id);
+		if (pua_dbf.end_transaction(pua_db) < 0)
+		{
+			LM_ERR("in end_transaction\n");
+			goto error;
+		}
 	}
 
-	free_results_puadb(res);
-	return 0;
-
 error:
 	if(etag.s)
 		pkg_free(etag.s);
@@ -692,7 +718,14 @@ error:
 		pkg_free(tuple_id);
 	}
 	free_results_puadb(res);
-	return -1;
+
+	if (dbmode == PUA_DB_ONLY && pua_dbf.abort_transaction)
+	{
+		if (pua_dbf.abort_transaction(pua_db) < 0)
+			LM_ERR("in abort_transaction\n");
+	}
+
+	return ret;
 }
 
 ua_pres_t* publish_cbparam(publ_info_t* publ,str* body,str* tuple_id,

modules_k/pua/send_subscribe.c

@@ -297,6 +297,16 @@ void subs_cback_func(struct cell *t, int cb_type, struct tmcb_params *ps)
 		LM_ERR("null callback parameter\n");
 		return;
 	}
+
+	if (dbmode == PUA_DB_ONLY && pua_dbf.start_transaction)
+	{
+		if (pua_dbf.start_transaction(pua_db) < 0)
+		{
+			LM_ERR("in start_transaction\n");
+			goto error;
+		}
+	}
+
 	LM_DBG("completed with status %d\n",ps->code) ;
 	hentity= (ua_pres_t*)(*ps->param);
 	hash_code= core_hash(hentity->pres_uri,hentity->watcher_uri,
@@ -676,6 +686,16 @@ done:
 		hentity->flag= flag;
 		run_pua_callbacks( hentity, msg);
 	}
+
+	if (dbmode == PUA_DB_ONLY && pua_dbf.end_transaction)
+	{
+		if (pua_dbf.end_transaction(pua_db) < 0)
+		{
+			LM_ERR("in end_transaction\n");
+			goto error;
+		}
+	}
+
 	goto end;
 
 error:	
@@ -685,6 +705,12 @@ error:
 	 	shm_free(presentity);
 	}
 
+	if (dbmode == PUA_DB_ONLY && pua_dbf.abort_transaction)
+	{
+		if (pua_dbf.abort_transaction(pua_db) < 0)
+			LM_ERR("in abort_transaction\n");
+	}
+
 end:
 
 	if(hentity)
@@ -889,7 +915,7 @@ int send_subscribe(subs_info_t* subs)
 	ua_pres_t* presentity= NULL;
 	str met= {"SUBSCRIBE", 9};
 	str* str_hdr= NULL;
-	int ret= 0;
+	int ret= -1;
 	unsigned int hash_code=0;
 	ua_pres_t* hentity= NULL;
 	int expires;
@@ -926,6 +952,15 @@ int send_subscribe(subs_info_t* subs)
 		return -1;
 	}
 
+	if (dbmode == PUA_DB_ONLY && pua_dbf.start_transaction)
+	{
+		if (pua_dbf.start_transaction(pua_db) < 0)
+		{
+			LM_ERR("in start_transaction\n");
+			goto error;
+		}
+	}
+
 	/* generation of hash and getting lock moved from here to further down */
 
 	if (dbmode==PUA_DB_ONLY)
@@ -980,8 +1015,7 @@ insert:
 		{
 			LM_ERR("while building callback"
 					" param\n");
-			ret= -1;
-			goto done;
+			goto error;
 		}
 		hentity->flag= flag;
 
@@ -1005,7 +1039,11 @@ insert:
 				uac_r.dialog = 0;
 			}
 			shm_free(hentity);
-			goto  done;
+
+			/* Although this is an error must not return -1 as the
+			   calling function must continue processing. */
+			ret = 0;
+			goto error;
 		}
 
 		/* Now create a temporary hash table entry.
@@ -1022,7 +1060,7 @@ insert:
 		if(presentity== NULL)
 		{
 			LM_ERR("no more share memory\n");
-			goto done;
+			goto error;
 		}
 		memset(presentity, 0, size);
 		size= sizeof(ua_pres_t);
@@ -1111,10 +1149,9 @@ insert:
 		if(td== NULL)
 		{
 			LM_ERR("while building tm dlg_t structure");
-			ret= -1;
 			if (dbmode!=PUA_DB_ONLY)
 				lock_release(&HashT->p_records[hash_code].lock);
-			goto done;
+			goto error;
 		}
 				
 		hentity= subs_cbparam_indlg(presentity, expires, REQ_OTHER);
@@ -1123,8 +1160,7 @@ insert:
 			LM_ERR("while building callback param\n");
 			if (dbmode!=PUA_DB_ONLY)
 				lock_release(&HashT->p_records[hash_code].lock);
-			ret= -1;
-			goto done;
+			goto error;
 		}
 		if (dbmode!=PUA_DB_ONLY)
 			lock_release(&HashT->p_records[hash_code].lock);
@@ -1139,14 +1175,33 @@ insert:
 			shm_free(hentity);
 			hentity= NULL;
 			LM_ERR("while sending request with t_request\n");
-			goto done;
+			goto error;
 		}
 	}
 
 
 done:
+	if (dbmode == PUA_DB_ONLY && pua_dbf.end_transaction)
+	{
+		if (pua_dbf.end_transaction(pua_db) < 0)
+		{
+			LM_ERR("in end_transaction\n");
+			goto error;
+		}
+	}
+
+	ret = 0;
+
+error:
 	pua_free_tm_dlg(td);
 	pkg_free(str_hdr);
 	free_results_puadb(res);
+
+	if (dbmode == PUA_DB_ONLY && pua_dbf.abort_transaction)
+	{
+		if (pua_dbf.abort_transaction(pua_db) < 0)
+			LM_ERR("in abort_transaction\n");
+	}
+
 	return ret;
 }