From f19c7f3edeb2bc4e566f7055dbcee96de4e30a9a Mon Sep 17 00:00:00 2001 From: XieHan Date: Tue, 2 Feb 2021 23:43:07 +0800 Subject: [PATCH] migrate WFNameService to windows branch --- CMakeLists_Headers.txt | 3 + src/factory/CMakeLists.txt | 2 + src/factory/MySQLTaskImpl.cc | 4 +- src/factory/WFTaskFactory.cc | 202 ------ src/factory/WFTaskFactory.inl | 471 ++++--------- src/manager/CMakeLists.txt | 1 + src/manager/DNSCache.cc | 10 +- src/manager/DNSCache.h | 188 +++--- src/manager/RouteManager.h | 2 +- src/manager/UpstreamManager.cc | 1127 ++++---------------------------- src/manager/UpstreamManager.h | 58 +- src/manager/WFGlobal.cc | 102 ++- src/manager/WFGlobal.h | 14 +- 13 files changed, 422 insertions(+), 1762 deletions(-) diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index 24d75827..55861061 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -78,6 +78,7 @@ 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 @@ -95,5 +96,7 @@ set(INCLUDE_HEADERS src/factory/WFAlgoTaskFactory.inl src/factory/Workflow.h src/factory/WFOperator.h + src/factory/WFNameService.h + src/factory/WFDNSResolver.h ) diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index aac274e2..e7a79722 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -8,6 +8,8 @@ set(SRC MySQLTaskImpl.cc WFTaskFactory.cc Workflow.cc + WFNameService.cc + WFDNSResolver.cc ) add_library(${PROJECT_NAME} OBJECT ${SRC}) diff --git a/src/factory/MySQLTaskImpl.cc b/src/factory/MySQLTaskImpl.cc index bdbdfeec..1264685b 100644 --- a/src/factory/MySQLTaskImpl.cc +++ b/src/factory/MySQLTaskImpl.cc @@ -471,13 +471,13 @@ bool ComplexMySQLTask::init_success() transaction_state_ = TRANSACTION_OUT; this->WFComplexClientTask::set_info(std::string("?maxconn=1&") + info + "|txn:" + transaction); - this->first_addr_only_ = true; + this->fixed_addr_ = true; } else { transaction_state_ = NO_TRANSACTION; this->WFComplexClientTask::set_info(info); - this->first_addr_only_ = false; + this->fixed_addr_ = false; } delete []info; diff --git a/src/factory/WFTaskFactory.cc b/src/factory/WFTaskFactory.cc index cd896030..f574ce7f 100644 --- a/src/factory/WFTaskFactory.cc +++ b/src/factory/WFTaskFactory.cc @@ -23,7 +23,6 @@ #include "PlatformSocket.h" #include "list.h" #include "rbtree.h" -#include "DNSRoutine.h" #include "WFGlobal.h" #include "WFTaskError.h" #include "WFTaskFactory.h" @@ -471,204 +470,3 @@ WFFileSyncTask *WFTaskFactory::create_fdsync_task(int fd, #endif -/********RouterTask*************/ - -void WFRouterTask::dispatch() -{ - insert_dns_ = true; - if (dns_cache_level_ != DNS_CACHE_LEVEL_0) - { - auto *dns_cache = WFGlobal::get_dns_cache(); - const DNSHandle *addr_handle = NULL; - - switch (dns_cache_level_) - { - case DNS_CACHE_LEVEL_1: - addr_handle = dns_cache->get_confident(host_, port_); - break; - - case DNS_CACHE_LEVEL_2: - addr_handle = dns_cache->get_ttl(host_, port_); - break; - - case DNS_CACHE_LEVEL_3: - addr_handle = dns_cache->get(host_, port_); - break; - - default: - break; - } - - if (addr_handle) - { - if (addr_handle->value.addrinfo) - { - auto *route_manager = WFGlobal::get_route_manager(); - struct addrinfo *addrinfo = addr_handle->value.addrinfo; - struct addrinfo first; - - if (first_addr_only_ && addrinfo->ai_next) - { - first = *addrinfo; - first.ai_next = NULL; - addrinfo = &first; - } - - if (route_manager->get(type_, addrinfo, - info_, &endpoint_params_, - route_result_) < 0) - { - this->state = WFT_STATE_SYS_ERROR; - this->error = errno; - } - else if (!route_result_.request_object) - { - //should not happen - this->state = WFT_STATE_SYS_ERROR; - this->error = EAGAIN; - } - else - this->state = WFT_STATE_SUCCESS; - - insert_dns_ = false; - } - - dns_cache->release(addr_handle); - } - } - - if (insert_dns_ && !host_.empty()) - { - char front = host_.front(); - char back = host_.back(); - struct in6_addr addr; - int ret; - - if (host_.find(':') != std::string::npos) - ret = inet_pton(AF_INET6, host_.c_str(), &addr); - else if (isdigit(back) && isdigit(front)) - ret = inet_pton(AF_INET, host_.c_str(), &addr); -#if defined(AF_UNIX) && !defined(_WIN32) - else if (front == '/') - ret = 1; -#endif - else - ret = 0; - - if (ret == 1) - { - DNSInput dns_in; - DNSOutput dns_out; - - dns_in.reset(host_, port_); - DNSRoutine::run(&dns_in, &dns_out); - dns_callback_internal(&dns_out, (unsigned int)-1, (unsigned int)-1); - insert_dns_ = false; - } - } - - if (insert_dns_) - { - auto&& cb = std::bind(&WFRouterTask::dns_callback, - this, - std::placeholders::_1); - WFDNSTask *dns_task = WFTaskFactory::create_dns_task(host_, port_, - std::move(cb)); - - series_of(this)->push_front(dns_task); - } - - this->subtask_done(); -} - -SubTask* WFRouterTask::done() -{ - SeriesWork *series = series_of(this); - - if (!insert_dns_) - { - if (callback_) - callback_(this); - - delete this; - } - - return series->pop(); -} - -void WFRouterTask::dns_callback_internal(DNSOutput *dns_out, - unsigned int ttl_default, - unsigned int ttl_min) -{ - int dns_error = dns_out->get_error(); - - if (dns_error) - { -#ifdef EAI_SYSTEM - if (dns_error == EAI_SYSTEM) - { - this->state = WFT_STATE_SYS_ERROR; - this->error = errno; - } - else -#endif - { - this->state = WFT_STATE_DNS_ERROR; - this->error = dns_error; - } - } - else - { - struct addrinfo *addrinfo = dns_out->move_addrinfo(); - - if (addrinfo) - { - auto *route_manager = WFGlobal::get_route_manager(); - auto *dns_cache = WFGlobal::get_dns_cache(); - const DNSHandle *addr_handle = dns_cache->put(host_, port_, - addrinfo, - (unsigned int)ttl_default, - (unsigned int)ttl_min); - - if (route_manager->get(type_, addrinfo, info_, &endpoint_params_, - route_result_) < 0) - { - this->state = WFT_STATE_SYS_ERROR; - this->error = errno; - } - else if (!route_result_.request_object) - { - //should not happen - this->state = WFT_STATE_SYS_ERROR; - this->error = EAGAIN; - } - else - this->state = WFT_STATE_SUCCESS; - - dns_cache->release(addr_handle); - } - else - { - //system promise addrinfo not null, here should not happen - this->state = WFT_STATE_SYS_ERROR; - this->error = EINVAL; - } - } -} - -void WFRouterTask::dns_callback(WFDNSTask *dns_task) -{ - if (dns_task->get_state() == WFT_STATE_SUCCESS) - dns_callback_internal(dns_task->get_output(), dns_ttl_default_, dns_ttl_min_); - else - { - this->state = dns_task->get_state(); - this->error = dns_task->get_error(); - } - - if (callback_) - callback_(this); - - delete this; -} - diff --git a/src/factory/WFTaskFactory.inl b/src/factory/WFTaskFactory.inl index 778a6e7b..4d75ec8a 100644 --- a/src/factory/WFTaskFactory.inl +++ b/src/factory/WFTaskFactory.inl @@ -13,8 +13,8 @@ See the License for the specific language governing permissions and limitations under the License. - Authors: Wu Jiaxu (wujiaxu@sogou-inc.com) - Xie Han (xiehan@sogou-inc.com) + Authors: Xie Han (xiehan@sogou-inc.com) + Wu Jiaxu (wujiaxu@sogou-inc.com) Li Yingxin (liyingxin@sogou-inc.com) */ @@ -30,11 +30,11 @@ #include "WFGlobal.h" #include "Workflow.h" #include "WFTask.h" -#include "UpstreamManager.h" #include "RouteManager.h" #include "URIParser.h" #include "WFTaskError.h" #include "EndpointParams.h" +#include "WFNameService.h" class __WFTimerTask : public WFTimerTask { @@ -131,102 +131,6 @@ WFTaskFactory::create_dynamic_task(dynamic_create_t create) return new __WFDynamicTask(std::move(create)); } -/**********WFComplexClientTask**********/ - -// If you design Derived WFComplexClientTask, You have two choices: -// 1) First choice will upstream by uri, then dns/dns-cache -// 2) Second choice will directly communicate without upstream/dns/dns-cache - -// 1) First choice: -// step 1. Child-Constructor call Father-Constructor to new WFComplexClientTask -// step 2. call init(uri) -// step 3. call set_type(type) -// step 4. call set_info(info) or do nothing with info - -// 2) Second choice: -// step 1. Child-Constructor call Father-Constructor to new WFComplexClientTask -// step 2. call init(type, addr, addrlen, info) - -// Some optional APIs for you to implement: -// [WFComplexTask] -// [ChildrenComplexTask] -// 1. init() -// init_succ() or init_failed(); // default: return true; -// 2. dispatch(); -// check_request(); // default: return true; -// route(); // default:DNS; goto 1; -// 3. message_out(); -// 4. message_in(); -// 5. keep_alive_timeout(); -// 6. done(); -// finish_once(); // default: return true; means this is user request. -// // If redirect or retry: goto 1; - -/* -DNS_CACHE_LEVEL_0 -> NO cache -DNS_CACHE_LEVEL_1 -> TTL MIN -DNS_CACHE_LEVEL_2 -> TTL [DEFAULT] -DNS_CACHE_LEVEL_3 -> Forever -*/ - -#define DNS_CACHE_LEVEL_0 0 -#define DNS_CACHE_LEVEL_1 1 -#define DNS_CACHE_LEVEL_2 2 -#define DNS_CACHE_LEVEL_3 3 - -class WFRouterTask : public WFGenericTask -{ -private: - using router_callback_t = std::function; - using WFDNSTask = WFThreadTask; - -public: - RouteManager::RouteResult route_result_; - - WFRouterTask(TransportType type, - const std::string& host, - unsigned short port, - const std::string& info, - int dns_cache_level, - unsigned int dns_ttl_default, - unsigned int dns_ttl_min, - const struct EndpointParams *endpoint_params, - bool first_addr_only, - router_callback_t&& callback) : - type_(type), - host_(host), - port_(port), - info_(info), - dns_cache_level_(dns_cache_level), - dns_ttl_default_(dns_ttl_default), - dns_ttl_min_(dns_ttl_min), - endpoint_params_(*endpoint_params), - first_addr_only_(first_addr_only), - callback_(std::move(callback)) - {} - -private: - virtual void dispatch(); - virtual SubTask *done(); - void dns_callback(WFDNSTask *dns_task); - void dns_callback_internal(DNSOutput *dns_task, - unsigned int ttl_default, - unsigned int ttl_min); - -private: - TransportType type_; - std::string host_; - unsigned short port_; - std::string info_; - int dns_cache_level_; - unsigned int dns_ttl_default_; - unsigned int dns_ttl_min_; - struct EndpointParams endpoint_params_; - bool first_addr_only_; - bool insert_dns_; - router_callback_t callback_; -}; - template class WFComplexClientTask : public WFClientTask { @@ -234,24 +138,24 @@ protected: using task_callback_t = std::function *)>; public: - WFComplexClientTask(int retry_max, task_callback_t&& callback): - WFClientTask(NULL, WFGlobal::get_scheduler(), - std::move(callback)), - retry_max_(retry_max), - first_addr_only_(false), - router_task_(NULL), - type_(TT_TCP), - retry_times_(0), - is_retry_(false), - redirect_(false) - {} + WFComplexClientTask(int retry_max, task_callback_t&& cb): + WFClientTask(NULL, WFGlobal::get_scheduler(), std::move(cb)) + { + type_ = TT_TCP; + fixed_addr_ = false; + retry_max_ = retry_max; + retry_times_ = 0; + redirect_ = false; + ns_policy_ = NULL; + router_task_ = NULL; + } protected: // new api for children virtual bool init_success() { return true; } virtual void init_failed() {} virtual bool check_request() { return true; } - virtual SubTask *route(); + virtual WFRouterTask *route(); virtual bool finish_once() { return true; } public: @@ -272,16 +176,12 @@ public: void init(const ParsedURI& uri) { - is_sockaddr_ = false; - init_state_ = 0; uri_ = uri; init_with_uri(); } void init(ParsedURI&& uri) { - is_sockaddr_ = false; - init_state_ = 0; uri_ = std::move(uri); init_with_uri(); } @@ -309,25 +209,6 @@ public: } protected: - void set_redirect() - { - redirect_ = true; - retry_times_ = 0; - } - - void set_retry(const ParsedURI& uri) - { - redirect_ = true; - init(uri); - retry_times_++; - } - - void set_retry() - { - redirect_ = true; - retry_times_++; - } - virtual void dispatch(); virtual SubTask *done(); @@ -347,13 +228,19 @@ protected: TransportType get_transport_type() const { return type_; } +protected: + TransportType type_; ParsedURI uri_; - - int retry_max_; - bool is_sockaddr_; - bool first_addr_only_; + std::string info_; + bool fixed_addr_; + bool redirect_; CTX ctx_; - SubTask *router_task_; + int retry_max_; + int retry_times_; + WFNSPolicy *ns_policy_; + WFRouterTask *router_task_; + RouteManager::RouteResult route_result_; + void *cookie_; public: CTX *get_mutable_ctx() { return &ctx_; } @@ -361,21 +248,8 @@ public: private: void init_with_uri(); bool set_port(); - void router_callback(SubTask *task); // default: DNS + void router_callback(WFRouterTask *task); void switch_callback(WFTimerTask *task); - - RouteManager::RouteResult route_result_; - UpstreamManager::UpstreamResult upstream_result_; - - TransportType type_; - std::string info_; - - int retry_times_; - - /* state 0: uninited or failed; 1: inited but not checked; 2: checked. */ - char init_state_; - bool is_retry_; - bool redirect_; }; template @@ -384,42 +258,34 @@ void WFComplexClientTask::init(TransportType type, socklen_t addrlen, const std::string& info) { - is_sockaddr_ = true; - init_state_ = 0; - type_ = type; - info_.assign(info); - struct addrinfo addrinfo; - const auto *params = &WFGlobal::get_global_settings()->endpoint_params; + if (redirect_) + { + ns_policy_ = NULL; + route_result_.clear(); + this->state = WFT_STATE_UNDEFINED; + this->error = 0; + this->timeout_reason = TOR_NOT_TIMEOUT; + } - addrinfo.ai_addrlen = addrlen; - addrinfo.ai_addr = (struct sockaddr *)addr; - addrinfo.ai_canonname = NULL; - addrinfo.ai_next = NULL; - addrinfo.ai_flags = 0; + const auto *params = &WFGlobal::get_global_settings()->endpoint_params; + struct addrinfo addrinfo = { }; addrinfo.ai_family = addr->sa_family; addrinfo.ai_socktype = SOCK_STREAM; - addrinfo.ai_protocol = 0; + addrinfo.ai_addr = (struct sockaddr *)addr; + addrinfo.ai_addrlen = addrlen; + type_ = type; + info_.assign(info); if (WFGlobal::get_route_manager()->get(type, &addrinfo, info_, params, - route_result_) < 0) + route_result_) < 0) { this->state = WFT_STATE_SYS_ERROR; this->error = errno; } - else if (!route_result_.request_object) - { - //should not happen - this->state = WFT_STATE_SYS_ERROR; - this->error = EAGAIN; - } - else - { - init_state_ = this->init_success() ? 1 : 0; + else if (this->init_success()) return; - } this->init_failed(); - return; } template @@ -469,170 +335,97 @@ bool WFComplexClientTask::set_port() template void WFComplexClientTask::init_with_uri() { - route_result_.clear(); - if (uri_.state == URI_STATE_SUCCESS && this->set_port()) + if (redirect_) { - int ret = UpstreamManager::choose(uri_, upstream_result_); + ns_policy_ = NULL; + route_result_.clear(); + this->state = WFT_STATE_UNDEFINED; + this->error = 0; + this->timeout_reason = TOR_NOT_TIMEOUT; + } - if (ret < 0) + if (uri_.state == URI_STATE_SUCCESS) + { + if (this->set_port()) { - this->state = WFT_STATE_SYS_ERROR; - this->error = errno; - } - else if (upstream_result_.state == UPSTREAM_ALL_DOWN) - { - this->state = WFT_STATE_TASK_ERROR; - this->error = WFT_ERR_UPSTREAM_UNAVAILABLE; - } - else - { - init_state_ = this->init_success() ? 1 : 0; - return; + if (this->init_success()) + return; } } + else if (uri_.state == URI_STATE_ERROR) + { + this->state = WFT_STATE_SYS_ERROR; + this->error = uri_.error; + } else { - if (uri_.state == URI_STATE_ERROR) - { - this->state = WFT_STATE_SYS_ERROR; - this->error = uri_.error; - } - else - { - this->state = WFT_STATE_TASK_ERROR; - this->error = WFT_ERR_URI_PARSE_FAILED; - } + this->state = WFT_STATE_TASK_ERROR; + this->error = WFT_ERR_URI_PARSE_FAILED; } this->init_failed(); - return; } template -SubTask *WFComplexClientTask::route() +WFRouterTask *WFComplexClientTask::route() { - unsigned int dns_ttl_default; - unsigned int dns_ttl_min; - const struct EndpointParams *endpoint_params; - - int dns_cache_level = (is_retry_ ? DNS_CACHE_LEVEL_1 - : DNS_CACHE_LEVEL_2); + WFNameService *ns = WFGlobal::get_name_service(); auto&& cb = std::bind(&WFComplexClientTask::router_callback, this, std::placeholders::_1); - - is_retry_ = false;//route means refresh DNS cache level - if (upstream_result_.state == UPSTREAM_SUCCESS) - { - const auto *params = upstream_result_.address_params; - - dns_ttl_default = params->dns_ttl_default; - dns_ttl_min = params->dns_ttl_min; - endpoint_params = ¶ms->endpoint_params; - } - else - { - const auto *params = WFGlobal::get_global_settings(); - - dns_ttl_default = params->dns_ttl_default; - dns_ttl_min = params->dns_ttl_min; - endpoint_params = ¶ms->endpoint_params; - } - - return new WFRouterTask(type_, uri_.host ? uri_.host : "", - uri_.port ? atoi(uri_.port) : 0, info_, - dns_cache_level, dns_ttl_default, dns_ttl_min, - endpoint_params, first_addr_only_, std::move(cb)); + struct WFNSParams params = { + .type = type_, + .uri = uri_, + .info = info_.c_str(), + .fixed_addr = fixed_addr_, + .retry_times = retry_times_, + }; + ns_policy_ = ns->get_policy(uri_.host ? uri_.host : ""); + return ns_policy_->create_router_task(¶ms, cb); } -/* - * router callback`s obligation: - * if success: - * 1. set route_result_ or call this->init() - * 2. series->push_front(ORIGIN_TASK) - * if failed: - * 1. this->finish_once() is optional; - * 2. this->callback() is necessary; -*/ template -void WFComplexClientTask::router_callback(SubTask *task) +void WFComplexClientTask::router_callback(WFRouterTask *task) { - WFRouterTask *router_task = static_cast(task); - int state = router_task->get_state(); - - if (state == WFT_STATE_SUCCESS) - route_result_ = router_task->route_result_; - else + this->state = task->get_state(); + if (this->state == WFT_STATE_SUCCESS) { - this->state = state; - this->error = router_task->get_error(); + route_result_ = std::move(*task->get_result()); + cookie_ = task->get_cookie(); } - - if (route_result_.request_object) - series_of(this)->push_front(this); - else + else if (this->state == WFT_STATE_UNDEFINED) { - UpstreamManager::notify_unavailable(upstream_result_.cookie); - - if (this->callback) - this->callback(this); - - if (redirect_) - { - init_state_ = this->init_success() ? 1 : 0; - redirect_ = false; - this->state = WFT_STATE_UNDEFINED; - this->error = 0; - this->timeout_reason = TOR_NOT_TIMEOUT; - series_of(this)->push_front(this); - } - else - delete this; + /* should not happend */ + this->state = WFT_STATE_SYS_ERROR; + this->error = ENOSYS; } + else + this->error = task->get_error(); } template void WFComplexClientTask::dispatch() { - // 1. children check_request() - if (init_state_ == 1) - init_state_ = this->check_request() ? 2 : 0; - - if (init_state_) + switch (this->state) { - if (route_result_.request_object) + case WFT_STATE_UNDEFINED: + if (this->check_request()) { - // 2. origin task dispatch() - this->set_request_object(route_result_.request_object); - this->WFClientTask::dispatch(); - return; + if (this->route_result_.request_object) + { + case WFT_STATE_SUCCESS: + this->set_request_object(route_result_.request_object); + this->WFClientTask::dispatch(); + return; + } + + router_task_ = this->route(); + series_of(this)->push_front(this); + series_of(this)->push_front(router_task_); } - if (is_sockaddr_ || uri_.state == URI_STATE_SUCCESS) - { - // 3. DNS route() or children route() - router_task_ = this->route(); - if (router_task_) - series_of(this)->push_front(router_task_); - else - { - this->state = WFT_STATE_TASK_ERROR; - this->error = WFT_ERR_ROUTE_FAILED; - } - } - else - { - if (uri_.state == URI_STATE_ERROR) - { - this->state = WFT_STATE_SYS_ERROR; - this->error = uri_.error; - } - else - { - this->state = WFT_STATE_TASK_ERROR; - this->error = WFT_ERR_URI_PARSE_FAILED; - } - } + default: + break; } this->subtask_done(); @@ -649,21 +442,15 @@ void WFComplexClientTask::switch_callback(WFTimerTask *) this->error = -this->error; } - // 4. children finish before user callback if (this->callback) this->callback(this); } if (redirect_) { - init_state_ = this->init_success() ? 1 : 0; redirect_ = false; clear_resp(); - this->target = NULL; - this->timeout_reason = TOR_NOT_TIMEOUT; - this->state = WFT_STATE_UNDEFINED; - this->error = 0; series_of(this)->push_front(this); } else @@ -675,48 +462,43 @@ SubTask *WFComplexClientTask::done() { SeriesWork *series = series_of(this); - // 1. routing if (router_task_) { router_task_ = NULL; return series->pop(); } - if (init_state_) - { - // 2. children can set_redirect() here - bool is_user_request = this->finish_once(); - // 3. complex task success - if (this->state == WFT_STATE_SUCCESS) - { - RouteManager::notify_available(route_result_.cookie, this->target); - UpstreamManager::notify_available(upstream_result_.cookie); - upstream_result_.clear(); - // 4. children message out sth. else - if (!is_user_request) - return this; - } - else if (this->state == WFT_STATE_SYS_ERROR) - { - RouteManager::notify_unavailable(route_result_.cookie, this->target); - UpstreamManager::notify_unavailable(upstream_result_.cookie); - // 5. complex task failed: retry - if (retry_times_ < retry_max_) - { - if (is_sockaddr_) - set_retry(); - else - set_retry(uri_); + bool is_user_request = this->finish_once(); - is_retry_ = true; // will influence next round dns cache time - } + if (ns_policy_ && route_result_.request_object) + { + if (this->state == WFT_STATE_SYS_ERROR) + ns_policy_->failed(&route_result_, cookie_, this->target); + else + ns_policy_->success(&route_result_, cookie_, this->target); + } + + if (this->state == WFT_STATE_SUCCESS) + { + if (!is_user_request) + return this; + } + else if (this->state == WFT_STATE_SYS_ERROR) + { + if (retry_times_ < retry_max_) + { + redirect_ = true; + this->state = WFT_STATE_UNDEFINED; + this->error = 0; + this->timeout_reason = 0; + retry_times_++; } } /* - * When target is NULL, it's very likely that we are still in the - * 'dispatch' thread. Running a timer will switch callback function - * to a handler thread, and this can prevent stack overflow. + * When target is NULL, it's very likely that we are in the caller's + * thread or DNS thread (dns failed). Running a timer will switch callback + * function to a handler thread, and this can prevent stack overflow. */ if (!this->target) { @@ -724,7 +506,6 @@ SubTask *WFComplexClientTask::done() this, std::placeholders::_1); WFTimerTask *timer = WFTaskFactory::create_timer_task(0, std::move(cb)); - series->push_front(timer); } else diff --git a/src/manager/CMakeLists.txt b/src/manager/CMakeLists.txt index ee916512..6a778613 100644 --- a/src/manager/CMakeLists.txt +++ b/src/manager/CMakeLists.txt @@ -6,6 +6,7 @@ set(SRC UpstreamManager.cc RouteManager.cc WFGlobal.cc + UpstreamPolicies.cc ) add_library(${PROJECT_NAME} OBJECT ${SRC}) diff --git a/src/manager/DNSCache.cc b/src/manager/DNSCache.cc index ba185602..fec5da4c 100644 --- a/src/manager/DNSCache.cc +++ b/src/manager/DNSCache.cc @@ -25,7 +25,7 @@ #define CONFIDENT_INC 10 #define TTL_INC 10 -const DNSHandle *DNSCache::get_inner(const HostPort& host_port, int type) +const DNSCache::DNSHandle *DNSCache::get_inner(const HostPort& host_port, int type) { const DNSHandle *handle = cache_pool_.get(host_port); @@ -73,10 +73,10 @@ const DNSHandle *DNSCache::get_inner(const HostPort& host_port, int type) return handle; } -const DNSHandle *DNSCache::put(const HostPort& host_port, - struct addrinfo *addrinfo, - unsigned int dns_ttl_default, - unsigned int dns_ttl_min) +const DNSCache::DNSHandle *DNSCache::put(const HostPort& host_port, + struct addrinfo *addrinfo, + unsigned int dns_ttl_default, + unsigned int dns_ttl_min) { int64_t expire_time; int64_t confident_time; diff --git a/src/manager/DNSCache.h b/src/manager/DNSCache.h index 776acd75..1b9a29ed 100644 --- a/src/manager/DNSCache.h +++ b/src/manager/DNSCache.h @@ -45,36 +45,75 @@ public: } }; -typedef std::pair HostPort; -typedef LRUHandle DNSHandle; - // RAII: NO. Release handle by user // Thread safety: YES // MUST call release when handle no longer used class DNSCache { +public: + using HostPort = std::pair; + using DNSHandle = LRUHandle; + public: // release handle by get/put - void release(DNSHandle *handle); - void release(const DNSHandle *handle); + void release(DNSHandle *handle) + { + cache_pool_.release(handle); + } + + void release(const DNSHandle *handle) + { + cache_pool_.release(handle); + } // get handler // Need call release when handle no longer needed //Handle *get(const KEY &key); - const DNSHandle *get(const HostPort& host_port); - const DNSHandle *get(const std::string& host, unsigned short port); - const DNSHandle *get(const char *host, unsigned short port); + const DNSHandle *get(const HostPort& host_port) + { + return cache_pool_.get(host_port); + } - const DNSHandle *get_ttl(const HostPort& host_port); - const DNSHandle *get_ttl(const std::string& host, unsigned short port); - const DNSHandle *get_ttl(const char *host, unsigned short port); + const DNSHandle *get(const std::string& host, unsigned short port) + { + return get(HostPort(host, port)); + } - const DNSHandle *get_confident(const HostPort& host_port); - const DNSHandle *get_confident(const std::string& host, unsigned short port); - const DNSHandle *get_confident(const char *host, unsigned short port); + const DNSHandle *get(const char *host, unsigned short port) + { + return get(std::string(host), port); + } + + const DNSHandle *get_ttl(const HostPort& host_port) + { + return get_inner(host_port, GET_TYPE_TTL); + } + + const DNSHandle *get_ttl(const std::string& host, unsigned short port) + { + return get_ttl(HostPort(host, port)); + } + + const DNSHandle *get_ttl(const char *host, unsigned short port) + { + return get_ttl(std::string(host), port); + } + + const DNSHandle *get_confident(const HostPort& host_port) + { + return get_inner(host_port, GET_TYPE_CONFIDENT); + } + + const DNSHandle *get_confident(const std::string& host, unsigned short port) + { + return get_confident(HostPort(host, port)); + } + + const DNSHandle *get_confident(const char *host, unsigned short port) + { + return get_confident(std::string(host), port); + } - // put copy - // Need call release when handle no longer needed const DNSHandle *put(const HostPort& host_port, struct addrinfo *addrinfo, unsigned int dns_ttl_default, @@ -84,18 +123,35 @@ public: unsigned short port, struct addrinfo *addrinfo, unsigned int dns_ttl_default, - unsigned int dns_ttl_min); + unsigned int dns_ttl_min) + { + return put(HostPort(host, port), addrinfo, dns_ttl_default, dns_ttl_min); + } const DNSHandle *put(const char *host, unsigned short port, struct addrinfo *addrinfo, unsigned int dns_ttl_default, - unsigned int dns_ttl_min); + unsigned int dns_ttl_min) + { + return put(std::string(host), port, addrinfo, dns_ttl_default, dns_ttl_min); + } // delete from cache, deleter delay called when all inuse-handle release. - void del(const HostPort& key); - void del(const std::string& host, unsigned short port); - void del(const char *host, unsigned short port); + void del(const HostPort& key) + { + cache_pool_.del(key); + } + + void del(const std::string& host, unsigned short port) + { + del(HostPort(host, port)); + } + + void del(const char *host, unsigned short port) + { + del(std::string(host), port); + } private: const DNSHandle *get_inner(const HostPort& host_port, int type); @@ -104,95 +160,5 @@ private: LRUCache cache_pool_; }; -//////////////////// - -inline void DNSCache::release(DNSHandle *handle) -{ - cache_pool_.release(handle); -} - -inline void DNSCache::release(const DNSHandle *handle) -{ - cache_pool_.release(handle); -} - -inline const DNSHandle *DNSCache::get(const HostPort& host_port) -{ - return cache_pool_.get(host_port); -} - -inline const DNSHandle *DNSCache::get(const std::string& host, unsigned short port) -{ - return get(HostPort(host, port)); -} - -inline const DNSHandle *DNSCache::get(const char *host, unsigned short port) -{ - return get(std::string(host), port); -} - -inline const DNSHandle *DNSCache::get_ttl(const HostPort& host_port) -{ - return get_inner(host_port, GET_TYPE_TTL); -} - -inline const DNSHandle *DNSCache::get_ttl(const std::string& host, unsigned short port) -{ - return get_ttl(HostPort(host, port)); -} - -inline const DNSHandle *DNSCache::get_ttl(const char *host, unsigned short port) -{ - return get_ttl(std::string(host), port); -} - -inline const DNSHandle *DNSCache::get_confident(const HostPort& host_port) -{ - return get_inner(host_port, GET_TYPE_CONFIDENT); -} - -inline const DNSHandle *DNSCache::get_confident(const std::string& host, unsigned short port) -{ - return get_confident(HostPort(host, port)); -} - -inline const DNSHandle *DNSCache::get_confident(const char *host, unsigned short port) -{ - return get_confident(std::string(host), port); -} - -inline const DNSHandle *DNSCache::put(const std::string& host, - unsigned short port, - struct addrinfo *addrinfo, - unsigned int dns_ttl_default, - unsigned int dns_ttl_min) -{ - return put(HostPort(host, port), addrinfo, dns_ttl_default, dns_ttl_min); -} - -inline const DNSHandle *DNSCache::put(const char *host, - unsigned short port, - struct addrinfo *addrinfo, - unsigned int dns_ttl_default, - unsigned int dns_ttl_min) -{ - return put(std::string(host), port, addrinfo, dns_ttl_default, dns_ttl_min); -} - -inline void DNSCache::del(const HostPort& key) -{ - cache_pool_.del(key); -} - -inline void DNSCache::del(const std::string& host, unsigned short port) -{ - del(HostPort(host, port)); -} - -inline void DNSCache::del(const char *host, unsigned short port) -{ - del(std::string(host), port); -} - #endif diff --git a/src/manager/RouteManager.h b/src/manager/RouteManager.h index cdcbe2bf..120009a1 100644 --- a/src/manager/RouteManager.h +++ b/src/manager/RouteManager.h @@ -37,7 +37,7 @@ public: CommSchedObject *request_object; public: - // RouteResult(): cookie(NULL), request_object(NULL) { } + RouteResult(): cookie(NULL), request_object(NULL) { } void clear() { cookie = NULL; request_object = NULL; } }; diff --git a/src/manager/UpstreamManager.cc b/src/manager/UpstreamManager.cc index a0c8ce75..6c6ae7e6 100644 --- a/src/manager/UpstreamManager.cc +++ b/src/manager/UpstreamManager.cc @@ -16,713 +16,10 @@ Authors: Wu Jiaxu (wujiaxu@sogou-inc.com) */ -#include -#include -#include -#include -#include #include -#include -#include -#include -#include -#include "list.h" -#include "rbtree.h" -#include "URIParser.h" -#include "StringUtil.h" -#include "RWLock.h" -#include "EndpointParams.h" #include "UpstreamManager.h" - -#define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() -#define MTTR_SECOND 30 -#define VIRTUAL_GROUP_SIZE 16 - -class UpstreamAddress; -class UpstreamGroup; -class Upstream; - -class UpstreamAddress -{ -public: - UpstreamGroup *group; - AddressParams params; - struct list_head list; - int64_t broken_timeout; - unsigned int consistent_hash[VIRTUAL_GROUP_SIZE]; - std::string address; - std::string host; - std::string port; - std::atomic fail_count; - unsigned short port_value; - -public: - UpstreamAddress(const std::string& address, - const struct AddressParams *address_params); -}; - -class UpstreamGroup -{ -public: - Upstream *upstream; - struct rb_node rb; - std::mutex mutex; - std::vector mains; - std::vector backups; - struct list_head breaker_list; - std::atomic nbreak; - std::atomic nalive; - int weight; - int group_id; - -public: - UpstreamGroup(int group_id, Upstream *us): - upstream(us), - nbreak(0), - nalive(0), - weight(0), - group_id(group_id) - { - INIT_LIST_HEAD(&this->breaker_list); - } - - const UpstreamAddress *get_one(); - const UpstreamAddress *get_one_backup(); -}; - -class Upstream -{ -public: - Upstream(); - ~Upstream(); - - int add(UpstreamAddress *ua); - int del(const std::string& address); - void disable_server(const std::string& address); - void enable_server(const std::string& address); - const UpstreamAddress *get(const ParsedURI& uri); - int set_select_callback(upstream_route_t&& select_callback); - int set_consistent_mode(upstream_route_t&& consistent_callback); - int set_attr(bool try_another, upstream_route_t rehash_callback); - void check_one_breaker(UpstreamGroup *group, int64_t cur_time); - void check_all_breaker(); - void get_all_main(std::vector& addr_list); - - static void notify_unavailable(UpstreamAddress *ua); - static void notify_available(UpstreamAddress *ua); - -protected: - RWLock rwlock_; - int total_weight_; - int available_weight_; - std::vector mains_; - std::unordered_map> server_map_; - struct rb_root group_map_; - upstream_route_t select_callback_; - upstream_route_t consistent_callback_; - UpstreamGroup *default_group_; - bool try_another_; - bool is_consistent_; - -private: - void lose_one_server(UpstreamGroup *group, const UpstreamAddress *ua); - void gain_one_server(UpstreamGroup *group, const UpstreamAddress *ua); - const UpstreamAddress *weighted_random_try_another() const; - const UpstreamAddress *consistent_hash_select(unsigned int hash) const; -}; - -const UpstreamAddress *UpstreamGroup::get_one() -{ - if (this->nalive == 0) - return NULL; - - std::lock_guard lock(this->mutex); - - std::random_shuffle(this->mains.begin(), this->mains.end()); - for (const auto *main : this->mains) - { - if (main->fail_count < main->params.max_fails) - return main; - } - - std::random_shuffle(this->backups.begin(), this->backups.end()); - for (const auto *backup : this->backups) - { - if (backup->fail_count < backup->params.max_fails) - return backup; - } - - return NULL; -} - -const UpstreamAddress *UpstreamGroup::get_one_backup() -{ - if (this->nalive == 0) - return NULL; - - std::lock_guard lock(this->mutex); - - std::random_shuffle(this->backups.begin(), this->backups.end()); - for (const auto *backup : this->backups) - { - if (backup->fail_count < backup->params.max_fails) - return backup; - } - - return NULL; -} - -static unsigned int __default_consistent_hash(const char *path, - const char *query, - const char *fragment) -{ - static std::hash std_hash; - std::string str(path); - - str += query; - str += fragment; - return (unsigned int)std_hash(str); -} - -UpstreamAddress::UpstreamAddress(const std::string& address, - const struct AddressParams *address_params) -{ - static std::hash std_hash; - std::vector arr = StringUtil::split(address, ':'); - - this->list.next = NULL; - this->fail_count = 0; - this->params = *address_params; - this->address = address; - for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++) - this->consistent_hash[i] = (unsigned int)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()); - } -} - -Upstream::Upstream(): - total_weight_(0), - available_weight_(0), - select_callback_(nullptr), - consistent_callback_(nullptr), - try_another_(false), - is_consistent_(false) -{ - group_map_.rb_node = NULL; - default_group_ = new UpstreamGroup(-1, this); - rb_link_node(&default_group_->rb, NULL, &group_map_.rb_node); - rb_insert_color(&default_group_->rb, &group_map_); -} - -Upstream::~Upstream() -{ - UpstreamGroup *group; - - while (group_map_.rb_node) - { - group = rb_entry(group_map_.rb_node, UpstreamGroup, rb); - rb_erase(group_map_.rb_node, &group_map_); - delete group; - } -} - -void Upstream::lose_one_server(UpstreamGroup *group, const UpstreamAddress *ua) -{ - if (--group->nalive == 0 && ua->params.group_id >= 0) - available_weight_ -= group->weight; - - if (ua->params.group_id < 0 && ua->params.server_type == 0) - available_weight_ -= ua->params.weight; -} - -void Upstream::gain_one_server(UpstreamGroup *group, const UpstreamAddress *ua) -{ - if (group->nalive++ == 0 && ua->params.group_id >= 0) - available_weight_ += group->weight; - - if (ua->params.group_id < 0 && ua->params.server_type == 0) - available_weight_ += ua->params.weight; -} - -int Upstream::add(UpstreamAddress *ua) -{ - int group_id = ua->params.group_id; - rb_node **p = &group_map_.rb_node; - rb_node *parent = NULL; - UpstreamGroup *group; - WriteLock lock(rwlock_); - - server_map_[ua->address].push_back(ua); - while (*p) - { - parent = *p; - group = rb_entry(*p, UpstreamGroup, rb); - - if (group_id < group->group_id) - p = &(*p)->rb_left; - else if (group_id > group->group_id) - p = &(*p)->rb_right; - else - break; - } - - if (*p == NULL) - { - group = new UpstreamGroup(group_id, this); - rb_link_node(&group->rb, parent, p); - rb_insert_color(&group->rb, &group_map_); - } - - if (ua->params.server_type == 0) - { - total_weight_ += ua->params.weight; - mains_.push_back(ua); - } - - group->mutex.lock(); - gain_one_server(group, ua); - ua->group = group; - if (ua->params.server_type == 0) - { - group->weight += ua->params.weight; - group->mains.push_back(ua); - } - else - group->backups.push_back(ua); - - group->mutex.unlock(); - - return 0; -} - -int Upstream::del(const std::string& address) -{ - WriteLock lock(rwlock_); - const auto map_it = server_map_.find(address); - - if (map_it != server_map_.cend()) - { - for (auto ua : map_it->second) - { - auto *group = ua->group; - std::vector *vec; - - if (ua->params.server_type == 0) - { - total_weight_ -= ua->params.weight; - vec = &group->mains; - } - else - vec = &group->backups; - - std::lock_guard lock(group->mutex); - - ua->group = NULL; - if (ua->fail_count < ua->params.max_fails) - lose_one_server(group, ua); - - if (ua->params.server_type == 0) - group->weight -= ua->params.weight; - - for (auto it = vec->begin(); it != vec->end(); ++it) - { - if (*it == ua) - { - vec->erase(it); - break; - } - } - } - - server_map_.erase(map_it); - } - - int n = (int)mains_.size(); - int new_n = 0; - - for (int i = 0; i < n; i++) - { - if (mains_[i]->address != address) - { - if (new_n != i) - mains_[new_n++] = mains_[i]; - else - new_n++; - } - } - - if (new_n < n) - { - mains_.resize(new_n); - return n - new_n; - } - - return 0; -} - -void Upstream::disable_server(const std::string& address) -{ - ReadLock lock(rwlock_); - const auto map_it = server_map_.find(address); - - if (map_it != server_map_.cend()) - { - for (auto ua : map_it->second) - { - auto *group = ua->group; - - if (group) - { - std::lock_guard lock(group->mutex); - - ua->fail_count = ua->params.max_fails; - if (ua->group == group && !ua->list.next) - { - ua->broken_timeout = GET_CURRENT_SECOND + MTTR_SECOND; - list_add_tail(&ua->list, &group->breaker_list); - group->nbreak++; - group->upstream->lose_one_server(group, ua); - } - } - else - ua->fail_count = ua->params.max_fails; - } - } -} - -void Upstream::enable_server(const std::string& address) -{ - ReadLock lock(rwlock_); - const auto map_it = server_map_.find(address); - - if (map_it != server_map_.cend()) - { - for (auto ua : map_it->second) - UpstreamManager::notify_available(ua); - } -} - -void Upstream::get_all_main(std::vector& addr_list) -{ - ReadLock lock(rwlock_); - - for (const auto *main : mains_) - addr_list.push_back(main->address); -} - -static inline const UpstreamAddress *__check_get_strong(const UpstreamAddress *ua) -{ - if (ua->fail_count >= ua->params.max_fails) - { - if (ua->params.group_id < 0) - ua = NULL; - else - ua = ua->group->get_one(); - } - - return ua; -} - -static inline const UpstreamAddress *__check_get_weak(const UpstreamAddress *ua) -{ - if (ua && ua->fail_count >= ua->params.max_fails && ua->params.group_id >= 0) - { - const auto *ret = ua->group->get_one(); - - if (ret) - ua = ret; - } - - return ua; -} - -static inline bool __is_alive_or_group_alive(const UpstreamAddress *ua) -{ - return (ua->params.group_id >= 0 && ua->group->nalive > 0) - || (ua->params.group_id < 0 && ua->fail_count < ua->params.max_fails); -} - -const UpstreamAddress *Upstream::weighted_random_try_another() const -{ - int temp_weight = available_weight_; - - if (temp_weight == 0) - return NULL; - - const UpstreamAddress *ua = NULL; - int x = rand() % temp_weight; - int s = 0; - - for (const auto *main : mains_) - { - if (__is_alive_or_group_alive(main)) - { - ua = main; - s += main->params.weight; - if (s > x) - break; - } - } - - return __check_get_weak(ua); -} - -const UpstreamAddress *Upstream::consistent_hash_select(unsigned int hash) const -{ - const UpstreamAddress *ua = NULL; - unsigned int min_dis = (unsigned int)-1; - - for (const auto *main : mains_) - { - if (__is_alive_or_group_alive(main)) - { - for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++) - { - unsigned int dis = std::min - (hash - main->consistent_hash[i], - main->consistent_hash[i] - hash); - - if (dis < min_dis) - { - min_dis = dis; - ua = main; - } - } - } - } - - return __check_get_weak(ua); -} - -void Upstream::check_one_breaker(UpstreamGroup *group, int64_t cur_time) -{ - struct list_head *pos, *tmp; - UpstreamAddress *ua; - - if (group->nbreak == 0) - return; - - std::lock_guard lock(group->mutex); - - list_for_each_safe(pos, tmp, &group->breaker_list) - { - ua = list_entry(pos, UpstreamAddress, list); - if (cur_time >= ua->broken_timeout) - { - if (ua->fail_count >= ua->params.max_fails) - { - ua->fail_count = ua->params.max_fails - 1; - if (ua->group == group) - gain_one_server(group, ua); - } - - list_del(pos); - ua->list.next = NULL; - group->nbreak--; - } - } -} - -void Upstream::check_all_breaker() -{ - if (!group_map_.rb_node) - return; - - struct rb_node *cur = rb_first(&group_map_); - UpstreamGroup *group; - int64_t cur_time = GET_CURRENT_SECOND; - - while (cur) - { - group = rb_entry(cur, UpstreamGroup, rb); - check_one_breaker(group, cur_time); - cur = rb_next(cur); - } -} - -const UpstreamAddress *Upstream::get(const ParsedURI& uri) -{ - unsigned int idx; - unsigned int hash_value; - const UpstreamAddress *ua; - - if (is_consistent_)// consistent mode - { - if (consistent_callback_) - hash_value = consistent_callback_(uri.path ? uri.path : "", - uri.query ? uri.query : "", - uri.fragment ? uri.fragment : ""); - else - hash_value = __default_consistent_hash(uri.path ? uri.path : "", - uri.query ? uri.query : "", - uri.fragment ? uri.fragment : ""); - - ReadLock lock(rwlock_); - check_all_breaker(); - ua = consistent_hash_select(hash_value); - } - else - { - ReadLock lock(rwlock_); - unsigned int n = (unsigned int)mains_.size(); - - if (n == 0) - return NULL; - else if (n == 1) - idx = 0; - else if (select_callback_) - { - idx = select_callback_(uri.path ? uri.path : "", - uri.query ? uri.query : "", - uri.fragment ? uri.fragment : ""); - - if (idx >= n) - idx %= n; - } - else - { - int x = 0; - int s = 0; - int temp_weight = total_weight_; - - if (temp_weight > 0) - x = rand() % temp_weight; - - for (idx = 0; idx < n; idx++) - { - s += mains_[idx]->params.weight; - if (s > x) - break; - } - - if (idx == n) - idx = n - 1; - } - - ua = mains_[idx]; - if (ua->fail_count >= ua->params.max_fails) - { - check_all_breaker(); - ua = __check_get_strong(ua); - - if (!ua && try_another_) - { - if (!select_callback_)// weighted random mode - ua = weighted_random_try_another(); - else// manual mode - { - if (consistent_callback_) - hash_value = consistent_callback_(uri.path ? uri.path : "", - uri.query ? uri.query : "", - uri.fragment ? uri.fragment : ""); - else - hash_value = __default_consistent_hash(uri.path ? uri.path : "", - uri.query ? uri.query : "", - uri.fragment ? uri.fragment : ""); - - ua = consistent_hash_select(hash_value); - } - } - } - } - - if (!ua) - ua = default_group_->get_one_backup();//get one backup from group[-1] - - return ua; -} - -int Upstream::set_select_callback(upstream_route_t&& select_callback) -{ - WriteLock lock(rwlock_); - - select_callback_ = std::move(select_callback); - return 0; -} - -int Upstream::set_consistent_mode(upstream_route_t&& consistent_callback) -{ - WriteLock lock(rwlock_); - - is_consistent_ = true; - consistent_callback_ = std::move(consistent_callback); - return 0; -} - -int Upstream::set_attr(bool try_another, upstream_route_t rehash_callback) -{ - WriteLock lock(rwlock_); - - try_another_ = try_another; - consistent_callback_ = std::move(rehash_callback); - return 0; -} - -void Upstream::notify_unavailable(UpstreamAddress *ua) -{ - auto *group = ua->group; - - if (group) - { - std::lock_guard lock(group->mutex); - - if (++ua->fail_count == ua->params.max_fails && ua->group == group && !ua->list.next) - { - ua->broken_timeout = GET_CURRENT_SECOND + MTTR_SECOND; - list_add_tail(&ua->list, &group->breaker_list); - group->nbreak++; - group->upstream->lose_one_server(group, ua); - } - } - else - ++ua->fail_count; -} - -void Upstream::notify_available(UpstreamAddress *ua) -{ - auto *group = ua->group; - - if (group) - { - std::lock_guard lock(group->mutex); - - if (ua->list.next) // in the list - { - if (ua->group == group && ua->fail_count >= ua->params.max_fails) - group->upstream->gain_one_server(group, ua); - - list_del(&ua->list); - ua->list.next = NULL; - } - - ua->fail_count = 0; - } - else - ua->fail_count = 0; -} +#include "WFNameService.h" +#include "UpstreamPolicies.h" class __UpstreamManager { @@ -733,286 +30,62 @@ public: return &kInstance; } - int upstream_create(const std::string& name, - upstream_route_t&& consistent_hash) + void add_policy_name(const std::string& name) { - Upstream *upstream = NULL; - { - WriteLock lock(rwlock_); - - if (upstream_map_.find(name) == upstream_map_.end()) - upstream = &upstream_map_[name]; - } - - if (upstream) - return upstream->set_consistent_mode(std::move(consistent_hash)); - - return -1; - } - - int upstream_create(const std::string& name, bool try_another) - { - Upstream *upstream = NULL; - { - WriteLock lock(rwlock_); - - if (upstream_map_.find(name) == upstream_map_.end()) - upstream = &upstream_map_[name]; - } - - if (upstream) - return upstream->set_attr(try_another, nullptr); - - return -1; - } - - int upstream_create(const std::string& name, - upstream_route_t&& select, - bool try_another, - upstream_route_t&& consistent_hash) - { - Upstream *upstream = NULL; - { - WriteLock lock(rwlock_); - - if (upstream_map_.find(name) == upstream_map_.end()) - upstream = &upstream_map_[name]; - } - - if (upstream) - { - upstream->set_select_callback(std::move(select)); - upstream->set_attr(try_another, std::move(consistent_hash)); - return 0; - } - - return 0; - } - - int upstream_add_server(const std::string& name, - const std::string& address, - const AddressParams *address_params) - { - auto *ua = new UpstreamAddress(address, address_params); - { - WriteLock lock(rwlock_); - - addresses_.push_back(ua); - } - - Upstream *upstream = NULL; - { - ReadLock lock(rwlock_); - auto it = upstream_map_.find(name); - - if (it != upstream_map_.end()) - upstream = &it->second; - } - - if (upstream) - return upstream->add(ua); - - return -1; - } - - int upstream_remove_server(const std::string& name, const std::string& address) - { - Upstream *upstream = NULL; - { - ReadLock lock(rwlock_); - auto it = upstream_map_.find(name); - - if (it != upstream_map_.end()) - upstream = &it->second; - } - - if (upstream) - return upstream->del(address); - - return -1; - } - - int upstream_replace_server(const std::string& name, - const std::string& address, - const AddressParams *address_params) - { - Upstream *upstream = NULL; - auto *ua = new UpstreamAddress(address, address_params); - WriteLock lock(rwlock_); - - addresses_.push_back(ua); - auto it = upstream_map_.find(name); - - if (it != upstream_map_.end()) - upstream = &it->second; - - if (upstream) - { - upstream->del(address); - return upstream->add(ua); - } - - return -1; - } - - int upstream_disable_server(const std::string& name, const std::string& address) - { - Upstream *upstream = NULL; - { - ReadLock lock(rwlock_); - auto it = upstream_map_.find(name); - - if (it != upstream_map_.end()) - upstream = &it->second; - } - - if (upstream) - { - upstream->disable_server(address); - return 0; - } - - return -1; - } - - int upstream_enable_server(const std::string& name, const std::string& address) - { - Upstream *upstream = NULL; - { - ReadLock lock(rwlock_); - auto it = upstream_map_.find(name); - - if (it != upstream_map_.end()) - upstream = &it->second; - } - - if (upstream) - { - upstream->enable_server(address); - return 0; - } - - return -1; - } - - std::vector upstream_main_address_list(const std::string& name) - { - std::vector addr_list; - Upstream *upstream = NULL; - { - ReadLock lock(rwlock_); - auto it = upstream_map_.find(name); - - if (it != upstream_map_.end()) - upstream = &it->second; - } - - if (upstream) - upstream->get_all_main(addr_list); - - return addr_list; - } - - int upstream_delete(const std::string& name) - { - WriteLock lock(rwlock_); - const auto it = upstream_map_.find(name); - - if (it == upstream_map_.end()) - return 0; - else - upstream_map_.erase(it); - - return 1; - } - - int upstream_choose(ParsedURI& uri, UpstreamManager::UpstreamResult& result) - { - result.cookie = NULL; - result.address_params = NULL; - result.state = UPSTREAM_NOTFOUND; - - if (uri.state != URI_STATE_SUCCESS || !uri.host) - return 0;// UPSTREAM_NOTFOUND - - Upstream *upstream; - { - ReadLock lock(rwlock_); - auto it = upstream_map_.find(uri.host); - - if (it == upstream_map_.end()) - return 0;// UPSTREAM_NOTFOUND - - upstream = &it->second; - } - - const auto *ua = upstream->get(uri); - - if (!ua) - { - result.state = UPSTREAM_ALL_DOWN; - return 0; - } - - char *host = NULL; - char *port = NULL; - - if (!ua->host.empty()) - { - host = strdup(ua->host.c_str()); - if (!host) - return -1; - } - - if (ua->port_value > 0) - { - port = strdup(ua->port.c_str()); - if (!port) - { - free(host); - return -1; - } - - free(uri.port); - uri.port = port; - } - - free(uri.host); - uri.host = host; - - result.state = UPSTREAM_SUCCESS; - result.address_params = &ua->params; - result.cookie = const_cast(ua); - return 0; + this->mutex.lock(); + this->upstream_names.push_back(name); + this->mutex.unlock(); } private: + __UpstreamManager() + { + } + ~__UpstreamManager() { - for (auto *ua : addresses_) - delete ua; + WFNSPolicy *policy; + WFNameService *ns = WFGlobal::get_name_service(); + + for (const std::string& name : this->upstream_names) + { + policy = ns->del_policy(name.c_str()); + delete policy; + } } -private: - RWLock rwlock_; - std::unordered_map upstream_map_; - std::vector addresses_; + std::mutex mutex; + std::vector upstream_names; }; int UpstreamManager::upstream_create_consistent_hash(const std::string& name, - upstream_route_t consitent_hash) + upstream_route_t consistent_hash) { - auto *manager = __UpstreamManager::get_instance(); + auto *ns = WFGlobal::get_name_service(); + UPSConsistentHashPolicy *policy = new UPSConsistentHashPolicy(std::move(consistent_hash)); - return manager->upstream_create(name, std::move(consitent_hash)); + if (ns->add_policy(name.c_str(), policy) >= 0) + { + __UpstreamManager::get_instance()->add_policy_name(name); + return 0; + } + + return -1; } int UpstreamManager::upstream_create_weighted_random(const std::string& name, bool try_another) { - auto *manager = __UpstreamManager::get_instance(); + auto *ns = WFGlobal::get_name_service(); + UPSWeightedRandomPolicy *policy = new UPSWeightedRandomPolicy(try_another); - return manager->upstream_create(name, try_another); + if (ns->add_policy(name.c_str(), policy) >= 0) + { + __UpstreamManager::get_instance()->add_policy_name(name); + return 0; + } + + return -1; } int UpstreamManager::upstream_create_manual(const std::string& name, @@ -1020,92 +93,118 @@ int UpstreamManager::upstream_create_manual(const std::string& name, bool try_another, upstream_route_t consitent_hash) { - auto *manager = __UpstreamManager::get_instance(); + auto *ns = WFGlobal::get_name_service(); + UPSManualPolicy *policy = new UPSManualPolicy(try_another, + std::move(select), + std::move(consitent_hash)); + if (ns->add_policy(name.c_str(), policy) >= 0) + { + __UpstreamManager::get_instance()->add_policy_name(name); + return 0; + } - return manager->upstream_create(name, std::move(select), - try_another, std::move(consitent_hash)); + return -1; } int UpstreamManager::upstream_add_server(const std::string& name, const std::string& address) { - auto *manager = __UpstreamManager::get_instance(); - - return manager->upstream_add_server(name, address, &ADDRESS_PARAMS_DEFAULT); + return UpstreamManager::upstream_add_server(name, address, + &ADDRESS_PARAMS_DEFAULT); } int UpstreamManager::upstream_add_server(const std::string& name, const std::string& address, const AddressParams *address_params) { - auto *manager = __UpstreamManager::get_instance(); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); - return manager->upstream_add_server(name, address, address_params); + if (policy) + { + policy->add_server(address, address_params); + return 0; + } + + errno = ENOENT; + return -1; } int UpstreamManager::upstream_remove_server(const std::string& name, const std::string& address) { - auto *manager = __UpstreamManager::get_instance(); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); - return manager->upstream_remove_server(name, address); + if (policy) + { + policy->remove_server(address); + return 0; + } + + errno = ENOENT; + return -1; } -int UpstreamManager::upstream_replace_server(const std::string& name, - const std::string& address, - const AddressParams *address_params) +std::vector +UpstreamManager::upstream_main_address_list(const std::string& name) { - auto *manager = __UpstreamManager::get_instance(); + std::vector address; + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); - return manager->upstream_replace_server(name, address, address_params); + if (policy) + policy->get_main_address(address); + + return address; } int UpstreamManager::upstream_disable_server(const std::string& name, const std::string& address) { - auto *manager = __UpstreamManager::get_instance(); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); - return manager->upstream_disable_server(name, address); + if (policy) + { + policy->disable_server(address); + return 0; + } + + errno = ENOENT; + return -1; } int UpstreamManager::upstream_enable_server(const std::string& name, const std::string& address) { - auto *manager = __UpstreamManager::get_instance(); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); - return manager->upstream_enable_server(name, address); + if (policy) + { + policy->enable_server(address); + return 0; + } + + errno = ENOENT; + return -1; } -std::vector UpstreamManager::upstream_main_address_list(const std::string& name) +int UpstreamManager::upstream_replace_server(const std::string& name, + const std::string& address, + const struct AddressParams *address_params) { - auto *manager = __UpstreamManager::get_instance(); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_cast(ns->get_policy(name.c_str())); - return manager->upstream_main_address_list(name); -} - -int UpstreamManager::upstream_delete(const std::string& name) -{ - auto *manager = __UpstreamManager::get_instance(); - - return manager->upstream_delete(name); -} - -int UpstreamManager::choose(ParsedURI& uri, UpstreamResult& result) -{ - auto *manager = __UpstreamManager::get_instance(); - - return manager->upstream_choose(uri, result); -} - -void UpstreamManager::notify_unavailable(void *cookie) -{ - if (cookie) - Upstream::notify_unavailable((UpstreamAddress *)cookie); -} - -void UpstreamManager::notify_available(void *cookie) -{ - if (cookie) - Upstream::notify_available((UpstreamAddress *)cookie); + if (policy) + { + policy->replace_server(address, address_params); + return 0; + } + + errno = ENOENT; + return -1; } diff --git a/src/manager/UpstreamManager.h b/src/manager/UpstreamManager.h index 87cd2b7b..2daa2ef2 100644 --- a/src/manager/UpstreamManager.h +++ b/src/manager/UpstreamManager.h @@ -23,6 +23,7 @@ #include #include "URIParser.h" #include "EndpointParams.h" +#include "WFGlobal.h" /** * @file UpstreamManager.h @@ -58,9 +59,9 @@ struct AddressParams * - 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 + 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 backup for any main + int group_id; ///< -1 means no group. Backup without group will be backup for any main }; /** @@ -68,13 +69,13 @@ struct AddressParams */ 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 + .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, }; /** @@ -84,7 +85,7 @@ static constexpr struct AddressParams ADDRESS_PARAMS_DEFAULT = * 1. Weighted-Random * 2. Consistent-Hash * 3. Manual-Select - * - Additional, we support Main-Backup & Group for server and working well in any mode. + * - Additional, we support Main-backup & Group for server and working well in any mode. * * @code{.cc} upstream_create_weighted_random("abc.sogou", true); //UPSTREAM_WIGHTED_RANDOM @@ -218,9 +219,9 @@ public: const std::string& address); /** - * @brief get all main address list from one upstream + * @brief get all main servers address list from one upstream * @param[in] name upstream name - * @return all main address list + * @return all main servers' address list * @warning If server servers has the same address in this upstream, then will appear in the vector multiply times */ static std::vector upstream_main_address_list(const std::string& name); @@ -234,39 +235,6 @@ public: const std::string& address, const struct AddressParams *address_params); -public: - /// @brief Internal use only - class UpstreamResult - { -public: - void *cookie; - const struct AddressParams *address_params; -#define UPSTREAM_SUCCESS 0 -#define UPSTREAM_NOTFOUND 1 -#define UPSTREAM_ALL_DOWN 2 - int state; - -public: - UpstreamResult(): - cookie(NULL), - address_params(NULL), - state(UPSTREAM_NOTFOUND) - {} - - void clear() - { - cookie = NULL; - address_params = NULL; - state = UPSTREAM_NOTFOUND; - } - }; - - /// @brief Internal use only - static int choose(ParsedURI& uri, UpstreamResult& result); - /// @brief Internal use only - static void notify_unavailable(void *cookie); - /// @brief Internal use only - static void notify_available(void *cookie); }; #endif diff --git a/src/manager/WFGlobal.cc b/src/manager/WFGlobal.cc index 2760461a..a55c4236 100644 --- a/src/manager/WFGlobal.cc +++ b/src/manager/WFGlobal.cc @@ -38,8 +38,8 @@ #include "RWLock.h" #include "WFTask.h" #include "WFTaskError.h" - -const char *library_version = "Workflow Library Version: 1.12.6"; +#include "WFNameService.h" +#include "WFDNSResolver.h" class __WFGlobal { @@ -104,30 +104,7 @@ public: } private: - __WFGlobal(): - settings_(GLOBAL_SETTINGS_DEFAULT) - { - static_scheme_port_["http"] = "80"; - static_scheme_port_["https"] = "443"; - static_scheme_port_["redis"] = "6379"; - static_scheme_port_["rediss"] = "6379"; - static_scheme_port_["mysql"] = "3306"; - static_scheme_port_["kafka"] = "9092"; - sync_count_ = 0; - sync_max_ = 0; - -#ifdef _WIN32 - WSADATA wsaData; - WORD wVersionRequested = MAKEWORD(2, 2); - int err = WSAStartup(wVersionRequested, &wsaData); - - if (err != 0) - abort(); - - if (LOBYTE(wsaData.wVersion) != 2 || HIBYTE(wsaData.wVersion) != 2) - abort(); -#endif - } + __WFGlobal(); #ifdef _WIN32 ~__WFGlobal() @@ -146,6 +123,52 @@ private: int sync_max_; }; +__WFGlobal::__WFGlobal() : settings_(GLOBAL_SETTINGS_DEFAULT) +{ + static_scheme_port_["http"] = "80"; + static_scheme_port_["Http"] = "80"; + static_scheme_port_["HTTP"] = "80"; + + static_scheme_port_["https"] = "443"; + static_scheme_port_["Https"] = "443"; + static_scheme_port_["HTTPs"] = "443"; + static_scheme_port_["HTTPS"] = "443"; + + static_scheme_port_["redis"] = "6379"; + static_scheme_port_["Redis"] = "6379"; + static_scheme_port_["REDIS"] = "6379"; + + static_scheme_port_["rediss"] = "6379"; + static_scheme_port_["Rediss"] = "6379"; + static_scheme_port_["REDISs"] = "6379"; + static_scheme_port_["REDISS"] = "6379"; + + static_scheme_port_["mysql"] = "3306"; + static_scheme_port_["Mysql"] = "3306"; + static_scheme_port_["MySql"] = "3306"; + static_scheme_port_["MySQL"] = "3306"; + static_scheme_port_["MYSQL"] = "3306"; + + static_scheme_port_["kafka"] = "9092"; + static_scheme_port_["Kafka"] = "9092"; + static_scheme_port_["KAFKA"] = "9092"; + + sync_count_ = 0; + sync_max_ = 0; + +#ifdef _WIN32 + WSADATA wsaData; + WORD wVersionRequested = MAKEWORD(2, 2); + int err = WSAStartup(wVersionRequested, &wsaData); + + if (err != 0) + abort(); + + if (LOBYTE(wsaData.wVersion) != 2 || HIBYTE(wsaData.wVersion) != 2) + abort(); +#endif +} + static __WFGlobal *_g_global = __WFGlobal::get_instance(); #if OPENSSL_VERSION_NUMBER < 0x10100000L @@ -514,6 +537,28 @@ private: Executor compute_executor_; }; +class __NameServiceManager +{ +public: + static __NameServiceManager *get_instance() + { + static __NameServiceManager kInstance; + return &kInstance; + } + +public: + WFNameService *get_name_service() { return &service_; } + +private: + static WFDNSResolver resolver_; + WFNameService service_; + +public: + __NameServiceManager() : service_(&__NameServiceManager::resolver_) { } +}; + +WFDNSResolver __NameServiceManager::resolver_; + CommScheduler *WFGlobal::get_scheduler() { return __CommManager::get_instance()->get_scheduler(); @@ -564,6 +609,11 @@ Executor *WFGlobal::get_dns_executor() return __CommManager::get_instance()->get_dns_executor(); } +WFNameService *WFGlobal::get_name_service() +{ + return __NameServiceManager::get_instance()->get_name_service(); +} + const char *WFGlobal::get_default_port(const std::string& scheme) { return __WFGlobal::get_instance()->get_default_port(scheme); diff --git a/src/manager/WFGlobal.h b/src/manager/WFGlobal.h index 02d00448..89fc9951 100644 --- a/src/manager/WFGlobal.h +++ b/src/manager/WFGlobal.h @@ -33,6 +33,7 @@ #include "RouteManager.h" #include "Executor.h" #include "EndpointParams.h" +#include "WFNameService.h" /** * @file WFGlobal.h @@ -107,29 +108,20 @@ public: static const char *get_error_string(int state, int error); public: - /// @brief Internal use only + // Internal usage only static CommScheduler *get_scheduler(); - /// @brief Internal use only static DNSCache *get_dns_cache(); - /// @brief Internal use only static RouteManager *get_route_manager(); - /// @brief Internal use only static SSL_CTX *get_ssl_client_ctx(); - /// @brief Internal use only static SSL_CTX *get_ssl_server_ctx(); - /// @brief Internal use only static ExecQueue *get_exec_queue(const std::string& queue_name); - /// @brief Internal use only static Executor *get_compute_executor(); - /// @brief Internal use only static IOService *get_io_service(); - /// @brief Internal use only static ExecQueue *get_dns_queue(); - /// @brief Internal use only static Executor *get_dns_executor(); - /// @brief Internal use only static void sync_operation_begin(); static void sync_operation_end(); + static WFNameService *get_name_service(); }; #endif