提交 236c4654 编写于 作者: Y yihaoDeng

fix case failure

上级 a1b81053
...@@ -299,7 +299,6 @@ typedef struct { ...@@ -299,7 +299,6 @@ typedef struct {
typedef struct { typedef struct {
char key[512]; char key[512];
SRpcCorEpSet *tscCorMgmtEpSet;
void *pDnodeConn; void *pDnodeConn;
} SRpcObj; } SRpcObj;
...@@ -319,6 +318,7 @@ typedef struct STscObj { ...@@ -319,6 +318,7 @@ typedef struct STscObj {
struct SSqlObj * sqlList; struct SSqlObj * sqlList;
struct SSqlStream *streamList; struct SSqlStream *streamList;
SRpcObj *pRpcObj; SRpcObj *pRpcObj;
SRpcCorEpSet *tscCorMgmtEpSet;
pthread_mutex_t mutex; pthread_mutex_t mutex;
int32_t numOfObj; // number of sqlObj from this tscObj int32_t numOfObj; // number of sqlObj from this tscObj
} STscObj; } STscObj;
...@@ -396,7 +396,7 @@ typedef struct SSqlStream { ...@@ -396,7 +396,7 @@ typedef struct SSqlStream {
void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable); void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable);
int tscAcquireRpc(const char *key, const char *user, const char *secret, SRpcCorEpSet *corMgmtEpSet, void **pRpcObj); int tscAcquireRpc(const char *key, const char *user, const char *secret,void **pRpcObj);
void tscReleaseRpc(void *param); void tscReleaseRpc(void *param);
void tscInitMsgsFp(); void tscInitMsgsFp();
......
...@@ -69,7 +69,7 @@ static void tscSetDnodeEpSet(SRpcEpSet* pEpSet, SVgroupInfo* pVgroupInfo) { ...@@ -69,7 +69,7 @@ static void tscSetDnodeEpSet(SRpcEpSet* pEpSet, SVgroupInfo* pVgroupInfo) {
} }
static void tscDumpMgmtEpSet(SSqlObj *pSql) { static void tscDumpMgmtEpSet(SSqlObj *pSql) {
SRpcCorEpSet *pCorEpSet = pSql->pTscObj->pRpcObj->tscCorMgmtEpSet; SRpcCorEpSet *pCorEpSet = pSql->pTscObj->tscCorMgmtEpSet;
taosCorBeginRead(&pCorEpSet->version); taosCorBeginRead(&pCorEpSet->version);
pSql->epSet = pCorEpSet->epSet; pSql->epSet = pCorEpSet->epSet;
taosCorEndRead(&pCorEpSet->version); taosCorEndRead(&pCorEpSet->version);
...@@ -94,7 +94,7 @@ bool tscEpSetIsEqual(SRpcEpSet *s1, SRpcEpSet *s2) { ...@@ -94,7 +94,7 @@ bool tscEpSetIsEqual(SRpcEpSet *s1, SRpcEpSet *s2) {
void tscUpdateMgmtEpSet(SSqlObj *pSql, SRpcEpSet *pEpSet) { void tscUpdateMgmtEpSet(SSqlObj *pSql, SRpcEpSet *pEpSet) {
// no need to update if equal // no need to update if equal
SRpcCorEpSet *pCorEpSet = pSql->pTscObj->pRpcObj->tscCorMgmtEpSet; SRpcCorEpSet *pCorEpSet = pSql->pTscObj->tscCorMgmtEpSet;
taosCorBeginWrite(&pCorEpSet->version); taosCorBeginWrite(&pCorEpSet->version);
pCorEpSet->epSet = *pEpSet; pCorEpSet->epSet = *pEpSet;
taosCorEndWrite(&pCorEpSet->version); taosCorEndWrite(&pCorEpSet->version);
...@@ -158,7 +158,7 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) { ...@@ -158,7 +158,7 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
if (epSet->numOfEps > 0) { if (epSet->numOfEps > 0) {
tscEpSetHtons(epSet); tscEpSetHtons(epSet);
//SRpcCorEpSet *pCorEpSet = pSql->pTscObj->pRpcObj->tscCorMgmtEpSet; //SRpcCorEpSet *pCorEpSet = pSql->pTscObj->tscCorMgmtEpSet;
//if (!tscEpSetIsEqual(&pCorEpSet->epSet, epSet)) { //if (!tscEpSetIsEqual(&pCorEpSet->epSet, epSet)) {
// tscTrace("%p updating epset: numOfEps: %d, inUse: %d", pSql, epSet->numOfEps, epSet->inUse); // tscTrace("%p updating epset: numOfEps: %d, inUse: %d", pSql, epSet->numOfEps, epSet->inUse);
// for (int8_t i = 0; i < epSet->numOfEps; i++) { // for (int8_t i = 0; i < epSet->numOfEps; i++) {
......
...@@ -94,7 +94,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa ...@@ -94,7 +94,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
snprintf(rpcKey, sizeof(rpcKey), "%s:%s:%s:%d", user, pass, ip, port); snprintf(rpcKey, sizeof(rpcKey), "%s:%s:%s:%d", user, pass, ip, port);
void *pRpcObj = NULL; void *pRpcObj = NULL;
if (tscAcquireRpc(rpcKey, user, secretEncrypt,&corMgmtEpSet, &pRpcObj) != 0) { if (tscAcquireRpc(rpcKey, user, secretEncrypt, &pRpcObj) != 0) {
terrno = TSDB_CODE_RPC_NETWORK_UNAVAIL; terrno = TSDB_CODE_RPC_NETWORK_UNAVAIL;
return NULL; return NULL;
} }
...@@ -105,6 +105,16 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa ...@@ -105,6 +105,16 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
tscReleaseRpc(pRpcObj); tscReleaseRpc(pRpcObj);
return NULL; return NULL;
} }
pObj->tscCorMgmtEpSet = malloc(sizeof(SRpcCorEpSet));
if (pObj->tscCorMgmtEpSet == NULL) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
tscReleaseRpc(pRpcObj);
free(pObj);
return NULL;
}
memcpy(pObj->tscCorMgmtEpSet, &corMgmtEpSet, sizeof(corMgmtEpSet));
pObj->signature = pObj; pObj->signature = pObj;
pObj->pRpcObj = (SRpcObj *)pRpcObj; pObj->pRpcObj = (SRpcObj *)pRpcObj;
tstrncpy(pObj->user, user, sizeof(pObj->user)); tstrncpy(pObj->user, user, sizeof(pObj->user));
......
...@@ -58,7 +58,6 @@ void tscFreeRpcObj(void *param) { ...@@ -58,7 +58,6 @@ void tscFreeRpcObj(void *param) {
SRpcObj *pRpcObj = (SRpcObj *)(param); SRpcObj *pRpcObj = (SRpcObj *)(param);
tscDebug("free rpcObj:%p and free pDnodeConn: %p", pRpcObj, pRpcObj->pDnodeConn); tscDebug("free rpcObj:%p and free pDnodeConn: %p", pRpcObj, pRpcObj->pDnodeConn);
rpcClose(pRpcObj->pDnodeConn); rpcClose(pRpcObj->pDnodeConn);
tfree(pRpcObj->tscCorMgmtEpSet);
} }
void tscReleaseRpc(void *param) { void tscReleaseRpc(void *param) {
...@@ -70,7 +69,7 @@ void tscReleaseRpc(void *param) { ...@@ -70,7 +69,7 @@ void tscReleaseRpc(void *param) {
pthread_mutex_unlock(&rpcObjMutex); pthread_mutex_unlock(&rpcObjMutex);
} }
int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncrypt, SRpcCorEpSet *corMgmtEpSet, void **ppRpcObj) { int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncrypt, void **ppRpcObj) {
pthread_mutex_lock(&rpcObjMutex); pthread_mutex_lock(&rpcObjMutex);
SRpcObj *pRpcObj = (SRpcObj *)taosCacheAcquireByKey(tscRpcCache, key, strlen(key)); SRpcObj *pRpcObj = (SRpcObj *)taosCacheAcquireByKey(tscRpcCache, key, strlen(key));
...@@ -109,13 +108,6 @@ int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncry ...@@ -109,13 +108,6 @@ int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncry
pthread_mutex_unlock(&rpcObjMutex); pthread_mutex_unlock(&rpcObjMutex);
return -1; return -1;
} }
pRpcObj->tscCorMgmtEpSet = malloc(sizeof(SRpcCorEpSet));
if (pRpcObj->tscCorMgmtEpSet == NULL) {
rpcClose(rpcObj.pDnodeConn);
pthread_mutex_unlock(&rpcObjMutex);
return -1;
}
memcpy(pRpcObj->tscCorMgmtEpSet, corMgmtEpSet, sizeof(*corMgmtEpSet));
*ppRpcObj = pRpcObj; *ppRpcObj = pRpcObj;
pthread_mutex_unlock(&rpcObjMutex); pthread_mutex_unlock(&rpcObjMutex);
......
...@@ -897,6 +897,7 @@ void tscCloseTscObj(void *param) { ...@@ -897,6 +897,7 @@ void tscCloseTscObj(void *param) {
pObj->signature = NULL; pObj->signature = NULL;
taosTmrStopA(&(pObj->pTimer)); taosTmrStopA(&(pObj->pTimer));
tfree(pObj->tscCorMgmtEpSet);
tscReleaseRpc(pObj->pRpcObj); tscReleaseRpc(pObj->pRpcObj);
pthread_mutex_destroy(&pObj->mutex); pthread_mutex_destroy(&pObj->mutex);
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册