rename UPSPolicy and add origin upstream api

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")

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})

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;

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);

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<EndpointAddress *> mains;
@@ -60,7 +60,7 @@ public:
 	std::atomic<int> 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<GovernancePolicy *>(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<UPSPolicy *>(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<server> 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<unsigned int (const char *, const char *, const char *)>;
 
-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;

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);

src/manager/UpstreamManager.h

@@ -23,6 +23,8 @@
 #include <functional>
 #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<NSPolicy *policy> 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