Merge remote-tracking branch 'upstream/master'

CMakeLists_Headers.txt

@@ -66,7 +66,6 @@ set(INCLUDE_HEADERS
 	src/manager/WFFuture.h
 	src/manager/WFFacilities.h
 	src/manager/WFFacilities.inl
-	src/manager/UpstreamPolicies.h
 	src/util/EncodeStream.h
 	src/util/LRUCache.h
 	src/util/StringUtil.h
@@ -83,8 +82,10 @@ set(INCLUDE_HEADERS
 	src/factory/WFAlgoTaskFactory.inl
 	src/factory/Workflow.h
 	src/factory/WFOperator.h
-	src/factory/WFNameService.h
-	src/factory/WFDNSResolver.h
+	src/nameservice/WFNameService.h
+	src/nameservice/WFDNSResolver.h
+	src/nameservice/ServiceGovernance.h
+	src/nameservice/UpstreamPolicies.h
 )
 
 if(KAFKA STREQUAL "y")

README.md

@@ -1,4 +1,4 @@
-[中文版](README_cn.md)
+[中文版入口](README_cn.md)
 
 ## Sogou C++ Workflow
 

docs/about-service-management.md

@@ -159,7 +159,7 @@ struct EndpointParams
     int ssl_connect_timeout;
 };
 
-// In UpstreamMananger.h
+// In ServiceGovernance.h
 struct AddressParams
 {
     struct EndpointParams endpoint_params; ///< Connection config

docs/en/about-service-management.md

@@ -172,7 +172,7 @@ struct EndpointParams
     int ssl_connect_timeout;
 };
 
-// In UpstreamMananger.h
+// In ServiceGovernance.h
 struct AddressParams
 {
     struct EndpointParams endpoint_params; ///< Connection config

src/CMakeLists.txt

@@ -9,6 +9,7 @@ add_subdirectory(manager)
 add_subdirectory(algorithm)
 add_subdirectory(protocol)
 add_subdirectory(factory)
+add_subdirectory(nameservice)
 add_subdirectory(server)
 add_subdirectory(client)
 
@@ -18,6 +19,7 @@ add_dependencies(manager LINK_HEADERS)
 add_dependencies(algorithm LINK_HEADERS)
 add_dependencies(protocol LINK_HEADERS)
 add_dependencies(factory LINK_HEADERS)
+add_dependencies(nameservice LINK_HEADERS)
 add_dependencies(server LINK_HEADERS)
 add_dependencies(client LINK_HEADERS)
 
@@ -29,6 +31,7 @@ add_library(
 	$<TARGET_OBJECTS:algorithm>
 	$<TARGET_OBJECTS:protocol>
 	$<TARGET_OBJECTS:factory>
+	$<TARGET_OBJECTS:nameservice>
 	$<TARGET_OBJECTS:server>
 	$<TARGET_OBJECTS:client>
 )
@@ -86,7 +89,7 @@ install(
 
 
 install (
-	TARGETS ${CMAKE_PROJECT_NAME} EXPORT ${CMAKE_PROJECT_NAME}-targets
+	TARGETS ${PROJECT_NAME} EXPORT ${CMAKE_PROJECT_NAME}-targets
 	ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
 	LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}
 	RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}

src/factory/CMakeLists.txt

@@ -8,8 +8,6 @@ set(SRC
 	MySQLTaskImpl.cc
 	WFTaskFactory.cc
 	Workflow.cc
-	WFNameService.cc
-	WFDNSResolver.cc
 )
 
 add_library(${PROJECT_NAME} OBJECT ${SRC})
@@ -20,3 +18,4 @@ if (KAFKA STREQUAL "y")
 	)
 	add_library("factory_kafka" OBJECT ${SRC})
 endif ()
+

src/factory/WFTask.h

@@ -346,6 +346,12 @@ public:
 	int get_state() const { return this->state; }
 	int get_error() const { return this->error; }
 
+public:
+	void set_callback(std::function<void (WFTimerTask *)> cb)
+	{
+		this->callback = std::move(cb);
+	}
+
 protected:
 	virtual SubTask *done()
 	{

src/kernel/SubTask.h

@@ -65,7 +65,7 @@ public:
 		this->subtasks_nr = n;
 	}
 
