From fa325465bfb3ecb2bf0f67ee6dc4ac94bafc605c Mon Sep 17 00:00:00 2001 From: XieHan Date: Sat, 30 Oct 2021 16:31:13 +0800 Subject: [PATCH 1/3] fix WFServiceGovernance's server ref bug --- src/nameservice/UpstreamPolicies.cc | 21 ++---- src/nameservice/UpstreamPolicies.h | 4 +- src/nameservice/WFServiceGovernance.cc | 91 ++++++++++---------------- src/nameservice/WFServiceGovernance.h | 12 +++- 4 files changed, 53 insertions(+), 75 deletions(-) diff --git a/src/nameservice/UpstreamPolicies.cc b/src/nameservice/UpstreamPolicies.cc index 6443f748..66b8b16b 100644 --- a/src/nameservice/UpstreamPolicies.cc +++ b/src/nameservice/UpstreamPolicies.cc @@ -465,26 +465,15 @@ int UPSWeightedRandomPolicy::remove_server_locked(const std::string& address) int UPSWeightedRandomPolicy::select_history_weight(WFNSTracing *tracing) { - if (!tracing || !tracing->data) + struct TracingData *tracing_data = (struct TracingData *)tracing->data; + + if (!tracing_data) return 0; - UPSAddrParams *params; - - if (!tracing->deleter) - { - auto *server = (EndpointAddress *)tracing->data; - params = (UPSAddrParams *)server->params; - return params->weight; - } - int ret = 0; - auto *v = (std::vector *)(tracing->data); - for (auto *server : (*v)) - { - params = (UPSAddrParams *)server->params; - ret += params->weight; - } + for (EndpointAddress *server : tracing_data->history) + ret += ((UPSAddrParams *)server->params)->weight; return ret; } diff --git a/src/nameservice/UpstreamPolicies.h b/src/nameservice/UpstreamPolicies.h index 2da0bca6..fc34a28e 100644 --- a/src/nameservice/UpstreamPolicies.h +++ b/src/nameservice/UpstreamPolicies.h @@ -16,8 +16,8 @@ Authors: Wu Jiaxu (wujiaxu@sogou-inc.com) */ -#ifndef _UPSTREAM_POLICIES_H_ -#define _UPSTREAM_POLICIES_H_ +#ifndef _UPSTREAMPOLICIES_H_ +#define _UPSTREAMPOLICIES_H_ #include #include diff --git a/src/nameservice/WFServiceGovernance.cc b/src/nameservice/WFServiceGovernance.cc index 6446fea9..d62f2f13 100644 --- a/src/nameservice/WFServiceGovernance.cc +++ b/src/nameservice/WFServiceGovernance.cc @@ -14,6 +14,7 @@ limitations under the License. Authors: Li Yingxin (liyingxin@sogou-inc.com) + Xie Han (xiehan@sogou-inc.com) */ #include @@ -123,9 +124,9 @@ EndpointAddress::EndpointAddress(const std::string& address, WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *params, router_callback_t callback) { + WFNSTracing *tracing = params->tracing; EndpointAddress *addr; WFRouterTask *task; - WFNSTracing *tracing = params->tracing; if (this->select(params->uri, tracing, &addr) && copy_host_port(params->uri, addr)) @@ -139,25 +140,16 @@ WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *p task = resolver->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min, endpoint_params, std::move(callback)); - if (!tracing->data) - tracing->data = addr; - else + struct TracingData *tracing_data = (struct TracingData *)tracing->data; + if (!tracing_data) { - std::vector *v; - - if (!tracing->deleter) - { - EndpointAddress *last_addr = (EndpointAddress *)tracing->data; - v = new std::vector; - v->push_back(last_addr); - tracing->deleter = WFServiceGovernance::tracing_deleter; - tracing->data = v; - } - else - v = (std::vector *)tracing->data; - - v->push_back(addr); + tracing_data = new TracingData; + tracing_data->sg = this; + tracing->data = tracing_data; + tracing->deleter = WFServiceGovernance::tracing_deleter; } + + tracing_data->history.push_back(addr); } else task = new WFSelectorFailTask(std::move(callback)); @@ -167,21 +159,31 @@ WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *p void WFServiceGovernance::tracing_deleter(void *data) { - delete (std::vector *)data; + struct TracingData *tracing_data = (struct TracingData *)data; + + for (EndpointAddress *addr : tracing_data->history) + { + if (--addr->ref == 0) + { + pthread_rwlock_wrlock(&tracing_data->sg->rwlock); + tracing_data->sg->pre_delete_server(addr); + pthread_rwlock_unlock(&tracing_data->sg->rwlock); + delete addr; + } + } + + delete tracing_data; } bool WFServiceGovernance::in_select_history(WFNSTracing *tracing, EndpointAddress *addr) { - if (!tracing || !tracing->data) + struct TracingData *tracing_data = (struct TracingData *)tracing->data; + + if (!tracing_data) return false; - if (!tracing->deleter) - return (EndpointAddress *)tracing->data == addr; - - auto *v = (std::vector *)(tracing->data); - - for (auto *server : (*v)) + for (EndpointAddress *server : tracing_data->history) { if (server == addr) return true; @@ -234,24 +236,14 @@ void WFServiceGovernance::success(RouteManager::RouteResult *result, WFNSTracing *tracing, CommTarget *target) { - EndpointAddress *server; - if (tracing->deleter) - { - auto *v = (std::vector *)(tracing->data); - server = (*v)[v->size() - 1]; - } - else - server = (EndpointAddress *)tracing->data; + struct TracingData *tracing_data = (struct TracingData *)tracing->data; + auto *v = &tracing_data->history; + EndpointAddress *server = (*v)[v->size() - 1]; pthread_rwlock_wrlock(&this->rwlock); this->recover_server_from_breaker(server); - if (--server->ref == 0) - { - this->pre_delete_server(server); - delete server; - } - pthread_rwlock_unlock(&this->rwlock); + this->WFNSPolicy::success(result, tracing, target); } @@ -259,26 +251,15 @@ void WFServiceGovernance::failed(RouteManager::RouteResult *result, WFNSTracing *tracing, CommTarget *target) { - EndpointAddress *server; - if (tracing->deleter) - { - auto *v = (std::vector *)(tracing->data); - server = (*v)[v->size() - 1]; - } - else - server = (EndpointAddress *)tracing->data; + struct TracingData *tracing_data = (struct TracingData *)tracing->data; + auto *v = &tracing_data->history; + EndpointAddress *server = (*v)[v->size() - 1]; pthread_rwlock_wrlock(&this->rwlock); if (++server->fail_count == server->params->max_fails) this->fuse_server_to_breaker(server); - - if (--server->ref == 0) - { - this->pre_delete_server(server); - delete server; - } - pthread_rwlock_unlock(&this->rwlock); + this->WFNSPolicy::failed(result, tracing, target); } diff --git a/src/nameservice/WFServiceGovernance.h b/src/nameservice/WFServiceGovernance.h index 4282637b..67b87141 100644 --- a/src/nameservice/WFServiceGovernance.h +++ b/src/nameservice/WFServiceGovernance.h @@ -14,10 +14,11 @@ limitations under the License. Authors: Li Yingxin (liyingxin@sogou-inc.com) + Xie Han (xiehan@sogou-inc.com) */ -#ifndef _SERVICE_GOVERNANCE_H_ -#define _SERVICE_GOVERNANCE_H_ +#ifndef _WFSERVICEGOVERNANCE_H_ +#define _WFSERVICEGOVERNANCE_H_ #include #include @@ -174,6 +175,13 @@ protected: WFNSTracing *tracing); void check_breaker(); void pre_delete_server(EndpointAddress *addr); + + struct TracingData + { + std::vector history; + WFServiceGovernance *sg; + }; + static void tracing_deleter(void *data); std::vector servers; From ca950fffaae6a0728f5ca23383c228d11986b754 Mon Sep 17 00:00:00 2001 From: XieHan Date: Sat, 30 Oct 2021 17:36:58 +0800 Subject: [PATCH 2/3] fix test case. --- src/nameservice/WFServiceGovernance.cc | 33 ++++++++------------------ test/upstream_unittest.cc | 2 ++ 2 files changed, 12 insertions(+), 23 deletions(-) diff --git a/src/nameservice/WFServiceGovernance.cc b/src/nameservice/WFServiceGovernance.cc index d62f2f13..364108ff 100644 --- a/src/nameservice/WFServiceGovernance.cc +++ b/src/nameservice/WFServiceGovernance.cc @@ -66,33 +66,19 @@ public: } }; -static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr) +static void copy_host_port(ParsedURI& uri, const EndpointAddress *addr) { - char *host = NULL; - char *port = NULL; - - if (!addr->host.empty()) + if (addr->host != uri.host) { - host = strdup(addr->host.c_str()); - if (!host) - return false; + free(uri.host); + uri.host = strdup(addr->host.c_str()); } - if (!addr->port.empty()) + if (addr->port != uri.port) { - port = strdup(addr->port.c_str()); - if (!port) - { - free(host); - return false; - } free(uri.port); - uri.port = port; + uri.port = strdup(addr->port.c_str()); } - - free(uri.host); - uri.host = host; - return true; } EndpointAddress::EndpointAddress(const std::string& address, @@ -128,8 +114,7 @@ WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *p EndpointAddress *addr; WFRouterTask *task; - if (this->select(params->uri, tracing, &addr) && - copy_host_port(params->uri, addr)) + if (this->select(params->uri, tracing, &addr)) { WFDnsResolver *resolver = WFGlobal::get_dns_resolver(); unsigned int dns_ttl_default = addr->params->dns_ttl_default; @@ -137,6 +122,8 @@ WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *p const struct EndpointParams *endpoint_params = &addr->params->endpoint_params; int dns_cache_level = params->retry_times == 0 ? DNS_CACHE_LEVEL_2 : DNS_CACHE_LEVEL_1; + + copy_host_port(params->uri, addr); task = resolver->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min, endpoint_params, std::move(callback)); @@ -340,7 +327,7 @@ bool WFServiceGovernance::select(const ParsedURI& uri, WFNSTracing *tracing, if (select_addr) { *addr = select_addr; - ++(*addr)->ref; + ++select_addr->ref; } pthread_rwlock_unlock(&this->rwlock); diff --git a/test/upstream_unittest.cc b/test/upstream_unittest.cc index 05835159..6a544faa 100644 --- a/test/upstream_unittest.cc +++ b/test/upstream_unittest.cc @@ -224,6 +224,7 @@ TEST(upstream_unittest, AddAndRemove) wait_group.wait(); EXPECT_TRUE(http_server1.start("127.0.0.1", 8001) == 0) << "http server start failed"; + ns->del_policy(name.c_str()); } TEST(upstream_unittest, FuseAndRecover) @@ -285,6 +286,7 @@ TEST(upstream_unittest, FuseAndRecover) series->start(); wait_group.wait(); + ns->del_policy("test_policy"); } TEST(upstream_unittest, TryAnother) From 59e1a8edc8c4c963ec1b2577165cb6e60f8d94ec Mon Sep 17 00:00:00 2001 From: XieHan Date: Sat, 30 Oct 2021 18:20:54 +0800 Subject: [PATCH 3/3] fix style --- src/nameservice/UpstreamPolicies.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/nameservice/UpstreamPolicies.cc b/src/nameservice/UpstreamPolicies.cc index 66b8b16b..02a4ca07 100644 --- a/src/nameservice/UpstreamPolicies.cc +++ b/src/nameservice/UpstreamPolicies.cc @@ -190,8 +190,8 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, WFNSTracing *tracing, if (select_addr) { - *addr = (EndpointAddress *)select_addr; - ++(*addr)->ref; + *addr = select_addr; + ++select_addr->ref; } pthread_rwlock_unlock(&this->rwlock);