diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index 08e1126b..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 @@ -83,8 +82,10 @@ set(INCLUDE_HEADERS src/factory/WFAlgoTaskFactory.inl src/factory/Workflow.h src/factory/WFOperator.h - src/factory/WFNameService.h - src/factory/WFDNSResolver.h + src/nameservice/WFNameService.h + src/nameservice/WFDNSResolver.h + src/nameservice/ServiceGovernance.h + src/nameservice/UpstreamPolicies.h ) if(KAFKA STREQUAL "y") diff --git a/README.md b/README.md index 2fa65a45..ae7ceafd 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -[中文版](README_cn.md) +[中文版入口](README_cn.md) ## Sogou C++ Workflow diff --git a/docs/about-service-management.md b/docs/about-service-management.md index 3f6bd073..2185dac4 100644 --- a/docs/about-service-management.md +++ b/docs/about-service-management.md @@ -159,7 +159,7 @@ struct EndpointParams int ssl_connect_timeout; }; -// In UpstreamMananger.h +// In ServiceGovernance.h struct AddressParams { struct EndpointParams endpoint_params; ///< Connection config diff --git a/docs/en/about-service-management.md b/docs/en/about-service-management.md index 59668ce4..49e340dd 100644 --- a/docs/en/about-service-management.md +++ b/docs/en/about-service-management.md @@ -172,7 +172,7 @@ struct EndpointParams int ssl_connect_timeout; }; -// In UpstreamMananger.h +// In ServiceGovernance.h struct AddressParams { struct EndpointParams endpoint_params; ///< Connection config diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 2e409e73..9877e423 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(nameservice) 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(nameservice LINK_HEADERS) add_dependencies(server LINK_HEADERS) add_dependencies(client LINK_HEADERS) @@ -29,6 +31,7 @@ add_library( $ $ $ + $ $ $ ) @@ -86,7 +89,7 @@ install( install ( - TARGETS ${CMAKE_PROJECT_NAME} EXPORT ${CMAKE_PROJECT_NAME}-targets + TARGETS ${PROJECT_NAME} EXPORT ${CMAKE_PROJECT_NAME}-targets ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR} LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR} diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index 74153a68..24689fa6 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}) @@ -20,3 +18,4 @@ if (KAFKA STREQUAL "y") ) add_library("factory_kafka" OBJECT ${SRC}) endif () + diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index 06f3c280..f4b17956 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -346,6 +346,12 @@ public: int get_state() const { return this->state; } int get_error() const { return this->error; } +public: + void set_callback(std::function cb) + { + this->callback = std::move(cb); + } + protected: virtual SubTask *done() { diff --git a/src/kernel/SubTask.h b/src/kernel/SubTask.h index dc7cf8cf..4d3f9ac7 100644 --- a/src/kernel/SubTask.h +++ b/src/kernel/SubTask.h @@ -65,7 +65,7 @@ public: this->subtasks_nr = n; } - SubTask **get_subtasks(size_t *n) + SubTask **get_subtasks(size_t *n) const { *n = this->subtasks_nr; return this->subtasks; 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/manager/UpstreamManager.cc b/src/manager/UpstreamManager.cc index e0027b92..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(); - UPSPolicy *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(); - UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); + UPSGroupPolicy *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())); + UPSGroupPolicy *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())); + 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(); - UPSPolicy *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(); - UPSPolicy *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/UpstreamManager.h b/src/manager/UpstreamManager.h index 2daa2ef2..ca8544e6 100644 --- a/src/manager/UpstreamManager.h +++ b/src/manager/UpstreamManager.h @@ -23,6 +23,8 @@ #include #include "URIParser.h" #include "EndpointParams.h" +#include "ServiceGovernance.h" +#include "UpstreamPolicies.h" #include "WFGlobal.h" /** @@ -34,50 +36,6 @@ * - This is working only for the current process. */ -/** - * @brief Functional for consistent-hash OR manual-select - * @details - * - path/query/fragment is empty string when uri not contain that region - * - path/query/fragment would never be NULL - */ -using upstream_route_t = std::function; - -/** - * @brief Server config for upstream - * @details - * When call UpstreamManager::upstream_add_server, you can set custom config for each target -*/ -struct AddressParams -{ - struct EndpointParams endpoint_params; ///< Connection config - unsigned int dns_ttl_default; ///< in seconds, DNS TTL when network request success - unsigned int dns_ttl_min; ///< in seconds, DNS TTL when network request fail -/** - * - The max_fails directive sets the number of consecutive unsuccessful attempts to communicate with the server. - * - After 30s following the server failure, upstream probe the server with some live client’s requests. - * - If the probes have been successful, the server is marked as a live one. - * - If max_fails is set to 1, it means server would out of upstream selection in 30 seconds when failed only once - */ - unsigned int max_fails; ///< [1, INT32_MAX] max_fails = 0 means max_fails = 1 - unsigned short weight; ///< [1, 65535] weight = 0 means weight = 1. only for main server - int server_type; ///< 0 for main and 1 for backup - int group_id; ///< -1 means no group. Backup without group will be backup for any main -}; - -/** - * @brief Default server config for upstream - */ -static constexpr struct AddressParams ADDRESS_PARAMS_DEFAULT = -{ - .endpoint_params = ENDPOINT_PARAMS_DEFAULT, - .dns_ttl_default = 12 * 3600, - .dns_ttl_min = 180, - .max_fails = 200, - .weight = 1, - .server_type = 0, /* 0 for main and 1 for backup. */ - .group_id = -1, -}; - /** * @brief Upstream Management Class * @details diff --git a/src/nameservice/CMakeLists.txt b/src/nameservice/CMakeLists.txt new file mode 100644 index 00000000..c249dc75 --- /dev/null +++ b/src/nameservice/CMakeLists.txt @@ -0,0 +1,12 @@ +cmake_minimum_required(VERSION 3.6) +project(nameservice) + +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 new file mode 100644 index 00000000..ea8e65e4 --- /dev/null +++ b/src/nameservice/ServiceGovernance.cc @@ -0,0 +1,398 @@ +/* + 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 +#include "URIParser.h" +#include "StringUtil.h" +#include "WFNameService.h" +#include "WFDNSResolver.h" +#include "ServiceGovernance.h" +#include "UpstreamManager.h" + +#define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() + +#define DNS_CACHE_LEVEL_1 1 +#define DNS_CACHE_LEVEL_2 2 + +PolicyAddrParams::PolicyAddrParams() +{ + const struct 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) : + 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: + 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.empty()) + { + 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, + PolicyAddrParams *address_params) +{ + std::vector arr = StringUtil::split(address, ':'); + + this->params = address_params; + if (this->params->max_fails == 0) + this->params->max_fails = 1; + + this->address = address; + this->fail_count = 0; + this->entry.list.next = NULL; + this->entry.ptr = this; + + if (arr.size() == 0) + this->host = ""; + else + this->host = arr[0]; + + if (arr.size() <= 1) + this->port = ""; + else + this->port = arr[1]; +} + +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->entry.list.next) + { + list_del(&addr->entry.list); + addr->entry.list.next = NULL; + this->recover_one_server(addr); + this->server_list_change(addr, RECOVER_SERVER); + } + pthread_mutex_unlock(&this->breaker_lock); +} + +inline void ServiceGovernance::fuse_server_to_breaker(EndpointAddress *addr) +{ + pthread_mutex_lock(&this->breaker_lock); + if (!addr->entry.list.next) + { + addr->broken_timeout = GET_CURRENT_SECOND + this->mttr_second; + list_add_tail(&addr->entry.list, &this->breaker_list); + this->fuse_one_server(addr); + this->server_list_change(addr, FUSE_SERVER); + } + 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; + struct EndpointAddress::address_entry *entry; + EndpointAddress *addr; + + list_for_each_safe(pos, tmp, &this->breaker_list) + { + entry = list_entry(pos, struct EndpointAddress::address_entry, + list); + addr = entry->ptr; + + 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); + this->server_list_change(addr, RECOVER_SERVER); + } + list_del(pos); + addr->entry.list.next = NULL; + } + } + } + pthread_mutex_unlock(&this->breaker_lock); +} + +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); + this->server_list_change(addr, ADD_SERVER); +} + +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_list_change(addr, REMOVE_SERVER); + } + } + + 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 *params) +{ + EndpointAddress *addr = new EndpointAddress(address, + new PolicyAddrParams(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 *params) +{ + int ret; + EndpointAddress *addr = new EndpointAddress(address, + new PolicyAddrParams(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_current_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/nameservice/ServiceGovernance.h b/src/nameservice/ServiceGovernance.h new file mode 100644 index 00000000..2397c187 --- /dev/null +++ b/src/nameservice/ServiceGovernance.h @@ -0,0 +1,183 @@ +/* + 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" + +#define MTTR_SECOND_DEFAULT 30 +#define VIRTUAL_GROUP_SIZE 16 + +struct AddressParams +{ + struct EndpointParams endpoint_params; ///< Connection config + unsigned int dns_ttl_default; ///< in seconds, DNS TTL when network request success + unsigned int dns_ttl_min; ///< in seconds, DNS TTL when network request fail +/** + * - The max_fails directive sets the number of consecutive unsuccessful attempts to communicate with the server. + * - After 30s following the server failure, upstream probe the server with some live client`s requests. + * - If the probes have been successful, the server is marked as a live one. + * - If max_fails is set to 1, it means server would out of upstream selection in 30 seconds when failed only once + */ + unsigned int max_fails; ///< [1, INT32_MAX] max_fails = 0 means max_fails = 1 + unsigned short weight; ///< [1, 65535] weight = 0 means weight = 1. only for main server + int server_type; ///< 0 for main and 1 for backup + int group_id; ///< -1 means no group. Backup without group will be backup for any main +}; + +static constexpr struct AddressParams ADDRESS_PARAMS_DEFAULT = +{ + .endpoint_params = ENDPOINT_PARAMS_DEFAULT, + .dns_ttl_default = 12 * 3600, + .dns_ttl_min = 180, + .max_fails = 200, + .weight = 1, + .server_type = 0, /* 0 for main and 1 for backup. */ + .group_id = -1, +}; + +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); +}; + +class EndpointAddress +{ +public: + std::string address; + std::string host; + std::string port; + 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; } +}; + +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); + + virtual void add_server(const std::string& address, + const struct AddressParams *params); + int remove_server(const std::string& address); + virtual int replace_server(const std::string& address, + const struct 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) + {} + void set_mttr_second(unsigned int second) { this->mttr_second = second; } + +public: + ServiceGovernance() : + breaker_lock(PTHREAD_MUTEX_INITIALIZER), + rwlock(PTHREAD_RWLOCK_INITIALIZER) + { + this->nalives = 0; + this->try_another = false; + this->mttr_second = MTTR_SECOND_DEFAULT; + 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; + unsigned int mttr_second; + +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 + diff --git a/src/manager/UpstreamPolicies.cc b/src/nameservice/UpstreamPolicies.cc similarity index 50% rename from src/manager/UpstreamPolicies.cc rename to src/nameservice/UpstreamPolicies.cc index a50fb81f..61c9e67a 100644 --- a/src/manager/UpstreamPolicies.cc +++ b/src/nameservice/UpstreamPolicies.cc @@ -21,361 +21,43 @@ #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 +UPSAddrParams::UPSAddrParams() : + PolicyAddrParams(&ADDRESS_PARAMS_DEFAULT) { -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; + UPSAddrParams(&ADDRESS_PARAMS_DEFAULT, ""); } -EndpointAddress::EndpointAddress(const std::string& address, - const struct AddressParams *address_params) +UPSAddrParams::UPSAddrParams(const struct AddressParams *params, + const std::string& address) : + PolicyAddrParams(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; + this->weight = params->weight; + this->server_type = params->server_type; + this->group_id = params->group_id; - if (this->params.max_fails == 0) - this->params.max_fails = 1; + if (this->group_id < 0) + this->group_id = -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()); - } + if (this->weight == 0) + this->weight = 1; } -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) +void UPSGroupPolicy::get_main_address(std::vector& addr_list) { + UPSAddrParams *params; pthread_rwlock_rdlock(&this->rwlock); for (const EndpointAddress *server : this->servers) - addr_list.push_back(server->address); + { + params = static_cast(server->params); + if (params->server_type == 0) + addr_list.push_back(server->address); + } pthread_rwlock_unlock(&this->rwlock); } @@ -400,6 +82,31 @@ UPSGroupPolicy::~UPSGroupPolicy() } } +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); @@ -421,7 +128,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); - 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); @@ -452,18 +159,19 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) 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 EndpointAddress *tmp = addr->group->get_one(); + const EndpointAddress *tmp = params->group->get_one(); if (tmp) addr = tmp; } @@ -482,7 +190,7 @@ const EndpointAddress *EndpointGroup::get_one() 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; @@ -494,7 +202,7 @@ const EndpointAddress *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; @@ -517,7 +225,7 @@ const EndpointAddress *EndpointGroup::get_one_backup() 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; @@ -530,7 +238,8 @@ const EndpointAddress *EndpointGroup::get_one_backup() 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; @@ -538,7 +247,7 @@ void UPSGroupPolicy::add_server_locked(EndpointAddress *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) @@ -562,12 +271,12 @@ void UPSGroupPolicy::add_server_locked(EndpointAddress *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); @@ -578,27 +287,29 @@ void UPSGroupPolicy::add_server_locked(EndpointAddress *addr) int UPSGroupPolicy::remove_server_locked(const std::string& address) { + UPSAddrParams *params; const auto map_it = this->server_map.find(address); if (map_it != this->server_map.cend()) { for (EndpointAddress *addr : map_it->second) { - EndpointGroup *group = addr->group; + 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) { @@ -640,6 +351,7 @@ int UPSGroupPolicy::remove_server_locked(const std::string& address) const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int hash) { + const UPSAddrParams *params; const EndpointAddress *addr = NULL; unsigned int min_dis = (unsigned int)-1; @@ -647,11 +359,13 @@ const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int h { 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 - server->consistent_hash[i], - server->consistent_hash[i] - hash); + (hash - params->consistent_hash[i], + params->consistent_hash[i] - hash); if (dis < min_dis) { @@ -667,22 +381,26 @@ const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int h 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) { + UPSAddrParams *params; const auto map_it = this->server_map.find(address); if (map_it != this->server_map.cend()) { for (EndpointAddress *addr : map_it->second) { - 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; } } @@ -694,6 +412,7 @@ const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& int x = 0; int s = 0; size_t idx; + UPSAddrParams *params; int temp_weight = this->total_weight; if (temp_weight > 0) @@ -701,7 +420,8 @@ const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& 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; } @@ -713,6 +433,7 @@ const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI& uri) { + UPSAddrParams *params; int temp_weight = this->available_weight; if (temp_weight == 0) return NULL; @@ -725,8 +446,9 @@ const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI { if (this->is_alive_or_group_alive(server)) { - addr = server; - s += server->params.weight; + addr = server; // TODO: please double check here + params = static_cast(server->params); + s += params->weight; if (s > x) break; } @@ -737,22 +459,26 @@ const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI void UPSWeightedRandomPolicy::recover_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; } 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 EndpointAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& uri) diff --git a/src/manager/UpstreamPolicies.h b/src/nameservice/UpstreamPolicies.h similarity index 57% rename from src/manager/UpstreamPolicies.h rename to src/nameservice/UpstreamPolicies.h index ef36907b..f1d94633 100644 --- a/src/manager/UpstreamPolicies.h +++ b/src/nameservice/UpstreamPolicies.h @@ -20,7 +20,6 @@ #define _UPSTREAM_POLICIES_H_ #include -#include #include #include #include "URIParser.h" @@ -29,34 +28,25 @@ #include "WFDNSResolver.h" #include "WFGlobal.h" #include "WFTaskError.h" -#include "UpstreamManager.h" +#include "ServiceGovernance.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() +using upstream_route_t = std::function; class EndpointGroup; -class UPSPolicy; class UPSGroupPolicy; -class EndpointAddress +class UPSAddrParams : public PolicyAddrParams { public: + unsigned short weight; + short server_type; + int group_id; 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); + UPSAddrParams(); + UPSAddrParams(const struct AddressParams *params, + const std::string& address); }; class EndpointGroup @@ -85,83 +75,19 @@ public: const EndpointAddress *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(); ~UPSGroupPolicy(); + virtual bool select(const ParsedURI& uri, EndpointAddress **addr); + virtual void add_server(const std::string& address, + const struct AddressParams *params); + virtual int replace_server(const std::string& address, + const struct AddressParams *params); + void get_main_address(std::vector& addr_list); + protected: struct rb_root group_map; EndpointGroup *default_group; @@ -170,17 +96,17 @@ private: 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 EndpointAddress *addr) { this->nalives--; - addr->group->nalives--; + UPSAddrParams *params = static_cast(addr->params); + params->group->nalives--; } - virtual bool select(const ParsedURI& uri, EndpointAddress **addr); - protected: virtual void add_server_locked(EndpointAddress *addr); virtual int remove_server_locked(const std::string& address); @@ -190,10 +116,11 @@ protected: 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)); } }; diff --git a/src/factory/WFDNSResolver.cc b/src/nameservice/WFDNSResolver.cc similarity index 100% rename from src/factory/WFDNSResolver.cc rename to src/nameservice/WFDNSResolver.cc diff --git a/src/factory/WFDNSResolver.h b/src/nameservice/WFDNSResolver.h similarity index 100% rename from src/factory/WFDNSResolver.h rename to src/nameservice/WFDNSResolver.h diff --git a/src/factory/WFNameService.cc b/src/nameservice/WFNameService.cc similarity index 100% rename from src/factory/WFNameService.cc rename to src/nameservice/WFNameService.cc diff --git a/src/factory/WFNameService.h b/src/nameservice/WFNameService.h similarity index 100% rename from src/factory/WFNameService.h rename to src/nameservice/WFNameService.h diff --git a/test/upstream_unittest.cc b/test/upstream_unittest.cc index b57fad25..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) @@ -105,8 +106,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 +127,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 +134,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, @@ -156,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( @@ -221,6 +231,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";