Merge pull request #860 from Barenboim/master

Optimize service governance codes structure

src/nameservice/WFDnsResolver.cc

@@ -41,18 +41,6 @@
 #include "WFDnsClient.h"
 #include "WFDnsResolver.h"
 
-/*
-DNS_CACHE_LEVEL_0	->	NO cache
-DNS_CACHE_LEVEL_1	->	TTL MIN
-DNS_CACHE_LEVEL_2	->	TTL [DEFAULT]
-DNS_CACHE_LEVEL_3	->	Forever
-*/
-
-#define DNS_CACHE_LEVEL_0		0
-#define DNS_CACHE_LEVEL_1		1
-#define DNS_CACHE_LEVEL_2		2
-#define DNS_CACHE_LEVEL_3		3
-
 #define HOSTS_LINEBUF_INIT_SIZE	128
 #define PORT_STR_MAX			5
 
@@ -223,116 +211,59 @@ static ThreadDnsTask *__create_thread_dns_task(const std::string& host,
 	return task;
 }
 
-class WFResolverTask : public WFRouterTask
-{
-public:
-	WFResolverTask(const struct WFNSParams *params, int dns_cache_level,
-				   unsigned int dns_ttl_default, unsigned int dns_ttl_min,
-				   const struct EndpointParams *endpoint_params,
-				   router_callback_t&& cb) :
-		WFRouterTask(std::move(cb))
-	{
-		type_ = params->type;
-		host_ = params->uri.host ? params->uri.host : "";
-		port_ = params->uri.port ? atoi(params->uri.port) : 0;
-		info_ = params->info;
-		dns_cache_level_ = dns_cache_level;
-		dns_ttl_default_ = dns_ttl_default;
-		dns_ttl_min_ = dns_ttl_min;
-		endpoint_params_ = *endpoint_params;
-		first_addr_only_ = params->fixed_addr;
-	}
-
-private:
-	virtual void dispatch();
-	virtual SubTask *done();
-	void thread_dns_callback(ThreadDnsTask *dns_task);
-	void dns_single_callback(WFDnsTask *dns_task);
-	static void dns_partial_callback(WFDnsTask *dns_task);
-	void dns_parallel_callback(const ParallelWork *pwork);
-	void dns_callback_internal(DnsOutput *dns_task,
-							   unsigned int ttl_default,
-							   unsigned int ttl_min);
-
-private:
-	TransportType type_;
-	std::string host_;
-	std::string info_;
-	unsigned short port_;
-	bool first_addr_only_;
-	bool query_dns_;
-	int dns_cache_level_;
-	unsigned int dns_ttl_default_;
-	unsigned int dns_ttl_min_;
-	struct EndpointParams endpoint_params_;
-};
-
 void WFResolverTask::dispatch()
 {
-	if (dns_cache_level_ != DNS_CACHE_LEVEL_0)
+	const ParsedURI& uri = ns_params_.uri;
+	host_ = uri.host ? uri.host : "";
+	port_ = uri.port ? atoi(uri.port) : 0;
+
+	DnsCache *dns_cache = WFGlobal::get_dns_cache();
+	const DnsCache::DnsHandle *addr_handle;
+	std::string hostname = host_;
+
+	if (ns_params_.retry_times == 0)
+		addr_handle = dns_cache->get_ttl(hostname, port_);
+	else
+		addr_handle = dns_cache->get_confident(hostname, port_);
+
+	if (addr_handle)
 	{
-		DnsCache *dns_cache = WFGlobal::get_dns_cache();
-		const DnsCache::DnsHandle *addr_handle;
+		RouteManager *route_manager = WFGlobal::get_route_manager();
+		struct addrinfo *addrinfo = addr_handle->value.addrinfo;
+		struct addrinfo first;
 
-		switch (dns_cache_level_)
+		if (ns_params_.fixed_addr && addrinfo->ai_next)
 		{
-		case DNS_CACHE_LEVEL_1:
-			addr_handle = dns_cache->get_confident(host_, port_);
-			break;
-
-		case DNS_CACHE_LEVEL_2:
-			addr_handle = dns_cache->get_ttl(host_, port_);
-			break;
-
-		case DNS_CACHE_LEVEL_3:
-			addr_handle = dns_cache->get(host_, port_);
-			break;
-
-		default:
-			addr_handle = NULL;
-			break;
+			first = *addrinfo;
+			first.ai_next = NULL;
+			addrinfo = &first;
 		}
 
-		if (addr_handle)
+		if (route_manager->get(ns_params_.type, addrinfo, ns_params_.info,
+							   &ep_params_, hostname, this->result) < 0)
 		{
-			RouteManager *route_manager = WFGlobal::get_route_manager();
-			struct addrinfo *addrinfo = addr_handle->value.addrinfo;
-			struct addrinfo first;
-
-			if (first_addr_only_ && addrinfo->ai_next)
-			{
-				first = *addrinfo;
-				first.ai_next = NULL;
-				addrinfo = &first;
-			}
-
-			if (route_manager->get(type_, addrinfo, info_, &endpoint_params_,
-								   host_, this->result) < 0)
-			{
-				this->state = WFT_STATE_SYS_ERROR;
-				this->error = errno;
-			}
-			else
-				this->state = WFT_STATE_SUCCESS;
-
-			dns_cache->release(addr_handle);
-			query_dns_ = false;
-			this->subtask_done();
-			return;
+			this->state = WFT_STATE_SYS_ERROR;
+			this->error = errno;
 		}
+		else
+			this->state = WFT_STATE_SUCCESS;
+
+		dns_cache->release(addr_handle);
+		this->subtask_done();
+		return;
 	}
 
-	if (!host_.empty())
+	if (*host_)
 	{
-		char front = host_.front();
-		char back = host_.back();
+		char front = host_[0];
+		char back = host_[hostname.size() - 1];
 		struct in6_addr addr;
 		int ret;
 
-		if (host_.find(':') != std::string::npos)
-			ret = inet_pton(AF_INET6, host_.c_str(), &addr);
+		if (strchr(host_, ':'))
+			ret = inet_pton(AF_INET6, host_, &addr);
 		else if (isdigit(back) && isdigit(front))
-			ret = inet_pton(AF_INET, host_.c_str(), &addr);
+			ret = inet_pton(AF_INET, host_, &addr);
 		else if (front == '/')
 			ret = 1;
 		else
@@ -340,13 +271,12 @@ void WFResolverTask::dispatch()
 
 		if (ret == 1)
 		{
-			DnsInput dns_in(host_, port_, true); // 'true' means numeric host
+			DnsInput dns_in(hostname, port_, true); // 'true' means numeric host
 			DnsOutput dns_out;
 
 			DnsRoutine::run(&dns_in, &dns_out);
 			__add_passive_flags((struct addrinfo *)dns_out.get_addrinfo());
 			dns_callback_internal(&dns_out, (unsigned int)-1, (unsigned int)-1);
-			query_dns_ = false;
 			this->subtask_done();
 			return;
 		}
@@ -356,7 +286,7 @@ void WFResolverTask::dispatch()
 	if (hosts)
 	{
 		struct addrinfo *ai;
-		int ret = __readaddrinfo(hosts, host_.c_str(), port_, &__ai_hints, &ai);
+		int ret = __readaddrinfo(hosts, host_, port_, &__ai_hints, &ai);
 
 		if (ret == 0)
 		{
@@ -364,7 +294,6 @@ void WFResolverTask::dispatch()
 			DnsRoutine::create(&out, ret, ai);
 			__add_passive_flags((struct addrinfo *)out.get_addrinfo());
 			dns_callback_internal(&out, dns_ttl_default_, dns_ttl_min_);
-			query_dns_ = false;
 			this->subtask_done();
 			return;
 		}
@@ -381,7 +310,7 @@ void WFResolverTask::dispatch()
 			auto&& cb = std::bind(&WFResolverTask::dns_single_callback,
 								  this,
 								  std::placeholders::_1);
-			WFDnsTask *dns_task = client->create_dns_task(host_, std::move(cb));
+			WFDnsTask *dns_task = client->create_dns_task(hostname, std::move(cb));
 
 			if (family == AF_INET6)
 				dns_task->get_req()->set_question_type(DNS_TYPE_AAAA);
@@ -401,10 +330,10 @@ void WFResolverTask::dispatch()
 			dctx[0].port = port_;
 			dctx[1].port = port_;
 
-			task_v4 = client->create_dns_task(host_, dns_partial_callback);
+			task_v4 = client->create_dns_task(hostname, dns_partial_callback);
 			task_v4->user_data = dctx;
 
-			task_v6 = client->create_dns_task(host_, dns_partial_callback);
+			task_v6 = client->create_dns_task(hostname, dns_partial_callback);
 			task_v6->get_req()->set_question_type(DNS_TYPE_AAAA);
 			task_v6->user_data = dctx + 1;
 
@@ -428,7 +357,7 @@ void WFResolverTask::dispatch()
 		auto&& cb = std::bind(&WFResolverTask::thread_dns_callback,
 							  this,
 							  std::placeholders::_1);
-		ThreadDnsTask *dns_task = __create_thread_dns_task(host_, port_,
+		ThreadDnsTask *dns_task = __create_thread_dns_task(hostname, port_,
 														   std::move(cb));
 		series_of(this)->push_front(dns_task);
 	}
@@ -477,12 +406,13 @@ void WFResolverTask::dns_callback_internal(DnsOutput *dns_out,
 		DnsCache *dns_cache = WFGlobal::get_dns_cache();
 		struct addrinfo *addrinfo = dns_out->move_addrinfo();
 		const DnsCache::DnsHandle *addr_handle;
+		std::string hostname = host_;
 
-		addr_handle = dns_cache->put(host_, port_, addrinfo,
+		addr_handle = dns_cache->put(hostname, port_, addrinfo,
 									 (unsigned int)ttl_default,
 									 (unsigned int)ttl_min);
-		if (route_manager->get(type_, addrinfo, info_, &endpoint_params_,
-							   host_, this->result) < 0)
+		if (route_manager->get(ns_params_.type, addrinfo, ns_params_.info,
+							   &ep_params_, hostname, this->result) < 0)
 		{
 			this->state = WFT_STATE_SYS_ERROR;
 			this->error = errno;
@@ -502,7 +432,7 @@ void WFResolverTask::dns_single_callback(WFDnsTask *dns_task)
 	{
 		struct addrinfo *ai = NULL;
 		int ret;
-		
+
 		ret = protocol::DnsUtil::getaddrinfo(dns_task->get_resp(), port_, &ai);
 		DnsOutput out;
 		DnsRoutine::create(&out, ret, ai);
@@ -601,27 +531,14 @@ void WFResolverTask::thread_dns_callback(ThreadDnsTask *dns_task)
 	delete this;
 }
 
-WFRouterTask *
-WFDnsResolver::create(const struct WFNSParams *params, int dns_cache_level,
-					  unsigned int dns_ttl_default, unsigned int dns_ttl_min,
-					  const struct EndpointParams *endpoint_params,
-					  router_callback_t&& callback)
-{
-	return new WFResolverTask(params, dns_cache_level, dns_ttl_default,
-							  dns_ttl_min, endpoint_params,
-							  std::move(callback));
-}
-
 WFRouterTask *WFDnsResolver::create_router_task(const struct WFNSParams *params,
 												router_callback_t callback)
 {
 	const struct WFGlobalSettings *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;
-	return create(params, dns_cache_level, dns_ttl_default, dns_ttl_min,
-				  endpoint_params, std::move(callback));
+	const struct EndpointParams *ep_params = &settings->endpoint_params;
+	return new WFResolverTask(params, dns_ttl_default, dns_ttl_min, ep_params,
+							  std::move(callback));
 }
 

src/nameservice/WFDnsResolver.h

@@ -19,21 +19,71 @@
 #ifndef _WFDNSRESOLVER_H_
 #define _WFDNSRESOLVER_H_
 
+#include <string>
+#include <functional>
+#include "Workflow.h"
+#include "DnsMessage.h"
+#include "DnsRoutine.h"
 #include "EndpointParams.h"
 #include "WFNameService.h"
-#include "WFResourcePool.h"
+
+class WFResolverTask : public WFRouterTask
+{
+public:
+	WFResolverTask(const struct WFNSParams *ns_params,
+				   unsigned int dns_ttl_default, unsigned int dns_ttl_min,
+				   const struct EndpointParams *ep_params,
+				   router_callback_t&& cb) :
+		WFRouterTask(std::move(cb)),
+		ns_params_(*ns_params),
+		ep_params_(*ep_params)
+	{
+		dns_ttl_default_ = dns_ttl_default;
+		dns_ttl_min_ = dns_ttl_min;
+		query_dns_ = false;
+	}
+
+	WFResolverTask(const struct WFNSParams *ns_params,
+				   router_callback_t&& cb) :
+		WFRouterTask(std::move(cb)),
+		ns_params_(*ns_params)
+	{
+		query_dns_ = false;
+	}
+
+protected:
+	virtual void dispatch();
+	virtual SubTask *done();
+
+private:
+	using DnsTask_thrd = WFThreadTask<DnsInput, DnsOutput>;
+	using DnsTask_net = WFNetworkTask<protocol::DnsRequest,
+									  protocol::DnsResponse>;
+	void thread_dns_callback(DnsTask_thrd *dns_task);
+	void dns_single_callback(DnsTask_net *dns_task);
+	static void dns_partial_callback(DnsTask_net *dns_task);
+	void dns_parallel_callback(const ParallelWork *pwork);
+	void dns_callback_internal(DnsOutput *dns_task,
+							   unsigned int ttl_default,
+							   unsigned int ttl_min);
+
+protected:
+	struct WFNSParams ns_params_;
+	unsigned int dns_ttl_default_;
+	unsigned int dns_ttl_min_;
+	struct EndpointParams ep_params_;
+
+private:
+	const char *host_;
+	unsigned short port_;
+	bool query_dns_;
+};
 
 class WFDnsResolver : public WFNSPolicy
 {
 public:
 	virtual WFRouterTask *create_router_task(const struct WFNSParams *params,
 											 router_callback_t callback);
-
-public:
-	WFRouterTask *create(const struct WFNSParams *params, int dns_cache_level,
-						 unsigned int dns_ttl_default, unsigned int dns_ttl_min,
-						 const struct EndpointParams *endpoint_params,
-						 router_callback_t&& callback);
 };
 
 #endif

src/nameservice/WFServiceGovernance.cc

@@ -17,6 +17,8 @@
            Xie Han (xiehan@sogou-inc.com)
 */
 
+#include <stdlib.h>
+#include <string.h>
 #include <stdint.h>
 #include <vector>
 #include <chrono>
@@ -50,38 +52,6 @@ PolicyAddrParams::PolicyAddrParams(const struct AddressParams *params) :
 	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 void copy_host_port(ParsedURI& uri, const EndpointAddress *addr)
-{
-	if (!addr->host.empty())
-	{
-		free(uri.host);
-		uri.host = strdup(addr->host.c_str());
-	}
-
-	if (!addr->port.empty())
-	{
-		free(uri.port);
-		uri.port = strdup(addr->port.c_str());
-	}
-}
-
 EndpointAddress::EndpointAddress(const std::string& address,
 								 PolicyAddrParams *address_params)
 {
@@ -108,41 +78,76 @@ EndpointAddress::EndpointAddress(const std::string& address,
 		this->port = arr[1];
 }
 
-WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *params,
-													  router_callback_t callback)
+class WFSGResolverTask : public WFResolverTask
 {
-	WFNSTracing *tracing = params->tracing;
-	EndpointAddress *addr;
-	WFRouterTask *task;
-
-	if (this->select(params->uri, tracing, &addr))
+public:
+	WFSGResolverTask(const struct WFNSParams *params,
+					 WFServiceGovernance *sg,
+					 router_callback_t&& cb) :
+		WFResolverTask(params, std::move(cb))
 	{
-		WFDnsResolver *resolver = WFGlobal::get_dns_resolver();
-		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;
+		sg_ = sg;
+	}
 
-		copy_host_port(params->uri, addr);
-		task = resolver->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min,
-								endpoint_params, std::move(callback));
+protected:
+	virtual void dispatch();
 
-		struct TracingData *tracing_data = (struct TracingData *)tracing->data;
+protected:
+	WFServiceGovernance *sg_;
+};
+
+static void copy_host_port(ParsedURI& uri, const EndpointAddress *addr)
+{
+	if (!addr->host.empty())
+	{
+		free(uri.host);
+		uri.host = strdup(addr->host.c_str());
+	}
+
+	if (!addr->port.empty())
+	{
+		free(uri.port);
+		uri.port = strdup(addr->port.c_str());
+	}
+}
+
+void WFSGResolverTask::dispatch()
+{
+	WFNSTracing *tracing = ns_params_.tracing;
+	EndpointAddress *addr;
+
+	if (sg_->select(ns_params_.uri, tracing, &addr))
+	{
+		auto *tracing_data = (WFServiceGovernance::TracingData *)tracing->data;
 		if (!tracing_data)
 		{
-			tracing_data = new TracingData;
-			tracing_data->sg = this;
+			tracing_data = new WFServiceGovernance::TracingData;
+			tracing_data->sg = sg_;
 			tracing->data = tracing_data;
 			tracing->deleter = WFServiceGovernance::tracing_deleter;
 		}
 
 		tracing_data->history.push_back(addr);
+
+		copy_host_port(ns_params_.uri, addr);
+		dns_ttl_default_ = addr->params->dns_ttl_default;
+		dns_ttl_min_ = addr->params->dns_ttl_min;
+		ep_params_ = addr->params->endpoint_params;
+		this->WFResolverTask::dispatch();
+		return;
 	}
 	else
-		task = new WFSelectorFailTask(std::move(callback));
+	{
+		this->state = WFT_STATE_TASK_ERROR;
+		this->error = WFT_ERR_UPSTREAM_UNAVAILABLE;
+		return this->subtask_done();
+	}
+}
 
-	return task;
+WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *params,
+													  router_callback_t callback)
+{
+	return new WFSGResolverTask(params, this, std::move(callback));
 }
 
 void WFServiceGovernance::tracing_deleter(void *data)

src/nameservice/WFServiceGovernance.h

@@ -193,6 +193,7 @@ protected:
 	pthread_rwlock_t rwlock;
 	std::atomic<int> nalives;
 	bool try_another;
+	friend class WFSGResolverTask;
 };
 
 #endif