modules/cdp: added loadbalancing across diameter peers within the same realm and with the same metric

modules/cdp/cdp_rpc.c

@@ -66,7 +66,10 @@ static void cdp_rpc_list_peers(rpc_t* rpc, void* ctx)
     void *peers_header;
     void *peers_container;
     void *peerdetail_container;
-    peer *i;
+    void *peerapplication_container;
+    peer *i, *j;
+    int c;
+    char buf[100];
 
     if (rpc->add(ctx, "{", &peers_header) < 0) {
             rpc->fault(ctx, 500, "Internal error creating top rpc");
@@ -91,19 +94,40 @@ static void cdp_rpc_list_peers(rpc_t* rpc, void* ctx)
     lock_get(peer_list_lock);
     i = peer_list->head;
     while (i) {
-            if (rpc->struct_add(peers_container, "S{",
+    		lock_get(i->lock);
+    		if (rpc->struct_add(peers_container, "S{",
                             "FQDN", &i->fqdn,
                             "Details", &peerdetail_container) < 0) {
                     rpc->fault(ctx, 500, "Internal error creating peers container struct");
+                    lock_release(i->lock);
                     return;
             }
-            if (rpc->struct_add(peerdetail_container, "ss",
+            if (rpc->struct_add(peerdetail_container, "ssd",
                     "State", dp_states[(int)i->state],
-                    "Disabled", i->disabled?"True":"False") < 0) {
+                    "Disabled", i->disabled?"True":"False",
+            		"Last used", i->last_selected) < 0) {
                     rpc->fault(ctx, 500, "Internal error creating peer detail container struct");
+                    lock_release(i->lock);
                     return;
             }
+            if (rpc->struct_add(peerdetail_container, "{", "Applications", &peerapplication_container) < 0) {
+            	rpc->fault(ctx, 500, "Internal error creating peer application container struct");
+            	lock_release(i->lock);
+            	return;
+            }
+
+            for (c = 0; c < i->applications_cnt; c++) {
+            	snprintf(buf, 100, "%d:%d", i->applications[c].id, i->applications[c].vendor);
+            	if (rpc->struct_add(peerapplication_container, "s",
+						"appid:vendorid", buf) < 0) {
+					rpc->fault(ctx, 500, "Internal error creating appid/vendorid information");
+					lock_release(i->lock);
+					return;
+				}
+            }
+            j=i;
             i = i->next;
+            lock_release(j->lock);
     }
     lock_release(peer_list_lock);
 }

modules/cdp/diameter_comm.c

@@ -195,7 +195,7 @@ AAAReturnCode AAASendMessageToPeer(
 			LM_ERR("AAASendMessageToPeer(): can't add transaction callback for answer.\n");
 	}
 
-//	if (!peer_send_msg(p,message))
+	p->last_selected = time(NULL);
 	if (!sm_process(p,Send_Message,message,0,0))
 		goto error;
 

modules/cdp/peer.h

@@ -111,6 +111,7 @@ typedef struct _peer_t{
 	int R_sock;				/**< socket used as receiver */
 	
 	time_t activity;		/**< timestamp of last activity */
+	time_t last_selected;	/**< timestamp this peer was last selected for routing - used in least recently used load balancing across metric */
 	int is_dynamic;			/**< whether this peer was accepted although it was not initially configured */
 	int disabled;			/**< administratively enable/disable a peer - ie remove/re-add from service dynamically */
 	int waitingDWA;			/**< if a Diameter Watch-dog Request was sent out and waiting for an answer */

modules/cdp/routing.c

@@ -48,7 +48,10 @@
 #include "peermanager.h"
 #include "diameter_api.h"
 
+#define LB_MAX_PEERS 20			/**< maximum peers that can be loadbalanced accross i.e. same metric */
+
 extern dp_config *config;		/**< Configuration for this diameter peer 	*/
+int gcount = 0;
 
 /**
  * Returns if the peer advertised support for an Application ID
@@ -74,11 +77,19 @@ int peer_handles_application(peer *p,int app_id,int vendor_id)
  */
 peer* get_first_connected_route(routing_entry *r,int app_id,int vendor_id)
 {
+	peer *peers[LB_MAX_PEERS];
+	int peer_count=0;
+	int prev_metric=0;
 	routing_entry *i;
 	peer *p;
+	int j;
+	time_t least_recent_time;
+
 	LM_DBG("get_first_connected_route in list %p for app_id %d and vendor_id %d\n",
 		r,app_id,vendor_id);
 	for(i=r;i;i=i->next){
+		if (peer_count >= LB_MAX_PEERS)
+			break;
 		p = get_peer_by_fqdn(&(i->fqdn));
 		if (!p)
 			LM_DBG("The peer %.*s does not seem to be connected or configured\n",
@@ -88,10 +99,31 @@ peer* get_first_connected_route(routing_entry *r,int app_id,int vendor_id)
 				(p->state==I_Open||p->state==R_Open)?"opened":"closed");
 		if (p && !p->disabled && (p->state==I_Open || p->state==R_Open) && peer_handles_application(p,app_id,vendor_id)) {
 			LM_DBG("The peer %.*s matches - will forward there\n",i->fqdn.len,i->fqdn.s);
-			return p;
+			if (peer_count!=0) {//check the metric
+				if (i->metric != prev_metric)
+					break;
+				//metric must be the same
+				peers[peer_count++] = p;
+			} else {//we're first
+				prev_metric = i->metric;
+				peers[peer_count++] = p;
+			}
 		}
 	}
-	return 0;
+
+	if (peer_count==0)
+		return 0;
+
+	least_recent_time = peers[0]->last_selected;
+	p = peers[0];
+	for (j=1; j<peer_count; j++) {
+		if (peers[j]->last_selected < least_recent_time) {
+			least_recent_time = peers[j]->last_selected;
+			p = peers[j];
+		}
+	}
+
+	return p;
 }
 
 /**
@@ -153,7 +185,10 @@ peer* get_routing_peer(AAAMessage *m)
 	if (destination_host.len){
 		/* There is a destination host present in the message try and route directly there */
 		p = get_peer_by_fqdn(&destination_host);
-		if (p && (p->state==I_Open || p->state==R_Open) && peer_handles_application(p,app_id,vendor_id)) return p;
+		if (p && (p->state==I_Open || p->state==R_Open) && peer_handles_application(p,app_id,vendor_id)) {
+			p->last_selected = time(NULL);
+			return p;
+		}
 		/* the destination host peer is not connected at the moment, try a normal route then */
 	}