#64 key keeper增加evdnsbase参数。

避免创建大量dnsbase，耗尽fd。ssl stream/tcp stream/proxy也做了相应修改。

5 files changed, 54 insertions, 24 deletions

diff --git a/platform/src/key_keeper.cpp b/platform/src/key_keeper.cpp
index 189d81d..d573112 100644
--- a/platform/src/key_keeper.cpp
+++ b/platform/src/key_keeper.cpp
@@ -43,6 +43,8 @@ struct key_keeper
 	X509* untrusted_ca_cert;
 	EVP_PKEY* untrusted_ca_key;
 	unsigned int no_cache;
+	struct key_keeper_stat stat;
+
 };
 
 
@@ -601,7 +603,7 @@ char* url_escape(char* url)
 	return _url;
 }
 
-void key_keeper_async_ask(struct future * f, struct key_keeper * keeper, const char* sni, int keyring_id, X509 * origin_cert, int is_cert_valid, struct event_base * evbase)
+void key_keeper_async_ask(struct future * f, struct key_keeper * keeper, const char* sni, int keyring_id, X509 * origin_cert, int is_cert_valid, struct event_base * evbase, struct evdns_base* dnsbase)
 {
 	struct promise* p = future_to_promise(f);
 	unsigned int len = 0;
@@ -618,12 +620,14 @@ void key_keeper_async_ask(struct future * f, struct key_keeper * keeper, const c
 	ctx->key_len = len;
 	promise_set_ctx(p, (void*)ctx, key_keeper_promise_free_ctx);
 	long int cb_rtn = 0;
+	keeper->stat.ask_times++;
 	if(!keeper->no_cache)
 	{
 		MESA_htable_search_cb(keeper->cert_cache, (const unsigned char*)(ctx->key), ctx->key_len, keyring_local_cache_query_cb, p, &cb_rtn);
 		if(cb_rtn == KEYRING_EXSITED)
 		{
 			//printf("KEYRING_EXSITED\n");
+			keeper->stat.cache_hit++;
 			return;
 		}
 	}
@@ -644,7 +648,7 @@ void key_keeper_async_ask(struct future * f, struct key_keeper * keeper, const c
 				promise_failed(p, FUTURE_ERROR_EXCEPTION, "url escape failed");
 				break;
 			}
-    		struct future* f_certstore_rpc = future_create("tfe_rpc", certstore_rpc_on_succ, certstore_rpc_on_fail, p);
+    		struct future* f_certstore_rpc = future_create("crt_store", certstore_rpc_on_succ, certstore_rpc_on_fail, p);
 			ctx->f_certstore_rpc = f_certstore_rpc;
 			char *url = NULL;
 
@@ -661,7 +665,7 @@ void key_keeper_async_ask(struct future * f, struct key_keeper * keeper, const c
 			}
 			TFE_LOG_DEBUG(keeper->logger, "CertStore query: %.100s", url);
 			curl_free(escaped_origin_cert_pem);
-    		tfe_rpc_async_ask(f_certstore_rpc, url, GET, DONE_CB, NULL, 0, evbase);
+    		tfe_rpc_async_ask(f_certstore_rpc, url, GET, DONE_CB, NULL, 0, evbase, dnsbase);
 			free(url);
 			break;
         }
@@ -699,3 +703,10 @@ void key_keeper_async_ask(struct future * f, struct key_keeper * keeper, const c
 	}
 	return;
 }
+void key_keeper_statistic(struct key_keeper *keeper, struct key_keeper_stat* result)
+{
+	keeper->stat.cached_num=MESA_htable_get_elem_num(keeper->cert_cache);
+	*result=keeper->stat;
+	return;
+}
+
diff --git a/platform/src/kni_acceptor.cpp b/platform/src/kni_acceptor.cpp
index 6d22351..c022f4c 100644
--- a/platform/src/kni_acceptor.cpp
+++ b/platform/src/kni_acceptor.cpp
@@ -249,14 +249,13 @@ void __kni_event_cb(evutil_socket_t fd, short what, void * user)
 	}
 
 	__cmsghdr = CMSG_FIRSTHDR(&__msghdr);
-	__fds = (int *) (CMSG_DATA(__cmsghdr));
-
 	if (unlikely(__cmsghdr == NULL))
 	{
 		TFE_LOG_ERROR(__ctx->logger, "Failed at fetch CMSG_FIRSTHDR() from incoming fds, close KNI connection.");
 		goto __close_kni_connection;
 	}
