add basic locking to callmaster struct

daemon/call.c

@@ -557,7 +557,9 @@ static void callmaster_timer(void *ptr) {
 	ZERO(hlp);
 	po = m->poller;
 
+	rwlock_lock_r(&m->lock);
 	g_hash_table_foreach(m->callhash, call_timer_iterator, &hlp);
+	rwlock_unlock_r(&m->lock);
 
 	memcpy(&m->stats, &m->statsps, sizeof(m->stats));
 	ZERO(m->statsps);
@@ -608,6 +610,7 @@ struct callmaster *callmaster_new(struct poller *p) {
 	if (!c->callhash)
 		goto fail;
 	c->poller = p;
+	rwlock_init(&c->lock);
 
 	poller_timer(p, callmaster_timer, &c->obj);
 
@@ -1183,7 +1186,9 @@ static void call_destroy(struct call *c) {
 	struct callmaster *m = c->callmaster;
 	struct callstream *s;
 
+	rwlock_lock_w(&m->lock);
 	g_hash_table_remove(m->callhash, c->callid); /* steal this ref */
+	rwlock_unlock_w(&m->lock);
 
 	if (redis_delete)
 		redis_delete(c);
@@ -1329,14 +1334,27 @@ static struct call *call_create(const char *callid, struct callmaster *m) {
 struct call *call_get_or_create(const char *callid, const char *viabranch, struct callmaster *m) {
 	struct call *c;
 
+restart:
+	rwlock_lock_r(&m->lock);
 	c = g_hash_table_lookup(m->callhash, callid);
 	if (!c) {
+		rwlock_unlock_r(&m->lock);
 		/* completely new call-id, create call */
 		c = call_create(callid, m);
+		rwlock_lock_w(&m->lock);
+		if (g_hash_table_lookup(m->callhash, callid)) {
+			/* preempted */
+			rwlock_unlock_w(&m->lock);
+			obj_put(c);
+			goto restart;
+		}
 		g_hash_table_insert(m->callhash, c->callid, obj_get(c));
+		rwlock_unlock_w(&m->lock);
 	}
-	else
+	else {
 		obj_hold(c);
+		rwlock_unlock_r(&m->lock);
+	}
 
 	if (viabranch && !g_hash_table_lookup(c->branches, viabranch))
 		g_hash_table_insert(c->branches, strdup(viabranch), (void *) 0x1);
@@ -1433,14 +1451,17 @@ char *call_lookup_udp(const char **out, struct callmaster *m) {
 	int num;
 	char *ret;
 
+	rwlock_lock_r(&m->lock);
 	c = g_hash_table_lookup(m->callhash, out[RE_UDP_UL_CALLID]);
 	if (!c || !g_hash_table_lookup(c->branches, out[RE_UDP_UL_VIABRANCH])) {
+		rwlock_unlock_r(&m->lock);
 		mylog(LOG_WARNING, LOG_PREFIX_CI "Got UDP LOOKUP for unknown call-id or unknown via-branch",
 			out[RE_UDP_UL_CALLID], out[RE_UDP_UL_VIABRANCH]);
 		asprintf(&ret, "%s 0 " IPF "\n", out[RE_UDP_COOKIE], IPP(m->ipv4));
 		return ret;
 	}
 	obj_hold(c);
+	rwlock_unlock_r(&m->lock);
 
 	c->log_info = out[RE_UDP_UL_CALLID];
 	strdupfree(&c->called_agent, "UNKNOWN(udp)");
@@ -1499,13 +1520,15 @@ char *call_lookup(const char **out, struct callmaster *m) {
 	int num;
 	char *ret;
 
+	rwlock_lock_r(&m->lock);
 	c = g_hash_table_lookup(m->callhash, out[RE_TCP_RL_CALLID]);
 	if (!c) {
+		rwlock_unlock_r(&m->lock);
 		mylog(LOG_WARNING, LOG_PREFIX_C "Got LOOKUP for unknown call-id", out[RE_TCP_RL_CALLID]);
 		return NULL;
 	}
-
 	obj_hold(c);
+	rwlock_unlock_r(&m->lock);
 
 	strdupfree(&c->called_agent, out[RE_TCP_RL_AGENT] ? : "UNKNOWN");
 	info_parse(out[RE_TCP_RL_INFO], &c->infohash);
@@ -1533,12 +1556,16 @@ char *call_delete_udp(const char **out, struct callmaster *m) {
 	DBG("got delete for callid '%s' and viabranch '%s'", 
 		out[RE_UDP_D_CALLID], out[RE_UDP_D_VIABRANCH]);
 
+	rwlock_lock_r(&m->lock);
 	c = g_hash_table_lookup(m->callhash, out[RE_UDP_D_CALLID]);
 	if (!c) {
+		rwlock_unlock_r(&m->lock);
 		mylog(LOG_INFO, LOG_PREFIX_C "Call-ID to delete not found", out[RE_UDP_D_CALLID]);
 		goto err;
 	}
 	obj_hold(c);
+	rwlock_unlock_r(&m->lock);
+
 	c->log_info = out[RE_UDP_D_VIABRANCH];
 
 	if (out[RE_UDP_D_FROMTAG] && *out[RE_UDP_D_FROMTAG]) {
@@ -1605,11 +1632,15 @@ out:
 void call_delete(const char **out, struct callmaster *m) {
 	struct call *c;
 
+	rwlock_lock_r(&m->lock);
 	c = g_hash_table_lookup(m->callhash, out[RE_TCP_D_CALLID]);
-	if (!c)
+	if (!c) {
+		rwlock_unlock_r(&m->lock);
 		return;
-
+	}
 	obj_hold(c);
+	rwlock_unlock_r(&m->lock);
+
 	/* delete whole list, as we don't have branches in tcp controller */
 	call_destroy(c);
 	obj_put(c);
@@ -1673,6 +1704,7 @@ static void call_status_iterator(void *key, void *val, void *ptr) {
 }
 
 void calls_status(struct callmaster *m, struct control_stream *s) {
+	rwlock_lock_r(&m->lock);
 	streambuf_printf(s->outbuf, "proxy %u %llu/%llu/%llu\n",
 		g_hash_table_size(m->callhash),
 		(long long unsigned int) m->stats.bytes,
@@ -1680,6 +1712,7 @@ void calls_status(struct callmaster *m, struct control_stream *s) {
 		(long long unsigned int) m->stats.bytes * 2 - m->stats.errors);
 
 	g_hash_table_foreach(m->callhash, call_status_iterator, s);
+	rwlock_unlock_r(&m->lock);
 }
 
 

daemon/call.h

@@ -11,6 +11,7 @@
 #include "control.h"
 #include "control_udp.h"
 #include "obj.h"
+#include "aux.h"
 
 struct poller;
 struct control_stream;
@@ -96,6 +97,7 @@ struct call {
 struct callmaster {
 	struct obj		obj;
 
+	rwlock_t		lock;
 	GHashTable		*callhash;
 	u_int16_t		lastport;
 	struct stats		statsps;