Merge pull request #503 from Barenboim/master

add asynchronous dns connections limit.

src/factory/DnsTaskImpl.cc

@@ -83,15 +83,6 @@ CommMessageIn *ComplexDnsTask::message_in()
 
 bool ComplexDnsTask::init_success()
 {
-	static constexpr struct EndpointParams ep =
-	{
-		.max_connections		=	65535,
-		.connect_timeout		=	10 * 1000,
-		.response_timeout		=	10 * 1000,
-		.ssl_connect_timeout	=	10 * 1000,
-		.use_tls_sni			=	false
-	};
-
 	if (uri_.scheme && strcasecmp(uri_.scheme, "dnss") == 0)
 		this->WFComplexClientTask::set_transport_type(TT_TCP_SSL);
 	else if (uri_.scheme && strcasecmp(uri_.scheme, "dns") != 0)
@@ -115,7 +106,8 @@ bool ComplexDnsTask::init_success()
 			return false;
 		}
 
-		ret = WFGlobal::get_route_manager()->get(type, addr, info_, &ep,
+		auto *ep = &WFGlobal::get_global_settings()->dns_server_params;
+		ret = WFGlobal::get_route_manager()->get(type, addr, info_, ep,
 												 uri_.host, route_result_);
 		freeaddrinfo(addr);
 		if (ret < 0)

src/factory/WFResourcePool.cc

@@ -13,8 +13,8 @@
   See the License for the specific language governing permissions and
   limitations under the License.
 
-  Author: Li Yingxin (liyingxin@sogou-inc.com)
-          Xie Han (xiehan@sogou-inc.com)
+  Authors: Li Yingxin (liyingxin@sogou-inc.com)
+           Xie Han (xiehan@sogou-inc.com)
 */
 
 #include <string.h>
@@ -60,16 +60,27 @@ WFConditional *WFResourcePool::get(SubTask *task, void **resbuf)
 	return new __WFConditional(task, resbuf, &this->data);
 }
 
-WFResourcePool::WFResourcePool(void *const *res, size_t n)
+void WFResourcePool::create(size_t n)
 {
 	this->data.res = new void *[n];
-	memcpy(this->data.res, res, n * sizeof (void *));
 	this->data.value = n;
 	this->data.index = 0;
 	INIT_LIST_HEAD(&this->data.wait_list);
 	this->data.pool = this;
 }
 
+WFResourcePool::WFResourcePool(void *const *res, size_t n)
+{
+	this->create(n);
+	memcpy(this->data.res, res, n * sizeof (void *));
+}
+
+WFResourcePool::WFResourcePool(size_t n)
+{
+	this->create(n);
+	memset(this->data.res, 0, n * sizeof (void *));
+}
+
 void WFResourcePool::post(void *res)
 {
 	struct WFResourcePool::Data *data = &this->data;

src/factory/WFResourcePool.h

@@ -13,8 +13,8 @@
   See the License for the specific language governing permissions and
   limitations under the License.
 
-  Author: Li Yingxin (liyingxin@sogou-inc.com)
-          Xie Han (xiehan@sogou-inc.com)
+  Authors: Li Yingxin (liyingxin@sogou-inc.com)
+           Xie Han (xiehan@sogou-inc.com)
 */
 
 #ifndef _WFRESOURCEPOOL_H_
@@ -58,8 +58,12 @@ private:
 private:
 	struct Data data;
 
+private:
+	void create(size_t n);
+
 public:
 	WFResourcePool(void *const *res, size_t n);
+	WFResourcePool(size_t n);
 	virtual ~WFResourcePool() { delete []this->data.res; }
 };
 

src/manager/WFGlobal.cc

@@ -543,18 +543,17 @@ public:
 	}
 
 public:
+	WFDnsResolver *get_dns_resolver() { return &resolver_; }
 	WFNameService *get_name_service() { return &service_; }
 
 private:
-	static WFDnsResolver resolver_;
+	WFDnsResolver resolver_;
 	WFNameService service_;
 
 public:
-	__NameServiceManager() : service_(&__NameServiceManager::resolver_) { }
+	__NameServiceManager() : service_(&resolver_) { }
 };
 
-WFDnsResolver __NameServiceManager::resolver_;
-
 #define MAX(x, y)	((x) >= (y) ? (x) : (y))
 #define HOSTS_LINEBUF_INIT_SIZE	128
 
@@ -805,6 +804,11 @@ WFNameService *WFGlobal::get_name_service()
 	return __NameServiceManager::get_instance()->get_name_service();
 }
 