-
+	
+	__fds = (int *) (CMSG_DATA(__cmsghdr));	
 	if (unlikely(__fds == NULL))
 	{
 		TFE_LOG_ERROR(__ctx->logger, "Failed at fetch CMSG_DATA() from incoming fds, close KNI connection.");
diff --git a/platform/src/proxy.cpp b/platform/src/proxy.cpp
index f5fd20f..1860581 100644
--- a/platform/src/proxy.cpp
+++ b/platform/src/proxy.cpp
@@ -16,6 +16,7 @@
 #include <pthread.h>
 
 #include <event2/event.h>
+#include <event2/dns.h>
 #include <event2/listener.h>
 #include <event2/bufferevent.h>
 #include <event2/bufferevent_ssl.h>
@@ -210,7 +211,8 @@ void tfe_proxy_work_thread_create_ctx(struct tfe_proxy * proxy)
 	{
 		proxy->work_threads[i] = ALLOC(struct tfe_thread_ctx, 1);
 		proxy->work_threads[i]->thread_id = i;
-		proxy->work_threads[i]->evbase = event_base_new();
+		proxy->work_threads[i]->evbase = event_base_new();	
+	    proxy->work_threads[i]->dnsbase = evdns_base_new(proxy->work_threads[i]->evbase, EVDNS_BASE_INITIALIZE_NAMESERVERS);
 	}
 	return;
 }
@@ -414,6 +416,12 @@ struct event_base * tfe_proxy_get_work_thread_evbase(unsigned int thread_id)
 	assert(thread_id < g_default_proxy->nr_work_threads);
 	return g_default_proxy->work_threads[thread_id]->evbase;
 }
+struct evdns_base* tfe_proxy_get_work_thread_dnsbase(unsigned int thread_id)
+{
+	assert(thread_id < g_default_proxy->nr_work_threads);
+	return g_default_proxy->work_threads[thread_id]->dnsbase;
+
+}
 
 struct event_base * tfe_proxy_get_gc_evbase(void)
 {
diff --git a/platform/src/ssl_stream.cpp b/platform/src/ssl_stream.cpp
index 0161843..eb9f919 100644
--- a/platform/src/ssl_stream.cpp
+++ b/platform/src/ssl_stream.cpp
@@ -90,6 +90,9 @@ enum ssl_stream_stat
 	SSL_NO_CHELLO,
 	SSL_NO_SNI,
 	SSL_FAKE_CRT,
+	KEY_KEEPER_CACHE_SIZE,
+	KEY_KEEPER_ASK,
+	KEY_KEEPER_HIT,
 	SSL_STAT_MAX
 };
 	
@@ -193,20 +196,20 @@ struct ssl_connect_server_ctx
 
 	evutil_socket_t fd_upstream;
 	evutil_socket_t fd_downstream;
