added support for retry in http router

core/cache.c

@@ -498,6 +498,27 @@ char *uwsgi_cache_get3(struct uwsgi_cache *uc, char *key, uint16_t keylen, uint6
         return NULL;
 }
 
+char *uwsgi_cache_get4(struct uwsgi_cache *uc, char *key, uint16_t keylen, uint64_t * valsize, uint64_t *hits) {
+
+        uint64_t index = uwsgi_cache_get_index(uc, key, keylen);
+
+        if (index) {
+                struct uwsgi_cache_item *uci = cache_item(index);
+                if (uci->flags & UWSGI_CACHE_FLAG_UNGETTABLE)
+                        return NULL;
+                *valsize = uci->valsize;
+                if (hits)
+                        *hits = uci->hits;
+                uci->hits++;
+                uc->hits++;
+                return uc->data + (uci->first_block * uc->blocksize);
+        }
+
+        uc->miss++;
+
+        return NULL;
+}
+
 
 int uwsgi_cache_del2(struct uwsgi_cache *uc, char *key, uint16_t keylen, uint64_t index, uint16_t flags) {
 

core/strings.c

@@ -1,5 +1,46 @@
 #include <uwsgi.h>
 
+char *uwsgi_str_split_nget(char *str, size_t len, char what, size_t pos, size_t *rlen) {
+	size_t i;
+	size_t current = 0;
+	char *choosen = str;
+	size_t choosen_len = 0;
+	*rlen = 0;
+	for(i=0;i<len;i++) {
+		if (!choosen) choosen = str + i;
+		if (str[i] == what) {
+			if (current == pos) {
+				if (choosen_len == 0) return NULL;
+				*rlen = choosen_len;
+				return choosen;
+			}
+			current++;
+			choosen = NULL;
+			choosen_len = 0;
+		}
+		else {
+			choosen_len ++;
+		}
+	}
+
+	if (current == pos) {
+		if (choosen_len == 0) return NULL;
+		*rlen = choosen_len;
+		return choosen;
+	}
+
+	return NULL;
+}
+
+size_t uwsgi_str_occurence(char *str, size_t len, char what) {
+	size_t count = 0;
+	size_t i;
+	for(i=0;i<len;i++) {
+		if (str[i] == what) count++;
+	}
+	return count;
+}
+
 // check if a string_list contains an item
 struct uwsgi_string_list *uwsgi_string_list_has_item(struct uwsgi_string_list *list, char *key, size_t keylen) {
         struct uwsgi_string_list *usl = list;

plugins/corerouter/cr_map.c

@@ -1,4 +1,4 @@
-#include "../../uwsgi.h"
+#include <uwsgi.h>
 
 #include "cr.h"
 
@@ -9,13 +9,26 @@ int uwsgi_cr_map_use_void(struct uwsgi_corerouter *ucr, struct corerouter_peer *
 }
 
 int uwsgi_cr_map_use_cache(struct uwsgi_corerouter *ucr, struct corerouter_peer *peer) {
+	uint64_t hits = 0;
 	uwsgi_rlock(ucr->cache->lock);
-	peer->instance_address = uwsgi_cache_get2(ucr->cache, peer->key, peer->key_len, &peer->instance_address_len);
+	char *value = uwsgi_cache_get4(ucr->cache, peer->key, peer->key_len, &peer->instance_address_len, &hits);
+	if (!value) goto end;
+	peer->tmp_socket_name = uwsgi_concat2n(value, peer->instance_address_len, "", 0);
+	size_t nodes = uwsgi_str_occurence(peer->tmp_socket_name, peer->instance_address_len, '|');
+	if (nodes > 0) {
+		size_t choosen_node = hits % (nodes+1);
+		peer->instance_address = uwsgi_str_split_nget(peer->tmp_socket_name, peer->instance_address_len, '|', choosen_node, &peer->instance_address_len);
+		if (!peer->instance_address) goto end;
+	}
+	else {
+		peer->instance_address = peer->tmp_socket_name;
+	}
 	char *cs_mod = uwsgi_str_contains(peer->instance_address, peer->instance_address_len, ',');
 	if (cs_mod) {
 		peer->modifier1 = uwsgi_str_num(cs_mod + 1, (peer->instance_address_len - (cs_mod - peer->instance_address)) - 1);
 		peer->instance_address_len = (cs_mod - peer->instance_address);
 	}
+end:
 	uwsgi_rwunlock(ucr->cache->lock);
 	return 0;
 }

plugins/http/http.c

@@ -430,6 +430,9 @@ ssize_t hr_instance_connected(struct corerouter_peer* peer) {
 
 	cr_peer_connected(peer, "hr_instance_connected()");
 
+	// we are connected, we cannot retry anymore
+        peer->can_retry = 0;
+
 	// prepare for write
 	peer->out_pos = 0;
 
@@ -705,7 +708,7 @@ ssize_t http_parse(struct corerouter_peer *main_peer) {
 			if (hr->websockets > 2 && hr->websocket_key_len > 0) {
 				hr->raw_body = 1;
 			}
-
+			new_peer->can_retry = 1;
                 	cr_connect(new_peer, hr_instance_connected);
 			break;
 		}
@@ -775,7 +778,31 @@ ssize_t hr_recv_stud4(struct corerouter_peer * main_peer) {
 
 }
 
+// retry connection to the backend
+static int hr_retry(struct corerouter_peer *peer) {
+
+        struct uwsgi_corerouter *ucr = peer->session->corerouter;
+
+        if (peer->instance_address_len > 0) goto retry;
+
+        if (ucr->mapper(ucr, peer)) {
+                return -1;
+        }
+
+        if (peer->instance_address_len == 0) {
+                return -1;
+        }
+
+retry:
+        // start async connect (again)
+        cr_connect(peer, hr_instance_connected);
+        return 0;
+}
+
+
 int http_alloc_session(struct uwsgi_corerouter *ucr, struct uwsgi_gateway_socket *ugs, struct corerouter_session *cs, struct sockaddr *sa, socklen_t s_len) {
+	// set the retry hook
+        cs->retry = hr_retry;
 	struct http_session *hr = (struct http_session *) cs;
 	// set the modifier1
 	cs->main_peer->modifier1 = uhttp.modifier1;

plugins/http/spdy3.c

@@ -568,6 +568,8 @@ static ssize_t spdy_inflate_http_headers(struct http_session *hr) {
         new_peer->out->buf[1] = (uint8_t) (pktsize & 0xff);
         new_peer->out->buf[2] = (uint8_t) ((pktsize >> 8) & 0xff);
 
+	new_peer->can_retry = 1;
+
 	cr_connect(new_peer, hr_instance_connected);
 
 	return 1;

uwsgi.h

@@ -2807,6 +2807,7 @@ int uwsgi_cache_set2(struct uwsgi_cache *, char *, uint16_t, char *, uint64_t, u
 int uwsgi_cache_del2(struct uwsgi_cache *, char *, uint16_t, uint64_t, uint16_t);
 char *uwsgi_cache_get2(struct uwsgi_cache *, char *, uint16_t, uint64_t *);
 char *uwsgi_cache_get3(struct uwsgi_cache *, char *, uint16_t, uint64_t *, uint64_t *);
+char *uwsgi_cache_get4(struct uwsgi_cache *, char *, uint16_t, uint64_t *, uint64_t *);
 uint32_t uwsgi_cache_exists2(struct uwsgi_cache *, char *, uint16_t);
 struct uwsgi_cache *uwsgi_cache_create(char *);
 struct uwsgi_cache *uwsgi_cache_by_name(char *);
@@ -2988,7 +2989,7 @@ time_t timegm(struct tm *);
 #endif
 
 size_t uwsgi_str_num(char *, int);
-
+size_t uwsgi_str_occurence(char *, size_t, char);
 
 int uwsgi_proto_uwsgi_parser(struct wsgi_request *);
 int uwsgi_proto_base_write(struct wsgi_request *, char *, size_t);
@@ -3029,6 +3030,8 @@ char *uwsgi_split3(char *, size_t, char, char **, size_t *, char **, size_t *, c
 char *uwsgi_split4(char *, size_t, char, char **, size_t *, char **, size_t *, char **, size_t *, char **, size_t *);
 char *uwsgi_netstring(char *, size_t, char **, size_t *);
 
+char *uwsgi_str_split_nget(char *, size_t, char, size_t, size_t *);
+
 int uwsgi_get_socket_num(struct uwsgi_socket *);
 struct uwsgi_socket *uwsgi_new_socket(char *);
 struct uwsgi_socket *uwsgi_new_shared_socket(char *);