提交 d3999c7b 编写于 作者: D dapan1121

fix bug

上级 7057c906
...@@ -73,6 +73,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row); ...@@ -73,6 +73,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row);
static void cqCreateStream(SCqContext *pContext, SCqObj *pObj); static void cqCreateStream(SCqContext *pContext, SCqObj *pObj);
int32_t cqObjRef = -1; int32_t cqObjRef = -1;
int32_t cqVnodeNum = 0;
void cqRmFromList(SCqObj *pObj) { void cqRmFromList(SCqObj *pObj) {
//LOCK in caller //LOCK in caller
...@@ -166,6 +167,8 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) { ...@@ -166,6 +167,8 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
return NULL; return NULL;
} }
atomic_add_fetch_32(&cqVnodeNum, 1);
cqCreateRef(); cqCreateRef();
pContext->tmrCtrl = taosTmrInit(0, 0, 0, "CQ"); pContext->tmrCtrl = taosTmrInit(0, 0, 0, "CQ");
...@@ -240,6 +243,13 @@ void cqClose(void *handle) { ...@@ -240,6 +243,13 @@ void cqClose(void *handle) {
if (hasCq == 0) { if (hasCq == 0) {
freeSCqContext(pContext); freeSCqContext(pContext);
} }
int32_t remainn = atomic_sub_fetch_32(&cqVnodeNum, 1);
if (remainn <= 0) {
int32_t ref = cqObjRef;
cqObjRef = -1;
taosCloseRef(ref);
}
} }
void cqStart(void *handle) { void cqStart(void *handle) {
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册