增加TCP Keepalive选项设置功能，避免部分fd因网络故障无法淘汰。

platform/include/internal/proxy.h

@@ -8,26 +8,14 @@ struct ssl_mgr;
 struct key_keeper;
 struct kni_acceptor;
 
-struct tfe_proxy
+struct tfe_proxy_tcp_options
 {
-	char name[TFE_SYMBOL_MAX];
+	int sz_rcv_buffer;
-	struct event_base * evbase;
+	int sz_snd_buffer;
-	struct event * sev[8];
+	int so_keepalive;
-	struct event * gcev;
+	int tcp_keepidle;
-
+	int tcp_keepintvl;
-	void * logger;
+	int tcp_keepcnt;
-
-	unsigned int nr_work_threads;
-	struct tfe_thread_ctx * work_threads[TFE_THREAD_MAX];
-
-	unsigned int nr_modules;
-	struct tfe_plugin * modules;
-
-	struct ssl_mgr * ssl_mgr_handler;
-	struct key_keeper * key_keeper_handler;
-	struct kni_acceptor * kni_acceptor_handler;
-
-	unsigned int tcp_all_passthrough;
 };
 
 struct tfe_proxy_accept_para
@@ -42,6 +30,31 @@ struct tfe_proxy_accept_para
 	bool passthrough;
 };
 
+struct tfe_proxy
+{
+	char name[TFE_SYMBOL_MAX];
+	struct event_base * evbase;
+	struct event * sev[8];
+	struct event * gcev;
+
+	void * logger;
+	unsigned int nr_work_threads;
+	struct tfe_thread_ctx * work_threads[TFE_THREAD_MAX];
+
+	unsigned int nr_modules;
+	struct tfe_plugin * modules;
+
+	struct ssl_mgr * ssl_mgr_handler;
+	struct key_keeper * key_keeper_handler;
+	struct kni_acceptor * kni_acceptor_handler;
+
+	/* DEBUG OPTIONS */
+	unsigned int tcp_all_passthrough;
+	struct tfe_proxy_tcp_options tcp_options;
+};
+
+
+
 struct tfe_thread_ctx * tfe_proxy_thread_ctx_acquire(struct tfe_proxy * ctx);
 void tfe_proxy_thread_ctx_release(struct tfe_thread_ctx * thread_ctx);
 

platform/src/proxy.cpp

@@ -182,8 +182,18 @@ __errout:
 
 int tfe_proxy_config(struct tfe_proxy * proxy, const char * profile)
 {
+	/* Worker threads */
 	MESA_load_profile_uint_def(profile, "main", "nr_worker_threads", &proxy->nr_work_threads, 1);
+	/* Debug */
 	MESA_load_profile_uint_def(profile, "debug", "passthrough_all_tcp", &proxy->tcp_all_passthrough, 0);
+	/* TCP options, -1 means unset, we should not call setsockopt */
+	MESA_load_profile_int_def(profile, "tcp", "sz_rcv_buffer", &proxy->tcp_options.sz_rcv_buffer, -1);
+	MESA_load_profile_int_def(profile, "tcp", "sz_snd_buffer", &proxy->tcp_options.sz_snd_buffer, -1);
+	MESA_load_profile_int_def(profile, "tcp", "so_keepalive", &proxy->tcp_options.so_keepalive, -1);
+	MESA_load_profile_int_def(profile, "tcp", "tcp_keepidle", &proxy->tcp_options.tcp_keepidle, -1);
+	MESA_load_profile_int_def(profile, "tcp", "tcp_keepintvl", &proxy->tcp_options.tcp_keepintvl, -1);
+	MESA_load_profile_int_def(profile, "tcp", "tcp_keepcnt", &proxy->tcp_options.tcp_keepcnt, -1);
+
 	return 0;
 }
 

platform/src/tcp_stream.cpp

@@ -30,6 +30,7 @@
 #include <ssl_stream.h>
 #include <tcp_stream.h>
 #include <proxy.h>
+#include <netinet/tcp.h>
 
 #ifndef TFE_CONFIG_OUTPUT_LIMIT_DEFAULT
 #define TFE_CONFIG_OUTPUT_LIMIT_DEFAULT                (1024 * 1024)
