From 3a14072832733d43bfec2b0be4d6704d44dd5673 Mon Sep 17 00:00:00 2001 From: holmes1412 Date: Tue, 19 Jan 2021 19:57:10 +0800 Subject: [PATCH] udpate GovernancePolicy GroupPolicy and SelectPolicy --- src/manager/GovernancePolicy.cc | 324 ++++++++++++++++++++++++++++++++ src/manager/GovernancePolicy.h | 274 +++++++++++++++++++++++++++ src/manager/GroupPolicy.cc | 265 ++++++++++++++++++++++++++ src/manager/SelectPolicy.cc | 111 +++++++++++ 4 files changed, 974 insertions(+) create mode 100644 src/manager/GovernancePolicy.cc create mode 100644 src/manager/GovernancePolicy.h create mode 100644 src/manager/GroupPolicy.cc create mode 100644 src/manager/SelectPolicy.cc diff --git a/src/manager/GovernancePolicy.cc b/src/manager/GovernancePolicy.cc new file mode 100644 index 00000000..d2ac8097 --- /dev/null +++ b/src/manager/GovernancePolicy.cc @@ -0,0 +1,324 @@ +#include "GovernancePolicy.h" +#include "StringUtil.h" + +#define DNS_CACHE_LEVEL_1 1 +#define DNS_CACHE_LEVEL_2 2 + +class WFSelectorFailTask : public WFRouterTask +{ +public: + WFSelectorFailTask(router_callback_t&& cb) + : WFRouterTask(std::move(cb)) + { + } + + virtual void dispatch() + { + this->state = WFT_STATE_TASK_ERROR; + this->error = WFT_ERR_UPSTREAM_UNAVAILABLE; + + return this->subtask_done(); + } +}; + +static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr) +{ + char *host = NULL; + char *port = NULL; + + if (!addr->host.empty()) + { + host = strdup(addr->host.c_str()); + if (!host) + return false; + } + + if (addr->port_value > 0) + { + port = strdup(addr->port.c_str()); + if (!port) + { + free(host); + return false; + } + free(uri.port); + uri.port = port; + } + + free(uri.host); + uri.host = host; + return true; +} + +EndpointAddress::EndpointAddress(const std::string& address, + const struct AddressParams *address_params) +{ + std::vector arr = StringUtil::split(address, ':'); + this->params = *address_params; + this->address = address; + + static std::hash std_hash; + for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++) + this->consistent_hash[i] = std_hash(address + "|v" + std::to_string(i)); + + if (this->params.weight == 0) + this->params.weight = 1; + + if (this->params.max_fails == 0) + this->params.max_fails = 1; + + if (this->params.group_id < 0) + this->params.group_id = -1; + + if (arr.size() == 0) + this->host = ""; + else + this->host = arr[0]; + + if (arr.size() <= 1) + { + this->port = ""; + this->port_value = 0; + } + else + { + this->port = arr[1]; + this->port_value = atoi(arr[1].c_str()); + } +} + +WFRouterTask *GovernancePolicy::create_router_task(const struct WFNSParams *params, + router_callback_t callback) +{ + EndpointAddress *addr = NULL; + WFRouterTask *task = NULL; + + if (this->select(params->uri, &addr) && copy_host_port(params->uri, addr)) + { + const auto *settings = WFGlobal::get_global_settings(); + unsigned int dns_ttl_default = settings->dns_ttl_default; + unsigned int dns_ttl_min = settings->dns_ttl_min; + const struct EndpointParams *endpoint_params = &settings->endpoint_params; + int dns_cache_level = params->retry_times == 0 ? DNS_CACHE_LEVEL_2 : + DNS_CACHE_LEVEL_1; + task = this->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min, + endpoint_params, std::move(callback)); + } + else + task = new WFSelectorFailTask(std::move(callback)); + + task->set_cookie(addr); + + return task; +} + +inline void GovernancePolicy::recover_server_from_breaker(EndpointAddress *addr) +{ + addr->fail_count = 0; + pthread_mutex_lock(&this->breaker_lock); + if (addr->list.next) + { + list_del(&addr->list); + addr->list.next = NULL; + this->recover_one_server(addr); + //this->server_list_change(); + } + pthread_mutex_unlock(&this->breaker_lock); +} + +inline void GovernancePolicy::fuse_server_to_breaker(EndpointAddress *addr) +{ + pthread_mutex_lock(&this->breaker_lock); + if (!addr->list.next) + { + addr->broken_timeout = GET_CURRENT_SECOND + MTTR_SECOND; + list_add_tail(&addr->list, &this->breaker_list); + this->fuse_one_server(addr); + //this->server_list_change(); + } + pthread_mutex_unlock(&this->breaker_lock); +} + +void GovernancePolicy::success(RouteManager::RouteResult *result, void *cookie, + CommTarget *target) +{ + pthread_rwlock_rdlock(&this->rwlock); + this->recover_server_from_breaker((EndpointAddress *)cookie); + pthread_rwlock_unlock(&this->rwlock); + + WFDNSResolver::success(result, NULL, target); +} + +void GovernancePolicy::failed(RouteManager::RouteResult *result, void *cookie, + CommTarget *target) +{ + EndpointAddress *server = (EndpointAddress *)cookie; + + pthread_rwlock_rdlock(&this->rwlock); + int fail_count = ++server->fail_count; + if (fail_count == server->params.max_fails) + this->fuse_server_to_breaker(server); + + pthread_rwlock_unlock(&this->rwlock); + + WFDNSResolver::failed(result, NULL, target); +} + +void GovernancePolicy::check_breaker() +{ + pthread_mutex_lock(&this->breaker_lock); + if (!list_empty(&this->breaker_list)) + { + int64_t cur_time = GET_CURRENT_SECOND; + struct list_head *pos, *tmp; + EndpointAddress *addr; + + list_for_each_safe(pos, tmp, &this->breaker_list) + { + addr = list_entry(pos, EndpointAddress, list); + if (cur_time >= addr->broken_timeout) + { + if (addr->fail_count >= addr->params.max_fails) + { + addr->fail_count = addr->params.max_fails - 1; + this->recover_one_server(addr); + } + list_del(pos); + addr->list.next = NULL; + } + } + } + pthread_mutex_unlock(&this->breaker_lock); + + //this->server_list_change(); +} + +const EndpointAddress *GovernancePolicy::first_stradegy(const ParsedURI& uri) +{ + unsigned int idx = rand() % this->servers.size(); + return this->servers[idx]; +} + +const EndpointAddress *GovernancePolicy::another_stradegy(const ParsedURI& uri) +{ + return this->first_stradegy(uri); +} + +bool GovernancePolicy::select(const ParsedURI& uri, EndpointAddress **addr) +{ + pthread_rwlock_rdlock(&this->rwlock); + unsigned int n = (unsigned int)this->servers.size(); + + if (n == 0) + { + pthread_rwlock_unlock(&this->rwlock); + return false; + } + + this->check_breaker(); + if (this->nalives == 0) + { + pthread_rwlock_unlock(&this->rwlock); + return false; + } + + // select_addr == NULL will only happened in consistent_hash + const EndpointAddress *select_addr = this->first_stradegy(uri); + + if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails) + { + if (this->try_another) + select_addr = this->another_stradegy(uri); + } + + pthread_rwlock_unlock(&this->rwlock); + + if (select_addr) + { + *addr = (EndpointAddress *)select_addr; + return true; + } + + return false; +} + +void GovernancePolicy::add_server(const std::string& address, + const AddressParams *address_params) +{ + 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) +{ + pthread_rwlock_wrlock(&this->rwlock); + const auto map_it = this->server_map.find(address); + if (map_it != this->server_map.cend()) + { + for (EndpointAddress *addr : map_it->second) + { + if (addr->fail_count < addr->params.max_fails) // or not: it has already been -- in nalives + this->fuse_one_server(addr); + } + + this->server_map.erase(map_it); + } + + size_t n = this->servers.size(); + size_t new_n = 0; + + for (size_t i = 0; i < n; i++) + { + if (this->servers[i]->address != address) + { + if (new_n != i) + this->servers[new_n++] = this->servers[i]; + else + new_n++; + } + } + + int ret = 0; + if (new_n < n) + { + this->servers.resize(new_n); + ret = n - new_n; + } + + pthread_rwlock_unlock(&this->rwlock); + return ret; +} + +void GovernancePolicy::enable_server(const std::string& address) +{ + pthread_rwlock_rdlock(&this->rwlock); + const auto map_it = this->server_map.find(address); + if (map_it != this->server_map.cend()) + { + for (EndpointAddress *addr : map_it->second) + this->recover_server_from_breaker(addr); + } + pthread_rwlock_unlock(&this->rwlock); +} + +void GovernancePolicy::disable_server(const std::string& address) +{ + pthread_rwlock_rdlock(&this->rwlock); + const auto map_it = this->server_map.find(address); + if (map_it != this->server_map.cend()) + { + for (EndpointAddress *addr : map_it->second) + { + addr->fail_count = addr->params.max_fails; + this->fuse_server_to_breaker(addr); + } + } + pthread_rwlock_unlock(&this->rwlock); +} + diff --git a/src/manager/GovernancePolicy.h b/src/manager/GovernancePolicy.h new file mode 100644 index 00000000..bd10faf4 --- /dev/null +++ b/src/manager/GovernancePolicy.h @@ -0,0 +1,274 @@ +#include "EndpointParams.h" +#include "WFNameService.h" +#include "WFDNSResolver.h" +#include "WFGlobal.h" +#include "WFTaskError.h" + +#include +#include + +#ifndef _GOVERNANCE_POLICY_H_ +#define _GOVERNANCE_POLICY_H_ + +#define MTTR_SECOND 30 +#define GET_CURRENT_SECOND std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count() +#define VIRTUAL_GROUP_SIZE 16 + +struct AddressParams +{ + struct EndpointParams endpoint_params; + unsigned int dns_ttl_default; + unsigned int dns_ttl_min; + unsigned int max_fails; + unsigned short weight; + int server_type; + int group_id; +}; + +class EndpointGroup; +class GovernancePolicy; +class GroupPolicy; + +class EndpointAddress +{ +public: + EndpointGroup *group; + AddressParams params; + std::string address; + std::string host; + std::string port; + short port_value; + struct list_head list; + std::atomic fail_count; + int64_t broken_timeout; + unsigned int consistent_hash[VIRTUAL_GROUP_SIZE]; + +public: + EndpointAddress(const std::string& address, + const struct AddressParams *address_params); +}; + +class EndpointGroup +{ +public: + int id; + GroupPolicy *policy; + struct rb_node rb; + pthread_mutex_t mutex; + std::vector mains; + std::vector backups; + std::atomic nalives; + int weight; + + EndpointGroup(int group_id, GroupPolicy *policy) + { + this->id = group_id; + this->policy = policy; + this->nalives = 0; + this->weight = 0; + pthread_mutex_init(&this->mutex, NULL); + } + + ~EndpointGroup() + { + pthread_mutex_destroy(&this->mutex); + } + +public: + const EndpointAddress *get_one(); + const EndpointAddress *get_one_backup(); +}; + +class GovernancePolicy : public WFDNSResolver +{ +public: + virtual WFRouterTask *create_router_task(const struct WFNSParams *params, + router_callback_t callback); + virtual void success(RouteManager::RouteResult *result, void *cookie, + CommTarget *target); + virtual void failed(RouteManager::RouteResult *result, void *cookie, + CommTarget *target); + + // 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); + + 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() + { + this->nalives = 0; + this->try_another = false; + pthread_rwlock_init(&this->rwlock, NULL); + pthread_mutex_init(&this->breaker_lock, NULL); + INIT_LIST_HEAD(&this->breaker_list); + } + + ~GovernancePolicy() + { + pthread_mutex_destroy(&this->breaker_lock); + pthread_rwlock_destroy(&this->rwlock); + for (EndpointAddress *addr : this->addresses) + delete addr; + } + +private: + virtual bool select(const ParsedURI& uri, EndpointAddress **addr); + + virtual void recover_one_server(const EndpointAddress *addr) + { + this->nalives++; + } + + virtual void fuse_one_server(const EndpointAddress *addr) + { + this->nalives--; + } + + void recover_server_from_breaker(EndpointAddress *addr); + void fuse_server_to_breaker(EndpointAddress *addr); + + struct list_head breaker_list; + pthread_mutex_t breaker_lock; + +protected: + virtual const EndpointAddress *first_stradegy(const ParsedURI& uri); + virtual const EndpointAddress *another_stradegy(const ParsedURI& uri); + void check_breaker(); + + std::vector servers; // current servers + std::vector addresses; // memory management + std::unordered_map> server_map; + pthread_rwlock_t rwlock; + std::atomic nalives; + bool try_another; +}; + +class GroupPolicy : public GovernancePolicy +{ +public: + GroupPolicy(); + ~GroupPolicy(); + 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) + { + this->nalives++; + addr->group->nalives++; + } + + virtual void fuse_one_server(const EndpointAddress *addr) + { + this->nalives--; + addr->group->nalives--; + } + // override: select() add_server() remove_server() + virtual bool select(const ParsedURI& uri, EndpointAddress **addr); + +protected: + const EndpointAddress *consistent_hash_with_group(unsigned int hash) const; + + inline const EndpointAddress *check_and_get(const EndpointAddress *addr) const // check_get_weak + { + if (addr && addr->fail_count >= addr->params.max_fails && addr->params.group_id >= 0) + { + const auto *ret = addr->group->get_one(); + + if (ret) + addr = ret; + } + return addr; + } + + inline bool is_alive_or_group_alive(const EndpointAddress *addr) const + { + return ((addr->params.group_id < 0 && addr->fail_count < addr->params.max_fails) || + (addr->params.group_id >= 0 && addr->group->nalives > 0)); + } +}; + +class WeightedRandomPolicy : public GroupPolicy +{ +public: + WeightedRandomPolicy(bool try_another) + { + this->total_weight = 0; + this->available_weight = 0; + this->try_another = try_another; + } + const EndpointAddress *first_stradegy(const ParsedURI& uri); + const EndpointAddress *another_stradegy(const ParsedURI& uri); + +protected: + int total_weight; + int available_weight; + +private: + virtual void recover_one_server(const EndpointAddress *addr); + virtual void fuse_one_server(const EndpointAddress *addr); +}; + +using select_t = std::function; + +class ConsistentHashPolicy : public GroupPolicy +{ +public: + ConsistentHashPolicy() + { + this->consistent_hash = this->default_consistent_hash; + } + + ConsistentHashPolicy(select_t consistent_hash) + { + this->consistent_hash = std::move(consistent_hash); + } + +protected: + const EndpointAddress *first_stradegy(const ParsedURI& uri); + +private: + select_t consistent_hash; + +public: + 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); + } +}; + +class ManualPolicy : public GroupPolicy +{ +public: + ManualPolicy(bool try_another, select_t select, select_t try_another_select) + { + this->try_another = try_another; + this->manual_select = select; + this->try_another_select = try_another_select; + } + + const EndpointAddress *first_stradegy(const ParsedURI& uri); + const EndpointAddress *another_stradegy(const ParsedURI& uri); + +private: + select_t manual_select; + select_t try_another_select; +}; + +#endif + diff --git a/src/manager/GroupPolicy.cc b/src/manager/GroupPolicy.cc new file mode 100644 index 00000000..550a474b --- /dev/null +++ b/src/manager/GroupPolicy.cc @@ -0,0 +1,265 @@ +#include "GovernancePolicy.h" + +#define VIRTUAL_GROUP_SIZE 16 + +GroupPolicy::GroupPolicy() +{ + this->group_map.rb_node = NULL; + this->default_group = new EndpointGroup(-1, this); + rb_link_node(&this->default_group->rb, NULL, &this->group_map.rb_node); + rb_insert_color(&this->default_group->rb, &this->group_map); +} + +GroupPolicy::~GroupPolicy() +{ + EndpointGroup *group; + + while (this->group_map.rb_node) + { + group = rb_entry(this->group_map.rb_node, EndpointGroup, rb); + rb_erase(this->group_map.rb_node, &this->group_map); + delete group; + } +} + +bool GroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr) +{ + pthread_rwlock_rdlock(&this->rwlock); + unsigned int n = (unsigned int)this->servers.size(); + + if (n == 0) + { + pthread_rwlock_unlock(&this->rwlock); + return false; + } + + this->check_breaker(); + if (this->nalives == 0) + { + pthread_rwlock_unlock(&this->rwlock); + return false; + } + + // select_addr == NULL will only happened in consistent_hash + const EndpointAddress *select_addr = this->first_stradegy(uri); + + if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails) + { + if (select_addr) + select_addr = select_addr->group->get_one(); + + if (!select_addr && this->try_another) + select_addr = this->another_stradegy(uri); + } + + if (!select_addr) + this->default_group->get_one_backup(); + + pthread_rwlock_unlock(&this->rwlock); + + if (select_addr) + { + *addr = (EndpointAddress *)select_addr; + return true; + } + + return false; +} + +const EndpointAddress *EndpointGroup::get_one() +{ + if (this->nalives == 0) + return NULL; + + const EndpointAddress *addr = NULL; + pthread_mutex_lock(&this->mutex); + + std::random_shuffle(this->mains.begin(), this->mains.end()); + for (size_t i = 0; i < this->mains.size(); i++) + { + if (this->mains[i]->fail_count < this->mains[i]->params.max_fails) + { + addr = this->mains[i]; + break; + } + } + + if (!addr) + { + std::random_shuffle(this->backups.begin(), this->backups.end()); + for (size_t i = 0; i < this->backups.size(); i++) + { + if (this->backups[i]->fail_count < this->backups[i]->params.max_fails) + { + addr = this->backups[i]; + break; + } + } + } + + pthread_mutex_unlock(&this->mutex); + return addr; +} + +const EndpointAddress *EndpointGroup::get_one_backup() +{ + if (this->nalives == 0) + return NULL; + + const EndpointAddress *addr = NULL; + pthread_mutex_lock(&this->mutex); + + std::random_shuffle(this->backups.begin(), this->backups.end()); + for (size_t i = 0; i < this->backups.size(); i++) + { + if (this->backups[i]->fail_count < this->backups[i]->params.max_fails) + { + addr = this->backups[i]; + break; + } + } + + pthread_mutex_unlock(&this->mutex); + return addr; +} + +void GroupPolicy::add_server(const std::string& address, + const AddressParams *address_params) +{ + 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); + + while (*p) + { + parent = *p; + group = rb_entry(*p, EndpointGroup, rb); + + if (group_id < group->id) + p = &(*p)->rb_left; + else if (group_id > group->id) + p = &(*p)->rb_right; + else + break; + } + + if (*p == NULL) + { + group = new EndpointGroup(group_id, this); + rb_link_node(&group->rb, parent, p); + rb_insert_color(&group->rb, &this->group_map); + } + + pthread_mutex_lock(&group->mutex); + this->recover_one_server(addr); + addr->group = group; + if (addr->params.server_type == 0) + { + group->mains.push_back(addr); + group->weight += addr->params.weight; // TODO + } + else + group->backups.push_back(addr); + pthread_mutex_unlock(&group->mutex); + + pthread_rwlock_unlock(&this->rwlock); + return; +} + +int GroupPolicy::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()) + { + for (EndpointAddress *addr : map_it->second) + { + EndpointGroup *group = addr->group; + std::vector *vec; + + if (addr->params.server_type == 0) + vec = &group->mains; + else + vec = &group->backups; + + //std::lock_guard lock(group->mutex); + pthread_mutex_lock(&group->mutex); + if (addr->fail_count < addr->params.max_fails) + this->fuse_one_server(addr); + + if (addr->params.server_type == 0) + group->weight -= addr->params.weight; // TODO + + for (auto it = vec->begin(); it != vec->end(); ++it) + { + if (*it == addr) + { + vec->erase(it); + break; + } + } + pthread_mutex_unlock(&group->mutex); + } + + this->server_map.erase(map_it); + } + + size_t n = this->servers.size(); + size_t new_n = 0; + + for (size_t i = 0; i < n; i++) + { + if (this->servers[i]->address != address) + { + if (new_n != i) + this->servers[new_n++] = this->servers[i]; + else + new_n++; + } + } + + int ret = 0; + if (new_n < n) + { + this->servers.resize(new_n); + ret = n - new_n; + } + + pthread_rwlock_unlock(&this->rwlock); + return ret; +} + +const EndpointAddress *GroupPolicy::consistent_hash_with_group(unsigned int hash) const +{ + const EndpointAddress *addr = NULL; + unsigned int min_dis = (unsigned int)-1; + + for (const EndpointAddress *server : this->servers) + { + if (this->is_alive_or_group_alive(server)) + { + for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++) + { + unsigned int dis = std::min + (hash - server->consistent_hash[i], + server->consistent_hash[i] - hash); + + if (dis < min_dis) + { + min_dis = dis; + addr = server; + } + } + } + } + + return this->check_and_get(addr); +} + diff --git a/src/manager/SelectPolicy.cc b/src/manager/SelectPolicy.cc new file mode 100644 index 00000000..e89bba42 --- /dev/null +++ b/src/manager/SelectPolicy.cc @@ -0,0 +1,111 @@ +#include "GovernancePolicy.h" + +const EndpointAddress *WeightedRandomPolicy::first_stradegy(const ParsedURI& uri) +{ + int x = 0; + int s = 0; + int idx, temp_weight; + + temp_weight = this->total_weight; + + if (temp_weight > 0) + x = rand() % temp_weight; + + for (idx = 0; idx < this->servers.size(); idx++) + { + s += this->servers[idx]->params.weight; + if (s > x) + break; + } + if (idx == this->servers.size()) + idx--; + + return this->servers[idx]; +} + +const EndpointAddress *WeightedRandomPolicy::another_stradegy(const ParsedURI& uri) +{ + int temp_weight = this->available_weight; + if (temp_weight == 0) + return NULL; + + const EndpointAddress *addr = NULL; + int x = rand() % temp_weight; + int s = 0; + + for (const EndpointAddress *server : this->servers) + { + if (this->is_alive_or_group_alive(server)) + { + addr = server; + s += server->params.weight; + if (s > x) + break; + } + } + return this->check_and_get(addr); +} + + +void WeightedRandomPolicy::recover_one_server(const EndpointAddress *addr) +{ + this->nalives++; + if (addr->group->nalives++ == 0 && addr->group->id > 0) + this->available_weight += addr->group->weight; + + if (addr->params.group_id < 0 && addr->params.server_type == 0) // TODO + this->available_weight += addr->params.weight; +} + +void WeightedRandomPolicy::fuse_one_server(const EndpointAddress *addr) +{ + this->nalives--; + if (--addr->group->nalives == 0 && addr->group->id > 0) + this->available_weight -= addr->group->weight; + + if (addr->params.group_id < 0 && addr->params.server_type == 0) // TODO + this->available_weight -= addr->params.weight; +} + +const EndpointAddress *ConsistentHashPolicy::first_stradegy(const ParsedURI& uri) +{ + unsigned int hash_value; + + if (this->consistent_hash) + hash_value = this->consistent_hash(uri.path ? uri.path : "", + uri.query ? uri.query : "", + uri.fragment ? uri.fragment : ""); + else + hash_value = this->default_consistent_hash(uri.path ? uri.path : "", + uri.query ? uri.query : "", + uri.fragment ? uri.fragment : ""); + return this->consistent_hash_with_group(hash_value); +} + +const EndpointAddress *ManualPolicy::first_stradegy(const ParsedURI& uri) +{ + int idx = this->manual_select(uri.path ? uri.path : "", + uri.query ? uri.query : "", + uri.fragment ? uri.fragment : ""); + + if (idx >= this->servers.size()) + idx %= this->servers.size(); + + return this->servers[idx]; +} + +const EndpointAddress *ManualPolicy::another_stradegy(const ParsedURI& uri) +{ + unsigned int hash_value; + + if (this->try_another_select) + hash_value = this->try_another_select(uri.path ? uri.path : "", + uri.query ? uri.query : "", + uri.fragment ? uri.fragment : ""); + else + hash_value = ConsistentHashPolicy::default_consistent_hash(uri.path ? uri.path : "", + uri.query ? uri.query : "", + uri.fragment ? uri.fragment : ""); + return this->consistent_hash_with_group(hash_value); +} +