-	SubTask **get_subtasks(size_t *n)
+	SubTask **get_subtasks(size_t *n) const
 	{
 		*n = this->subtasks_nr;
 		return this->subtasks;

src/manager/CMakeLists.txt

@@ -6,7 +6,6 @@ set(SRC
 	UpstreamManager.cc
 	RouteManager.cc
 	WFGlobal.cc
-	UpstreamPolicies.cc
 )
 
 add_library(${PROJECT_NAME} OBJECT ${SRC})

src/manager/UpstreamManager.cc

@@ -119,7 +119,7 @@ int UpstreamManager::upstream_add_server(const std::string& name,
 										 const AddressParams *address_params)
 {
 	WFNameService *ns = WFGlobal::get_name_service();
-	UPSPolicy *policy = dynamic_cast<UPSPolicy *>(ns->get_policy(name.c_str()));
+	UPSGroupPolicy *policy = dynamic_cast<UPSGroupPolicy *>(ns->get_policy(name.c_str()));
 
 	if (policy)
 	{
@@ -135,7 +135,7 @@ int UpstreamManager::upstream_remove_server(const std::string& name,
 											const std::string& address)
 {
 	WFNameService *ns = WFGlobal::get_name_service();
-	UPSPolicy *policy = dynamic_cast<UPSPolicy *>(ns->get_policy(name.c_str()));
+	UPSGroupPolicy *policy = dynamic_cast<UPSGroupPolicy *>(ns->get_policy(name.c_str()));
 
 	if (policy)
 	{
@@ -152,7 +152,7 @@ UpstreamManager::upstream_main_address_list(const std::string& name)
 {
 	std::vector<std::string> address;
 	WFNameService *ns = WFGlobal::get_name_service();
-	UPSPolicy *policy = dynamic_cast<UPSPolicy *>(ns->get_policy(name.c_str()));
+	UPSGroupPolicy *policy = dynamic_cast<UPSGroupPolicy *>(ns->get_policy(name.c_str()));
 
 	if (policy)
 		policy->get_main_address(address);
@@ -164,7 +164,7 @@ int UpstreamManager::upstream_disable_server(const std::string& name,
 											 const std::string& address)
 {
 	WFNameService *ns = WFGlobal::get_name_service();
-	UPSPolicy *policy = dynamic_cast<UPSPolicy *>(ns->get_policy(name.c_str()));
+	UPSGroupPolicy *policy = dynamic_cast<UPSGroupPolicy *>(ns->get_policy(name.c_str()));
 
 	if (policy)
 	{
@@ -180,7 +180,7 @@ int UpstreamManager::upstream_enable_server(const std::string& name,
 											const std::string& address)
 {
 	WFNameService *ns = WFGlobal::get_name_service();
-	UPSPolicy *policy = dynamic_cast<UPSPolicy *>(ns->get_policy(name.c_str()));
+	UPSGroupPolicy *policy = dynamic_cast<UPSGroupPolicy *>(ns->get_policy(name.c_str()));
 
 	if (policy)
 	{
@@ -197,7 +197,7 @@ int UpstreamManager::upstream_replace_server(const std::string& name,
 											 const struct AddressParams *address_params)
 {
 	WFNameService *ns = WFGlobal::get_name_service();
-	UPSPolicy *policy = dynamic_cast<UPSPolicy *>(ns->get_policy(name.c_str()));
+	UPSGroupPolicy *policy = dynamic_cast<UPSGroupPolicy *>(ns->get_policy(name.c_str()));
 
 	if (policy)
 	{

src/manager/UpstreamManager.h

@@ -23,6 +23,8 @@
 #include <functional>
 #include "URIParser.h"
 #include "EndpointParams.h"
+#include "ServiceGovernance.h"
+#include "UpstreamPolicies.h"
 #include "WFGlobal.h"
 
 /**
@@ -34,50 +36,6 @@
  * - This is working only for the current process.
  */
 
-/**
- * @brief   Functional for consistent-hash OR manual-select
- * @details
- * - path/query/fragment is empty string when uri not contain that region
- * - path/query/fragment would never be NULL
- */
-using upstream_route_t = std::function<unsigned int (const char *, const char *, const char *)>;
-
-/**
- * @brief   Server config for upstream
- * @details
- * When call UpstreamManager::upstream_add_server, you can set custom config for each target
-*/
-struct AddressParams
-{
-	struct EndpointParams endpoint_params; ///< Connection config
-	unsigned int dns_ttl_default;          ///< in seconds, DNS TTL when network request success
-	unsigned int dns_ttl_min;              ///< in seconds, DNS TTL when network request fail
-/**
- * - The max_fails directive sets the number of consecutive unsuccessful attempts to communicate with the server.
- * - After 30s following the server failure, upstream probe the server with some live client’s requests.
- * - If the probes have been successful, the server is marked as a live one.
- * - If max_fails is set to 1, it means server would out of upstream selection in 30 seconds when failed only once
- */
-	unsigned int max_fails;                ///< [1, INT32_MAX] max_fails = 0 means max_fails = 1
-	unsigned short weight;                 ///< [1, 65535] weight = 0 means weight = 1. only for main server
-	int server_type;                       ///< 0 for main and 1 for backup
-	int group_id;                          ///< -1 means no group. Backup without group will be backup for any main
-};
-
-/**
- * @brief   Default server config for upstream
- */
-static constexpr struct AddressParams ADDRESS_PARAMS_DEFAULT =
-{
-	.endpoint_params	=	ENDPOINT_PARAMS_DEFAULT,
-	.dns_ttl_default	=	12 * 3600,
-	.dns_ttl_min		=	180,
-	.max_fails			=	200,
-	.weight				=	1,
-	.server_type		=	0,	/* 0 for main and 1 for backup. */
-	.group_id			=	-1,
-};
-
 /**
  * @brief   Upstream Management Class
  * @details

src/nameservice/CMakeLists.txt

@@ -0,0 +1,12 @@
+cmake_minimum_required(VERSION 3.6)
+project(nameservice)
+
+set(SRC
+	WFNameService.cc
+	WFDNSResolver.cc
+	ServiceGovernance.cc
+	UpstreamPolicies.cc
+)
+
+add_library(${PROJECT_NAME} OBJECT ${SRC})
+

src/nameservice/ServiceGovernance.cc

@@ -0,0 +1,398 @@
+/*
+  Copyright (c) 2021 Sogou, Inc.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+  Authors: Li Yingxin (liyingxin@sogou-inc.com)
+*/
+
+#include <vector>
+#include "URIParser.h"
+#include "StringUtil.h"
+#include "WFNameService.h"
+#include "WFDNSResolver.h"
+#include "ServiceGovernance.h"
+#include "UpstreamManager.h"
+
+#define GET_CURRENT_SECOND  std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now().time_since_epoch()).count()
+
+#define DNS_CACHE_LEVEL_1		1
+#define DNS_CACHE_LEVEL_2		2
+
+PolicyAddrParams::PolicyAddrParams()
+{
+	const struct AddressParams *params = &ADDRESS_PARAMS_DEFAULT;
+	this->endpoint_params = params->endpoint_params;
+	this->dns_ttl_default = params->dns_ttl_default;
+	this->dns_ttl_min = params->dns_ttl_min;
+	this->max_fails = params->max_fails;
+}
+
+PolicyAddrParams::PolicyAddrParams(const struct AddressParams *params) :
+	endpoint_params(params->endpoint_params)
+{
+	this->dns_ttl_default = params->dns_ttl_default;
+	this->dns_ttl_min = params->dns_ttl_min;
+	this->max_fails = params->max_fails;
+}
+
+class WFSelectorFailTask : public WFRouterTask
+{
+public:
+	WFSelectorFailTask(router_callback_t&& cb) :
+		WFRouterTask(std::move(cb))
+	{
+	}
+
+	virtual void dispatch()
+	{
+		this->state = WFT_STATE_TASK_ERROR;
+		this->error = WFT_ERR_UPSTREAM_UNAVAILABLE;
+
+		return this->subtask_done();
+	}
+};
+
+static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr)
+{
+	char *host = NULL;
+	char *port = NULL;
+
+	if (!addr->host.empty())
+	{
+		host = strdup(addr->host.c_str());
+		if (!host)
+			return false;
+	}
+
+	if (!addr->port.empty())
+	{
+		port = strdup(addr->port.c_str());
+		if (!port)
+		{
+			free(host);
+			return false;
+		}
+		free(uri.port);
+		uri.port = port;
+	}
+
+	free(uri.host);
+	uri.host = host;
+	return true;
+}
+
+EndpointAddress::EndpointAddress(const std::string& address,
+								 PolicyAddrParams *address_params)
+{
+	std::vector<std::string> arr = StringUtil::split(address, ':');
+
+	this->params = address_params;
+	if (this->params->max_fails == 0)
+		this->params->max_fails = 1;
+
+	this->address = address;
+	this->fail_count = 0;
+	this->entry.list.next = NULL;
+	this->entry.ptr = this;
+
+	if (arr.size() == 0)
+		this->host = "";
+	else
+		this->host = arr[0];
+
+	if (arr.size() <= 1)
+		this->port = "";
+	else
+		this->port = arr[1];
+}
+
+WFRouterTask *ServiceGovernance::create_router_task(const struct WFNSParams *params,
+													router_callback_t callback)
+{
+	EndpointAddress *addr;
+	WFRouterTask *task;
+
+	if (this->select(params->uri, &addr) && copy_host_port(params->uri, addr))
+	{
+		unsigned int dns_ttl_default = addr->params->dns_ttl_default;
+		unsigned int dns_ttl_min = addr->params->dns_ttl_min;
+		const struct EndpointParams *endpoint_params = &addr->params->endpoint_params;
+		int dns_cache_level = params->retry_times == 0 ? DNS_CACHE_LEVEL_2 :
+														 DNS_CACHE_LEVEL_1;
+		task = this->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min,
+							endpoint_params, std::move(callback));
+		task->set_cookie(addr);
+	}
+	else
+		task = new WFSelectorFailTask(std::move(callback));
+
+	return task;
+}
+
+inline void ServiceGovernance::recover_server_from_breaker(EndpointAddress *addr)
+{
+	addr->fail_count = 0;
+	pthread_mutex_lock(&this->breaker_lock);
+	if (addr->entry.list.next)
+	{
+		list_del(&addr->entry.list);
+		addr->entry.list.next = NULL;
+		this->recover_one_server(addr);
+		this->server_list_change(addr, RECOVER_SERVER);
+	}
+	pthread_mutex_unlock(&this->breaker_lock);
+}
+
+inline void ServiceGovernance::fuse_server_to_breaker(EndpointAddress *addr)
+{
+	pthread_mutex_lock(&this->breaker_lock);
+	if (!addr->entry.list.next)
+	{
+		addr->broken_timeout = GET_CURRENT_SECOND + this->mttr_second;
+		list_add_tail(&addr->entry.list, &this->breaker_list);
+		this->fuse_one_server(addr);
+		this->server_list_change(addr, FUSE_SERVER);
+	}
+	pthread_mutex_unlock(&this->breaker_lock);
+}
+
+void ServiceGovernance::success(RouteManager::RouteResult *result, void *cookie,
+								CommTarget *target)
+{
+	pthread_rwlock_rdlock(&this->rwlock);
+	this->recover_server_from_breaker((EndpointAddress *)cookie);
+	pthread_rwlock_unlock(&this->rwlock);
+
+	WFDNSResolver::success(result, NULL, target);
+}
+
+void ServiceGovernance::failed(RouteManager::RouteResult *result, void *cookie,
+							   CommTarget *target)
+{
+	EndpointAddress *server = (EndpointAddress *)cookie;
+
+	pthread_rwlock_rdlock(&this->rwlock);
+	size_t fail_count = ++server->fail_count;
+	if (fail_count == server->params->max_fails)
+		this->fuse_server_to_breaker(server);
+
+	pthread_rwlock_unlock(&this->rwlock);
+
+	WFDNSResolver::failed(result, NULL, target);
+}
+
+void ServiceGovernance::check_breaker()
+{
+	pthread_mutex_lock(&this->breaker_lock);
+	if (!list_empty(&this->breaker_list))
+	{
+		int64_t cur_time = GET_CURRENT_SECOND;
+		struct list_head *pos, *tmp;
+		struct EndpointAddress::address_entry *entry;
+		EndpointAddress *addr;
+
+		list_for_each_safe(pos, tmp, &this->breaker_list)
+		{
+			entry = list_entry(pos, struct EndpointAddress::address_entry,
+							   list);
+			addr = entry->ptr;
+
+			if (cur_time >= addr->broken_timeout)
+			{
+				if (addr->fail_count >= addr->params->max_fails)
+				{
+					addr->fail_count = addr->params->max_fails - 1;
+					this->recover_one_server(addr);
+					this->server_list_change(addr, RECOVER_SERVER);
+				}
+				list_del(pos);
+				addr->entry.list.next = NULL;
+			}
+		}
+	}
+	pthread_mutex_unlock(&this->breaker_lock);
+}
+
+const EndpointAddress *ServiceGovernance::first_stradegy(const ParsedURI& uri)
+{
+	unsigned int idx = rand() % this->servers.size();
+	return this->servers[idx];
+}
+
+const EndpointAddress *ServiceGovernance::another_stradegy(const ParsedURI& uri)
+{
+	return this->first_stradegy(uri);
+}
+
+bool ServiceGovernance::select(const ParsedURI& uri, EndpointAddress **addr)
+{
+	pthread_rwlock_rdlock(&this->rwlock);
+	unsigned int n = (unsigned int)this->servers.size();
+
+	if (n == 0)
+	{
+		pthread_rwlock_unlock(&this->rwlock);
+		return false;
+	}
+
+	this->check_breaker();
+	if (this->nalives == 0)
+	{
+		pthread_rwlock_unlock(&this->rwlock);
+		return false;
+	}
+
+	// select_addr == NULL will only happened in consistent_hash
+	const EndpointAddress *select_addr = this->first_stradegy(uri);
+
+	if (!select_addr ||
+		select_addr->fail_count >= select_addr->params->max_fails)
+	{
+		if (this->try_another)
+			select_addr = this->another_stradegy(uri);
+	}
+
+	pthread_rwlock_unlock(&this->rwlock);
+
+	if (select_addr)
+	{
+		*addr = (EndpointAddress *)select_addr;
+		return true;
+	}
+
+	return false;
+}
+
+void ServiceGovernance::add_server_locked(EndpointAddress *addr)
+{
+	this->addresses.push_back(addr);
+	this->server_map[addr->address].push_back(addr);
+	this->servers.push_back(addr);
+	this->recover_one_server(addr);
+	this->server_list_change(addr, ADD_SERVER);
+}
+
+int ServiceGovernance::remove_server_locked(const std::string& address)
+{
+	const auto map_it = this->server_map.find(address);
+	if (map_it != this->server_map.cend())
+	{
+		for (EndpointAddress *addr : map_it->second)
+		{
+			// or not: it has already been -- in nalives
+			if (addr->fail_count < addr->params->max_fails)
+			{
+				this->fuse_one_server(addr);
+				this->server_list_change(addr, REMOVE_SERVER);
+			}
+		}
+
+		this->server_map.erase(map_it);
+	}
+
+	size_t n = this->servers.size();
+	size_t new_n = 0;
+
+	for (size_t i = 0; i < n; i++)
+	{
+		if (this->servers[i]->address != address)
+		{
+			if (new_n != i)
+				this->servers[new_n++] = this->servers[i];
+			else
+				new_n++;
+		}
+	}
+
+	int ret = 0;
+	if (new_n < n)
+	{
+		this->servers.resize(new_n);
+		ret = n - new_n;
+	}
+
+	return ret;
+}
+
+void ServiceGovernance::add_server(const std::string& address,
+								   const AddressParams *params)
+{
+	EndpointAddress *addr = new EndpointAddress(address,
+									new PolicyAddrParams(params));
+
+	pthread_rwlock_wrlock(&this->rwlock);
+	this->add_server_locked(addr);
+	pthread_rwlock_unlock(&this->rwlock);
+}
+
+int ServiceGovernance::remove_server(const std::string& address)
+{
+	int ret;
+	pthread_rwlock_wrlock(&this->rwlock);
+	ret = this->remove_server_locked(address);
+	pthread_rwlock_unlock(&this->rwlock);
+	return ret;
+}
+
+int ServiceGovernance::replace_server(const std::string& address,
+									  const AddressParams *params)
+{
+	int ret;
+	EndpointAddress *addr = new EndpointAddress(address,
+									new PolicyAddrParams(params));
+
+	pthread_rwlock_wrlock(&this->rwlock);
+	this->add_server_locked(addr);
+	ret = this->remove_server_locked(address);
+	pthread_rwlock_unlock(&this->rwlock);
+	return ret;
+}
+
+void ServiceGovernance::enable_server(const std::string& address)
+{
+	pthread_rwlock_rdlock(&this->rwlock);
+	const auto map_it = this->server_map.find(address);
+	if (map_it != this->server_map.cend())
+	{
+		for (EndpointAddress *addr : map_it->second)
+			this->recover_server_from_breaker(addr);
+	}
+	pthread_rwlock_unlock(&this->rwlock);
+}
+
+void ServiceGovernance::disable_server(const std::string& address)
+{
+	pthread_rwlock_rdlock(&this->rwlock);
+	const auto map_it = this->server_map.find(address);
+	if (map_it != this->server_map.cend())
+	{
+		for (EndpointAddress *addr : map_it->second)
+		{
+			addr->fail_count = addr->params->max_fails;
+			this->fuse_server_to_breaker(addr);
+		}
+	}
+	pthread_rwlock_unlock(&this->rwlock);
+}
+
+void ServiceGovernance::get_current_address(std::vector<std::string>& addr_list)
+{
+	pthread_rwlock_rdlock(&this->rwlock);
+
+	for (const EndpointAddress *server : this->servers)
+		addr_list.push_back(server->address);
+
+	pthread_rwlock_unlock(&this->rwlock);
+}
+

src/nameservice/ServiceGovernance.h

@@ -0,0 +1,183 @@
+/*
+  Copyright (c) 2021 Sogou, Inc.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+  Authors: Li Yingxin (liyingxin@sogou-inc.com)
+*/
+
+#ifndef _SERVICE_GOVERNANCE_H_
+#define _SERVICE_GOVERNANCE_H_
+
+#include <pthread.h>
+#include <unordered_map>
+#include <vector>
+#include <atomic>
+#include "URIParser.h"
+#include "EndpointParams.h"
+#include "WFNameService.h"
+#include "WFDNSResolver.h"
+#include "WFGlobal.h"
+#include "WFTaskError.h"
+
+#define MTTR_SECOND_DEFAULT 30
+#define VIRTUAL_GROUP_SIZE  16
+
+struct AddressParams
+{
+	struct EndpointParams endpoint_params; ///< Connection config
+	unsigned int dns_ttl_default;          ///< in seconds, DNS TTL when network request success
+	unsigned int dns_ttl_min;              ///< in seconds, DNS TTL when network request fail
+/**
+ * - The max_fails directive sets the number of consecutive unsuccessful attempts to communicate with the server.
+ * - After 30s following the server failure, upstream probe the server with some live client`s requests.
+ * - If the probes have been successful, the server is marked as a live one.
+ * - If max_fails is set to 1, it means server would out of upstream selection in 30 seconds when failed only once
+ */
+	unsigned int max_fails;                ///< [1, INT32_MAX] max_fails = 0 means max_fails = 1
+	unsigned short weight;                 ///< [1, 65535] weight = 0 means weight = 1. only for main server
+	int server_type;                       ///< 0 for main and 1 for backup
+	int group_id;                          ///< -1 means no group. Backup without group will be backup for any main
+};
+
+static constexpr struct AddressParams ADDRESS_PARAMS_DEFAULT =
+{
+	.endpoint_params	=	ENDPOINT_PARAMS_DEFAULT,
+	.dns_ttl_default	=	12 * 3600,
+	.dns_ttl_min		=	180,
+	.max_fails			=	200,
+	.weight				=	1,
+	.server_type		=	0,	/* 0 for main and 1 for backup. */
+	.group_id			=	-1,
+};
+
+enum ServerChangeState
+{
+	ADD_SERVER		=	0,
+	REMOVE_SERVER	=	1,
+	RECOVER_SERVER	=	2,
+	FUSE_SERVER		=	3,
+};
+
+class PolicyAddrParams
+{
+public:
+	struct EndpointParams endpoint_params;
+	unsigned int dns_ttl_default;
+	unsigned int dns_ttl_min;
+	unsigned int max_fails;
+
+	PolicyAddrParams();
+	PolicyAddrParams(const struct AddressParams *params);
+};
+
+class EndpointAddress
+{
+public:
+	std::string address;
+	std::string host;
+	std::string port;
+	std::atomic<unsigned int> fail_count;
+	long long broken_timeout;
+	PolicyAddrParams *params;
+
+	struct address_entry
+	{
+		struct list_head list;
+		EndpointAddress *ptr;
+	} entry;
+
+public:
+	EndpointAddress(const std::string& address, PolicyAddrParams *params);
+	virtual ~EndpointAddress() { delete this->params; }
+};
+
+class ServiceGovernance : public WFDNSResolver
+{
+public:
+	virtual WFRouterTask *create_router_task(const struct WFNSParams *params,
+											 router_callback_t callback);
+	virtual void success(RouteManager::RouteResult *result, void *cookie,
+					 	 CommTarget *target);
+	virtual void failed(RouteManager::RouteResult *result, void *cookie,
+						CommTarget *target);
+
+	virtual void add_server(const std::string& address,
+							const struct AddressParams *params);
+	int remove_server(const std::string& address);
+	virtual int replace_server(const std::string& address,
+							   const struct AddressParams *params);
+
+	void enable_server(const std::string& address);
+	void disable_server(const std::string& address);
+	virtual void get_current_address(std::vector<std::string>& addr_list);
+	virtual void server_list_change(const EndpointAddress *address, int state)
+	{}
+	void set_mttr_second(unsigned int second) { this->mttr_second = second; }
+
+public:
+	ServiceGovernance() :
+		breaker_lock(PTHREAD_MUTEX_INITIALIZER),
+		rwlock(PTHREAD_RWLOCK_INITIALIZER)
+	{
+		this->nalives = 0;
+		this->try_another = false;
+		this->mttr_second = MTTR_SECOND_DEFAULT;
+		INIT_LIST_HEAD(&this->breaker_list);
+	}
+
+	virtual ~ServiceGovernance()
+	{
+		for (EndpointAddress *addr : this->addresses)
+			delete addr;
+	}
+
+private:
+	virtual bool select(const ParsedURI& uri, EndpointAddress **addr);
+
+	virtual void recover_one_server(const EndpointAddress *addr)
+	{
+		this->nalives++;
+	}
+
+	virtual void fuse_one_server(const EndpointAddress *addr)
+	{
+		this->nalives--;
+	}
+
+	virtual void add_server_locked(EndpointAddress *addr);
+	virtual int remove_server_locked(const std::string& address);
+
+	void recover_server_from_breaker(EndpointAddress *addr);
+	void fuse_server_to_breaker(EndpointAddress *addr);
+
+	struct list_head breaker_list;
+	pthread_mutex_t breaker_lock;
+	unsigned int mttr_second;
+
+protected:
+	virtual const EndpointAddress *first_stradegy(const ParsedURI& uri);
+	virtual const EndpointAddress *another_stradegy(const ParsedURI& uri);
+	void check_breaker();
+
+	std::vector<EndpointAddress *> servers; // current servers
+	std::vector<EndpointAddress *> addresses; // memory management
+	std::unordered_map<std::string,
+					   std::vector<EndpointAddress *>> server_map;
+	pthread_rwlock_t rwlock;
+	std::atomic<int> nalives;
+	bool try_another;
+};
+
+#endif
+

src/nameservice/UpstreamPolicies.cc

@@ -21,361 +21,43 @@
 #include "URIParser.h"
 #include "StringUtil.h"
 #include "UpstreamPolicies.h"
-#include "WFDNSResolver.h"
 
-#define DNS_CACHE_LEVEL_1		1
-#define DNS_CACHE_LEVEL_2		2
-
-class WFSelectorFailTask : public WFRouterTask
+UPSAddrParams::UPSAddrParams() :
+	PolicyAddrParams(&ADDRESS_PARAMS_DEFAULT)
 {
-public:
-	WFSelectorFailTask(router_callback_t&& cb)
-		: WFRouterTask(std::move(cb))
-	{
-	}
-
-	virtual void dispatch()
-	{
-		this->state = WFT_STATE_TASK_ERROR;
-		this->error = WFT_ERR_UPSTREAM_UNAVAILABLE;
-
-		return this->subtask_done();
-	}
-};
-
-static bool copy_host_port(ParsedURI& uri, const EndpointAddress *addr)
-{
-	char *host = NULL;
-	char *port = NULL;
-
-	if (!addr->host.empty())
-	{
-		host = strdup(addr->host.c_str());
-		if (!host)
-			return false;
-	}
-
-	if (addr->port_value > 0)
-	{
-		port = strdup(addr->port.c_str());
-		if (!port)
-		{
-			free(host);
-			return false;
-		}
-		free(uri.port);
-		uri.port = port;
-	}
-
-	free(uri.host);
-	uri.host = host;
-	return true;
+	UPSAddrParams(&ADDRESS_PARAMS_DEFAULT, "");
 }
 
-EndpointAddress::EndpointAddress(const std::string& address,
-								 const struct AddressParams *address_params)
+UPSAddrParams::UPSAddrParams(const struct AddressParams *params,
+							 const std::string& address) :
+	PolicyAddrParams(params)
 {
-	std::vector<std::string> arr = StringUtil::split(address, ':');
-	this->params = *address_params;
-	this->address = address;
-	this->list.next = NULL;
-	this->fail_count = 0;
-
 	static std::hash<std::string> std_hash;
 	for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++)
 		this->consistent_hash[i] = std_hash(address + "|v" + std::to_string(i));
 
-	if (this->params.weight == 0)
-		this->params.weight = 1;
+	this->weight = params->weight;
+	this->server_type = params->server_type;
+	this->group_id = params->group_id;
 
-	if (this->params.max_fails == 0)
-		this->params.max_fails = 1;
+	if (this->group_id < 0)
+		this->group_id = -1;
 
-	if (this->params.group_id < 0)
-		this->params.group_id = -1;
-
-	if (arr.size() == 0)
-		this->host = "";
-	else
-		this->host = arr[0];
-
-	if (arr.size() <= 1)
-	{
-		this->port = "";
-		this->port_value = 0;
-	}
-	else
-	{
-		this->port = arr[1];
-		this->port_value = atoi(arr[1].c_str());
-	}
+	if (this->weight == 0)
+		this->weight = 1;
 }
 
-WFRouterTask *UPSPolicy::create_router_task(const struct WFNSParams *params,
-											router_callback_t callback)
-{
-	EndpointAddress *addr;
-	WFRouterTask *task;
-
-	if (this->select(params->uri, &addr) && copy_host_port(params->uri, addr))
-	{
-		unsigned int dns_ttl_default = addr->params.dns_ttl_default;
-		unsigned int dns_ttl_min = addr->params.dns_ttl_min;
-		const struct EndpointParams *endpoint_params = &addr->params.endpoint_params;
-		int dns_cache_level = params->retry_times == 0 ? DNS_CACHE_LEVEL_2 :
-														 DNS_CACHE_LEVEL_1;
-		task = this->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min,
-							endpoint_params, std::move(callback));
-		task->set_cookie(addr);
-	}
-	else
-		task = new WFSelectorFailTask(std::move(callback));
-
-	return task;
-}
-
-inline void UPSPolicy::recover_server_from_breaker(EndpointAddress *addr)
-{
-	addr->fail_count = 0;
-	pthread_mutex_lock(&this->breaker_lock);
-	if (addr->list.next)
-	{
-		list_del(&addr->list);
-		addr->list.next = NULL;
-		this->recover_one_server(addr);
-		//this->server_list_change();
-	}
-	pthread_mutex_unlock(&this->breaker_lock);
-}
-
-inline void UPSPolicy::fuse_server_to_breaker(EndpointAddress *addr)
-{
-	pthread_mutex_lock(&this->breaker_lock);
-	if (!addr->list.next)
-	{
-		addr->broken_timeout = GET_CURRENT_SECOND + MTTR_SECOND;
-		list_add_tail(&addr->list, &this->breaker_list);
-		this->fuse_one_server(addr);
-		//this->server_list_change();
-	}
-	pthread_mutex_unlock(&this->breaker_lock);
-}
-
-void UPSPolicy::success(RouteManager::RouteResult *result, void *cookie,
-					 		   CommTarget *target)
-{
-	pthread_rwlock_rdlock(&this->rwlock);
-	this->recover_server_from_breaker((EndpointAddress *)cookie);
-	pthread_rwlock_unlock(&this->rwlock);
-
-	WFDNSResolver::success(result, NULL, target);
-}
-
-void UPSPolicy::failed(RouteManager::RouteResult *result, void *cookie,
-							  CommTarget *target)
-{
-	EndpointAddress *server = (EndpointAddress *)cookie;
-
-	pthread_rwlock_rdlock(&this->rwlock);
-	size_t fail_count = ++server->fail_count;
-	if (fail_count == server->params.max_fails)
-		this->fuse_server_to_breaker(server);
-
-	pthread_rwlock_unlock(&this->rwlock);
-
-	WFDNSResolver::failed(result, NULL, target);
-}
-
-void UPSPolicy::check_breaker()
-{
-	pthread_mutex_lock(&this->breaker_lock);
-	if (!list_empty(&this->breaker_list))
-	{
-		int64_t cur_time = GET_CURRENT_SECOND;
-		struct list_head *pos, *tmp;
-		EndpointAddress *addr;
-
-		list_for_each_safe(pos, tmp, &this->breaker_list)
-		{
-			addr = list_entry(pos, EndpointAddress, list);
-			if (cur_time >= addr->broken_timeout)
-			{
-				if (addr->fail_count >= addr->params.max_fails)
-				{
-					addr->fail_count = addr->params.max_fails - 1;
-					this->recover_one_server(addr);
-				}
-				list_del(pos);
-				addr->list.next = NULL;
-			}
-		}
-	}
-	pthread_mutex_unlock(&this->breaker_lock);
-	
-	//this->server_list_change();
-}
-
-const EndpointAddress *UPSPolicy::first_stradegy(const ParsedURI& uri)
-{
-	unsigned int idx = rand() % this->servers.size();
-	return this->servers[idx];
-}
-
-const EndpointAddress *UPSPolicy::another_stradegy(const ParsedURI& uri)
-{
-	return this->first_stradegy(uri);
-}
-
-bool UPSPolicy::select(const ParsedURI& uri, EndpointAddress **addr)
-{
-	pthread_rwlock_rdlock(&this->rwlock);
-	unsigned int n = (unsigned int)this->servers.size();
-
-	if (n == 0)
-	{
-		pthread_rwlock_unlock(&this->rwlock);
-		return false;
-	}
-
-	this->check_breaker();
-	if (this->nalives == 0)
-	{
-		pthread_rwlock_unlock(&this->rwlock);
-		return false;
-	}
-
-	// select_addr == NULL will only happened in consistent_hash
-	const EndpointAddress *select_addr = this->first_stradegy(uri);
-
-	if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails)
-	{
-		if (this->try_another)
-			select_addr = this->another_stradegy(uri);
-	}
-
-	pthread_rwlock_unlock(&this->rwlock);
-
-	if (select_addr)
-	{
-		*addr = (EndpointAddress *)select_addr;
-		return true;
-	}
-
-	return false;
-}
-
-void UPSPolicy::add_server_locked(EndpointAddress *addr)
-{
-	this->addresses.push_back(addr);
-	this->server_map[addr->address].push_back(addr);
-	this->servers.push_back(addr);
-	this->recover_one_server(addr);
-}
-
-int UPSPolicy::remove_server_locked(const std::string& address)
-{
-	const auto map_it = this->server_map.find(address);
-	if (map_it != this->server_map.cend())
-	{
-		for (EndpointAddress *addr : map_it->second)
-		{
-			// or not: it has already been -- in nalives
-			if (addr->fail_count < addr->params.max_fails)
-				this->fuse_one_server(addr);
-		}
-
-		this->server_map.erase(map_it);
-	}
-
-	size_t n = this->servers.size();
-	size_t new_n = 0;
-
-	for (size_t i = 0; i < n; i++)
-	{
-		if (this->servers[i]->address != address)
-		{
-			if (new_n != i)
-				this->servers[new_n++] = this->servers[i];
-			else
-				new_n++;
-		}
-	}
-
-	int ret = 0;
-	if (new_n < n)
-	{
-		this->servers.resize(new_n);
-		ret = n - new_n;
-	}
-
-	return ret;
-}
-
-void UPSPolicy::add_server(const std::string& address,
-						   const AddressParams *address_params)
-{
-	EndpointAddress *addr = new EndpointAddress(address, address_params);
-
-	pthread_rwlock_wrlock(&this->rwlock);
-	this->add_server_locked(addr);
-	pthread_rwlock_unlock(&this->rwlock);
-}
-
-int UPSPolicy::remove_server(const std::string& address)
-{
-	int ret;
-	pthread_rwlock_wrlock(&this->rwlock);
-	ret = this->remove_server_locked(address);
-	pthread_rwlock_unlock(&this->rwlock);
-	return ret;
-}
-
-int UPSPolicy::replace_server(const std::string& address,
-							  const AddressParams *address_params)
-{
-	int ret;
-	EndpointAddress *addr = new EndpointAddress(address, address_params);
-
-	pthread_rwlock_wrlock(&this->rwlock);
-	this->add_server_locked(addr);
-	ret = this->remove_server_locked(address);
-	pthread_rwlock_unlock(&this->rwlock);
-	return ret;
-}
-
-void UPSPolicy::enable_server(const std::string& address)
-{
-	pthread_rwlock_rdlock(&this->rwlock);
-	const auto map_it = this->server_map.find(address);
-	if (map_it != this->server_map.cend())
-	{
-		for (EndpointAddress *addr : map_it->second)
-			this->recover_server_from_breaker(addr);
-	}
-	pthread_rwlock_unlock(&this->rwlock);
-}
-
-void UPSPolicy::disable_server(const std::string& address)
-{
-	pthread_rwlock_rdlock(&this->rwlock);
-	const auto map_it = this->server_map.find(address);
-	if (map_it != this->server_map.cend())
-	{
-		for (EndpointAddress *addr : map_it->second)
-		{
-			addr->fail_count = addr->params.max_fails;
-			this->fuse_server_to_breaker(addr);
-		}
-	}
-	pthread_rwlock_unlock(&this->rwlock);
-}
-
-void UPSPolicy::get_main_address(std::vector<std::string>& addr_list)
+void UPSGroupPolicy::get_main_address(std::vector<std::string>& addr_list)
 {
+	UPSAddrParams *params;
 	pthread_rwlock_rdlock(&this->rwlock);
 
 	for (const EndpointAddress *server : this->servers)
-		addr_list.push_back(server->address);
+	{
+		params = static_cast<UPSAddrParams *>(server->params);
+		if (params->server_type == 0)
+			addr_list.push_back(server->address);
+	}
 
 	pthread_rwlock_unlock(&this->rwlock);
 }
@@ -400,6 +82,31 @@ UPSGroupPolicy::~UPSGroupPolicy()
     }
 }
 
+void UPSGroupPolicy::add_server(const std::string& address,
+								const AddressParams *params)
+{
+	EndpointAddress *addr = new EndpointAddress(address,
+									new UPSAddrParams(params, address));
+
+	pthread_rwlock_wrlock(&this->rwlock);
+	this->add_server_locked(addr);
+	pthread_rwlock_unlock(&this->rwlock);
+}
+
+int UPSGroupPolicy::replace_server(const std::string& address,
+								   const AddressParams *params)
+{
+	int ret;
+	EndpointAddress *addr = new EndpointAddress(address,
+									new UPSAddrParams(params, address));
+
+	pthread_rwlock_wrlock(&this->rwlock);
+	this->add_server_locked(addr);
+	ret = this->remove_server_locked(address);
+	pthread_rwlock_unlock(&this->rwlock);
+	return ret;
+}
+
 bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr)
 {
 	pthread_rwlock_rdlock(&this->rwlock);
@@ -421,7 +128,7 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr)
 	// select_addr == NULL will only happened in consistent_hash
 	const EndpointAddress *select_addr = this->first_stradegy(uri);
 
-	if (!select_addr || select_addr->fail_count >= select_addr->params.max_fails)
+	if (!select_addr || select_addr->fail_count >= select_addr->params->max_fails)
 	{
 		if (select_addr)
 			select_addr = this->check_and_get(select_addr, true);
@@ -452,18 +159,19 @@ bool UPSGroupPolicy::select(const ParsedURI& uri, EndpointAddress **addr)
 inline const EndpointAddress *UPSGroupPolicy::check_and_get(const EndpointAddress *addr,
 															bool flag)
 {
-	if (flag == true) // && addr->fail_count >= addr->params.max_fails
+	UPSAddrParams *params = static_cast<UPSAddrParams *>(addr->params);
+	if (flag == true) // && addr->fail_count >= addr->params->max_fails
 	{
-		if (addr->params.group_id == -1)
+		if (params->group_id == -1)
 			return NULL;
 
-		return addr->group->get_one();
+		return params->group->get_one();
 	}
 
-	if (addr && addr->fail_count >= addr->params.max_fails &&
-		addr->params.group_id >= 0)
+	if (addr && addr->fail_count >= addr->params->max_fails &&
+		params->group_id >= 0)
 	{
-		const EndpointAddress *tmp = addr->group->get_one();
+		const EndpointAddress *tmp = params->group->get_one();
 		if (tmp)
 			addr = tmp;
 	}
@@ -482,7 +190,7 @@ const EndpointAddress *EndpointGroup::get_one()
 	std::random_shuffle(this->mains.begin(), this->mains.end());
 	for (size_t i = 0; i < this->mains.size(); i++)
 	{
-		if (this->mains[i]->fail_count < this->mains[i]->params.max_fails)
+		if (this->mains[i]->fail_count < this->mains[i]->params->max_fails)
 		{
 			addr = this->mains[i];
 			break;
@@ -494,7 +202,7 @@ const EndpointAddress *EndpointGroup::get_one()
 		std::random_shuffle(this->backups.begin(), this->backups.end());
 		for (size_t i = 0; i < this->backups.size(); i++)
 		{
-			if (this->backups[i]->fail_count < this->backups[i]->params.max_fails)
+			if (this->backups[i]->fail_count < this->backups[i]->params->max_fails)
 			{
 				addr = this->backups[i];
 				break;
@@ -517,7 +225,7 @@ const EndpointAddress *EndpointGroup::get_one_backup()
 	std::random_shuffle(this->backups.begin(), this->backups.end());
 	for (size_t i = 0; i < this->backups.size(); i++)
 	{
-		if (this->backups[i]->fail_count < this->backups[i]->params.max_fails)
+		if (this->backups[i]->fail_count < this->backups[i]->params->max_fails)
 		{
 			addr = this->backups[i];
 			break;
@@ -530,7 +238,8 @@ const EndpointAddress *EndpointGroup::get_one_backup()
 
 void UPSGroupPolicy::add_server_locked(EndpointAddress *addr)
 {
-	int group_id = addr->params.group_id;
+	UPSAddrParams *params = static_cast<UPSAddrParams *>(addr->params);
+	int group_id = params->group_id;
 	rb_node **p = &this->group_map.rb_node;
 	rb_node *parent = NULL;
 	EndpointGroup *group;
@@ -538,7 +247,7 @@ void UPSGroupPolicy::add_server_locked(EndpointAddress *addr)
 	this->addresses.push_back(addr);
 	this->server_map[addr->address].push_back(addr);
 
-	if (addr->params.server_type == 0)
+	if (params->server_type == 0)
 		this->servers.push_back(addr);
 
 	while (*p)
@@ -562,12 +271,12 @@ void UPSGroupPolicy::add_server_locked(EndpointAddress *addr)
 	}
 
 	pthread_mutex_lock(&group->mutex);
-	addr->group = group;
+	params->group = group;
 	this->recover_one_server(addr);
-	if (addr->params.server_type == 0)
+	if (params->server_type == 0)
 	{
 		group->mains.push_back(addr);
-		group->weight += addr->params.weight;
+		group->weight += params->weight;
 	}
 	else
 		group->backups.push_back(addr);
@@ -578,27 +287,29 @@ void UPSGroupPolicy::add_server_locked(EndpointAddress *addr)
 
 int UPSGroupPolicy::remove_server_locked(const std::string& address)
 {
+	UPSAddrParams *params;
 	const auto map_it = this->server_map.find(address);
 
 	if (map_it != this->server_map.cend())
 	{
 		for (EndpointAddress *addr : map_it->second)
 		{
-			EndpointGroup *group = addr->group;
+			params = static_cast<UPSAddrParams *>(addr->params);
+			EndpointGroup *group = params->group;
 			std::vector<EndpointAddress *> *vec;
 
-			if (addr->params.server_type == 0)
+			if (params->server_type == 0)
 				vec = &group->mains;
 			else
 				vec = &group->backups;
 
 			//std::lock_guard<std::mutex> lock(group->mutex);
 			pthread_mutex_lock(&group->mutex);
-			if (addr->fail_count < addr->params.max_fails)
+			if (addr->fail_count < params->max_fails)
 				this->fuse_one_server(addr);
 
-			if (addr->params.server_type == 0)
-				group->weight -= addr->params.weight;
+			if (params->server_type == 0)
+				group->weight -= params->weight;
 
 			for (auto it = vec->begin(); it != vec->end(); ++it)
 			{
@@ -640,6 +351,7 @@ int UPSGroupPolicy::remove_server_locked(const std::string& address)
 
 const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int hash)
 {
+	const UPSAddrParams *params;
 	const EndpointAddress *addr = NULL;
 	unsigned int min_dis = (unsigned int)-1;
 
@@ -647,11 +359,13 @@ const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int h
 	{
 		if (this->is_alive_or_group_alive(server))
 		{
+			params = static_cast<UPSAddrParams *>(server->params);
+
 			for (int i = 0; i < VIRTUAL_GROUP_SIZE; i++)
 			{
 				unsigned int dis = std::min<unsigned int>
-								   (hash - server->consistent_hash[i],
-								   server->consistent_hash[i] - hash);
+								   (hash - params->consistent_hash[i],
+								   params->consistent_hash[i] - hash);
 
 				if (dis < min_dis)
 				{
@@ -667,22 +381,26 @@ const EndpointAddress *UPSGroupPolicy::consistent_hash_with_group(unsigned int h
 
 void UPSWeightedRandomPolicy::add_server_locked(EndpointAddress *addr)
 {
+	UPSAddrParams *params = static_cast<UPSAddrParams *>(addr->params);
+
 	UPSGroupPolicy::add_server_locked(addr);
-	if (addr->params.server_type == 0)
-		this->total_weight += addr->params.weight;
+	if (params->server_type == 0)
+		this->total_weight += params->weight;
 	return;
 }
 
 int UPSWeightedRandomPolicy::remove_server_locked(const std::string& address)
 {
+	UPSAddrParams *params;
 	const auto map_it = this->server_map.find(address);
 
 	if (map_it != this->server_map.cend())
 	{
 		for (EndpointAddress *addr : map_it->second)
 		{
-			if (addr->params.server_type == 0)
-				this->total_weight -= addr->params.weight;
+			params = static_cast<UPSAddrParams *>(addr->params);
+			if (params->server_type == 0)
+				this->total_weight -= params->weight;
 		}
 	}
 
@@ -694,6 +412,7 @@ const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI&
 	int x = 0;
 	int s = 0;
 	size_t idx;
+	UPSAddrParams *params;
 	int temp_weight = this->total_weight;
 
 	if (temp_weight > 0)
@@ -701,7 +420,8 @@ const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI&
 
 	for (idx = 0; idx < this->servers.size(); idx++)
 	{
-		s += this->servers[idx]->params.weight;
+		params = static_cast<UPSAddrParams *>(this->servers[idx]->params);
+		s += params->weight;
 		if (s > x)
 			break;
 	}
@@ -713,6 +433,7 @@ const EndpointAddress *UPSWeightedRandomPolicy::first_stradegy(const ParsedURI&
 
 const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI& uri)
 {
+	UPSAddrParams *params;
 	int temp_weight = this->available_weight;
 	if (temp_weight == 0)
 		return NULL;
@@ -725,8 +446,9 @@ const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI
 	{
 		if (this->is_alive_or_group_alive(server))
 		{
-			addr = server;
-			s += server->params.weight;
+			addr = server; // TODO: please double check here
+			params = static_cast<UPSAddrParams *>(server->params);
+			s += params->weight;
 			if (s > x)
 				break;
 		}
@@ -737,22 +459,26 @@ const EndpointAddress *UPSWeightedRandomPolicy::another_stradegy(const ParsedURI
 
 void UPSWeightedRandomPolicy::recover_one_server(const EndpointAddress *addr)
 {
-	this->nalives++;
-	if (addr->group->nalives++ == 0 && addr->group->id > 0)
-		this->available_weight += addr->group->weight;
+	UPSAddrParams *params = static_cast<UPSAddrParams *>(addr->params);
 
-	if (addr->params.group_id < 0 && addr->params.server_type == 0)
-		this->available_weight += addr->params.weight;
+	this->nalives++;
+	if (params->group->nalives++ == 0 && params->group->id > 0)
+		this->available_weight += params->group->weight;
+
+	if (params->group_id < 0 && params->server_type == 0)
+		this->available_weight += params->weight;
 }
 
 void UPSWeightedRandomPolicy::fuse_one_server(const EndpointAddress *addr)
 {
-	this->nalives--;
-	if (--addr->group->nalives == 0 && addr->group->id > 0)
-		this->available_weight -= addr->group->weight;
+	UPSAddrParams *params = static_cast<UPSAddrParams *>(addr->params);
 
-	if (addr->params.group_id < 0 && addr->params.server_type == 0)
-		this->available_weight -= addr->params.weight;
+	this->nalives--;
+	if (--params->group->nalives == 0 && params->group->id > 0)
+		this->available_weight -= params->group->weight;
+
+	if (params->group_id < 0 && params->server_type == 0)
+		this->available_weight -= params->weight;
 }
 
 const EndpointAddress *UPSConsistentHashPolicy::first_stradegy(const ParsedURI& uri)

src/nameservice/UpstreamPolicies.h

@@ -20,7 +20,6 @@
 #define _UPSTREAM_POLICIES_H_ 
 
 #include <pthread.h>
-#include <unordered_map>
 #include <vector>
 #include <atomic>
 #include "URIParser.h"
@@ -29,34 +28,25 @@
 #include "WFDNSResolver.h"
 #include "WFGlobal.h"
 #include "WFTaskError.h"
-#include "UpstreamManager.h"
+#include "ServiceGovernance.h"
 
-#define MTTR_SECOND			30
-#define VIRTUAL_GROUP_SIZE  16
-
-#define GET_CURRENT_SECOND  std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now().time_since_epoch()).count()
+using upstream_route_t = std::function<unsigned int (const char *, const char *, const char *)>;
 
 class EndpointGroup;
-class UPSPolicy;
 class UPSGroupPolicy;
 
-class EndpointAddress
+class UPSAddrParams : public PolicyAddrParams
 {
 public:
+	unsigned short weight;
+	short server_type;
+	int group_id;
 	EndpointGroup *group;
-	AddressParams params;
-	std::string address;
-	std::string host;
-	std::string port;
-	short port_value;
-	struct list_head list;
-	std::atomic<unsigned int> fail_count;
-	long long broken_timeout;
 	unsigned int consistent_hash[VIRTUAL_GROUP_SIZE];
 
-public:
-	EndpointAddress(const std::string& address,
-					const struct AddressParams *address_params);
+	UPSAddrParams();
+	UPSAddrParams(const struct AddressParams *params,
+				  const std::string& address);
 };
 
 class EndpointGroup
@@ -85,83 +75,19 @@ public:
 	const EndpointAddress *get_one_backup();
 };
 
-class UPSPolicy : public WFDNSResolver
-{
-public:
-	virtual WFRouterTask *create_router_task(const struct WFNSParams *params,
-											 router_callback_t callback);
-	virtual void success(RouteManager::RouteResult *result, void *cookie,
-					 	 CommTarget *target);
-	virtual void failed(RouteManager::RouteResult *result, void *cookie,
-						CommTarget *target);
-
-	void add_server(const std::string& address, const AddressParams *address_params);
-	int remove_server(const std::string& address);
-	int replace_server(const std::string& address, const AddressParams *address_params);
-
-	virtual void enable_server(const std::string& address);
-	virtual void disable_server(const std::string& address);
-	virtual void get_main_address(std::vector<std::string>& addr_list);
-	// virtual void server_list_change(/* std::vector<server> status */) {}
-
-public:
-	UPSPolicy() :
-		breaker_lock(PTHREAD_MUTEX_INITIALIZER),
-		rwlock(PTHREAD_RWLOCK_INITIALIZER)
-	{
-		this->nalives = 0;
-		this->try_another = false;
-		INIT_LIST_HEAD(&this->breaker_list);
-	}
-
-	virtual ~UPSPolicy()
-	{
-		for (EndpointAddress *addr : this->addresses)
-			delete addr;
-	}
-
-private:
-	virtual bool select(const ParsedURI& uri, EndpointAddress **addr);
-
-	virtual void recover_one_server(const EndpointAddress *addr)
-	{
-		this->nalives++;
-	}
-
-	virtual void fuse_one_server(const EndpointAddress *addr)
-	{
-		this->nalives--;
-	}
-
-	virtual void add_server_locked(EndpointAddress *addr);
-	virtual int remove_server_locked(const std::string& address);
-
-	void recover_server_from_breaker(EndpointAddress *addr);
-	void fuse_server_to_breaker(EndpointAddress *addr);
-
-	struct list_head breaker_list;
-	pthread_mutex_t breaker_lock;
-
-protected:
-	virtual const EndpointAddress *first_stradegy(const ParsedURI& uri);
-	virtual const EndpointAddress *another_stradegy(const ParsedURI& uri);
-	void check_breaker();
-
-	std::vector<EndpointAddress *> servers; // current servers
-	std::vector<EndpointAddress *> addresses; // memory management
-	std::unordered_map<std::string,
-					   std::vector<EndpointAddress *>> server_map;
-	pthread_rwlock_t rwlock;
-	std::atomic<int> nalives;
-	bool try_another;
-};
-
-class UPSGroupPolicy : public UPSPolicy
+class UPSGroupPolicy : public ServiceGovernance
 {
 public:
 	UPSGroupPolicy();
 	~UPSGroupPolicy();
 
+	virtual bool select(const ParsedURI& uri, EndpointAddress **addr);
+	virtual void add_server(const std::string& address,
+							const struct AddressParams *params);
+	virtual int replace_server(const std::string& address,
+							   const struct AddressParams *params);
+	void get_main_address(std::vector<std::string>& addr_list);
+
 protected:
 	struct rb_root group_map;
 	EndpointGroup *default_group;
@@ -170,17 +96,17 @@ private:
 	virtual void recover_one_server(const EndpointAddress *addr)
 	{
 		this->nalives++;
-		addr->group->nalives++;
+		UPSAddrParams *params = static_cast<UPSAddrParams *>(addr->params);
+		params->group->nalives++;
 	}
 
 	virtual void fuse_one_server(const EndpointAddress *addr)
 	{
 		this->nalives--;
-		addr->group->nalives--;
+		UPSAddrParams *params = static_cast<UPSAddrParams *>(addr->params);
+		params->group->nalives--;
 	}
 
-	virtual bool select(const ParsedURI& uri, EndpointAddress **addr);
-
 protected:
 	virtual void add_server_locked(EndpointAddress *addr);
 	virtual int remove_server_locked(const std::string& address);
@@ -190,10 +116,11 @@ protected:
 
 	inline bool is_alive_or_group_alive(const EndpointAddress *addr) const
 	{
-		return ((addr->params.group_id < 0 &&
-					addr->fail_count < addr->params.max_fails) || 
-				(addr->params.group_id >= 0 &&
-					addr->group->nalives > 0));
+		UPSAddrParams *params = static_cast<UPSAddrParams *>(addr->params);
+		return ((params->group_id < 0 &&
+					addr->fail_count < addr->params->max_fails) ||
+				(params->group_id >= 0 &&
+					params->group->nalives > 0));
 	}
 };
 

test/upstream_unittest.cc

@@ -1,5 +1,5 @@
 /*
-  Copyright (c) 2020 Sogou, Inc.
+  Copyright (c) 2021 Sogou, Inc.
 
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
@@ -21,10 +21,11 @@
 #include "workflow/WFHttpServer.h"
 #include "workflow/WFTaskFactory.h"
 #include "workflow/WFFacilities.h"
+#include "workflow/UpstreamPolicies.h"
 
-#define REDIRECT_MAX	3
-#define RETRY_MAX		3
-#define MTTR			30
+#define REDIRECT_MAX	2
+#define RETRY_MAX		2
+#define MTTR			2
 #define MAX_FAILS		200
 
 static void __http_process1(WFHttpTask *task)
@@ -105,8 +106,6 @@ TEST(upstream_unittest, BasicPolicy)
 {
 	WFFacilities::WaitGroup wait_group(3);
 
-	register_upstream_hosts();
-
 	char url[3][30] = {"http://weighted.random", "http://hash", "http://manual"};
 
 	http_callback_t cb = std::bind(basic_callback, std::placeholders::_1,
@@ -128,7 +127,6 @@ TEST(upstream_unittest, EnableAndDisable)
 
 	UpstreamManager::upstream_disable_server("weighted.random", "127.0.0.1:8001");
 
-	//fprintf(stderr, "disable server and try......................\n");
 	std::string url = "http://weighted.random";
 	WFHttpTask *task = WFTaskFactory::create_http_task(url, REDIRECT_MAX, RETRY_MAX,
 											  		   [&wait_group, &url](WFHttpTask *task){
@@ -136,7 +134,6 @@ TEST(upstream_unittest, EnableAndDisable)
 		EXPECT_EQ(state, WFT_STATE_TASK_ERROR);
 		EXPECT_EQ(task->get_error(), WFT_ERR_UPSTREAM_UNAVAILABLE);
 		UpstreamManager::upstream_enable_server("weighted.random", "127.0.0.1:8001");
-		//fprintf(stderr, "ensable server and try......................\n");
 		auto *task2 = WFTaskFactory::create_http_task(url, REDIRECT_MAX, RETRY_MAX,
 													  std::bind(basic_callback,
 													  			std::placeholders::_1,
@@ -156,10 +153,23 @@ TEST(upstream_unittest, FuseAndRecover)
 	WFHttpTask *task;
 	SeriesWork *series;
 	protocol::HttpRequest *req;
-	std::string url = "http://weighted.random";
+	std::string url = "http://test_policy";
 	int batch = MAX_FAILS + 50;
-	int timeout = (MTTR + 3) * 1000000;
+	int timeout = (MTTR + 1) * 1000000;
+
+	UPSWeightedRandomPolicy test_policy(false);
+	test_policy.set_mttr_second(MTTR);
+	AddressParams address_params = ADDRESS_PARAMS_DEFAULT;
 	
+	address_params.weight = 1000;
+	test_policy.add_server("127.0.0.1:8001", &address_params);
+
+	address_params.weight = 1;
+	test_policy.add_server("127.0.0.1:8002", &address_params);
+
+	auto *ns = WFGlobal::get_name_service();
+	EXPECT_EQ(ns->add_policy("test_policy", &test_policy), 0);
+
 	http_server1.stop();
 	fprintf(stderr, "server 1 stopped start %d tasks to fuse it\n", batch);
 	ParallelWork *pwork = Workflow::create_parallel_work(
@@ -221,6 +231,8 @@ int main(int argc, char* argv[])
 {
 	::testing::InitGoogleTest(&argc, argv);
 
+	register_upstream_hosts();
+
 	EXPECT_TRUE(http_server1.start("127.0.0.1", 8001) == 0)
 				<< "http server start failed";