-	struct event_base * evbase;
+	unsigned int thread_id;
 	struct future * f_peek_chello;
 	struct timespec start,end;
 };
 
 struct ssl_connect_client_ctx
 {
+	unsigned int thread_id;
 	int keyring_id;
 	struct ssl_stream * origin_ssl;
 	X509 * origin_crt;
 	int is_origin_crt_verify_passed;
 	struct ssl_mgr * ssl_mgr;
 	evutil_socket_t fd_downstream;
-	struct event_base * evbase;
 
 	struct future * f_ask_keyring;
 	struct bufferevent * bev_down;
@@ -241,6 +244,12 @@ ssl_stream_gc_cb(evutil_socket_t fd, short what, void * arg)
 	int i=0;
 	ssl_sess_cache_stat(mgr->up_sess_cache, &(mgr->stat_val[SSL_UP_CACHE_SZ]), &(mgr->stat_val[SSL_UP_CACHE_QUERY]), &(mgr->stat_val[SSL_UP_CACHE_HIT]));
 	ssl_sess_cache_stat(mgr->down_sess_cache, &(mgr->stat_val[SSL_DOWN_CACHE_SZ]), &(mgr->stat_val[SSL_DOWN_CACHE_QUERY]), &(mgr->stat_val[SSL_DOWN_CACHE_HIT]));
+	struct key_keeper_stat keeper_stat;
+	key_keeper_statistic(mgr->key_keeper, &keeper_stat);
+	mgr->stat_val[KEY_KEEPER_ASK]=keeper_stat.ask_times;
+	mgr->stat_val[KEY_KEEPER_HIT]=keeper_stat.cache_hit;
+	mgr->stat_val[KEY_KEEPER_CACHE_SIZE]=keeper_stat.cached_num;
+	
 	for(i=0;i<SSL_STAT_MAX;i++)
 	{
 		FS_operate(mgr->fs_handle, mgr->fs_id[i], 0, FS_OP_SET, ATOMIC_READ(&(mgr->stat_val[i])));
@@ -283,6 +292,9 @@ void ssl_stat_init(struct ssl_mgr * mgr)
 	spec[SSL_NO_CHELLO]="ssl_no_chlo";
 	spec[SSL_NO_SNI]="ssl_no_sni";
 	spec[SSL_FAKE_CRT]="ssl_fk_crt";
+	spec[KEY_KEEPER_ASK]="kyr_ask";
+	spec[KEY_KEEPER_HIT]="kyr_hit";
+	spec[KEY_KEEPER_CACHE_SIZE]="kyr_cache";
 
 	for(i=0;i<SSL_STAT_MAX;i++)
 	{
@@ -1034,6 +1046,7 @@ static void peek_chello_on_succ(future_result_t * result, void * user)
 {
 	struct promise * p = (struct promise *) user;
 	struct ssl_connect_server_ctx * ctx = (struct ssl_connect_server_ctx *) promise_get_ctx(p);
+	struct event_base* evbase=tfe_proxy_get_work_thread_evbase(ctx->thread_id);
 
 	struct ssl_chello * chello = ssl_peek_result_release_chello(result);//chello has been saved in ssl_stream.
 	if(chello->sni==NULL)
@@ -1042,7 +1055,7 @@ static void peek_chello_on_succ(future_result_t * result, void * user)
 	}
 	clock_gettime(CLOCK_MONOTONIC, &(ctx->start));
 	ctx->s_stream = ssl_stream_new(ctx->mgr, ctx->fd_upstream, CONN_DIR_UPSTREAM, chello, NULL, NULL);
-	ctx->bev = bufferevent_openssl_socket_new(ctx->evbase, ctx->fd_upstream,
+	ctx->bev = bufferevent_openssl_socket_new(evbase, ctx->fd_upstream,
 		ctx->s_stream->ssl, BUFFEREVENT_SSL_CONNECTING, BEV_OPT_DEFER_CALLBACKS);
 
 	bufferevent_openssl_set_allow_dirty_shutdown(ctx->bev, 1);
@@ -1065,7 +1078,7 @@ static void peek_chello_on_fail(enum e_future_error err, const char * what, void
 }
 
 void ssl_async_upstream_create(struct future * f, struct ssl_mgr * mgr, evutil_socket_t fd_upstream,
-	evutil_socket_t fd_downstream, struct event_base * evbase)
+	evutil_socket_t fd_downstream, unsigned int thread_id)
 {
 	struct promise * p = future_to_promise(f);
 	struct ssl_connect_server_ctx * ctx = ALLOC(struct ssl_connect_server_ctx, 1);
@@ -1079,10 +1092,10 @@ void ssl_async_upstream_create(struct future * f, struct ssl_mgr * mgr, evutil_s
 		promise_failed(p, FUTURE_ERROR_EXCEPTION, "upstream fd closed");
 		return;
 	}
-
+	struct event_base* evbase=tfe_proxy_get_work_thread_evbase(thread_id);
 	ctx->fd_downstream = fd_downstream;
 	ctx->fd_upstream = fd_upstream;
-	ctx->evbase = evbase;
+	ctx->thread_id = thread_id;
 	ctx->mgr = mgr;
 	promise_set_ctx(p, ctx, wrap_ssl_connect_server_ctx_free);
 
@@ -1498,16 +1511,14 @@ void ask_keyring_on_succ(void * result, void * user)
 
 	struct keyring * kyr = NULL;
 	struct ssl_mgr * mgr = ctx->ssl_mgr;
-
-
+	struct event_base* evbase=tfe_proxy_get_work_thread_evbase(ctx->thread_id);
 
 	kyr = key_keeper_release_keyring(result);	//kyr will be freed at ssl downstream closing.
-
 	
 	clock_gettime(CLOCK_MONOTONIC, &(ctx->start));
 	ctx->downstream = ssl_stream_new(mgr, ctx->fd_downstream, CONN_DIR_DOWNSTREAM, NULL, kyr,
 										ctx->origin_ssl?ctx->origin_ssl->alpn_selected:NULL);
-	ctx->bev_down = bufferevent_openssl_socket_new(ctx->evbase, ctx->fd_downstream, ctx->downstream->ssl,
+	ctx->bev_down = bufferevent_openssl_socket_new(evbase, ctx->fd_downstream, ctx->downstream->ssl,
 		BUFFEREVENT_SSL_ACCEPTING, BEV_OPT_DEFER_CALLBACKS);
 	bufferevent_openssl_set_allow_dirty_shutdown(ctx->bev_down, 1);
 	
@@ -1532,7 +1543,7 @@ void ask_keyring_on_fail(enum e_future_error error, const char * what, void * us
  * Create a SSL stream for the incoming connection, based on the upstream.
  */
 void ssl_async_downstream_create(struct future * f, struct ssl_mgr * mgr, struct ssl_stream * upstream,
-	evutil_socket_t fd_downstream, int keyring_id, struct event_base * evbase)
+	evutil_socket_t fd_downstream, int keyring_id, unsigned int thread_id)
 {
 
 	assert(upstream->dir == CONN_DIR_UPSTREAM);
@@ -1541,7 +1552,10 @@ void ssl_async_downstream_create(struct future * f, struct ssl_mgr * mgr, struct
 	ctx->keyring_id = keyring_id;
 	ctx->ssl_mgr = mgr;
 	ctx->fd_downstream = fd_downstream;
-	ctx->evbase = evbase;
+	ctx->thread_id = thread_id;
+	struct event_base * evbase=tfe_proxy_get_work_thread_evbase(thread_id);
+	struct evdns_base* dnsbase=tfe_proxy_get_work_thread_dnsbase(thread_id);
+
 	if (upstream != NULL)
 	{
 		ctx->origin_ssl = upstream;
@@ -1555,7 +1569,7 @@ void ssl_async_downstream_create(struct future * f, struct ssl_mgr * mgr, struct
 	ctx->f_ask_keyring = future_create("ask_kyr",ask_keyring_on_succ, ask_keyring_on_fail, p);
 	ctx->is_origin_crt_verify_passed = upstream->is_peer_cert_verify_passed;
 	key_keeper_async_ask(ctx->f_ask_keyring, mgr->key_keeper, sni, keyring_id, ctx->origin_crt, ctx->is_origin_crt_verify_passed,
-		evbase);
+		evbase, dnsbase);
 	return;
 }
 
diff --git a/platform/src/tcp_stream.cpp b/platform/src/tcp_stream.cpp
index b844027..247d579 100644
--- a/platform/src/tcp_stream.cpp
+++ b/platform/src/tcp_stream.cpp
@@ -763,7 +763,6 @@ void ssl_downstream_create_on_fail(enum e_future_error err, const char * what, v
 void ssl_upstream_create_on_success(future_result_t * result, void * user)
 {
 	struct tfe_stream_private * _stream = (struct tfe_stream_private *) user;
-	struct event_base * ev_base = _stream->thread_ref->evbase;
 
 	struct ssl_stream * upstream = ssl_upstream_create_result_release_stream(result);
 	struct bufferevent * bev = ssl_upstream_create_result_release_bev(result);
@@ -785,7 +784,7 @@ void ssl_upstream_create_on_success(future_result_t * result, void * user)
 		ssl_downstream_create_on_fail, _stream);
 
 	ssl_async_downstream_create(_stream->future_downstream_create, _stream->ssl_mgr,
-		_stream->ssl_upstream, _stream->defer_fd_downstream, _stream->keyring_id, ev_base);
+		_stream->ssl_upstream, _stream->defer_fd_downstream, _stream->keyring_id, _stream->thread_ref->thread_id);
 }
 
 void ssl_upstream_create_on_fail(enum e_future_error err, const char * what, void * user)
@@ -1021,7 +1020,6 @@ void __stream_fd_option_setup(struct tfe_stream_private * _stream, evutil_socket
 int tfe_stream_init_by_fds(struct tfe_stream * stream, evutil_socket_t fd_downstream, evutil_socket_t fd_upstream)
 {
 	struct tfe_stream_private * _stream = container_of(stream, struct tfe_stream_private, head);
-	struct event_base * ev_base = _stream->thread_ref->evbase;
 
 	_stream->defer_fd_downstream = fd_downstream;
 	_stream->defer_fd_upstream = fd_upstream;
@@ -1074,7 +1072,7 @@ int tfe_stream_init_by_fds(struct tfe_stream * stream, evutil_socket_t fd_downst
 
 		/* Defer setup conn_downstream & conn_upstream in async callbacks. */
 		ssl_async_upstream_create(_stream->future_upstream_create,
-			_stream->ssl_mgr, fd_upstream, fd_downstream, ev_base);
+			_stream->ssl_mgr, fd_upstream, fd_downstream, _stream->thread_ref->thread_id);
 
 		TFE_PROXY_STAT_INCREASE(STAT_STREAM_TCP_SSL, 1);
 	}