From 4649472a40ea74238913679790e97ad3183a962c Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 21 Jul 2025 15:14:16 +0800 Subject: [PATCH 1/2] Remove kafka '-fno-rtti' flag. (#1759) --- BUILD | 136 +++++++++++++------------------ src/factory/CMakeLists.txt | 1 - src/factory/xmake.lua | 1 - src/include/workflow/dns_types.h | 1 + src/protocol/CMakeLists.txt | 2 - src/protocol/KafkaDataTypes.cc | 8 ++ src/protocol/KafkaDataTypes.h | 6 ++ src/protocol/KafkaResult.cc | 6 -- src/protocol/KafkaResult.h | 6 +- src/protocol/xmake.lua | 17 +--- 10 files changed, 81 insertions(+), 103 deletions(-) create mode 120000 src/include/workflow/dns_types.h diff --git a/BUILD b/BUILD index cb791ed9..efaaa052 100644 --- a/BUILD +++ b/BUILD @@ -196,30 +196,6 @@ cc_library( visibility = ["//visibility:public"], ) -cc_library( - name = 'kafka_message', - hdrs = [ - 'src/factory/KafkaTaskImpl.inl', - 'src/protocol/KafkaDataTypes.h', - 'src/protocol/KafkaMessage.h', - 'src/protocol/KafkaResult.h', - 'src/protocol/kafka_parser.h', - ], - includes = [ - 'src/factory', - 'src/protocol', - ], - srcs = [ - 'src/factory/KafkaTaskImpl.cc', - 'src/protocol/KafkaMessage.cc', - 'src/protocol/KafkaResult.cc', - ], - copts = ['-fno-rtti'], - deps = [ - ':common', - ], -) - cc_library( name = 'kafka', hdrs = [ @@ -239,10 +215,12 @@ cc_library( 'src/client/WFKafkaClient.cc', 'src/protocol/KafkaDataTypes.cc', 'src/protocol/kafka_parser.c', + 'src/factory/KafkaTaskImpl.cc', + 'src/protocol/KafkaMessage.cc', + 'src/protocol/KafkaResult.cc', ], deps = [ ':common', - ':kafka_message', ], visibility = ["//visibility:public"], linkopts = [ @@ -276,37 +254,37 @@ cc_library( ) cc_binary( - name = 'helloworld', - srcs = ['tutorial/tutorial-00-helloworld.cc'], - deps = [':http'], + name = 'helloworld', + srcs = ['tutorial/tutorial-00-helloworld.cc'], + deps = [':http'], ) cc_binary( - name = 'wget', - srcs = ['tutorial/tutorial-01-wget.cc'], - deps = [':http'], + name = 'wget', + srcs = ['tutorial/tutorial-01-wget.cc'], + deps = [':http'], ) cc_binary( - name = 'redis_cli', - srcs = ['tutorial/tutorial-02-redis_cli.cc'], - deps = [':redis'], + name = 'redis_cli', + srcs = ['tutorial/tutorial-02-redis_cli.cc'], + deps = [':redis'], ) cc_binary( - name = 'wget_to_redis', - srcs = ['tutorial/tutorial-03-wget_to_redis.cc'], - deps = [':http', 'redis'], + name = 'wget_to_redis', + srcs = ['tutorial/tutorial-03-wget_to_redis.cc'], + deps = [':http', 'redis'], ) cc_binary( - name = 'http_echo_server', - srcs = ['tutorial/tutorial-04-http_echo_server.cc'], - deps = [':http'], + name = 'http_echo_server', + srcs = ['tutorial/tutorial-04-http_echo_server.cc'], + deps = [':http'], ) cc_binary( - name = 'http_proxy', - srcs = ['tutorial/tutorial-05-http_proxy.cc'], - deps = [':http'], + name = 'http_proxy', + srcs = ['tutorial/tutorial-05-http_proxy.cc'], + deps = [':http'], ) cc_binary( @@ -316,21 +294,21 @@ cc_binary( ) cc_binary( - name = 'sort_task', - srcs = ['tutorial/tutorial-07-sort_task.cc'], - deps = [':common'], + name = 'sort_task', + srcs = ['tutorial/tutorial-07-sort_task.cc'], + deps = [':common'], ) cc_binary( - name = 'matrix_multiply', - srcs = ['tutorial/tutorial-08-matrix_multiply.cc'], - deps = [':common'], + name = 'matrix_multiply', + srcs = ['tutorial/tutorial-08-matrix_multiply.cc'], + deps = [':common'], ) cc_binary( - name = 'http_file_server', - srcs = ['tutorial/tutorial-09-http_file_server.cc'], - deps = [':http'], + name = 'http_file_server', + srcs = ['tutorial/tutorial-09-http_file_server.cc'], + deps = [':http'], ) cc_library( @@ -340,45 +318,45 @@ cc_library( ) cc_binary( - name = 'server', - srcs = [ - 'tutorial/tutorial-10-user_defined_protocol/server.cc', - 'tutorial/tutorial-10-user_defined_protocol/message.cc', - ], - deps = [':common', ':user_hdrs'], + name = 'server', + srcs = [ + 'tutorial/tutorial-10-user_defined_protocol/server.cc', + 'tutorial/tutorial-10-user_defined_protocol/message.cc', + ], + deps = [':common', ':user_hdrs'], ) cc_binary( - name = 'client', - srcs = [ - 'tutorial/tutorial-10-user_defined_protocol/client.cc', - 'tutorial/tutorial-10-user_defined_protocol/message.cc', - ], - deps = [':common', ':user_hdrs'], + name = 'client', + srcs = [ + 'tutorial/tutorial-10-user_defined_protocol/client.cc', + 'tutorial/tutorial-10-user_defined_protocol/message.cc', + ], + deps = [':common', ':user_hdrs'], ) cc_binary( - name = 'graph_task', - srcs = ['tutorial/tutorial-11-graph_task.cc'], - deps = [':http'], + name = 'graph_task', + srcs = ['tutorial/tutorial-11-graph_task.cc'], + deps = [':http'], ) cc_binary( - name = 'mysql_cli', - srcs = ['tutorial/tutorial-12-mysql_cli.cc'], - deps = [':mysql'], + name = 'mysql_cli', + srcs = ['tutorial/tutorial-12-mysql_cli.cc'], + deps = [':mysql'], ) cc_binary( - name = 'kafka_cli', - srcs = ['tutorial/tutorial-13-kafka_cli.cc'], - deps = [':kafka', ':workflow_hdrs'], + name = 'kafka_cli', + srcs = ['tutorial/tutorial-13-kafka_cli.cc'], + deps = [':kafka', ':workflow_hdrs'], ) cc_binary( - name = 'consul_cli', - srcs = ['tutorial/tutorial-14-consul_cli.cc'], - deps = [':consul'], + name = 'consul_cli', + srcs = ['tutorial/tutorial-14-consul_cli.cc'], + deps = [':consul'], ) cc_binary( @@ -409,9 +387,9 @@ cc_binary( ) cc_binary( - name = 'redis_subscriber', - srcs = ['tutorial/tutorial-18-redis_subscriber.cc'], - deps = [':redis'], + name = 'redis_subscriber', + srcs = ['tutorial/tutorial-18-redis_subscriber.cc'], + deps = [':redis'], ) cc_binary( diff --git a/src/factory/CMakeLists.txt b/src/factory/CMakeLists.txt index c8efcaf7..6a247ba8 100644 --- a/src/factory/CMakeLists.txt +++ b/src/factory/CMakeLists.txt @@ -33,5 +33,4 @@ if (KAFKA STREQUAL "y") KafkaTaskImpl.cc ) add_library("factory_kafka" OBJECT ${SRC}) - set_property(SOURCE KafkaTaskImpl.cc APPEND PROPERTY COMPILE_OPTIONS "-fno-rtti") endif () diff --git a/src/factory/xmake.lua b/src/factory/xmake.lua index e1e76b2c..9e7aa559 100644 --- a/src/factory/xmake.lua +++ b/src/factory/xmake.lua @@ -13,7 +13,6 @@ target("kafka_factory") if has_config("kafka") then add_files("KafkaTaskImpl.cc") set_kind("object") - add_cxxflags("-fno-rtti") add_deps("factory") add_packages("zlib", "snappy", "zstd", "lz4") else diff --git a/src/include/workflow/dns_types.h b/src/include/workflow/dns_types.h new file mode 120000 index 00000000..666230ef --- /dev/null +++ b/src/include/workflow/dns_types.h @@ -0,0 +1 @@ +../../protocol/dns_types.h \ No newline at end of file diff --git a/src/protocol/CMakeLists.txt b/src/protocol/CMakeLists.txt index 9c777648..86e71a1e 100644 --- a/src/protocol/CMakeLists.txt +++ b/src/protocol/CMakeLists.txt @@ -43,6 +43,4 @@ if (KAFKA STREQUAL "y") KafkaResult.cc ) add_library("protocol_kafka" OBJECT ${SRC}) - set_property(SOURCE KafkaMessage.cc APPEND PROPERTY COMPILE_OPTIONS "-fno-rtti") - set_property(SOURCE KafkaResult.cc APPEND PROPERTY COMPILE_OPTIONS "-fno-rtti") endif () diff --git a/src/protocol/KafkaDataTypes.cc b/src/protocol/KafkaDataTypes.cc index 870366bb..9a0be8ae 100644 --- a/src/protocol/KafkaDataTypes.cc +++ b/src/protocol/KafkaDataTypes.cc @@ -664,4 +664,12 @@ const KafkaMeta *get_meta(const char *topic, KafkaMetaList *meta_list) return NULL; } +KafkaSnappySink::~KafkaSnappySink() +{ +} + +KafkaSnappySource::~KafkaSnappySource() +{ +} + } /* namespace protocol */ diff --git a/src/protocol/KafkaDataTypes.h b/src/protocol/KafkaDataTypes.h index 325b332b..ca6013a0 100644 --- a/src/protocol/KafkaDataTypes.h +++ b/src/protocol/KafkaDataTypes.h @@ -1609,6 +1609,9 @@ public: private: KafkaBuffer *buffer; + +public: + virtual ~KafkaSnappySink(); }; class KafkaSnappySource : public snappy::Source @@ -1643,6 +1646,9 @@ private: KafkaBuffer *buffer; size_t buf_size; size_t pos; + +public: + virtual ~KafkaSnappySource(); }; } diff --git a/src/protocol/KafkaResult.cc b/src/protocol/KafkaResult.cc index f573ab5b..6292ded4 100644 --- a/src/protocol/KafkaResult.cc +++ b/src/protocol/KafkaResult.cc @@ -27,12 +27,6 @@ enum KAFKA_STATUS_END, }; -KafkaResult::KafkaResult() -{ - this->resp_vec = NULL; - this->resp_num = 0; -} - KafkaResult::~KafkaResult() { delete []this->resp_vec; diff --git a/src/protocol/KafkaResult.h b/src/protocol/KafkaResult.h index d2d16318..3ce575d7 100644 --- a/src/protocol/KafkaResult.h +++ b/src/protocol/KafkaResult.h @@ -43,7 +43,11 @@ public: void set_resp(KafkaResponse&& resp, size_t i); public: - KafkaResult(); + KafkaResult() + { + this->resp_vec = NULL; + this->resp_num = 0; + } virtual ~KafkaResult(); diff --git a/src/protocol/xmake.lua b/src/protocol/xmake.lua index e2809936..cadd3ff8 100644 --- a/src/protocol/xmake.lua +++ b/src/protocol/xmake.lua @@ -36,23 +36,14 @@ target("protocol") set_kind("object") add_deps("basic_protocol", "mysql_protocol", "redis_protocol") -target("kafka_message") - if has_config("kafka") then - add_files("KafkaMessage.cc", - "KafkaResult.cc") - set_kind("object") - add_cxxflags("-fno-rtti") - add_packages("lz4", "zstd", "zlib", "snappy") - else - set_kind("phony") - end - target("kafka_protocol") if has_config("kafka") then set_kind("object") add_files("kafka_parser.c", - "KafkaDataTypes.cc") - add_deps("kafka_message", "protocol") + "KafkaDataTypes.cc", + "KafkaMessage.cc", + "KafkaResult.cc") + add_deps("basic_protocol") add_packages("zlib", "snappy", "zstd", "lz4") else set_kind("phony") From f20375dbc48c543ec675560e134c375118d0438b Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Mon, 21 Jul 2025 16:53:01 +0800 Subject: [PATCH 2/2] Merge vectors when sending on SSL connection. (#1758) --- src/kernel/Communicator.cc | 64 +++++++++++++++++++++++++++++++++++-- src/manager/RouteManager.cc | 22 +------------ src/server/WFServer.cc | 12 ------- 3 files changed, 62 insertions(+), 36 deletions(-) diff --git a/src/kernel/Communicator.cc b/src/kernel/Communicator.cc index 9abef974..f85445dd 100644 --- a/src/kernel/Communicator.cc +++ b/src/kernel/Communicator.cc @@ -119,6 +119,59 @@ static int __create_ssl(SSL_CTX *ssl_ctx, struct CommConnEntry *entry) return -1; } +#define SSL_WRITEV_BUFSIZE 2048 + +static int __ssl_writev(SSL *ssl, struct iovec vectors[], int cnt) +{ + if (vectors[0].iov_len < SSL_WRITEV_BUFSIZE && cnt > 1) + { + char *p = (char *)SSL_get_app_data(ssl); + size_t nleft = SSL_WRITEV_BUFSIZE; + size_t n; + int i; + + if (!p) + { + p = (char *)malloc(SSL_WRITEV_BUFSIZE); + if (!p) + return -1; + + if (SSL_set_app_data(ssl, p) <= 0) + { + free(p); + return -1; + } + } + + n = vectors[0].iov_len; + memcpy(p, vectors[0].iov_base, n); + vectors[0].iov_base = p; + + p += n; + nleft -= n; + for (i = 1; i < cnt; i++) + { + if (vectors[i].iov_len < nleft) + n = vectors[i].iov_len; + else + n = nleft; + + memcpy(p, vectors[i].iov_base, n); + vectors[i].iov_base = (char *)vectors[i].iov_base + n; + vectors[i].iov_len -= n; + + p += n; + nleft -= n; + if (nleft == 0) + break; + } + + vectors[0].iov_len = SSL_WRITEV_BUFSIZE - nleft; + } + + return SSL_write(ssl, vectors[0].iov_base, vectors[0].iov_len); +} + static void __release_conn(struct CommConnEntry *entry) { delete entry->conn; @@ -126,7 +179,10 @@ static void __release_conn(struct CommConnEntry *entry) pthread_mutex_destroy(&entry->mutex); if (entry->ssl) + { + free(SSL_get_app_data(entry->ssl)); SSL_free(entry->ssl); + } close(entry->sockfd); free(entry); @@ -446,7 +502,7 @@ int Communicator::send_message_sync(struct iovec vectors[], int cnt, } else if (vectors->iov_len > 0) { - n = SSL_write(entry->ssl, vectors->iov_base, vectors->iov_len); + n = __ssl_writev(entry->ssl, vectors, cnt); if (n <= 0) return cnt; } @@ -457,12 +513,14 @@ int Communicator::send_message_sync(struct iovec vectors[], int cnt, { if ((size_t)n >= vectors[i].iov_len) n -= vectors[i].iov_len; - else + else if (n > 0) { vectors[i].iov_base = (char *)vectors[i].iov_base + n; vectors[i].iov_len -= n; - break; + return cnt - i; } + else + break; } vectors += i; diff --git a/src/manager/RouteManager.cc b/src/manager/RouteManager.cc index 081e350f..6280ca39 100644 --- a/src/manager/RouteManager.cc +++ b/src/manager/RouteManager.cc @@ -78,27 +78,7 @@ private: #endif }; -class RouteTargetTCP_SSL : public RouteTargetTCP -{ -private: - virtual int create_connect_fd() - { - const struct sockaddr *addr; - socklen_t addrlen; - int sockfd; - - this->get_addr(&addr, &addrlen); - sockfd = socket(addr->sa_family, SOCK_STREAM, 0); - if (sockfd >= 0) - { - int nodelay = 1; - setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, - &nodelay, sizeof (int)); - } - - return sockfd; - } -}; +using RouteTargetTCP_SSL = RouteTargetTCP; /* To support TLS SNI. */ class RouteTargetTCP_TLS_SNI : public RouteTargetTCP_SSL diff --git a/src/server/WFServer.cc b/src/server/WFServer.cc index 7c275388..bfd7e194 100644 --- a/src/server/WFServer.cc +++ b/src/server/WFServer.cc @@ -183,18 +183,6 @@ WFConnection *WFServerBase::new_connection(int accept_fd) int reuse = 1; setsockopt(accept_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof (int)); - - if (this->get_ssl_ctx()) - { - if (this->params.transport_type == TT_TCP || - this->params.transport_type == TT_TCP_SSL) - { - int nodelay = 1; - setsockopt(accept_fd, IPPROTO_TCP, TCP_NODELAY, - &nodelay, sizeof (int)); - } - } - return new WFServerConnection(&this->conn_count); }