GithubStar/workflow
1
0
Fork 0
mirror of https://github.com/sogou/workflow.git synced 2026-10-29 03:27:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #613 from Barenboim/master

Browse Source 
fix WFServiceGovernance's server ref bug
This commit is contained in:
xiehan
2021-10-30 18:47:37 +08:00
committed by GitHub
parent f81b71472c 59e1a8edc8
commit efbebcf448
5 changed files with 67 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/nameservice/UpstreamPolicies.cc
+7 -18
View File
@@ -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);
@@ -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<EndpointAddress *> *)(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;
}
src/nameservice/UpstreamPolicies.h
+2 -2
View File
@@ -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 <utility>
#include <vector>
src/nameservice/WFServiceGovernance.cc
+46 -78
View File
@@ -14,6 +14,7 @@
limitations under the License.
Authors: Li Yingxin (liyingxin@sogou-inc.com)
Xie Han (xiehan@sogou-inc.com)
*/
#include <vector>
@@ -65,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,
@@ -123,12 +110,11 @@ 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))
if (this->select(params->uri, tracing, &addr))
{
WFDnsResolver *resolver = WFGlobal::get_dns_resolver();
unsigned int dns_ttl_default = addr->params->dns_ttl_default;
@@ -136,28 +122,21 @@ 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));
if (!tracing->data)
tracing->data = addr;
else
struct TracingData *tracing_data = (struct TracingData *)tracing->data;
if (!tracing_data)
{
std::vector<EndpointAddress *> *v;
if (!tracing->deleter)
{
EndpointAddress *last_addr = (EndpointAddress *)tracing->data;
v = new std::vector<EndpointAddress *>;
v->push_back(last_addr);
tracing->deleter = WFServiceGovernance::tracing_deleter;
tracing->data = v;
}
else
v = (std::vector<EndpointAddress *> *)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 +146,31 @@ WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *p
void WFServiceGovernance::tracing_deleter(void *data)
{
delete (std::vector<EndpointAddress *> *)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<EndpointAddress *> *)(tracing->data);
for (auto *server : (*v))
for (EndpointAddress *server : tracing_data->history)
{
if (server == addr)
return true;
@@ -234,24 +223,14 @@ void WFServiceGovernance::success(RouteManager::RouteResult *result,
WFNSTracing *tracing,
CommTarget *target)
{
EndpointAddress *server;
if (tracing->deleter)
{
auto *v = (std::vector<EndpointAddress *> *)(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 +238,15 @@ void WFServiceGovernance::failed(RouteManager::RouteResult *result,
WFNSTracing *tracing,
CommTarget *target)
{
EndpointAddress *server;
if (tracing->deleter)
{
auto *v = (std::vector<EndpointAddress *> *)(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);
}
@@ -359,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);
src/nameservice/WFServiceGovernance.h
+10 -2
View File
@@ -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 <pthread.h>
#include <unordered_map>
@@ -174,6 +175,13 @@ protected:
WFNSTracing *tracing);
void check_breaker();
void pre_delete_server(EndpointAddress *addr);
struct TracingData
{
std::vector<EndpointAddress *> history;
WFServiceGovernance *sg;
};
static void tracing_deleter(void *data);
std::vector<EndpointAddress *> servers;
test/upstream_unittest.cc
+2
View File
@@ -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)