diff --git a/CMakeLists_Headers.txt b/CMakeLists_Headers.txt index 63585d21..254260fc 100644 --- a/CMakeLists_Headers.txt +++ b/CMakeLists_Headers.txt @@ -84,7 +84,7 @@ set(INCLUDE_HEADERS src/factory/WFOperator.h src/factory/WFNameService.h src/factory/WFDNSResolver.h - src/manager/GovernancePolicy.h + src/manager/UPSPolicy.h ) if(KAFKA STREQUAL "y") diff --git a/src/manager/CMakeLists.txt b/src/manager/CMakeLists.txt index 2596d755..c1663af4 100644 --- a/src/manager/CMakeLists.txt +++ b/src/manager/CMakeLists.txt @@ -3,12 +3,12 @@ project(manager) set(SRC DNSCache.cc - UpstreamManager.cc +# UpstreamManager.cc RouteManager.cc WFGlobal.cc - GovernancePolicy.cc - GroupPolicy.cc - SelectPolicy.cc + UPSPolicy.cc + UPSGroupPolicy.cc + UPSSelectPolicy.cc ) add_library(${PROJECT_NAME} OBJECT ${SRC}) diff --git a/src/manager/GroupPolicy.cc b/src/manager/UPSGroupPolicy.cc similarity index 88% rename from src/manager/GroupPolicy.cc rename to src/manager/UPSGroupPolicy.cc index 550a474b..6fa851cc 100644 --- a/src/manager/GroupPolicy.cc +++ b/src/manager/UPSGroupPolicy.cc @@ -1,8 +1,8 @@ -#include "GovernancePolicy.h" +#include "UPSPolicy.h" #define VIRTUAL_GROUP_SIZE 16 -GroupPolicy::GroupPolicy() +UPSGroupPolicy::UPSGroupPolicy() { this->group_map.rb_node = NULL; this->default_group = new EndpointGroup(-1, this); @@ -10,7 +10,7 @@ GroupPolicy::GroupPolicy() rb_insert_color(&this->default_group->rb, &this->group_map); } -GroupPolicy::~GroupPolicy() +UPSGroupPolicy::~UPSGroupPolicy() { EndpointGroup *group; @@ -22,7 +22,7 @@ GroupPolicy::~GroupPolicy() } } -bool GroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) +bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) { pthread_rwlock_rdlock(&this->rwlock); unsigned int n = (unsigned int)this->servers.size(); @@ -123,16 +123,13 @@ const EndpointAddress *EndpointGroup::get_one_backup() return addr; } -void GroupPolicy::add_server(const std::string& address, - const AddressParams *address_params) +void UPSGroupPolicy::__add_server(EndpointAddress *addr) { - EndpointAddress *addr = new EndpointAddress(address, address_params); int group_id = addr->params.group_id; rb_node **p = &this->group_map.rb_node; rb_node *parent = NULL; EndpointGroup *group; - pthread_rwlock_wrlock(&this->rwlock); this->addresses.push_back(addr); this->server_map[addr->address].push_back(addr); @@ -168,13 +165,11 @@ void GroupPolicy::add_server(const std::string& address, group->backups.push_back(addr); pthread_mutex_unlock(&group->mutex); - pthread_rwlock_unlock(&this->rwlock); return; } -int GroupPolicy::remove_server(const std::string& address) +int UPSGroupPolicy::__remove_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()) @@ -232,11 +227,10 @@ int GroupPolicy::remove_server(const std::string& address) ret = n - new_n; } - pthread_rwlock_unlock(&this->rwlock); return ret; } -const EndpointAddress *GroupPolicy::consistent_hash_with_group(unsigned int hash) const +const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int hash) const { const EndpointAddress *addr = NULL; unsigned int min_dis = (unsigned int)-1; diff --git a/src/manager/GovernancePolicy.cc b/src/manager/UPSPolicy.cc similarity index 80% rename from src/manager/GovernancePolicy.cc rename to src/manager/UPSPolicy.cc index d2ac8097..913eb7ee 100644 --- a/src/manager/GovernancePolicy.cc +++ b/src/manager/UPSPolicy.cc @@ -1,4 +1,4 @@ -#include "GovernancePolicy.h" +#include "UPSPolicy.h" #include "StringUtil.h" #define DNS_CACHE_LEVEL_1 1 @@ -21,7 +21,7 @@ public: } }; -static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr) +static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr) { char *host = NULL; char *port = NULL; @@ -87,7 +87,7 @@ EndpointAddress::EndpointAddress(const std::string& address, } } -WFRouterTask *GovernancePolicy::create_router_task(const struct WFNSParams *params, +WFRouterTask *UPSPolicy::create_router_task(const struct WFNSParams *params, router_callback_t callback) { EndpointAddress *addr = NULL; @@ -112,7 +112,7 @@ WFRouterTask *GovernancePolicy::create_router_task(const struct WFNSParams *para return task; } -inline void GovernancePolicy::recover_server_from_breaker(EndpointAddress *addr) +inline void UPSPolicy::recover_server_from_breaker(EndpointAddress *addr) { addr->fail_count = 0; pthread_mutex_lock(&this->breaker_lock); @@ -126,7 +126,7 @@ inline void GovernancePolicy::recover_server_from_breaker(EndpointAddress *addr) pthread_mutex_unlock(&this->breaker_lock); } -inline void GovernancePolicy::fuse_server_to_breaker(EndpointAddress *addr) +inline void UPSPolicy::fuse_server_to_breaker(EndpointAddress *addr) { pthread_mutex_lock(&this->breaker_lock); if (!addr->list.next) @@ -139,7 +139,7 @@ inline void GovernancePolicy::fuse_server_to_breaker(EndpointAddress *addr) pthread_mutex_unlock(&this->breaker_lock); } -void GovernancePolicy::success(RouteManager::RouteResult *result, void *cookie, +void UPSPolicy::success(RouteManager::RouteResult *result, void *cookie, CommTarget *target) { pthread_rwlock_rdlock(&this->rwlock); @@ -149,7 +149,7 @@ void GovernancePolicy::success(RouteManager::RouteResult *result, void *cookie, WFDNSResolver::success(result, NULL, target); } -void GovernancePolicy::failed(RouteManager::RouteResult *result, void *cookie, +void UPSPolicy::failed(RouteManager::RouteResult *result, void *cookie, CommTarget *target) { EndpointAddress *server = (EndpointAddress *)cookie; @@ -164,7 +164,7 @@ void GovernancePolicy::failed(RouteManager::RouteResult *result, void *cookie, WFDNSResolver::failed(result, NULL, target); } -void GovernancePolicy::check_breaker() +void UPSPolicy::check_breaker() { pthread_mutex_lock(&this->breaker_lock); if (!list_empty(&this->breaker_list)) @@ -193,18 +193,18 @@ void GovernancePolicy::check_breaker() //this->server_list_change(); } -const EndpointAddress *GovernancePolicy::first_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSPolicy::first_stradegy(const ParsedURI& uri) { unsigned int idx = rand() % this->servers.size(); return this->servers[idx]; } -const EndpointAddress *GovernancePolicy::another_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSPolicy::another_stradegy(const ParsedURI& uri) { return this->first_stradegy(uri); } -bool GovernancePolicy::select(const ParsedURI& uri, EndpointAddress **addr) +bool UPSPolicy::select(const ParsedURI& uri, EndpointAddress **addr) { pthread_rwlock_rdlock(&this->rwlock); unsigned int n = (unsigned int)this->servers.size(); @@ -242,22 +242,16 @@ bool GovernancePolicy::select(const ParsedURI& uri, EndpointAddress **addr) return false; } -void GovernancePolicy::add_server(const std::string& address, - const AddressParams *address_params) +void UPSPolicy::__add_server(EndpointAddress *addr) { - auto *addr = new EndpointAddress(address, address_params); - - pthread_rwlock_wrlock(&this->rwlock); this->addresses.push_back(addr); this->server_map[addr->address].push_back(addr); this->servers.push_back(addr); this->recover_one_server(addr); - pthread_rwlock_unlock(&this->rwlock); } -int GovernancePolicy::remove_server(const std::string& address) +int UPSPolicy::__remove_server(const std::string& address) { - pthread_rwlock_wrlock(&this->rwlock); const auto map_it = this->server_map.find(address); if (map_it != this->server_map.cend()) { @@ -291,11 +285,42 @@ int GovernancePolicy::remove_server(const std::string& address) 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(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(address); pthread_rwlock_unlock(&this->rwlock); return ret; } -void GovernancePolicy::enable_server(const std::string& address) +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(addr); + ret = this->__remove_server(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); @@ -307,7 +332,7 @@ void GovernancePolicy::enable_server(const std::string& address) pthread_rwlock_unlock(&this->rwlock); } -void GovernancePolicy::disable_server(const std::string& address) +void UPSPolicy::disable_server(const std::string& address) { pthread_rwlock_rdlock(&this->rwlock); const auto map_it = this->server_map.find(address); diff --git a/src/manager/GovernancePolicy.h b/src/manager/UPSPolicy.h similarity index 83% rename from src/manager/GovernancePolicy.h rename to src/manager/UPSPolicy.h index bd10faf4..9da9e543 100644 --- a/src/manager/GovernancePolicy.h +++ b/src/manager/UPSPolicy.h @@ -26,8 +26,8 @@ struct AddressParams }; class EndpointGroup; -class GovernancePolicy; -class GroupPolicy; +class UPSPolicy; +class UPSGroupPolicy; class EndpointAddress { @@ -52,7 +52,7 @@ class EndpointGroup { public: int id; - GroupPolicy *policy; + UPSGroupPolicy *policy; struct rb_node rb; pthread_mutex_t mutex; std::vector mains; @@ -60,7 +60,7 @@ public: std::atomic nalives; int weight; - EndpointGroup(int group_id, GroupPolicy *policy) + EndpointGroup(int group_id, UPSGroupPolicy *policy) { this->id = group_id; this->policy = policy; @@ -79,7 +79,7 @@ public: const EndpointAddress *get_one_backup(); }; -class GovernancePolicy : public WFDNSResolver +class UPSPolicy : public WFDNSResolver { public: virtual WFRouterTask *create_router_task(const struct WFNSParams *params, @@ -89,17 +89,17 @@ public: virtual void failed(RouteManager::RouteResult *result, void *cookie, CommTarget *target); - // GovernancePolicy *gp = dynamic_cast(policy); gp->add_server(...); - virtual void add_server(const std::string& address, const AddressParams *address_params); - virtual int remove_server(const std::string& address); + // UPSPolicy *gp = dynamic_cast(policy); gp->add_server(...); + 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 server_list_change(/* std::vector status */) {} -private: public: - GovernancePolicy() + UPSPolicy() { this->nalives = 0; this->try_another = false; @@ -108,7 +108,7 @@ public: INIT_LIST_HEAD(&this->breaker_list); } - ~GovernancePolicy() + ~UPSPolicy() { pthread_mutex_destroy(&this->breaker_lock); pthread_rwlock_destroy(&this->rwlock); @@ -129,6 +129,9 @@ private: this->nalives--; } + virtual void __add_server(EndpointAddress *addr); + virtual int __remove_server(const std::string& address); + void recover_server_from_breaker(EndpointAddress *addr); void fuse_server_to_breaker(EndpointAddress *addr); @@ -149,15 +152,13 @@ protected: bool try_another; }; -class GroupPolicy : public GovernancePolicy +class UPSGroupPolicy : public UPSPolicy { public: - GroupPolicy(); - ~GroupPolicy(); + UPSGroupPolicy(); + ~UPSGroupPolicy(); struct rb_root group_map; EndpointGroup *default_group; - virtual void add_server(const std::string& address, const AddressParams *address_params); - virtual int remove_server(const std::string& address); private: virtual void recover_one_server(const EndpointAddress *addr) @@ -173,6 +174,8 @@ private: } // override: select() add_server() remove_server() virtual bool select(const ParsedURI& uri, EndpointAddress **addr); + virtual void __add_server(EndpointAddress *addr); + virtual int __remove_server(const std::string& address); protected: const EndpointAddress *consistent_hash_with_group(unsigned int hash) const; @@ -196,10 +199,10 @@ protected: } }; -class WeightedRandomPolicy : public GroupPolicy +class UPSWeightedRandomPolicy : public UPSGroupPolicy { public: - WeightedRandomPolicy(bool try_another) + UPSWeightedRandomPolicy(bool try_another) { this->total_weight = 0; this->available_weight = 0; @@ -219,15 +222,15 @@ private: using select_t = std::function; -class ConsistentHashPolicy : public GroupPolicy +class UPSConsistentHashPolicy : public UPSGroupPolicy { public: - ConsistentHashPolicy() + UPSConsistentHashPolicy() { this->consistent_hash = this->default_consistent_hash; } - ConsistentHashPolicy(select_t consistent_hash) + UPSConsistentHashPolicy(select_t consistent_hash) { this->consistent_hash = std::move(consistent_hash); } @@ -252,10 +255,10 @@ public: } }; -class ManualPolicy : public GroupPolicy +class UPSManualPolicy : public UPSGroupPolicy { public: - ManualPolicy(bool try_another, select_t select, select_t try_another_select) + UPSManualPolicy(bool try_another, select_t select, select_t try_another_select) { this->try_another = try_another; this->manual_select = select; diff --git a/src/manager/SelectPolicy.cc b/src/manager/UPSSelectPolicy.cc similarity index 77% rename from src/manager/SelectPolicy.cc rename to src/manager/UPSSelectPolicy.cc index e89bba42..5d76976d 100644 --- a/src/manager/SelectPolicy.cc +++ b/src/manager/UPSSelectPolicy.cc @@ -1,6 +1,6 @@ -#include "GovernancePolicy.h" +#include "UPSPolicy.h" -const EndpointAddress *WeightedRandomPolicy::first_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI& uri) { int x = 0; int s = 0; @@ -23,7 +23,7 @@ const EndpointAddress *WeightedRandomPolicy::first_stradegy(const ParsedURI& uri return this->servers[idx]; } -const EndpointAddress *WeightedRandomPolicy::another_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI& uri) { int temp_weight = this->available_weight; if (temp_weight == 0) @@ -47,7 +47,7 @@ const EndpointAddress *WeightedRandomPolicy::another_stradegy(const ParsedURI& u } -void WeightedRandomPolicy::recover_one_server(const EndpointAddress *addr) +void UPSWeightedRandomPolicy::recover_one_server(const EndpointAddress *addr) { this->nalives++; if (addr->group->nalives++ == 0 && addr->group->id > 0) @@ -57,7 +57,7 @@ void WeightedRandomPolicy::recover_one_server(const EndpointAddress *addr) this->available_weight += addr->params.weight; } -void WeightedRandomPolicy::fuse_one_server(const EndpointAddress *addr) +void UPSWeightedRandomPolicy::fuse_one_server(const EndpointAddress *addr) { this->nalives--; if (--addr->group->nalives == 0 && addr->group->id > 0) @@ -67,7 +67,7 @@ void WeightedRandomPolicy::fuse_one_server(const EndpointAddress *addr) this->available_weight -= addr->params.weight; } -const EndpointAddress *ConsistentHashPolicy::first_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& uri) { unsigned int hash_value; @@ -82,7 +82,7 @@ const EndpointAddress *ConsistentHashPolicy::first_stradegy(const ParsedURI& uri return this->consistent_hash_with_group(hash_value); } -const EndpointAddress *ManualPolicy::first_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSManualPolicy::first_stradegy(const ParsedURI& uri) { int idx = this->manual_select(uri.path ? uri.path : "", uri.query ? uri.query : "", @@ -94,7 +94,7 @@ const EndpointAddress *ManualPolicy::first_stradegy(const ParsedURI& uri) return this->servers[idx]; } -const EndpointAddress *ManualPolicy::another_stradegy(const ParsedURI& uri) +const EndpointAddress *UPSManualPolicy::another_stradegy(const ParsedURI& uri) { unsigned int hash_value; @@ -103,7 +103,7 @@ const EndpointAddress *ManualPolicy::another_stradegy(const ParsedURI& uri) uri.query ? uri.query : "", uri.fragment ? uri.fragment : ""); else - hash_value = ConsistentHashPolicy::default_consistent_hash(uri.path ? uri.path : "", + hash_value = UPSConsistentHashPolicy::default_consistent_hash(uri.path ? uri.path : "", uri.query ? uri.query : "", uri.fragment ? uri.fragment : ""); return this->consistent_hash_with_group(hash_value); diff --git a/src/manager/UpstreamManager.cc b/src/manager/UpstreamManager.cc index b5adad77..09cbac1e 100644 --- a/src/manager/UpstreamManager.cc +++ b/src/manager/UpstreamManager.cc @@ -16,1036 +16,36 @@ Authors: Wu Jiaxu (wujiaxu@sogou-inc.com) */ -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include "list.h" -#include "rbtree.h" -#include "URIParser.h" -#include "StringUtil.h" -#include "EndpointParams.h" #include "UpstreamManager.h" +#include "workflow/WFNameService.h" +#include "UPSPolicy.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 - -namespace //anoymous namespace, for safe, avoid conflict +// existed: -1; success: 0; +UpstreamManager::~UpstreamManager() { -// RAII: YES -class ReadLock -{ -public: - ReadLock(pthread_rwlock_t& rwlock): rwlock_(&rwlock) { pthread_rwlock_rdlock(rwlock_); } - ReadLock(pthread_rwlock_t *rwlock): rwlock_(rwlock) { pthread_rwlock_rdlock(rwlock_); } - ~ReadLock() { pthread_rwlock_unlock(rwlock_); } - -private: - pthread_rwlock_t *rwlock_; -}; - -// RAII: YES -class WriteLock -{ -public: - WriteLock(pthread_rwlock_t& rwlock): rwlock_(&rwlock) { pthread_rwlock_wrlock(rwlock_); } - WriteLock(pthread_rwlock_t *rwlock): rwlock_(rwlock) { pthread_rwlock_wrlock(rwlock_); } - ~WriteLock() { pthread_rwlock_unlock(rwlock_); } - -private: - pthread_rwlock_t *rwlock_; -}; - + for (const NSPolicy *policy : this->policies) + delete policy; + pthread_mutex_destroy(&this->lock); } -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: - pthread_rwlock_t 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 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] = 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(): - rwlock_(PTHREAD_RWLOCK_INITIALIZER), - 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; -} - -class __UpstreamManager -{ -public: - static __UpstreamManager *get_instance() - { - static __UpstreamManager kInstance; - return &kInstance; - } - - int upstream_create(const std::string& name, - 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) - 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; - } - -private: - __UpstreamManager(): - rwlock_(PTHREAD_RWLOCK_INITIALIZER) - {} - - ~__UpstreamManager() - { - for (auto *ua : addresses_) - delete ua; - } - -private: - pthread_rwlock_t rwlock_; - std::unordered_map upstream_map_; - std::vector addresses_; -}; - 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(); - - return manager->upstream_create(name, std::move(consitent_hash)); + UPSConsistentHashPolicy *policy = new UPSConsistentHashPolicy(std::move(consistent_hash)); + pthread_mutex_lock(&this->lock); + this->policies->push_back(policy); + pthread_mutex_unlock(&this->lock); + return WFGlobal::get_name_service()->add_policy(name, policy); } int UpstreamManager::upstream_create_weighted_random(const std::string& name, bool try_another) { - auto *manager = __UpstreamManager::get_instance(); - - return manager->upstream_create(name, try_another); + UPSWeightedRandomPolicy *policy = new UPSWeightedRandomPolicy(try_another); + pthread_mutex_lock(&this->lock); + this->policies->push_back(policy); + pthread_mutex_unlock(&this->lock); + return WFGlobal::get_name_service()->add_policy(name, policy); } int UpstreamManager::upstream_create_manual(const std::string& name, @@ -1053,92 +53,93 @@ int UpstreamManager::upstream_create_manual(const std::string& name, bool try_another, upstream_route_t consitent_hash) { - auto *manager = __UpstreamManager::get_instance(); - - return manager->upstream_create(name, std::move(select), - try_another, std::move(consitent_hash)); + UPSManualPolicy *policy = new UPSManualPolicy(try_another, + std::move(select), + std::move(consitent_hash)); + pthread_mutex_lock(&this->lock); + this->policies->push_back(policy); + pthread_mutex_unlock(&this->lock); + return WFGlobal::get_name_service()->add_policy(name, policy); } 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 this->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(); - - return manager->upstream_add_server(name, address, address_params); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_castns->get_policy(name); + if (policy) + { + policy->add_server(address, address_params); + return 0; + } + return -1; } int UpstreamManager::upstream_remove_server(const std::string& name, const std::string& address) { - auto *manager = __UpstreamManager::get_instance(); - - return manager->upstream_remove_server(name, address); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_castns->get_policy(name); + if (policy) + { + policy->remove_server(address); + return 0; + } + 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(); - - return manager->upstream_replace_server(name, address, address_params); + std::vector address; + //TODO + return address } int UpstreamManager::upstream_disable_server(const std::string& name, const std::string& address) { - auto *manager = __UpstreamManager::get_instance(); - - return manager->upstream_disable_server(name, address); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_castns->get_policy(name); + if (policy) + { + policy->disable_server(address); + return 0; + } + return -1; } int UpstreamManager::upstream_enable_server(const std::string& name, const std::string& address) { - auto *manager = __UpstreamManager::get_instance(); - - return manager->upstream_enable_server(name, address); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_castns->get_policy(name); + if (policy) + { + policy->enable_server(address); + return 0; + } + 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(); - - 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); + WFNameService *ns = WFGlobal::get_name_service(); + UPSPolicy *policy = dynamic_castns->get_policy(name); + if (policy) + { + policy->replace_server(address, address_params); + return 0; + } + return -1; } diff --git a/src/manager/UpstreamManager.h b/src/manager/UpstreamManager.h index ecc1c81a..63761215 100644 --- a/src/manager/UpstreamManager.h +++ b/src/manager/UpstreamManager.h @@ -23,6 +23,8 @@ #include #include "URIParser.h" #include "EndpointParams.h" +#include "WFGlobal.h" +#include "WFNameService.h" /** * @file UpstreamManager.h @@ -234,6 +236,14 @@ public: const std::string& address, const struct AddressParams *address_params); + UpstreamManager() + { + pthread_mutex_init(&this->lock, NULL); + } +private: + std::vector policies; + pthread_mutex_t lock; +/* public: /// @brief Internal use only class UpstreamResult @@ -267,6 +277,7 @@ public: static void notify_unavailable(void *cookie); /// @brief Internal use only static void notify_available(void *cookie); +*/ }; #endif