From 85d7b58d10524c2d401073cad9d71a5b822e54b7 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Sun, 7 Feb 2021 20:32:33 +0800 Subject: [PATCH 01/10] add service for governance and other features --- CMakeLists_Headers.txt | 1 + src/CMakeLists.txt | 3 + src/manager/UpstreamManager.cc | 12 +- src/manager/UpstreamPolicies.cc | 446 ++++--------------------------- src/manager/UpstreamPolicies.h | 133 ++------- src/service/CMakeLists.txt | 9 + src/service/ServiceGovernance.cc | 370 +++++++++++++++++++++++++ src/service/ServiceGovernance.h | 128 +++++++++ 8 files changed, 599 insertions(+), 503 deletions(-) create mode 100644 src/service/CMakeLists.txt create mode 100644 src/service/ServiceGovernance.cc create mode 100644 src/service/ServiceGovernance.h diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index 08e1126b..a2b7baff 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -85,6 +85,7 @@ set(INCLUDE_HEADERS src/factory/WFOperator.h src/factory/WFNameService.h src/factory/WFDNSResolver.h + src/service/ServiceGovernance.h ) if(KAFKA STREQUAL "y") diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 4536237c..ee2c002b 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -9,6 +9,7 @@ add_subdirectory(manager) add_subdirectory(algorithm) add_subdirectory(protocol) add_subdirectory(factory) +add_subdirectory(service) add_subdirectory(server) add_subdirectory(client) @@ -18,6 +19,7 @@ add_dependencies(manager LINK_HEADERS) add_dependencies(algorithm LINK_HEADERS) add_dependencies(protocol LINK_HEADERS) add_dependencies(factory LINK_HEADERS) +add_dependencies(service LINK_HEADERS) add_dependencies(server LINK_HEADERS) add_dependencies(client LINK_HEADERS) @@ -29,6 +31,7 @@ add_library( $ $ $ + $ $ $ ) diff --git a/src/manager/UpstreamManager.cc b/src/manager/UpstreamManager.cc index e0027b92..e6e9789f 100644 --- a/src/manager/UpstreamManager.cc +++ b/src/manager/UpstreamManager.cc @@ -119,7 +119,7 @@ int UpstreamManager::upstream_add_server(const std::string& name, const AddressParams *address_params) { WFNameService *ns = WFGlobal::get_name_service(); - UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); + ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { @@ -135,7 +135,7 @@ int UpstreamManager::upstream_remove_server(const std::string& name, const std::string& address) { WFNameService *ns = WFGlobal::get_name_service(); - UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); + ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { @@ -152,7 +152,7 @@ UpstreamManager::upstream_main_address_list(const std::string& name) { std::vector address; WFNameService *ns = WFGlobal::get_name_service(); - UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); + ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) policy->get_main_address(address); @@ -164,7 +164,7 @@ int UpstreamManager::upstream_disable_server(const std::string& name, const std::string& address) { WFNameService *ns = WFGlobal::get_name_service(); - UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); + ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { @@ -180,7 +180,7 @@ int UpstreamManager::upstream_enable_server(const std::string& name, const std::string& address) { WFNameService *ns = WFGlobal::get_name_service(); - UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); + ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { @@ -197,7 +197,7 @@ int UpstreamManager::upstream_replace_server(const std::string& name, const struct AddressParams *address_params) { WFNameService *ns = WFGlobal::get_name_service(); - UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); + ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { diff --git a/src/manager/UpstreamPolicies.cc b/src/manager/UpstreamPolicies.cc index a50fb81f..607f32db 100644 --- a/src/manager/UpstreamPolicies.cc +++ b/src/manager/UpstreamPolicies.cc @@ -21,363 +21,20 @@ #include "URIParser.h" #include "StringUtil.h" #include "UpstreamPolicies.h" -#include "WFDNSResolver.h" -#define DNS_CACHE_LEVEL_1 1 -#define DNS_CACHE_LEVEL_2 2 - -class WFSelectorFailTask : public WFRouterTask +UPSAddress::UPSAddress(const std::string& address, + const struct AddressParams *address_params) : + EndpointAddress(address, address_params) { -public: - WFSelectorFailTask(router_callback_t&& cb) - : WFRouterTask(std::move(cb)) - { - } - - virtual void dispatch() - { - this->state = WFT_STATE_TASK_ERROR; - this->error = WFT_ERR_UPSTREAM_UNAVAILABLE; - - return this->subtask_done(); - } -}; - -static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr) -{ - char *host = NULL; - char *port = NULL; - - if (!addr->host.empty()) - { - host = strdup(addr->host.c_str()); - if (!host) - return false; - } - - if (addr->port_value > 0) - { - port = strdup(addr->port.c_str()); - if (!port) - { - free(host); - return false; - } - free(uri.port); - uri.port = port; - } - - free(uri.host); - uri.host = host; - return true; -} - -EndpointAddress::EndpointAddress(const std::string& address, - const struct AddressParams *address_params) -{ - std::vector arr = StringUtil::split(address, ':'); - this->params = *address_params; - this->address = address; - this->list.next = NULL; - this->fail_count = 0; - static std::hash std_hash; for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++) this->consistent_hash[i] = std_hash(address + "|v" + std::to_string(i)); - if (this->params.weight == 0) - this->params.weight = 1; - - if (this->params.max_fails == 0) - this->params.max_fails = 1; - if (this->params.group_id < 0) this->params.group_id = -1; - if (arr.size() == 0) - this->host = ""; - else - this->host = arr[0]; - - if (arr.size() <= 1) - { - this->port = ""; - this->port_value = 0; - } - else - { - this->port = arr[1]; - this->port_value = atoi(arr[1].c_str()); - } -} - -WFRouterTask *UPSPolicy::create_router_task(const struct WFNSParams *params, - router_callback_t callback) -{ - EndpointAddress *addr; - WFRouterTask *task; - - if (this->select(params->uri, &addr) && copy_host_port(params->uri, addr)) - { - unsigned int dns_ttl_default = addr->params.dns_ttl_default; - unsigned int dns_ttl_min = addr->params.dns_ttl_min; - 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; - task = this->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min, - endpoint_params, std::move(callback)); - task->set_cookie(addr); - } - else - task = new WFSelectorFailTask(std::move(callback)); - - return task; -} - -inline void UPSPolicy::recover_server_from_breaker(EndpointAddress *addr) -{ - addr->fail_count = 0; - pthread_mutex_lock(&this->breaker_lock); - if (addr->list.next) - { - list_del(&addr->list); - addr->list.next = NULL; - this->recover_one_server(addr); - //this->server_list_change(); - } - pthread_mutex_unlock(&this->breaker_lock); -} - -inline void UPSPolicy::fuse_server_to_breaker(EndpointAddress *addr) -{ - pthread_mutex_lock(&this->breaker_lock); - if (!addr->list.next) - { - addr->broken_timeout = GET_CURRENT_SECOND + MTTR_SECOND; - list_add_tail(&addr->list, &this->breaker_list); - this->fuse_one_server(addr); - //this->server_list_change(); - } - pthread_mutex_unlock(&this->breaker_lock); -} - -void UPSPolicy::success(RouteManager::RouteResult *result, void *cookie, - CommTarget *target) -{ - pthread_rwlock_rdlock(&this->rwlock); - this->recover_server_from_breaker((EndpointAddress *)cookie); - pthread_rwlock_unlock(&this->rwlock); - - WFDNSResolver::success(result, NULL, target); -} - -void UPSPolicy::failed(RouteManager::RouteResult *result, void *cookie, - CommTarget *target) -{ - EndpointAddress *server = (EndpointAddress *)cookie; - - pthread_rwlock_rdlock(&this->rwlock); - size_t fail_count = ++server->fail_count; - if (fail_count == server->params.max_fails) - this->fuse_server_to_breaker(server); - - pthread_rwlock_unlock(&this->rwlock); - - WFDNSResolver::failed(result, NULL, target); -} - -void UPSPolicy::check_breaker() -{ - pthread_mutex_lock(&this->breaker_lock); - if (!list_empty(&this->breaker_list)) - { - int64_t cur_time = GET_CURRENT_SECOND; - struct list_head *pos, *tmp; - EndpointAddress *addr; - - list_for_each_safe(pos, tmp, &this->breaker_list) - { - addr = list_entry(pos, EndpointAddress, list); - if (cur_time >= addr->broken_timeout) - { - if (addr->fail_count >= addr->params.max_fails) - { - addr->fail_count = addr->params.max_fails - 1; - this->recover_one_server(addr); - } - list_del(pos); - addr->list.next = NULL; - } - } - } - pthread_mutex_unlock(&this->breaker_lock); - - //this->server_list_change(); -} - -const EndpointAddress *UPSPolicy::first_stradegy(const ParsedURI& uri) -{ - unsigned int idx = rand() % this->servers.size(); - return this->servers[idx]; -} - -const EndpointAddress *UPSPolicy::another_stradegy(const ParsedURI& uri) -{ - return this->first_stradegy(uri); -} - -bool UPSPolicy::select(const ParsedURI& uri, EndpointAddress **addr) -{ - pthread_rwlock_rdlock(&this->rwlock); - unsigned int n = (unsigned int)this->servers.size(); - - if (n == 0) - { - pthread_rwlock_unlock(&this->rwlock); - return false; - } - - this->check_breaker(); - if (this->nalives == 0) - { - pthread_rwlock_unlock(&this->rwlock); - return false; - } - - // select_addr == NULL will only happened in consistent_hash - const EndpointAddress *select_addr = this->first_stradegy(uri); - - if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails) - { - if (this->try_another) - select_addr = this->another_stradegy(uri); - } - - pthread_rwlock_unlock(&this->rwlock); - - if (select_addr) - { - *addr = (EndpointAddress *)select_addr; - return true; - } - - return false; -} - -void UPSPolicy::add_server_locked(EndpointAddress *addr) -{ - this->addresses.push_back(addr); - this->server_map[addr->address].push_back(addr); - this->servers.push_back(addr); - this->recover_one_server(addr); -} - -int UPSPolicy::remove_server_locked(const std::string& address) -{ - const auto map_it = this->server_map.find(address); - if (map_it != this->server_map.cend()) - { - for (EndpointAddress *addr : map_it->second) - { - // or not: it has already been -- in nalives - if (addr->fail_count < addr->params.max_fails) - this->fuse_one_server(addr); - } - - this->server_map.erase(map_it); - } - - size_t n = this->servers.size(); - size_t new_n = 0; - - for (size_t i = 0; i < n; i++) - { - if (this->servers[i]->address != address) - { - if (new_n != i) - this->servers[new_n++] = this->servers[i]; - else - new_n++; - } - } - - int ret = 0; - if (new_n < n) - { - this->servers.resize(new_n); - ret = n - new_n; - } - - return ret; -} - -void UPSPolicy::add_server(const std::string& address, - const AddressParams *address_params) -{ - EndpointAddress *addr = new EndpointAddress(address, address_params); - - pthread_rwlock_wrlock(&this->rwlock); - this->add_server_locked(addr); - pthread_rwlock_unlock(&this->rwlock); -} - -int UPSPolicy::remove_server(const std::string& address) -{ - int ret; - pthread_rwlock_wrlock(&this->rwlock); - ret = this->remove_server_locked(address); - pthread_rwlock_unlock(&this->rwlock); - return ret; -} - -int UPSPolicy::replace_server(const std::string& address, - const AddressParams *address_params) -{ - int ret; - EndpointAddress *addr = new EndpointAddress(address, address_params); - - pthread_rwlock_wrlock(&this->rwlock); - this->add_server_locked(addr); - ret = this->remove_server_locked(address); - pthread_rwlock_unlock(&this->rwlock); - return ret; -} - -void UPSPolicy::enable_server(const std::string& address) -{ - pthread_rwlock_rdlock(&this->rwlock); - const auto map_it = this->server_map.find(address); - if (map_it != this->server_map.cend()) - { - for (EndpointAddress *addr : map_it->second) - this->recover_server_from_breaker(addr); - } - pthread_rwlock_unlock(&this->rwlock); -} - -void UPSPolicy::disable_server(const std::string& address) -{ - pthread_rwlock_rdlock(&this->rwlock); - const auto map_it = this->server_map.find(address); - if (map_it != this->server_map.cend()) - { - for (EndpointAddress *addr : map_it->second) - { - addr->fail_count = addr->params.max_fails; - this->fuse_server_to_breaker(addr); - } - } - pthread_rwlock_unlock(&this->rwlock); -} - -void UPSPolicy::get_main_address(std::vector& addr_list) -{ - pthread_rwlock_rdlock(&this->rwlock); - - for (const EndpointAddress *server : this->servers) - addr_list.push_back(server->address); - - pthread_rwlock_unlock(&this->rwlock); + if (this->params.weight == 0) + this->params.weight = 1; } UPSGroupPolicy::UPSGroupPolicy() @@ -400,7 +57,7 @@ UPSGroupPolicy::~UPSGroupPolicy() } } -bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) +bool UPSGroupPolicy::select(const ParsedURI& uri, UPSAddress **addr) { pthread_rwlock_rdlock(&this->rwlock); unsigned int n = (unsigned int)this->servers.size(); @@ -419,7 +76,7 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) } // select_addr == NULL will only happened in consistent_hash - const EndpointAddress *select_addr = this->first_stradegy(uri); + const UPSAddress *select_addr = static_cast(this->first_stradegy(uri)); if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails) { @@ -428,7 +85,7 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) if (!select_addr && this->try_another) { - select_addr = this->another_stradegy(uri); + select_addr = static_cast(this->another_stradegy(uri)); select_addr = this->check_and_get(select_addr, false); } } @@ -440,7 +97,7 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) if (select_addr) { - *addr = (EndpointAddress *)select_addr; + *addr = (UPSAddress *)select_addr; return true; } @@ -449,8 +106,8 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) // flag true : guarantee addr != NULL, and please return an available one // flag false : means addr maybe useful but want one any way. addr may be NULL -inline const EndpointAddress *UPSGroupPolicy::check_and_get(const EndpointAddress *addr, - bool flag) +inline const UPSAddress *UPSGroupPolicy::check_and_get(const UPSAddress *addr, + bool flag) { if (flag == true) // && addr->fail_count >= addr->params.max_fails { @@ -463,7 +120,7 @@ inline const EndpointAddress *UPSGroupPolicy::check_and_get(const EndpointAddres if (addr && addr->fail_count >= addr->params.max_fails && addr->params.group_id >= 0) { - const EndpointAddress *tmp = addr->group->get_one(); + const UPSAddress *tmp = addr->group->get_one(); if (tmp) addr = tmp; } @@ -471,12 +128,12 @@ inline const EndpointAddress *UPSGroupPolicy::check_and_get(const EndpointAddres return addr; } -const EndpointAddress *EndpointGroup::get_one() +const UPSAddress *EndpointGroup::get_one() { if (this->nalives == 0) return NULL; - const EndpointAddress *addr = NULL; + const UPSAddress *addr = NULL; pthread_mutex_lock(&this->mutex); std::random_shuffle(this->mains.begin(), this->mains.end()); @@ -506,12 +163,12 @@ const EndpointAddress *EndpointGroup::get_one() return addr; } -const EndpointAddress *EndpointGroup::get_one_backup() +const UPSAddress *EndpointGroup::get_one_backup() { if (this->nalives == 0) return NULL; - const EndpointAddress *addr = NULL; + const UPSAddress *addr = NULL; pthread_mutex_lock(&this->mutex); std::random_shuffle(this->backups.begin(), this->backups.end()); @@ -528,7 +185,7 @@ const EndpointAddress *EndpointGroup::get_one_backup() return addr; } -void UPSGroupPolicy::add_server_locked(EndpointAddress *addr) +void UPSGroupPolicy::add_server_locked(UPSAddress *addr) { int group_id = addr->params.group_id; rb_node **p = &this->group_map.rb_node; @@ -578,14 +235,16 @@ void UPSGroupPolicy::add_server_locked(EndpointAddress *addr) int UPSGroupPolicy::remove_server_locked(const std::string& address) { + UPSAddress *addr; const auto map_it = this->server_map.find(address); if (map_it != this->server_map.cend()) { - for (EndpointAddress *addr : map_it->second) + for (EndpointAddress *server : map_it->second) { + addr = static_cast(server); EndpointGroup *group = addr->group; - std::vector *vec; + std::vector *vec; if (addr->params.server_type == 0) vec = &group->mains; @@ -638,34 +297,36 @@ int UPSGroupPolicy::remove_server_locked(const std::string& address) return ret; } -const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int hash) +const UPSAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int hash) { - const EndpointAddress *addr = NULL; + const UPSAddress *addr; + const UPSAddress *select_addr = NULL; unsigned int min_dis = (unsigned int)-1; for (const EndpointAddress *server : this->servers) { - if (this->is_alive_or_group_alive(server)) + addr = static_cast(server); + if (this->is_alive_or_group_alive(addr)) { for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++) { unsigned int dis = std::min - (hash - server->consistent_hash[i], - server->consistent_hash[i] - hash); + (hash - addr->consistent_hash[i], + addr->consistent_hash[i] - hash); if (dis < min_dis) { min_dis = dis; - addr = server; + select_addr = addr; } } } } - return this->check_and_get(addr, false); + return this->check_and_get(select_addr, false); } -void UPSWeightedRandomPolicy::add_server_locked(EndpointAddress *addr) +void UPSWeightedRandomPolicy::add_server_locked(UPSAddress *addr) { UPSGroupPolicy::add_server_locked(addr); if (addr->params.server_type == 0) @@ -675,12 +336,14 @@ void UPSWeightedRandomPolicy::add_server_locked(EndpointAddress *addr) int UPSWeightedRandomPolicy::remove_server_locked(const std::string& address) { + UPSAddress *addr; const auto map_it = this->server_map.find(address); if (map_it != this->server_map.cend()) { - for (EndpointAddress *addr : map_it->second) + for (EndpointAddress *server : map_it->second) { + addr = static_cast(server); if (addr->params.server_type == 0) this->total_weight -= addr->params.weight; } @@ -689,11 +352,12 @@ int UPSWeightedRandomPolicy::remove_server_locked(const std::string& address) return UPSGroupPolicy::remove_server_locked(address); } -const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& uri) +const UPSAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& uri) { int x = 0; int s = 0; size_t idx; + const UPSAddress *addr; int temp_weight = this->total_weight; if (temp_weight > 0) @@ -708,34 +372,37 @@ const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& if (idx == this->servers.size()) idx--; - return this->servers[idx]; + addr = static_cast(this->servers[idx]); + return addr; } -const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI& uri) +const UPSAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI& uri) { int temp_weight = this->available_weight; if (temp_weight == 0) return NULL; - const EndpointAddress *addr = NULL; + const UPSAddress *addr; + const UPSAddress *select_addr = NULL; int x = rand() % temp_weight; int s = 0; for (const EndpointAddress *server : this->servers) { - if (this->is_alive_or_group_alive(server)) + addr = static_cast(server); + if (this->is_alive_or_group_alive(addr)) { - addr = server; - s += server->params.weight; + select_addr = addr; + s += addr->params.weight; if (s > x) break; } } - return this->check_and_get(addr, false); + return this->check_and_get(select_addr, false); } -void UPSWeightedRandomPolicy::recover_one_server(const EndpointAddress *addr) +void UPSWeightedRandomPolicy::recover_one_server(const UPSAddress *addr) { this->nalives++; if (addr->group->nalives++ == 0 && addr->group->id > 0) @@ -745,7 +412,7 @@ void UPSWeightedRandomPolicy::recover_one_server(const EndpointAddress *addr) this->available_weight += addr->params.weight; } -void UPSWeightedRandomPolicy::fuse_one_server(const EndpointAddress *addr) +void UPSWeightedRandomPolicy::fuse_one_server(const UPSAddress *addr) { this->nalives--; if (--addr->group->nalives == 0 && addr->group->id > 0) @@ -755,7 +422,7 @@ void UPSWeightedRandomPolicy::fuse_one_server(const EndpointAddress *addr) this->available_weight -= addr->params.weight; } -const EndpointAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& uri) +const UPSAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& uri) { unsigned int hash_value; @@ -770,7 +437,7 @@ const EndpointAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& return this->consistent_hash_with_group(hash_value); } -const EndpointAddress *UPSManualPolicy::first_stradegy(const ParsedURI& uri) +const UPSAddress *UPSManualPolicy::first_stradegy(const ParsedURI& uri) { unsigned int idx = this->manual_select(uri.path ? uri.path : "", uri.query ? uri.query : "", @@ -779,10 +446,11 @@ const EndpointAddress *UPSManualPolicy::first_stradegy(const ParsedURI& uri) if (idx >= this->servers.size()) idx %= this->servers.size(); - return this->servers[idx]; + const UPSAddress *addr = static_cast(this->servers[idx]); + return addr; } -const EndpointAddress *UPSManualPolicy::another_stradegy(const ParsedURI& uri) +const UPSAddress *UPSManualPolicy::another_stradegy(const ParsedURI& uri) { unsigned int hash_value; @@ -791,8 +459,10 @@ const EndpointAddress *UPSManualPolicy::another_stradegy(const ParsedURI& uri) uri.query ? uri.query : "", uri.fragment ? uri.fragment : ""); else - hash_value = UPSConsistentHashPolicy::default_consistent_hash(uri.path ? uri.path : "", - uri.query ? uri.query : "", - uri.fragment ? uri.fragment : ""); + hash_value = UPSConsistentHashPolicy::default_consistent_hash( + uri.path ? uri.path : "", + uri.query ? uri.query : "", + uri.fragment ? uri.fragment : ""); return this->consistent_hash_with_group(hash_value); } + diff --git a/src/manager/UpstreamPolicies.h b/src/manager/UpstreamPolicies.h index ef36907b..b09efc05 100644 --- a/src/manager/UpstreamPolicies.h +++ b/src/manager/UpstreamPolicies.h @@ -20,7 +20,6 @@ #define _UPSTREAM_POLICIES_H_ #include -#include #include #include #include "URIParser.h" @@ -30,33 +29,20 @@ #include "WFGlobal.h" #include "WFTaskError.h" #include "UpstreamManager.h" - -#define MTTR_SECOND 30 -#define VIRTUAL_GROUP_SIZE 16 - -#define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() +#include "ServiceGovernance.h" class EndpointGroup; -class UPSPolicy; class UPSGroupPolicy; -class EndpointAddress +class UPSAddress : public EndpointAddress { public: EndpointGroup *group; - AddressParams params; - std::string address; - std::string host; - std::string port; - short port_value; - struct list_head list; - std::atomic fail_count; - long long broken_timeout; unsigned int consistent_hash[VIRTUAL_GROUP_SIZE]; public: - EndpointAddress(const std::string& address, - const struct AddressParams *address_params); + UPSAddress(const std::string& address, + const struct AddressParams *address_params); }; class EndpointGroup @@ -66,8 +52,8 @@ public: UPSGroupPolicy *policy; struct rb_node rb; pthread_mutex_t mutex; - std::vector mains; - std::vector backups; + std::vector mains; + std::vector backups; std::atomic nalives; int weight; @@ -81,82 +67,11 @@ public: } public: - const EndpointAddress *get_one(); - const EndpointAddress *get_one_backup(); + const UPSAddress *get_one(); + const UPSAddress *get_one_backup(); }; -class UPSPolicy : public WFDNSResolver -{ -public: - virtual WFRouterTask *create_router_task(const struct WFNSParams *params, - router_callback_t callback); - virtual void success(RouteManager::RouteResult *result, void *cookie, - CommTarget *target); - virtual void failed(RouteManager::RouteResult *result, void *cookie, - CommTarget *target); - - void add_server(const std::string& address, const AddressParams *address_params); - int remove_server(const std::string& address); - int replace_server(const std::string& address, const AddressParams *address_params); - - virtual void enable_server(const std::string& address); - virtual void disable_server(const std::string& address); - virtual void get_main_address(std::vector& addr_list); - // virtual void server_list_change(/* std::vector status */) {} - -public: - UPSPolicy() : - breaker_lock(PTHREAD_MUTEX_INITIALIZER), - rwlock(PTHREAD_RWLOCK_INITIALIZER) - { - this->nalives = 0; - this->try_another = false; - INIT_LIST_HEAD(&this->breaker_list); - } - - virtual ~UPSPolicy() - { - for (EndpointAddress *addr : this->addresses) - delete addr; - } - -private: - virtual bool select(const ParsedURI& uri, EndpointAddress **addr); - - virtual void recover_one_server(const EndpointAddress *addr) - { - this->nalives++; - } - - virtual void fuse_one_server(const EndpointAddress *addr) - { - this->nalives--; - } - - virtual void add_server_locked(EndpointAddress *addr); - virtual int remove_server_locked(const std::string& address); - - void recover_server_from_breaker(EndpointAddress *addr); - void fuse_server_to_breaker(EndpointAddress *addr); - - struct list_head breaker_list; - pthread_mutex_t breaker_lock; - -protected: - virtual const EndpointAddress *first_stradegy(const ParsedURI& uri); - virtual const EndpointAddress *another_stradegy(const ParsedURI& uri); - void check_breaker(); - - std::vector servers; // current servers - std::vector addresses; // memory management - std::unordered_map> server_map; - pthread_rwlock_t rwlock; - std::atomic nalives; - bool try_another; -}; - -class UPSGroupPolicy : public UPSPolicy +class UPSGroupPolicy : public ServiceGovernance { public: UPSGroupPolicy(); @@ -167,28 +82,28 @@ protected: EndpointGroup *default_group; private: - virtual void recover_one_server(const EndpointAddress *addr) + virtual void recover_one_server(const UPSAddress *addr) { this->nalives++; addr->group->nalives++; } - virtual void fuse_one_server(const EndpointAddress *addr) + virtual void fuse_one_server(const UPSAddress *addr) { this->nalives--; addr->group->nalives--; } - virtual bool select(const ParsedURI& uri, EndpointAddress **addr); + virtual bool select(const ParsedURI& uri, UPSAddress **addr); protected: - virtual void add_server_locked(EndpointAddress *addr); + virtual void add_server_locked(UPSAddress *addr); virtual int remove_server_locked(const std::string& address); - const EndpointAddress *consistent_hash_with_group(unsigned int hash); - const EndpointAddress *check_and_get(const EndpointAddress *addr, bool flag); + const UPSAddress *consistent_hash_with_group(unsigned int hash); + const UPSAddress *check_and_get(const UPSAddress *addr, bool flag); - inline bool is_alive_or_group_alive(const EndpointAddress *addr) const + inline bool is_alive_or_group_alive(const UPSAddress *addr) const { return ((addr->params.group_id < 0 && addr->fail_count < addr->params.max_fails) || @@ -206,17 +121,17 @@ public: this->available_weight = 0; this->try_another = try_another; } - const EndpointAddress *first_stradegy(const ParsedURI& uri); - const EndpointAddress *another_stradegy(const ParsedURI& uri); + const UPSAddress *first_stradegy(const ParsedURI& uri); + const UPSAddress *another_stradegy(const ParsedURI& uri); protected: int total_weight; int available_weight; private: - virtual void recover_one_server(const EndpointAddress *addr); - virtual void fuse_one_server(const EndpointAddress *addr); - virtual void add_server_locked(EndpointAddress *addr); + virtual void recover_one_server(const UPSAddress *addr); + virtual void fuse_one_server(const UPSAddress *addr); + virtual void add_server_locked(UPSAddress *addr); virtual int remove_server_locked(const std::string& address); }; @@ -234,7 +149,7 @@ public: } protected: - const EndpointAddress *first_stradegy(const ParsedURI& uri); + const UPSAddress *first_stradegy(const ParsedURI& uri); private: upstream_route_t consistent_hash; @@ -264,8 +179,8 @@ public: this->try_another_select = try_another_select; } - const EndpointAddress *first_stradegy(const ParsedURI& uri); - const EndpointAddress *another_stradegy(const ParsedURI& uri); + const UPSAddress *first_stradegy(const ParsedURI& uri); + const UPSAddress *another_stradegy(const ParsedURI& uri); private: upstream_route_t manual_select; diff --git a/src/service/CMakeLists.txt b/src/service/CMakeLists.txt new file mode 100644 index 00000000..8ad8c714 --- /dev/null +++ b/src/service/CMakeLists.txt @@ -0,0 +1,9 @@ +cmake_minimum_required(VERSION 3.6) +project(service) + +set(SRC + ServiceGovernance.cc +) + +add_library(${PROJECT_NAME} OBJECT ${SRC}) + diff --git a/src/service/ServiceGovernance.cc b/src/service/ServiceGovernance.cc new file mode 100644 index 00000000..e3fc8271 --- /dev/null +++ b/src/service/ServiceGovernance.cc @@ -0,0 +1,370 @@ +/* + Copyright (c) 2021 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Authors: Li Yingxin (liyingxin@sogou-inc.com) +*/ + +#include "URIParser.h" +#include "StringUtil.h" +#include "WFDNSResolver.h" +#include "ServiceGovernance.h" + +#define DNS_CACHE_LEVEL_1 1 +#define DNS_CACHE_LEVEL_2 2 + +class WFSelectorFailTask : public WFRouterTask +{ +public: + WFSelectorFailTask(router_callback_t&& cb) + : WFRouterTask(std::move(cb)) + { + } + + virtual void dispatch() + { + this->state = WFT_STATE_TASK_ERROR; + this->error = WFT_ERR_UPSTREAM_UNAVAILABLE; + + return this->subtask_done(); + } +}; + +static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr) +{ + char *host = NULL; + char *port = NULL; + + if (!addr->host.empty()) + { + host = strdup(addr->host.c_str()); + if (!host) + return false; + } + + if (addr->port_value > 0) + { + port = strdup(addr->port.c_str()); + if (!port) + { + free(host); + return false; + } + free(uri.port); + uri.port = port; + } + + free(uri.host); + uri.host = host; + return true; +} + +EndpointAddress::EndpointAddress(const std::string& address, + const struct AddressParams *address_params) +{ + std::vector arr = StringUtil::split(address, ':'); + this->params = *address_params; + this->address = address; + this->list.next = NULL; + this->fail_count = 0; + + if (this->params.max_fails == 0) + this->params.max_fails = 1; + + if (arr.size() == 0) + this->host = ""; + else + this->host = arr[0]; + + if (arr.size() <= 1) + { + this->port = ""; + this->port_value = 0; + } + else + { + this->port = arr[1]; + this->port_value = atoi(arr[1].c_str()); + } +} + +WFRouterTask *ServiceGovernance::create_router_task(const struct WFNSParams *params, + router_callback_t callback) +{ + EndpointAddress *addr; + WFRouterTask *task; + + if (this->select(params->uri, &addr) && copy_host_port(params->uri, addr)) + { + unsigned int dns_ttl_default = addr->params.dns_ttl_default; + unsigned int dns_ttl_min = addr->params.dns_ttl_min; + 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; + task = this->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min, + endpoint_params, std::move(callback)); + task->set_cookie(addr); + } + else + task = new WFSelectorFailTask(std::move(callback)); + + return task; +} + +inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr) +{ + addr->fail_count = 0; + pthread_mutex_lock(&this->breaker_lock); + if (addr->list.next) + { + list_del(&addr->list); + addr->list.next = NULL; + this->recover_one_server(addr); + //this->server_list_change(); + } + pthread_mutex_unlock(&this->breaker_lock); +} + +inline void ServiceGovernance::fuse_server_to_breaker(EndpointAddress *addr) +{ + pthread_mutex_lock(&this->breaker_lock); + if (!addr->list.next) + { + addr->broken_timeout = GET_CURRENT_SECOND + MTTR_SECOND; + list_add_tail(&addr->list, &this->breaker_list); + this->fuse_one_server(addr); + //this->server_list_change(); + } + pthread_mutex_unlock(&this->breaker_lock); +} + +void ServiceGovernance::success(RouteManager::RouteResult *result, void *cookie, + CommTarget *target) +{ + pthread_rwlock_rdlock(&this->rwlock); + this->recover_server_from_breaker((EndpointAddress *)cookie); + pthread_rwlock_unlock(&this->rwlock); + + WFDNSResolver::success(result, NULL, target); +} + +void ServiceGovernance::failed(RouteManager::RouteResult *result, void *cookie, + CommTarget *target) +{ + EndpointAddress *server = (EndpointAddress *)cookie; + + pthread_rwlock_rdlock(&this->rwlock); + size_t fail_count = ++server->fail_count; + if (fail_count == server->params.max_fails) + this->fuse_server_to_breaker(server); + + pthread_rwlock_unlock(&this->rwlock); + + WFDNSResolver::failed(result, NULL, target); +} + +void ServiceGovernance::check_breaker() +{ + pthread_mutex_lock(&this->breaker_lock); + if (!list_empty(&this->breaker_list)) + { + int64_t cur_time = GET_CURRENT_SECOND; + struct list_head *pos, *tmp; + EndpointAddress *addr; + + list_for_each_safe(pos, tmp, &this->breaker_list) + { + addr = list_entry(pos, EndpointAddress, list); + if (cur_time >= addr->broken_timeout) + { + if (addr->fail_count >= addr->params.max_fails) + { + addr->fail_count = addr->params.max_fails - 1; + this->recover_one_server(addr); + } + list_del(pos); + addr->list.next = NULL; + } + } + } + pthread_mutex_unlock(&this->breaker_lock); + + //this->server_list_change(); +} + +const EndpointAddress *ServiceGovernance::first_stradegy(const ParsedURI& uri) +{ + unsigned int idx = rand() % this->servers.size(); + return this->servers[idx]; +} + +const EndpointAddress *ServiceGovernance::another_stradegy(const ParsedURI& uri) +{ + return this->first_stradegy(uri); +} + +bool ServiceGovernance::select(const ParsedURI& uri, EndpointAddress **addr) +{ + pthread_rwlock_rdlock(&this->rwlock); + unsigned int n = (unsigned int)this->servers.size(); + + if (n == 0) + { + pthread_rwlock_unlock(&this->rwlock); + return false; + } + + this->check_breaker(); + if (this->nalives == 0) + { + pthread_rwlock_unlock(&this->rwlock); + return false; + } + + // select_addr == NULL will only happened in consistent_hash + const EndpointAddress *select_addr = this->first_stradegy(uri); + + if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails) + { + if (this->try_another) + select_addr = this->another_stradegy(uri); + } + + pthread_rwlock_unlock(&this->rwlock); + + if (select_addr) + { + *addr = (EndpointAddress *)select_addr; + return true; + } + + return false; +} + +void ServiceGovernance::add_server_locked(EndpointAddress *addr) +{ + this->addresses.push_back(addr); + this->server_map[addr->address].push_back(addr); + this->servers.push_back(addr); + this->recover_one_server(addr); +} + +int ServiceGovernance::remove_server_locked(const std::string& address) +{ + const auto map_it = this->server_map.find(address); + if (map_it != this->server_map.cend()) + { + for (EndpointAddress *addr : map_it->second) + { + // or not: it has already been -- in nalives + if (addr->fail_count < addr->params.max_fails) + this->fuse_one_server(addr); + } + + this->server_map.erase(map_it); + } + + size_t n = this->servers.size(); + size_t new_n = 0; + + for (size_t i = 0; i < n; i++) + { + if (this->servers[i]->address != address) + { + if (new_n != i) + this->servers[new_n++] = this->servers[i]; + else + new_n++; + } + } + + int ret = 0; + if (new_n < n) + { + this->servers.resize(new_n); + ret = n - new_n; + } + + return ret; +} + +void ServiceGovernance::add_server(const std::string& address, + const AddressParams *address_params) +{ + EndpointAddress *addr = new EndpointAddress(address, address_params); + + pthread_rwlock_wrlock(&this->rwlock); + this->add_server_locked(addr); + pthread_rwlock_unlock(&this->rwlock); +} + +int ServiceGovernance::remove_server(const std::string& address) +{ + int ret; + pthread_rwlock_wrlock(&this->rwlock); + ret = this->remove_server_locked(address); + pthread_rwlock_unlock(&this->rwlock); + return ret; +} + +int ServiceGovernance::replace_server(const std::string& address, + const AddressParams *address_params) +{ + int ret; + EndpointAddress *addr = new EndpointAddress(address, address_params); + + pthread_rwlock_wrlock(&this->rwlock); + this->add_server_locked(addr); + ret = this->remove_server_locked(address); + pthread_rwlock_unlock(&this->rwlock); + return ret; +} + +void ServiceGovernance::enable_server(const std::string& address) +{ + pthread_rwlock_rdlock(&this->rwlock); + const auto map_it = this->server_map.find(address); + if (map_it != this->server_map.cend()) + { + for (EndpointAddress *addr : map_it->second) + this->recover_server_from_breaker(addr); + } + pthread_rwlock_unlock(&this->rwlock); +} + +void ServiceGovernance::disable_server(const std::string& address) +{ + pthread_rwlock_rdlock(&this->rwlock); + const auto map_it = this->server_map.find(address); + if (map_it != this->server_map.cend()) + { + for (EndpointAddress *addr : map_it->second) + { + addr->fail_count = addr->params.max_fails; + this->fuse_server_to_breaker(addr); + } + } + pthread_rwlock_unlock(&this->rwlock); +} + +void ServiceGovernance::get_main_address(std::vector& addr_list) +{ + pthread_rwlock_rdlock(&this->rwlock); + + for (const EndpointAddress *server : this->servers) + addr_list.push_back(server->address); + + pthread_rwlock_unlock(&this->rwlock); +} + diff --git a/src/service/ServiceGovernance.h b/src/service/ServiceGovernance.h new file mode 100644 index 00000000..27df62d7 --- /dev/null +++ b/src/service/ServiceGovernance.h @@ -0,0 +1,128 @@ +/* + Copyright (c) 2021 Sogou, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + + Authors: Li Yingxin (liyingxin@sogou-inc.com) +*/ + +#ifndef _SERVICE_GOVERNANCE_H_ +#define _SERVICE_GOVERNANCE_H_ + +#include +#include +#include +#include +#include "URIParser.h" +#include "EndpointParams.h" +#include "WFNameService.h" +#include "WFDNSResolver.h" +#include "WFGlobal.h" +#include "WFTaskError.h" +#include "UpstreamManager.h" + +#define MTTR_SECOND 30 +#define VIRTUAL_GROUP_SIZE 16 + +#define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() + +class EndpointAddress +{ +public: + AddressParams params; + std::string address; + std::string host; + std::string port; + short port_value; //TODO + struct list_head list; + std::atomic fail_count; + long long broken_timeout; + +public: + EndpointAddress(const std::string& address, + const struct AddressParams *address_params); +}; + +class ServiceGovernance : public WFDNSResolver +{ +public: + virtual WFRouterTask *create_router_task(const struct WFNSParams *params, + router_callback_t callback); + virtual void success(RouteManager::RouteResult *result, void *cookie, + CommTarget *target); + virtual void failed(RouteManager::RouteResult *result, void *cookie, + CommTarget *target); + + void add_server(const std::string& address, const AddressParams *address_params); + int remove_server(const std::string& address); + int replace_server(const std::string& address, const AddressParams *address_params); + + virtual void enable_server(const std::string& address); + virtual void disable_server(const std::string& address); + virtual void get_main_address(std::vector& addr_list); + // virtual void server_list_change(/* std::vector status */) {} + +public: + ServiceGovernance() : + breaker_lock(PTHREAD_MUTEX_INITIALIZER), + rwlock(PTHREAD_RWLOCK_INITIALIZER) + { + this->nalives = 0; + this->try_another = false; + INIT_LIST_HEAD(&this->breaker_list); + } + + virtual ~ServiceGovernance() + { + for (EndpointAddress *addr : this->addresses) + delete addr; + } + +private: + virtual bool select(const ParsedURI& uri, EndpointAddress **addr); + + virtual void recover_one_server(const EndpointAddress *addr) + { + this->nalives++; + } + + virtual void fuse_one_server(const EndpointAddress *addr) + { + this->nalives--; + } + + virtual void add_server_locked(EndpointAddress *addr); + virtual int remove_server_locked(const std::string& address); + + void recover_server_from_breaker(EndpointAddress *addr); + void fuse_server_to_breaker(EndpointAddress *addr); + + struct list_head breaker_list; + pthread_mutex_t breaker_lock; + +protected: + virtual const EndpointAddress *first_stradegy(const ParsedURI& uri); + virtual const EndpointAddress *another_stradegy(const ParsedURI& uri); + void check_breaker(); + + std::vector servers; // current servers + std::vector addresses; // memory management + std::unordered_map> server_map; + pthread_rwlock_t rwlock; + std::atomic nalives; + bool try_another; +}; + +#endif + From 18eecb62a9a278cb6a15b62f652b163f47b9551c Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Thu, 11 Feb 2021 21:50:22 +0800 Subject: [PATCH 02/10] update AddressParams --- src/manager/UpstreamManager.cc | 2 +- src/manager/UpstreamPolicies.cc | 242 +++++++++++++++++++------------ src/manager/UpstreamPolicies.h | 68 +++++---- src/service/ServiceGovernance.cc | 77 ++++++---- src/service/ServiceGovernance.h | 47 ++++-- 5 files changed, 277 insertions(+), 159 deletions(-) diff --git a/src/manager/UpstreamManager.cc b/src/manager/UpstreamManager.cc index e6e9789f..caac2efe 100644 --- a/src/manager/UpstreamManager.cc +++ b/src/manager/UpstreamManager.cc @@ -152,7 +152,7 @@ UpstreamManager::upstream_main_address_list(const std::string& name) { std::vector address; WFNameService *ns = WFGlobal::get_name_service(); - ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); + UPSGroupPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) policy->get_main_address(address); diff --git a/src/manager/UpstreamPolicies.cc b/src/manager/UpstreamPolicies.cc index 607f32db..61c9e67a 100644 --- a/src/manager/UpstreamPolicies.cc +++ b/src/manager/UpstreamPolicies.cc @@ -22,19 +22,44 @@ #include "StringUtil.h" #include "UpstreamPolicies.h" -UPSAddress::UPSAddress(const std::string& address, - const struct AddressParams *address_params) : - EndpointAddress(address, address_params) +UPSAddrParams::UPSAddrParams() : + PolicyAddrParams(&ADDRESS_PARAMS_DEFAULT) +{ + UPSAddrParams(&ADDRESS_PARAMS_DEFAULT, ""); +} + +UPSAddrParams::UPSAddrParams(const struct AddressParams *params, + const std::string& address) : + PolicyAddrParams(params) { static std::hash std_hash; for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++) this->consistent_hash[i] = std_hash(address + "|v" + std::to_string(i)); - if (this->params.group_id < 0) - this->params.group_id = -1; + this->weight = params->weight; + this->server_type = params->server_type; + this->group_id = params->group_id; - if (this->params.weight == 0) - this->params.weight = 1; + if (this->group_id < 0) + this->group_id = -1; + + if (this->weight == 0) + this->weight = 1; +} + +void UPSGroupPolicy::get_main_address(std::vector& addr_list) +{ + UPSAddrParams *params; + pthread_rwlock_rdlock(&this->rwlock); + + for (const EndpointAddress *server : this->servers) + { + params = static_cast(server->params); + if (params->server_type == 0) + addr_list.push_back(server->address); + } + + pthread_rwlock_unlock(&this->rwlock); } UPSGroupPolicy::UPSGroupPolicy() @@ -57,7 +82,32 @@ UPSGroupPolicy::~UPSGroupPolicy() } } -bool UPSGroupPolicy::select(const ParsedURI& uri, UPSAddress **addr) +void UPSGroupPolicy::add_server(const std::string& address, + const AddressParams *params) +{ + EndpointAddress *addr = new EndpointAddress(address, + new UPSAddrParams(params, address)); + + pthread_rwlock_wrlock(&this->rwlock); + this->add_server_locked(addr); + pthread_rwlock_unlock(&this->rwlock); +} + +int UPSGroupPolicy::replace_server(const std::string& address, + const AddressParams *params) +{ + int ret; + EndpointAddress *addr = new EndpointAddress(address, + new UPSAddrParams(params, address)); + + pthread_rwlock_wrlock(&this->rwlock); + this->add_server_locked(addr); + ret = this->remove_server_locked(address); + pthread_rwlock_unlock(&this->rwlock); + return ret; +} + +bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) { pthread_rwlock_rdlock(&this->rwlock); unsigned int n = (unsigned int)this->servers.size(); @@ -76,16 +126,16 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, UPSAddress **addr) } // select_addr == NULL will only happened in consistent_hash - const UPSAddress *select_addr = static_cast(this->first_stradegy(uri)); + const EndpointAddress *select_addr = this->first_stradegy(uri); - if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails) + if (!select_addr || select_addr->fail_count >= select_addr->params->max_fails) { if (select_addr) select_addr = this->check_and_get(select_addr, true); if (!select_addr && this->try_another) { - select_addr = static_cast(this->another_stradegy(uri)); + select_addr = this->another_stradegy(uri); select_addr = this->check_and_get(select_addr, false); } } @@ -97,7 +147,7 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, UPSAddress **addr) if (select_addr) { - *addr = (UPSAddress *)select_addr; + *addr = (EndpointAddress *)select_addr; return true; } @@ -106,21 +156,22 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, UPSAddress **addr) // flag true : guarantee addr != NULL, and please return an available one // flag false : means addr maybe useful but want one any way. addr may be NULL -inline const UPSAddress *UPSGroupPolicy::check_and_get(const UPSAddress *addr, - bool flag) +inline const EndpointAddress *UPSGroupPolicy::check_and_get(const EndpointAddress *addr, + bool flag) { - if (flag == true) // && addr->fail_count >= addr->params.max_fails + UPSAddrParams *params = static_cast(addr->params); + if (flag == true) // && addr->fail_count >= addr->params->max_fails { - if (addr->params.group_id == -1) + if (params->group_id == -1) return NULL; - return addr->group->get_one(); + return params->group->get_one(); } - if (addr && addr->fail_count >= addr->params.max_fails && - addr->params.group_id >= 0) + if (addr && addr->fail_count >= addr->params->max_fails && + params->group_id >= 0) { - const UPSAddress *tmp = addr->group->get_one(); + const EndpointAddress *tmp = params->group->get_one(); if (tmp) addr = tmp; } @@ -128,18 +179,18 @@ inline const UPSAddress *UPSGroupPolicy::check_and_get(const UPSAddress *addr, return addr; } -const UPSAddress *EndpointGroup::get_one() +const EndpointAddress *EndpointGroup::get_one() { if (this->nalives == 0) return NULL; - const UPSAddress *addr = NULL; + const EndpointAddress *addr = NULL; pthread_mutex_lock(&this->mutex); std::random_shuffle(this->mains.begin(), this->mains.end()); for (size_t i = 0; i < this->mains.size(); i++) { - if (this->mains[i]->fail_count < this->mains[i]->params.max_fails) + if (this->mains[i]->fail_count < this->mains[i]->params->max_fails) { addr = this->mains[i]; break; @@ -151,7 +202,7 @@ const UPSAddress *EndpointGroup::get_one() std::random_shuffle(this->backups.begin(), this->backups.end()); for (size_t i = 0; i < this->backups.size(); i++) { - if (this->backups[i]->fail_count < this->backups[i]->params.max_fails) + if (this->backups[i]->fail_count < this->backups[i]->params->max_fails) { addr = this->backups[i]; break; @@ -163,18 +214,18 @@ const UPSAddress *EndpointGroup::get_one() return addr; } -const UPSAddress *EndpointGroup::get_one_backup() +const EndpointAddress *EndpointGroup::get_one_backup() { if (this->nalives == 0) return NULL; - const UPSAddress *addr = NULL; + const EndpointAddress *addr = NULL; pthread_mutex_lock(&this->mutex); std::random_shuffle(this->backups.begin(), this->backups.end()); for (size_t i = 0; i < this->backups.size(); i++) { - if (this->backups[i]->fail_count < this->backups[i]->params.max_fails) + if (this->backups[i]->fail_count < this->backups[i]->params->max_fails) { addr = this->backups[i]; break; @@ -185,9 +236,10 @@ const UPSAddress *EndpointGroup::get_one_backup() return addr; } -void UPSGroupPolicy::add_server_locked(UPSAddress *addr) +void UPSGroupPolicy::add_server_locked(EndpointAddress *addr) { - int group_id = addr->params.group_id; + UPSAddrParams *params = static_cast(addr->params); + int group_id = params->group_id; rb_node **p = &this->group_map.rb_node; rb_node *parent = NULL; EndpointGroup *group; @@ -195,7 +247,7 @@ void UPSGroupPolicy::add_server_locked(UPSAddress *addr) this->addresses.push_back(addr); this->server_map[addr->address].push_back(addr); - if (addr->params.server_type == 0) + if (params->server_type == 0) this->servers.push_back(addr); while (*p) @@ -219,12 +271,12 @@ void UPSGroupPolicy::add_server_locked(UPSAddress *addr) } pthread_mutex_lock(&group->mutex); - addr->group = group; + params->group = group; this->recover_one_server(addr); - if (addr->params.server_type == 0) + if (params->server_type == 0) { group->mains.push_back(addr); - group->weight += addr->params.weight; + group->weight += params->weight; } else group->backups.push_back(addr); @@ -235,29 +287,29 @@ void UPSGroupPolicy::add_server_locked(UPSAddress *addr) int UPSGroupPolicy::remove_server_locked(const std::string& address) { - UPSAddress *addr; + UPSAddrParams *params; const auto map_it = this->server_map.find(address); if (map_it != this->server_map.cend()) { - for (EndpointAddress *server : map_it->second) + for (EndpointAddress *addr : map_it->second) { - addr = static_cast(server); - EndpointGroup *group = addr->group; - std::vector *vec; + params = static_cast(addr->params); + EndpointGroup *group = params->group; + std::vector *vec; - if (addr->params.server_type == 0) + if (params->server_type == 0) vec = &group->mains; else vec = &group->backups; //std::lock_guard lock(group->mutex); pthread_mutex_lock(&group->mutex); - if (addr->fail_count < addr->params.max_fails) + if (addr->fail_count < params->max_fails) this->fuse_one_server(addr); - if (addr->params.server_type == 0) - group->weight -= addr->params.weight; + if (params->server_type == 0) + group->weight -= params->weight; for (auto it = vec->begin(); it != vec->end(); ++it) { @@ -297,67 +349,70 @@ int UPSGroupPolicy::remove_server_locked(const std::string& address) return ret; } -const UPSAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int hash) +const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int hash) { - const UPSAddress *addr; - const UPSAddress *select_addr = NULL; + const UPSAddrParams *params; + const EndpointAddress *addr = NULL; unsigned int min_dis = (unsigned int)-1; for (const EndpointAddress *server : this->servers) { - addr = static_cast(server); - if (this->is_alive_or_group_alive(addr)) + if (this->is_alive_or_group_alive(server)) { + params = static_cast(server->params); + for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++) { unsigned int dis = std::min - (hash - addr->consistent_hash[i], - addr->consistent_hash[i] - hash); + (hash - params->consistent_hash[i], + params->consistent_hash[i] - hash); if (dis < min_dis) { min_dis = dis; - select_addr = addr; + addr = server; } } } } - return this->check_and_get(select_addr, false); + return this->check_and_get(addr, false); } -void UPSWeightedRandomPolicy::add_server_locked(UPSAddress *addr) +void UPSWeightedRandomPolicy::add_server_locked(EndpointAddress *addr) { + UPSAddrParams *params = static_cast(addr->params); + UPSGroupPolicy::add_server_locked(addr); - if (addr->params.server_type == 0) - this->total_weight += addr->params.weight; + if (params->server_type == 0) + this->total_weight += params->weight; return; } int UPSWeightedRandomPolicy::remove_server_locked(const std::string& address) { - UPSAddress *addr; + UPSAddrParams *params; const auto map_it = this->server_map.find(address); if (map_it != this->server_map.cend()) { - for (EndpointAddress *server : map_it->second) + for (EndpointAddress *addr : map_it->second) { - addr = static_cast(server); - if (addr->params.server_type == 0) - this->total_weight -= addr->params.weight; + params = static_cast(addr->params); + if (params->server_type == 0) + this->total_weight -= params->weight; } } return UPSGroupPolicy::remove_server_locked(address); } -const UPSAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& uri) { int x = 0; int s = 0; size_t idx; - const UPSAddress *addr; + UPSAddrParams *params; int temp_weight = this->total_weight; if (temp_weight > 0) @@ -365,64 +420,68 @@ const UPSAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& uri) for (idx = 0; idx < this->servers.size(); idx++) { - s += this->servers[idx]->params.weight; + params = static_cast(this->servers[idx]->params); + s += params->weight; if (s > x) break; } if (idx == this->servers.size()) idx--; - addr = static_cast(this->servers[idx]); - return addr; + return this->servers[idx]; } -const UPSAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI& uri) { + UPSAddrParams *params; int temp_weight = this->available_weight; if (temp_weight == 0) return NULL; - const UPSAddress *addr; - const UPSAddress *select_addr = NULL; + const EndpointAddress *addr = NULL; int x = rand() % temp_weight; int s = 0; for (const EndpointAddress *server : this->servers) { - addr = static_cast(server); - if (this->is_alive_or_group_alive(addr)) + if (this->is_alive_or_group_alive(server)) { - select_addr = addr; - s += addr->params.weight; + addr = server; // TODO: please double check here + params = static_cast(server->params); + s += params->weight; if (s > x) break; } } - return this->check_and_get(select_addr, false); + return this->check_and_get(addr, false); } -void UPSWeightedRandomPolicy::recover_one_server(const UPSAddress *addr) +void UPSWeightedRandomPolicy::recover_one_server(const EndpointAddress *addr) { + UPSAddrParams *params = static_cast(addr->params); + this->nalives++; - if (addr->group->nalives++ == 0 && addr->group->id > 0) - this->available_weight += addr->group->weight; + if (params->group->nalives++ == 0 && params->group->id > 0) + this->available_weight += params->group->weight; - if (addr->params.group_id < 0 && addr->params.server_type == 0) - this->available_weight += addr->params.weight; + if (params->group_id < 0 && params->server_type == 0) + this->available_weight += params->weight; } -void UPSWeightedRandomPolicy::fuse_one_server(const UPSAddress *addr) +void UPSWeightedRandomPolicy::fuse_one_server(const EndpointAddress *addr) { - this->nalives--; - if (--addr->group->nalives == 0 && addr->group->id > 0) - this->available_weight -= addr->group->weight; + UPSAddrParams *params = static_cast(addr->params); - if (addr->params.group_id < 0 && addr->params.server_type == 0) - this->available_weight -= addr->params.weight; + this->nalives--; + if (--params->group->nalives == 0 && params->group->id > 0) + this->available_weight -= params->group->weight; + + if (params->group_id < 0 && params->server_type == 0) + this->available_weight -= params->weight; } -const UPSAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& uri) { unsigned int hash_value; @@ -437,7 +496,7 @@ const UPSAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& uri) return this->consistent_hash_with_group(hash_value); } -const UPSAddress *UPSManualPolicy::first_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSManualPolicy::first_stradegy(const ParsedURI& uri) { unsigned int idx = this->manual_select(uri.path ? uri.path : "", uri.query ? uri.query : "", @@ -446,11 +505,10 @@ const UPSAddress *UPSManualPolicy::first_stradegy(const ParsedURI& uri) if (idx >= this->servers.size()) idx %= this->servers.size(); - const UPSAddress *addr = static_cast(this->servers[idx]); - return addr; + return this->servers[idx]; } -const UPSAddress *UPSManualPolicy::another_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSManualPolicy::another_stradegy(const ParsedURI& uri) { unsigned int hash_value; @@ -459,10 +517,8 @@ const UPSAddress *UPSManualPolicy::another_stradegy(const ParsedURI& uri) uri.query ? uri.query : "", uri.fragment ? uri.fragment : ""); else - hash_value = UPSConsistentHashPolicy::default_consistent_hash( - uri.path ? uri.path : "", - uri.query ? uri.query : "", - uri.fragment ? uri.fragment : ""); + hash_value = UPSConsistentHashPolicy::default_consistent_hash(uri.path ? uri.path : "", + uri.query ? uri.query : "", + uri.fragment ? uri.fragment : ""); return this->consistent_hash_with_group(hash_value); } - diff --git a/src/manager/UpstreamPolicies.h b/src/manager/UpstreamPolicies.h index b09efc05..82d5a9da 100644 --- a/src/manager/UpstreamPolicies.h +++ b/src/manager/UpstreamPolicies.h @@ -34,15 +34,18 @@ class EndpointGroup; class UPSGroupPolicy; -class UPSAddress : public EndpointAddress +class UPSAddrParams : public PolicyAddrParams { public: + unsigned short weight; + short server_type; + int group_id; EndpointGroup *group; unsigned int consistent_hash[VIRTUAL_GROUP_SIZE]; -public: - UPSAddress(const std::string& address, - const struct AddressParams *address_params); + UPSAddrParams(); + UPSAddrParams(const struct AddressParams *params, const std::string& address); + virtual ~UPSAddrParams() {} }; class EndpointGroup @@ -52,8 +55,8 @@ public: UPSGroupPolicy *policy; struct rb_node rb; pthread_mutex_t mutex; - std::vector mains; - std::vector backups; + std::vector mains; + std::vector backups; std::atomic nalives; int weight; @@ -67,8 +70,8 @@ public: } public: - const UPSAddress *get_one(); - const UPSAddress *get_one_backup(); + const EndpointAddress *get_one(); + const EndpointAddress *get_one_backup(); }; class UPSGroupPolicy : public ServiceGovernance @@ -77,38 +80,45 @@ public: UPSGroupPolicy(); ~UPSGroupPolicy(); + void get_main_address(std::vector& addr_list); + protected: struct rb_root group_map; EndpointGroup *default_group; private: - virtual void recover_one_server(const UPSAddress *addr) + virtual void recover_one_server(const EndpointAddress *addr) { this->nalives++; - addr->group->nalives++; + UPSAddrParams *params = static_cast(addr->params); + params->group->nalives++; } - virtual void fuse_one_server(const UPSAddress *addr) + virtual void fuse_one_server(const EndpointAddress *addr) { this->nalives--; - addr->group->nalives--; + UPSAddrParams *params = static_cast(addr->params); + params->group->nalives--; } - virtual bool select(const ParsedURI& uri, UPSAddress **addr); + virtual bool select(const ParsedURI& uri, EndpointAddress **addr); + virtual void add_server(const std::string& address, const AddressParams *params); + virtual int replace_server(const std::string& address, const AddressParams *params); protected: - virtual void add_server_locked(UPSAddress *addr); + virtual void add_server_locked(EndpointAddress *addr); virtual int remove_server_locked(const std::string& address); - const UPSAddress *consistent_hash_with_group(unsigned int hash); - const UPSAddress *check_and_get(const UPSAddress *addr, bool flag); + const EndpointAddress *consistent_hash_with_group(unsigned int hash); + const EndpointAddress *check_and_get(const EndpointAddress *addr, bool flag); - inline bool is_alive_or_group_alive(const UPSAddress *addr) const + inline bool is_alive_or_group_alive(const EndpointAddress *addr) const { - return ((addr->params.group_id < 0 && - addr->fail_count < addr->params.max_fails) || - (addr->params.group_id >= 0 && - addr->group->nalives > 0)); + UPSAddrParams *params = static_cast(addr->params); + return ((params->group_id < 0 && + addr->fail_count < addr->params->max_fails) || + (params->group_id >= 0 && + params->group->nalives > 0)); } }; @@ -121,17 +131,17 @@ public: this->available_weight = 0; this->try_another = try_another; } - const UPSAddress *first_stradegy(const ParsedURI& uri); - const UPSAddress *another_stradegy(const ParsedURI& uri); + const EndpointAddress *first_stradegy(const ParsedURI& uri); + const EndpointAddress *another_stradegy(const ParsedURI& uri); protected: int total_weight; int available_weight; private: - virtual void recover_one_server(const UPSAddress *addr); - virtual void fuse_one_server(const UPSAddress *addr); - virtual void add_server_locked(UPSAddress *addr); + virtual void recover_one_server(const EndpointAddress *addr); + virtual void fuse_one_server(const EndpointAddress *addr); + virtual void add_server_locked(EndpointAddress *addr); virtual int remove_server_locked(const std::string& address); }; @@ -149,7 +159,7 @@ public: } protected: - const UPSAddress *first_stradegy(const ParsedURI& uri); + const EndpointAddress *first_stradegy(const ParsedURI& uri); private: upstream_route_t consistent_hash; @@ -179,8 +189,8 @@ public: this->try_another_select = try_another_select; } - const UPSAddress *first_stradegy(const ParsedURI& uri); - const UPSAddress *another_stradegy(const ParsedURI& uri); + const EndpointAddress *first_stradegy(const ParsedURI& uri); + const EndpointAddress *another_stradegy(const ParsedURI& uri); private: upstream_route_t manual_select; diff --git a/src/service/ServiceGovernance.cc b/src/service/ServiceGovernance.cc index e3fc8271..930af8b2 100644 --- a/src/service/ServiceGovernance.cc +++ b/src/service/ServiceGovernance.cc @@ -24,6 +24,19 @@ #define DNS_CACHE_LEVEL_1 1 #define DNS_CACHE_LEVEL_2 2 +PolicyAddrParams::PolicyAddrParams() +{ + PolicyAddrParams(ADDRESS_PARAMS_DEFAULT); +} + +PolicyAddrParams::PolicyAddrParams(const struct AddressParams *params) : + endpoint_params(params->endpoint_params) +{ + this->dns_ttl_default = params->dns_ttl_default; + this->dns_ttl_min = params->dns_ttl_min; + this->max_fails = params->max_fails; +} + class WFSelectorFailTask : public WFRouterTask { public: @@ -71,16 +84,17 @@ static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr) } EndpointAddress::EndpointAddress(const std::string& address, - const struct AddressParams *address_params) + PolicyAddrParams *address_params) { std::vector arr = StringUtil::split(address, ':'); - this->params = *address_params; - this->address = address; - this->list.next = NULL; - this->fail_count = 0; - if (this->params.max_fails == 0) - this->params.max_fails = 1; + this->params = address_params; + if (this->params->max_fails == 0) + this->params->max_fails = 1; + + this->address = address; + this->fail_count = 0; + this->list.next = NULL; if (arr.size() == 0) this->host = ""; @@ -107,9 +121,9 @@ WFRouterTask *ServiceGovernance::create_router_task(const struct WFNSParams *par if (this->select(params->uri, &addr) && copy_host_port(params->uri, addr)) { - unsigned int dns_ttl_default = addr->params.dns_ttl_default; - unsigned int dns_ttl_min = addr->params.dns_ttl_min; - const struct EndpointParams *endpoint_params = &addr->params.endpoint_params; + unsigned int dns_ttl_default = addr->params->dns_ttl_default; + unsigned int dns_ttl_min = addr->params->dns_ttl_min; + 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; task = this->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min, @@ -122,6 +136,12 @@ WFRouterTask *ServiceGovernance::create_router_task(const struct WFNSParams *par return task; } +void ServiceGovernance::server_list_change(const EndpointAddress *address, int state) +{ + fprintf(stderr, "server_list_change(). addr: %s state: %d\n", + address->address.c_str(), state); +} + inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr) { addr->fail_count = 0; @@ -131,7 +151,7 @@ inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr list_del(&addr->list); addr->list.next = NULL; this->recover_one_server(addr); - //this->server_list_change(); + this->server_list_change(addr, RECOVER_SERVER); } pthread_mutex_unlock(&this->breaker_lock); } @@ -144,7 +164,7 @@ inline void ServiceGovernance::fuse_server_to_breaker(EndpointAddress *addr) addr->broken_timeout = GET_CURRENT_SECOND + MTTR_SECOND; list_add_tail(&addr->list, &this->breaker_list); this->fuse_one_server(addr); - //this->server_list_change(); + this->server_list_change(addr, FUSE_SERVER); } pthread_mutex_unlock(&this->breaker_lock); } @@ -166,7 +186,7 @@ void ServiceGovernance::failed(RouteManager::RouteResult *result, void *cookie, pthread_rwlock_rdlock(&this->rwlock); size_t fail_count = ++server->fail_count; - if (fail_count == server->params.max_fails) + if (fail_count == server->params->max_fails) this->fuse_server_to_breaker(server); pthread_rwlock_unlock(&this->rwlock); @@ -188,19 +208,18 @@ void ServiceGovernance::check_breaker() addr = list_entry(pos, EndpointAddress, list); if (cur_time >= addr->broken_timeout) { - if (addr->fail_count >= addr->params.max_fails) + if (addr->fail_count >= addr->params->max_fails) { - addr->fail_count = addr->params.max_fails - 1; + addr->fail_count = addr->params->max_fails - 1; this->recover_one_server(addr); + this->server_list_change(addr, RECOVER_SERVER); } list_del(pos); addr->list.next = NULL; } } } - pthread_mutex_unlock(&this->breaker_lock); - - //this->server_list_change(); + pthread_mutex_unlock(&this->breaker_lock); } const EndpointAddress *ServiceGovernance::first_stradegy(const ParsedURI& uri) @@ -235,7 +254,7 @@ bool ServiceGovernance::select(const ParsedURI& uri, EndpointAddress **addr) // select_addr == NULL will only happened in consistent_hash const EndpointAddress *select_addr = this->first_stradegy(uri); - if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails) + if (!select_addr || select_addr->fail_count >= select_addr->params->max_fails) { if (this->try_another) select_addr = this->another_stradegy(uri); @@ -258,6 +277,7 @@ void ServiceGovernance::add_server_locked(EndpointAddress *addr) this->server_map[addr->address].push_back(addr); this->servers.push_back(addr); this->recover_one_server(addr); + this->server_list_change(addr, ADD_SERVER); } int ServiceGovernance::remove_server_locked(const std::string& address) @@ -268,8 +288,11 @@ int ServiceGovernance::remove_server_locked(const std::string& address) for (EndpointAddress *addr : map_it->second) { // or not: it has already been -- in nalives - if (addr->fail_count < addr->params.max_fails) + if (addr->fail_count < addr->params->max_fails) + { this->fuse_one_server(addr); + this->server_list_change(addr, REMOVE_SERVER); + } } this->server_map.erase(map_it); @@ -300,9 +323,10 @@ int ServiceGovernance::remove_server_locked(const std::string& address) } void ServiceGovernance::add_server(const std::string& address, - const AddressParams *address_params) + const AddressParams *params) { - EndpointAddress *addr = new EndpointAddress(address, address_params); + EndpointAddress *addr = new EndpointAddress(address, + new PolicyAddrParams(params)); pthread_rwlock_wrlock(&this->rwlock); this->add_server_locked(addr); @@ -319,10 +343,11 @@ int ServiceGovernance::remove_server(const std::string& address) } int ServiceGovernance::replace_server(const std::string& address, - const AddressParams *address_params) + const AddressParams *params) { int ret; - EndpointAddress *addr = new EndpointAddress(address, address_params); + EndpointAddress *addr = new EndpointAddress(address, + new PolicyAddrParams(params)); pthread_rwlock_wrlock(&this->rwlock); this->add_server_locked(addr); @@ -351,14 +376,14 @@ void ServiceGovernance::disable_server(const std::string& address) { for (EndpointAddress *addr : map_it->second) { - addr->fail_count = addr->params.max_fails; + addr->fail_count = addr->params->max_fails; this->fuse_server_to_breaker(addr); } } pthread_rwlock_unlock(&this->rwlock); } -void ServiceGovernance::get_main_address(std::vector& addr_list) +void ServiceGovernance::get_current_address(std::vector& addr_list) { pthread_rwlock_rdlock(&this->rwlock); diff --git a/src/service/ServiceGovernance.h b/src/service/ServiceGovernance.h index 27df62d7..bd7557f7 100644 --- a/src/service/ServiceGovernance.h +++ b/src/service/ServiceGovernance.h @@ -31,26 +31,53 @@ #include "WFTaskError.h" #include "UpstreamManager.h" -#define MTTR_SECOND 30 +#define MTTR_SECOND 5 #define VIRTUAL_GROUP_SIZE 16 #define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() +enum ServerChangeState +{ + ADD_SERVER = 0, + REMOVE_SERVER = 1, + RECOVER_SERVER = 2, + FUSE_SERVER = 3, +}; + +class PolicyAddrParams +{ +public: + struct EndpointParams endpoint_params; + unsigned int dns_ttl_default; + unsigned int dns_ttl_min; + unsigned int max_fails; + + PolicyAddrParams(); + PolicyAddrParams(const struct AddressParams *params); +}; +/* +struct address_list +{ + struct list_head node; + EndpointAddress *ptr; +}; +*/ class EndpointAddress { public: - AddressParams params; std::string address; std::string host; std::string port; - short port_value; //TODO + unsigned short port_value; //TODO struct list_head list; +// struct address_list list; std::atomic fail_count; long long broken_timeout; + PolicyAddrParams *params; public: - EndpointAddress(const std::string& address, - const struct AddressParams *address_params); + EndpointAddress(const std::string& address, PolicyAddrParams *params); + virtual ~EndpointAddress() { delete this->params; } }; class ServiceGovernance : public WFDNSResolver @@ -63,14 +90,14 @@ public: virtual void failed(RouteManager::RouteResult *result, void *cookie, CommTarget *target); - void add_server(const std::string& address, const AddressParams *address_params); - int remove_server(const std::string& address); - int replace_server(const std::string& address, const AddressParams *address_params); + virtual void add_server(const std::string& address, const AddressParams *params); + virtual int remove_server(const std::string& address); + virtual int replace_server(const std::string& address, const AddressParams *params); virtual void enable_server(const std::string& address); virtual void disable_server(const std::string& address); - virtual void get_main_address(std::vector& addr_list); - // virtual void server_list_change(/* std::vector status */) {} + virtual void get_current_address(std::vector& addr_list); + virtual void server_list_change(const EndpointAddress *address, int state); public: ServiceGovernance() : From c41fdd74d6d92e191823ac5eed9f23af41ad9ce5 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Thu, 11 Feb 2021 23:07:09 +0800 Subject: [PATCH 03/10] fix delete bug --- src/manager/UpstreamPolicies.h | 1 - src/service/ServiceGovernance.h | 2 +- test/upstream_unittest.cc | 6 ++---- 3 files changed, 3 insertions(+), 6 deletions(-) diff --git a/src/manager/UpstreamPolicies.h b/src/manager/UpstreamPolicies.h index 82d5a9da..301ba399 100644 --- a/src/manager/UpstreamPolicies.h +++ b/src/manager/UpstreamPolicies.h @@ -45,7 +45,6 @@ public: UPSAddrParams(); UPSAddrParams(const struct AddressParams *params, const std::string& address); - virtual ~UPSAddrParams() {} }; class EndpointGroup diff --git a/src/service/ServiceGovernance.h b/src/service/ServiceGovernance.h index bd7557f7..56455144 100644 --- a/src/service/ServiceGovernance.h +++ b/src/service/ServiceGovernance.h @@ -31,7 +31,7 @@ #include "WFTaskError.h" #include "UpstreamManager.h" -#define MTTR_SECOND 5 +#define MTTR_SECOND 30 #define VIRTUAL_GROUP_SIZE 16 #define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() diff --git a/test/upstream_unittest.cc b/test/upstream_unittest.cc index b57fad25..fe6de63c 100644 --- a/test/upstream_unittest.cc +++ b/test/upstream_unittest.cc @@ -105,8 +105,6 @@ TEST(upstream_unittest, BasicPolicy) { WFFacilities::WaitGroup wait_group(3); - register_upstream_hosts(); - char url[3][30] = {"http://weighted.random", "http://hash", "http://manual"}; http_callback_t cb = std::bind(basic_callback, std::placeholders::_1, @@ -128,7 +126,6 @@ TEST(upstream_unittest, EnableAndDisable) UpstreamManager::upstream_disable_server("weighted.random", "127.0.0.1:8001"); - //fprintf(stderr, "disable server and try......................\n"); std::string url = "http://weighted.random"; WFHttpTask *task = WFTaskFactory::create_http_task(url, REDIRECT_MAX, RETRY_MAX, [&wait_group, &url](WFHttpTask *task){ @@ -136,7 +133,6 @@ TEST(upstream_unittest, EnableAndDisable) EXPECT_EQ(state, WFT_STATE_TASK_ERROR); EXPECT_EQ(task->get_error(), WFT_ERR_UPSTREAM_UNAVAILABLE); UpstreamManager::upstream_enable_server("weighted.random", "127.0.0.1:8001"); - //fprintf(stderr, "ensable server and try......................\n"); auto *task2 = WFTaskFactory::create_http_task(url, REDIRECT_MAX, RETRY_MAX, std::bind(basic_callback, std::placeholders::_1, @@ -221,6 +217,8 @@ int main(int argc, char* argv[]) { ::testing::InitGoogleTest(&argc, argv); + register_upstream_hosts(); + EXPECT_TRUE(http_server1.start("127.0.0.1", 8001) == 0) << "http server start failed"; From 2bd99ef5371a9d31f70229f64dac0b114b12022d Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Sun, 14 Feb 2021 17:23:05 +0800 Subject: [PATCH 04/10] adjust breaker_list node and add set_mttr_second() --- src/service/ServiceGovernance.cc | 37 ++++++++++++++++---------------- src/service/ServiceGovernance.h | 22 +++++++++++-------- 2 files changed, 31 insertions(+), 28 deletions(-) diff --git a/src/service/ServiceGovernance.cc b/src/service/ServiceGovernance.cc index 930af8b2..a9efe701 100644 --- a/src/service/ServiceGovernance.cc +++ b/src/service/ServiceGovernance.cc @@ -66,7 +66,7 @@ static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr) return false; } - if (addr->port_value > 0) + if (!addr->port.empty()) { port = strdup(addr->port.c_str()); if (!port) @@ -94,7 +94,8 @@ EndpointAddress::EndpointAddress(const std::string& address, this->address = address; this->fail_count = 0; - this->list.next = NULL; + this->list.node.next = NULL; + this->list.ptr = this; if (arr.size() == 0) this->host = ""; @@ -102,15 +103,9 @@ EndpointAddress::EndpointAddress(const std::string& address, this->host = arr[0]; if (arr.size() <= 1) - { this->port = ""; - this->port_value = 0; - } else - { this->port = arr[1]; - this->port_value = atoi(arr[1].c_str()); - } } WFRouterTask *ServiceGovernance::create_router_task(const struct WFNSParams *params, @@ -146,10 +141,10 @@ inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr { addr->fail_count = 0; pthread_mutex_lock(&this->breaker_lock); - if (addr->list.next) + if (addr->list.node.next) { - list_del(&addr->list); - addr->list.next = NULL; + list_del(&addr->list.node); + addr->list.node.next = NULL; this->recover_one_server(addr); this->server_list_change(addr, RECOVER_SERVER); } @@ -159,10 +154,10 @@ inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr inline void ServiceGovernance::fuse_server_to_breaker(EndpointAddress *addr) { pthread_mutex_lock(&this->breaker_lock); - if (!addr->list.next) + if (!addr->list.node.next) { - addr->broken_timeout = GET_CURRENT_SECOND + MTTR_SECOND; - list_add_tail(&addr->list, &this->breaker_list); + addr->broken_timeout = GET_CURRENT_SECOND + this->mttr_second; + list_add_tail(&addr->list.node, &this->breaker_list); this->fuse_one_server(addr); this->server_list_change(addr, FUSE_SERVER); } @@ -201,11 +196,14 @@ void ServiceGovernance::check_breaker() { int64_t cur_time = GET_CURRENT_SECOND; struct list_head *pos, *tmp; + struct address_list *node; EndpointAddress *addr; list_for_each_safe(pos, tmp, &this->breaker_list) { - addr = list_entry(pos, EndpointAddress, list); + node = list_entry(pos, struct address_list, node); + addr = node->ptr; + if (cur_time >= addr->broken_timeout) { if (addr->fail_count >= addr->params->max_fails) @@ -215,7 +213,7 @@ void ServiceGovernance::check_breaker() this->server_list_change(addr, RECOVER_SERVER); } list_del(pos); - addr->list.next = NULL; + addr->list.node.next = NULL; } } } @@ -254,7 +252,8 @@ bool ServiceGovernance::select(const ParsedURI& uri, EndpointAddress **addr) // select_addr == NULL will only happened in consistent_hash const EndpointAddress *select_addr = this->first_stradegy(uri); - if (!select_addr || select_addr->fail_count >= select_addr->params->max_fails) + if (!select_addr || + select_addr->fail_count >= select_addr->params->max_fails) { if (this->try_another) select_addr = this->another_stradegy(uri); @@ -326,7 +325,7 @@ void ServiceGovernance::add_server(const std::string& address, const AddressParams *params) { EndpointAddress *addr = new EndpointAddress(address, - new PolicyAddrParams(params)); + new PolicyAddrParams(params)); pthread_rwlock_wrlock(&this->rwlock); this->add_server_locked(addr); @@ -347,7 +346,7 @@ int ServiceGovernance::replace_server(const std::string& address, { int ret; EndpointAddress *addr = new EndpointAddress(address, - new PolicyAddrParams(params)); + new PolicyAddrParams(params)); pthread_rwlock_wrlock(&this->rwlock); this->add_server_locked(addr); diff --git a/src/service/ServiceGovernance.h b/src/service/ServiceGovernance.h index 56455144..8bb330a5 100644 --- a/src/service/ServiceGovernance.h +++ b/src/service/ServiceGovernance.h @@ -31,7 +31,7 @@ #include "WFTaskError.h" #include "UpstreamManager.h" -#define MTTR_SECOND 30 +#define MTTR_SECOND_DEFAULT 30 #define VIRTUAL_GROUP_SIZE 16 #define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() @@ -55,22 +55,23 @@ public: PolicyAddrParams(); PolicyAddrParams(const struct AddressParams *params); }; -/* + +class EndpointAddress; + struct address_list { struct list_head node; EndpointAddress *ptr; +// address_list(EndpointAddress *addr) { ptr = addr; node.next = NULL; } }; -*/ + class EndpointAddress { public: std::string address; std::string host; std::string port; - unsigned short port_value; //TODO - struct list_head list; -// struct address_list list; + struct address_list list; std::atomic fail_count; long long broken_timeout; PolicyAddrParams *params; @@ -91,13 +92,14 @@ public: CommTarget *target); virtual void add_server(const std::string& address, const AddressParams *params); - virtual int remove_server(const std::string& address); + int remove_server(const std::string& address); virtual int replace_server(const std::string& address, const AddressParams *params); - virtual void enable_server(const std::string& address); - virtual void disable_server(const std::string& address); + void enable_server(const std::string& address); + void disable_server(const std::string& address); virtual void get_current_address(std::vector& addr_list); virtual void server_list_change(const EndpointAddress *address, int state); + void set_mttr_second(unsigned int second) { this->mttr_second = second; } public: ServiceGovernance() : @@ -106,6 +108,7 @@ public: { this->nalives = 0; this->try_another = false; + this->mttr_second = MTTR_SECOND_DEFAULT; INIT_LIST_HEAD(&this->breaker_list); } @@ -136,6 +139,7 @@ private: struct list_head breaker_list; pthread_mutex_t breaker_lock; + unsigned int mttr_second; protected: virtual const EndpointAddress *first_stradegy(const ParsedURI& uri); From 75847572efb73da8d4af86a40ee519ad7d51485b Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Sun, 14 Feb 2021 17:41:03 +0800 Subject: [PATCH 05/10] move WFNameService and WFDNSResolver to service --- CMakeLists_Headers.txt | 4 ++-- src/factory/CMakeLists.txt | 2 -- src/service/CMakeLists.txt | 2 ++ src/{factory => service}/WFDNSResolver.cc | 0 src/{factory => service}/WFDNSResolver.h | 0 src/{factory => service}/WFNameService.cc | 0 src/{factory => service}/WFNameService.h | 0 7 files changed, 4 insertions(+), 4 deletions(-) rename src/{factory => service}/WFDNSResolver.cc (100%) rename src/{factory => service}/WFDNSResolver.h (100%) rename src/{factory => service}/WFNameService.cc (100%) rename src/{factory => service}/WFNameService.h (100%) diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index a2b7baff..aa30c3a4 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -83,8 +83,8 @@ set(INCLUDE_HEADERS src/factory/WFAlgoTaskFactory.inl src/factory/Workflow.h src/factory/WFOperator.h - src/factory/WFNameService.h - src/factory/WFDNSResolver.h + src/service/WFNameService.h + src/service/WFDNSResolver.h src/service/ServiceGovernance.h ) diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index 74153a68..2bb2c02d 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -8,8 +8,6 @@ set(SRC MySQLTaskImpl.cc WFTaskFactory.cc Workflow.cc - WFNameService.cc - WFDNSResolver.cc ) add_library(${PROJECT_NAME} OBJECT ${SRC}) diff --git a/src/service/CMakeLists.txt b/src/service/CMakeLists.txt index 8ad8c714..ea8dbb83 100644 --- a/src/service/CMakeLists.txt +++ b/src/service/CMakeLists.txt @@ -2,6 +2,8 @@ cmake_minimum_required(VERSION 3.6) project(service) set(SRC + WFNameService.cc + WFDNSResolver.cc ServiceGovernance.cc ) diff --git a/src/factory/WFDNSResolver.cc b/src/service/WFDNSResolver.cc similarity index 100% rename from src/factory/WFDNSResolver.cc rename to src/service/WFDNSResolver.cc diff --git a/src/factory/WFDNSResolver.h b/src/service/WFDNSResolver.h similarity index 100% rename from src/factory/WFDNSResolver.h rename to src/service/WFDNSResolver.h diff --git a/src/factory/WFNameService.cc b/src/service/WFNameService.cc similarity index 100% rename from src/factory/WFNameService.cc rename to src/service/WFNameService.cc diff --git a/src/factory/WFNameService.h b/src/service/WFNameService.h similarity index 100% rename from src/factory/WFNameService.h rename to src/service/WFNameService.h From 0ade638e0679cb7083ee672c7196ebc84376e335 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Sun, 14 Feb 2021 17:52:34 +0800 Subject: [PATCH 06/10] rename service to nameservice and remove the implement of server_list_change() --- CMakeLists_Headers.txt | 6 +++--- src/CMakeLists.txt | 6 +++--- src/{service => nameservice}/CMakeLists.txt | 2 +- src/{service => nameservice}/ServiceGovernance.cc | 6 ------ src/{service => nameservice}/ServiceGovernance.h | 2 +- src/{service => nameservice}/WFDNSResolver.cc | 0 src/{service => nameservice}/WFDNSResolver.h | 0 src/{service => nameservice}/WFNameService.cc | 0 src/{service => nameservice}/WFNameService.h | 0 9 files changed, 8 insertions(+), 14 deletions(-) rename src/{service => nameservice}/CMakeLists.txt (87%) rename src/{service => nameservice}/ServiceGovernance.cc (98%) rename src/{service => nameservice}/ServiceGovernance.h (99%) rename src/{service => nameservice}/WFDNSResolver.cc (100%) rename src/{service => nameservice}/WFDNSResolver.h (100%) rename src/{service => nameservice}/WFNameService.cc (100%) rename src/{service => nameservice}/WFNameService.h (100%) diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index aa30c3a4..fbb9584f 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -83,9 +83,9 @@ set(INCLUDE_HEADERS src/factory/WFAlgoTaskFactory.inl src/factory/Workflow.h src/factory/WFOperator.h - src/service/WFNameService.h - src/service/WFDNSResolver.h - src/service/ServiceGovernance.h + src/nameservice/WFNameService.h + src/nameservice/WFDNSResolver.h + src/nameservice/ServiceGovernance.h ) if(KAFKA STREQUAL "y") diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index ee2c002b..1319eebe 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -9,7 +9,7 @@ add_subdirectory(manager) add_subdirectory(algorithm) add_subdirectory(protocol) add_subdirectory(factory) -add_subdirectory(service) +add_subdirectory(nameservice) add_subdirectory(server) add_subdirectory(client) @@ -19,7 +19,7 @@ add_dependencies(manager LINK_HEADERS) add_dependencies(algorithm LINK_HEADERS) add_dependencies(protocol LINK_HEADERS) add_dependencies(factory LINK_HEADERS) -add_dependencies(service LINK_HEADERS) +add_dependencies(nameservice LINK_HEADERS) add_dependencies(server LINK_HEADERS) add_dependencies(client LINK_HEADERS) @@ -31,7 +31,7 @@ add_library( $ $ $ - $ + $ $ $ ) diff --git a/src/service/CMakeLists.txt b/src/nameservice/CMakeLists.txt similarity index 87% rename from src/service/CMakeLists.txt rename to src/nameservice/CMakeLists.txt index ea8dbb83..ae07f38b 100644 --- a/src/service/CMakeLists.txt +++ b/src/nameservice/CMakeLists.txt @@ -1,5 +1,5 @@ cmake_minimum_required(VERSION 3.6) -project(service) +project(nameservice) set(SRC WFNameService.cc diff --git a/src/service/ServiceGovernance.cc b/src/nameservice/ServiceGovernance.cc similarity index 98% rename from src/service/ServiceGovernance.cc rename to src/nameservice/ServiceGovernance.cc index a9efe701..b51acc08 100644 --- a/src/service/ServiceGovernance.cc +++ b/src/nameservice/ServiceGovernance.cc @@ -131,12 +131,6 @@ WFRouterTask *ServiceGovernance::create_router_task(const struct WFNSParams *par return task; } -void ServiceGovernance::server_list_change(const EndpointAddress *address, int state) -{ - fprintf(stderr, "server_list_change(). addr: %s state: %d\n", - address->address.c_str(), state); -} - inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr) { addr->fail_count = 0; diff --git a/src/service/ServiceGovernance.h b/src/nameservice/ServiceGovernance.h similarity index 99% rename from src/service/ServiceGovernance.h rename to src/nameservice/ServiceGovernance.h index 8bb330a5..f4083e31 100644 --- a/src/service/ServiceGovernance.h +++ b/src/nameservice/ServiceGovernance.h @@ -98,7 +98,7 @@ public: void enable_server(const std::string& address); void disable_server(const std::string& address); virtual void get_current_address(std::vector& addr_list); - virtual void server_list_change(const EndpointAddress *address, int state); + virtual void server_list_change(const EndpointAddress *address, int state) {} void set_mttr_second(unsigned int second) { this->mttr_second = second; } public: diff --git a/src/service/WFDNSResolver.cc b/src/nameservice/WFDNSResolver.cc similarity index 100% rename from src/service/WFDNSResolver.cc rename to src/nameservice/WFDNSResolver.cc diff --git a/src/service/WFDNSResolver.h b/src/nameservice/WFDNSResolver.h similarity index 100% rename from src/service/WFDNSResolver.h rename to src/nameservice/WFDNSResolver.h diff --git a/src/service/WFNameService.cc b/src/nameservice/WFNameService.cc similarity index 100% rename from src/service/WFNameService.cc rename to src/nameservice/WFNameService.cc diff --git a/src/service/WFNameService.h b/src/nameservice/WFNameService.h similarity index 100% rename from src/service/WFNameService.h rename to src/nameservice/WFNameService.h From f0eaabf7292554c29b93d236454ba0e5bb392cac Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Sun, 14 Feb 2021 20:50:00 +0800 Subject: [PATCH 07/10] update some functions to public and accelerate unittest --- src/manager/UpstreamManager.cc | 10 +++++----- src/manager/UpstreamPolicies.h | 12 +++++++----- src/nameservice/ServiceGovernance.cc | 4 ++-- test/upstream_unittest.cc | 26 ++++++++++++++++++++------ 4 files changed, 34 insertions(+), 18 deletions(-) diff --git a/src/manager/UpstreamManager.cc b/src/manager/UpstreamManager.cc index caac2efe..aa36ba78 100644 --- a/src/manager/UpstreamManager.cc +++ b/src/manager/UpstreamManager.cc @@ -119,7 +119,7 @@ int UpstreamManager::upstream_add_server(const std::string& name, const AddressParams *address_params) { WFNameService *ns = WFGlobal::get_name_service(); - ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); + UPSGroupPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { @@ -135,7 +135,7 @@ int UpstreamManager::upstream_remove_server(const std::string& name, const std::string& address) { WFNameService *ns = WFGlobal::get_name_service(); - ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); + UPSGroupPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { @@ -164,7 +164,7 @@ int UpstreamManager::upstream_disable_server(const std::string& name, const std::string& address) { WFNameService *ns = WFGlobal::get_name_service(); - ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); + UPSGroupPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { @@ -180,7 +180,7 @@ int UpstreamManager::upstream_enable_server(const std::string& name, const std::string& address) { WFNameService *ns = WFGlobal::get_name_service(); - ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); + UPSGroupPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { @@ -197,7 +197,7 @@ int UpstreamManager::upstream_replace_server(const std::string& name, const struct AddressParams *address_params) { WFNameService *ns = WFGlobal::get_name_service(); - ServiceGovernance *policy = dynamic_cast(ns->get_policy(name.c_str())); + UPSGroupPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); if (policy) { diff --git a/src/manager/UpstreamPolicies.h b/src/manager/UpstreamPolicies.h index 301ba399..0b63e94c 100644 --- a/src/manager/UpstreamPolicies.h +++ b/src/manager/UpstreamPolicies.h @@ -44,7 +44,8 @@ public: unsigned int consistent_hash[VIRTUAL_GROUP_SIZE]; UPSAddrParams(); - UPSAddrParams(const struct AddressParams *params, const std::string& address); + UPSAddrParams(const struct AddressParams *params, + const std::string& address); }; class EndpointGroup @@ -79,6 +80,11 @@ public: UPSGroupPolicy(); ~UPSGroupPolicy(); + virtual bool select(const ParsedURI& uri, EndpointAddress **addr); + virtual void add_server(const std::string& address, + const AddressParams *params); + virtual int replace_server(const std::string& address, + const AddressParams *params); void get_main_address(std::vector& addr_list); protected: @@ -100,10 +106,6 @@ private: params->group->nalives--; } - virtual bool select(const ParsedURI& uri, EndpointAddress **addr); - virtual void add_server(const std::string& address, const AddressParams *params); - virtual int replace_server(const std::string& address, const AddressParams *params); - protected: virtual void add_server_locked(EndpointAddress *addr); virtual int remove_server_locked(const std::string& address); diff --git a/src/nameservice/ServiceGovernance.cc b/src/nameservice/ServiceGovernance.cc index b51acc08..c0907faa 100644 --- a/src/nameservice/ServiceGovernance.cc +++ b/src/nameservice/ServiceGovernance.cc @@ -40,8 +40,8 @@ PolicyAddrParams::PolicyAddrParams(const struct AddressParams *params) : class WFSelectorFailTask : public WFRouterTask { public: - WFSelectorFailTask(router_callback_t&& cb) - : WFRouterTask(std::move(cb)) + WFSelectorFailTask(router_callback_t&& cb) : + WFRouterTask(std::move(cb)) { } diff --git a/test/upstream_unittest.cc b/test/upstream_unittest.cc index fe6de63c..e321d804 100644 --- a/test/upstream_unittest.cc +++ b/test/upstream_unittest.cc @@ -1,5 +1,5 @@ /* - Copyright (c) 2020 Sogou, Inc. + Copyright (c) 2021 Sogou, Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. @@ -21,10 +21,11 @@ #include "workflow/WFHttpServer.h" #include "workflow/WFTaskFactory.h" #include "workflow/WFFacilities.h" +#include "workflow/UpstreamPolicies.h" -#define REDIRECT_MAX 3 -#define RETRY_MAX 3 -#define MTTR 30 +#define REDIRECT_MAX 2 +#define RETRY_MAX 2 +#define MTTR 2 #define MAX_FAILS 200 static void __http_process1(WFHttpTask *task) @@ -152,10 +153,23 @@ TEST(upstream_unittest, FuseAndRecover) WFHttpTask *task; SeriesWork *series; protocol::HttpRequest *req; - std::string url = "http://weighted.random"; + std::string url = "http://test_policy"; int batch = MAX_FAILS + 50; - int timeout = (MTTR + 3) * 1000000; + int timeout = (MTTR + 1) * 1000000; + + UPSWeightedRandomPolicy test_policy(false); + test_policy.set_mttr_second(MTTR); + AddressParams address_params = ADDRESS_PARAMS_DEFAULT; + address_params.weight = 1000; + test_policy.add_server("127.0.0.1:8001", &address_params); + + address_params.weight = 1; + test_policy.add_server("127.0.0.1:8002", &address_params); + + auto *ns = WFGlobal::get_name_service(); + EXPECT_EQ(ns->add_policy("test_policy", &test_policy), 0); + http_server1.stop(); fprintf(stderr, "server 1 stopped start %d tasks to fuse it\n", batch); ParallelWork *pwork = Workflow::create_parallel_work( From 2789384eb7c66fe2e2d4d20e9c164282ac1ded78 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Mon, 15 Feb 2021 01:11:39 +0800 Subject: [PATCH 08/10] rename address_entry and move UpstreamPolicies to service --- CMakeLists_Headers.txt | 2 +- src/manager/CMakeLists.txt | 1 - src/nameservice/CMakeLists.txt | 1 + src/nameservice/ServiceGovernance.cc | 22 +++++++++---------- src/nameservice/ServiceGovernance.h | 15 ++++++++----- .../UpstreamPolicies.cc | 0 .../UpstreamPolicies.h | 0 7 files changed, 22 insertions(+), 19 deletions(-) rename src/{manager => nameservice}/UpstreamPolicies.cc (100%) rename src/{manager => nameservice}/UpstreamPolicies.h (100%) diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index fbb9584f..23a378c1 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -66,7 +66,6 @@ set(INCLUDE_HEADERS src/manager/WFFuture.h src/manager/WFFacilities.h src/manager/WFFacilities.inl - src/manager/UpstreamPolicies.h src/util/EncodeStream.h src/util/LRUCache.h src/util/StringUtil.h @@ -86,6 +85,7 @@ set(INCLUDE_HEADERS src/nameservice/WFNameService.h src/nameservice/WFDNSResolver.h src/nameservice/ServiceGovernance.h + src/nameservice/UpstreamPolicies.h ) if(KAFKA STREQUAL "y") diff --git a/src/manager/CMakeLists.txt b/src/manager/CMakeLists.txt index 2f22b886..6a282556 100644 --- a/src/manager/CMakeLists.txt +++ b/src/manager/CMakeLists.txt @@ -6,7 +6,6 @@ set(SRC UpstreamManager.cc RouteManager.cc WFGlobal.cc - UpstreamPolicies.cc ) add_library(${PROJECT_NAME} OBJECT ${SRC}) diff --git a/src/nameservice/CMakeLists.txt b/src/nameservice/CMakeLists.txt index ae07f38b..c249dc75 100644 --- a/src/nameservice/CMakeLists.txt +++ b/src/nameservice/CMakeLists.txt @@ -5,6 +5,7 @@ set(SRC WFNameService.cc WFDNSResolver.cc ServiceGovernance.cc + UpstreamPolicies.cc ) add_library(${PROJECT_NAME} OBJECT ${SRC}) diff --git a/src/nameservice/ServiceGovernance.cc b/src/nameservice/ServiceGovernance.cc index c0907faa..39cfa013 100644 --- a/src/nameservice/ServiceGovernance.cc +++ b/src/nameservice/ServiceGovernance.cc @@ -94,8 +94,8 @@ EndpointAddress::EndpointAddress(const std::string& address, this->address = address; this->fail_count = 0; - this->list.node.next = NULL; - this->list.ptr = this; + this->entry.list.next = NULL; + this->entry.ptr = this; if (arr.size() == 0) this->host = ""; @@ -135,10 +135,10 @@ inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr { addr->fail_count = 0; pthread_mutex_lock(&this->breaker_lock); - if (addr->list.node.next) + if (addr->entry.list.next) { - list_del(&addr->list.node); - addr->list.node.next = NULL; + list_del(&addr->entry.list); + addr->entry.list.next = NULL; this->recover_one_server(addr); this->server_list_change(addr, RECOVER_SERVER); } @@ -148,10 +148,10 @@ inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr inline void ServiceGovernance::fuse_server_to_breaker(EndpointAddress *addr) { pthread_mutex_lock(&this->breaker_lock); - if (!addr->list.node.next) + if (!addr->entry.list.next) { addr->broken_timeout = GET_CURRENT_SECOND + this->mttr_second; - list_add_tail(&addr->list.node, &this->breaker_list); + list_add_tail(&addr->entry.list, &this->breaker_list); this->fuse_one_server(addr); this->server_list_change(addr, FUSE_SERVER); } @@ -190,13 +190,13 @@ void ServiceGovernance::check_breaker() { int64_t cur_time = GET_CURRENT_SECOND; struct list_head *pos, *tmp; - struct address_list *node; + struct address_entry *entry; EndpointAddress *addr; list_for_each_safe(pos, tmp, &this->breaker_list) { - node = list_entry(pos, struct address_list, node); - addr = node->ptr; + entry = list_entry(pos, struct address_entry, list); + addr = entry->ptr; if (cur_time >= addr->broken_timeout) { @@ -207,7 +207,7 @@ void ServiceGovernance::check_breaker() this->server_list_change(addr, RECOVER_SERVER); } list_del(pos); - addr->list.node.next = NULL; + addr->entry.list.next = NULL; } } } diff --git a/src/nameservice/ServiceGovernance.h b/src/nameservice/ServiceGovernance.h index f4083e31..bd0660a3 100644 --- a/src/nameservice/ServiceGovernance.h +++ b/src/nameservice/ServiceGovernance.h @@ -58,9 +58,9 @@ public: class EndpointAddress; -struct address_list +struct address_entry { - struct list_head node; + struct list_head list; EndpointAddress *ptr; // address_list(EndpointAddress *addr) { ptr = addr; node.next = NULL; } }; @@ -71,7 +71,7 @@ public: std::string address; std::string host; std::string port; - struct address_list list; + struct address_entry entry; std::atomic fail_count; long long broken_timeout; PolicyAddrParams *params; @@ -91,14 +91,17 @@ public: virtual void failed(RouteManager::RouteResult *result, void *cookie, CommTarget *target); - virtual void add_server(const std::string& address, const AddressParams *params); + virtual void add_server(const std::string& address, + const AddressParams *params); int remove_server(const std::string& address); - virtual int replace_server(const std::string& address, const AddressParams *params); + virtual int replace_server(const std::string& address, + const AddressParams *params); void enable_server(const std::string& address); void disable_server(const std::string& address); virtual void get_current_address(std::vector& addr_list); - virtual void server_list_change(const EndpointAddress *address, int state) {} + virtual void server_list_change(const EndpointAddress *address, int state) + {} void set_mttr_second(unsigned int second) { this->mttr_second = second; } public: diff --git a/src/manager/UpstreamPolicies.cc b/src/nameservice/UpstreamPolicies.cc similarity index 100% rename from src/manager/UpstreamPolicies.cc rename to src/nameservice/UpstreamPolicies.cc diff --git a/src/manager/UpstreamPolicies.h b/src/nameservice/UpstreamPolicies.h similarity index 100% rename from src/manager/UpstreamPolicies.h rename to src/nameservice/UpstreamPolicies.h From a872444db03869d9f6af495e7b5e8bf85186c593 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Tue, 16 Feb 2021 00:54:46 +0800 Subject: [PATCH 09/10] move address_entry into EndpointAddress --- src/nameservice/ServiceGovernance.cc | 11 +++++++---- src/nameservice/ServiceGovernance.h | 16 ++++++---------- 2 files changed, 13 insertions(+), 14 deletions(-) diff --git a/src/nameservice/ServiceGovernance.cc b/src/nameservice/ServiceGovernance.cc index 39cfa013..6e61fc44 100644 --- a/src/nameservice/ServiceGovernance.cc +++ b/src/nameservice/ServiceGovernance.cc @@ -16,8 +16,10 @@ Authors: Li Yingxin (liyingxin@sogou-inc.com) */ +#include #include "URIParser.h" #include "StringUtil.h" +#include "WFNameService.h" #include "WFDNSResolver.h" #include "ServiceGovernance.h" @@ -159,7 +161,7 @@ inline void ServiceGovernance::fuse_server_to_breaker(EndpointAddress *addr) } void ServiceGovernance::success(RouteManager::RouteResult *result, void *cookie, - CommTarget *target) + CommTarget *target) { pthread_rwlock_rdlock(&this->rwlock); this->recover_server_from_breaker((EndpointAddress *)cookie); @@ -169,7 +171,7 @@ void ServiceGovernance::success(RouteManager::RouteResult *result, void *cookie, } void ServiceGovernance::failed(RouteManager::RouteResult *result, void *cookie, - CommTarget *target) + CommTarget *target) { EndpointAddress *server = (EndpointAddress *)cookie; @@ -190,12 +192,13 @@ void ServiceGovernance::check_breaker() { int64_t cur_time = GET_CURRENT_SECOND; struct list_head *pos, *tmp; - struct address_entry *entry; + struct EndpointAddress::address_entry *entry; EndpointAddress *addr; list_for_each_safe(pos, tmp, &this->breaker_list) { - entry = list_entry(pos, struct address_entry, list); + entry = list_entry(pos, struct EndpointAddress::address_entry, + list); addr = entry->ptr; if (cur_time >= addr->broken_timeout) diff --git a/src/nameservice/ServiceGovernance.h b/src/nameservice/ServiceGovernance.h index bd0660a3..b8a55674 100644 --- a/src/nameservice/ServiceGovernance.h +++ b/src/nameservice/ServiceGovernance.h @@ -56,26 +56,22 @@ public: PolicyAddrParams(const struct AddressParams *params); }; -class EndpointAddress; - -struct address_entry -{ - struct list_head list; - EndpointAddress *ptr; -// address_list(EndpointAddress *addr) { ptr = addr; node.next = NULL; } -}; - class EndpointAddress { public: std::string address; std::string host; std::string port; - struct address_entry entry; std::atomic fail_count; long long broken_timeout; PolicyAddrParams *params; + struct address_entry + { + struct list_head list; + EndpointAddress *ptr; + } entry; + public: EndpointAddress(const std::string& address, PolicyAddrParams *params); virtual ~EndpointAddress() { delete this->params; } From 5e25a6129f9d8aa6448162225f736660171b716f Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Thu, 18 Feb 2021 21:54:52 +0800 Subject: [PATCH 10/10] update PolicyAddrParams constructor --- src/factory/CMakeLists.txt | 1 + src/nameservice/ServiceGovernance.cc | 11 ++++++++--- src/nameservice/ServiceGovernance.h | 6 +++--- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index 2bb2c02d..24689fa6 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -18,3 +18,4 @@ if (KAFKA STREQUAL "y") ) add_library("factory_kafka" OBJECT ${SRC}) endif () + diff --git a/src/nameservice/ServiceGovernance.cc b/src/nameservice/ServiceGovernance.cc index 6e61fc44..d6ae2d6d 100644 --- a/src/nameservice/ServiceGovernance.cc +++ b/src/nameservice/ServiceGovernance.cc @@ -22,16 +22,21 @@ #include "WFNameService.h" #include "WFDNSResolver.h" #include "ServiceGovernance.h" +#include "UpstreamManager.h" #define DNS_CACHE_LEVEL_1 1 #define DNS_CACHE_LEVEL_2 2 PolicyAddrParams::PolicyAddrParams() { - PolicyAddrParams(ADDRESS_PARAMS_DEFAULT); + const AddressParams *params = &ADDRESS_PARAMS_DEFAULT; + this->endpoint_params = params->endpoint_params; + this->dns_ttl_default = params->dns_ttl_default; + this->dns_ttl_min = params->dns_ttl_min; + this->max_fails = params->max_fails; } -PolicyAddrParams::PolicyAddrParams(const struct AddressParams *params) : +PolicyAddrParams::PolicyAddrParams(const AddressParams *params) : endpoint_params(params->endpoint_params) { this->dns_ttl_default = params->dns_ttl_default; @@ -214,7 +219,7 @@ void ServiceGovernance::check_breaker() } } } - pthread_mutex_unlock(&this->breaker_lock); + pthread_mutex_unlock(&this->breaker_lock); } const EndpointAddress *ServiceGovernance::first_stradegy(const ParsedURI& uri) diff --git a/src/nameservice/ServiceGovernance.h b/src/nameservice/ServiceGovernance.h index b8a55674..a87e65e2 100644 --- a/src/nameservice/ServiceGovernance.h +++ b/src/nameservice/ServiceGovernance.h @@ -17,7 +17,7 @@ */ #ifndef _SERVICE_GOVERNANCE_H_ -#define _SERVICE_GOVERNANCE_H_ +#define _SERVICE_GOVERNANCE_H_ #include #include @@ -31,7 +31,7 @@ #include "WFTaskError.h" #include "UpstreamManager.h" -#define MTTR_SECOND_DEFAULT 30 +#define MTTR_SECOND_DEFAULT 30 #define VIRTUAL_GROUP_SIZE 16 #define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() @@ -53,7 +53,7 @@ public: unsigned int max_fails; PolicyAddrParams(); - PolicyAddrParams(const struct AddressParams *params); + PolicyAddrParams(const AddressParams *params); }; class EndpointAddress