xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into feature/TD-1413

This commit is contained in:
Hongze Cheng 2020-11-30 10:38:07 +00:00
parent f429ad5875 71bed5816a
commit 53ea89b076
15 changed files with 125 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/client/inc/tsclient.h
View File

@ -333,7 +333,7 @@ typedef struct STscObj {
char superAuth : 1;
uint32_t connId;
uint64_t rid; // ref ID returned by taosAddRef
struct SSqlObj * pHb;
int64_t hbrid;
struct SSqlObj * sqlList;
struct SSqlStream *streamList;
void* pDnodeConn;
@ -373,7 +373,7 @@ typedef struct SSqlObj {
struct SSqlObj **pSubs;
struct SSqlObj *prev, *next;
struct SSqlObj **self;
int64_t self;
} SSqlObj;
typedef struct SSqlStream {
@ -507,7 +507,7 @@ static FORCE_INLINE void tscGetResultColumnChr(SSqlRes* pRes, SFieldInfo* pField
}
extern SCacheObj* tscMetaCache;
extern SCacheObj* tscObjCache;
extern int tscObjRef;
extern void * tscTmr;
extern void * tscQhandle;
extern int tscKeepConn[];

7
src/client/src/tscLocal.c
View File

@ -825,8 +825,11 @@ static int32_t tscProcessClientVer(SSqlObj *pSql) {
static int32_t tscProcessServStatus(SSqlObj *pSql) {
STscObj* pObj = pSql->pTscObj;
if (pObj->pHb != NULL) {
if (pObj->pHb->res.code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
SSqlObj* pHb = (SSqlObj*)taosAcquireRef(tscObjRef, pObj->hbrid);
if (pHb != NULL) {
int32_t code = pHb->res.code;
taosReleaseRef(tscObjRef, pObj->hbrid);
if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL) {
pSql->res.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
return pSql->res.code;
}

49
src/client/src/tscServer.c
View File

@ -175,10 +175,10 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
if (pRsp->streamId) tscKillStream(pObj, htonl(pRsp->streamId));
}
} else {
tscDebug("%p heartbeat failed, code:%s", pObj->pHb, tstrerror(code));
tscDebug("%" PRId64 " heartbeat failed, code:%s", pObj->hbrid, tstrerror(code));
}
if (pObj->pHb != NULL) {
if (pObj->hbrid != 0) {
int32_t waitingDuring = tsShellActivityTimer * 500;
tscDebug("%p send heartbeat in %dms", pSql, waitingDuring);
@ -193,20 +193,12 @@ void tscProcessActivityTimer(void *handle, void *tmrId) {
STscObj *pObj = taosAcquireRef(tscRefId, rid);
if (pObj == NULL) return;
SSqlObj* pHB = pObj->pHb;
void** p = taosCacheAcquireByKey(tscObjCache, &pHB, sizeof(TSDB_CACHE_PTR_TYPE));
if (p == NULL) {
tscWarn("%p HB object has been released already", pHB);
taosReleaseRef(tscRefId, pObj->rid);
return;
}
assert(*pHB->self == pHB);
SSqlObj* pHB = taosAcquireRef(tscObjRef, pObj->hbrid);
assert(pHB->self == pObj->hbrid);
pHB->retry = 0;
int32_t code = tscProcessSql(pHB);
taosCacheRelease(tscObjCache, (void**) &p, false);
taosReleaseRef(tscObjRef, pObj->hbrid);
if (code != TSDB_CODE_SUCCESS) {
tscError("%p failed to sent HB to server, reason:%s", pHB, tstrerror(code));
@ -236,7 +228,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
.msgType = pSql->cmd.msgType,
.pCont = pMsg,
.contLen = pSql->cmd.payloadLen,
.ahandle = pSql,
.ahandle = (void*)pSql->self,
.handle = NULL,
.code = 0
};
@ -247,26 +239,24 @@ int tscSendMsgToServer(SSqlObj *pSql) {
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
TSDB_CACHE_PTR_TYPE handle = (TSDB_CACHE_PTR_TYPE) rpcMsg->ahandle;
void** p = taosCacheAcquireByKey(tscObjCache, &handle, sizeof(TSDB_CACHE_PTR_TYPE));
if (p == NULL) {
SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, handle);
if (pSql == NULL) {
rpcFreeCont(rpcMsg->pCont);
return;
}
SSqlObj* pSql = *p;
assert(pSql != NULL);
assert(pSql->self == handle);
STscObj *pObj = pSql->pTscObj;
SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd;
assert(*pSql->self == pSql);
pSql->rpcRid = -1;
if (pObj->signature != pObj) {
tscDebug("%p DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command, pObj, pObj->signature);
taosCacheRelease(tscObjCache, (void**) &p, true);
taosRemoveRef(tscObjRef, pSql->self);
taosReleaseRef(tscObjRef, pSql->self);
rpcFreeCont(rpcMsg->pCont);
return;
}
@ -276,10 +266,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
tscDebug("%p sqlObj needs to be released or DB connection is closed, cmd:%d type:%d, pObj:%p signature:%p",
pSql, pCmd->command, pQueryInfo->type, pObj, pObj->signature);
void** p1 = p;
taosCacheRelease(tscObjCache, (void**) &p1, false);
taosCacheRelease(tscObjCache, (void**) &p, true);
taosRemoveRef(tscObjRef, pSql->self);
taosReleaseRef(tscObjRef, pSql->self);
rpcFreeCont(rpcMsg->pCont);
return;
}
@ -322,7 +310,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
// if there is an error occurring, proceed to the following error handling procedure.
if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosCacheRelease(tscObjCache, (void**) &p, false);
taosReleaseRef(tscObjRef, pSql->self);
rpcFreeCont(rpcMsg->pCont);
return;
}
@ -390,11 +378,10 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
(*pSql->fp)(pSql->param, pSql, rpcMsg->code);
}
void** p1 = p;
taosCacheRelease(tscObjCache, (void**) &p1, false);
taosReleaseRef(tscObjRef, pSql->self);
if (shouldFree) { // in case of table-meta/vgrouplist query, automatically free it
taosCacheRelease(tscObjCache, (void **)&p, true);
taosRemoveRef(tscObjRef, pSql->self);
tscDebug("%p sqlObj is automatically freed", pSql);
}
@ -2020,7 +2007,7 @@ int tscProcessShowRsp(SSqlObj *pSql) {
// TODO multithread problem
static void createHBObj(STscObj* pObj) {
if (pObj->pHb != NULL) {
if (pObj->hbrid != 0) {
return;
}
@ -2052,7 +2039,7 @@ static void createHBObj(STscObj* pObj) {
registerSqlObj(pSql);
tscDebug("%p HB is allocated, pObj:%p", pSql, pObj);
pObj->pHb = pSql;
pObj->hbrid = pSql->self;
}
int tscProcessConnectRsp(SSqlObj *pSql) {

18
src/client/src/tscSql.c
View File

@ -276,8 +276,8 @@ void taos_close(TAOS *taos) {
pObj->signature = NULL;
taosTmrStopA(&(pObj->pTimer));
SSqlObj* pHb = pObj->pHb;
if (pHb != NULL && atomic_val_compare_exchange_ptr(&pObj->pHb, pHb, 0) == pHb) {
SSqlObj* pHb = (SSqlObj*)taosAcquireRef(tscObjRef, pObj->hbrid);
if (pHb != NULL) {
if (pHb->rpcRid > 0) { // wait for rsp from dnode
rpcCancelRequest(pHb->rpcRid);
pHb->rpcRid = -1;
@ -285,6 +285,7 @@ void taos_close(TAOS *taos) {
tscDebug("%p HB is freed", pHb);
taos_free_result(pHb);
taosReleaseRef(tscObjRef, pHb->self);
}
int32_t ref = T_REF_DEC(pObj);
@ -606,8 +607,7 @@ void taos_free_result(TAOS_RES *res) {
bool freeNow = tscKillQueryInDnode(pSql);
if (freeNow) {
tscDebug("%p free sqlObj in cache", pSql);
SSqlObj** p = pSql->self;
taosCacheRelease(tscObjCache, (void**) &p, true);
taosReleaseRef(tscObjRef, pSql->self);
}
}
@ -700,13 +700,7 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
continue;
}
void** p = taosCacheAcquireByKey(tscObjCache, &pSub, sizeof(TSDB_CACHE_PTR_TYPE));
if (p == NULL) {
continue;
}
SSqlObj* pSubObj = (SSqlObj*) (*p);
assert(pSubObj->self == (SSqlObj**) p);
SSqlObj* pSubObj = pSub;
pSubObj->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
if (pSubObj->rpcRid > 0) {
@ -715,7 +709,7 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
}
tscQueueAsyncRes(pSubObj);
taosCacheRelease(tscObjCache, (void**) &p, false);
taosReleaseRef(tscObjRef, pSubObj->self);
}
tscDebug("%p super table query cancelled", pSql);

4
src/client/src/tscSub.c
View File

@ -179,8 +179,8 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
fail:
tscError("tscCreateSubscription failed at line %d, reason: %s", line, tstrerror(code));
if (pSql != NULL) {
if (pSql->self != NULL) {
taos_free_result(pSql);
if (pSql->self != 0) {
taosReleaseRef(tscObjRef, pSql->self);
} else {
tscFreeSqlObj(pSql);
}

3
src/client/src/tscSubquery.c
View File

@ -2198,6 +2198,9 @@ int32_t tscHandleInsertRetry(SSqlObj* pSql) {
STableDataBlocks* pTableDataBlock = taosArrayGetP(pCmd->pDataBlocks, pSupporter->index);
int32_t code = tscCopyDataBlockToPayload(pSql, pTableDataBlock);
// free the data block created from insert sql string
pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks);
if ((pRes->code = code)!= TSDB_CODE_SUCCESS) {
tscQueueAsyncRes(pSql);
return code; // here the pSql may have been released already.

12
src/client/src/tscSystem.c
View File

@ -15,7 +15,7 @@
#include "os.h"
#include "taosmsg.h"
#include "tcache.h"
#include "tref.h"
#include "trpc.h"
#include "ttimer.h"
#include "tutil.h"
@ -30,7 +30,7 @@
// global, not configurable
SCacheObj* tscMetaCache;
SCacheObj* tscObjCache;
int tscObjRef = -1;
void * tscTmr;
void * tscQhandle;
void * tscCheckDiskUsageTmr;
@ -143,7 +143,7 @@ void taos_init_imp(void) {
int64_t refreshTime = 10; // 10 seconds by default
if (tscMetaCache == NULL) {
tscMetaCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, refreshTime, false, tscFreeTableMetaHelper, "tableMeta");
tscObjCache = taosCacheInit(TSDB_CACHE_PTR_KEY, refreshTime / 2, false, tscFreeRegisteredSqlObj, "sqlObj");
tscObjRef = taosOpenRef(4096, tscFreeRegisteredSqlObj);
}
tscRefId = taosOpenRef(200, tscCloseTscObj);
@ -166,9 +166,9 @@ void taos_cleanup(void) {
taosCacheCleanup(m);
}
m = tscObjCache;
if (m != NULL && atomic_val_compare_exchange_ptr(&tscObjCache, m, 0) == m) {
taosCacheCleanup(m);
int refId = atomic_exchange_32(&tscObjRef, -1);
if (refId != -1) {
taosCloseRef(refId);
}
m = tscQhandle;

35
src/client/src/tscUtil.c
View File

@ -447,20 +447,18 @@ static void tscFreeSubobj(SSqlObj* pSql) {
void tscFreeRegisteredSqlObj(void *pSql) {
assert(pSql != NULL);
SSqlObj** p = (SSqlObj**)pSql;
STscObj* pTscObj = (*p)->pTscObj;
SSqlObj* p = *(SSqlObj**)pSql;
STscObj* pTscObj = p->pTscObj;
assert((*p)->self != 0 && (*p)->self == (p));
SSqlObj* ptr = *p;
tscFreeSqlObj(*p);
assert(p->self != 0);
tscFreeSqlObj(p);
int32_t ref = T_REF_DEC(pTscObj);
assert(ref >= 0);
tscDebug("%p free sqlObj completed, tscObj:%p ref:%d", ptr, pTscObj, ref);
tscDebug("%p free sqlObj completed, tscObj:%p ref:%d", p, pTscObj, ref);
if (ref == 0) {
tscDebug("%p all sqlObj freed, free tscObj:%p", ptr, pTscObj);
tscDebug("%p all sqlObj freed, free tscObj:%p", p, pTscObj);
taosRemoveRef(tscRefId, pTscObj->rid);
}
}
@ -840,7 +838,6 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SArray* pTableDataBlockList) {
// the length does not include the SSubmitBlk structure
pBlocks->dataLen = htonl(finalLen);
dataBuf->numOfTables += 1;
}
@ -1565,19 +1562,6 @@ void tscGetSrcColumnInfo(SSrcColumnInfo* pColInfo, SQueryInfo* pQueryInfo) {
}
}
void tscSetFreeHeatBeat(STscObj* pObj) {
if (pObj == NULL || pObj->signature != pObj || pObj->pHb == NULL) {
return;
}
SSqlObj* pHeatBeat = pObj->pHb;
assert(pHeatBeat == pHeatBeat->signature);
// to denote the heart-beat timer close connection and free all allocated resources
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pHeatBeat->cmd, 0);
pQueryInfo->type = TSDB_QUERY_TYPE_FREE_RESOURCE;
}
/*
* the following four kinds of SqlObj should not be freed
* 1. SqlObj for stream computing
@ -1596,7 +1580,7 @@ bool tscShouldBeFreed(SSqlObj* pSql) {
}
STscObj* pTscObj = pSql->pTscObj;
if (pSql->pStream != NULL || pTscObj->pHb == pSql || pSql->pSubscription != NULL) {
if (pSql->pStream != NULL || pTscObj->hbrid == pSql->self || pSql->pSubscription != NULL) {
return false;
}
@ -1888,13 +1872,10 @@ void tscResetForNextRetrieve(SSqlRes* pRes) {
}
void registerSqlObj(SSqlObj* pSql) {
int32_t DEFAULT_LIFE_TIME = 2 * 600 * 1000; // 1200 sec
int32_t ref = T_REF_INC(pSql->pTscObj);
tscDebug("%p add to tscObj:%p, ref:%d", pSql, pSql->pTscObj, ref);
TSDB_CACHE_PTR_TYPE p = (TSDB_CACHE_PTR_TYPE) pSql;
pSql->self = taosCachePut(tscObjCache, &p, sizeof(TSDB_CACHE_PTR_TYPE), &p, sizeof(TSDB_CACHE_PTR_TYPE), DEFAULT_LIFE_TIME);
pSql->self = taosAddRef(tscObjRef, pSql);
}
SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cmd) {

18
src/cq/src/cqMain.c
View File

@ -40,15 +40,14 @@
typedef struct {
int32_t vgId;
int32_t master;
int32_t num; // number of continuous streams
char user[TSDB_USER_LEN];
char pass[TSDB_PASSWORD_LEN];
char db[TSDB_DB_NAME_LEN];
FCqWrite cqWrite;
void *ahandle;
int32_t num; // number of continuous streams
struct SCqObj *pHead;
void *dbConn;
int32_t master;
void *tmrCtrl;
pthread_mutex_t mutex;
} SCqContext;
@ -90,7 +89,6 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
tstrncpy(pContext->db, db, sizeof(pContext->db));
pContext->vgId = pCfg->vgId;
pContext->cqWrite = pCfg->cqWrite;
pContext->ahandle = ahandle;
tscEmbedded = 1;
pthread_mutex_init(&pContext->mutex, NULL);
@ -241,8 +239,12 @@ static void doCreateStream(void *param, TAOS_RES *result, int32_t code) {
SCqObj* pObj = (SCqObj*)param;
SCqContext* pContext = pObj->pContext;
SSqlObj* pSql = (SSqlObj*)result;
pContext->dbConn = pSql->pTscObj;
if (atomic_val_compare_exchange_ptr(&(pContext->dbConn), NULL, pSql->pTscObj) != NULL) {
taos_close(pSql->pTscObj);
}
pthread_mutex_lock(&pContext->mutex);
cqCreateStream(pContext, pObj);
pthread_mutex_unlock(&pContext->mutex);
}
static void cqProcessCreateTimer(void *param, void *tmrId) {
@ -253,7 +255,9 @@ static void cqProcessCreateTimer(void *param, void *tmrId) {
cDebug("vgId:%d, try connect to TDengine", pContext->vgId);
taos_connect_a(NULL, pContext->user, pContext->pass, pContext->db, 0, doCreateStream, param, NULL);
} else {
pthread_mutex_lock(&pContext->mutex);
cqCreateStream(pContext, pObj);
pthread_mutex_unlock(&pContext->mutex);
}
}
@ -267,6 +271,7 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
}
pObj->tmrId = 0;
if (pObj->pStream == NULL) {
pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, 0, pObj, NULL);
if (pObj->pStream) {
pContext->num++;
@ -274,6 +279,7 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
} else {
cError("vgId:%d, id:%d CQ:%s, failed to open", pContext->vgId, pObj->tid, pObj->sqlStr);
}
}
}
static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
@ -334,7 +340,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
pHead->version = 0;
// write into vnode write queue
pContext->cqWrite(pContext->ahandle, pHead, TAOS_QTYPE_CQ, NULL);
pContext->cqWrite(pContext->vgId, pHead, TAOS_QTYPE_CQ, NULL);
free(buffer);
}

2
src/cq/test/cqtest.c
View File

@ -24,7 +24,7 @@
int64_t ver = 0;
void *pCq = NULL;
int writeToQueue(void *pVnode, void *data, int type, void *pMsg) {
int writeToQueue(int32_t vgId, void *data, int type, void *pMsg) {
return 0;
}

2
src/inc/tcq.h
View File

@ -21,7 +21,7 @@ extern "C" {
#include "tdataformat.h"
typedef int32_t (*FCqWrite)(void *ahandle, void *pHead, int32_t qtype, void *pMsg);
typedef int32_t (*FCqWrite)(int32_t vgId, void *pHead, int32_t qtype, void *pMsg);
typedef struct {
int32_t vgId;

8
src/tsdb/src/tsdbRWHelper.c
View File

@ -925,6 +925,8 @@ static int tsdbInsertSuperBlock(SRWHelper *pHelper, SCompBlock *pCompBlock, int
ASSERT(pHelper->pCompInfo->blocks[0].keyLast < pHelper->pCompInfo->blocks[1].keyFirst);
}
ASSERT((blkIdx == pIdx->numOfBlocks -1) || (!pCompBlock->last));
tsdbDebug("vgId:%d tid:%d a super block is inserted at index %d", REPO_ID(pHelper->pRepo), pHelper->tableInfo.tid,
blkIdx);
@ -1050,6 +1052,8 @@ static int tsdbUpdateSuperBlock(SRWHelper *pHelper, SCompBlock *pCompBlock, int
pIdx->maxKey = blockAtIdx(pHelper, pIdx->numOfBlocks - 1)->keyLast;
pIdx->hasLast = (uint32_t)blockAtIdx(pHelper, pIdx->numOfBlocks - 1)->last;
ASSERT((blkIdx == pIdx->numOfBlocks-1) || (!pCompBlock->last));
tsdbDebug("vgId:%d tid:%d a super block is updated at index %d", REPO_ID(pHelper->pRepo), pHelper->tableInfo.tid,
blkIdx);
@ -1630,11 +1634,7 @@ static int tsdbProcessMergeCommit(SRWHelper *pHelper, SCommitIter *pCommitIter,
pCfg->update);
if (pDataCols->numOfRows == 0) break;
if (tblkIdx == pIdx->numOfBlocks - 1) {
if (tsdbWriteBlockToProperFile(pHelper, pDataCols, &compBlock) < 0) return -1;
} else {
if (tsdbWriteBlockToFile(pHelper, helperDataF(pHelper), pDataCols, &compBlock, false, true) < 0) return -1;
}
if (round == 0) {
if (oBlock.last && pHelper->hasOldLastBlock) pHelper->hasOldLastBlock = false;

22
src/util/src/tfile.c
View File

@ -93,17 +93,23 @@ int64_t tfRead(int64_t tfd, void *buf, int64_t count) {
return ret;
}
int64_t tfFsync(int64_t tfd) {
int32_t tfFsync(int64_t tfd) {
void *p = taosAcquireRef(tsFileRsetId, tfd);
if (p == NULL) return -1;
int32_t fd = (int32_t)(uintptr_t)p;
return fsync(fd);
int32_t code = fsync(fd);
taosReleaseRef(tsFileRsetId, tfd);
return code;
}
bool tfValid(int64_t tfd) {
void *p = taosAcquireRef(tsFileRsetId, tfd);
return p != NULL;
if (p == NULL) return false;
taosReleaseRef(tsFileRsetId, tfd);
return true;
}
int64_t tfLseek(int64_t tfd, int64_t offset, int32_t whence) {
@ -111,7 +117,10 @@ int64_t tfLseek(int64_t tfd, int64_t offset, int32_t whence) {
if (p == NULL) return -1;
int32_t fd = (int32_t)(uintptr_t)p;
return taosLSeek(fd, offset, whence);
int64_t ret = taosLSeek(fd, offset, whence);
taosReleaseRef(tsFileRsetId, tfd);
return ret;
}
int32_t tfFtruncate(int64_t tfd, int64_t length) {
@ -119,5 +128,8 @@ int32_t tfFtruncate(int64_t tfd, int64_t length) {
if (p == NULL) return -1;
int32_t fd = (int32_t)(uintptr_t)p;
return taosFtruncate(fd, length);
int32_t code = taosFtruncate(fd, length);
taosReleaseRef(tsFileRsetId, tfd);
return code;
}

31
src/util/src/ttimer.c
View File

@ -560,6 +560,37 @@ void taosTmrCleanUp(void* handle) {
tmrDebug("%s timer controller is cleaned up.", ctrl->label);
ctrl->label[0] = 0;
// cancel all timers of this controller
for (size_t i = 0; i < timerMap.size; i++) {
timer_list_t* list = timerMap.slots + i;
lockTimerList(list);
tmr_obj_t* t = list->timers;
tmr_obj_t* prev = NULL;
while (t != NULL) {
tmr_obj_t* next = t->mnext;
if (t->ctrl != ctrl) {
prev = t;
t = next;
continue;
}
uint8_t state = atomic_val_compare_exchange_8(&t->state, TIMER_STATE_WAITING, TIMER_STATE_CANCELED);
if (state == TIMER_STATE_WAITING) {
removeFromWheel(t);
}
timerDecRef(t);
if (prev == NULL) {
list->timers = next;
} else {
prev->mnext = next;
}
t = next;
}
unlockTimerList(list);
}
pthread_mutex_lock(&tmrCtrlMutex);
ctrl->next = unusedTmrCtrl;
numOfTmrCtrl--;

2
src/vnode/src/vnodeMain.c
View File

@ -264,7 +264,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
strcpy(cqCfg.pass, tsInternalPass);
strcpy(cqCfg.db, pVnode->db);
cqCfg.vgId = vnode;
cqCfg.cqWrite = vnodeWriteToWQueue;
cqCfg.cqWrite = vnodeWriteToCache;
pVnode->cq = cqOpen(pVnode, &cqCfg);
if (pVnode->cq == NULL) {
vnodeCleanUp(pVnode);