mirrors/TarsCloud_TarsCpp
1
0
Fork 0
mirror of https://gitee.com/TarsCloud/TarsCpp.git synced 2025-01-05 17:42:24 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix hash call bug

Browse Source
This commit is contained in:
ruanshudong 2021-03-23 22:19:27 +08:00
parent f8b8fa17b9
commit 5edc05c8a4
2 changed files with 47 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
servant/libservant/AdapterProxy.cpp
View File

@ -485,14 +485,14 @@ int AdapterProxy::getConTimeout()
return _objectProxy->getConTimeout();
}
bool AdapterProxy::checkActive(bool bForceConnect)
bool AdapterProxy::checkActive(bool connecting)
{
time_t now = TNOW;
TLOGTARS("[AdapterProxy::checkActive,"
<< _objectProxy->name() << "," << _trans->getEndpointInfo().desc() << ","
<< (_activeStatus ? "active" : "inactive")
<< (bForceConnect ? ",forceConnect" : "")
<< (connecting ? ", connecting" : "")
<< ",freqtimeout:" << _frequenceFailInvoke
<< ",timeout:" << _timeoutInvoke
<< ",connExcCnt:" << _connExcCnt
@ -500,35 +500,35 @@ bool AdapterProxy::checkActive(bool bForceConnect)
_trans->checkTimeout();
//强制重试
if (bForceConnect)
{
//正在连接也算有效的连接, 避免第一次都没有连接上时, 没有办法发送数据
if (_trans->isConnecting() || _trans->hasConnected())
{
return true;
}
// //强制重试
// if (bForceConnect)
// {
// //正在连接也算有效的连接, 避免第一次都没有连接上时, 没有办法发送数据
// if (_trans->isConnecting() || _trans->hasConnected())
// {
// return true;
// }
_nextRetryTime = now + _objectProxy->checkTimeoutInfo().tryTimeInterval;
// _nextRetryTime = now + _objectProxy->checkTimeoutInfo().tryTimeInterval;
//连接没有建立或者连接无效, 重新建立连接
if (!_trans->isValid())
{
try
{
_trans->reconnect();
}
catch (exception & ex)
{
_activeStatus = false;
_trans->close();
// //连接没有建立或者连接无效, 重新建立连接
// if (!_trans->isValid())
// {
// try
// {
// _trans->reconnect();
// }
// catch (exception & ex)
// {
// _activeStatus = false;
// _trans->close();
TLOGERROR("[AdapterProxy::checkActive connect obj:" << _objectProxy->name() << ",desc:" << _trans->getEndpointInfo().desc() << " ex:" << ex.what() << endl);
}
}
// TLOGERROR("[AdapterProxy::checkActive connect obj:" << _objectProxy->name() << ",desc:" << _trans->getEndpointInfo().desc() << " ex:" << ex.what() << endl);
// }
// }
return (_trans->hasConnected() || _trans->isConnecting());
}
// return (_trans->hasConnected() || _trans->isConnecting());
// }
//失效且没有到下次重试时间, 直接返回不可用
if ((!_activeStatus) && (now < _nextRetryTime))
@ -559,8 +559,16 @@ bool AdapterProxy::checkActive(bool bForceConnect)
}
}
if(connecting && _activeStatus) {
//hash模式, 且是第一次连接(_activeStatus=true, 即没有失败过), 返回已经连接或者正在连接的, 这样保证第一次hash不会错且连接挂过以后, 不会马上就使用, 直到连接成功才使用!
return (_trans->hasConnected() || _trans->isConnecting());
}
else {
return _trans->hasConnected();
}
//已经建立连接了才返回
return _trans->hasConnected();
// return _trans->hasConnected();
}
void AdapterProxy::setConTimeout(bool bConTimeout)

22
servant/libservant/EndpointManager.cpp
View File

