diff --git a/src/manager/WFGlobal.cc b/src/manager/WFGlobal.cc index 77dde95f..f00ff715 100644 --- a/src/manager/WFGlobal.cc +++ b/src/manager/WFGlobal.cc @@ -14,8 +14,8 @@ limitations under the License. Authors: Wu Jiaxu (wujiaxu@sogou-inc.com) - Liu Kai (liukaidx@sogou-inc.com) Xie Han (xiehan@sogou-inc.com) + Liu Kai (liukaidx@sogou-inc.com) */ #include @@ -35,13 +35,9 @@ #include #include #include "CommScheduler.h" -#include "DnsCache.h" -#include "RouteManager.h" #include "Executor.h" -#include "WFTask.h" +#include "WFResourcePool.h" #include "WFTaskError.h" -#include "WFNameService.h" -#include "WFDnsResolver.h" #include "WFDnsClient.h" #include "WFGlobal.h" @@ -54,16 +50,6 @@ public: return &kInstance; } - const WFGlobalSettings *get_global_settings() const - { - return &settings_; - } - - void set_global_settings(const WFGlobalSettings *settings) - { - settings_ = *settings; - } - const char *get_default_port(const std::string& scheme) { const auto it = static_scheme_port_.find(scheme); @@ -114,7 +100,6 @@ private: __WFGlobal(); private: - struct WFGlobalSettings settings_; std::unordered_map static_scheme_port_; std::unordered_map user_scheme_port_; std::mutex user_scheme_port_mutex_; @@ -123,7 +108,7 @@ private: int sync_max_; }; -__WFGlobal::__WFGlobal() : settings_(GLOBAL_SETTINGS_DEFAULT) +__WFGlobal::__WFGlobal() { static_scheme_port_["dns"] = "53"; static_scheme_port_["Dns"] = "53"; @@ -246,10 +231,10 @@ private: SSL_CTX *ssl_client_ctx_; }; -class IOServer : public IOService +class __FileIOService : public IOService { public: - IOServer(CommScheduler *scheduler): + __FileIOService(CommScheduler *scheduler): scheduler_(scheduler), flag_(true) {} @@ -298,13 +283,19 @@ private: bool flag_; }; -class __DnsManager +class __ThreadDnsManager { public: + static __ThreadDnsManager *get_instance() + { + static __ThreadDnsManager kInstance; + return &kInstance; + } + ExecQueue *get_dns_queue() { return &dns_queue_; } Executor *get_dns_executor() { return &dns_executor_; } - __DnsManager() + __ThreadDnsManager() { int ret; @@ -312,14 +303,12 @@ public: if (ret < 0) abort(); - ret = dns_executor_.init(__WFGlobal::get_instance()-> - get_global_settings()-> - dns_threads); + ret = dns_executor_.init(WFGlobal::get_global_settings()->dns_threads); if (ret < 0) abort(); } - ~__DnsManager() + ~__ThreadDnsManager() { dns_executor_.deinit(); dns_queue_.deinit(); @@ -340,49 +329,36 @@ public: } CommScheduler *get_scheduler() { return &scheduler_; } - RouteManager *get_route_manager() { return &route_manager_; } IOService *get_io_service() { - if (!io_flag_) + if (!fio_flag_) { - io_mutex_.lock(); - if (!io_flag_) + fio_mutex_.lock(); + if (!fio_flag_) { - io_server_ = new IOServer(&scheduler_); + fio_service_ = new __FileIOService(&scheduler_); //todo EAGAIN 65536->2 - if (io_server_->init(8192) < 0) + if (fio_service_->init(8192) < 0) abort(); - if (io_server_->bind() < 0) + if (fio_service_->bind() < 0) abort(); - io_flag_ = true; + fio_flag_ = true; } - io_mutex_.unlock(); + fio_mutex_.unlock(); } - return io_server_; - } - - ExecQueue *get_dns_queue() - { - return get_dns_manager_safe()->get_dns_queue(); - } - - Executor *get_dns_executor() - { - return get_dns_manager_safe()->get_dns_executor(); + return fio_service_; } private: __CommManager(): - io_server_(NULL), - io_flag_(false), - dns_manager_(NULL), - dns_flag_(false) + fio_service_(NULL), + fio_flag_(false) { - const auto *settings = __WFGlobal::get_instance()->get_global_settings(); + const auto *settings = WFGlobal::get_global_settings(); if (scheduler_.init(settings->poller_threads, settings->handler_threads) < 0) abort(); @@ -392,63 +368,20 @@ private: ~__CommManager() { - if (dns_manager_) - delete dns_manager_; - + // scheduler_.deinit() will triger fio_service to stop scheduler_.deinit(); - if (io_server_) + if (fio_service_) { - io_server_->deinit(); - delete io_server_; + fio_service_->deinit(); + delete fio_service_; } } - __DnsManager *get_dns_manager_safe() - { - if (!dns_flag_) - { - dns_mutex_.lock(); - if (!dns_flag_) - { - dns_manager_ = new __DnsManager(); - dns_flag_ = true; - } - - dns_mutex_.unlock(); - } - - return dns_manager_; - } - private: CommScheduler scheduler_; - RouteManager route_manager_; - IOServer *io_server_; - volatile bool io_flag_; - std::mutex io_mutex_; - __DnsManager *dns_manager_; - volatile bool dns_flag_; - std::mutex dns_mutex_; -}; - -class __DnsCache -{ -public: - static __DnsCache *get_instance() - { - static __DnsCache kInstance; - return &kInstance; - } - - DnsCache *get_dns_cache() { return &dns_cache_; } - -private: - __DnsCache() { } - - ~__DnsCache() { } - -private: - DnsCache dns_cache_; + __FileIOService *fio_service_; + volatile bool fio_flag_; + std::mutex fio_mutex_; }; class __ExecManager @@ -503,9 +436,7 @@ private: __ExecManager(): rwlock_(PTHREAD_RWLOCK_INITIALIZER) { - int compute_threads = __WFGlobal::get_instance()-> - get_global_settings()-> - compute_threads; + int compute_threads = WFGlobal::get_global_settings()->compute_threads; if (compute_threads <= 0) compute_threads = sysconf(_SC_NPROCESSORS_ONLN); @@ -531,27 +462,6 @@ private: Executor compute_executor_; }; -class __NameServiceManager -{ -public: - static __NameServiceManager *get_instance() - { - static __NameServiceManager kInstance; - return &kInstance; - } - -public: - WFDnsResolver *get_dns_resolver() { return &resolver_; } - WFNameService *get_name_service() { return &service_; } - -private: - WFDnsResolver resolver_; - WFNameService service_; - -public: - __NameServiceManager() : service_(&resolver_) { } -}; - #define MAX(x, y) ((x) >= (y) ? (x) : (y)) #define HOSTS_LINEBUF_INIT_SIZE 128 @@ -668,14 +578,16 @@ public: } public: - WFDnsClient *get_dns_client() { return client; } + WFDnsClient *get_dns_client() { return client_; } + WFResourcePool *get_dns_respool() { return &respool_; }; private: - __DnsClientManager() + __DnsClientManager() : respool_(WFGlobal::get_global_settings()-> + dns_server_params.max_connections) { const char *path = WFGlobal::get_global_settings()->resolv_conf_path; - client = NULL; + client_ = NULL; if (path && path[0]) { int ndots = 1; @@ -688,47 +600,39 @@ private: if (url.size() == 0) url = "8.8.8.8"; - client = new WFDnsClient; - if (client->init(url, search, ndots, attempts, rotate) >= 0) + client_ = new WFDnsClient; + if (client_->init(url, search, ndots, attempts, rotate) >= 0) return; - delete client; - client = NULL; + delete client_; + client_ = NULL; } } ~__DnsClientManager() { - if (client) + if (client_) { - client->deinit(); - delete client; + client_->deinit(); + delete client_; } } - WFDnsClient *client; + WFDnsClient *client_; + WFResourcePool respool_; }; -WFDnsClient *WFGlobal::get_dns_client() -{ - return __DnsClientManager::get_instance()->get_dns_client(); -} +struct WFGlobalSettings WFGlobal::settings_ = GLOBAL_SETTINGS_DEFAULT; +RouteManager WFGlobal::route_manager_; +DnsCache WFGlobal::dns_cache_; +WFDnsResolver WFGlobal::dns_resolver_; +WFNameService WFGlobal::name_service_(&WFGlobal::dns_resolver_); CommScheduler *WFGlobal::get_scheduler() { return __CommManager::get_instance()->get_scheduler(); } -DnsCache *WFGlobal::get_dns_cache() -{ - return __DnsCache::get_instance()->get_dns_cache(); -} - -RouteManager *WFGlobal::get_route_manager() -{ - return __CommManager::get_instance()->get_route_manager(); -} - SSL_CTX *WFGlobal::get_ssl_client_ctx() { return __SSLManager::get_instance()->get_ssl_client_ctx(); @@ -756,22 +660,22 @@ IOService *WFGlobal::get_io_service() ExecQueue *WFGlobal::get_dns_queue() { - return __CommManager::get_instance()->get_dns_queue(); + return __ThreadDnsManager::get_instance()->get_dns_queue(); } Executor *WFGlobal::get_dns_executor() { - return __CommManager::get_instance()->get_dns_executor(); + return __ThreadDnsManager::get_instance()->get_dns_executor(); } -WFNameService *WFGlobal::get_name_service() +WFDnsClient *WFGlobal::get_dns_client() { - return __NameServiceManager::get_instance()->get_name_service(); + return __DnsClientManager::get_instance()->get_dns_client(); } -WFDnsResolver *WFGlobal::get_dns_resolver() +WFResourcePool *WFGlobal::get_dns_respool() { - return __NameServiceManager::get_instance()->get_dns_resolver(); + return __DnsClientManager::get_instance()->get_dns_respool(); } const char *WFGlobal::get_default_port(const std::string& scheme) @@ -785,16 +689,6 @@ void WFGlobal::register_scheme_port(const std::string& scheme, __WFGlobal::get_instance()->register_scheme_port(scheme, port); } -const WFGlobalSettings *WFGlobal::get_global_settings() -{ - return __WFGlobal::get_instance()->get_global_settings(); -} - -void WORKFLOW_library_init(const WFGlobalSettings *settings) -{ - __WFGlobal::get_instance()->set_global_settings(settings); -} - void WFGlobal::sync_operation_begin() { __WFGlobal::get_instance()->sync_operation_begin(); @@ -973,3 +867,8 @@ const char *WFGlobal::get_error_string(int state, int error) return "Unknown"; } +void WORKFLOW_library_init(const struct WFGlobalSettings *settings) +{ + WFGlobal::set_global_settings(settings); +} + diff --git a/src/manager/WFGlobal.h b/src/manager/WFGlobal.h index b51005f8..26cfabd8 100644 --- a/src/manager/WFGlobal.h +++ b/src/manager/WFGlobal.h @@ -24,13 +24,14 @@ #include #endif -#include #include +#include #include "CommScheduler.h" #include "DnsCache.h" #include "RouteManager.h" #include "Executor.h" #include "EndpointParams.h" +#include "WFResourcePool.h" #include "WFNameService.h" #include "WFDnsResolver.h" #include "EndpointParams.h" @@ -95,7 +96,8 @@ public: * @param[in] port default port value * @warning No effect when scheme is "http"/"https"/"redis"/"rediss"/"mysql"/"kafka" */ - static void register_scheme_port(const std::string& scheme, unsigned short port); + static void register_scheme_port(const std::string& scheme, + unsigned short port); /** * @brief get default port string for one scheme string * @param[in] scheme scheme string @@ -109,15 +111,21 @@ public: * @return current global settings const pointer * @note returnval never NULL */ - static const struct WFGlobalSettings *get_global_settings(); + static const struct WFGlobalSettings *get_global_settings() + { + return &settings_; + } + + static void set_global_settings(const struct WFGlobalSettings *settings) + { + settings_ = *settings; + } static const char *get_error_string(int state, int error); // Internal usage only public: static class CommScheduler *get_scheduler(); - static class DnsCache *get_dns_cache(); - static class RouteManager *get_route_manager(); static SSL_CTX *get_ssl_client_ctx(); static SSL_CTX *new_ssl_server_ctx(); static class ExecQueue *get_exec_queue(const std::string& queue_name); @@ -125,13 +133,39 @@ public: static class IOService *get_io_service(); static class ExecQueue *get_dns_queue(); static class Executor *get_dns_executor(); - static class WFNameService *get_name_service(); - static class WFDnsResolver *get_dns_resolver(); static class WFDnsClient *get_dns_client(); + static class WFResourcePool *get_dns_respool(); + + static class RouteManager *get_route_manager() + { + return &route_manager_; + } + + static class DnsCache *get_dns_cache() + { + return &dns_cache_; + } + + static class WFDnsResolver *get_dns_resolver() + { + return &dns_resolver_; + } + + static class WFNameService *get_name_service() + { + return &name_service_; + } public: static void sync_operation_begin(); static void sync_operation_end(); + +private: + static struct WFGlobalSettings settings_; + static RouteManager route_manager_; + static DnsCache dns_cache_; + static WFDnsResolver dns_resolver_; + static WFNameService name_service_; }; #endif diff --git a/src/nameservice/WFDnsResolver.cc b/src/nameservice/WFDnsResolver.cc index 427d565c..13534928 100644 --- a/src/nameservice/WFDnsResolver.cc +++ b/src/nameservice/WFDnsResolver.cc @@ -295,7 +295,7 @@ void WFResolverTask::dispatch() if (addr_handle) { - auto *route_manager = WFGlobal::get_route_manager(); + RouteManager *route_manager = WFGlobal::get_route_manager(); struct addrinfo *addrinfo = addr_handle->value.addrinfo; struct addrinfo first; @@ -374,7 +374,7 @@ void WFResolverTask::dispatch() if (client) { static int family = __default_family(); - WFDnsResolver *resolver = WFGlobal::get_dns_resolver(); + WFResourcePool *respool = WFGlobal::get_dns_respool(); if (family == AF_INET || family == AF_INET6) { @@ -386,7 +386,7 @@ void WFResolverTask::dispatch() if (family == AF_INET6) dns_task->get_req()->set_question_type(DNS_TYPE_AAAA); - WFConditional *cond = resolver->respool.get(dns_task); + WFConditional *cond = respool->get(dns_task); series_of(this)->push_front(cond); } else @@ -415,8 +415,8 @@ void WFResolverTask::dispatch() pwork = Workflow::create_parallel_work(std::move(cb)); pwork->set_context(dctx); - WFConditional *cond_v4 = resolver->respool.get(task_v4); - WFConditional *cond_v6 = resolver->respool.get(task_v6); + WFConditional *cond_v4 = respool->get(task_v4); + WFConditional *cond_v6 = respool->get(task_v6); pwork->add_series(Workflow::create_series_work(cond_v4, nullptr)); pwork->add_series(Workflow::create_series_work(cond_v6, nullptr)); @@ -473,8 +473,8 @@ void WFResolverTask::dns_callback_internal(DnsOutput *dns_out, } else { - auto *route_manager = WFGlobal::get_route_manager(); - auto *dns_cache = WFGlobal::get_dns_cache(); + RouteManager *route_manager = WFGlobal::get_route_manager(); + DnsCache *dns_cache = WFGlobal::get_dns_cache(); struct addrinfo *addrinfo = dns_out->move_addrinfo(); const DnsCache::DnsHandle *addr_handle; @@ -496,7 +496,7 @@ void WFResolverTask::dns_callback_internal(DnsOutput *dns_out, void WFResolverTask::dns_single_callback(WFDnsTask *dns_task) { - WFGlobal::get_dns_resolver()->respool.post(NULL); + WFGlobal::get_dns_respool()->post(NULL); if (dns_task->get_state() == WFT_STATE_SUCCESS) { @@ -520,22 +520,19 @@ void WFResolverTask::dns_single_callback(WFDnsTask *dns_task) void WFResolverTask::dns_partial_callback(WFDnsTask *dns_task) { - WFGlobal::get_dns_resolver()->respool.post(NULL); + WFGlobal::get_dns_respool()->post(NULL); struct DnsContext *ctx = (struct DnsContext *)dns_task->user_data; + ctx->ai = NULL; ctx->state = dns_task->get_state(); ctx->error = dns_task->get_error(); if (ctx->state == WFT_STATE_SUCCESS) { - auto *resp = dns_task->get_resp(); - ctx->ai = NULL; + protocol::DnsResponse *resp = dns_task->get_resp(); ctx->eai_error = DnsUtil::getaddrinfo(resp, ctx->port, &ctx->ai); } else - { ctx->eai_error = EAI_NONAME; - ctx->ai = NULL; - } } void WFResolverTask::dns_parallel_callback(const ParallelWork *pwork) @@ -615,7 +612,7 @@ WFDnsResolver::create(const struct WFNSParams *params, int dns_cache_level, WFRouterTask *WFDnsResolver::create_router_task(const struct WFNSParams *params, router_callback_t callback) { - const auto *settings = WFGlobal::get_global_settings(); + 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; @@ -625,8 +622,3 @@ 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) -{ -} - diff --git a/src/nameservice/WFDnsResolver.h b/src/nameservice/WFDnsResolver.h index 163f2306..0cd52382 100644 --- a/src/nameservice/WFDnsResolver.h +++ b/src/nameservice/WFDnsResolver.h @@ -34,13 +34,6 @@ public: unsigned int dns_ttl_default, unsigned int dns_ttl_min, const struct EndpointParams *endpoint_params, router_callback_t&& callback); - -private: - WFResourcePool respool; - -public: - WFDnsResolver(); - friend class WFResolverTask; }; #endif