From 294452eef7e75df33d658e019a9542578b8b6af2 Mon Sep 17 00:00:00 2001 From: Unbit Date: Tue, 25 Jun 2013 07:56:01 +0200 Subject: [PATCH] added support for retry in http router --- core/cache.c | 21 +++++++++++++++++++ core/strings.c | 41 +++++++++++++++++++++++++++++++++++++ plugins/corerouter/cr_map.c | 17 +++++++++++++-- plugins/http/http.c | 29 +++++++++++++++++++++++++- plugins/http/spdy3.c | 2 ++ uwsgi.h | 5 ++++- 6 files changed, 111 insertions(+), 4 deletions(-) diff --git a/core/cache.c b/core/cache.c index db3b9d0d..d4e89214 100644 --- a/core/cache.c +++ b/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) { diff --git a/core/strings.c b/core/strings.c index 4a8932e8..300af344 100644 --- a/core/strings.c +++ b/core/strings.c @@ -1,5 +1,46 @@ #include +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 #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; } diff --git a/plugins/http/http.c b/plugins/http/http.c index 0a9dc1cd..ae7574ed 100644 --- a/plugins/http/http.c +++ b/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; diff --git a/plugins/http/spdy3.c b/plugins/http/spdy3.c index 516b2685..4edb7aa2 100644 --- a/plugins/http/spdy3.c +++ b/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; diff --git a/uwsgi.h b/uwsgi.h index 99d49af2..6299d1ac 100644 --- a/uwsgi.h +++ b/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 *);