@@ -799,6 +800,74 @@ __errout:
 	return NULL;
 }
 
+void __stream_fd_option_setup(struct tfe_stream_private * _stream, evutil_socket_t fd)
+{
+	struct tfe_stream * stream = &_stream->head;
+	struct tfe_proxy_tcp_options * tcp_options = &_stream->proxy_ref->tcp_options;
+
+	/* Make it non-blocking */
+	evutil_make_socket_nonblocking(fd);
+
+	/* Recv Buffer */
+	if (tcp_options->sz_rcv_buffer >= 0)
+	{
+		if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (const void *) &tcp_options->sz_rcv_buffer, sizeof(int)) == -1)
+		{
+			TFE_STREAM_LOG_ERROR(stream, "setsockopt(SO_RCVBUF, %d) failed, ignored: %s",
+				tcp_options->sz_rcv_buffer, strerror(errno));
+		}
+	}
+
+	/* Send Buffer */
+	if (tcp_options->sz_snd_buffer >= 0)
+	{
+		if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (const void *) &tcp_options->sz_snd_buffer, sizeof(int)) == -1)
+		{
+			TFE_STREAM_LOG_ERROR(stream, "setsockopt(SO_SNDBUF, %d) failed, ignored: %s",
+				tcp_options->sz_snd_buffer, strerror(errno));
+		}
+	}
+
+	/* Keep-alive */
+	if (tcp_options->so_keepalive > 0)
+	{
+		if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (const void *) &tcp_options->so_keepalive, sizeof(int)) == -1)
+		{
+			TFE_STREAM_LOG_ERROR(stream, "setsockopt(SO_KEEPALIVE, %d) failed, ignored: %s",
+				tcp_options->so_keepalive, strerror(errno));
+		}
+	}
+
+	if (tcp_options->tcp_keepcnt > 0)
+	{
+		if (setsockopt(fd, SOL_SOCKET, TCP_KEEPCNT, (const void *) &tcp_options->tcp_keepcnt, sizeof(int)) == -1)
+		{
+			TFE_STREAM_LOG_ERROR(stream, "setsockopt(TCP_KEEPCNT, %d) failed, ignored: %s",
+				tcp_options->tcp_keepcnt, strerror(errno));
+		}
+	}
+
+	if (tcp_options->tcp_keepintvl > 0)
+	{
+		if (setsockopt(fd, SOL_SOCKET, TCP_KEEPINTVL, (const void *) &tcp_options->tcp_keepintvl, sizeof(int)) == -1)
+		{
+			TFE_STREAM_LOG_ERROR(stream, "setsockopt(TCP_KEEPINTVL, %d) failed, ignored: %s",
+				tcp_options->tcp_keepintvl, strerror(errno));
+		}
+	}
+
+	if (tcp_options->tcp_keepidle > 0)
+	{
+		if (setsockopt(fd, SOL_SOCKET, TCP_KEEPIDLE, (const void *) &tcp_options->tcp_keepidle, sizeof(int)) == -1)
+		{
+			TFE_STREAM_LOG_ERROR(stream, "setsockopt(TCP_KEEPIDLE, %d) failed, ignored: %s",
+				tcp_options->tcp_keepidle, strerror(errno));
+		}
+	}
+
+	return;
+}
+
 void 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);
@@ -807,8 +876,8 @@ void tfe_stream_init_by_fds(struct tfe_stream * stream, evutil_socket_t fd_downs
 	_stream->defer_fd_downstream = fd_downstream;
 	_stream->defer_fd_upstream = fd_upstream;
 
-	evutil_make_socket_nonblocking(fd_downstream);
+	__stream_fd_option_setup(_stream, fd_downstream);
-	evutil_make_socket_nonblocking(fd_upstream);
+	__stream_fd_option_setup(_stream, fd_upstream);
 
 	_stream->head.addr = __stream_addr_create_by_fds(stream, fd_downstream);
 	_stream->str_stream_addr = tfe_stream_addr_to_str(_stream->head.addr);