From: Marco Capetta <mcapetta@sipwise.com>
Date: Mon, 16 Dec 2019 14:15:50 +0100
Subject: [PATCH] tm: new weight-based call-termination distribution *
 Modified funtion t_load_contact adding the 'mode' parameter:   - if equals
 to 0 or missing,
 it behaves as before (standard behavior)   - if equals to 1,
 contacts are ordered using a proportional algorithm   that use to q-value to
 calculate the weight of each contact. Higher   is the q value,
 higher is the probability to be contacted as first.

 * Added new function ki_t_load_contact_mode to be able to set the
   'mode' paramenter also from KEMI.

 * Additionally a small code refactory has been done in order to
   keep the code easier to read and to avoid duplications.

Change-Id: Ic39a70e43004422e2b6e5db4788df68d9e223129
---
 src/modules/tm/doc/functions.xml |  33 +++--
 src/modules/tm/t_serial.c        | 265 +++++++++++++++++++++++++++++++--------
 src/modules/tm/t_serial.h        |   4 +-
 src/modules/tm/tm.c              |   9 +-
 4 files changed, 252 insertions(+), 59 deletions(-)

diff --git a/src/modules/tm/doc/functions.xml b/src/modules/tm/doc/functions.xml
index 489147a..1935fe3 100644
--- a/src/modules/tm/doc/functions.xml
+++ b/src/modules/tm/doc/functions.xml
@@ -1154,7 +1154,7 @@ failure_route[1] {
 
 	<section id="tm.f.t_load_contacts">
 		<title>
-		<function moreinfo="none">t_load_contacts()</function>
+		<function moreinfo="none">t_load_contacts([mode])</function>
 		</title>
 		<para>
 		  This is the first of the three functions that can be used
@@ -1178,18 +1178,35 @@ failure_route[1] {
 		</para>
 		<para>
 		  If the current destination set contains more than one branch,
-		  the function sorts them
-		  according to increasing value of the q parameter and
-		  then stores the branches in reverse order into the XAVP.
+		  the function sorts them according to the algorithm selected
+		  with the 'mode' paramenter and then stores the branches in
+		  reverse order into the XAVP.
 		</para>
 		<para>
 		  The q parameter of a branch contains a value from range 0-1.0
 		  and it  expresses relative preference of the branch
 		  among all branches in the destination set.
 		  The higher the q value the more preference the
-		  user agent gave to the branch. Branches with higher q
-		  values will be tried before branches with lower ones
-		  when serial forking takes place.
+		  user agent gave to the branch.
+		  The parameter 'mode' controls which type of algorithm kamailio
+		  will apply to sort the branches based on the q values:
+		</para>
+		<para>
+		  - '0' or missing: branches with higher q values will be tried
+		  before branches with lower ones when serial forking takes place
+		  Equals q values will result in a parallel forking.
+		</para>
+		<para>
+		  - '1': branches are ordered using a proportional algorithm
+          that uses to q value as the weight of the branch. Higher is the
+          q value, higher is the probability to be tried as first. Equals
+          q values means equals probability to be tried. Differently from
+          the previous mode there is no possibility to have parallel forking.
+          Branches with q values equals to 0 or lower are not considered by
+          the ordering algorithm, but just added at the end of the list
+          as backup option if all other branches fail.
+          This algorithm can be useful in case of endpoints in ACTIVE-ACTIVE
+          configuration to load-balance the calls.
 		</para>
 		<para>
 		  After calling <function>t_load_contacts()</function>, function
@@ -1197,7 +1214,7 @@ failure_route[1] {
 		  also <function>t_next_contact_flow()</function> need
 		  to be called
 		  one or more times in order to retrieve the branches based
-		  on their q value.
+		  on their order.
 		</para>
 		<para>
 		  Function returns 1 if loading of contacts succeeded or
diff --git a/src/modules/tm/t_serial.c b/src/modules/tm/t_serial.c
index 68b8afc..8592a9c 100644
--- a/src/modules/tm/t_serial.c
+++ b/src/modules/tm/t_serial.c
@@ -33,6 +33,8 @@
 #include "../../core/parser/msg_parser.h"
 #include "../../core/ut.h"
 #include "../../core/xavp.h"
+#include "../../core/sr_module.h"
+#include "../../core/rand/kam_rand.h"
 #include "config.h"
 #include "t_funcs.h"
 #include "t_reply.h"
@@ -40,6 +42,9 @@
 
 /* usr_avp flag for sequential forking */
 #define Q_FLAG      (1<<2)
+/* t_load_contacts modes/algorithms */
+#define T_LOAD_STANDARD     0
+#define T_LOAD_PROPORTIONAL 1
 
 extern str ulattrs_xavp_name;
 
@@ -57,6 +62,7 @@ struct contact {
 	unsigned short q_flag;
 	struct contact *next;
 	sr_xavp_t *ulattrs;
+	unsigned short q_index;
 };
 
 struct instance_list {
@@ -101,7 +107,8 @@ static str ua_name = {"ua", 2};
 
 void add_contacts_avp(str *uri, str *dst_uri, str *path, str *sock_str,
 		unsigned int flags, unsigned int q_flag, str *instance,
-		str *ruid, str *location_ua, sr_xavp_t *ulattrs_xavp)
+		str *ruid, str *location_ua, sr_xavp_t *ulattrs_xavp,
+		sr_xavp_t **pxavp)
 {
 	sr_xavp_t *record;
 	sr_xval_t val;
@@ -159,26 +166,168 @@ void add_contacts_avp(str *uri, str *dst_uri, str *path, str *sock_str,
 
 	val.type = SR_XTYPE_XAVP;
 	val.v.xavp = record;
-	if(xavp_add_value(&contacts_avp, &val, NULL)==NULL) {
-		/* failed to add xavps to root list */
-		LM_ERR("failed to add xavps to root list\n");
-		xavp_destroy_list(&record);
+	if(pxavp) {
+		if((*pxavp = xavp_add_value_after(&contacts_avp, &val, *pxavp))==NULL) {
+			/* failed to add xavps to the end of the list */
+			LM_ERR("failed to add xavps to the end of the list\n");
+			xavp_destroy_list(&record);
+		}
+	}
+	else {
+		if(xavp_add_value(&contacts_avp, &val, NULL)==NULL) {
+			/* failed to add xavps to root list */
+			LM_ERR("failed to add xavps to root list\n");
+			xavp_destroy_list(&record);
+		}
 	}
 }
 
-/* 
+/*
+ * Socket preparation for 'add_contacts_avp' function
+ */
+int add_contacts_avp_preparation(struct contact *curr, char *sock_buf, sr_xavp_t **pxavp)
+{
+	str sock_str;
+	int len;
+
+	if (curr->sock) {
+		len = MAX_SOCKET_STR - 1;
+		if (socket2str(sock_buf, &len, curr->sock) < 0) {
+			LM_ERR("failed to convert socket to str\n");
+			return -1;
+		}
+		sock_buf[len] = 0;
+		sock_str.s = sock_buf;
+		sock_str.len = len + 1;
+	} else {
+		sock_str.s = 0;
+		sock_str.len = 0;
+	}
+
+	add_contacts_avp(&(curr->uri), &(curr->dst_uri), &(curr->path),
+			&sock_str, curr->flags, curr->q_flag,
+			&(curr->instance), &(curr->ruid), &(curr->location_ua),
+			curr->ulattrs, pxavp);
+
+	return 0;
+}
+
+/*
  * Loads contacts in destination set into contacts_avp in reverse
  * priority order and associated each contact with Q_FLAG telling if
- * contact is the last one in its priority class.  Finally, removes
- * all branches from destination set.
+ * contact is the last one in its priority class.
  */
-int ki_t_load_contacts(struct sip_msg* msg)
+int t_load_contacts_standard(struct contact *contacts, char *sock_buf)
+{
+	struct contact *curr;
+
+	/* Assign values for q_flags */
+	curr = contacts;
+	curr->q_flag = 0;
+	while (curr->next) {
+		if (curr->q < curr->next->q) {
+			curr->next->q_flag = Q_FLAG;
+		} else {
+			curr->next->q_flag = 0;
+		}
+		curr = curr->next;
+	}
+
+	/* Add contacts to contacts_avp */
+	curr = contacts;
+	while (curr) {
+		if (add_contacts_avp_preparation(curr, sock_buf, NULL) < 0) {
+			return -1;
+		}
+
+		curr = curr->next;
+	}
+
+	return 0;
+}
+
+/*
+ * Loads contacts in destination set into contacts_avp in reverse
+ * proportional order. Each contact is associated with Q_FLAG beacuse
+ * only one contact at a time has to ring.
+ */
+int t_load_contacts_proportional(struct contact *contacts, char *sock_buf, int n, unsigned short q_total)
+{
+	int q_remove, n_rand, idx;
+	struct contact *curr;
+	sr_xavp_t *lxavp = NULL;
+
+	/* Add contacts with q-value NOT equals to 0 and NOT negative to contacts_avp */
+	for (idx = 0; idx < n; idx++) {
+		q_remove = 0;
+
+		/* Generate a random number from 0 to (q_total -1) */
+		n_rand = kam_rand() % q_total;
+
+		curr = contacts;
+		while (curr) {
+			if (curr->q <= 0) {
+				curr = curr->next;
+				continue;
+			}
+
+			if (q_remove != 0) {
+				/* ALREADY FOUND */
+				curr->q_index -= q_remove;
+			}
+			else if (curr->q_index > n_rand) {
+				/* FOUND */
+				LM_DBG("proportionally selected contact with uri: %s (q: %d, random: %d, q_index: %d, q_total: %d)\n", curr->uri.s, curr->q, n_rand, curr->q_index, q_total);
+				q_remove = curr->q;
+				q_total -= q_remove;
+				curr->q_index -= q_remove;
+				curr->q_flag = Q_FLAG;
+
+				if (add_contacts_avp_preparation(curr, sock_buf, &lxavp) < 0) {
+					return -1;
+				}
+			}
+
+			curr = curr->next;
+		}
+	}
+
+	/* Add contacts with q-value equals to 0 or negative to contacts_avp */
+	curr = contacts;
+	while (curr) {
+		if (curr->q > 0) {
+			curr = curr->next;
+			continue;
+		}
+
+		LM_DBG("proportionally added backup contact with uri: %s (q: %d)\n", curr->uri.s, curr->q);
+		curr->q_flag = Q_FLAG;
+
+		if (add_contacts_avp_preparation(curr, sock_buf, &lxavp) < 0) {
+			return -1;
+		}
+
+		curr = curr->next;
+	}
+
+	return 0;
+}
+
+/*
+ * Loads contacts in destination set and process it. Then call
+ * 't_load_contacts_proportional' or 't_load_contacts_standard'
+ * function based on the selected ordering machanism. Finally,
+ * removes all branches from destination set.
+ */
+int ki_t_load_contacts_mode(struct sip_msg* msg, int mode)
 {
 	branch_t *branch;
-	str *ruri, sock_str;
+	str *ruri;
 	struct contact *contacts, *next, *prev, *curr;
-	int first_idx, idx, len;
+	int first_idx, idx;
 	char sock_buf[MAX_SOCKET_STR];
+	unsigned short q_total = 0;
+	int n_elements = 0;
 
 	/* Check if contacts_avp has been defined */
 	if (contacts_avp.len == 0) {
@@ -251,6 +400,17 @@ int ki_t_load_contacts(struct sip_msg* msg)
 		first_idx = 1;
 	}
 
+	contacts->q_index = contacts->q;
+	if (mode == T_LOAD_PROPORTIONAL) {
+		/* Save in q_index the index to check for the proportional order
+		   Don't consider elements with Q value 0 or negative */
+		if (contacts->q > 0) {
+			q_total += contacts->q;
+			n_elements += 1;
+		}
+		contacts->q_index = q_total;
+	}
+
 	contacts->next = (struct contact *)0;
 
 	/* Insert (remaining) branches to contact list in increasing q order */
@@ -283,13 +443,25 @@ int ki_t_load_contacts(struct sip_msg* msg)
 		{
 			next->ulattrs = xavp_get_by_index(&ulattrs_xavp_name, idx + 1, NULL);
 		}
+
+		next->q_index = next->q;
+		if (mode == T_LOAD_PROPORTIONAL) {
+			/* Save in q_index the index to check for the proportional order
+			   Don't consider elements with Q value 0 or negative */
+			if (next->q > 0) {
+				q_total += next->q;
+				n_elements += 1;
+			}
+			next->q_index = q_total;
+		}
+
 		next->next = (struct contact *)0;
 
 		prev = (struct contact *)0;
 		curr = contacts;
 		while (curr &&
-				((curr->q < next->q) ||
-				 ((curr->q == next->q) && (next->path.len == 0)))) {
+				((curr->q_index < next->q_index) ||
+				 ((curr->q_index == next->q_index) && (next->path.len == 0)))) {
 			prev = curr;
 			curr = curr->next;
 		}
@@ -306,43 +478,17 @@ int ki_t_load_contacts(struct sip_msg* msg)
 		}
 	}
 
-	/* Assign values for q_flags */
-	curr = contacts;
-	curr->q_flag = 0;
-	while (curr->next) {
-		if (curr->q < curr->next->q) {
-			curr->next->q_flag = Q_FLAG;
-		} else {
-			curr->next->q_flag = 0;
+	if (mode == T_LOAD_PROPORTIONAL) {
+		if (t_load_contacts_proportional(contacts, sock_buf, n_elements, q_total) < 0) {
+			free_contact_list(contacts);
+			return -1;
 		}
-		curr = curr->next;
 	}
-
-	/* Add contacts to contacts_avp */
-	curr = contacts;
-	while (curr) {
-
-		if (curr->sock) {
-			len = MAX_SOCKET_STR - 1;
-			if (socket2str(sock_buf, &len, curr->sock) < 0) {
-				LM_ERR("failed to convert socket to str\n");
-				free_contact_list(contacts);
-				return -1;
-			}
-			sock_buf[len] = 0;
-			sock_str.s = sock_buf;
-			sock_str.len = len + 1;
-		} else {
-			sock_str.s = 0;
-			sock_str.len = 0;
+	else {
+		if (t_load_contacts_standard(contacts, sock_buf) < 0) {
+			free_contact_list(contacts);
+			return -1;
 		}
-
-		add_contacts_avp(&(curr->uri), &(curr->dst_uri), &(curr->path),
-				&sock_str, curr->flags, curr->q_flag,
-				&(curr->instance), &(curr->ruid), &(curr->location_ua),
-				curr->ulattrs);
-
-		curr = curr->next;
 	}
 
 	/* Clear all branches */
@@ -357,9 +503,30 @@ int ki_t_load_contacts(struct sip_msg* msg)
 	return 1;
 }
 
-int t_load_contacts(struct sip_msg* msg, char* key, char* value)
+int ki_t_load_contacts(struct sip_msg* msg)
+{
+	return ki_t_load_contacts_mode(msg, T_LOAD_STANDARD);
+}
+
+int t_load_contacts(struct sip_msg* msg, char* mode, char* value)
 {
-	return ki_t_load_contacts(msg);
+	int i = T_LOAD_STANDARD;
+
+	if(mode) {
+		if(get_int_fparam(&i, msg, (fparam_t*)mode)<0) return -1;
+
+		if ((i != T_LOAD_STANDARD) && (i != T_LOAD_PROPORTIONAL)) {
+			LM_ERR("invalid load_contact mode: %d, please use 0 (standard) or 1 (proportional)\n", i);
+			return -1;
+		}
+		LM_DBG("load_contact mode selected: %d\n", i);
+	}
+	else
+	{
+		LM_DBG("load_contact mode not selected, using: %d\n", T_LOAD_STANDARD);
+	}
+
+	return ki_t_load_contacts_mode(msg, i);
 }
 
 void add_contact_flows_avp(str *uri, str *dst_uri, str *path, str *sock_str,
diff --git a/src/modules/tm/t_serial.h b/src/modules/tm/t_serial.h
index 3830413..b6bd0ea 100644
--- a/src/modules/tm/t_serial.h
+++ b/src/modules/tm/t_serial.h
@@ -30,12 +30,14 @@
 
 extern int fr_inv_timer_next;
 
-int t_load_contacts(struct sip_msg* msg, char* key, char* value);
+int t_load_contacts(struct sip_msg* msg, char* mode, char* value);
 
 int t_next_contacts(struct sip_msg* msg, char* key, char* value);
 
 int t_next_contact_flow(struct sip_msg* msg, char* key, char* value);
 
+int ki_t_load_contacts_mode(struct sip_msg* msg, int mode);
+
 int ki_t_load_contacts(struct sip_msg* msg);
 
 int ki_t_next_contacts(struct sip_msg* msg);
diff --git a/src/modules/tm/tm.c b/src/modules/tm/tm.c
index 98e37fa..01b818f 100644
--- a/src/modules/tm/tm.c
+++ b/src/modules/tm/tm.c
@@ -402,7 +402,9 @@ static cmd_export_t cmds[]={
 	{"t_get_status_code", w_t_get_status_code,      0, 0, 0,
 		REQUEST_ROUTE|ONREPLY_ROUTE|FAILURE_ROUTE|BRANCH_ROUTE },
 
-	{"t_load_contacts", t_load_contacts,            0, 0, 0,
+	{"t_load_contacts", t_load_contacts,          0, 0, 0,
+		REQUEST_ROUTE | FAILURE_ROUTE},
+	{"t_load_contacts", t_load_contacts,          1, fixup_var_int_1, 0,
 		REQUEST_ROUTE | FAILURE_ROUTE},
 	{"t_next_contacts", t_next_contacts,            0, 0, 0,
 		REQUEST_ROUTE | FAILURE_ROUTE},
@@ -3034,6 +3036,11 @@ static sr_kemi_t tm_kemi_exports[] = {
 		{ SR_KEMIP_NONE, SR_KEMIP_NONE, SR_KEMIP_NONE,
 			SR_KEMIP_NONE, SR_KEMIP_NONE, SR_KEMIP_NONE }
 	},
+	{ str_init("tm"), str_init("ki_t_load_contacts_mode"),
+		SR_KEMIP_INT, ki_t_load_contacts_mode,
+		{ SR_KEMIP_INT, SR_KEMIP_NONE, SR_KEMIP_NONE,
+			SR_KEMIP_NONE, SR_KEMIP_NONE, SR_KEMIP_NONE }
+	},
 	{ str_init("tm"), str_init("t_next_contacts"),
 		SR_KEMIP_INT, ki_t_next_contacts,
 		{ SR_KEMIP_NONE, SR_KEMIP_NONE, SR_KEMIP_NONE,
