Browse code

dispatcher: algorithm 13

- latency optimized round-robin with failover
- optionally congestion can be use instead of latency

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,79 @@ 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 cc is enabled, the latency is the congestion ms instead of the estimated latency.
2191
+		if (ds_dest->attrs.congestion_control)
2192
+			gw_latency = ds_dest->latency_stats.estimate - ds_dest->latency_stats.average;
2193
+		if(!gw_inactive) {
2194
+			if(gw_latency > gw_priority && gw_priority > 0)
2195
+				latency_proirity_handicap = gw_latency / gw_priority;
2196
+			ds_dest->attrs.rpriority = gw_priority - latency_proirity_handicap;
2197
+			if(ds_dest->attrs.rpriority < 1 && gw_priority > 0)
2198
+				ds_dest->attrs.rpriority = 1;
2199
+			if(ds_dest->attrs.rpriority > active_priority) {
2200
+				hash = z;
2201
+				active_priority = ds_dest->attrs.rpriority;
2202
+			}
2203
+			ds_sorted[y].idx = z;
2204
+			ds_sorted[y].priority = ds_dest->attrs.rpriority;
2205
+			LM_DBG("[active]idx[%d]uri[%.*s]priority[%d-%d=%d]latency[%dms]flag[%d]\n",
2206
+				z, ds_dest->uri.len, ds_dest->uri.s,
2207
+				gw_priority, latency_proirity_handicap,
2208
+				ds_dest->attrs.rpriority, gw_latency, ds_dest->flags);
2209
+		} else {
2210
+			ds_sorted[y].idx = -1;
2211
+			ds_sorted[y].priority = -1;
2212
+			LM_DBG("[inactive]idx[%d]uri[%.*s]priority[%d]latency[%dms]flag[%d]",
2213
+				z, ds_dest->uri.len, ds_dest->uri.s,
2214
+				gw_priority, gw_latency, ds_dest->flags);
2215
+		}
2216
+		if(ds_use_default != 0 && idx->nr != 1)
2217
+			z = (z + 1) % (idx->nr - 1);
2218
+		else
2219
+			z = (z + 1) % idx->nr;
2220
+	}
2221
+	idx->last = hash % idx->nr;
2222
+	LM_DBG("priority[%d]gateway_selected[%d]next_index[%d]\n", active_priority, hash, idx->last);
2223
+	ds_sorted_by_priority(ds_sorted, idx->nr);
2224
+	ds_manage_routes_fill_reodered_xavp(ds_sorted, idx, rstate);
2225
+	return hash;
2226
+}
2227
+
2128 2228
 /**
2129 2229
  *
2130 2230
  */
... ...
@@ -2135,6 +2235,7 @@ int ds_manage_routes(sip_msg_t *msg, ds_select_state_t *rstate)
2135 2235
 	ds_set_t *idx = NULL;
2136 2236
 	int ulast = 0;
2137 2237
 	int vlast = 0;
2238
+	int xavp_filled = 0;
2138 2239
 