@ -404,7 +404,7 @@ void QueryEpBase::refreshReg(GetEndpointType type, const string & sName)
case E_STATION:
{
iRet = _queryFPrx->findObjectByIdInSameStation(_objName,sName,activeEp,inactiveEp, ServerConfig::Context);
break;
break;
}
case E_SET:
{
@ -955,7 +955,7 @@ AdapterProxy* EndpointManager::getHashProxyForWeight(int64_t hashCode, bool bSta
}
//被hash到的节点在主控是active的才走在流程
if (_vRegProxys[iIndex]->isActiveInReg() && _vRegProxys[iIndex]->checkActive())
if (_vRegProxys[iIndex]->isActiveInReg() && _vRegProxys[iIndex]->checkActive(true))
{
return _vRegProxys[iIndex];
}
@ -981,7 +981,7 @@ AdapterProxy* EndpointManager::getHashProxyForWeight(int64_t hashCode, bool bSta
hash = hash % thisHash.size();
}
if (thisHash[hash]->checkActive(false))
if (thisHash[hash]->checkActive(true))
{
return thisHash[hash];
}
@ -1057,7 +1057,7 @@ AdapterProxy* EndpointManager::getConHashProxyForWeight(int64_t hashCode, bool b
}
//被hash到的节点在主控是active的才走在流程
if (_vRegProxys[iIndex]->isActiveInReg() && _vRegProxys[iIndex]->checkActive())
if (_vRegProxys[iIndex]->isActiveInReg() && _vRegProxys[iIndex]->checkActive(true))
{
return _vRegProxys[iIndex];
}
@ -1084,7 +1084,7 @@ AdapterProxy* EndpointManager::getConHashProxyForWeight(int64_t hashCode, bool b
hash = hash % thisHash.size();
}
if (thisHash[hash]->checkActive())
if (thisHash[hash]->checkActive(false))
{
return thisHash[hash];
}
@ -1352,7 +1352,7 @@ AdapterProxy* EndpointManager::getHashProxyForNormal(int64_t hashCode)
}
//被hash到的节点在主控是active的才走在流程
if (_vRegProxys[hash]->isActiveInReg() && _vRegProxys[hash]->checkActive())
if (_vRegProxys[hash]->isActiveInReg() && _vRegProxys[hash]->checkActive(true))
{
return _vRegProxys[hash];
}
@ -1378,7 +1378,7 @@ AdapterProxy* EndpointManager::getHashProxyForNormal(int64_t hashCode)
hash = hash % thisHash.size();
}
if (thisHash[hash]->checkActive())
if (thisHash[hash]->checkActive(true))
{
return thisHash[hash];
}
@ -1450,7 +1450,7 @@ AdapterProxy* EndpointManager::getConHashProxyForNormal(int64_t hashCode)
}
//被hash到的节点在主控是active的才走在流程
if (_vRegProxys[iIndex]->isActiveInReg() && _vRegProxys[iIndex]->checkActive())
if (_vRegProxys[iIndex]->isActiveInReg() && _vRegProxys[iIndex]->checkActive(true))
{
return _vRegProxys[iIndex];
}
@ -1477,7 +1477,7 @@ AdapterProxy* EndpointManager::getConHashProxyForNormal(int64_t hashCode)
hash = hash % thisHash.size();
}
if (thisHash[hash]->checkActive())
if (thisHash[hash]->checkActive(true))
{
return thisHash[hash];
}
@ -1568,7 +1568,7 @@ AdapterProxy* EndpointManager::getWeightedForNormal(bool bStaticWeighted)
iProxyIndex = _staticRouterCache[_lastSWeightPosition];
if(_activeWeightProxy[iProxyIndex]->checkActive())
if(_activeWeightProxy[iProxyIndex]->checkActive(false))
{
return _activeWeightProxy[iProxyIndex];
}
@ -1623,7 +1623,7 @@ AdapterProxy* EndpointManager::getWeightedForNormal(bool bStaticWeighted)
if(_lastRoundPosition >= _activeProxys.size())
_lastRoundPosition = 0;
if(_activeProxys[_lastRoundPosition]->checkActive())
if(_activeProxys[_lastRoundPosition]->checkActive(false))
{
return _activeProxys[_lastRoundPosition];
}