Browse code

dispatcher: algorithm 13

latency optimized round-robin with failover

Julien Chavanton authored on 29/09/2020 21:50:03
Showing 3 changed files
... ...
@@ -78,6 +78,7 @@
78 78
 #define DS_ALG_CALLLOAD 10
79 79
 #define DS_ALG_RELWEIGHT 11
80 80
 #define DS_ALG_PARALLEL 12
81
+#define DS_ALG_LATENCY 13
81 82
 
82 83
 /* increment call load */
83 84
 #define DS_LOAD_INC(dgrp, didx) do { \
... ...
@@ -2066,6 +2067,32 @@ int ds_select_dst_limit(sip_msg_t *msg, int set, int alg, uint32_t limit,
2066 2067
 	return ret;
2067 2068
 }
2068 2069
 
2070
+typedef struct sorted_ds {
2071
+	int idx;
2072
+	int priority;
2073
+} sorted_ds_t;
2074
+
2075
+int ds_manage_routes_fill_reodered_xavp(sorted_ds_t *ds_sorted, ds_set_t *idx, ds_select_state_t *rstate)
2076
+{
2077
+	int i;
2078
+	for(i=0; i < idx->nr && rstate->cnt < rstate->limit; i++) {
2079
+		if(ds_sorted[i].idx < 0 || ds_skip_dst(idx->dlist[i].flags)
2080
+				|| (ds_use_default != 0 && ds_sorted[i].idx == (idx->nr - 1))) {
2081
+			continue;
2082
+		}
2083
+		if(ds_add_xavp_record(idx, ds_sorted[i].idx, rstate->setid, rstate->alg,
2084
+					&rstate->lxavp)<0) {
2085
+			LM_ERR("failed to add destination in the xavp (%d/%d)\n",
2086
+					ds_sorted[i].idx, rstate->setid);
2087
+			return -1;
2088
+		}
2089
+		LM_ERR("destination added in the xavp (%d/%d)\n",
2090
+					ds_sorted[i].idx, rstate->setid);
2091
+		rstate->cnt++;
2092
+	}
2093
+	return 0;
2094
+}
2095
+
2069 2096
 int ds_manage_routes_fill_xavp(unsigned int hash, ds_set_t *idx, ds_select_state_t *rstate)
