aaa_diameter: Add management for "unreplied requests"

This becomes necessary both when the event_route is not defined, as well
as in various error cases, e.g. script writer omits to invoke
dm_send_answer() or an internal error occurs before the answer is built.

Finally, make sure to protect the requests list, since it is actually
managed by multiple threads part of the "Diameter peer" OpenSIPS worker:

* dm_update_unreplied_req() - called by the peer's Server Thread
    (i.e. "put a freshly received request on hold")
* dm_remove_unreplied_req() - called by the peer's Main Thread
    (i.e. "the script writer just built a new reply, msg can be freed")

modules/aaa_diameter/aaa_impl.c

@@ -40,6 +40,20 @@ struct local_rules_definition {
 	int max;
 };
 
+struct fd_msg_list {
+	struct msg *req;
+	unsigned int timeout_jf;
+	struct list_head list;
+};
+
+/* for now, the purpose of this list is only to avoid dangling Diameter
+ * requests (server-side), in case the opensips.cfg has no event_route
+ * to consume them */
+struct list_head dm_unreplied_req;
+gen_lock_t dm_unreplied_req_lk;
+unsigned int dm_unreplied_req_timeout = 120; /* sec */
+
+
 struct _dm_dict dm_dict;
 
 /* Workaround until we find a way of looking up a single enum val in fD */
@@ -140,6 +154,70 @@ int init_mutex_cond(pthread_mutex_t *mutex, pthread_cond_t *cond)
 }
 
 
+static inline void dm_update_unreplied_req(struct msg *req)
+{
+	struct list_head *it, *aux;
+	struct fd_msg_list *rit, *ml;
+	unsigned int now = get_ticks();
+
+	lock_get(&dm_unreplied_req_lk);
+
+	list_for_each_safe (it, aux, &dm_unreplied_req) {
+		rit = list_entry(it, struct fd_msg_list, list);
+
+		if (rit->timeout_jf <= now) {
+			LM_DBG("Diameter request timeout (unhandled), cleaning up\n");
+			list_del(&rit->list);
+			fd_msg_free(rit->req);
+			pkg_free(rit);
+		} else {
+			break;
+		}
+	}
+
+	lock_release(&dm_unreplied_req_lk);
+
+	ml = pkg_malloc(sizeof *ml);
+	if (!ml) {
+		LM_ERR("oom\n");
+		return;
+	}
+	memset(ml, 0, sizeof *ml);
+
+	ml->req = req;
+	ml->timeout_jf = get_ticks() + dm_unreplied_req_timeout;
+
+	lock_get(&dm_unreplied_req_lk);
+	list_add_tail(&ml->list, &dm_unreplied_req);
+	lock_release(&dm_unreplied_req_lk);
+}
+
+
+int dm_remove_unreplied_req(struct msg *req)
+{
+	struct list_head *it, *aux;
+	struct fd_msg_list *rit;
+
+	lock_get(&dm_unreplied_req_lk);
+
+	list_for_each_safe (it, aux, &dm_unreplied_req) {
+		rit = list_entry(it, struct fd_msg_list, list);
+
+		if (rit->req == req) {
+			list_del(&rit->list);
+			lock_release(&dm_unreplied_req_lk);
+			LM_DBG("matched unreplied req, removing from list\n");
+			pkg_free(rit);
+			return 0;
+		}
+	}
+
+	lock_release(&dm_unreplied_req_lk);
+	LM_DBG("failed to match unreplied req (already cleaned up?!)\n");
+	return -1;
+}
+
+
 static int dm_acc_reply(struct msg ** msg, struct avp * avp, struct session * sess, void * data, enum disp_action * act)
 {
 	struct avp *a = NULL;
@@ -469,11 +547,13 @@ static int dm_receive_req(struct msg **_msg, struct avp * avp, struct session *
 
 	init_str(&avp_arr, cJSON_PrintUnformatted(avps));
 
+	/* keep the request for a while in order to be able to generate the answer */
+	dm_update_unreplied_req(msg);
+
 	if (dm_dispatch_event_req(msg, &tid, hdr->msg_appl, hdr->msg_code, &avp_arr))
 		LM_ERR("failed to dispatch DM Request (tid: %.*s, %d/%d)\n", tid.len,
 		        tid.s, hdr->msg_appl, hdr->msg_code);
 
-	/* for now, keep the message in order to be able to generate the reply */
 	goto out;
 
 error:

modules/aaa_diameter/aaa_impl.h

@@ -137,10 +137,14 @@ struct dm_cond {
 };
 int init_mutex_cond(pthread_mutex_t *mutex, pthread_cond_t *cond);
 
+extern struct list_head dm_unreplied_req;
+extern gen_lock_t dm_unreplied_req_lk;
+extern unsigned int dm_unreplied_req_timeout;
 extern char *dm_conf_filename;
 extern char *extra_avps_file;
 extern struct _dm_dict dm_dict;
 extern int dm_answer_timeout;
+int dm_remove_unreplied_req(struct msg *req);
 
 int freeDiameter_init(void);
 

modules/aaa_diameter/peer.c

@@ -62,6 +62,9 @@ int dm_init_peer(void)
 	msg_send_cond = &wrap->cond;
 
 	init_mutex_cond(msg_send_lk, msg_send_cond);
+
+	INIT_LIST_HEAD(&dm_unreplied_req);
+	lock_init(&dm_unreplied_req_lk);
 	return 0;
 }
 
@@ -510,6 +513,12 @@ static int dm_custom_rpl(struct dm_message *dm)
 	struct msg *ans = (struct msg *)dm->fd_req;
 	int rc;
 
+	if (dm_remove_unreplied_req(ans) != 0) {
+		LM_ERR("unable to build answer, request is no longer available "
+		        "(timeout: %d s)\n", dm_unreplied_req_timeout);
+		return -1;
+	}
+
 	rc = fd_msg_new_answer_from_req(fd_g_config->cnf_dict, &ans, 0);
 	if (rc != 0) {
 		LM_ERR("failed to create answer message, error: %d\n", rc);