Skip to content

TDengine
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
TDengine
TDengine
提交 9db65e7a 编写于 作者: L Liu Jicong
浏览文件
操作

merge from 3.0

上级 c6230453 cd1e59a9
展开全部 隐藏空白更改
内联 并排
Showing with 1002 addition and 1230 deletion
2.0/src/query/src/queryMain.c
浏览文件 @ 9db65e7a
......@@ -461,7 +461,7 @@ int32_t qKillQuery(qinfo_t qinfo) {
}
qDebug("QInfo:0x%"PRIx64" query killed", pQInfo->qId);
setQueryKilled(pQInfo);
setTaskKilled(pQInfo);
// Wait for the query executing thread being stopped/
// Once the query is stopped, the owner of qHandle will be cleared immediately.
......@@ -634,7 +634,7 @@ int32_t qKillQueryByQId(void* pMgmt, int64_t qId, int32_t waitMs, int32_t waitCo
return TSDB_CODE_QRY_INVALID_QHANDLE;
}
qWarn("QId:0x%"PRIx64" be killed(no memory commit).", pQInfo->qId);
setQueryKilled(pQInfo);
setTaskKilled(pQInfo);
// wait query stop
int32_t loop = 0;
......
include/libs/executor/executor.h
浏览文件 @ 9db65e7a
......@@ -24,6 +24,7 @@ extern "C" {
typedef void* qTaskInfo_t;
typedef void* DataSinkHandle;
struct SRpcMsg;
struct SSubplan;
/**
......@@ -45,7 +46,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, void* input);
* @param qId
* @return
*/
int32_t qCreateExecTask(void* readHandle, int32_t vgId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle);
int32_t qCreateExecTask(void* readHandle, int32_t vgId, uint64_t taskId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle);
/**
* The main task execution function, including query on both table and multiple tables,
......@@ -208,6 +209,8 @@ void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle);
*/
void** qDeregisterQInfo(void* pMgmt, void* pQInfo);
void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet);
#ifdef __cplusplus
}
#endif
......
include/libs/qworker/qworker.h
浏览文件 @ 9db65e7a
......@@ -49,9 +49,10 @@ typedef struct {
} SQWorkerStat;
typedef int32_t (*putReqToQueryQFp)(void *, struct SRpcMsg *);
typedef int32_t (*sendReqToDnodeFp)(void *, struct SEpSet *, struct SRpcMsg *);
int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, putReqToQueryQFp fp);
int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj,
putReqToQueryQFp fp1, sendReqToDnodeFp fp2);
int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
......@@ -65,6 +66,8 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessFetchRsp(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
......
include/os/os.h
浏览文件 @ 9db65e7a
......@@ -51,6 +51,7 @@ extern "C" {
#include <libgen.h>
#include <sys/mman.h>
#include <sys/prctl.h>
#include "osAtomic.h"
#include "osDef.h"
......
include/os/osDef.h
浏览文件 @ 9db65e7a
......@@ -181,7 +181,8 @@ extern "C" {
#endif
#else
// Windows
#define setThreadName(name)
// #define setThreadName(name)
#define setThreadName(name) do { prctl(PR_SET_NAME, (name)); } while (0)
#endif
#if defined(_WIN32)
......
include/util/tqueue.h
浏览文件 @ 9db65e7a
......@@ -15,6 +15,7 @@
#ifndef _TD_UTIL_QUEUE_H
#define _TD_UTIL_QUEUE_H
#include "os.h"
#ifdef __cplusplus
extern "C" {
......@@ -40,13 +41,13 @@ shall be used to set up the protection.
typedef struct STaosQueue STaosQueue;
typedef struct STaosQset STaosQset;
typedef struct STaosQall STaosQall;
typedef void (*FProcessItem)(void *ahandle, void *pItem);
typedef void (*FProcessItems)(void *ahandle, STaosQall *qall, int32_t numOfItems);
typedef void (*FItem)(void *ahandle, void *pItem);
typedef void (*FItems)(void *ahandle, STaosQall *qall, int32_t numOfItems);
STaosQueue *taosOpenQueue();
void taosCloseQueue(STaosQueue *queue);
void taosSetQueueFp(STaosQueue *queue, FProcessItem itemFp, FProcessItems itemsFp);
void *taosAllocateQitem(int32_t size);
void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp);
void * taosAllocateQitem(int32_t size);
void taosFreeQitem(void *pItem);
int32_t taosWriteQitem(STaosQueue *queue, void *pItem);
int32_t taosReadQitem(STaosQueue *queue, void **ppItem);
......@@ -66,8 +67,11 @@ int32_t taosAddIntoQset(STaosQset *qset, STaosQueue *queue, void *ahandle);
void taosRemoveFromQset(STaosQset *qset, STaosQueue *queue);
int32_t taosGetQueueNumber(STaosQset *qset);
int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FProcessItem *itemFp);
int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FProcessItems *itemsFp);
int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FItem *itemFp);
int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FItems *itemsFp);
int32_t taosReadQitemFromQsetByThread(STaosQset *qset, void **ppItem, void **ahandle, FItem *itemFp, int32_t threadId);
void taosResetQsetThread(STaosQset *qset, void *pItem);
int32_t taosGetQueueItemsNumber(STaosQueue *queue);
int32_t taosGetQsetItemsNumber(STaosQset *qset);
......
include/util/tworker.h
浏览文件 @ 9db65e7a
......@@ -15,57 +15,61 @@
#ifndef _TD_UTIL_WORKER_H
#define _TD_UTIL_WORKER_H
#include "tqueue.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct SWorkerPool SWorkerPool;
typedef struct SMWorkerPool SMWorkerPool;
typedef struct SQWorkerPool SQWorkerPool;
typedef struct SWWorkerPool SWWorkerPool;
typedef struct SWorker {
int32_t id; // worker ID
pthread_t thread; // thread
SWorkerPool *pool;
} SWorker;
typedef struct SQWorker {
int32_t id; // worker ID
pthread_t thread; // thread
SQWorkerPool *pool;
} SQWorker, SFWorker;
typedef struct SWorkerPool {
typedef struct SQWorkerPool {
int32_t max; // max number of workers
int32_t min; // min number of workers
int32_t num; // current number of workers
STaosQset *qset;
const char *name;
SWorker *workers;
STaosQset * qset;
const char * name;
SQWorker * workers;
pthread_mutex_t mutex;
} SWorkerPool;
} SQWorkerPool, SFWorkerPool;
typedef struct SMWorker {
typedef struct SWWorker {
int32_t id; // worker id
pthread_t thread; // thread
STaosQall *qall;
STaosQset *qset; // queue set
SMWorkerPool *pool;
} SMWorker;
STaosQall * qall;
STaosQset * qset; // queue set
SWWorkerPool *pool;
} SWWorker;
typedef struct SMWorkerPool {
typedef struct SWWorkerPool {
int32_t max; // max number of workers
int32_t nextId; // from 0 to max-1, cyclic
const char *name;
SMWorker *workers;
const char * name;
SWWorker * workers;
pthread_mutex_t mutex;
} SMWorkerPool;
} SWWorkerPool;
int32_t tQWorkerInit(SQWorkerPool *pool);
void tQWorkerCleanup(SQWorkerPool *pool);
STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp);
void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue);
int32_t tWorkerInit(SWorkerPool *pool);
void tWorkerCleanup(SWorkerPool *pool);
STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp);
void tWorkerFreeQueue(SWorkerPool *pool, STaosQueue *queue);
int32_t tFWorkerInit(SFWorkerPool *pool);
void tFWorkerCleanup(SFWorkerPool *pool);
STaosQueue *tFWorkerAllocQueue(SFWorkerPool *pool, void *ahandle, FItem fp);
void tFWorkerFreeQueue(SFWorkerPool *pool, STaosQueue *queue);
int32_t tMWorkerInit(SMWorkerPool *pool);
void tMWorkerCleanup(SMWorkerPool *pool);
STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems fp);
void tMWorkerFreeQueue(SMWorkerPool *pool, STaosQueue *queue);
int32_t tWWorkerInit(SWWorkerPool *pool);
void tWWorkerCleanup(SWWorkerPool *pool);
STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp);
void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue);
#ifdef __cplusplus
}
......
source/client/test/clientTests.cpp
浏览文件 @ 9db65e7a
......@@ -570,6 +570,7 @@ TEST(testCase, create_topic_Test) {
//taos_close(pConn);
//}
#if 0
TEST(testCase, tmq_subscribe_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
......@@ -594,6 +595,7 @@ TEST(testCase, tmq_subscribe_Test) {
//if (msg == NULL) break;
}
}
#endif
TEST(testCase, tmq_consume_Test) {
}
......
source/dnode/mgmt/impl/inc/dndEnv.h
浏览文件 @ 9db65e7a
......@@ -31,8 +31,8 @@ typedef struct {
SDnode *pDnode;
STaosQueue *queue;
union {
SWorkerPool pool;
SMWorkerPool mpool;
SQWorkerPool pool;
SWWorkerPool mpool;
};
} SDnodeWorker;
......@@ -109,10 +109,10 @@ typedef struct {
int32_t openVnodes;
int32_t totalVnodes;
SRWLatch latch;
SWorkerPool queryPool;
SWorkerPool fetchPool;
SMWorkerPool syncPool;
SMWorkerPool writePool;
SQWorkerPool queryPool;
SFWorkerPool fetchPool;
SWWorkerPool syncPool;
SWWorkerPool writePool;
} SVnodesMgmt;
typedef struct {
......
source/dnode/mgmt/impl/src/dndEnv.c
浏览文件 @ 9db65e7a
......@@ -289,6 +289,7 @@ int32_t dndInit(const SDnodeEnvCfg *pCfg) {
.charset = pCfg->charset,
.nthreads = pCfg->numOfCommitThreads,
.putReqToVQueryQFp = dndPutReqToVQueryQ,
.sendReqToDnodeFp = dndSendReqToDnode
};
if (vnodeInit(&vnodeOpt) != 0) {
......
source/dnode/mgmt/impl/src/dndMgmt.c
浏览文件 @ 9db65e7a
......@@ -487,6 +487,8 @@ static void *dnodeThreadRoutine(void *param) {
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
int32_t ms = pDnode->cfg.statusInterval * 1000;
setThreadName("dnode-hb");
while (true) {
pthread_testcancel();
taosMsleep(ms);
......@@ -527,7 +529,7 @@ int32_t dndInitMgmt(SDnode *pDnode) {
}
if (pMgmt->dropped) {
dError("dnode will not start for its already dropped");
dError("dnode not start since its already dropped");
return -1;
}
......
source/dnode/mgmt/impl/src/dndTransport.c
浏览文件 @ 9db65e7a
......@@ -25,9 +25,9 @@
#include "dndMnode.h"
#include "dndVnodes.h"
#define INTERNAL_USER "_internal"
#define INTERNAL_USER "_dnd"
#define INTERNAL_CKEY "_key"
#define INTERNAL_SECRET "_secret"
#define INTERNAL_SECRET "_pwd"
static void dndInitMsgFp(STransMgmt *pMgmt) {
// Requests handled by DNODE
......@@ -120,8 +120,9 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
// Requests handled by VNODE
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SUBMIT)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY)] = dndProcessVnodeQueryMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY_CONTINUE)] = dndProcessVnodeQueryMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY_CONTINUE)] = dndProcessVnodeQueryMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH_RSP)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_UPDATE_TAG_VAL)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLE_META)] = dndProcessVnodeFetchMsg;
......@@ -159,17 +160,18 @@ static void dndProcessResponse(void *parent, SRpcMsg *pRsp, SEpSet *pEpSet) {
if (dndGetStat(pDnode) == DND_STAT_STOPPED) {
if (pRsp == NULL || pRsp->pCont == NULL) return;
dTrace("RPC %p, rsp:%s is ignored since dnode is stopping", pRsp->handle, TMSG_INFO(msgType));
dTrace("RPC %p, rsp:%s ignored since dnode exiting, app:%p", pRsp->handle, TMSG_INFO(msgType), pRsp->ahandle);
rpcFreeCont(pRsp->pCont);
return;
}
DndMsgFp fp = pMgmt->msgFp[TMSG_INDEX(msgType)];
if (fp != NULL) {
dTrace("RPC %p, rsp:%s will be processed, code:0x%x", pRsp->handle, TMSG_INFO(msgType), pRsp->code & 0XFFFF);
dTrace("RPC %p, rsp:%s will be processed, code:0x%x app:%p", pRsp->handle, TMSG_INFO(msgType), pRsp->code & 0XFFFF,
pRsp->ahandle);
(*fp)(pDnode, pRsp, pEpSet);
} else {
dError("RPC %p, rsp:%s not processed", pRsp->handle, TMSG_INFO(msgType));
dError("RPC %p, rsp:%s not processed, app:%p", pRsp->handle, TMSG_INFO(msgType), pRsp->ahandle);
rpcFreeCont(pRsp->pCont);
}
}
......@@ -196,7 +198,7 @@ static int32_t dndInitClient(SDnode *pDnode) {
pMgmt->clientRpc = rpcOpen(&rpcInit);
if (pMgmt->clientRpc == NULL) {
dError("failed to init rpc client");
dError("failed to init dnode rpc client");
return -1;
}
......@@ -219,40 +221,39 @@ static void dndProcessRequest(void *param, SRpcMsg *pReq, SEpSet *pEpSet) {
tmsg_t msgType = pReq->msgType;
if (msgType == TDMT_DND_NETWORK_TEST) {
dTrace("RPC %p, network test req, app:%p will be processed, code:0x%x", pReq->handle, pReq->ahandle, pReq->code);
dTrace("RPC %p, network test req will be processed, app:%p", pReq->handle, pReq->ahandle);
dndProcessStartupReq(pDnode, pReq);
return;
}
if (dndGetStat(pDnode) == DND_STAT_STOPPED) {
dError("RPC %p, req:%s app:%p is ignored since dnode exiting", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_DND_OFFLINE};
dError("RPC %p, req:%s ignored since dnode exiting, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_DND_OFFLINE, .ahandle = pReq->ahandle};
rpcSendResponse(&rspMsg);
rpcFreeCont(pReq->pCont);
return;
} else if (dndGetStat(pDnode) != DND_STAT_RUNNING) {
dError("RPC %p, req:%s app:%p is ignored since dnode not running", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_APP_NOT_READY};
dError("RPC %p, req:%s ignored since dnode not running, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pReq->ahandle};
rpcSendResponse(&rspMsg);
rpcFreeCont(pReq->pCont);
return;
}
if (pReq->pCont == NULL) {
dTrace("RPC %p, req:%s app:%p not processed since content is null", pReq->handle, TMSG_INFO(msgType),
pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_DND_INVALID_MSG_LEN};
dTrace("RPC %p, req:%s not processed since its empty, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_DND_INVALID_MSG_LEN, .ahandle = pReq->ahandle};
rpcSendResponse(&rspMsg);
return;
}
DndMsgFp fp = pMgmt->msgFp[TMSG_INDEX(msgType)];
if (fp != NULL) {
dTrace("RPC %p, req:%s app:%p will be processed", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
dTrace("RPC %p, req:%s will be processed, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
(*fp)(pDnode, pReq, pEpSet);
} else {
dError("RPC %p, req:%s app:%p is not processed since no handle", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED};
dError("RPC %p, req:%s not processed since no handle, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pReq->ahandle};
rpcSendResponse(&rspMsg);
rpcFreeCont(pReq->pCont);
}
......@@ -292,24 +293,24 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char
SDnode *pDnode = parent;
if (dndAuthInternalReq(parent, user, spi, encrypt, secret, ckey) == 0) {
dTrace("user:%s, get auth from internal mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
return 0;
}
if (dndGetUserAuthFromMnode(pDnode, user, spi, encrypt, secret, ckey) == 0) {
dTrace("user:%s, get auth from internal mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
return 0;
}
if (terrno != TSDB_CODE_APP_NOT_READY) {
dTrace("failed to get user auth from internal mnode since %s", terrstr());
dTrace("failed to get user auth from mnode since %s", terrstr());
return -1;
}
SAuthReq *pReq = rpcMallocCont(sizeof(SAuthReq));
tstrncpy(pReq->user, user, TSDB_USER_LEN);
SRpcMsg rpcMsg = {.pCont = pReq, .contLen = sizeof(SAuthReq), .msgType = TDMT_MND_AUTH};
SRpcMsg rpcMsg = {.pCont = pReq, .contLen = sizeof(SAuthReq), .msgType = TDMT_MND_AUTH, .ahandle = (void *)9528};
SRpcMsg rpcRsp = {0};
dTrace("user:%s, send user auth req to other mnodes, spi:%d encrypt:%d", user, pReq->spi, pReq->encrypt);
dndSendMsgToMnodeRecv(pDnode, &rpcMsg, &rpcRsp);
......@@ -353,7 +354,7 @@ static int32_t dndInitServer(SDnode *pDnode) {
pMgmt->serverRpc = rpcOpen(&rpcInit);
if (pMgmt->serverRpc == NULL) {
dError("failed to init rpc server");
dError("failed to init dnode rpc server");
return -1;
}
......
source/dnode/mgmt/impl/src/dndVnodes.c
浏览文件 @ 9db65e7a
......@@ -253,7 +253,7 @@ static int32_t dndGetVnodesFromFile(SDnode *pDnode, SWrapperCfg **ppCfgs, int32_
}
for (int32_t i = 0; i < vnodesNum; ++i) {
cJSON *vnode = cJSON_GetArrayItem(vnodes, i);
cJSON * vnode = cJSON_GetArrayItem(vnodes, i);
SWrapperCfg *pCfg = &pCfgs[i];
cJSON *vgId = cJSON_GetObjectItem(vnode, "vgId");
......@@ -382,7 +382,7 @@ static void *dnodeOpenVnodeFunc(void *param) {
dndReportStartup(pDnode, "open-vnodes", stepDesc);
SVnodeCfg cfg = {.pDnode = pDnode, .pTfs = pDnode->pTfs, .vgId = pCfg->vgId};
SVnode *pImpl = vnodeOpen(pCfg->path, &cfg);
SVnode * pImpl = vnodeOpen(pCfg->path, &cfg);
if (pImpl == NULL) {
dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
pThread->failed++;
......@@ -737,15 +737,9 @@ int32_t dndProcessCompactVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
return 0;
}
static void dndProcessVnodeQueryQueue(SVnodeObj *pVnode, SRpcMsg *pMsg) {
SRpcMsg *pRsp = NULL;
vnodeProcessQueryReq(pVnode->pImpl, pMsg, &pRsp);
}
static void dndProcessVnodeQueryQueue(SVnodeObj *pVnode, SRpcMsg *pMsg) { vnodeProcessQueryMsg(pVnode->pImpl, pMsg); }
static void dndProcessVnodeFetchQueue(SVnodeObj *pVnode, SRpcMsg *pMsg) {
SRpcMsg *pRsp = NULL;
vnodeProcessFetchReq(pVnode->pImpl, pMsg, &pRsp);
}
static void dndProcessVnodeFetchQueue(SVnodeObj *pVnode, SRpcMsg *pMsg) { vnodeProcessFetchMsg(pVnode->pImpl, pMsg); }
static void dndProcessVnodeWriteQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t numOfMsgs) {
SArray *pArray = taosArrayInit(numOfMsgs, sizeof(SRpcMsg *));
......@@ -916,27 +910,27 @@ static int32_t dndInitVnodeWorkers(SDnode *pDnode) {
int32_t maxWriteThreads = TMAX(pDnode->env.numOfCores, 1);
int32_t maxSyncThreads = TMAX(pDnode->env.numOfCores / 2, 1);
SWorkerPool *pPool = &pMgmt->queryPool;
pPool->name = "vnode-query";
pPool->min = minQueryThreads;
pPool->max = maxQueryThreads;
if (tWorkerInit(pPool) != 0) return -1;
SQWorkerPool *pQPool = &pMgmt->queryPool;
pQPool->name = "vnode-query";
pQPool->min = minQueryThreads;
pQPool->max = maxQueryThreads;
if (tQWorkerInit(pQPool) != 0) return -1;
pPool = &pMgmt->fetchPool;
pPool->name = "vnode-fetch";
pPool->min = minFetchThreads;
pPool->max = maxFetchThreads;
if (tWorkerInit(pPool) != 0) return -1;
SFWorkerPool *pFPool = &pMgmt->fetchPool;
pFPool->name = "vnode-fetch";
pFPool->min = minFetchThreads;
pFPool->max = maxFetchThreads;
if (tFWorkerInit(pFPool) != 0) return -1;
SMWorkerPool *pMPool = &pMgmt->writePool;
pMPool->name = "vnode-write";
pMPool->max = maxWriteThreads;
if (tMWorkerInit(pMPool) != 0) return -1;
SWWorkerPool *pWPool = &pMgmt->writePool;
pWPool->name = "vnode-write";
pWPool->max = maxWriteThreads;
if (tWWorkerInit(pWPool) != 0) return -1;
pMPool = &pMgmt->syncPool;
pMPool->name = "vnode-sync";
pMPool->max = maxSyncThreads;
if (tMWorkerInit(pMPool) != 0) return -1;
pWPool = &pMgmt->syncPool;
pWPool->name = "vnode-sync";
pWPool->max = maxSyncThreads;
if (tWWorkerInit(pWPool) != 0) return -1;
dDebug("vnode workers is initialized");
return 0;
......@@ -944,21 +938,21 @@ static int32_t dndInitVnodeWorkers(SDnode *pDnode) {
static void dndCleanupVnodeWorkers(SDnode *pDnode) {
SVnodesMgmt *pMgmt = &pDnode->vmgmt;
tWorkerCleanup(&pMgmt->fetchPool);
tWorkerCleanup(&pMgmt->queryPool);
tMWorkerCleanup(&pMgmt->writePool);
tMWorkerCleanup(&pMgmt->syncPool);
tFWorkerCleanup(&pMgmt->fetchPool);
tQWorkerCleanup(&pMgmt->queryPool);
tWWorkerCleanup(&pMgmt->writePool);
tWWorkerCleanup(&pMgmt->syncPool);
dDebug("vnode workers is closed");
}
static int32_t dndAllocVnodeQueue(SDnode *pDnode, SVnodeObj *pVnode) {
SVnodesMgmt *pMgmt = &pDnode->vmgmt;
pVnode->pWriteQ = tMWorkerAllocQueue(&pMgmt->writePool, pVnode, (FProcessItems)dndProcessVnodeWriteQueue);
pVnode->pApplyQ = tMWorkerAllocQueue(&pMgmt->writePool, pVnode, (FProcessItems)dndProcessVnodeApplyQueue);
pVnode->pSyncQ = tMWorkerAllocQueue(&pMgmt->syncPool, pVnode, (FProcessItems)dndProcessVnodeSyncQueue);
pVnode->pFetchQ = tWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FProcessItem)dndProcessVnodeFetchQueue);
pVnode->pQueryQ = tWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FProcessItem)dndProcessVnodeQueryQueue);
pVnode->pWriteQ = tWWorkerAllocQueue(&pMgmt->writePool, pVnode, (FItems)dndProcessVnodeWriteQueue);
pVnode->pApplyQ = tWWorkerAllocQueue(&pMgmt->writePool, pVnode, (FItems)dndProcessVnodeApplyQueue);
pVnode->pSyncQ = tWWorkerAllocQueue(&pMgmt->syncPool, pVnode, (FItems)dndProcessVnodeSyncQueue);
pVnode->pFetchQ = tFWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItem)dndProcessVnodeFetchQueue);
pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)dndProcessVnodeQueryQueue);
if (pVnode->pApplyQ == NULL || pVnode->pWriteQ == NULL || pVnode->pSyncQ == NULL || pVnode->pFetchQ == NULL ||
pVnode->pQueryQ == NULL) {
......@@ -971,11 +965,11 @@ static int32_t dndAllocVnodeQueue(SDnode *pDnode, SVnodeObj *pVnode) {
static void dndFreeVnodeQueue(SDnode *pDnode, SVnodeObj *pVnode) {
SVnodesMgmt *pMgmt = &pDnode->vmgmt;
tWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ);
tWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ);
tMWorkerFreeQueue(&pMgmt->writePool, pVnode->pWriteQ);
tMWorkerFreeQueue(&pMgmt->writePool, pVnode->pApplyQ);
tMWorkerFreeQueue(&pMgmt->syncPool, pVnode->pSyncQ);
tQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ);
tFWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ);
tWWorkerFreeQueue(&pMgmt->writePool, pVnode->pWriteQ);
tWWorkerFreeQueue(&pMgmt->writePool, pVnode->pApplyQ);
tWWorkerFreeQueue(&pMgmt->syncPool, pVnode->pSyncQ);
pVnode->pWriteQ = NULL;
pVnode->pApplyQ = NULL;
pVnode->pSyncQ = NULL;
......
source/dnode/mgmt/impl/src/dndWorker.c
浏览文件 @ 9db65e7a
......@@ -31,28 +31,28 @@ int32_t dndInitWorker(SDnode *pDnode, SDnodeWorker *pWorker, EWorkerType type, c
pWorker->pDnode = pDnode;
if (pWorker->type == DND_WORKER_SINGLE) {
SWorkerPool *pPool = &pWorker->pool;
SQWorkerPool *pPool = &pWorker->pool;
pPool->name = name;
pPool->min = minNum;
pPool->max = maxNum;
if (tWorkerInit(pPool) != 0) {
if (tQWorkerInit(pPool) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pWorker->queue = tWorkerAllocQueue(pPool, pDnode, (FProcessItem)queueFp);
pWorker->queue = tQWorkerAllocQueue(pPool, pDnode, (FItem)queueFp);
if (pWorker->queue == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
} else if (pWorker->type == DND_WORKER_MULTI) {
SMWorkerPool *pPool = &pWorker->mpool;
SWWorkerPool *pPool = &pWorker->mpool;
pPool->name = name;
pPool->max = maxNum;
if (tMWorkerInit(pPool) != 0) {
if (tWWorkerInit(pPool) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pWorker->queue = tMWorkerAllocQueue(pPool, pDnode, (FProcessItems)queueFp);
pWorker->queue = tWWorkerAllocQueue(pPool, pDnode, (FItems)queueFp);
if (pWorker->queue == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
......@@ -70,11 +70,11 @@ void dndCleanupWorker(SDnodeWorker *pWorker) {
}
if (pWorker->type == DND_WORKER_SINGLE) {
tWorkerCleanup(&pWorker->pool);
tWorkerFreeQueue(&pWorker->pool, pWorker->queue);
tQWorkerCleanup(&pWorker->pool);
tQWorkerFreeQueue(&pWorker->pool, pWorker->queue);
} else if (pWorker->type == DND_WORKER_MULTI) {
tMWorkerCleanup(&pWorker->mpool);
tMWorkerFreeQueue(&pWorker->mpool, pWorker->queue);
tWWorkerCleanup(&pWorker->mpool);
tWWorkerFreeQueue(&pWorker->mpool, pWorker->queue);
} else {
}
}
......
source/dnode/mnode/impl/src/mndTrans.c
浏览文件 @ 9db65e7a
......@@ -921,7 +921,7 @@ static int32_t mndProcessTransMsg(SMnodeMsg *pMsg) {
void mndTransPullup(SMnode *pMnode) {
STrans *pTrans = NULL;
void *pIter = NULL;
void * pIter = NULL;
while (1) {
pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
......@@ -930,4 +930,6 @@ void mndTransPullup(SMnode *pMnode) {
mndTransExecute(pMnode, pTrans);
sdbRelease(pMnode->pSdb, pTrans);
}
sdbWriteFile(pMnode->pSdb);
}
source/dnode/mnode/sdb/src/sdb.c
浏览文件 @ 9db65e7a
......@@ -64,11 +64,7 @@ SSdb *sdbInit(SSdbOpt *pOption) {
void sdbCleanup(SSdb *pSdb) {
mDebug("start to cleanup sdb");
if (pSdb->curVer > pSdb->lastCommitVer) {
mDebug("write sdb file for current ver:%" PRId64 " larger than last commit ver:%" PRId64, pSdb->curVer,
pSdb->lastCommitVer);
sdbWriteFile(pSdb);
}
sdbWriteFile(pSdb);
if (pSdb->currDir != NULL) {
tfree(pSdb->currDir);
......
source/dnode/mnode/sdb/src/sdbFile.c
浏览文件 @ 9db65e7a
......@@ -221,7 +221,7 @@ PARSE_SDB_DATA_ERROR:
return code;
}
int32_t sdbWriteFile(SSdb *pSdb) {
static int32_t sdbWriteFileImp(SSdb *pSdb) {
int32_t code = 0;
char tmpfile[PATH_MAX] = {0};
......@@ -229,7 +229,8 @@ int32_t sdbWriteFile(SSdb *pSdb) {
char curfile[PATH_MAX] = {0};
snprintf(curfile, sizeof(curfile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);
mDebug("start to write file:%s", curfile);
mDebug("start to write file:%s, current ver:%" PRId64 ", commit ver:%" PRId64, curfile, pSdb->curVer,
pSdb->lastCommitVer);
FileFd fd = taosOpenFileCreateWriteTrunc(tmpfile);
if (fd <= 0) {
......@@ -323,12 +324,20 @@ int32_t sdbWriteFile(SSdb *pSdb) {
return code;
}
int32_t sdbWriteFile(SSdb *pSdb) {
if (pSdb->curVer == pSdb->lastCommitVer) {
return 0;
}
return sdbWriteFileImp(pSdb);
}
int32_t sdbDeploy(SSdb *pSdb) {
if (sdbRunDeployFp(pSdb) != 0) {
return -1;
}
if (sdbWriteFile(pSdb) != 0) {
if (sdbWriteFileImp(pSdb) != 0) {
return -1;
}
......
source/dnode/vnode/inc/tq.h
浏览文件 @ 9db65e7a
......@@ -318,8 +318,8 @@ int tqRegisterContext(STqGroup*, void* ahandle);
int tqSendLaunchQuery(STqMsgItem*, int64_t offset);
#endif
int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg** ppRsp);
int32_t tqProcessSetConnReq(STQ* pTq, char* msg, SRpcMsg** ppRsp);
int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessSetConnReq(STQ* pTq, char* msg);
#ifdef __cplusplus
}
......
source/dnode/vnode/inc/tsdb.h
浏览文件 @ 9db65e7a
......@@ -75,7 +75,6 @@ typedef struct STsdbQueryCond {
} STsdbQueryCond;
typedef struct {
void *pTable;
TSKEY lastKey;
uint64_t uid;
} STableKeyInfo;
......@@ -104,8 +103,7 @@ typedef void* tsdbReadHandleT;
* @param qinfo query info handle from query processor
* @return
*/
tsdbReadHandleT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, uint64_t qId,
void *pRef);
tsdbReadHandleT *tsdbQueryTables(STsdb *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, uint64_t qId, uint64_t taskId);
/**
* Get the last row of the given query time window for all the tables in STableGroupInfo object.
......@@ -142,7 +140,7 @@ bool isTsdbCacheLastRow(tsdbReadHandleT* pTsdbReadHandle);
*/
int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len,
int16_t tagNameRelType, const char* tbnameCond, STableGroupInfo* pGroupInfo,
SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId);
SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId, uint64_t taskId);
/**
* get num of rows in mem table
*
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 9db65e7a
......@@ -33,6 +33,7 @@ extern "C" {
typedef struct SVnode SVnode;
typedef struct SDnode SDnode;
typedef int32_t (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
typedef struct STqCfg {
// TODO
......@@ -64,6 +65,7 @@ typedef struct {
const char *charset;
uint16_t nthreads; // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
PutReqToVQueryQFp putReqToVQueryQFp;
SendReqToDnodeFp sendReqToDnodeFp;
} SVnodeOpt;
typedef struct STqReadHandle {
......@@ -159,20 +161,18 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
*
* @param pVnode The vnode object.
* @param pMsg The request message
* @param pRsp The response message
* @return int 0 for success, -1 for failure
*/
int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg);
/**
* @brief Process a fetch message.
*
* @param pVnode The vnode object.
* @param pMsg The request message
* @param pRsp The response message
* @return int 0 for success, -1 for failure
*/
int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg);
/* ------------------------ SVnodeCfg ------------------------ */
/**
......
source/dnode/vnode/src/inc/vnd.h
浏览文件 @ 9db65e7a
......@@ -55,6 +55,7 @@ typedef struct SVnodeMgr {
// For vnode Mgmt
SDnode* pDnode;
PutReqToVQueryQFp putReqToVQueryQFp;
SendReqToDnodeFp sendReqToDnodeFp;
} SVnodeMgr;
extern SVnodeMgr vnodeMgr;
......@@ -85,6 +86,7 @@ struct SVnode {
int vnodeScheduleTask(SVnodeTask* task);
int32_t vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq);
void vnodeSendReqToDnode(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq);
// For Log
extern int32_t vDebugFlag;
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 9db65e7a
......@@ -667,8 +667,8 @@ int tqItemSSize() {
}
#endif
int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg** ppRsp) {
SMqConsumeReq* pReq = pMsg->pCont;
int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
SMqConsumeReq* pReq = pMsg->pCont;
SRpcMsg rpcMsg;
int64_t reqId = pReq->reqId;
int64_t consumerId = pReq->consumerId;
......@@ -767,7 +767,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg, SRpcMsg** ppRsp) {
return 0;
}
int32_t tqProcessSetConnReq(STQ* pTq, char* msg, SRpcMsg** ppRsp) {
int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
SMqSetCVgReq req;
tDecodeSMqSetCVgReq(msg, &req);
STqConsumerHandle* pConsumer = calloc(sizeof(STqConsumerHandle), 1);
......
source/dnode/vnode/src/vnd/vnodeMgr.c
浏览文件 @ 9db65e7a
......@@ -26,6 +26,7 @@ int vnodeInit(const SVnodeOpt *pOption) {
vnodeMgr.stop = false;
vnodeMgr.putReqToVQueryQFp = pOption->putReqToVQueryQFp;
vnodeMgr.sendReqToDnodeFp = pOption->sendReqToDnodeFp;
// Start commit handers
if (pOption->nthreads > 0) {
......@@ -96,8 +97,14 @@ int32_t vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq) {
return (*vnodeMgr.putReqToVQueryQFp)(pVnode->pDnode, pReq);
}
void vnodeSendReqToDnode(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq) {
(*vnodeMgr.sendReqToDnodeFp)(pVnode->pDnode, epSet, pReq);
}
/* ------------------------ STATIC METHODS ------------------------ */
static void* loop(void* arg) {
setThreadName("vnode-commit");
SVnodeTask* pTask;
for (;;) {
pthread_mutex_lock(&(vnodeMgr.mutex));
......
source/dnode/vnode/src/vnd/vnodeQuery.c
浏览文件 @ 9db65e7a
......@@ -17,14 +17,14 @@
#include "vnd.h"
static int32_t vnodeGetTableList(SVnode *pVnode, SRpcMsg *pMsg);
static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg);
int vnodeQueryOpen(SVnode *pVnode) {
return qWorkerInit(NODE_TYPE_VNODE, pVnode->vgId, NULL, (void **)&pVnode->pQuery, pVnode,
(putReqToQueryQFp)vnodePutReqToVQueryQ);
(putReqToQueryQFp)vnodePutReqToVQueryQ, (sendReqToDnodeFp)vnodeSendReqToDnode);
}
int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
vTrace("message in query queue is processing");
switch (pMsg->msgType) {
......@@ -38,11 +38,13 @@ int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
}
}
int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) {
vTrace("message in fetch queue is processing");
switch (pMsg->msgType) {
case TDMT_VND_FETCH:
return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg);
case TDMT_VND_FETCH_RSP:
return qWorkerProcessFetchRsp(pVnode, pVnode->pQuery, pMsg);
case TDMT_VND_RES_READY:
return qWorkerProcessReadyMsg(pVnode, pVnode->pQuery, pMsg);
case TDMT_VND_TASKS_STATUS:
......@@ -57,24 +59,24 @@ int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
return vnodeGetTableList(pVnode, pMsg);
// return qWorkerProcessShowFetchMsg(pVnode->pMeta, pVnode->pQuery, pMsg);
case TDMT_VND_TABLE_META:
return vnodeGetTableMeta(pVnode, pMsg, pRsp);
return vnodeGetTableMeta(pVnode, pMsg);
case TDMT_VND_CONSUME:
return tqProcessConsumeReq(pVnode->pTq, pMsg, pRsp);
return tqProcessConsumeReq(pVnode->pTq, pMsg);
default:
vError("unknown msg type:%d in fetch queue", pMsg->msgType);
return TSDB_CODE_VND_APP_ERROR;
}
}
static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
STableInfoReq * pReq = (STableInfoReq *)(pMsg->pCont);
STbCfg * pTbCfg = NULL;
STbCfg * pStbCfg = NULL;
tb_uid_t uid;
int32_t nCols;
int32_t nTagCols;
SSchemaWrapper *pSW;
STableMetaRsp * pTbMetaMsg = NULL;
SSchemaWrapper *pSW = NULL;
STableMetaRsp *pTbMetaMsg = NULL;
SSchema * pTagSchema;
SRpcMsg rpcMsg;
int msgLen = 0;
......@@ -145,15 +147,22 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
_exit:
free(pSW->pSchema);
free(pSW);
free(pTbCfg->name);
free(pTbCfg);
if (pTbCfg->type == META_SUPER_TABLE) {
free(pTbCfg->stbCfg.pTagSchema);
} else if (pTbCfg->type == META_SUPER_TABLE) {
kvRowFree(pTbCfg->ctbCfg.pTag);
if (pSW != NULL) {
tfree(pSW->pSchema);
tfree(pSW);
}
if (pTbCfg) {
tfree(pTbCfg->name);
if (pTbCfg->type == META_SUPER_TABLE) {
free(pTbCfg->stbCfg.pTagSchema);
} else if (pTbCfg->type == META_SUPER_TABLE) {
kvRowFree(pTbCfg->ctbCfg.pTag);
}
tfree(pTbCfg);
}
rpcMsg.handle = pMsg->handle;
rpcMsg.ahandle = pMsg->ahandle;
rpcMsg.pCont = pTbMetaMsg;
......
source/dnode/vnode/src/vnd/vnodeWrite.c
浏览文件 @ 9db65e7a
......@@ -115,7 +115,7 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
}
break;
case TDMT_VND_MQ_SET_CONN: {
if (tqProcessSetConnReq(pVnode->pTq, POINTER_SHIFT(ptr, sizeof(SMsgHead)), NULL) < 0) {
if (tqProcessSetConnReq(pVnode->pTq, POINTER_SHIFT(ptr, sizeof(SMsgHead))) < 0) {
// TODO: handle error
}
} break;
......
source/libs/executor/inc/executil.h
浏览文件 @ 9db65e7a
......@@ -38,7 +38,7 @@
#define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t))
#define GET_RES_EXT_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t) + POINTER_BYTES)
#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.queryId)
#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.str)
#define curTimeWindowIndex(_winres) ((_winres)->curIndex)
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 9db65e7a
......@@ -239,7 +239,7 @@ typedef struct STaskIdInfo {
uint64_t queryId; // this is also a request id
uint64_t subplanId;
uint64_t templateId;
uint64_t taskId; // this is a subplan id
char *str;
} STaskIdInfo;
typedef struct SExecTaskInfo {
......@@ -250,9 +250,8 @@ typedef struct SExecTaskInfo {
STaskCostInfo cost;
int64_t owner; // if it is in execution
int32_t code;
uint64_t totalRows; // total number of rows
STableGroupInfo tableqinfoGroupInfo; // this is a group array list, including SArray<STableQueryInfo*> structure
pthread_mutex_t lock; // used to synchronize the rsp/query threads
char *sql; // query sql string
jmp_buf env; //
struct SOperatorInfo *pRoot;
......@@ -378,8 +377,10 @@ typedef struct SExchangeInfo {
SSDataBlock *pResult;
int32_t current;
uint64_t rowsOfCurrentSource;
uint64_t bytes; // total load bytes from remote
uint64_t totalRows;
uint64_t totalSize; // total load bytes from remote
uint64_t totalRows; // total number of rows
uint64_t totalElapsed;// total elapsed time
} SExchangeInfo;
typedef struct STableScanInfo {
......@@ -622,8 +623,6 @@ int32_t createIndirectQueryFuncExprFromMsg(SQueryTableReq *pQueryMsg, int32_t nu
int32_t createQueryFilter(char *data, uint16_t len, SFilterInfo** pFilters);
SGroupbyExpr *createGroupbyExprFromMsg(SQueryTableReq *pQueryMsg, SColIndex *pColIndex, int32_t *code);
SQInfo *createQInfoImpl(SQueryTableReq *pQueryMsg, SGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, SFilterInfo* pFilters, int32_t vgId, char* sql, uint64_t qId, struct SUdfInfo* pUdfInfo);
int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, void* sourceOptr, SQInfo* pQInfo, STaskParam* param, char* start,
int32_t prevResultLen, void* merger);
......@@ -639,32 +638,29 @@ int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg);
bool isTaskKilled(SExecTaskInfo *pTaskInfo);
int32_t checkForQueryBuf(size_t numOfTables);
bool checkNeedToCompressQueryCol(SQInfo *pQInfo);
bool doBuildResCheck(SQInfo* pQInfo);
void setQueryStatus(STaskRuntimeEnv *pRuntimeEnv, int8_t status);
bool onlyQueryTags(STaskAttr* pQueryAttr);
//void destroyUdfInfo(struct SUdfInfo* pUdfInfo);
bool isValidQInfo(void *param);
int32_t doDumpQueryResult(SQInfo *pQInfo, char *data, int8_t compressed, int32_t *compLen);
size_t getResultSize(SQInfo *pQInfo, int64_t *numOfRows);
void setQueryKilled(SQInfo *pQInfo);
void setTaskKilled(SExecTaskInfo *pTaskInfo);
void publishOperatorProfEvent(SOperatorInfo* operatorInfo, EQueryProfEventType eventType);
void publishQueryAbortEvent(SExecTaskInfo * pTaskInfo, int32_t code);
void calculateOperatorProfResults(SQInfo* pQInfo);
void queryCostStatis(SQInfo *pQInfo);
void queryCostStatis(SExecTaskInfo *pTaskInfo);
void doDestroyTask(SQInfo *pQInfo);
void doDestroyTask(SExecTaskInfo *pTaskInfo);
void freeQueryAttr(STaskAttr *pQuery);
int32_t getMaximumIdleDurationSec();
void doInvokeUdf(struct SUdfInfo* pUdfInfo, SQLFunctionCtx *pCtx, int32_t idx, int32_t type);
void setTaskStatus(SExecTaskInfo *pTaskInfo, int8_t status);
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle);
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle, uint64_t taskId);
#endif // TDENGINE_EXECUTORIMPL_H
source/libs/executor/src/dataDispatcher.c
浏览文件 @ 9db65e7a
......@@ -121,11 +121,19 @@ static void toDataCacheEntry(const SDataDispatchHandle* pHandle, const SInputDat
}
static bool allocBuf(SDataDispatchHandle* pDispatcher, const SInputData* pInput, SDataDispatchBuf* pBuf) {
if (taosQueueSize(pDispatcher->pDataBlocks) >= pDispatcher->pManager->cfg.maxDataBlockNumPerQuery) {
uint32_t capacity = pDispatcher->pManager->cfg.maxDataBlockNumPerQuery;
if (taosQueueSize(pDispatcher->pDataBlocks) > capacity) {
qError("SinkNode queue is full, no capacity, max:%d, current:%d, no capacity", capacity,
taosQueueSize(pDispatcher->pDataBlocks));
return false;
}
pBuf->allocSize = DATA_META_LENGTH(pInput->pTableRetrieveTsMap) + pDispatcher->schema.resultRowSize * pInput->pData->info.rows;
pBuf->pData = malloc(pBuf->allocSize);
if (pBuf->pData == NULL) {
qError("SinkNode failed to malloc memory, size:%d, code:%d", pBuf->allocSize, TAOS_SYSTEM_ERROR(errno));
}
return NULL != pBuf->pData;
}
......
source/libs/executor/src/executor.c
浏览文件 @ 9db65e7a
......@@ -18,20 +18,20 @@
#include "executorimpl.h"
#include "planner.h"
static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, uint64_t reqId) {
static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id) {
ASSERT(pOperator != NULL);
if (pOperator->operatorType != OP_StreamScan) {
if (pOperator->numOfDownstream == 0) {
qError("failed to find stream scan operator to set the input data block, reqId:0x%" PRIx64, reqId);
qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);
return TSDB_CODE_QRY_APP_ERROR;
}
if (pOperator->numOfDownstream > 1) { // not handle this in join query
qError("join not supported for stream block scan, reqId:0x%" PRIx64, reqId);
qError("join not supported for stream block scan, %s" PRIx64, id);
return TSDB_CODE_QRY_APP_ERROR;
}
return doSetStreamBlock(pOperator->pDownstream[0], input, reqId);
return doSetStreamBlock(pOperator->pDownstream[0], input, id);
} else {
SStreamBlockScanInfo* pInfo = pOperator->info;
tqReadHandleSetMsg(pInfo->readerHandle, input, 0);
......@@ -52,9 +52,9 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, void* input) {
int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, GET_TASKID(pTaskInfo));
if (code != TSDB_CODE_SUCCESS) {
qError("failed to set the stream block data, reqId:0x%"PRIx64, GET_TASKID(pTaskInfo));
qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
} else {
qDebug("set the stream block successfully, reqId:0x%"PRIx64, GET_TASKID(pTaskInfo));
qDebug("%s set the stream block successfully", GET_TASKID(pTaskInfo));
}
return code;
......@@ -81,7 +81,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle) {
}
qTaskInfo_t pTaskInfo = NULL;
code = qCreateExecTask(streamReadHandle, 0, plan, &pTaskInfo, NULL);
code = qCreateExecTask(streamReadHandle, 0, 0, plan, &pTaskInfo, NULL);
if (code != TSDB_CODE_SUCCESS) {
// TODO: destroy SSubplan & pTaskInfo
terrno = code;
......
source/libs/executor/src/executorMain.c
浏览文件 @ 9db65e7a
......@@ -69,11 +69,11 @@ void freeParam(STaskParam *param) {
tfree(param->prevResult);
}
int32_t qCreateExecTask(void* readHandle, int32_t vgId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle) {
int32_t qCreateExecTask(void* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle) {
assert(readHandle != NULL && pSubplan != NULL);
SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo;
int32_t code = createExecTaskInfoImpl(pSubplan, pTask, readHandle);
int32_t code = createExecTaskInfoImpl(pSubplan, pTask, readHandle, taskId);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
......@@ -141,16 +141,10 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
int64_t threadId = taosGetSelfPthreadId();
// todo: remove it.
if (tinfo == NULL) {
return TSDB_CODE_SUCCESS;
}
*pRes = NULL;
int64_t curOwner = 0;
if ((curOwner = atomic_val_compare_exchange_64(&pTaskInfo->owner, 0, threadId)) != 0) {
qError("QInfo:0x%" PRIx64 "-%p qhandle is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo,
qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo,
(void*)curOwner);
pTaskInfo->code = TSDB_CODE_QRY_IN_EXEC;
return pTaskInfo->code;
......@@ -161,7 +155,7 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
}
if (isTaskKilled(pTaskInfo)) {
qDebug("QInfo:0x%" PRIx64 " it is already killed, abort", GET_TASKID(pTaskInfo));
qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));
return TSDB_CODE_SUCCESS;
}
......@@ -170,12 +164,12 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
if (ret != TSDB_CODE_SUCCESS) {
publishQueryAbortEvent(pTaskInfo, ret);
pTaskInfo->code = ret;
qDebug("QInfo:0x%" PRIx64 " query abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo),
qDebug("%s task abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo),
tstrerror(pTaskInfo->code));
return pTaskInfo->code;
}
qDebug("QInfo:0x%" PRIx64 " query task is launched", GET_TASKID(pTaskInfo));
qDebug("%s execTask is launched", GET_TASKID(pTaskInfo));
bool newgroup = false;
publishOperatorProfEvent(pTaskInfo->pRoot, QUERY_PROF_BEFORE_OPERATOR_EXEC);
......@@ -184,66 +178,25 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
st = taosGetTimestampUs();
*pRes = pTaskInfo->pRoot->exec(pTaskInfo->pRoot, &newgroup);
pTaskInfo->cost.elapsedTime += (taosGetTimestampUs() - st);
uint64_t el = (taosGetTimestampUs() - st);
pTaskInfo->cost.elapsedTime += el;
publishOperatorProfEvent(pTaskInfo->pRoot, QUERY_PROF_AFTER_OPERATOR_EXEC);
if (NULL == *pRes) {
*useconds = pTaskInfo->cost.elapsedTime;
}
qDebug("QInfo:0x%" PRIx64 " query paused, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d",
GET_TASKID(pTaskInfo), 0, 0L, 0);
int32_t current = (*pRes != NULL)? (*pRes)->info.rows:0;
pTaskInfo->totalRows += current;
qDebug("%s task suspended, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms",
GET_TASKID(pTaskInfo), current, pTaskInfo->totalRows, 0, el/1000.0);
atomic_store_64(&pTaskInfo->owner, 0);
return pTaskInfo->code;
}
int32_t qRetrieveQueryResultInfo(qTaskInfo_t qinfo, bool* buildRes, void* pRspContext) {
SQInfo *pQInfo = (SQInfo *)qinfo;
if (pQInfo == NULL || !isValidQInfo(pQInfo)) {
qError("QInfo invalid qhandle");
return TSDB_CODE_QRY_INVALID_QHANDLE;
}
*buildRes = false;
if (IS_QUERY_KILLED(pQInfo)) {
qDebug("QInfo:0x%"PRIx64" query is killed, code:0x%08x", pQInfo->qId, pQInfo->code);
return pQInfo->code;
}
int32_t code = TSDB_CODE_SUCCESS;
if (tsRetrieveBlockingModel) {
pQInfo->rspContext = pRspContext;
tsem_wait(&pQInfo->ready);
*buildRes = true;
code = pQInfo->code;
} else {
STaskRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv;
STaskAttr *pQueryAttr = pQInfo->runtimeEnv.pQueryAttr;
pthread_mutex_lock(&pQInfo->lock);
assert(pQInfo->rspContext == NULL);
if (pQInfo->dataReady == QUERY_RESULT_READY) {
*buildRes = true;
qDebug("QInfo:0x%"PRIx64" retrieve result info, rowsize:%d, rows:%d, code:%s", pQInfo->qId, pQueryAttr->resultRowSize,
GET_NUM_OF_RESULTS(pRuntimeEnv), tstrerror(pQInfo->code));
} else {
*buildRes = false;
qDebug("QInfo:0x%"PRIx64" retrieve req set query return result after paused", pQInfo->qId);
pQInfo->rspContext = pRspContext;
assert(pQInfo->rspContext != NULL);
}
code = pQInfo->code;
pthread_mutex_unlock(&pQInfo->lock);
}
return code;
}
void* qGetResultRetrieveMsg(qTaskInfo_t qinfo) {
SQInfo* pQInfo = (SQInfo*) qinfo;
assert(pQInfo != NULL);
......@@ -252,18 +205,18 @@ void* qGetResultRetrieveMsg(qTaskInfo_t qinfo) {
}
int32_t qKillTask(qTaskInfo_t qinfo) {
SQInfo *pQInfo = (SQInfo *)qinfo;
SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo;
if (pQInfo == NULL || !isValidQInfo(pQInfo)) {
if (pTaskInfo == NULL) {
return TSDB_CODE_QRY_INVALID_QHANDLE;
}
qDebug("QInfo:0x%"PRIx64" query killed", pQInfo->qId);
setQueryKilled(pQInfo);
qDebug("%s execTask killed", GET_TASKID(pTaskInfo));
setTaskKilled(pTaskInfo);
// Wait for the query executing thread being stopped/
// Once the query is stopped, the owner of qHandle will be cleared immediately.
while (pQInfo->owner != 0) {
while (pTaskInfo->owner != 0) {
taosMsleep(100);
}
......@@ -271,14 +224,14 @@ int32_t qKillTask(qTaskInfo_t qinfo) {
}
int32_t qAsyncKillTask(qTaskInfo_t qinfo) {
SQInfo *pQInfo = (SQInfo *)qinfo;
SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo;
if (pQInfo == NULL || !isValidQInfo(pQInfo)) {
if (pTaskInfo == NULL) {
return TSDB_CODE_QRY_INVALID_QHANDLE;
}
qDebug("QInfo:0x%"PRIx64" query async killed", pQInfo->qId);
setQueryKilled(pQInfo);
qDebug("%s execTask async killed", GET_TASKID(pTaskInfo));
setTaskKilled(pTaskInfo);
return TSDB_CODE_SUCCESS;
}
......@@ -293,145 +246,12 @@ int32_t qIsTaskCompleted(qTaskInfo_t qinfo) {
return isTaskKilled(pTaskInfo) || Q_STATUS_EQUAL(pTaskInfo->status, TASK_OVER);
}
void qDestroyTask(qTaskInfo_t qHandle) {
SQInfo* pQInfo = (SQInfo*) qHandle;
if (!isValidQInfo(pQInfo)) {
return;
}
qDebug("QInfo:0x%"PRIx64" query completed", pQInfo->qId);
queryCostStatis(pQInfo); // print the query cost summary
doDestroyTask(pQInfo);
}
void* qOpenTaskMgmt(int32_t vgId) {
const int32_t refreshHandleInterval = 30; // every 30 seconds, refresh handle pool
char cacheName[128] = {0};
sprintf(cacheName, "qhandle_%d", vgId);
STaskMgmt* pTaskMgmt = calloc(1, sizeof(STaskMgmt));
if (pTaskMgmt == NULL) {
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
return NULL;
}
pTaskMgmt->qinfoPool = taosCacheInit(TSDB_CACHE_PTR_KEY, refreshHandleInterval, true, freeqinfoFn, cacheName);
pTaskMgmt->closed = false;
pTaskMgmt->vgId = vgId;
pthread_mutex_init(&pTaskMgmt->lock, NULL);
qDebug("vgId:%d, open queryTaskMgmt success", vgId);
return pTaskMgmt;
}
void qTaskMgmtNotifyClosing(void* pQMgmt) {
if (pQMgmt == NULL) {
return;
}
STaskMgmt* pQueryMgmt = pQMgmt;
qInfo("vgId:%d, set querymgmt closed, wait for all queries cancelled", pQueryMgmt->vgId);
pthread_mutex_lock(&pQueryMgmt->lock);
pQueryMgmt->closed = true;
pthread_mutex_unlock(&pQueryMgmt->lock);
taosCacheRefresh(pQueryMgmt->qinfoPool, taskMgmtKillTaskFn, NULL);
}
void qQueryMgmtReOpen(void *pQMgmt) {
if (pQMgmt == NULL) {
return;
}
STaskMgmt *pQueryMgmt = pQMgmt;
qInfo("vgId:%d, set querymgmt reopen", pQueryMgmt->vgId);
pthread_mutex_lock(&pQueryMgmt->lock);
pQueryMgmt->closed = false;
pthread_mutex_unlock(&pQueryMgmt->lock);
}
void qCleanupTaskMgmt(void* pQMgmt) {
if (pQMgmt == NULL) {
return;
}
STaskMgmt* pQueryMgmt = pQMgmt;
int32_t vgId = pQueryMgmt->vgId;
assert(pQueryMgmt->closed);
SCacheObj* pqinfoPool = pQueryMgmt->qinfoPool;
pQueryMgmt->qinfoPool = NULL;
taosCacheCleanup(pqinfoPool);
pthread_mutex_destroy(&pQueryMgmt->lock);
tfree(pQueryMgmt);
qDebug("vgId:%d, queryMgmt cleanup completed", vgId);
}
void** qRegisterTask(void* pMgmt, uint64_t qId, void *qInfo) {
if (pMgmt == NULL) {
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
return NULL;
}
STaskMgmt *pQueryMgmt = pMgmt;
if (pQueryMgmt->qinfoPool == NULL) {
qError("QInfo:0x%"PRIx64"-%p failed to add qhandle into qMgmt, since qMgmt is closed", qId, (void*)qInfo);
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
return NULL;
}
pthread_mutex_lock(&pQueryMgmt->lock);
if (pQueryMgmt->closed) {
pthread_mutex_unlock(&pQueryMgmt->lock);
qError("QInfo:0x%"PRIx64"-%p failed to add qhandle into cache, since qMgmt is colsing", qId, (void*)qInfo);
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
return NULL;
} else {
void** handle = taosCachePut(pQueryMgmt->qinfoPool, &qId, sizeof(qId), &qInfo, sizeof(TSDB_CACHE_PTR_TYPE),
(getMaximumIdleDurationSec()*1000));
pthread_mutex_unlock(&pQueryMgmt->lock);
return handle;
}
}
void** qAcquireTask(void* pMgmt, uint64_t _key) {
STaskMgmt *pQueryMgmt = pMgmt;
if (pQueryMgmt->closed) {
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
return NULL;
}
if (pQueryMgmt->qinfoPool == NULL) {
terrno = TSDB_CODE_QRY_INVALID_QHANDLE;
return NULL;
}
void** handle = taosCacheAcquireByKey(pQueryMgmt->qinfoPool, &_key, sizeof(_key));
if (handle == NULL || *handle == NULL) {
terrno = TSDB_CODE_QRY_INVALID_QHANDLE;
return NULL;
} else {
return handle;
}
}
void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle) {
STaskMgmt *pQueryMgmt = pMgmt;
if (pQueryMgmt->qinfoPool == NULL) {
return NULL;
}
void qDestroyTask(qTaskInfo_t qTaskHandle) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) qTaskHandle;
qDebug("%s execTask completed, numOfRows:%"PRId64, GET_TASKID(pTaskInfo), pTaskInfo->totalRows);
taosCacheRelease(pQueryMgmt->qinfoPool, pQInfo, freeHandle);
return 0;
queryCostStatis(pTaskInfo); // print the query cost summary
doDestroyTask(pTaskInfo);
}
#if 0
......@@ -445,8 +265,8 @@ int32_t qKillQueryByQId(void* pMgmt, int64_t qId, int32_t waitMs, int32_t waitCo
if (pQInfo == NULL || !isValidQInfo(pQInfo)) {
return TSDB_CODE_QRY_INVALID_QHANDLE;
}
qWarn("QId:0x%"PRIx64" be killed(no memory commit).", pQInfo->qId);
setQueryKilled(pQInfo);
qWarn("%s be killed(no memory commit).", pQInfo->qId);
setTaskKilled(pQInfo);
// wait query stop
int32_t loop = 0;
......
source/libs/executor/test/executorTests.cpp
浏览文件 @ 9db65e7a
......@@ -219,7 +219,7 @@ TEST(testCase, build_executor_tree_Test) {
SExecTaskInfo* pTaskInfo = nullptr;
DataSinkHandle sinkHandle = nullptr;
int32_t code = qCreateExecTask((void*) 1, 2, NULL, (void**) &pTaskInfo, &sinkHandle);
int32_t code = qCreateExecTask((void*) 1, 2, 1, NULL, (void**) &pTaskInfo, &sinkHandle);
}
#pragma GCC diagnostic pop
\ No newline at end of file
source/libs/planner/src/logicPlan.c
浏览文件 @ 9db65e7a
......@@ -413,7 +413,6 @@ static void doDestroyQueryNode(SQueryPlanNode* pQueryNode) {
tfree(pQueryTableInfo->tableName);
}
printf("----------->Free:%p\n", pQueryNode->pExpr);
taosArrayDestroy(pQueryNode->pExpr);
tfree(pQueryNode->pExtInfo);
......
source/libs/planner/src/physicalPlanJson.c
浏览文件 @ 9db65e7a
......@@ -1125,8 +1125,8 @@ int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
*str = cJSON_Print(json);
cJSON_Delete(json);
/*printf("====Physical plan:====\n");*/
/*printf("%s\n", *str);*/
// printf("====Physical plan:====\n");
// printf("%s\n", *str);
*len = strlen(*str) + 1;
return TSDB_CODE_SUCCESS;
}
......
source/libs/qworker/inc/qworkerInt.h
浏览文件 @ 9db65e7a
......@@ -137,6 +137,7 @@ typedef struct SQWorkerMgmt {
SHashObj *ctxHash; //key: queryId+taskId, value: SQWTaskCtx
void *nodeObj;
putReqToQueryQFp putToQueueFp;
sendReqToDnodeFp sendReqFp;
} SQWorkerMgmt;
#define QW_FPARAMS_DEF SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId
......@@ -173,17 +174,17 @@ typedef struct SQWorkerMgmt {
#define QW_SCH_ELOG(param, ...) qError("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__)
#define QW_SCH_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64" " param, mgmt, sId, __VA_ARGS__)
#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_DLOG_E(param) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
#define QW_SCH_TASK_ELOG(param, ...) qError("QW:%p SID:%"PRIx64",QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_ELOG(param, ...) qError("QW:%p SID:%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_WLOG(param, ...) qWarn("QW:%p SID:%"PRIx64",QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64",QID:0x%"PRIx64",TID:%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_SCH_TASK_DLOG(param, ...) qDebug("QW:%p SID:%"PRIx64",QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, sId, qId, tId, __VA_ARGS__)
#define QW_LOCK_DEBUG(...) do { if (gQWDebug.lockDebug) { qDebug(__VA_ARGS__); } } while (0)
......
source/libs/qworker/inc/qworkerMsg.h
浏览文件 @ 9db65e7a
......@@ -32,7 +32,7 @@ int32_t qwProcessDrop(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t t
int32_t qwBuildAndSendDropRsp(void *connection, int32_t code);
int32_t qwBuildAndSendCancelRsp(SRpcMsg *pMsg, int32_t code);
int32_t qwBuildAndSendFetchRsp(void *connection, SRetrieveTableRsp *pRsp, int32_t dataLength, int32_t code);
void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len);
void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete);
int32_t qwBuildAndSendCQueryMsg(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, void *connection);
int32_t qwBuildAndSendSchSinkMsg(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, void *connection);
int32_t qwBuildAndSendReadyRsp(void *connection, int32_t code);
......
source/libs/qworker/src/qworker.c
浏览文件 @ 9db65e7a
......@@ -456,7 +456,7 @@ _return:
QW_RET(code);
}
int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) {
int32_t code = 0;
bool qcontinue = true;
SSDataBlock* pRes = NULL;
......@@ -467,11 +467,11 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
DataSinkHandle sinkHandle = ctx->sinkHandle;
while (true) {
QW_TASK_DLOG("start to execTask in executor, loopIdx:%d", i++);
QW_TASK_DLOG("start to execTask, loopIdx:%d", i++);
code = qExecTask(*taskHandle, &pRes, &useconds);
if (code) {
QW_TASK_ELOG("qExecTask failed, code:%x", code);
QW_TASK_ELOG("qExecTask failed, code:%s", tstrerror(code));
QW_ERR_JRET(code);
}
......@@ -485,6 +485,10 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
if (TASK_TYPE_TEMP == ctx->taskType) {
qwFreeTaskHandle(QW_FPARAMS(), taskHandle);
}
if (queryEnd) {
*queryEnd = true;
}
break;
}
......@@ -492,7 +496,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx) {
SInputData inputData = {.pData = pRes, .pTableRetrieveTsMap = NULL};
code = dsPutDataBlock(sinkHandle, &inputData, &qcontinue);
if (code) {
QW_TASK_ELOG("dsPutDataBlock failed, code:%x", code);
QW_TASK_ELOG("dsPutDataBlock failed, code:%s", tstrerror(code));
QW_ERR_JRET(code);
}
......@@ -587,12 +591,7 @@ int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void
QW_ERR_RET(code);
}
queryEnd = pOutput->queryEnd;
pOutput->queryEnd = false;
if (DS_BUF_EMPTY == pOutput->bufStatus && queryEnd) {
pOutput->queryEnd = true;
if (DS_BUF_EMPTY == pOutput->bufStatus && pOutput->queryEnd) {
QW_SCH_TASK_DLOG("task all fetched, status:%d", JOB_TASK_STATUS_SUCCEED);
QW_ERR_RET(qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCCEED));
}
......@@ -974,7 +973,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType) {
QW_ERR_JRET(code);
}
code = qCreateExecTask(qwMsg->node, 0, (struct SSubplan *)plan, &pTaskInfo, &sinkHandle);
code = qCreateExecTask(qwMsg->node, 0, tId, (struct SSubplan *)plan, &pTaskInfo, &sinkHandle);
if (code) {
QW_TASK_ELOG("qCreateExecTask failed, code:%s", tstrerror(code));
QW_ERR_JRET(code);
......@@ -996,7 +995,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType) {
atomic_store_ptr(&ctx->sinkHandle, sinkHandle);
if (pTaskInfo && sinkHandle) {
QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx));
QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx, NULL));
}
_return:
......@@ -1083,6 +1082,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
SQWPhaseOutput output = {0};
void *rsp = NULL;
int32_t dataLen = 0;
bool queryEnd = false;
do {
QW_ERR_JRET(qwHandlePrePhaseEvents(QW_FPARAMS(), QW_PHASE_PRE_CQUERY, &input, &output));
......@@ -1102,7 +1102,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
DataSinkHandle sinkHandle = ctx->sinkHandle;
QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx));
QW_ERR_JRET(qwExecTask(QW_FPARAMS(), ctx, &queryEnd));
if (QW_IS_EVENT_RECEIVED(ctx, QW_EVENT_FETCH)) {
SOutputData sOutput = {0};
......@@ -1114,13 +1114,10 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
// RC WARNING
atomic_store_8(&ctx->queryContinue, 1);
}
if (sOutput.queryEnd) {
needStop = true;
}
if (rsp) {
qwBuildFetchRsp(rsp, &sOutput, dataLen);
bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_FETCH);
......@@ -1131,6 +1128,10 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
}
}
if (queryEnd) {
needStop = true;
}
_return:
if (NULL == ctx) {
......@@ -1196,7 +1197,8 @@ int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg) {
if (NULL == rsp) {
QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_FETCH);
} else {
qwBuildFetchRsp(rsp, &sOutput, dataLen);
bool qComplete = (DS_BUF_EMPTY == sOutput.bufStatus && sOutput.queryEnd);
qwBuildFetchRsp(rsp, &sOutput, dataLen, qComplete);
}
if ((!sOutput.queryEnd) && (DS_BUF_LOW == sOutput.bufStatus || DS_BUF_EMPTY == sOutput.bufStatus)) {
......@@ -1311,12 +1313,13 @@ _return:
QW_RET(code);
}
int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, putReqToQueryQFp fp) {
if (NULL == qWorkerMgmt || NULL == nodeObj || NULL == fp) {
int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj,
putReqToQueryQFp fp1, sendReqToDnodeFp fp2) {
if (NULL == qWorkerMgmt || NULL == nodeObj || NULL == fp1 || NULL == fp2) {
qError("invalid param to init qworker");
QW_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
SQWorkerMgmt *mgmt = calloc(1, sizeof(SQWorkerMgmt));
if (NULL == mgmt) {
qError("calloc %d failed", (int32_t)sizeof(SQWorkerMgmt));
......@@ -1359,7 +1362,8 @@ int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qW
mgmt->nodeType = nodeType;
mgmt->nodeId = nodeId;
mgmt->nodeObj = nodeObj;
mgmt->putToQueueFp = fp;
mgmt->putToQueueFp = fp1;
mgmt->sendReqFp = fp2;
*qWorkerMgmt = mgmt;
......
source/libs/qworker/src/qworkerMsg.c
浏览文件 @ 9db65e7a
......@@ -26,11 +26,11 @@ int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp) {
return TSDB_CODE_SUCCESS;
}
void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len) {
void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete) {
SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)msg;
rsp->useconds = htobe64(input->useconds);
rsp->completed = input->queryEnd;
rsp->completed = qComplete;
rsp->precision = input->precision;
rsp->compressed = input->compressed;
rsp->compLen = htonl(len);
......@@ -258,12 +258,12 @@ int32_t qwBuildAndSendCQueryMsg(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId,
int32_t code = (*mgmt->putToQueueFp)(mgmt->nodeObj, &pNewMsg);
if (TSDB_CODE_SUCCESS != code) {
QW_SCH_TASK_ELOG("put query continue msg to queue failed, code:%x", code);
QW_SCH_TASK_ELOG("put query continue msg to queue failed, vgId:%d, code:%s", mgmt->nodeId, tstrerror(code));
rpcFreeCont(req);
QW_ERR_RET(code);
}
QW_SCH_TASK_DLOG("put query continue msg to query queue, vgId:%d", mgmt->nodeId);
QW_SCH_TASK_DLOG("put task continue exec msg to query queue, vgId:%d", mgmt->nodeId);
return TSDB_CODE_SUCCESS;
}
......@@ -421,6 +421,11 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerProcessFetchRsp(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
qProcessFetchRsp(NULL, pMsg, NULL);
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
return TSDB_CODE_QRY_INVALID_INPUT;
......
source/libs/qworker/test/qworkerTests.cpp
浏览文件 @ 9db65e7a
......@@ -1081,7 +1081,7 @@ TEST(rcTest, shortExecshortDelay) {
qwtTestStop = false;
qwtTestQuitThreadNum = 0;
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL);
ASSERT_EQ(code, 0);
qwtTestMaxExecTaskUsec = 0;
......@@ -1162,7 +1162,7 @@ TEST(rcTest, longExecshortDelay) {
qwtTestStop = false;
qwtTestQuitThreadNum = 0;
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL);
ASSERT_EQ(code, 0);
qwtTestMaxExecTaskUsec = 1000000;
......@@ -1245,7 +1245,7 @@ TEST(rcTest, shortExeclongDelay) {
qwtTestStop = false;
qwtTestQuitThreadNum = 0;
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL);
ASSERT_EQ(code, 0);
qwtTestMaxExecTaskUsec = 0;
......
source/libs/transport/inc/transComm.h
浏览文件 @ 9db65e7a
......@@ -182,7 +182,7 @@ typedef struct {
#define TRANS_RESERVE_SIZE (sizeof(STranConnCtx))
#define TRANS_MSG_OVERHEAD (sizeof(STransMsgHead) + sizeof(STransDigestMsg))
#define TRANS_MSG_OVERHEAD (sizeof(STransMsgHead))
#define transHeadFromCont(cont) ((STransMsgHead*)((char*)cont - sizeof(STransMsgHead)))
#define transContFromHead(msg) (msg + sizeof(STransMsgHead))
#define transMsgLenFromCont(contLen) (contLen + sizeof(STransMsgHead))
......@@ -201,6 +201,9 @@ bool transDecompressMsg(char* msg, int32_t len, int32_t* flen);
void transConnCtxDestroy(STransConnCtx* ctx);
void transFreeMsg(void* msg);
//
typedef struct SConnBuffer {
char* buf;
int len;
......@@ -208,4 +211,9 @@ typedef struct SConnBuffer {
int left;
} SConnBuffer;
int transInitBuffer(SConnBuffer* buf);
int transClearBuffer(SConnBuffer* buf);
int transDestroyBuffer(SConnBuffer* buf);
int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf);
#endif
source/libs/transport/src/rpcMain.c
浏览文件 @ 9db65e7a
......@@ -242,11 +242,14 @@ void *rpcOpen(const SRpcInit *pInit) {
pRpc = (SRpcInfo *)calloc(1, sizeof(SRpcInfo));
if (pRpc == NULL) return NULL;
if (pInit->label) tstrncpy(pRpc->label, pInit->label, strlen(pInit->label));
if (pInit->label) tstrncpy(pRpc->label, pInit->label, tListLen(pInit->label));
pRpc->connType = pInit->connType;
if (pRpc->connType == TAOS_CONN_CLIENT) {
pRpc->numOfThreads = pInit->numOfThreads;
if (pRpc->numOfThreads >= 10) {
pRpc->numOfThreads = 10;
}
} else {
pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
}
......@@ -769,8 +772,8 @@ static SRpcConn *rpcAllocateServerConn(SRpcInfo *pRpc, SRecvInfo *pRecv) {
}
taosHashPut(pRpc->hash, hashstr, size, (char *)&pConn, POINTER_BYTES);
tDebug("%s %p server connection is allocated, uid:0x%x sid:%d key:%s spi:%d", pRpc->label, pConn, pConn->linkUid, sid,
hashstr, pConn->spi);
tDebug("%s %p server connection is allocated, uid:0x%x sid:%d key:%s spi:%d", pRpc->label, pConn, pConn->linkUid,
sid, hashstr, pConn->spi);
}
return pConn;
......@@ -1357,9 +1360,14 @@ static void rpcSendMsgToPeer(SRpcConn *pConn, void *msg, int msgLen) {
tDebug("%s, %s is sent to %s:%hu, len:%d sig:0x%08x:0x%08x:%d", pConn->info, TMSG_INFO(pHead->msgType),
pConn->peerFqdn, pConn->peerPort, msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
} else {
if (pHead->code == 0) pConn->secured = 1; // for success response, set link as secured
tDebug("%s, %s is sent to 0x%x:%hu, code:0x%x len:%d sig:0x%08x:0x%08x:%d", pConn->info, TMSG_INFO(pHead->msgType),
pConn->peerIp, pConn->peerPort, htonl(pHead->code), msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
if (pHead->code == 0) {
pConn->secured = 1; // for success response, set link as secured
}
char ipport[40] = {0};
taosIpPort2String(pConn->peerIp, pConn->peerPort, ipport);
tDebug("%s, %s is sent to %s, code:0x%x len:%d sig:0x%08x:0x%08x:%d", pConn->info, TMSG_INFO(pHead->msgType),
ipport, htonl(pHead->code), msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
}
// tTrace("connection type is: %d", pConn->connType);
......
source/libs/transport/src/trans.c
浏览文件 @ 9db65e7a
......@@ -30,7 +30,8 @@ void* rpcOpen(const SRpcInit* pInit) {
tstrncpy(pRpc->label, pInit->label, strlen(pInit->label));
}
pRpc->cfp = pInit->cfp;
pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
// pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
pRpc->numOfThreads = pInit->numOfThreads;
pRpc->connType = pInit->connType;
pRpc->idleTime = pInit->idleTime;
pRpc->tcphandle = (*taosInitHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);
......@@ -55,7 +56,13 @@ void* rpcMallocCont(int contLen) {
}
return start + sizeof(STransMsgHead);
}
void rpcFreeCont(void* cont) { return; }
void rpcFreeCont(void* cont) {
// impl
if (cont == NULL) {
return;
}
free((char*)cont - TRANS_MSG_OVERHEAD);
}
void* rpcReallocCont(void* ptr, int contLen) { return NULL; }
void rpcSendRedirectRsp(void* pConn, const SEpSet* pEpSet) {}
......
source/libs/transport/src/transCli.c
浏览文件 @ 9db65e7a
......@@ -31,6 +31,7 @@ typedef struct SCliConn {
char secured;
uint64_t expireTime;
int8_t notifyCount; // timers already notify to client
int32_t ref;
} SCliConn;
typedef struct SCliMsg {
......@@ -105,15 +106,28 @@ static void* clientThread(void* arg);
static void clientHandleResp(SCliConn* conn) {
STransConnCtx* pCtx = ((SCliMsg*)conn->data)->ctx;
SRpcInfo* pRpc = pCtx->pTransInst;
SRpcMsg rpcMsg;
rpcMsg.pCont = conn->readBuf.buf;
rpcMsg.contLen = conn->readBuf.len;
STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf);
pHead->code = htonl(pHead->code);
pHead->msgLen = htonl(pHead->msgLen);
SRpcMsg rpcMsg;
rpcMsg.contLen = transContLenFromMsg(pHead->msgLen);
rpcMsg.pCont = transContFromHead((char*)pHead);
rpcMsg.code = pHead->code;
rpcMsg.msgType = pHead->msgType;
rpcMsg.ahandle = pCtx->ahandle;
tDebug("conn %p handle resp", conn);
(pRpc->cfp)(NULL, &rpcMsg, NULL);
conn->notifyCount += 1;
SCliThrdObj* pThrd = conn->hostThrd;
tfree(conn->data);
// buf alread translated to rpcMsg.pCont
transClearBuffer(&conn->readBuf);
uv_read_start((uv_stream_t*)conn->stream, clientAllocBufferCb, clientReadCb);
addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn);
// start thread's timer of conn pool if not active
......@@ -123,16 +137,24 @@ static void clientHandleResp(SCliConn* conn) {
destroyTransConnCtx(pCtx);
}
static void clientHandleExcept(SCliConn* pConn) {
if (pConn->data == NULL) {
clientConnDestroy(pConn, true);
return;
}
tDebug("conn %p destroy", pConn);
SCliMsg* pMsg = pConn->data;
transFreeMsg((pMsg->msg.pCont));
pMsg->msg.pCont = NULL;
STransConnCtx* pCtx = pMsg->ctx;
SRpcInfo* pRpc = pCtx->pTransInst;
SRpcMsg rpcMsg;
SRpcMsg rpcMsg = {0};
rpcMsg.ahandle = pCtx->ahandle;
rpcMsg.code = -1;
// SRpcInfo* pRpc = pMsg->ctx->pRpc;
(pRpc->cfp)(NULL, &rpcMsg, NULL);
tfree(pConn->data);
pConn->notifyCount += 1;
destroyTransConnCtx(pCtx);
clientConnDestroy(pConn, true);
......@@ -201,12 +223,17 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
}
queue* h = QUEUE_HEAD(&plist->conn);
QUEUE_REMOVE(h);
return QUEUE_DATA(h, SCliConn, conn);
SCliConn* conn = QUEUE_DATA(h, SCliConn, conn);
QUEUE_INIT(&conn->conn);
return conn;
}
static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn) {
char key[128] = {0};
tstrncpy(key, ip, strlen(ip));
tstrncpy(key + strlen(key), (char*)(&port), sizeof(port));
tDebug("conn %p added to conn pool, read buf cap: %d", conn, conn->readBuf.cap);
SRpcInfo* pRpc = ((SCliThrdObj*)conn->hostThrd)->pTransInst;
conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
......@@ -225,40 +252,18 @@ static bool clientReadComplete(SConnBuffer* data) {
if (msgLen > data->len) {
data->left = msgLen - data->len;
return false;
} else {
} else if (msgLen == data->len) {
data->left = 0;
return true;
}
} else {
return false;
}
}
static void clientAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
// impl later
static const int CAPACITY = 512;
static void clientAllocBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
SCliConn* conn = handle->data;
SConnBuffer* pBuf = &conn->readBuf;
if (pBuf->cap == 0) {
pBuf->buf = (char*)calloc(1, CAPACITY * sizeof(char));
pBuf->len = 0;
pBuf->cap = CAPACITY;
pBuf->left = -1;
buf->base = pBuf->buf;
buf->len = CAPACITY;
} else {
if (pBuf->len >= pBuf->cap) {
if (pBuf->left == -1) {
pBuf->cap *= 2;
pBuf->buf = realloc(pBuf->buf, pBuf->cap);
} else if (pBuf->len + pBuf->left > pBuf->cap) {
pBuf->cap = pBuf->len + pBuf->left;
pBuf->buf = realloc(pBuf->buf, pBuf->cap);
}
}
buf->base = pBuf->buf + pBuf->len;
buf->len = pBuf->cap - pBuf->len;
}
transAllocBuffer(pBuf, buf);
}
static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
// impl later
......@@ -267,6 +272,7 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
if (nread > 0) {
pBuf->len += nread;
if (clientReadComplete(pBuf)) {
uv_read_stop((uv_stream_t*)conn->stream);
tDebug("conn %p read complete", conn);
clientHandleResp(conn);
} else {
......@@ -276,10 +282,9 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
}
assert(nread <= 0);
if (nread == 0) {
tError("conn %p closed", conn);
return;
}
if (nread < 0) {
if (nread < 0 || nread == UV_EOF) {
tError("conn %p read error: %s", conn, uv_err_name(nread));
clientHandleExcept(conn);
}
......@@ -288,39 +293,46 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
}
static void clientConnDestroy(SCliConn* conn, bool clear) {
tDebug("conn %p destroy", conn);
if (clear) {
uv_close((uv_handle_t*)conn->stream, NULL);
//
conn->ref--;
if (conn->ref == 0) {
tDebug("conn %p remove from conn pool", conn);
QUEUE_REMOVE(&conn->conn);
tDebug("conn %p remove from conn pool successfully", conn);
if (clear) {
uv_close((uv_handle_t*)conn->stream, clientDestroy);
}
}
free(conn->stream);
free(conn->readBuf.buf);
free(conn->writeReq);
free(conn);
}
static void clientDestroy(uv_handle_t* handle) {
SCliConn* conn = handle->data;
// QUEUE_REMOVE(&conn->conn);
clientConnDestroy(conn, false);
// transDestroyBuffer(&conn->readBuf);
free(conn->stream);
free(conn->writeReq);
tDebug("conn %p destroy successfully", conn);
free(conn);
// clientConnDestroy(conn, false);
}
static void clientWriteCb(uv_write_t* req, int status) {
SCliConn* pConn = req->data;
if (status == 0) {
tDebug("conn %p data already was written out", pConn);
SCliMsg* pMsg = pConn->data;
if (pMsg != NULL) {
transFreeMsg((pMsg->msg.pCont));
pMsg->msg.pCont = NULL;
}
} else {
tError("conn %p failed to write: %s", pConn, uv_err_name(status));
clientHandleExcept(pConn);
return;
}
SCliThrdObj* pThrd = pConn->hostThrd;
// if (pConn->stream == NULL) {
// pConn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
// uv_tcp_init(pThrd->loop, (uv_tcp_t*)pConn->stream);
// pConn->stream->data = pConn;
//}
uv_read_start((uv_stream_t*)pConn->stream, clientAllocReadBufferCb, clientReadCb);
// impl later
uv_read_start((uv_stream_t*)pConn->stream, clientAllocBufferCb, clientReadCb);
}
static void clientWrite(SCliConn* pConn) {
......@@ -365,14 +377,11 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
tDebug("conn %p get from conn pool", conn);
conn->data = pMsg;
conn->writeReq->data = conn;
conn->readBuf.len = 0;
memset(conn->readBuf.buf, 0, conn->readBuf.cap);
conn->readBuf.left = -1;
transDestroyBuffer(&conn->readBuf);
clientWrite(conn);
} else {
SCliConn* conn = calloc(1, sizeof(SCliConn));
conn->ref++;
// read/write stream handle
conn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream));
......@@ -381,6 +390,7 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
// write req handle
conn->writeReq = malloc(sizeof(uv_write_t));
conn->writeReq->data = conn;
QUEUE_INIT(&conn->conn);
conn->connReq.data = conn;
......
source/libs/transport/src/transComm.c
浏览文件 @ 9db65e7a
......@@ -191,4 +191,58 @@ void transConnCtxDestroy(STransConnCtx* ctx) {
free(ctx->ip);
free(ctx);
}
void transFreeMsg(void* msg) {
if (msg == NULL) {
return;
}
free((char*)msg - sizeof(STransMsgHead));
}
int transInitBuffer(SConnBuffer* buf) {
transClearBuffer(buf);
return 0;
}
int transClearBuffer(SConnBuffer* buf) {
memset(buf, 0, sizeof(*buf));
return 0;
}
int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf) {
/*
* formate of data buffer:
* |<--------------------------data from socket------------------------------->|
* |<------STransMsgHead------->|<-------------------other data--------------->|
*/
static const int CAPACITY = 1024;
SConnBuffer* p = connBuf;
if (p->cap == 0) {
p->buf = (char*)calloc(CAPACITY, sizeof(char));
p->len = 0;
p->cap = CAPACITY;
p->left = -1;
uvBuf->base = p->buf;
uvBuf->len = CAPACITY;
} else {
if (p->len >= p->cap) {
if (p->left == -1) {
p->cap *= 2;
p->buf = realloc(p->buf, p->cap);
} else if (p->len + p->left > p->cap) {
p->cap = p->len + p->left;
p->buf = realloc(p->buf, p->len + p->left);
}
}
uvBuf->base = p->buf + p->len;
uvBuf->len = p->cap - p->len;
}
return 0;
}
int transDestroyBuffer(SConnBuffer* buf) {
if (buf->cap > 0) {
tfree(buf->buf);
}
transClearBuffer(buf);
}
#endif
source/libs/transport/src/transSrv.c
浏览文件 @ 9db65e7a
......@@ -17,7 +17,7 @@
#include "transComm.h"
typedef struct SConn {
typedef struct SSrvConn {
uv_tcp_t* pTcp;
uv_write_t* pWriter;
uv_timer_t* pTimer;
......@@ -26,13 +26,14 @@ typedef struct SConn {
queue queue;
int ref;
int persist; // persist connection or not
SConnBuffer connBuf; // read buf,
SConnBuffer readBuf; // read buf,
int inType;
void* pTransInst; // rpc init
void* ahandle; //
void* hostThrd;
void* pSrvMsg;
SRpcMsg sendMsg;
// SRpcMsg sendMsg;
// del later
char secured;
int spi;
......@@ -40,7 +41,13 @@ typedef struct SConn {
char user[TSDB_UNI_LEN]; // user ID for the link
char secret[TSDB_PASSWORD_LEN];
char ckey[TSDB_PASSWORD_LEN]; // ciphering key
} SConn;
} SSrvConn;
typedef struct SSrvMsg {
SSrvConn* pConn;
SRpcMsg msg;
queue q;
} SSrvMsg;
typedef struct SWorkThrdObj {
pthread_t thread;
......@@ -48,8 +55,8 @@ typedef struct SWorkThrdObj {
int fd;
uv_loop_t* loop;
uv_async_t* workerAsync; //
queue conn;
pthread_mutex_t connMtx;
queue msg;
pthread_mutex_t msgMtx;
void* pTransInst;
} SWorkThrdObj;
......@@ -68,9 +75,9 @@ typedef struct SServerObj {
static const char* notify = "a";
// refactor later
static int transAddAuthPart(SConn* pConn, char* msg, int msgLen);
static int transAddAuthPart(SSrvConn* pConn, char* msg, int msgLen);
static int uvAuthMsg(SConn* pConn, char* msg, int msgLen);
static int uvAuthMsg(SSrvConn* pConn, char* msg, int msgLen);
static void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
static void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
......@@ -82,12 +89,13 @@ static void uvOnAcceptCb(uv_stream_t* stream, int status);
static void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf);
static void uvWorkerAsyncCb(uv_async_t* handle);
static void uvPrepareSendData(SConn* conn, uv_buf_t* wb);
static void uvPrepareSendData(SSrvMsg* msg, uv_buf_t* wb);
static void uvStartSendResp(SSrvMsg* msg);
static void destroySrvMsg(SSrvConn* conn);
// check whether already read complete packet
static bool readComplete(SConnBuffer* buf);
static SConn* createConn();
static void destroyConn(SConn* conn, bool clear /*clear handle or not*/);
static bool readComplete(SConnBuffer* buf);
static SSrvConn* createConn();
static void destroyConn(SSrvConn* conn, bool clear /*clear handle or not*/);
static void uvDestroyConn(uv_handle_t* handle);
......@@ -105,31 +113,9 @@ void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b
* |<--------------------------data from socket------------------------------->|
* |<------STransMsgHead------->|<-------------------other data--------------->|
*/
static const int CAPACITY = 1024;
SConn* conn = handle->data;
SConnBuffer* pBuf = &conn->connBuf;
if (pBuf->cap == 0) {
pBuf->buf = (char*)calloc(CAPACITY, sizeof(char));
pBuf->len = 0;
pBuf->cap = CAPACITY;
pBuf->left = -1;
buf->base = pBuf->buf;
buf->len = CAPACITY;
} else {
if (pBuf->len >= pBuf->cap) {
if (pBuf->left == -1) {
pBuf->cap *= 2;
pBuf->buf = realloc(pBuf->buf, pBuf->cap);
} else if (pBuf->len + pBuf->left > pBuf->cap) {
pBuf->cap = pBuf->len + pBuf->left;
pBuf->buf = realloc(pBuf->buf, pBuf->len + pBuf->left);
}
}
buf->base = pBuf->buf + pBuf->len;
buf->len = pBuf->cap - pBuf->len;
}
SSrvConn* conn = handle->data;
SConnBuffer* pBuf = &conn->readBuf;
transAllocBuffer(pBuf, buf);
}
// check data read from socket completely or not
......@@ -159,7 +145,7 @@ static bool readComplete(SConnBuffer* data) {
// // impl later
// STransMsgHead* pHead = (STransMsgHead*)pRecv->msg;
// SRpcInfo* pRpc = (SRpcInfo*)pRecv->shandle;
// SConn* pConn = pRecv->thandle;
// SSrvConn* pConn = pRecv->thandle;
// tDump(pRecv->msg, pRecv->msgLen);
// terrno = 0;
// // SRpcReqContext* pContest;
......@@ -167,7 +153,7 @@ static bool readComplete(SConnBuffer* data) {
// // do auth and check
//}
static int uvAuthMsg(SConn* pConn, char* msg, int len) {
static int uvAuthMsg(SSrvConn* pConn, char* msg, int len) {
STransMsgHead* pHead = (STransMsgHead*)msg;
int code = 0;
......@@ -222,14 +208,14 @@ static int uvAuthMsg(SConn* pConn, char* msg, int len) {
// refers specifically to query or insert timeout
static void uvHandleActivityTimeout(uv_timer_t* handle) {
SConn* conn = handle->data;
SSrvConn* conn = handle->data;
tDebug("%p timeout since no activity", conn);
}
static void uvHandleReq(SConn* pConn) {
static void uvHandleReq(SSrvConn* pConn) {
SRecvInfo info;
SRecvInfo* p = &info;
SConnBuffer* pBuf = &pConn->connBuf;
SConnBuffer* pBuf = &pConn->readBuf;
p->msg = pBuf->buf;
p->msgLen = pBuf->len;
p->ip = 0;
......@@ -255,7 +241,6 @@ static void uvHandleReq(SConn* pConn) {
pHead->code = htonl(pHead->code);
int32_t dlen = 0;
SRpcMsg rpcMsg;
if (transDecompressMsg(NULL, 0, NULL)) {
// add compress later
// pHead = rpcDecompressRpcMsg(pHead);
......@@ -264,6 +249,8 @@ static void uvHandleReq(SConn* pConn) {
// impl later
//
}
SRpcMsg rpcMsg;
rpcMsg.contLen = transContLenFromMsg(pHead->msgLen);
rpcMsg.pCont = pHead->content;
rpcMsg.msgType = pHead->msgType;
......@@ -271,6 +258,7 @@ static void uvHandleReq(SConn* pConn) {
rpcMsg.ahandle = NULL;
rpcMsg.handle = pConn;
transClearBuffer(&pConn->readBuf);
pConn->ref++;
(*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL);
// uv_timer_start(pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0);
......@@ -280,8 +268,8 @@ static void uvHandleReq(SConn* pConn) {
void uvOnReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
// opt
SConn* conn = cli->data;
SConnBuffer* pBuf = &conn->connBuf;
SSrvConn* conn = cli->data;
SConnBuffer* pBuf = &conn->readBuf;
if (nread > 0) {
pBuf->len += nread;
tDebug("conn %p read summroy, total read: %d, current read: %d", conn, pBuf->len, (int)nread);
......@@ -294,11 +282,12 @@ void uvOnReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
return;
}
if (nread == 0) {
tDebug("conn %p except read", conn);
// destroyConn(conn, true);
return;
}
if (nread != UV_EOF) {
if (nread < 0 || nread != UV_EOF) {
if (conn->ref > 1) {
conn->ref++; // ref > 1 signed that write is in progress
}
tDebug("conn %p read error: %s", conn, uv_err_name(nread));
destroyConn(conn, true);
}
......@@ -310,21 +299,21 @@ void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b
void uvOnTimeoutCb(uv_timer_t* handle) {
// opt
SConn* pConn = handle->data;
SSrvConn* pConn = handle->data;
tDebug("conn %p time out", pConn);
}
void uvOnWriteCb(uv_write_t* req, int status) {
SConn* conn = req->data;
SSrvConn* conn = req->data;
SSrvMsg* smsg = conn->pSrvMsg;
destroySrvMsg(conn);
SConnBuffer* buf = &conn->connBuf;
buf->len = 0;
memset(buf->buf, 0, buf->cap);
buf->left = -1;
transClearBuffer(&conn->readBuf);
if (status == 0) {
tDebug("conn %p data already was written on stream", conn);
} else {
tDebug("conn %p failed to write data, %s", conn, uv_err_name(status));
//
destroyConn(conn, true);
}
// opt
......@@ -337,16 +326,16 @@ static void uvOnPipeWriteCb(uv_write_t* req, int status) {
}
}
static void uvPrepareSendData(SConn* conn, uv_buf_t* wb) {
static void uvPrepareSendData(SSrvMsg* smsg, uv_buf_t* wb) {
// impl later;
tDebug("conn %p prepare to send resp", conn);
SRpcMsg* pMsg = &conn->sendMsg;
tDebug("conn %p prepare to send resp", smsg->pConn);
SRpcMsg* pMsg = &smsg->msg;
if (pMsg->pCont == 0) {
pMsg->pCont = (void*)rpcMallocCont(0);
pMsg->contLen = 0;
}
STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
pHead->msgType = conn->inType + 1;
pHead->msgType = smsg->pConn->inType + 1;
// add more info
char* msg = (char*)pHead;
int32_t len = transMsgLenFromCont(pMsg->contLen);
......@@ -357,28 +346,55 @@ static void uvPrepareSendData(SConn* conn, uv_buf_t* wb) {
wb->base = msg;
wb->len = len;
}
static void uvStartSendResp(SSrvMsg* smsg) {
// impl
uv_buf_t wb;
uvPrepareSendData(smsg, &wb);
SSrvConn* pConn = smsg->pConn;
uv_timer_stop(pConn->pTimer);
pConn->pSrvMsg = smsg;
// conn->pWriter->data = smsg;
uv_write(pConn->pWriter, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnWriteCb);
// SRpcMsg* rpcMsg = smsg->msg;
return;
}
static void destroySrvMsg(SSrvConn* conn) {
SSrvMsg* smsg = conn->pSrvMsg;
if (smsg == NULL) {
return;
}
transFreeMsg(smsg->msg.pCont);
free(conn->pSrvMsg);
conn->pSrvMsg = NULL;
}
void uvWorkerAsyncCb(uv_async_t* handle) {
SWorkThrdObj* pThrd = handle->data;
SConn* conn = NULL;
SSrvConn* conn = NULL;
queue wq;
// batch process to avoid to lock/unlock frequently
pthread_mutex_lock(&pThrd->connMtx);
QUEUE_MOVE(&pThrd->conn, &wq);
pthread_mutex_unlock(&pThrd->connMtx);
pthread_mutex_lock(&pThrd->msgMtx);
QUEUE_MOVE(&pThrd->msg, &wq);
pthread_mutex_unlock(&pThrd->msgMtx);
while (!QUEUE_IS_EMPTY(&wq)) {
queue* head = QUEUE_HEAD(&wq);
QUEUE_REMOVE(head);
SConn* conn = QUEUE_DATA(head, SConn, queue);
if (conn == NULL) {
tError("except occurred, do nothing");
return;
SSrvMsg* msg = QUEUE_DATA(head, SSrvMsg, q);
if (msg == NULL) {
tError("except occurred, continue");
continue;
}
uv_buf_t wb;
uvPrepareSendData(conn, &wb);
uv_timer_stop(conn->pTimer);
uvStartSendResp(msg);
// uv_buf_t wb;
// uvPrepareSendData(msg, &wb);
// uv_timer_stop(conn->pTimer);
uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb);
// uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb);
}
}
......@@ -431,7 +447,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
uv_handle_type pending = uv_pipe_pending_type(pipe);
assert(pending == UV_TCP);
SConn* pConn = createConn();
SSrvConn* pConn = createConn();
pConn->pTransInst = pThrd->pTransInst;
/* init conn timer*/
......@@ -480,8 +496,8 @@ static bool addHandleToWorkloop(void* arg) {
pThrd->pipe->data = pThrd;
QUEUE_INIT(&pThrd->conn);
pthread_mutex_init(&pThrd->connMtx, NULL);
QUEUE_INIT(&pThrd->msg);
pthread_mutex_init(&pThrd->msgMtx, NULL);
pThrd->workerAsync = malloc(sizeof(uv_async_t));
uv_async_init(pThrd->loop, pThrd->workerAsync, uvWorkerAsyncCb);
......@@ -519,34 +535,42 @@ void* workerThread(void* arg) {
uv_run(pThrd->loop, UV_RUN_DEFAULT);
}
static SConn* createConn() {
SConn* pConn = (SConn*)calloc(1, sizeof(SConn));
static SSrvConn* createConn() {
SSrvConn* pConn = (SSrvConn*)calloc(1, sizeof(SSrvConn));
tDebug("conn %p created", pConn);
++pConn->ref;
return pConn;
}
static void destroyConn(SConn* conn, bool clear) {
static void destroyConn(SSrvConn* conn, bool clear) {
if (conn == NULL) {
return;
}
if (--conn->ref == 0) {
// SRpcMsg* pMsg = &conn->sendMsg;
// transFreeMsg(pMsg->pCont);
// pMsg->pCont = NULL;
tDebug("conn %p try to destroy", conn);
if (--conn->ref > 0) {
return;
}
transDestroyBuffer(&conn->readBuf);
destroySrvMsg(conn);
if (clear) {
uv_close((uv_handle_t*)conn->pTcp, NULL);
uv_close((uv_handle_t*)conn->pTcp, uvDestroyConn);
}
}
static void uvDestroyConn(uv_handle_t* handle) {
SSrvConn* conn = handle->data;
tDebug("conn %p destroy", conn);
uv_timer_stop(conn->pTimer);
free(conn->pTimer);
free(conn->pTcp);
free(conn->connBuf.buf);
// free(conn->pTcp);
free(conn->pWriter);
free(conn);
}
static void uvDestroyConn(uv_handle_t* handle) {
SConn* conn = handle->data;
destroyConn(conn, false);
}
static int transAddAuthPart(SConn* pConn, char* msg, int msgLen) {
static int transAddAuthPart(SSrvConn* pConn, char* msg, int msgLen) {
STransMsgHead* pHead = (STransMsgHead*)msg;
if (pConn->spi && pConn->secured == 0) {
......@@ -628,6 +652,7 @@ void destroyWorkThrd(SWorkThrdObj* pThrd) {
pthread_join(pThrd->thread, NULL);
// free(srv->pipe[i]);
free(pThrd->loop);
pthread_mutex_destroy(&pThrd->msgMtx);
free(pThrd);
}
void taosCloseServer(void* arg) {
......@@ -644,17 +669,20 @@ void taosCloseServer(void* arg) {
}
void rpcSendResponse(const SRpcMsg* pMsg) {
SConn* pConn = pMsg->handle;
SSrvConn* pConn = pMsg->handle;
SWorkThrdObj* pThrd = pConn->hostThrd;
// opt later
pConn->sendMsg = *pMsg;
pthread_mutex_lock(&pThrd->connMtx);
QUEUE_PUSH(&pThrd->conn, &pConn->queue);
pthread_mutex_unlock(&pThrd->connMtx);
SSrvMsg* srvMsg = calloc(1, sizeof(SSrvMsg));
srvMsg->pConn = pConn;
srvMsg->msg = *pMsg;
pthread_mutex_lock(&pThrd->msgMtx);
QUEUE_PUSH(&pThrd->msg, &srvMsg->q);
pthread_mutex_unlock(&pThrd->msgMtx);
tDebug("conn %p start to send resp", pConn);
uv_async_send(pConn->pWorkerAsync);
uv_async_send(pThrd->workerAsync);
}
#endif
source/libs/transport/test/rclient.c
浏览文件 @ 9db65e7a
......@@ -64,6 +64,7 @@ static void *sendRequest(void *param) {
// tsem_wait(&pInfo->rspSem);
tsem_wait(&pInfo->rspSem);
tDebug("recv response succefully");
// usleep(100000000);
}
......
source/util/src/tqueue.c
浏览文件 @ 9db65e7a
......@@ -13,35 +13,36 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "taoserror.h"
#define _DEFAULT_SOURCE
#include "tqueue.h"
#include "taoserror.h"
#include "ulog.h"
typedef struct STaosQnode STaosQnode;
typedef struct STaosQnode {
STaosQnode *next;
STaosQueue *queue;
char item[];
} STaosQnode;
typedef struct STaosQueue {
int32_t itemSize;
int32_t numOfItems;
STaosQnode *head;
STaosQnode *tail;
STaosQueue *next; // for queue set
STaosQset *qset; // for queue set
void *ahandle; // for queue set
FProcessItem itemFp;
FProcessItems itemsFp;
int32_t threadId;
STaosQnode * head;
STaosQnode * tail;
STaosQueue * next; // for queue set
STaosQset * qset; // for queue set
void * ahandle; // for queue set
FItem itemFp;
FItems itemsFp;
pthread_mutex_t mutex;
} STaosQueue;
typedef struct STaosQset {
STaosQueue *head;
STaosQueue *current;
STaosQueue * head;
STaosQueue * current;
pthread_mutex_t mutex;
int32_t numOfQueues;
int32_t numOfItems;
......@@ -56,19 +57,23 @@ typedef struct STaosQall {
} STaosQall;
STaosQueue *taosOpenQueue() {
STaosQueue *queue = calloc(sizeof(STaosQueue), 1);
STaosQueue *queue = calloc(1, sizeof(STaosQueue));
if (queue == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pthread_mutex_init(&queue->mutex, NULL);
if (pthread_mutex_init(&queue->mutex, NULL) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
uTrace("queue:%p is opened", queue);
queue->threadId = -1;
uDebug("queue:%p is opened", queue);
return queue;
}
void taosSetQueueFp(STaosQueue *queue, FProcessItem itemFp, FProcessItems itemsFp) {
void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp) {
if (queue == NULL) return;
queue->itemFp = itemFp;
queue->itemsFp = itemsFp;
......@@ -77,7 +82,7 @@ void taosSetQueueFp(STaosQueue *queue, FProcessItem itemFp, FProcessItems itemsF
void taosCloseQueue(STaosQueue *queue) {
if (queue == NULL) return;
STaosQnode *pTemp;
STaosQset *qset;
STaosQset * qset;
pthread_mutex_lock(&queue->mutex);
STaosQnode *pNode = queue->head;
......@@ -85,7 +90,9 @@ void taosCloseQueue(STaosQueue *queue) {
qset = queue->qset;
pthread_mutex_unlock(&queue->mutex);
if (queue->qset) taosRemoveFromQset(qset, queue);
if (queue->qset) {
taosRemoveFromQset(qset, queue);
}
while (pNode) {
pTemp = pNode;
......@@ -96,7 +103,7 @@ void taosCloseQueue(STaosQueue *queue) {
pthread_mutex_destroy(&queue->mutex);
free(queue);
uTrace("queue:%p is closed", queue);
uDebug("queue:%p is closed", queue);
}
bool taosQueueEmpty(STaosQueue *queue) {
......@@ -120,19 +127,23 @@ int32_t taosQueueSize(STaosQueue *queue) {
}
void *taosAllocateQitem(int32_t size) {
STaosQnode *pNode = (STaosQnode *)calloc(sizeof(STaosQnode) + size, 1);
STaosQnode *pNode = calloc(1, sizeof(STaosQnode) + size);
if (pNode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
if (pNode == NULL) return NULL;
uTrace("item:%p, node:%p is allocated", pNode->item, pNode);
return (void *)pNode->item;
}
void taosFreeQitem(void *param) {
if (param == NULL) return;
void taosFreeQitem(void *pItem) {
if (pItem == NULL) return;
char *temp = (char *)param;
char *temp = pItem;
temp -= sizeof(STaosQnode);
uTrace("item:%p, node:%p is freed", param, temp);
uTrace("item:%p, node:%p is freed", pItem, temp);
free(temp);
}
......@@ -175,7 +186,7 @@ int32_t taosReadQitem(STaosQueue *queue, void **ppItem) {
queue->numOfItems--;
if (queue->qset) atomic_sub_fetch_32(&queue->qset->numOfItems, 1);
code = 1;
uDebug("item:%p is read out from queue:%p, items:%d", *ppItem, queue, queue->numOfItems);
uTrace("item:%p is read out from queue:%p, items:%d", *ppItem, queue, queue->numOfItems);
}
pthread_mutex_unlock(&queue->mutex);
......@@ -183,7 +194,7 @@ int32_t taosReadQitem(STaosQueue *queue, void **ppItem) {
return code;
}
STaosQall *taosAllocateQall() { return calloc(sizeof(STaosQall), 1); }
STaosQall *taosAllocateQall() { return calloc(1, sizeof(STaosQall)); }
void taosFreeQall(STaosQall *qall) { free(qall); }
......@@ -238,7 +249,7 @@ int32_t taosGetQitem(STaosQall *qall, void **ppItem) {
void taosResetQitems(STaosQall *qall) { qall->current = qall->start; }
STaosQset *taosOpenQset() {
STaosQset *qset = (STaosQset *)calloc(sizeof(STaosQset), 1);
STaosQset *qset = calloc(sizeof(STaosQset), 1);
if (qset == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
......@@ -247,7 +258,7 @@ STaosQset *taosOpenQset() {
pthread_mutex_init(&qset->mutex, NULL);
tsem_init(&qset->sem, 0, 0);
uTrace("qset:%p is opened", qset);
uDebug("qset:%p is opened", qset);
return qset;
}
......@@ -268,7 +279,7 @@ void taosCloseQset(STaosQset *qset) {
pthread_mutex_destroy(&qset->mutex);
tsem_destroy(&qset->sem);
free(qset);
uTrace("qset:%p is closed", qset);
uDebug("qset:%p is closed", qset);
}
// tsem_post 'qset->sem', so that reader threads waiting for it
......@@ -338,12 +349,12 @@ void taosRemoveFromQset(STaosQset *qset, STaosQueue *queue) {
pthread_mutex_unlock(&qset->mutex);
uTrace("queue:%p is removed from qset:%p", queue, qset);
uDebug("queue:%p is removed from qset:%p", queue, qset);
}
int32_t taosGetQueueNumber(STaosQset *qset) { return qset->numOfQueues; }
int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FProcessItem *itemFp) {
int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FItem *itemFp) {
STaosQnode *pNode = NULL;
int32_t code = 0;
......@@ -365,6 +376,7 @@ int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FP
*ppItem = pNode->item;
if (ahandle) *ahandle = queue->ahandle;
if (itemFp) *itemFp = queue->itemFp;
queue->head = pNode->next;
if (queue->head == NULL) queue->tail = NULL;
queue->numOfItems--;
......@@ -382,7 +394,7 @@ int32_t taosReadQitemFromQset(STaosQset *qset, void **ppItem, void **ahandle, FP
return code;
}
int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FProcessItems *itemsFp) {
int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahandle, FItems *itemsFp) {
STaosQueue *queue;
int32_t code = 0;
......@@ -411,7 +423,9 @@ int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahand
queue->tail = NULL;
queue->numOfItems = 0;
atomic_sub_fetch_32(&qset->numOfItems, qall->numOfItems);
for (int32_t j = 1; j < qall->numOfItems; ++j) tsem_wait(&qset->sem);
for (int32_t j = 1; j < qall->numOfItems; ++j) {
tsem_wait(&qset->sem);
}
}
pthread_mutex_unlock(&queue->mutex);
......@@ -423,6 +437,65 @@ int32_t taosReadAllQitemsFromQset(STaosQset *qset, STaosQall *qall, void **ahand
return code;
}
int32_t taosReadQitemFromQsetByThread(STaosQset *qset, void **ppItem, void **ahandle, FItem *itemFp, int32_t threadId) {
STaosQnode *pNode = NULL;
int32_t code = -1;
tsem_wait(&qset->sem);
pthread_mutex_lock(&qset->mutex);
for (int32_t i = 0; i < qset->numOfQueues; ++i) {
if (qset->current == NULL) qset->current = qset->head;
STaosQueue *queue = qset->current;
if (queue) qset->current = queue->next;
if (queue == NULL) break;
if (queue->head == NULL) continue;
if (queue->threadId != -1 && queue->threadId != threadId) {
code = 0;
continue;
}
pthread_mutex_lock(&queue->mutex);
if (queue->head) {
pNode = queue->head;
pNode->queue = queue;
queue->threadId = threadId;
*ppItem = pNode->item;
if (ahandle) *ahandle = queue->ahandle;
if (itemFp) *itemFp = queue->itemFp;
queue->head = pNode->next;
if (queue->head == NULL) queue->tail = NULL;
queue->numOfItems--;
atomic_sub_fetch_32(&qset->numOfItems, 1);
code = 1;
uTrace("item:%p is read out from queue:%p, items:%d", *ppItem, queue, queue->numOfItems);
}
pthread_mutex_unlock(&queue->mutex);
if (pNode) break;
}
pthread_mutex_unlock(&qset->mutex);
return code;
}
void taosResetQsetThread(STaosQset *qset, void *pItem) {
if (pItem == NULL) return;
STaosQnode *pNode = (STaosQnode *)((char *)pItem - sizeof(STaosQnode));
pthread_mutex_lock(&qset->mutex);
pNode->queue->threadId = -1;
for (int32_t i = 0; i < pNode->queue->numOfItems; ++i) {
tsem_post(&qset->sem);
}
pthread_mutex_unlock(&qset->mutex);
}
int32_t taosGetQueueItemsNumber(STaosQueue *queue) {
if (!queue) return 0;
......
source/util/src/tworker.c
浏览文件 @ 9db65e7a
......@@ -14,38 +14,46 @@
*/
#define _DEFAULT_SOURCE
#include "os.h"
#include "ulog.h"
#include "tqueue.h"
#include "tworker.h"
#include "taoserror.h"
#include "ulog.h"
typedef void* (*ThreadFp)(void *param);
typedef void *(*ThreadFp)(void *param);
int32_t tWorkerInit(SWorkerPool *pool) {
int32_t tQWorkerInit(SQWorkerPool *pool) {
pool->qset = taosOpenQset();
pool->workers = calloc(sizeof(SWorker), pool->max);
pthread_mutex_init(&pool->mutex, NULL);
for (int i = 0; i < pool->max; ++i) {
SWorker *worker = pool->workers + i;
pool->workers = calloc(sizeof(SQWorker), pool->max);
if (pool->workers == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
if (pthread_mutex_init(&pool->mutex, NULL)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
for (int32_t i = 0; i < pool->max; ++i) {
SQWorker *worker = pool->workers + i;
worker->id = i;
worker->pool = pool;
}
uInfo("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max);
uDebug("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max);
return 0;
}
void tWorkerCleanup(SWorkerPool *pool) {
for (int i = 0; i < pool->max; ++i) {
SWorker *worker = pool->workers + i;
void tQWorkerCleanup(SQWorkerPool *pool) {
for (int32_t i = 0; i < pool->max; ++i) {
SQWorker *worker = pool->workers + i;
if (worker == NULL) continue;
if (taosCheckPthreadValid(worker->thread)) {
taosQsetThreadResume(pool->qset);
}
}
for (int i = 0; i < pool->max; ++i) {
SWorker *worker = pool->workers + i;
for (int32_t i = 0; i < pool->max; ++i) {
SQWorker *worker = pool->workers + i;
if (worker == NULL) continue;
if (taosCheckPthreadValid(worker->thread)) {
pthread_join(worker->thread, NULL);
......@@ -56,15 +64,15 @@ void tWorkerCleanup(SWorkerPool *pool) {
taosCloseQset(pool->qset);
pthread_mutex_destroy(&pool->mutex);
uInfo("worker:%s is closed", pool->name);
uDebug("worker:%s is closed", pool->name);
}
static void *tWorkerThreadFp(SWorker *worker) {
SWorkerPool *pool = worker->pool;
FProcessItem fp = NULL;
static void *tQWorkerThreadFp(SQWorker *worker) {
SQWorkerPool *pool = worker->pool;
FItem fp = NULL;
void *msg = NULL;
void *ahandle = NULL;
void * msg = NULL;
void * ahandle = NULL;
int32_t code = 0;
taosBlockSIGPIPE();
......@@ -77,7 +85,7 @@ static void *tWorkerThreadFp(SWorker *worker) {
break;
}
if (fp) {
if (fp != NULL) {
(*fp)(ahandle, msg);
}
}
......@@ -85,11 +93,12 @@ static void *tWorkerThreadFp(SWorker *worker) {
return NULL;
}
STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp) {
STaosQueue *tWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp, ThreadFp threadFp) {
pthread_mutex_lock(&pool->mutex);
STaosQueue *queue = taosOpenQueue();
if (queue == NULL) {
pthread_mutex_unlock(&pool->mutex);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
......@@ -99,14 +108,18 @@ STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp)
// spawn a thread to process queue
if (pool->num < pool->max) {
do {
SWorker *worker = pool->workers + pool->num;
SQWorker *worker = pool->workers + pool->num;
pthread_attr_t thAttr;
pthread_attr_init(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&worker->thread, &thAttr, (ThreadFp)tWorkerThreadFp, worker) != 0) {
if (pthread_create(&worker->thread, &thAttr, threadFp, worker) != 0) {
uError("worker:%s:%d failed to create thread to process since %s", pool->name, worker->id, strerror(errno));
taosCloseQueue(queue);
terrno = TSDB_CODE_OUT_OF_MEMORY;
queue = NULL;
break;
}
pthread_attr_destroy(&thAttr);
......@@ -121,19 +134,73 @@ STaosQueue *tWorkerAllocQueue(SWorkerPool *pool, void *ahandle, FProcessItem fp)
return queue;
}
void tWorkerFreeQueue(SWorkerPool *pool, STaosQueue *queue) {
STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) {
return tWorkerAllocQueue(pool, ahandle, fp, (ThreadFp)tQWorkerThreadFp);
}
void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) {
taosCloseQueue(queue);
uDebug("worker:%s, queue:%p is freed", pool->name, queue);
}
int32_t tMWorkerInit(SMWorkerPool *pool) {
int32_t tFWorkerInit(SFWorkerPool *pool) { return tQWorkerInit((SQWorkerPool *)pool); }
void tFWorkerCleanup(SFWorkerPool *pool) { tQWorkerCleanup(pool); }
static void *tFWorkerThreadFp(SQWorker *worker) {
SQWorkerPool *pool = worker->pool;
FItem fp = NULL;
void * msg = NULL;
void * ahandle = NULL;
int32_t code = 0;
taosBlockSIGPIPE();
setThreadName(pool->name);
uDebug("worker:%s:%d is running", pool->name, worker->id);
while (1) {
code = taosReadQitemFromQsetByThread(pool->qset, (void **)&msg, &ahandle, &fp, worker->id);
if (code < 0) {
uDebug("worker:%s:%d qset:%p, got no message and exiting", pool->name, worker->id, pool->qset);
break;
} else if (code == 0) {
// uTrace("worker:%s:%d qset:%p, got no message and continue", pool->name, worker->id, pool->qset);
continue;
}
if (fp != NULL) {
(*fp)(ahandle, msg);
}
taosResetQsetThread(pool->qset, msg);
}
return NULL;
}
STaosQueue *tFWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) {
return tWorkerAllocQueue(pool, ahandle, fp, (ThreadFp)tFWorkerThreadFp);
}
void tFWorkerFreeQueue(SFWorkerPool *pool, STaosQueue *queue) { tQWorkerFreeQueue(pool, queue); }
int32_t tWWorkerInit(SWWorkerPool *pool) {
pool->nextId = 0;
pool->workers = calloc(sizeof(SMWorker), pool->max);
if (pool->workers == NULL) return -1;
pool->workers = calloc(sizeof(SWWorker), pool->max);
if (pool->workers == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
if (pthread_mutex_init(&pool->mutex, NULL) != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pthread_mutex_init(&pool->mutex, NULL);
for (int32_t i = 0; i < pool->max; ++i) {
SMWorker *worker = pool->workers + i;
SWWorker *worker = pool->workers + i;
worker->id = i;
worker->qall = NULL;
worker->qset = NULL;
......@@ -144,16 +211,18 @@ int32_t tMWorkerInit(SMWorkerPool *pool) {
return 0;
}
void tMWorkerCleanup(SMWorkerPool *pool) {
void tWWorkerCleanup(SWWorkerPool *pool) {
for (int32_t i = 0; i < pool->max; ++i) {
SMWorker *worker = pool->workers + i;
SWWorker *worker = pool->workers + i;
if (taosCheckPthreadValid(worker->thread)) {
if (worker->qset) taosQsetThreadResume(worker->qset);
if (worker->qset) {
taosQsetThreadResume(worker->qset);
}
}
}
for (int32_t i = 0; i < pool->max; ++i) {
SMWorker *worker = pool->workers + i;
SWWorker *worker = pool->workers + i;
if (taosCheckPthreadValid(worker->thread)) {
pthread_join(worker->thread, NULL);
taosFreeQall(worker->qall);
......@@ -167,12 +236,12 @@ void tMWorkerCleanup(SMWorkerPool *pool) {
uInfo("worker:%s is closed", pool->name);
}
static void *tWriteWorkerThreadFp(SMWorker *worker) {
SMWorkerPool *pool = worker->pool;
FProcessItems fp = NULL;
static void *tWWorkerThreadFp(SWWorker *worker) {
SWWorkerPool *pool = worker->pool;
FItems fp = NULL;
void *msg = NULL;
void *ahandle = NULL;
void * msg = NULL;
void * ahandle = NULL;
int32_t numOfMsgs = 0;
int32_t qtype = 0;
......@@ -187,7 +256,7 @@ static void *tWriteWorkerThreadFp(SMWorker *worker) {
break;
}
if (fp) {
if (fp != NULL) {
(*fp)(ahandle, worker->qall, numOfMsgs);
}
}
......@@ -195,13 +264,14 @@ static void *tWriteWorkerThreadFp(SMWorker *worker) {
return NULL;
}
STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems fp) {
STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp) {
pthread_mutex_lock(&pool->mutex);
SMWorker *worker = pool->workers + pool->nextId;
SWWorker *worker = pool->workers + pool->nextId;
STaosQueue *queue = taosOpenQueue();
if (queue == NULL) {
pthread_mutex_unlock(&pool->mutex);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
......@@ -221,17 +291,19 @@ STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems
taosCloseQset(worker->qset);
taosCloseQueue(queue);
pthread_mutex_unlock(&pool->mutex);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pthread_attr_t thAttr;
pthread_attr_init(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&worker->thread, &thAttr, (ThreadFp)tWriteWorkerThreadFp, worker) != 0) {
if (pthread_create(&worker->thread, &thAttr, (ThreadFp)tWWorkerThreadFp, worker) != 0) {
uError("worker:%s:%d failed to create thread to process since %s", pool->name, worker->id, strerror(errno));
taosFreeQall(worker->qall);
taosCloseQset(worker->qset);
taosCloseQueue(queue);
terrno = TSDB_CODE_OUT_OF_MEMORY;
queue = NULL;
} else {
uDebug("worker:%s:%d is launched, max:%d", pool->name, worker->id, pool->max);
......@@ -250,7 +322,7 @@ STaosQueue *tMWorkerAllocQueue(SMWorkerPool *pool, void *ahandle, FProcessItems
return queue;
}
void tMWorkerFreeQueue(SMWorkerPool *pool, STaosQueue *queue) {
void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) {
taosCloseQueue(queue);
uDebug("worker:%s, queue:%p is freed", pool->name, queue);
}
tests/script/tsim/dnode/basic1.sim
浏览文件 @ 9db65e7a
......@@ -178,9 +178,9 @@ if $rows != 3 then
endi
sql select * from st
#if $rows != 15 then
# return -1
#endi
if $rows != 15 then
return -1
endi
print =============== drop dnode
sql drop dnode 2;
......
tests/script/tsim/table/basic1.sim
浏览文件 @ 9db65e7a
......@@ -139,9 +139,9 @@ endi
print =============== query data frpm st
sql select * from st
#if $rows != 21 then
# return -1
#endi
if $rows != 21 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode1 -s start
......@@ -200,8 +200,8 @@ endi
print =============== query data frpm st
sql select * from st
#if $rows != 21 then
# return -1
#endi
if $rows != 21 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或