GithubStar/workflow
1
0
Fork 0
mirror of https://github.com/sogou/workflow.git synced 2026-10-22 23:39:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

migrate WFNameService to windows branch

Browse Source
This commit is contained in:
XieHan
2021-02-02 23:43:07 +08:00
parent c712b10718
commit f19c7f3ede
13 changed files with 422 additions and 1762 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists_Headers.txt
+3
View File
@@ -78,6 +78,7 @@ set(INCLUDE_HEADERS
src/manager/WFFuture.h
src/manager/WFFacilities.h
src/manager/WFFacilities.inl
src/manager/UpstreamPolicies.h
src/util/EncodeStream.h
src/util/LRUCache.h
src/util/StringUtil.h
@@ -95,5 +96,7 @@ set(INCLUDE_HEADERS
src/factory/WFAlgoTaskFactory.inl
src/factory/Workflow.h
src/factory/WFOperator.h
src/factory/WFNameService.h
src/factory/WFDNSResolver.h
)
src/factory/CMakeLists.txt
+2
View File
@@ -8,6 +8,8 @@ set(SRC
MySQLTaskImpl.cc
WFTaskFactory.cc
Workflow.cc
WFNameService.cc
WFDNSResolver.cc
)
add_library(${PROJECT_NAME} OBJECT ${SRC})
src/factory/MySQLTaskImpl.cc
+2 -2
View File
@@ -471,13 +471,13 @@ bool ComplexMySQLTask::init_success()
transaction_state_ = TRANSACTION_OUT;
this->WFComplexClientTask::set_info(std::string("?maxconn=1&") +
info + "|txn:" + transaction);
this->first_addr_only_ = true;
this->fixed_addr_ = true;
}
else
{
transaction_state_ = NO_TRANSACTION;
this->WFComplexClientTask::set_info(info);
this->first_addr_only_ = false;
this->fixed_addr_ = false;
}
delete []info;
src/factory/WFTaskFactory.cc
-202
View File
@@ -23,7 +23,6 @@
#include "PlatformSocket.h"
#include "list.h"
#include "rbtree.h"
#include "DNSRoutine.h"
#include "WFGlobal.h"
#include "WFTaskError.h"
#include "WFTaskFactory.h"
@@ -471,204 +470,3 @@ WFFileSyncTask *WFTaskFactory::create_fdsync_task(int fd,
#endif
/********RouterTask*************/
void WFRouterTask::dispatch()
{
insert_dns_ = true;
if (dns_cache_level_ != DNS_CACHE_LEVEL_0)
{
auto *dns_cache = WFGlobal::get_dns_cache();
const DNSHandle *addr_handle = NULL;
switch (dns_cache_level_)
{
case DNS_CACHE_LEVEL_1:
addr_handle = dns_cache->get_confident(host_, port_);
break;
case DNS_CACHE_LEVEL_2:
addr_handle = dns_cache->get_ttl(host_, port_);
break;
case DNS_CACHE_LEVEL_3:
addr_handle = dns_cache->get(host_, port_);
break;
default:
break;
}
if (addr_handle)
{
if (addr_handle->value.addrinfo)
{
auto *route_manager = WFGlobal::get_route_manager();
struct addrinfo *addrinfo = addr_handle->value.addrinfo;
struct addrinfo first;
if (first_addr_only_ && addrinfo->ai_next)
{
first = *addrinfo;
first.ai_next = NULL;
addrinfo = &first;
}
if (route_manager->get(type_, addrinfo,
info_, &endpoint_params_,
route_result_) < 0)
{
this->state = WFT_STATE_SYS_ERROR;
this->error = errno;
}
else if (!route_result_.request_object)
{
//should not happen
this->state = WFT_STATE_SYS_ERROR;
this->error = EAGAIN;
}
else
this->state = WFT_STATE_SUCCESS;
insert_dns_ = false;
}
dns_cache->release(addr_handle);
}
}
if (insert_dns_ && !host_.empty())
{
char front = host_.front();
char back = host_.back();
struct in6_addr addr;
int ret;
if (host_.find(':') != std::string::npos)
ret = inet_pton(AF_INET6, host_.c_str(), &addr);
else if (isdigit(back) && isdigit(front))
ret = inet_pton(AF_INET, host_.c_str(), &addr);
#if defined(AF_UNIX) && !defined(_WIN32)
else if (front == '/')
ret = 1;
#endif
else
ret = 0;
if (ret == 1)
{
DNSInput dns_in;
DNSOutput dns_out;
dns_in.reset(host_, port_);
DNSRoutine::run(&dns_in, &dns_out);
dns_callback_internal(&dns_out, (unsigned int)-1, (unsigned int)-1);
insert_dns_ = false;
}
}
if (insert_dns_)
{
auto&& cb = std::bind(&WFRouterTask::dns_callback,
this,
std::placeholders::_1);
WFDNSTask *dns_task = WFTaskFactory::create_dns_task(host_, port_,
std::move(cb));
series_of(this)->push_front(dns_task);
}
this->subtask_done();
}
SubTask* WFRouterTask::done()
{
SeriesWork *series = series_of(this);
if (!insert_dns_)
{
if (callback_)
callback_(this);
delete this;
}
return series->pop();
}
void WFRouterTask::dns_callback_internal(DNSOutput *dns_out,
unsigned int ttl_default,
unsigned int ttl_min)
{
int dns_error = dns_out->get_error();
if (dns_error)
{
#ifdef EAI_SYSTEM
if (dns_error == EAI_SYSTEM)
{
this->state = WFT_STATE_SYS_ERROR;
this->error = errno;
}
else
#endif
{
this->state = WFT_STATE_DNS_ERROR;
this->error = dns_error;
}
}
else
{
struct addrinfo *addrinfo = dns_out->move_addrinfo();
if (addrinfo)
{
auto *route_manager = WFGlobal::get_route_manager();
auto *dns_cache = WFGlobal::get_dns_cache();
const DNSHandle *addr_handle = dns_cache->put(host_, port_,
addrinfo,
(unsigned int)ttl_default,
(unsigned int)ttl_min);
if (route_manager->get(type_, addrinfo, info_, &endpoint_params_,
route_result_) < 0)
{
this->state = WFT_STATE_SYS_ERROR;
this->error = errno;
}
else if (!route_result_.request_object)
{
//should not happen
this->state = WFT_STATE_SYS_ERROR;
this->error = EAGAIN;
}
else
this->state = WFT_STATE_SUCCESS;
dns_cache->release(addr_handle);
}
else
{
//system promise addrinfo not null, here should not happen
this->state = WFT_STATE_SYS_ERROR;
this->error = EINVAL;
}
}
}
void WFRouterTask::dns_callback(WFDNSTask *dns_task)
{
if (dns_task->get_state() == WFT_STATE_SUCCESS)
dns_callback_internal(dns_task->get_output(), dns_ttl_default_, dns_ttl_min_);
else
{
this->state = dns_task->get_state();
this->error = dns_task->get_error();
}
if (callback_)
callback_(this);
delete this;
}
src/factory/WFTaskFactory.inl
+126 -345
View File
@@ -13,8 +13,8 @@
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)
Wu Jiaxu (wujiaxu@sogou-inc.com)
Li Yingxin (liyingxin@sogou-inc.com)
*/
@@ -30,11 +30,11 @@
#include "WFGlobal.h"
#include "Workflow.h"
#include "WFTask.h"
#include "UpstreamManager.h"
#include "RouteManager.h"
#include "URIParser.h"
#include "WFTaskError.h"
#include "EndpointParams.h"
#include "WFNameService.h"
class __WFTimerTask : public WFTimerTask
{
@@ -131,102 +131,6 @@ WFTaskFactory::create_dynamic_task(dynamic_create_t create)
return new __WFDynamicTask(std::move(create));
}
/**********WFComplexClientTask**********/
// If you design Derived WFComplexClientTask, You have two choices:
// 1) First choice will upstream by uri, then dns/dns-cache
// 2) Second choice will directly communicate without upstream/dns/dns-cache
// 1) First choice:
// step 1. Child-Constructor call Father-Constructor to new WFComplexClientTask
// step 2. call init(uri)
// step 3. call set_type(type)
// step 4. call set_info(info) or do nothing with info
// 2) Second choice:
// step 1. Child-Constructor call Father-Constructor to new WFComplexClientTask
// step 2. call init(type, addr, addrlen, info)
// Some optional APIs for you to implement:
// [WFComplexTask]
// [ChildrenComplexTask]
// 1. init()
// init_succ() or init_failed(); // default: return true;
// 2. dispatch();
// check_request(); // default: return true;
// route(); // default:DNS; goto 1;
// 3. message_out();
// 4. message_in();
// 5. keep_alive_timeout();
// 6. done();
// finish_once(); // default: return true; means this is user request.
// // If redirect or retry: goto 1;
/*
DNS_CACHE_LEVEL_0 -> NO cache
DNS_CACHE_LEVEL_1 -> TTL MIN
DNS_CACHE_LEVEL_2 -> TTL [DEFAULT]
DNS_CACHE_LEVEL_3 -> Forever
*/
#define DNS_CACHE_LEVEL_0 0
#define DNS_CACHE_LEVEL_1 1
#define DNS_CACHE_LEVEL_2 2
#define DNS_CACHE_LEVEL_3 3
class WFRouterTask : public WFGenericTask
{
private:
using router_callback_t = std::function<void (WFRouterTask *)>;
using WFDNSTask = WFThreadTask<DNSInput, DNSOutput>;
public:
RouteManager::RouteResult route_result_;
WFRouterTask(TransportType type,
const std::string& host,
unsigned short port,
const std::string& info,
int dns_cache_level,
unsigned int dns_ttl_default,
unsigned int dns_ttl_min,
const struct EndpointParams *endpoint_params,
bool first_addr_only,
router_callback_t&& callback) :
type_(type),
host_(host),
port_(port),
info_(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_(first_addr_only),
callback_(std::move(callback))
{}
private:
virtual void dispatch();
virtual SubTask *done();
void dns_callback(WFDNSTask *dns_task);
void dns_callback_internal(DNSOutput *dns_task,
unsigned int ttl_default,
unsigned int ttl_min);
private:
TransportType type_;
std::string host_;
unsigned short port_;
std::string info_;
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_;
router_callback_t callback_;
};
template<class REQ, class RESP, typename CTX = bool>
class WFComplexClientTask : public WFClientTask<REQ, RESP>
{
@@ -234,24 +138,24 @@ protected:
using task_callback_t = std::function<void (WFNetworkTask<REQ, RESP> *)>;
public:
WFComplexClientTask(int retry_max, task_callback_t&& callback):
WFClientTask<REQ, RESP>(NULL, WFGlobal::get_scheduler(),
std::move(callback)),
retry_max_(retry_max),
first_addr_only_(false),
router_task_(NULL),
type_(TT_TCP),
retry_times_(0),
is_retry_(false),
redirect_(false)
{}
WFComplexClientTask(int retry_max, task_callback_t&& cb):
WFClientTask<REQ, RESP>(NULL, WFGlobal::get_scheduler(), std::move(cb))
{
type_ = TT_TCP;
fixed_addr_ = false;
retry_max_ = retry_max;
retry_times_ = 0;
redirect_ = false;
ns_policy_ = NULL;
router_task_ = NULL;
}
protected:
// new api for children
virtual bool init_success() { return true; }
virtual void init_failed() {}
virtual bool check_request() { return true; }
virtual SubTask *route();
virtual WFRouterTask *route();
virtual bool finish_once() { return true; }
public:
@@ -272,16 +176,12 @@ public:
void init(const ParsedURI& uri)
{
is_sockaddr_ = false;
init_state_ = 0;
uri_ = uri;
init_with_uri();
}
void init(ParsedURI&& uri)
{
is_sockaddr_ = false;
init_state_ = 0;
uri_ = std::move(uri);
init_with_uri();
}
@@ -309,25 +209,6 @@ public:
}
protected:
void set_redirect()
{
redirect_ = true;
retry_times_ = 0;
}
void set_retry(const ParsedURI& uri)
{
redirect_ = true;
init(uri);
retry_times_++;
}
void set_retry()
{
redirect_ = true;
retry_times_++;
}
virtual void dispatch();
virtual SubTask *done();
@@ -347,13 +228,19 @@ protected:
TransportType get_transport_type() const { return type_; }
protected:
TransportType type_;
ParsedURI uri_;
int retry_max_;
bool is_sockaddr_;
bool first_addr_only_;
std::string info_;
bool fixed_addr_;
bool redirect_;
CTX ctx_;
SubTask *router_task_;
int retry_max_;
int retry_times_;
WFNSPolicy *ns_policy_;
WFRouterTask *router_task_;
RouteManager::RouteResult route_result_;
void *cookie_;
public:
CTX *get_mutable_ctx() { return &ctx_; }
@@ -361,21 +248,8 @@ public:
private:
void init_with_uri();
bool set_port();
void router_callback(SubTask *task); // default: DNS
void router_callback(WFRouterTask *task);
void switch_callback(WFTimerTask *task);
RouteManager::RouteResult route_result_;
UpstreamManager::UpstreamResult upstream_result_;
TransportType type_;
std::string info_;
int retry_times_;
/* state 0: uninited or failed; 1: inited but not checked; 2: checked. */
char init_state_;
bool is_retry_;
bool redirect_;
};
template<class REQ, class RESP, typename CTX>
@@ -384,42 +258,34 @@ void WFComplexClientTask<REQ, RESP, CTX>::init(TransportType type,
socklen_t addrlen,
const std::string& info)
{
is_sockaddr_ = true;
init_state_ = 0;
type_ = type;
info_.assign(info);
struct addrinfo addrinfo;
const auto *params = &WFGlobal::get_global_settings()->endpoint_params;
if (redirect_)
{
ns_policy_ = NULL;
route_result_.clear();
this->state = WFT_STATE_UNDEFINED;
this->error = 0;
this->timeout_reason = TOR_NOT_TIMEOUT;
}
addrinfo.ai_addrlen = addrlen;
addrinfo.ai_addr = (struct sockaddr *)addr;
addrinfo.ai_canonname = NULL;
addrinfo.ai_next = NULL;
addrinfo.ai_flags = 0;
const auto *params = &WFGlobal::get_global_settings()->endpoint_params;
struct addrinfo addrinfo = { };
addrinfo.ai_family = addr->sa_family;
addrinfo.ai_socktype = SOCK_STREAM;
addrinfo.ai_protocol = 0;
addrinfo.ai_addr = (struct sockaddr *)addr;
addrinfo.ai_addrlen = addrlen;
type_ = type;
info_.assign(info);
if (WFGlobal::get_route_manager()->get(type, &addrinfo, info_, params,
route_result_) < 0)
route_result_) < 0)
{
this->state = WFT_STATE_SYS_ERROR;
this->error = errno;
}
else if (!route_result_.request_object)
{
//should not happen
this->state = WFT_STATE_SYS_ERROR;
this->error = EAGAIN;
}
else
{
init_state_ = this->init_success() ? 1 : 0;
else if (this->init_success())
return;
}
this->init_failed();
return;
}
template<class REQ, class RESP, typename CTX>
@@ -469,170 +335,97 @@ bool WFComplexClientTask<REQ, RESP, CTX>::set_port()
template<class REQ, class RESP, typename CTX>
void WFComplexClientTask<REQ, RESP, CTX>::init_with_uri()
{
route_result_.clear();
if (uri_.state == URI_STATE_SUCCESS && this->set_port())
if (redirect_)
{
int ret = UpstreamManager::choose(uri_, upstream_result_);
ns_policy_ = NULL;
route_result_.clear();
this->state = WFT_STATE_UNDEFINED;
this->error = 0;
this->timeout_reason = TOR_NOT_TIMEOUT;
}
if (ret < 0)
if (uri_.state == URI_STATE_SUCCESS)
{
if (this->set_port())
{
this->state = WFT_STATE_SYS_ERROR;
this->error = errno;
}
else if (upstream_result_.state == UPSTREAM_ALL_DOWN)
{
this->state = WFT_STATE_TASK_ERROR;
this->error = WFT_ERR_UPSTREAM_UNAVAILABLE;
}
else
{
init_state_ = this->init_success() ? 1 : 0;
return;
if (this->init_success())
return;
}
}
else if (uri_.state == URI_STATE_ERROR)
{
this->state = WFT_STATE_SYS_ERROR;
this->error = uri_.error;
}
else
{
if (uri_.state == URI_STATE_ERROR)
{
this->state = WFT_STATE_SYS_ERROR;
this->error = uri_.error;
}
else
{
this->state = WFT_STATE_TASK_ERROR;
this->error = WFT_ERR_URI_PARSE_FAILED;
}
this->state = WFT_STATE_TASK_ERROR;
this->error = WFT_ERR_URI_PARSE_FAILED;
}
this->init_failed();
return;
}
template<class REQ, class RESP, typename CTX>
SubTask *WFComplexClientTask<REQ, RESP, CTX>::route()
WFRouterTask *WFComplexClientTask<REQ, RESP, CTX>::route()
{
unsigned int dns_ttl_default;
unsigned int dns_ttl_min;
const struct EndpointParams *endpoint_params;
int dns_cache_level = (is_retry_ ? DNS_CACHE_LEVEL_1
: DNS_CACHE_LEVEL_2);
WFNameService *ns = WFGlobal::get_name_service();
auto&& cb = std::bind(&WFComplexClientTask::router_callback,
this,
std::placeholders::_1);
is_retry_ = false;//route means refresh DNS cache level
if (upstream_result_.state == UPSTREAM_SUCCESS)
{
const auto *params = upstream_result_.address_params;
dns_ttl_default = params->dns_ttl_default;
dns_ttl_min = params->dns_ttl_min;
endpoint_params = &params->endpoint_params;
}
else
{
const auto *params = WFGlobal::get_global_settings();
dns_ttl_default = params->dns_ttl_default;
dns_ttl_min = params->dns_ttl_min;
endpoint_params = &params->endpoint_params;
}
return new WFRouterTask(type_, uri_.host ? uri_.host : "",
uri_.port ? atoi(uri_.port) : 0, info_,
dns_cache_level, dns_ttl_default, dns_ttl_min,
endpoint_params, first_addr_only_, std::move(cb));
struct WFNSParams params = {
.type = type_,
.uri = uri_,
.info = info_.c_str(),
.fixed_addr = fixed_addr_,
.retry_times = retry_times_,
};
ns_policy_ = ns->get_policy(uri_.host ? uri_.host : "");
return ns_policy_->create_router_task(&params, cb);
}
/*
* router callback`s obligation:
* if success:
* 1. set route_result_ or call this->init()
* 2. series->push_front(ORIGIN_TASK)
* if failed:
* 1. this->finish_once() is optional;
* 2. this->callback() is necessary;
*/
template<class REQ, class RESP, typename CTX>
void WFComplexClientTask<REQ, RESP, CTX>::router_callback(SubTask *task)
void WFComplexClientTask<REQ, RESP, CTX>::router_callback(WFRouterTask *task)
{
WFRouterTask *router_task = static_cast<WFRouterTask *>(task);
int state = router_task->get_state();
if (state == WFT_STATE_SUCCESS)
route_result_ = router_task->route_result_;
else
this->state = task->get_state();
if (this->state == WFT_STATE_SUCCESS)
{
this->state = state;
this->error = router_task->get_error();
route_result_ = std::move(*task->get_result());
cookie_ = task->get_cookie();
}
if (route_result_.request_object)
series_of(this)->push_front(this);
else
else if (this->state == WFT_STATE_UNDEFINED)
{
UpstreamManager::notify_unavailable(upstream_result_.cookie);
if (this->callback)
this->callback(this);
if (redirect_)
{
init_state_ = this->init_success() ? 1 : 0;
redirect_ = false;
this->state = WFT_STATE_UNDEFINED;
this->error = 0;
this->timeout_reason = TOR_NOT_TIMEOUT;
series_of(this)->push_front(this);
}
else
delete this;
/* should not happend */
this->state = WFT_STATE_SYS_ERROR;
this->error = ENOSYS;
}
else
this->error = task->get_error();
}
template<class REQ, class RESP, typename CTX>
void WFComplexClientTask<REQ, RESP, CTX>::dispatch()
{
// 1. children check_request()
if (init_state_ == 1)
init_state_ = this->check_request() ? 2 : 0;
if (init_state_)
switch (this->state)
{
if (route_result_.request_object)
case WFT_STATE_UNDEFINED:
if (this->check_request())
{
// 2. origin task dispatch()
this->set_request_object(route_result_.request_object);
this->WFClientTask<REQ, RESP>::dispatch();
return;
if (this->route_result_.request_object)
{
case WFT_STATE_SUCCESS:
this->set_request_object(route_result_.request_object);
this->WFClientTask<REQ, RESP>::dispatch();
return;
}
router_task_ = this->route();
series_of(this)->push_front(this);
series_of(this)->push_front(router_task_);
}
if (is_sockaddr_ || uri_.state == URI_STATE_SUCCESS)
{
// 3. DNS route() or children route()
router_task_ = this->route();
if (router_task_)
series_of(this)->push_front(router_task_);
else
{
this->state = WFT_STATE_TASK_ERROR;
this->error = WFT_ERR_ROUTE_FAILED;
}
}
else
{
if (uri_.state == URI_STATE_ERROR)
{
this->state = WFT_STATE_SYS_ERROR;
this->error = uri_.error;
}
else
{
this->state = WFT_STATE_TASK_ERROR;
this->error = WFT_ERR_URI_PARSE_FAILED;
}
}
default:
break;
}
this->subtask_done();
@@ -649,21 +442,15 @@ void WFComplexClientTask<REQ, RESP, CTX>::switch_callback(WFTimerTask *)
this->error = -this->error;
}
// 4. children finish before user callback
if (this->callback)
this->callback(this);
}
if (redirect_)
{
init_state_ = this->init_success() ? 1 : 0;
redirect_ = false;
clear_resp();
this->target = NULL;
this->timeout_reason = TOR_NOT_TIMEOUT;
this->state = WFT_STATE_UNDEFINED;
this->error = 0;
series_of(this)->push_front(this);
}
else
@@ -675,48 +462,43 @@ SubTask *WFComplexClientTask<REQ, RESP, CTX>::done()
{
SeriesWork *series = series_of(this);
// 1. routing
if (router_task_)
{
router_task_ = NULL;
return series->pop();
}
if (init_state_)
{
// 2. children can set_redirect() here
bool is_user_request = this->finish_once();
// 3. complex task success
if (this->state == WFT_STATE_SUCCESS)
{
RouteManager::notify_available(route_result_.cookie, this->target);
UpstreamManager::notify_available(upstream_result_.cookie);
upstream_result_.clear();
// 4. children message out sth. else
if (!is_user_request)
return this;
}
else if (this->state == WFT_STATE_SYS_ERROR)
{
RouteManager::notify_unavailable(route_result_.cookie, this->target);
UpstreamManager::notify_unavailable(upstream_result_.cookie);
// 5. complex task failed: retry
if (retry_times_ < retry_max_)
{
if (is_sockaddr_)
set_retry();
else
set_retry(uri_);
bool is_user_request = this->finish_once();
is_retry_ = true; // will influence next round dns cache time
}
if (ns_policy_ && route_result_.request_object)
{
if (this->state == WFT_STATE_SYS_ERROR)
ns_policy_->failed(&route_result_, cookie_, this->target);
else
ns_policy_->success(&route_result_, cookie_, this->target);
}
if (this->state == WFT_STATE_SUCCESS)
{
if (!is_user_request)
return this;
}
else if (this->state == WFT_STATE_SYS_ERROR)
{
if (retry_times_ < retry_max_)
{
redirect_ = true;
this->state = WFT_STATE_UNDEFINED;
this->error = 0;
this->timeout_reason = 0;
retry_times_++;
}
}
/*
* When target is NULL, it's very likely that we are still in the
* 'dispatch' thread. Running a timer will switch callback function
* to a handler thread, and this can prevent stack overflow.
* When target is NULL, it's very likely that we are in the caller's
* thread or DNS thread (dns failed). Running a timer will switch callback
* function to a handler thread, and this can prevent stack overflow.
*/
if (!this->target)
{
@@ -724,7 +506,6 @@ SubTask *WFComplexClientTask<REQ, RESP, CTX>::done()
this,
std::placeholders::_1);
WFTimerTask *timer = WFTaskFactory::create_timer_task(0, std::move(cb));
series->push_front(timer);
}
else
src/manager/CMakeLists.txt
+1
View File
@@ -6,6 +6,7 @@ set(SRC
UpstreamManager.cc
RouteManager.cc
WFGlobal.cc
UpstreamPolicies.cc
)
add_library(${PROJECT_NAME} OBJECT ${SRC})
src/manager/DNSCache.cc
+5 -5
View File
@@ -25,7 +25,7 @@
#define CONFIDENT_INC 10
#define TTL_INC 10
const DNSHandle *DNSCache::get_inner(const HostPort& host_port, int type)
const DNSCache::DNSHandle *DNSCache::get_inner(const HostPort& host_port, int type)
{
const DNSHandle *handle = cache_pool_.get(host_port);
@@ -73,10 +73,10 @@ const DNSHandle *DNSCache::get_inner(const HostPort& host_port, int type)
return handle;
}
const DNSHandle *DNSCache::put(const HostPort& host_port,
struct addrinfo *addrinfo,
unsigned int dns_ttl_default,
unsigned int dns_ttl_min)
const DNSCache::DNSHandle *DNSCache::put(const HostPort& host_port,
struct addrinfo *addrinfo,
unsigned int dns_ttl_default,
unsigned int dns_ttl_min)
{
int64_t expire_time;
int64_t confident_time;
src/manager/DNSCache.h
+77 -111
View File
@@ -45,36 +45,75 @@ public:
}
};
typedef std::pair<std::string, unsigned short> HostPort;
typedef LRUHandle<HostPort, DNSCacheValue> DNSHandle;
// RAII: NO. Release handle by user
// Thread safety: YES
// MUST call release when handle no longer used
class DNSCache
{
public:
using HostPort = std::pair<std::string, unsigned short>;
using DNSHandle = LRUHandle<HostPort, DNSCacheValue>;
public:
// release handle by get/put
void release(DNSHandle *handle);
void release(const DNSHandle *handle);
void release(DNSHandle *handle)
{
cache_pool_.release(handle);
}
void release(const DNSHandle *handle)
{
cache_pool_.release(handle);
}
// get handler
// Need call release when handle no longer needed
//Handle *get(const KEY &key);
const DNSHandle *get(const HostPort& host_port);
const DNSHandle *get(const std::string& host, unsigned short port);
const DNSHandle *get(const char *host, unsigned short port);
const DNSHandle *get(const HostPort& host_port)
{
return cache_pool_.get(host_port);
}
const DNSHandle *get_ttl(const HostPort& host_port);
const DNSHandle *get_ttl(const std::string& host, unsigned short port);
const DNSHandle *get_ttl(const char *host, unsigned short port);
const DNSHandle *get(const std::string& host, unsigned short port)
{
return get(HostPort(host, port));
}
const DNSHandle *get_confident(const HostPort& host_port);
const DNSHandle *get_confident(const std::string& host, unsigned short port);
const DNSHandle *get_confident(const char *host, unsigned short port);
const DNSHandle *get(const char *host, unsigned short port)
{
return get(std::string(host), port);
}
const DNSHandle *get_ttl(const HostPort& host_port)
{
return get_inner(host_port, GET_TYPE_TTL);
}
const DNSHandle *get_ttl(const std::string& host, unsigned short port)
{
return get_ttl(HostPort(host, port));
}
const DNSHandle *get_ttl(const char *host, unsigned short port)
{
return get_ttl(std::string(host), port);
}
const DNSHandle *get_confident(const HostPort& host_port)
{
return get_inner(host_port, GET_TYPE_CONFIDENT);
}
const DNSHandle *get_confident(const std::string& host, unsigned short port)
{
return get_confident(HostPort(host, port));
}
const DNSHandle *get_confident(const char *host, unsigned short port)
{
return get_confident(std::string(host), port);
}
// put copy
// Need call release when handle no longer needed
const DNSHandle *put(const HostPort& host_port,
struct addrinfo *addrinfo,
unsigned int dns_ttl_default,
@@ -84,18 +123,35 @@ public:
unsigned short port,
struct addrinfo *addrinfo,
unsigned int dns_ttl_default,
unsigned int dns_ttl_min);
unsigned int dns_ttl_min)
{
return put(HostPort(host, port), addrinfo, dns_ttl_default, dns_ttl_min);
}
const DNSHandle *put(const char *host,
unsigned short port,
struct addrinfo *addrinfo,
unsigned int dns_ttl_default,
unsigned int dns_ttl_min);
unsigned int dns_ttl_min)
{
return put(std::string(host), port, addrinfo, dns_ttl_default, dns_ttl_min);
}
// delete from cache, deleter delay called when all inuse-handle release.
void del(const HostPort& key);
void del(const std::string& host, unsigned short port);
void del(const char *host, unsigned short port);
void del(const HostPort& key)
{
cache_pool_.del(key);
}
void del(const std::string& host, unsigned short port)
{
del(HostPort(host, port));
}
void del(const char *host, unsigned short port)
{
del(std::string(host), port);
}
private:
const DNSHandle *get_inner(const HostPort& host_port, int type);
@@ -104,95 +160,5 @@ private:
LRUCache<HostPort, DNSCacheValue, ValueDeleter> cache_pool_;
};
////////////////////
inline void DNSCache::release(DNSHandle *handle)
{
cache_pool_.release(handle);
}
inline void DNSCache::release(const DNSHandle *handle)
{
cache_pool_.release(handle);
}
inline const DNSHandle *DNSCache::get(const HostPort& host_port)
{
return cache_pool_.get(host_port);
}
inline const DNSHandle *DNSCache::get(const std::string& host, unsigned short port)
{
return get(HostPort(host, port));
}
inline const DNSHandle *DNSCache::get(const char *host, unsigned short port)
{
return get(std::string(host), port);
}
inline const DNSHandle *DNSCache::get_ttl(const HostPort& host_port)
{
return get_inner(host_port, GET_TYPE_TTL);
}
inline const DNSHandle *DNSCache::get_ttl(const std::string& host, unsigned short port)
{
return get_ttl(HostPort(host, port));
}
inline const DNSHandle *DNSCache::get_ttl(const char *host, unsigned short port)
{
return get_ttl(std::string(host), port);
}
inline const DNSHandle *DNSCache::get_confident(const HostPort& host_port)
{
return get_inner(host_port, GET_TYPE_CONFIDENT);
}
inline const DNSHandle *DNSCache::get_confident(const std::string& host, unsigned short port)
{
return get_confident(HostPort(host, port));
}
inline const DNSHandle *DNSCache::get_confident(const char *host, unsigned short port)
{
return get_confident(std::string(host), port);
}
inline const DNSHandle *DNSCache::put(const std::string& host,
unsigned short port,
struct addrinfo *addrinfo,
unsigned int dns_ttl_default,
unsigned int dns_ttl_min)
{
return put(HostPort(host, port), addrinfo, dns_ttl_default, dns_ttl_min);
}
inline const DNSHandle *DNSCache::put(const char *host,
unsigned short port,
struct addrinfo *addrinfo,
unsigned int dns_ttl_default,
unsigned int dns_ttl_min)
{
return put(std::string(host), port, addrinfo, dns_ttl_default, dns_ttl_min);
}
inline void DNSCache::del(const HostPort& key)
{
cache_pool_.del(key);
}
inline void DNSCache::del(const std::string& host, unsigned short port)
{
del(HostPort(host, port));
}
inline void DNSCache::del(const char *host, unsigned short port)
{
del(std::string(host), port);
}
#endif
src/manager/RouteManager.h
+1 -1
View File
@@ -37,7 +37,7 @@ public:
CommSchedObject *request_object;
public:
// RouteResult(): cookie(NULL), request_object(NULL) { }
RouteResult(): cookie(NULL), request_object(NULL) { }
void clear() { cookie = NULL; request_object = NULL; }
};
src/manager/UpstreamManager.cc
+113 -1014
View File
File diff suppressed because it is too large Load Diff
src/manager/UpstreamManager.h
+13 -45
View File
@@ -23,6 +23,7 @@
#include <functional>
#include "URIParser.h"
#include "EndpointParams.h"
#include "WFGlobal.h"
/**
* @file UpstreamManager.h
@@ -58,9 +59,9 @@ struct AddressParams
* - If max_fails is set to 1, it means server would out of upstream selection in 30 seconds when failed only once
*/
unsigned int max_fails; ///< [1, INT32_MAX] max_fails = 0 means max_fails = 1
unsigned short weight; ///< [1, 65535] weight = 0 means weight = 1. only for main
unsigned short weight; ///< [1, 65535] weight = 0 means weight = 1. only for main server
int server_type; ///< 0 for main and 1 for backup
int group_id; ///< -1 means no group. Backup without group will backup for any main
int group_id; ///< -1 means no group. Backup without group will be backup for any main
};
/**
@@ -68,13 +69,13 @@ struct AddressParams
*/
static constexpr struct AddressParams ADDRESS_PARAMS_DEFAULT =
{
/* .endpoint_params = */ ENDPOINT_PARAMS_DEFAULT,
/* .dns_ttl_default = */ 12 * 3600,
/* .dns_ttl_min = */ 180,
/* .max_fails = */ 200,
/* .weight = */ 1,
/* .server_type = */ 0, /* 0 for main and 1 for backup */
/* .group_id = */ -1
.endpoint_params = ENDPOINT_PARAMS_DEFAULT,
.dns_ttl_default = 12 * 3600,
.dns_ttl_min = 180,
.max_fails = 200,
.weight = 1,
.server_type = 0, /* 0 for main and 1 for backup. */
.group_id = -1,
};
/**
@@ -84,7 +85,7 @@ static constexpr struct AddressParams ADDRESS_PARAMS_DEFAULT =
* 1. Weighted-Random
* 2. Consistent-Hash
* 3. Manual-Select
* - Additional, we support Main-Backup & Group for server and working well in any mode.
* - Additional, we support Main-backup & Group for server and working well in any mode.
*
* @code{.cc}
upstream_create_weighted_random("abc.sogou", true); //UPSTREAM_WIGHTED_RANDOM
@@ -218,9 +219,9 @@ public:
const std::string& address);
/**
* @brief get all main address list from one upstream
* @brief get all main servers address list from one upstream
* @param[in] name upstream name
* @return all main address list
* @return all main servers' address list
* @warning If server servers has the same address in this upstream, then will appear in the vector multiply times
*/
static std::vector<std::string> upstream_main_address_list(const std::string& name);
@@ -234,39 +235,6 @@ public:
const std::string& address,
const struct AddressParams *address_params);
public:
/// @brief Internal use only
class UpstreamResult
{
public:
void *cookie;
const struct AddressParams *address_params;
#define UPSTREAM_SUCCESS 0
#define UPSTREAM_NOTFOUND 1
#define UPSTREAM_ALL_DOWN 2
int state;
public:
UpstreamResult():
cookie(NULL),
address_params(NULL),
state(UPSTREAM_NOTFOUND)
{}
void clear()
{
cookie = NULL;
address_params = NULL;
state = UPSTREAM_NOTFOUND;
}
};
/// @brief Internal use only
static int choose(ParsedURI& uri, UpstreamResult& result);
/// @brief Internal use only
static void notify_unavailable(void *cookie);
/// @brief Internal use only
static void notify_available(void *cookie);
};
#endif
src/manager/WFGlobal.cc
+76 -26
View File
@@ -38,8 +38,8 @@
#include "RWLock.h"
#include "WFTask.h"
#include "WFTaskError.h"
const char *library_version = "Workflow Library Version: 1.12.6";
#include "WFNameService.h"
#include "WFDNSResolver.h"
class __WFGlobal
{
@@ -104,30 +104,7 @@ public:
}
private:
__WFGlobal():
settings_(GLOBAL_SETTINGS_DEFAULT)
{
static_scheme_port_["http"] = "80";
static_scheme_port_["https"] = "443";
static_scheme_port_["redis"] = "6379";
static_scheme_port_["rediss"] = "6379";
static_scheme_port_["mysql"] = "3306";
static_scheme_port_["kafka"] = "9092";
sync_count_ = 0;
sync_max_ = 0;
#ifdef _WIN32
WSADATA wsaData;
WORD wVersionRequested = MAKEWORD(2, 2);
int err = WSAStartup(wVersionRequested, &wsaData);
if (err != 0)
abort();
if (LOBYTE(wsaData.wVersion) != 2 || HIBYTE(wsaData.wVersion) != 2)
abort();
#endif
}
__WFGlobal();
#ifdef _WIN32
~__WFGlobal()
@@ -146,6 +123,52 @@ private:
int sync_max_;
};
__WFGlobal::__WFGlobal() : settings_(GLOBAL_SETTINGS_DEFAULT)
{
static_scheme_port_["http"] = "80";
static_scheme_port_["Http"] = "80";
static_scheme_port_["HTTP"] = "80";
static_scheme_port_["https"] = "443";
static_scheme_port_["Https"] = "443";
static_scheme_port_["HTTPs"] = "443";
static_scheme_port_["HTTPS"] = "443";
static_scheme_port_["redis"] = "6379";
static_scheme_port_["Redis"] = "6379";
static_scheme_port_["REDIS"] = "6379";
static_scheme_port_["rediss"] = "6379";
static_scheme_port_["Rediss"] = "6379";
static_scheme_port_["REDISs"] = "6379";
static_scheme_port_["REDISS"] = "6379";
static_scheme_port_["mysql"] = "3306";
static_scheme_port_["Mysql"] = "3306";
static_scheme_port_["MySql"] = "3306";
static_scheme_port_["MySQL"] = "3306";
static_scheme_port_["MYSQL"] = "3306";
static_scheme_port_["kafka"] = "9092";
static_scheme_port_["Kafka"] = "9092";
static_scheme_port_["KAFKA"] = "9092";
sync_count_ = 0;
sync_max_ = 0;
#ifdef _WIN32
WSADATA wsaData;
WORD wVersionRequested = MAKEWORD(2, 2);
int err = WSAStartup(wVersionRequested, &wsaData);
if (err != 0)
abort();
if (LOBYTE(wsaData.wVersion) != 2 || HIBYTE(wsaData.wVersion) != 2)
abort();
#endif
}
static __WFGlobal *_g_global = __WFGlobal::get_instance();
#if OPENSSL_VERSION_NUMBER < 0x10100000L
@@ -514,6 +537,28 @@ private:
Executor compute_executor_;
};
class __NameServiceManager
{
public:
static __NameServiceManager *get_instance()
{
static __NameServiceManager kInstance;
return &kInstance;
}
public:
WFNameService *get_name_service() { return &service_; }
private:
static WFDNSResolver resolver_;
WFNameService service_;
public:
__NameServiceManager() : service_(&__NameServiceManager::resolver_) { }
};
WFDNSResolver __NameServiceManager::resolver_;
CommScheduler *WFGlobal::get_scheduler()
{
return __CommManager::get_instance()->get_scheduler();
@@ -564,6 +609,11 @@ Executor *WFGlobal::get_dns_executor()
return __CommManager::get_instance()->get_dns_executor();
}
WFNameService *WFGlobal::get_name_service()
{
return __NameServiceManager::get_instance()->get_name_service();
}
const char *WFGlobal::get_default_port(const std::string& scheme)
{
return __WFGlobal::get_instance()->get_default_port(scheme);
src/manager/WFGlobal.h
+3 -11
View File
@@ -33,6 +33,7 @@
#include "RouteManager.h"
#include "Executor.h"
#include "EndpointParams.h"
#include "WFNameService.h"
/**
* @file WFGlobal.h
@@ -107,29 +108,20 @@ public:
static const char *get_error_string(int state, int error);
public:
/// @brief Internal use only
// Internal usage only
static CommScheduler *get_scheduler();
/// @brief Internal use only
static DNSCache *get_dns_cache();
/// @brief Internal use only
static RouteManager *get_route_manager();
/// @brief Internal use only
static SSL_CTX *get_ssl_client_ctx();
/// @brief Internal use only
static SSL_CTX *get_ssl_server_ctx();
/// @brief Internal use only
static ExecQueue *get_exec_queue(const std::string& queue_name);
/// @brief Internal use only
static Executor *get_compute_executor();
/// @brief Internal use only
static IOService *get_io_service();
/// @brief Internal use only
static ExecQueue *get_dns_queue();
/// @brief Internal use only
static Executor *get_dns_executor();
/// @brief Internal use only
static void sync_operation_begin();
static void sync_operation_end();
static WFNameService *get_name_service();
};
#endif