2070 2097
 {
2071 2098
 	int i;
... ...
@@ -2125,6 +2152,76 @@ int ds_manage_routes_fill_xavp(unsigned int hash, ds_set_t *idx, ds_select_state
2125 2152
 	return 0;
2126 2153
 }
2127 2154
 
2155
+
2156
+void ds_sorted_by_priority(sorted_ds_t * sorted_ds, int size) {
2157
+	int i,ii;
2158
+	for(i=0;i<size;++i) {
2159
+		for(ii=1;ii<size;++ii) {
2160
+			sorted_ds_t temp;
2161
+			if(sorted_ds[ii-1].priority < sorted_ds[ii].priority) {
2162
+				temp.idx = sorted_ds[ii].idx;
2163
+				temp.priority = sorted_ds[ii].priority;
2164
+				sorted_ds[ii].idx = sorted_ds[ii-1].idx;
2165
+				sorted_ds[ii].priority = sorted_ds[ii-1].priority;
2166
+				sorted_ds[ii-1].idx = temp.idx;
2167
+				sorted_ds[ii-1].priority = temp.priority;
2168
+			}
2169
+		}
2170
+	}
2171
+}
2172
+
2173
+int ds_manage_route_algo13(ds_set_t *idx, ds_select_state_t *rstate) {
2174
+	int hash = idx->last;
2175
+	int y = 0;
2176
+	int z = hash;
2177
+	int active_priority = 0;
2178
+	sorted_ds_t *ds_sorted = pkg_malloc(sizeof(sorted_ds_t) * idx->nr);
2179
+	if(ds_sorted == NULL) {
2180
+		LM_ERR("no more pkg\n");
2181
+		return -1;
2182
+	}
2183
+
2184
+	for(y=0; y<idx->nr ;y++) {
2185
+		int latency_proirity_handicap = 0;
2186
+		ds_dest_t * ds_dest = &idx->dlist[z];
2187
+		int gw_priority = ds_dest->priority;
2188
+		int gw_latency = ds_dest->latency_stats.estimate;
2189
+		int gw_inactive = ds_skip_dst(ds_dest->flags);
2190
+		if(!gw_inactive) {
2191
+			if(gw_latency > gw_priority && gw_priority > 0)
2192
+				latency_proirity_handicap = gw_latency / gw_priority;
2193
+			ds_dest->attrs.rpriority = gw_priority - latency_proirity_handicap;
2194
+			if(ds_dest->attrs.rpriority < 1 && gw_priority > 0)
2195
+				ds_dest->attrs.rpriority = 1;
2196
+			if(ds_dest->attrs.rpriority > active_priority) {
2197
+				hash = z;
2198
+				active_priority = ds_dest->attrs.rpriority;
2199
+			}
2200
+			ds_sorted[y].idx = z;
2201
+			ds_sorted[y].priority = ds_dest->attrs.rpriority;
2202
+			LM_DBG("[active]idx[%d]uri[%.*s]priority[%d-%d=%d]latency[%dms]flag[%d]\n",
2203
+				z, ds_dest->uri.len, ds_dest->uri.s,
2204
+				gw_priority, latency_proirity_handicap,
2205
+				ds_dest->attrs.rpriority, gw_latency, ds_dest->flags);
2206
+		} else {
2207
+			ds_sorted[y].idx = -1;
2208
+			ds_sorted[y].priority = -1;
2209
+			LM_DBG("[inactive]idx[%d]uri[%.*s]priority[%d]latency[%dms]flag[%d]",
2210
+				z, ds_dest->uri.len, ds_dest->uri.s,
2211
+				gw_priority, gw_latency, ds_dest->flags);
2212
+		}
2213
+		if(ds_use_default != 0 && idx->nr != 1)
2214
+			z = (z + 1) % (idx->nr - 1);
2215
+		else
2216
+			z = (z + 1) % idx->nr;
2217
+	}
2218
+	idx->last = hash % idx->nr;
2219
+	LM_DBG("priority[%d]gateway_selected[%d]next_index[%d]\n", active_priority, hash, idx->last);
2220
+	ds_sorted_by_priority(ds_sorted, idx->nr);
2221
+	ds_manage_routes_fill_reodered_xavp(ds_sorted, idx, rstate);
2222
+	return hash;
2223
+}
2224
+
2128 2225
 /**
2129 2226
  *
2130 2227
  */
... ...
@@ -2135,6 +2232,7 @@ int ds_manage_routes(sip_msg_t *msg, ds_select_state_t *rstate)
2135 2232
 	ds_set_t *idx = NULL;
2136 2233
 	int ulast = 0;
2137 2234
 	int vlast = 0;
2235
+	int xavp_filled = 0;
2138 2236
 
2139 2237
 	if(msg == NULL) {
2140 2238
 		LM_ERR("bad parameters\n");
... ...
@@ -2270,6 +2368,13 @@ int ds_manage_routes(sip_msg_t *msg, ds_select_state_t *rstate)
2270 2368
 		case DS_ALG_PARALLEL: /* 12 - parallel dispatching */
2271 2369
 			hash = 0;
2272 2370
 			break;
2371
+		case DS_ALG_LATENCY: /* 13 - latency optimized round-robin with failover */
2372
+			hash = ds_manage_route_algo13(idx, rstate);
2373
+			if (hash < 0)
2374
+				return -1;
2375
+			xavp_filled = 1;
2376
+			ulast = 1;
2377
+			break;
2273 2378
 		default:
2274 2379
 			LM_WARN("algo %d not implemented - using first entry...\n",
2275 2380
 					rstate->alg);
... ...
@@ -2285,7 +2390,7 @@ int ds_manage_routes(sip_msg_t *msg, ds_select_state_t *rstate)
2285 2390
 	i = hash;
2286 2391
 
2287 2392
 	/* if selected address is inactive, find next active */
2288
-	while(ds_skip_dst(idx->dlist[i].flags)) {
2393
+	while(!xavp_filled && ds_skip_dst(idx->dlist[i].flags)) {
2289 2394
 		if(ds_use_default != 0 && idx->nr != 1)
2290 2395
 			i = (i + 1) % (idx->nr - 1);
2291 2396
 		else
... ...
@@ -2344,8 +2449,11 @@ int ds_manage_routes(sip_msg_t *msg, ds_select_state_t *rstate)
2344 2449
 		return 1;
2345 2450
 	}
2346 2451
 
2347
-	if (ds_manage_routes_fill_xavp(hash, idx, rstate) == -1)
2348
-		return -1;
2452
+	if(!xavp_filled) {
2453
+		if(ds_manage_routes_fill_xavp(hash, idx, rstate) == -1){
2454
+			return -1;
2455
+		}
2456
+	}
2349 2457
 
2350 2458
 	/* add default dst to last position in XAVP list */
2351 2459
 	if(ds_use_default != 0 && hash != idx->nr - 1
... ...
@@ -2715,12 +2823,22 @@ int ds_update_latency(int group, str *address, int code)
2715 2823
 			int latency_ms;
2716 2824
 			/* Destination address found, this is the gateway that was pinged. */
2717 2825
 			state = ds_dest->flags;
2826
+			if (!(state & DS_PROBING_DST)) {
2827
+				i++;
2828
+				continue;
2829
+			}
2718 2830
 			if (code == 408 && latency_stats->timeout < UINT32_MAX)
2719 2831
 				latency_stats->timeout++;
2720 2832
 			gettimeofday(&now, NULL);
2721 2833
 			latency_ms = (now.tv_sec - latency_stats->start.tv_sec)*1000
2722 2834
 		            + (now.tv_usec - latency_stats->start.tv_usec)/1000;
2723
-			latency_stats_update(latency_stats, latency_ms);
2835
+			if (code != 408)
2836
+				latency_stats_update(latency_stats, latency_ms);
2837
+
2838
+			LM_DBG("[%d]latency[%d]avg[%.2f][%.*s]code[%d]rweight[%d]\n",
2839
+					latency_stats->count, latency_ms,
2840
+					latency_stats->average, address->len, address->s,
2841
+					code, ds_dest->attrs.rweight);
2724 2842
 
2725 2843
 			/* Adjusting weight using congestion detection based on latency estimator. */
2726 2844
 			if (ds_dest->attrs.congestion_control && ds_dest->attrs.weight > 0) {
... ...
@@ -190,6 +190,7 @@ typedef struct _ds_attrs {
190 190
 	int congestion_control;
191 191
 	str ping_from;
192 192
 	str obproxy;
193
+	int rpriority;
193 194
 } ds_attrs_t;
194 195
 
195 196
 typedef struct _ds_latency_stats {
... ...
@@ -1261,6 +1261,34 @@ modparam("dispatcher", "reload_delta", 1)
1261 1261
 				making sense in this case.
1262 1262
 				</para>
1263 1263
 			</listitem>
1264
+			<listitem>
1265
+				<para>
1266
+				<quote>13</quote> - latency optimized dispatching
1267
+				</para>
1268
+				<para>
1269
+				- The algorithm will load balance using round-robin prioritizing the gateways with the highest priority.
1270
+				</para>
1271
+				<para>
1272
+				- If ds_ping_latency_stats is active the algorithm be able to adjust the priority of the gateway automaticaly,
1273
+				the priority will be lowered by 1 point every time the latency ms is as high as the priority.
1274
+				</para>
1275
+				<example>
1276
+				<title><function>latency_optimized_dispatching</function> usage</title>
1277
+				<programlisting format="linespecific">
1278
+Using this simple formula : ADJUSTED_PRIORITY = PRIORITY - (ESTIMATED_LATENCY/PRIORITY)
1279
+
1280
+GATEWAY | PRIORITY | ESTIMATED | ADJUSTED | LOAD
1281
+   #    |          |  LATENCY  | PRIORITY | DISTRIBUTION
1282
+   1    |    30    |    21     |    30    | 33%
1283
+   2    |    30    |    91     |    27    | 0%
1284
+   3    |    30    |    61     |    28    | 0%
1285
+   4    |    30    |    19     |    30    | 33%
1286
+   5    |    30    |    32     |    29    | 0%
1287
+   6    |    30    |    0      |    30    | 33%
1288
+   7    |    30    |    201    |    24    | 0%
1289
+				</programlisting>
1290
+				</example>
1291
+			</listitem>
1264 1292
 			<listitem>
1265 1293
 				<para>
1266 1294
 				<quote>X</quote> - if the algorithm is not implemented, the