implement delayed call/branch deletion

Squashed commit of the following:

commit 5539a3dcb4
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Wed Aug 27 08:02:00 2014 -0400

    update redis mod version

commit fa372760a2
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Tue Aug 26 15:42:00 2014 -0400

    delay branch deletion

commit 87af9f48f5
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Tue Aug 26 13:52:28 2014 -0400

    delay deletion of entire call

daemon/call.c

@@ -34,12 +34,20 @@
 
 
 
+#ifndef DELETE_DELAY
+#define DELETE_DELAY 30
+#endif
+
+
+
+
 
 typedef int (*rewrite_func)(str *, struct packet_stream *);
 
 /* also serves as array index for callstream->peers[] */
 struct iterator_helper {
-	GSList			*del;
+	GSList			*del_timeout;
+	GSList			*del_scheduled;
 	struct stream_fd	*ports[0x10000];
 };
 struct xmlrpc_helper {
@@ -287,6 +295,7 @@ static const struct mediaproxy_srtp __mps_null = {
 static void unkernelize(struct packet_stream *);
 static void __stream_unkernelize(struct packet_stream *ps);
 static void stream_unkernelize(struct packet_stream *ps);
+static void __monologue_destroy(struct call_monologue *monologue);
 
 
 
@@ -871,6 +880,53 @@ done:
 
 
 
+/* called with call->master_lock held in R */
+static int call_timer_delete_monologues(struct call *c) {
+	GSList *i;
+	struct call_monologue *ml;
+	int ret = 0;
+	time_t min_deleted = 0;
+
+	/* we need a write lock here */
+	rwlock_unlock_r(&c->master_lock);
+	rwlock_lock_w(&c->master_lock);
+
+	for (i = c->monologues; i; i = i->next) {
+		ml = i->data;
+
+		if (!ml->deleted)
+			continue;
+		if (ml->deleted > poller_now) {
+			if (!min_deleted || ml->deleted < min_deleted)
+				min_deleted = ml->deleted;
+			continue;
+		}
+
+		__monologue_destroy(ml);
+		ml->deleted = 0;
+
+		if (!g_hash_table_size(c->tags)) {
+			ilog(LOG_INFO, "Call branch '"STR_FORMAT"' deleted, no more branches remaining",
+					STR_FMT(&ml->tag));
+			ret = 1; /* destroy call */
+			goto out;
+		}
+
+		ilog(LOG_INFO, "Call branch "STR_FORMAT" deleted",
+				STR_FMT(&ml->tag));
+	}
+
+out:
+	c->ml_deleted = min_deleted;
+
+	rwlock_unlock_w(&c->master_lock);
+	rwlock_lock_r(&c->master_lock);
+
+	return ret;
+}
+
+
+
 /* called with callmaster->hashlock held */
 static void call_timer_iterator(void *key, void *val, void *ptr) {
 	struct call *c = val;
@@ -883,6 +939,16 @@ static void call_timer_iterator(void *key, void *val, void *ptr) {
 	struct stream_fd *sfd;
 
 	rwlock_lock_r(&c->master_lock);
+	log_info_call(c);
+
+	if (c->deleted && poller_now >= c->deleted
+			&& c->last_signal <= c->deleted)
+		goto delete;
+
+	if (c->ml_deleted && poller_now >= c->ml_deleted) {
+		if (call_timer_delete_monologues(c))
+			goto delete;
+	}
 
 	if (!c->streams)
 		goto drop;
@@ -925,17 +991,19 @@ next:
 	if (good)
 		goto out;
 
-	log_info_call(c);
-	ilog(LOG_INFO, "Closing call branch due to timeout");
-	log_info_clear();
+	ilog(LOG_INFO, "Closing call due to timeout");
 
 drop:
-	rwlock_unlock_r(&c->master_lock);
-	hlp->del = g_slist_prepend(hlp->del, obj_get(c));
-	return;
+	hlp->del_timeout = g_slist_prepend(hlp->del_timeout, obj_get(c));
+	goto out;
+
+delete:
+	hlp->del_scheduled = g_slist_prepend(hlp->del_scheduled, obj_get(c));
+	goto out;
 
 out:
 	rwlock_unlock_r(&c->master_lock);
+	log_info_clear();
 }
 
 void xmlrpc_kill_calls(void *p) {
@@ -1033,11 +1101,10 @@ void kill_calls_timer(GSList *list, struct callmaster *m) {
 	struct xmlrpc_helper *xh = NULL;
 
 	if (!list)
-		return; /* shouldn't happen */
+		return;
 
-	ca = list->data;
-	m = ca->callmaster; /* same callmaster for all of them */
-	url = m->conf.b2b_url;
+	/* if m is NULL, it's the scheduled deletions, otherwise it's the timeouts */
+	url = m ? m->conf.b2b_url : NULL;
 	if (url) {
 		xh = g_slice_alloc(sizeof(*xh));
 		xh->c = g_string_chunk_new(64);
@@ -1187,10 +1254,8 @@ next:
 		if (hlp.ports[j])
 			obj_put(hlp.ports[j]);
 
-	if (!hlp.del)
-		return;
-
-	kill_calls_timer(hlp.del, m);
+	kill_calls_timer(hlp.del_scheduled, NULL);
+	kill_calls_timer(hlp.del_timeout, m);
 }
 #undef DS
 
@@ -1903,6 +1968,7 @@ int monologue_offer_answer(struct call_monologue *monologue, GQueue *streams,
 	struct endpoint_map *em;
 
 	monologue->call->last_signal = poller_now;
+	monologue->call->deleted = 0;
 
 	/* we must have a complete dialogue, even though the to-tag (other_ml->tag)
 	 * may not be known yet */
@@ -2421,6 +2487,9 @@ static void __monologue_unkernelize(struct call_monologue *monologue) {
 	if (!monologue)
 		return;
 
+	monologue->deleted = 0; /* not really related, but indicates activity, so cancel
+				   any pending deletion */
+
 	for (l = monologue->medias.head; l; l = l->next) {
 		media = l->data;
 
@@ -2582,16 +2651,17 @@ int call_delete_branch(struct callmaster *m, const str *callid, const str *branc
 	}
 */
 
-	__monologue_destroy(ml);
-	if (g_hash_table_size(c->tags)) {
-		ilog(LOG_INFO, "Call branch deleted (other branches still active)");
-		goto success_unlock;
-	}
+	ilog(LOG_INFO, "Scheduling deletion of call branch '"STR_FORMAT"' in %d seconds",
+			STR_FMT(&ml->tag), DELETE_DELAY);
+	ml->deleted = poller_now + 30;
+	if (!c->ml_deleted || c->ml_deleted > ml->deleted)
+		c->ml_deleted = ml->deleted;
+	goto success_unlock;
 
 del_all:
+	ilog(LOG_INFO, "Scheduling deletion of entire call in %d seconds", DELETE_DELAY);
+	c->deleted = poller_now + DELETE_DELAY;
 	rwlock_unlock_w(&c->master_lock);
-	ilog(LOG_INFO, "Deleting full call");
-	call_destroy(c);
 	goto success;
 
 success_unlock:

daemon/call.h

@@ -278,6 +278,7 @@ struct call_monologue {
 
 	str			tag;	
 	time_t			created;	/* RO */
+	time_t			deleted;
 	GHashTable		*other_tags;
 	struct call_monologue	*active_dialogue;
 
@@ -304,6 +305,8 @@ struct call {
 	str			callid;	
 	time_t			created;
 	time_t			last_signal;
+	time_t			deleted;
+	time_t			ml_deleted;
 	unsigned char		tos;
 };
 

daemon/main.c

@@ -28,6 +28,11 @@
 
 
 
+#define REDIS_MODULE_VERSION "redis/5"
+
+
+
+
 #define die(x...) do { fprintf(stderr, x); exit(-1); } while(0)
 #define dlresolve(n) do {								\
 	n ## _mod = dlsym(dlh, "mod_" #n);						\
@@ -516,7 +521,7 @@ no_kernel:
 		if (!dlh)
 			die("Failed to open redis plugin, aborting (%s)\n", dlerror());
 		strp = dlsym(dlh, "__module_version");
-		if (!strp || !*strp || strcmp(*strp, "redis/4"))
+		if (!strp || !*strp || strcmp(*strp, REDIS_MODULE_VERSION))
 			die("Incorrect redis module version: %s\n", *strp);
 		redis_mod_verify(dlh);
 		mc.redis = redis_new_mod(redis_ip, redis_port, redis_db);