b2b_entities: add support for locked_by per entities

Author: razvancrainea

modules/b2b_entities/b2b_entities.c

@@ -35,6 +35,7 @@
 #include "../../dprint.h"
 #include "../../error.h"
 #include "../../ut.h"
+#include "../../pt.h"
 #include "../../script_cb.h"
 #include "../../parser/parse_from.h"
 #include "../dialog/dlg_load.h"
@@ -416,6 +417,7 @@ int b2b_update_b2bl_param(enum b2b_entity_type type, str* key,
 	b2b_dlg_t* dlg;
 	b2b_table table;
 	unsigned int hash_index, local_index;
+	int unlock = 1;
 
 	if(!param)
 	{
@@ -442,17 +444,23 @@ int b2b_update_b2bl_param(enum b2b_entity_type type, str* key,
 		LM_ERR("Wrong format for b2b key [%.*s]\n", key->len, key->s);
 		return -1;
 	}
-	lock_get(&table[hash_index].lock);
+	if (table[hash_index].locked_by != process_no)
+		lock_get(&table[hash_index].lock);
+	else
+		unlock = 0;
+
 	dlg = b2b_search_htable(table, hash_index, local_index);
 	if(dlg == NULL)
 	{
 		LM_ERR("No dialog found\n");
-		lock_release(&table[hash_index].lock);
+		if (unlock)
+			lock_release(&table[hash_index].lock);
 		return -1;
 	}
 	memcpy(dlg->param.s, param->s, param->len);
 	dlg->param.len = param->len;
-	lock_release(&table[hash_index].lock);
+	if (unlock)
+		lock_release(&table[hash_index].lock);
 
 	if (b2be_cluster)
 		replicate_entity_update(dlg, type, hash_index, param, NULL);

modules/b2b_entities/b2be_clustering.c

@@ -37,6 +37,7 @@
 #include "b2be_clustering.h"
 #include "b2b_entities.h"
 #include "b2be_db.h"
+#include "../../pt.h"
 
 #define NO_REPL_CONSTANT_STRS 10
 
@@ -453,9 +454,12 @@ int receive_entity_create(bin_packet_t *packet, b2b_dlg_t *dlg, int type,
 		new_dlg->legs = new_leg;
 
 	lock_get(&htable[h_idx].lock);
+	htable[h_idx].locked_by = process_no;
 
 	b2b_run_cb(new_dlg, type, B2BCB_RECV_EVENT, B2B_EVENT_CREATE, packet);
 
+	htable[h_idx].locked_by = -1;
+
 	new_key = b2b_htable_insert(htable, new_dlg, hash_index, type, 1, 1);
 	if (new_key == NULL) {
 		LM_ERR("Failed to insert new record\n");
@@ -547,7 +551,9 @@ int receive_entity_update(bin_packet_t *packet)
 	if (packet->type == REPL_ENTITY_UPDATE) {
 		unpack_update_fields(packet, dlg);
 
+		htable[hash_index].locked_by = process_no;
 		b2b_run_cb(dlg, type, B2BCB_RECV_EVENT, B2B_EVENT_UPDATE, packet);
+		htable[hash_index].locked_by = -1;
 	} else {
 		rc = recv_b2bl_param_update(packet, dlg);
 	}
@@ -601,7 +607,9 @@ int receive_entity_delete(bin_packet_t *packet)
 		return 0;
 	}
 
+	htable[hash_index].locked_by = process_no;
 	b2b_run_cb(dlg, type, B2BCB_RECV_EVENT, B2B_EVENT_DELETE, packet);
+	htable[hash_index].locked_by = -1;
 
 	b2b_entity_db_delete(type, dlg);
 	b2b_delete_record(dlg, htable, hash_index);

modules/b2b_entities/dlg.h

@@ -117,6 +117,7 @@ typedef struct b2b_entry
 {
 	b2b_dlg_t* first;
 	gen_lock_t lock;
+	int locked_by;
 	int checked;
 }b2b_entry_t;