+WFDnsResolver *WFGlobal::get_dns_resolver()
+{
+	return __NameServiceManager::get_instance()->get_dns_resolver();
+}
+
 const char *WFGlobal::get_default_port(const std::string& scheme)
 {
 	return __WFGlobal::get_instance()->get_default_port(scheme);

src/manager/WFGlobal.h

@@ -32,6 +32,7 @@
 #include "Executor.h"
 #include "EndpointParams.h"
 #include "WFNameService.h"
+#include "WFDnsResolver.h"
 
 /**
  * @file    WFGlobal.h
@@ -46,6 +47,7 @@
 struct WFGlobalSettings
 {
 	struct EndpointParams endpoint_params;
+	struct EndpointParams dns_server_params;
 	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
 	int dns_threads;
@@ -62,6 +64,7 @@ struct WFGlobalSettings
 static constexpr struct WFGlobalSettings GLOBAL_SETTINGS_DEFAULT =
 {
 	.endpoint_params	=	ENDPOINT_PARAMS_DEFAULT,
+	.dns_server_params	=	ENDPOINT_PARAMS_DEFAULT,
 	.dns_ttl_default	=	12 * 3600,
 	.dns_ttl_min		=	180,
 	.dns_threads		=	4,
@@ -109,8 +112,8 @@ public:
 
 	static const char *get_error_string(int state, int error);
 
-public:
 	// Internal usage only
+public:
 	static CommScheduler *get_scheduler();
 	static DnsCache *get_dns_cache();
 	static RouteManager *get_route_manager();
@@ -122,7 +125,10 @@ public:
 	static ExecQueue *get_dns_queue();
 	static Executor *get_dns_executor();
 	static WFNameService *get_name_service();
+	static WFDnsResolver *get_dns_resolver();
 	static class WFDnsClient *get_dns_client();
+
+public:
 	static void sync_operation_begin();
 	static void sync_operation_end();
 };

src/nameservice/WFDnsResolver.cc

@@ -13,9 +13,9 @@
   See the License for the specific language governing permissions and
   limitations under the License.
 
-  Authors: Wu Jiaxu (wujiaxu@sogou-inc.com)
-           Xie Han (xiehan@sogou-inc.com)
+  Authors: Xie Han (xiehan@sogou-inc.com)
            Liu Kai (liukaidx@sogou-inc.com)
+           Wu Jiaxu (wujiaxu@sogou-inc.com)
 */
 
 #include <sys/types.h>
@@ -34,6 +34,7 @@
 #include "WFGlobal.h"
 #include "WFTaskError.h"
 #include "WFTaskFactory.h"
+#include "WFResourcePool.h"
 #include "WFNameService.h"
 #include "DnsUtil.h"
 #include "WFDnsClient.h"
@@ -265,24 +266,22 @@ static ThreadDnsTask *__create_thread_dns_task(const std::string& host,
 
 class WFResolverTask : public WFRouterTask
 {
-	static void dns_partial_callback(WFDnsTask *dns_task);
-
 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)
+		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:
@@ -290,6 +289,7 @@ private:
 	virtual SubTask *done();
 	void thread_dns_callback(ThreadDnsTask *dns_task);
 	void dns_single_callback(WFDnsTask *dns_task);
+	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,
@@ -298,14 +298,14 @@ private:
 private:
 	TransportType type_;
 	std::string host_;
-	unsigned short port_;
 	std::string info_;
+	unsigned short port_;
+	bool first_addr_only_;
+	bool insert_dns_;
 	int dns_cache_level_;
 	unsigned int dns_ttl_default_;
 	unsigned int dns_ttl_min_;
 	struct EndpointParams endpoint_params_;
-	bool first_addr_only_;
-	bool insert_dns_;
 };
 
 void WFResolverTask::dispatch()
@@ -420,6 +420,8 @@ void WFResolverTask::dispatch()
 	}
 	else if (insert_dns_)
 	{
+		WFDnsResolver *resolver = WFGlobal::get_dns_resolver();
+
 		if (family == AF_INET || family == AF_INET6)
 		{
 			auto&& cb = std::bind(&WFResolverTask::dns_single_callback,
@@ -430,10 +432,17 @@ void WFResolverTask::dispatch()
 			if (family == AF_INET6)
 				dns_task->get_req()->set_question_type(DNS_TYPE_AAAA);
 
-			series_of(this)->push_front(dns_task);
+			WFConditional *cond = resolver->get_cond(dns_task);
+			series_of(this)->push_front(cond);
 		}
 		else
 		{
+			auto&& cb_v4 = std::bind(&WFResolverTask::dns_partial_callback,
+									 this,
+									 std::placeholders::_1);
+			auto&& cb_v6 = std::bind(&WFResolverTask::dns_partial_callback,
+									 this,
+									 std::placeholders::_1);
 			struct DnsContext *dctx = new struct DnsContext[2];
 			WFDnsTask *task_v4;
 			WFDnsTask *task_v6;
@@ -444,10 +453,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(host_, std::move(cb_v4));
 			task_v4->user_data = dctx;
 
-			task_v6 = client->create_dns_task(host_, dns_partial_callback);
+			task_v6 = client->create_dns_task(host_, std::move(cb_v6));
 			task_v6->get_req()->set_question_type(DNS_TYPE_AAAA);
 			task_v6->user_data = dctx + 1;
 
@@ -457,8 +466,11 @@ void WFResolverTask::dispatch()
 
 			pwork = Workflow::create_parallel_work(std::move(cb));
 			pwork->set_context(dctx);
-			pwork->add_series(Workflow::create_series_work(task_v4, nullptr));
-			pwork->add_series(Workflow::create_series_work(task_v6, nullptr));
+
+			WFConditional *cond_v4 = resolver->get_cond(task_v4);
+			WFConditional *cond_v6 = resolver->get_cond(task_v6);
+			pwork->add_series(Workflow::create_series_work(cond_v4, nullptr));
+			pwork->add_series(Workflow::create_series_work(cond_v6, nullptr));
 
 			series_of(this)->push_front(pwork);
 		}
@@ -526,6 +538,8 @@ void WFResolverTask::dns_callback_internal(DnsOutput *dns_out,
 
 void WFResolverTask::dns_single_callback(WFDnsTask *dns_task)
 {
+	WFGlobal::get_dns_resolver()->post_cond();
+
 	if (dns_task->get_state() == WFT_STATE_SUCCESS)
 	{
 		struct addrinfo *ai = NULL;
@@ -548,8 +562,9 @@ void WFResolverTask::dns_single_callback(WFDnsTask *dns_task)
 
 void WFResolverTask::dns_partial_callback(WFDnsTask *dns_task)
 {
-	struct DnsContext *ctx = (struct DnsContext *)dns_task->user_data;
+	WFGlobal::get_dns_resolver()->post_cond();
 
+	struct DnsContext *ctx = (struct DnsContext *)dns_task->user_data;
 	ctx->state = dns_task->get_state();
 	ctx->error = dns_task->get_error();
 	if (ctx->state == WFT_STATE_SUCCESS)
@@ -634,9 +649,9 @@ WFDnsResolver::create(const struct WFNSParams *params, int dns_cache_level,
 					  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));
+	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,
@@ -652,3 +667,8 @@ WFRouterTask *WFDnsResolver::create_router_task(const struct WFNSParams *params,
 				  endpoint_params, std::move(callback));
 }
 
+WFDnsResolver::WFDnsResolver() :
+	respool(WFGlobal::get_global_settings()->dns_server_params.max_connections)
+{
+}
+

src/nameservice/WFDnsResolver.h

@@ -21,6 +21,7 @@
 
 #include "EndpointParams.h"
 #include "WFNameService.h"
+#include "WFResourcePool.h"
 
 class WFDnsResolver : public WFNSPolicy
 {
@@ -28,11 +29,27 @@ public:
 	virtual WFRouterTask *create_router_task(const struct WFNSParams *params,
 											 router_callback_t callback);
 
-protected:
+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);
+
+private:
+	WFResourcePool respool;
+
+private:
+	WFConditional *get_cond(SubTask *task)
+	{
+		static void *buf;
+		return this->respool.get(task, &buf);
+	}
+
+	void post_cond() { this->respool.post(NULL); }
+
+public:
+	WFDnsResolver();
+	friend class WFResolverTask;
 };
 
 #endif

src/nameservice/WFServiceGovernance.cc

@@ -127,13 +127,14 @@ WFRouterTask *WFServiceGovernance::create_router_task(const struct WFNSParams *p
 	if (this->select(params->uri, tracing, &addr) &&
 		copy_host_port(params->uri, addr))
 	{
+		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;
-		task = this->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min,
-							endpoint_params, std::move(callback));
+		task = resolver->create(params, dns_cache_level, dns_ttl_default, dns_ttl_min,
+								endpoint_params, std::move(callback));
 
 		if (!tracing->data)
 			tracing->data = addr;
@@ -230,7 +231,7 @@ void WFServiceGovernance::success(RouteManager::RouteResult *result,
 	this->recover_server_from_breaker(server);
 	pthread_rwlock_unlock(&this->rwlock);
 
-	WFDnsResolver::success(result, tracing, target);
+	WFNSPolicy::success(result, tracing, target);
 }
 
 void WFServiceGovernance::failed(RouteManager::RouteResult *result,
@@ -253,7 +254,7 @@ void WFServiceGovernance::failed(RouteManager::RouteResult *result,
 
 	pthread_rwlock_unlock(&this->rwlock);
 
-	WFDnsResolver::failed(result, tracing, target);
+	WFNSPolicy::failed(result, tracing, target);
 }
 
 void WFServiceGovernance::check_breaker()

src/nameservice/WFServiceGovernance.h

@@ -102,7 +102,7 @@ public:
 	virtual ~EndpointAddress() { delete this->params; }
 };
 
-class WFServiceGovernance : public WFDnsResolver
+class WFServiceGovernance : public WFNSPolicy
 {
 public:
 	virtual WFRouterTask *create_router_task(const struct WFNSParams *params,