2139 2240
 	if(msg == NULL) {
2140 2241
 		LM_ERR("bad parameters\n");
... ...
@@ -2270,6 +2371,15 @@ int ds_manage_routes(sip_msg_t *msg, ds_select_state_t *rstate)
2270 2371
 		case DS_ALG_PARALLEL: /* 12 - parallel dispatching */
2271 2372
 			hash = 0;
2272 2373
 			break;
2374
+		case DS_ALG_LATENCY: /* 13 - latency optimized round-robin with failover */
2375
+			lock_get(&idx->lock);
2376
+			hash = ds_manage_route_algo13(idx, rstate);
2377
+			lock_release(&idx->lock);
2378
+			if (hash < 0)
2379
+				return -1;
2380
+			xavp_filled = 1;
2381
+			ulast = 1;
2382
+			break;
2273 2383
 		default:
2274 2384
 			LM_WARN("algo %d not implemented - using first entry...\n",
2275 2385
 					rstate->alg);
... ...
@@ -2285,7 +2395,7 @@ int ds_manage_routes(sip_msg_t *msg, ds_select_state_t *rstate)
2285 2395
 	i = hash;
2286 2396
 
2287 2397
 	/* if selected address is inactive, find next active */
2288
-	while(ds_skip_dst(idx->dlist[i].flags)) {
2398
+	while(!xavp_filled && ds_skip_dst(idx->dlist[i].flags)) {
2289 2399
 		if(ds_use_default != 0 && idx->nr != 1)
2290 2400
 			i = (i + 1) % (idx->nr - 1);
2291 2401
 		else
... ...
@@ -2344,8 +2454,11 @@ int ds_manage_routes(sip_msg_t *msg, ds_select_state_t *rstate)
2344 2454
 		return 1;
2345 2455
 	}
2346 2456
 
2347
-	if (ds_manage_routes_fill_xavp(hash, idx, rstate) == -1)
2348
-		return -1;
2457
+	if(!xavp_filled) {
2458
+		if(ds_manage_routes_fill_xavp(hash, idx, rstate) == -1){
2459
+			return -1;
2460
+		}
2461
+	}
2349 2462
 
2350 2463
 	/* add default dst to last position in XAVP list */
2351 2464
 	if(ds_use_default != 0 && hash != idx->nr - 1
... ...
@@ -2715,12 +2828,22 @@ int ds_update_latency(int group, str *address, int code)
2715 2828
 			int latency_ms;
2716 2829
 			/* Destination address found, this is the gateway that was pinged. */
2717 2830
 			state = ds_dest->flags;
2831
+			if (!(state & DS_PROBING_DST)) {
2832
+				i++;
2833
+				continue;
2834
+			}
2718 2835
 			if (code == 408 && latency_stats->timeout < UINT32_MAX)
2719 2836
 				latency_stats->timeout++;
2720 2837
 			gettimeofday(&now, NULL);
2721 2838
 			latency_ms = (now.tv_sec - latency_stats->start.tv_sec)*1000
2722 2839
 		            + (now.tv_usec - latency_stats->start.tv_usec)/1000;
2723
-			latency_stats_update(latency_stats, latency_ms);
2840
+			if (code != 408)
2841
+				latency_stats_update(latency_stats, latency_ms);
2842
+
2843
+			LM_DBG("[%d]latency[%d]avg[%.2f][%.*s]code[%d]rweight[%d]\n",
2844
+					latency_stats->count, latency_ms,
2845
+					latency_stats->average, address->len, address->s,
2846
+					code, ds_dest->attrs.rweight);
2724 2847
 
2725 2848
 			/* Adjusting weight using congestion detection based on latency estimator. */
2726 2849
 			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 {
... ...
@@ -1241,7 +1241,7 @@ modparam("dispatcher", "reload_delta", 1)
1241 1241
 				</para>
1242 1242
 				<para>
1243 1243
 				Using this algorithm, you can also enable congestion control by setting the
1244
-				attibute 'cc=1', when 'cc' is enabled the 'weight' attribute will also be
1244
+				attribute 'cc=1', when 'cc' is enabled the 'weight' attribute will also be
1245 1245
 				used to control congestion tolerance. When facing congestion the weight of
1246 1246
 				a gateway is lowered by 1 for every ms of estimated congestion, a 'rweight'
1247 1247
 				value of 50 is recommended. See the example "configuring load balancing with
... ...
@@ -1261,6 +1261,43 @@ 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 will adjust the priority of the gateway automatically,
1273
+				the priority will be lowered by 1 point every time the latency ms is as high as the priority.
1274
+				</para>
1275
+				<para>
1276
+				- If the attribute 'cc=1' is set, the latency used is congestion ms : estimate (current latency ms) - average (normal condition latency ms).
1277
+				</para>
1278
+				<example>
1279
+				<title><function>latency_optimized_dispatching</function> usage</title>
1280
+				<programlisting format="linespecific">
1281
+Using this simple formula :
1282
+    ADJUSTED_PRIORITY = PRIORITY - (ESTIMATED_LATENCY_MS/PRIORITY)
1283
+
1284
+GATEWAY | PRIORITY | ESTIMATED | ADJUSTED | LOAD
1285
+   #    |          |  LATENCY  | PRIORITY | DISTRIBUTION
1286
+   1    |    30    |    21     |    30    | 33%
1287
+   2    |    30    |    91     |    27    | 0%
1288
+   3    |    30    |    61     |    28    | 0%
1289
+   4    |    30    |    19     |    30    | 33%
1290
+   5    |    30    |    32     |    29    | 0%
1291
+   6    |    30    |    0      |    30    | 33%
1292
+   7    |    30    |    201    |    24    | 0%
1293
+
1294
+
1295
+With congestion control the formula becomes :
1296
+    CONGESTION_MS = CURRENT_LATENCY_MS - NORMAL_CONDITION_LATENCY_MS
1297
+    ADJUSTED_PRIORITY = PRIORITY - (CONGESTION_MS/PRIORITY)
1298
+				</programlisting>
1299
+				</example>
1300
+			</listitem>
1264 1301
 			<listitem>
1265 1302
 				<para>
1266 1303
 				<quote>X</quote> - if the algorithm is not implemented, the