Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 d0e83605 编写于 作者: S Shengliang Guan
浏览文件
操作

Merge remote-tracking branch 'origin/3.0' into fix/tsim

上级 0ba9d225 dfaabf3d
显示空白变更内容
内联 并排
Showing with 3049 addition and 2528 deletion
include/common/tglobal.h
浏览文件 @ d0e83605
...@@ -64,6 +64,7 @@ extern int32_t tsNumOfMnodeQueryThreads; ...@@ -64,6 +64,7 @@ extern int32_t tsNumOfMnodeQueryThreads;
extern int32_t tsNumOfMnodeFetchThreads; extern int32_t tsNumOfMnodeFetchThreads;
extern int32_t tsNumOfMnodeReadThreads; extern int32_t tsNumOfMnodeReadThreads;
extern int32_t tsNumOfVnodeQueryThreads; extern int32_t tsNumOfVnodeQueryThreads;
extern int32_t tsNumOfVnodeStreamThreads;
extern int32_t tsNumOfVnodeFetchThreads; extern int32_t tsNumOfVnodeFetchThreads;
extern int32_t tsNumOfVnodeWriteThreads; extern int32_t tsNumOfVnodeWriteThreads;
extern int32_t tsNumOfVnodeSyncThreads; extern int32_t tsNumOfVnodeSyncThreads;
......
include/common/tmsgcb.h
浏览文件 @ d0e83605
...@@ -34,6 +34,7 @@ typedef enum { ...@@ -34,6 +34,7 @@ typedef enum {
WRITE_QUEUE, WRITE_QUEUE,
APPLY_QUEUE, APPLY_QUEUE,
SYNC_QUEUE, SYNC_QUEUE,
STREAM_QUEUE,
QUEUE_MAX, QUEUE_MAX,
} EQueueType; } EQueueType;
......
include/libs/executor/executor.h
浏览文件 @ d0e83605
...@@ -187,8 +187,11 @@ int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset); ...@@ -187,8 +187,11 @@ int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);
void* qStreamExtractMetaMsg(qTaskInfo_t tinfo); void* qStreamExtractMetaMsg(qTaskInfo_t tinfo);
void* qExtractReaderFromStreamScanner(void* scanner); void* qExtractReaderFromStreamScanner(void* scanner);
int32_t qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner); int32_t qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner);
int32_t qStreamInput(qTaskInfo_t tinfo, void* pItem);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif
......
include/libs/stream/tstream.h
浏览文件 @ d0e83605
...@@ -13,6 +13,7 @@ ...@@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "executor.h"
#include "os.h" #include "os.h"
#include "query.h" #include "query.h"
#include "tdatablock.h" #include "tdatablock.h"
...@@ -120,7 +121,6 @@ static FORCE_INLINE void* streamQueueCurItem(SStreamQueue* queue) { return queue ...@@ -120,7 +121,6 @@ static FORCE_INLINE void* streamQueueCurItem(SStreamQueue* queue) { return queue
static FORCE_INLINE void* streamQueueNextItem(SStreamQueue* queue) { static FORCE_INLINE void* streamQueueNextItem(SStreamQueue* queue) {
int8_t dequeueFlag = atomic_exchange_8(&queue->status, STREAM_QUEUE__PROCESSING); int8_t dequeueFlag = atomic_exchange_8(&queue->status, STREAM_QUEUE__PROCESSING);
if (dequeueFlag == STREAM_QUEUE__FAILED) { if (dequeueFlag == STREAM_QUEUE__FAILED) {
ASSERT(0);
ASSERT(queue->qItem != NULL); ASSERT(queue->qItem != NULL);
return streamQueueCurItem(queue); return streamQueueCurItem(queue);
} else { } else {
...@@ -307,14 +307,18 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem ...@@ -307,14 +307,18 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem
atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__FAILED); atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__FAILED);
return -1; return -1;
} }
qInfo("task %d %p submit enqueue %p %p %p", pTask->taskId, pTask, pItem, pSubmitClone, pSubmitClone->data); qDebug("task %d %p submit enqueue %p %p %p", pTask->taskId, pTask, pItem, pSubmitClone, pSubmitClone->data);
taosWriteQitem(pTask->inputQueue->queue, pSubmitClone); taosWriteQitem(pTask->inputQueue->queue, pSubmitClone);
// qStreamInput(pTask->exec.executor, pSubmitClone);
} else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) { } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
taosWriteQitem(pTask->inputQueue->queue, pItem); taosWriteQitem(pTask->inputQueue->queue, pItem);
// qStreamInput(pTask->exec.executor, pItem);
} else if (pItem->type == STREAM_INPUT__CHECKPOINT) { } else if (pItem->type == STREAM_INPUT__CHECKPOINT) {
taosWriteQitem(pTask->inputQueue->queue, pItem); taosWriteQitem(pTask->inputQueue->queue, pItem);
// qStreamInput(pTask->exec.executor, pItem);
} else if (pItem->type == STREAM_INPUT__TRIGGER) { } else if (pItem->type == STREAM_INPUT__TRIGGER) {
taosWriteQitem(pTask->inputQueue->queue, pItem); taosWriteQitem(pTask->inputQueue->queue, pItem);
// qStreamInput(pTask->exec.executor, pItem);
} }
if (pItem->type != STREAM_INPUT__TRIGGER && pItem->type != STREAM_INPUT__CHECKPOINT && pTask->triggerParam != 0) { if (pItem->type != STREAM_INPUT__TRIGGER && pItem->type != STREAM_INPUT__CHECKPOINT && pTask->triggerParam != 0) {
......
include/libs/sync/sync.h
浏览文件 @ d0e83605
...@@ -26,6 +26,8 @@ extern "C" { ...@@ -26,6 +26,8 @@ extern "C" {
extern bool gRaftDetailLog; extern bool gRaftDetailLog;
#define SYNC_RESP_TTL_MS 5000
#define SYNC_MAX_BATCH_SIZE 500 #define SYNC_MAX_BATCH_SIZE 500
#define SYNC_INDEX_BEGIN 0 #define SYNC_INDEX_BEGIN 0
#define SYNC_INDEX_INVALID -1 #define SYNC_INDEX_INVALID -1
......
source/common/src/tglobal.c
浏览文件 @ d0e83605
...@@ -55,6 +55,7 @@ int32_t tsNumOfMnodeQueryThreads = 2; ...@@ -55,6 +55,7 @@ int32_t tsNumOfMnodeQueryThreads = 2;
int32_t tsNumOfMnodeFetchThreads = 1; int32_t tsNumOfMnodeFetchThreads = 1;
int32_t tsNumOfMnodeReadThreads = 1; int32_t tsNumOfMnodeReadThreads = 1;
int32_t tsNumOfVnodeQueryThreads = 2; int32_t tsNumOfVnodeQueryThreads = 2;
int32_t tsNumOfVnodeStreamThreads = 2;
int32_t tsNumOfVnodeFetchThreads = 4; int32_t tsNumOfVnodeFetchThreads = 4;
int32_t tsNumOfVnodeWriteThreads = 2; int32_t tsNumOfVnodeWriteThreads = 2;
int32_t tsNumOfVnodeSyncThreads = 2; int32_t tsNumOfVnodeSyncThreads = 2;
...@@ -412,6 +413,10 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { ...@@ -412,6 +413,10 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
tsNumOfVnodeQueryThreads = TMAX(tsNumOfVnodeQueryThreads, 2); tsNumOfVnodeQueryThreads = TMAX(tsNumOfVnodeQueryThreads, 2);
if (cfgAddInt32(pCfg, "numOfVnodeQueryThreads", tsNumOfVnodeQueryThreads, 1, 1024, 0) != 0) return -1; if (cfgAddInt32(pCfg, "numOfVnodeQueryThreads", tsNumOfVnodeQueryThreads, 1, 1024, 0) != 0) return -1;
tsNumOfVnodeStreamThreads = tsNumOfCores / 4;
tsNumOfVnodeStreamThreads = TMAX(tsNumOfVnodeStreamThreads, 4);
if (cfgAddInt32(pCfg, "numOfVnodeStreamThreads", tsNumOfVnodeStreamThreads, 1, 1024, 0) != 0) return -1;
tsNumOfVnodeFetchThreads = tsNumOfCores / 4; tsNumOfVnodeFetchThreads = tsNumOfCores / 4;
tsNumOfVnodeFetchThreads = TMAX(tsNumOfVnodeFetchThreads, 4); tsNumOfVnodeFetchThreads = TMAX(tsNumOfVnodeFetchThreads, 4);
if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 1, 1024, 0) != 0) return -1; if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 1, 1024, 0) != 0) return -1;
...@@ -587,6 +592,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { ...@@ -587,6 +592,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsNumOfMnodeQueryThreads = cfgGetItem(pCfg, "numOfMnodeQueryThreads")->i32; tsNumOfMnodeQueryThreads = cfgGetItem(pCfg, "numOfMnodeQueryThreads")->i32;
tsNumOfMnodeReadThreads = cfgGetItem(pCfg, "numOfMnodeReadThreads")->i32; tsNumOfMnodeReadThreads = cfgGetItem(pCfg, "numOfMnodeReadThreads")->i32;
tsNumOfVnodeQueryThreads = cfgGetItem(pCfg, "numOfVnodeQueryThreads")->i32; tsNumOfVnodeQueryThreads = cfgGetItem(pCfg, "numOfVnodeQueryThreads")->i32;
tsNumOfVnodeStreamThreads = cfgGetItem(pCfg, "numOfVnodeStreamThreads")->i32;
tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32; tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32;
tsNumOfVnodeWriteThreads = cfgGetItem(pCfg, "numOfVnodeWriteThreads")->i32; tsNumOfVnodeWriteThreads = cfgGetItem(pCfg, "numOfVnodeWriteThreads")->i32;
tsNumOfVnodeSyncThreads = cfgGetItem(pCfg, "numOfVnodeSyncThreads")->i32; tsNumOfVnodeSyncThreads = cfgGetItem(pCfg, "numOfVnodeSyncThreads")->i32;
......
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
浏览文件 @ d0e83605
...@@ -31,6 +31,7 @@ typedef struct SVnodeMgmt { ...@@ -31,6 +31,7 @@ typedef struct SVnodeMgmt {
const char *path; const char *path;
const char *name; const char *name;
SQWorkerPool queryPool; SQWorkerPool queryPool;
SQWorkerPool streamPool;
SWWorkerPool fetchPool; SWWorkerPool fetchPool;
SWWorkerPool syncPool; SWWorkerPool syncPool;
SWWorkerPool writePool; SWWorkerPool writePool;
...@@ -61,6 +62,7 @@ typedef struct { ...@@ -61,6 +62,7 @@ typedef struct {
STaosQueue *pSyncQ; STaosQueue *pSyncQ;
STaosQueue *pApplyQ; STaosQueue *pApplyQ;
STaosQueue *pQueryQ; STaosQueue *pQueryQ;
STaosQueue *pStreamQ;
STaosQueue *pFetchQ; STaosQueue *pFetchQ;
} SVnodeObj; } SVnodeObj;
...@@ -105,6 +107,7 @@ int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); ...@@ -105,6 +107,7 @@ int32_t vmPutMsgToWriteQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutMsgToSyncQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToStreamQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t vmPutMsgToMonitorQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
......
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @ d0e83605
...@@ -357,16 +357,16 @@ SArray *vmGetMsgHandles() { ...@@ -357,16 +357,16 @@ SArray *vmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER_RSP, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
......
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
浏览文件 @ d0e83605
...@@ -92,6 +92,7 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { ...@@ -92,6 +92,7 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
while (!taosQueueEmpty(pVnode->pApplyQ)) taosMsleep(10); while (!taosQueueEmpty(pVnode->pApplyQ)) taosMsleep(10);
while (!taosQueueEmpty(pVnode->pQueryQ)) taosMsleep(10); while (!taosQueueEmpty(pVnode->pQueryQ)) taosMsleep(10);
while (!taosQueueEmpty(pVnode->pFetchQ)) taosMsleep(10); while (!taosQueueEmpty(pVnode->pFetchQ)) taosMsleep(10);
while (!taosQueueEmpty(pVnode->pStreamQ)) taosMsleep(10);
dTrace("vgId:%d, vnode queue is empty", pVnode->vgId); dTrace("vgId:%d, vnode queue is empty", pVnode->vgId);
vmFreeQueue(pMgmt, pVnode); vmFreeQueue(pMgmt, pVnode);
......
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
浏览文件 @ d0e83605
...@@ -81,6 +81,23 @@ static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { ...@@ -81,6 +81,23 @@ static void vmProcessQueryQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
} }
static void vmProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
SVnodeObj *pVnode = pInfo->ahandle;
const STraceId *trace = &pMsg->info.traceId;
dGTrace("vgId:%d, msg:%p get from vnode-stream queue", pVnode->vgId, pMsg);
int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);
if (code != 0) {
if (terrno != 0) code = terrno;
dGError("vgId:%d, msg:%p failed to stream since %s", pVnode->vgId, pMsg, terrstr());
vmSendRsp(pMsg, code);
}
dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);
rpcFreeCont(pMsg->pCont);
taosFreeQitem(pMsg);
}
static void vmProcessFetchQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { static void vmProcessFetchQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SVnodeObj *pVnode = pInfo->ahandle; SVnodeObj *pVnode = pInfo->ahandle;
SRpcMsg *pMsg = NULL; SRpcMsg *pMsg = NULL;
...@@ -140,6 +157,10 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp ...@@ -140,6 +157,10 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
dGTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg); dGTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg);
taosWriteQitem(pVnode->pQueryQ, pMsg); taosWriteQitem(pVnode->pQueryQ, pMsg);
break; break;
case STREAM_QUEUE:
dGTrace("vgId:%d, msg:%p put into vnode-stream queue", pVnode->vgId, pMsg);
taosWriteQitem(pVnode->pStreamQ, pMsg);
break;
case FETCH_QUEUE: case FETCH_QUEUE:
dGTrace("vgId:%d, msg:%p put into vnode-fetch queue", pVnode->vgId, pMsg); dGTrace("vgId:%d, msg:%p put into vnode-fetch queue", pVnode->vgId, pMsg);
taosWriteQitem(pVnode->pFetchQ, pMsg); taosWriteQitem(pVnode->pFetchQ, pMsg);
...@@ -174,6 +195,8 @@ int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsg ...@@ -174,6 +195,8 @@ int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsg
int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); } int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); }
int32_t vmPutMsgToStreamQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_QUEUE); }
int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
const STraceId *trace = &pMsg->info.traceId; const STraceId *trace = &pMsg->info.traceId;
dGTrace("msg:%p, put into vnode-mgmt queue", pMsg); dGTrace("msg:%p, put into vnode-mgmt queue", pMsg);
...@@ -234,6 +257,9 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { ...@@ -234,6 +257,9 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {
case FETCH_QUEUE: case FETCH_QUEUE:
size = taosQueueItemSize(pVnode->pFetchQ); size = taosQueueItemSize(pVnode->pFetchQ);
break; break;
case STREAM_QUEUE:
size = taosQueueItemSize(pVnode->pStreamQ);
break;
default: default:
break; break;
} }
...@@ -247,10 +273,11 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { ...@@ -247,10 +273,11 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
pVnode->pSyncQ = tWWorkerAllocQueue(&pMgmt->syncPool, pVnode, (FItems)vmProcessSyncQueue); pVnode->pSyncQ = tWWorkerAllocQueue(&pMgmt->syncPool, pVnode, (FItems)vmProcessSyncQueue);
pVnode->pApplyQ = tWWorkerAllocQueue(&pMgmt->applyPool, pVnode->pImpl, (FItems)vnodeApplyWriteMsg); pVnode->pApplyQ = tWWorkerAllocQueue(&pMgmt->applyPool, pVnode->pImpl, (FItems)vnodeApplyWriteMsg);
pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue); pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue);
pVnode->pStreamQ = tQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue);
pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue); pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue);
if (pVnode->pWriteQ == NULL || pVnode->pSyncQ == NULL || pVnode->pApplyQ == NULL || pVnode->pQueryQ == NULL || if (pVnode->pWriteQ == NULL || pVnode->pSyncQ == NULL || pVnode->pApplyQ == NULL || pVnode->pQueryQ == NULL ||
pVnode->pFetchQ == NULL) { pVnode->pStreamQ == NULL || pVnode->pFetchQ == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
...@@ -259,6 +286,7 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { ...@@ -259,6 +286,7 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
dDebug("vgId:%d, sync-queue:%p is alloced", pVnode->vgId, pVnode->pSyncQ); dDebug("vgId:%d, sync-queue:%p is alloced", pVnode->vgId, pVnode->pSyncQ);
dDebug("vgId:%d, apply-queue:%p is alloced", pVnode->vgId, pVnode->pApplyQ); dDebug("vgId:%d, apply-queue:%p is alloced", pVnode->vgId, pVnode->pApplyQ);
dDebug("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ); dDebug("vgId:%d, query-queue:%p is alloced", pVnode->vgId, pVnode->pQueryQ);
dDebug("vgId:%d, stream-queue:%p is alloced", pVnode->vgId, pVnode->pStreamQ);
dDebug("vgId:%d, fetch-queue:%p is alloced", pVnode->vgId, pVnode->pFetchQ); dDebug("vgId:%d, fetch-queue:%p is alloced", pVnode->vgId, pVnode->pFetchQ);
return 0; return 0;
} }
...@@ -268,11 +296,13 @@ void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { ...@@ -268,11 +296,13 @@ void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
tWWorkerFreeQueue(&pMgmt->applyPool, pVnode->pApplyQ); tWWorkerFreeQueue(&pMgmt->applyPool, pVnode->pApplyQ);
tWWorkerFreeQueue(&pMgmt->syncPool, pVnode->pSyncQ); tWWorkerFreeQueue(&pMgmt->syncPool, pVnode->pSyncQ);
tQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ); tQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ);
tQWorkerFreeQueue(&pMgmt->streamPool, pVnode->pStreamQ);
tWWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ); tWWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ);
pVnode->pWriteQ = NULL; pVnode->pWriteQ = NULL;
pVnode->pSyncQ = NULL; pVnode->pSyncQ = NULL;
pVnode->pApplyQ = NULL; pVnode->pApplyQ = NULL;
pVnode->pQueryQ = NULL; pVnode->pQueryQ = NULL;
pVnode->pStreamQ = NULL;
pVnode->pFetchQ = NULL; pVnode->pFetchQ = NULL;
dDebug("vgId:%d, queue is freed", pVnode->vgId); dDebug("vgId:%d, queue is freed", pVnode->vgId);
} }
...@@ -284,6 +314,12 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) { ...@@ -284,6 +314,12 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) {
pQPool->max = tsNumOfVnodeQueryThreads; pQPool->max = tsNumOfVnodeQueryThreads;
if (tQWorkerInit(pQPool) != 0) return -1; if (tQWorkerInit(pQPool) != 0) return -1;
SQWorkerPool *pStreamPool = &pMgmt->streamPool;
pStreamPool->name = "vnode-stream";
pStreamPool->min = tsNumOfVnodeStreamThreads;
pStreamPool->max = tsNumOfVnodeStreamThreads;
if (tQWorkerInit(pStreamPool) != 0) return -1;
SWWorkerPool *pFPool = &pMgmt->fetchPool; SWWorkerPool *pFPool = &pMgmt->fetchPool;
pFPool->name = "vnode-fetch"; pFPool->name = "vnode-fetch";
pFPool->max = tsNumOfVnodeFetchThreads; pFPool->max = tsNumOfVnodeFetchThreads;
...@@ -333,6 +369,7 @@ void vmStopWorker(SVnodeMgmt *pMgmt) { ...@@ -333,6 +369,7 @@ void vmStopWorker(SVnodeMgmt *pMgmt) {
tWWorkerCleanup(&pMgmt->applyPool); tWWorkerCleanup(&pMgmt->applyPool);
tWWorkerCleanup(&pMgmt->syncPool); tWWorkerCleanup(&pMgmt->syncPool);
tQWorkerCleanup(&pMgmt->queryPool); tQWorkerCleanup(&pMgmt->queryPool);
tQWorkerCleanup(&pMgmt->streamPool);
tWWorkerCleanup(&pMgmt->fetchPool); tWWorkerCleanup(&pMgmt->fetchPool);
dDebug("vnode workers are closed"); dDebug("vnode workers are closed");
} }
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @ d0e83605
...@@ -218,6 +218,7 @@ bool mndIsMnode(SMnode *pMnode, int32_t dnodeId) { ...@@ -218,6 +218,7 @@ bool mndIsMnode(SMnode *pMnode, int32_t dnodeId) {
} }
void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) { void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) {
#if 0
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int32_t totalMnodes = sdbGetSize(pSdb, SDB_MNODE); int32_t totalMnodes = sdbGetSize(pSdb, SDB_MNODE);
void *pIter = NULL; void *pIter = NULL;
...@@ -237,6 +238,9 @@ void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) { ...@@ -237,6 +238,9 @@ void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) {
addEpIntoEpSet(pEpSet, pObj->pDnode->fqdn, pObj->pDnode->port); addEpIntoEpSet(pEpSet, pObj->pDnode->fqdn, pObj->pDnode->port);
sdbRelease(pSdb, pObj); sdbRelease(pSdb, pObj);
} }
#else
syncGetRetryEpSet(pMnode->syncMgmt.sync, pEpSet);
#endif
} }
static int32_t mndSetCreateMnodeRedoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) { static int32_t mndSetCreateMnodeRedoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
......
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
浏览文件 @ d0e83605
...@@ -94,12 +94,15 @@ int32_t tsdbLastRowReaderOpen(void* pVnode, int32_t type, SArray* pTableIdList, ...@@ -94,12 +94,15 @@ int32_t tsdbLastRowReaderOpen(void* pVnode, int32_t type, SArray* pTableIdList,
int32_t tsdbLastrowReaderClose(void* pReader) { int32_t tsdbLastrowReaderClose(void* pReader) {
SLastrowReader* p = pReader; SLastrowReader* p = pReader;
if (p->pSchema != NULL) {
for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) { for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {
taosMemoryFreeClear(p->transferBuf[i]); taosMemoryFreeClear(p->transferBuf[i]);
} }
taosMemoryFree(p->pSchema);
taosMemoryFree(p->transferBuf); taosMemoryFree(p->transferBuf);
taosMemoryFree(p->pSchema);
}
taosMemoryFree(pReader); taosMemoryFree(pReader);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
......
source/dnode/vnode/src/vnd/vnodeSync.c
浏览文件 @ d0e83605
...@@ -427,6 +427,7 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c ...@@ -427,6 +427,7 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c
syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncGetVgId(pVnode->sync), pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state,
syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType)); syncUtilState2String(cbMeta.state), pMsg->msgType, TMSG_INFO(pMsg->msgType));
if (cbMeta.code == 0) {
SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen}; SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen); rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen); memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
...@@ -434,6 +435,14 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c ...@@ -434,6 +435,14 @@ static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta c
rpcMsg.info.conn.applyIndex = cbMeta.index; rpcMsg.info.conn.applyIndex = cbMeta.index;
rpcMsg.info.conn.applyTerm = cbMeta.term; rpcMsg.info.conn.applyTerm = cbMeta.term;
tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg); tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
} else {
SRpcMsg rsp = {.code = cbMeta.code, .info = pMsg->info};
vError("vgId:%d, sync commit error, msgtype:%d,%s, error:0x%X, errmsg:%s", syncGetVgId(pVnode->sync), pMsg->msgType,
TMSG_INFO(pMsg->msgType), cbMeta.code, tstrerror(cbMeta.code));
if (rsp.info.handle != NULL) {
tmsgSendRsp(&rsp);
}
}
} }
static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { static void vnodeSyncPreCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
......
source/libs/executor/CMakeLists.txt
浏览文件 @ d0e83605
...@@ -8,7 +8,7 @@ add_library(executor STATIC ${EXECUTOR_SRC}) ...@@ -8,7 +8,7 @@ add_library(executor STATIC ${EXECUTOR_SRC})
# ) # )
target_link_libraries(executor target_link_libraries(executor
PRIVATE os util common function parser planner qcom vnode scalar nodes index PRIVATE os util common function parser planner qcom vnode scalar nodes index stream
) )
target_include_directories( target_include_directories(
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ d0e83605
...@@ -39,6 +39,7 @@ extern "C" { ...@@ -39,6 +39,7 @@ extern "C" {
#include "tmsg.h" #include "tmsg.h"
#include "tpagedbuf.h" #include "tpagedbuf.h"
#include "tstreamUpdate.h" #include "tstreamUpdate.h"
#include "tstream.h"
#include "vnode.h" #include "vnode.h"
#include "executorInt.h" #include "executorInt.h"
...@@ -139,12 +140,14 @@ typedef struct STaskIdInfo { ...@@ -139,12 +140,14 @@ typedef struct STaskIdInfo {
} STaskIdInfo; } STaskIdInfo;
typedef struct { typedef struct {
//TODO remove prepareStatus
STqOffsetVal prepareStatus; // for tmq STqOffsetVal prepareStatus; // for tmq
STqOffsetVal lastStatus; // for tmq STqOffsetVal lastStatus; // for tmq
void* metaBlk; // for tmq fetching meta void* metaBlk; // for tmq fetching meta
SSDataBlock* pullOverBlk; // for streaming SSDataBlock* pullOverBlk; // for streaming
SWalFilterCond cond; SWalFilterCond cond;
int64_t lastScanUid; int64_t lastScanUid;
SStreamQueue* inputQueue;
} SStreamTaskInfo; } SStreamTaskInfo;
typedef struct SExecTaskInfo { typedef struct SExecTaskInfo {
......
source/libs/executor/src/executor.c
浏览文件 @ d0e83605
...@@ -60,8 +60,6 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu ...@@ -60,8 +60,6 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock); taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock);
taosArrayPush(pInfo->pBlockLists, &p); taosArrayPush(pInfo->pBlockLists, &p);
} }
/*} else if (type == STREAM_INPUT__TABLE_SCAN) {*/
/*ASSERT(pInfo->blockType == STREAM_INPUT__TABLE_SCAN);*/
} else { } else {
ASSERT(0); ASSERT(0);
} }
......
source/libs/executor/src/executorMain.c
浏览文件 @ d0e83605
...@@ -44,6 +44,13 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, ...@@ -44,6 +44,13 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
goto _error; goto _error;
} }
if (model == OPTR_EXEC_MODEL_STREAM) {
(*pTask)->streamInfo.inputQueue = streamQueueOpen();
if ((*pTask)->streamInfo.inputQueue == NULL) {
goto _error;
}
}
SDataSinkMgtCfg cfg = {.maxDataBlockNum = 1000, .maxDataBlockNumPerQuery = 100}; SDataSinkMgtCfg cfg = {.maxDataBlockNum = 1000, .maxDataBlockNumPerQuery = 100};
code = dsDataSinkMgtInit(&cfg); code = dsDataSinkMgtInit(&cfg);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
...@@ -252,6 +259,13 @@ int32_t qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner) { ...@@ -252,6 +259,13 @@ int32_t qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner) {
} }
} }
int32_t qStreamInput(qTaskInfo_t tinfo, void* pItem) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
ASSERT(pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM);
taosWriteQitem(pTaskInfo->streamInfo.inputQueue->queue, pItem);
return 0;
}
void* qExtractReaderFromStreamScanner(void* scanner) { void* qExtractReaderFromStreamScanner(void* scanner) {
SStreamScanInfo* pInfo = scanner; SStreamScanInfo* pInfo = scanner;
return (void*)pInfo->tqReader; return (void*)pInfo->tqReader;
......
source/libs/executor/src/scanoperator.c
浏览文件 @ d0e83605
...@@ -1202,15 +1202,6 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock ...@@ -1202,15 +1202,6 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
taosArrayDestroy(pBlock->pDataBlock); taosArrayDestroy(pBlock->pDataBlock);
ASSERT(pInfo->pRes->pDataBlock != NULL); ASSERT(pInfo->pRes->pDataBlock != NULL);
#if 0
if (pInfo->pRes->pDataBlock == NULL) {
// TODO add log
updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
pOperator->status = OP_EXEC_DONE;
pTaskInfo->code = terrno;
return -1;
}
#endif
// currently only the tbname pseudo column // currently only the tbname pseudo column
if (pInfo->numOfPseudoExpr > 0) { if (pInfo->numOfPseudoExpr > 0) {
...@@ -1231,11 +1222,6 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { ...@@ -1231,11 +1222,6 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SStreamScanInfo* pInfo = pOperator->info; SStreamScanInfo* pInfo = pOperator->info;
/*pTaskInfo->code = pOperator->fpSet._openFn(pOperator);*/
/*if (pTaskInfo->code != TSDB_CODE_SUCCESS || pOperator->status == OP_EXEC_DONE) {*/
/*return NULL;*/
/*}*/
qDebug("stream scan called"); qDebug("stream scan called");
if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__LOG) { if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__LOG) {
while (1) { while (1) {
...@@ -1425,15 +1411,6 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { ...@@ -1425,15 +1411,6 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
qDebug("scan rows: %d", pBlockInfo->rows); qDebug("scan rows: %d", pBlockInfo->rows);
return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes; return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
#if 0
} else if (pInfo->blockType == STREAM_INPUT__TABLE_SCAN) {
ASSERT(0);
// check reader last status
// if not match, reset status
SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
return pResult && pResult->info.rows > 0 ? pResult : NULL;
#endif
} else { } else {
ASSERT(0); ASSERT(0);
return NULL; return NULL;
......
source/libs/executor/src/timewindowoperator.c
浏览文件 @ d0e83605
...@@ -1341,9 +1341,9 @@ static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup, ...@@ -1341,9 +1341,9 @@ static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup,
uint64_t groupId = *(uint64_t*)key; uint64_t groupId = *(uint64_t*)key;
ASSERT(keyLen == GET_RES_WINDOW_KEY_LEN(sizeof(TSKEY))); ASSERT(keyLen == GET_RES_WINDOW_KEY_LEN(sizeof(TSKEY)));
TSKEY ts = *(int64_t*)((char*)key + sizeof(uint64_t)); TSKEY ts = *(int64_t*)((char*)key + sizeof(uint64_t));
SResultRowInfo dumyInfo; STimeWindow win;
dumyInfo.cur.pageId = -1; win.skey = ts;
STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, ts, pInterval, TSDB_ORDER_ASC); win.ekey = taosTimeAdd(win.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
SWinRes winRe = { SWinRes winRe = {
.ts = win.skey, .ts = win.skey,
.groupId = groupId, .groupId = groupId,
......
source/libs/stream/inc/streamInc.h
浏览文件 @ d0e83605
...@@ -42,6 +42,9 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) ...@@ -42,6 +42,9 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq); int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq);
int32_t streamAppendQueueItem(SStreamQueueItem* dst, SStreamQueueItem* elem);
void streamFreeQitem(SStreamQueueItem* data);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif
......
source/libs/stream/src/stream.c
浏览文件 @ d0e83605
...@@ -97,7 +97,7 @@ int32_t streamLaunchByWrite(SStreamTask* pTask, int32_t vgId) { ...@@ -97,7 +97,7 @@ int32_t streamLaunchByWrite(SStreamTask* pTask, int32_t vgId) {
.pCont = pRunReq, .pCont = pRunReq,
.contLen = sizeof(SStreamTaskRunReq), .contLen = sizeof(SStreamTaskRunReq),
}; };
tmsgPutToQueue(pTask->pMsgCb, FETCH_QUEUE, &msg); tmsgPutToQueue(pTask->pMsgCb, STREAM_QUEUE, &msg);
} }
return 0; return 0;
} }
......
source/libs/stream/src/streamData.c
浏览文件 @ d0e83605
...@@ -97,3 +97,29 @@ void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit) { ...@@ -97,3 +97,29 @@ void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit) {
taosMemoryFree(pDataSubmit->dataRef); taosMemoryFree(pDataSubmit->dataRef);
} }
} }
int32_t streamAppendQueueItem(SStreamQueueItem* dst, SStreamQueueItem* elem) {
ASSERT(elem);
if (dst->type == elem->type && dst->type == STREAM_INPUT__DATA_BLOCK) {
SStreamDataBlock* pBlock = (SStreamDataBlock*)dst;
SStreamDataBlock* pBlockSrc = (SStreamDataBlock*)elem;
taosArrayAddAll(pBlock->blocks, pBlockSrc->blocks);
return 0;
} else {
return -1;
}
}
void streamFreeQitem(SStreamQueueItem* data) {
int8_t type = data->type;
if (type == STREAM_INPUT__TRIGGER) {
blockDataDestroy(((SStreamTrigger*)data)->pBlock);
taosFreeQitem(data);
} else if (type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__DATA_RETRIEVE) {
taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)tDeleteSSDataBlock);
taosFreeQitem(data);
} else if (type == STREAM_INPUT__DATA_SUBMIT) {
streamDataSubmitRefDec((SStreamDataSubmit*)data);
taosFreeQitem(data);
}
}
source/libs/stream/src/streamDispatch.c
浏览文件 @ d0e83605
...@@ -251,7 +251,7 @@ int32_t streamBuildDispatchMsg(SStreamTask* pTask, const SStreamDataBlock* data, ...@@ -251,7 +251,7 @@ int32_t streamBuildDispatchMsg(SStreamTask* pTask, const SStreamDataBlock* data,
ASSERT(vgId > 0 || vgId == SNODE_HANDLE); ASSERT(vgId > 0 || vgId == SNODE_HANDLE);
req.taskId = downstreamTaskId; req.taskId = downstreamTaskId;
qInfo("dispatch from task %d (child id %d) to down stream task %d in vnode %d", pTask->taskId, pTask->selfChildId, qDebug("dispatch from task %d (child id %d) to down stream task %d in vnode %d", pTask->taskId, pTask->selfChildId,
downstreamTaskId, vgId); downstreamTaskId, vgId);
// serialize // serialize
...@@ -298,6 +298,7 @@ int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb) { ...@@ -298,6 +298,7 @@ int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb) {
SStreamDataBlock* pBlock = streamQueueNextItem(pTask->outputQueue); SStreamDataBlock* pBlock = streamQueueNextItem(pTask->outputQueue);
if (pBlock == NULL) { if (pBlock == NULL) {
qDebug("stream stop dispatching since no output: task %d", pTask->taskId);
atomic_store_8(&pTask->outputStatus, TASK_OUTPUT_STATUS__NORMAL); atomic_store_8(&pTask->outputStatus, TASK_OUTPUT_STATUS__NORMAL);
return 0; return 0;
} }
......
source/libs/stream/src/streamExec.c
浏览文件 @ d0e83605
...@@ -75,10 +75,35 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) ...@@ -75,10 +75,35 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
static SArray* streamExecForQall(SStreamTask* pTask, SArray* pRes) { static SArray* streamExecForQall(SStreamTask* pTask, SArray* pRes) {
while (1) { while (1) {
void* data = streamQueueNextItem(pTask->inputQueue); int32_t cnt = 0;
void* data = NULL;
while (1) {
SStreamQueueItem* qItem = streamQueueNextItem(pTask->inputQueue);
if (qItem == NULL) {
qDebug("stream exec over, queue empty");
break;
}
if (data == NULL) {
data = qItem;
streamQueueProcessSuccess(pTask->inputQueue);
continue;
} else {
if (streamAppendQueueItem(data, qItem) < 0) {
streamQueueProcessFail(pTask->inputQueue);
break;
} else {
cnt++;
streamQueueProcessSuccess(pTask->inputQueue);
taosArrayDestroy(((SStreamDataBlock*)qItem)->blocks);
taosFreeQitem(qItem);
}
}
}
if (data == NULL) break; if (data == NULL) break;
qDebug("stream task %d exec begin, batch msg: %d", pTask->taskId, cnt);
streamTaskExecImpl(pTask, data, pRes); streamTaskExecImpl(pTask, data, pRes);
qDebug("stream task %d exec end", pTask->taskId);
if (pTask->taskStatus == TASK_STATUS__DROPPING) { if (pTask->taskStatus == TASK_STATUS__DROPPING) {
taosArrayDestroyEx(pRes, (FDelete)tDeleteSSDataBlock); taosArrayDestroyEx(pRes, (FDelete)tDeleteSSDataBlock);
...@@ -95,27 +120,16 @@ static SArray* streamExecForQall(SStreamTask* pTask, SArray* pRes) { ...@@ -95,27 +120,16 @@ static SArray* streamExecForQall(SStreamTask* pTask, SArray* pRes) {
qRes->type = STREAM_INPUT__DATA_BLOCK; qRes->type = STREAM_INPUT__DATA_BLOCK;
qRes->blocks = pRes; qRes->blocks = pRes;
if (streamTaskOutput(pTask, qRes) < 0) { if (streamTaskOutput(pTask, qRes) < 0) {
streamQueueProcessFail(pTask->inputQueue); /*streamQueueProcessFail(pTask->inputQueue);*/
taosArrayDestroyEx(pRes, (FDelete)tDeleteSSDataBlock); taosArrayDestroyEx(pRes, (FDelete)tDeleteSSDataBlock);
taosFreeQitem(qRes); taosFreeQitem(qRes);
return NULL; return NULL;
} }
streamQueueProcessSuccess(pTask->inputQueue); /*streamQueueProcessSuccess(pTask->inputQueue);*/
pRes = taosArrayInit(0, sizeof(SSDataBlock)); pRes = taosArrayInit(0, sizeof(SSDataBlock));
} }
int8_t type = ((SStreamQueueItem*)data)->type; streamFreeQitem(data);
if (type == STREAM_INPUT__TRIGGER) {
blockDataDestroy(((SStreamTrigger*)data)->pBlock);
taosFreeQitem(data);
} else if (type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__DATA_RETRIEVE) {
taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)tDeleteSSDataBlock);
taosFreeQitem(data);
} else if (type == STREAM_INPUT__DATA_SUBMIT) {
ASSERT(pTask->isDataScan);
streamDataSubmitRefDec((SStreamDataSubmit*)data);
taosFreeQitem(data);
}
} }
return pRes; return pRes;
} }
...@@ -129,6 +143,7 @@ int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) { ...@@ -129,6 +143,7 @@ int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) {
atomic_val_compare_exchange_8(&pTask->execStatus, TASK_EXEC_STATUS__IDLE, TASK_EXEC_STATUS__EXECUTING); atomic_val_compare_exchange_8(&pTask->execStatus, TASK_EXEC_STATUS__IDLE, TASK_EXEC_STATUS__EXECUTING);
if (execStatus == TASK_EXEC_STATUS__IDLE) { if (execStatus == TASK_EXEC_STATUS__IDLE) {
// first run // first run
qDebug("stream exec, enter exec status");
pRes = streamExecForQall(pTask, pRes); pRes = streamExecForQall(pTask, pRes);
if (pRes == NULL) goto FAIL; if (pRes == NULL) goto FAIL;
...@@ -136,11 +151,13 @@ int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) { ...@@ -136,11 +151,13 @@ int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) {
atomic_store_8(&pTask->execStatus, TASK_EXEC_STATUS__CLOSING); atomic_store_8(&pTask->execStatus, TASK_EXEC_STATUS__CLOSING);
// second run, make sure inputQ and qall are cleared // second run, make sure inputQ and qall are cleared
qDebug("stream exec, enter closing status");
pRes = streamExecForQall(pTask, pRes); pRes = streamExecForQall(pTask, pRes);
if (pRes == NULL) goto FAIL; if (pRes == NULL) goto FAIL;
taosArrayDestroyEx(pRes, (FDelete)tDeleteSSDataBlock); taosArrayDestroyEx(pRes, (FDelete)tDeleteSSDataBlock);
atomic_store_8(&pTask->execStatus, TASK_EXEC_STATUS__IDLE); atomic_store_8(&pTask->execStatus, TASK_EXEC_STATUS__IDLE);
qDebug("stream exec, return result");
return 0; return 0;
} else if (execStatus == TASK_EXEC_STATUS__CLOSING) { } else if (execStatus == TASK_EXEC_STATUS__CLOSING) {
continue; continue;
......
source/libs/sync/inc/syncRaftEntry.h
浏览文件 @ d0e83605
...@@ -26,6 +26,7 @@ extern "C" { ...@@ -26,6 +26,7 @@ extern "C" {
#include "syncInt.h" #include "syncInt.h"
#include "syncMessage.h" #include "syncMessage.h"
#include "taosdef.h" #include "taosdef.h"
#include "tskiplist.h"
typedef struct SSyncRaftEntry { typedef struct SSyncRaftEntry {
uint32_t bytes; uint32_t bytes;
...@@ -58,29 +59,52 @@ void syncEntryLog(const SSyncRaftEntry* pObj); ...@@ -58,29 +59,52 @@ void syncEntryLog(const SSyncRaftEntry* pObj);
void syncEntryLog2(char* s, const SSyncRaftEntry* pObj); void syncEntryLog2(char* s, const SSyncRaftEntry* pObj);
//----------------------------------- //-----------------------------------
typedef struct SRaftEntryCache { typedef struct SRaftEntryHashCache {
SHashObj* pEntryHash; SHashObj* pEntryHash;
int32_t maxCount; int32_t maxCount;
int32_t currentCount; int32_t currentCount;
TdThreadMutex mutex; TdThreadMutex mutex;
SSyncNode* pSyncNode; SSyncNode* pSyncNode;
} SRaftEntryHashCache;
SRaftEntryHashCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount);
void raftCacheDestroy(SRaftEntryHashCache* pCache);
int32_t raftCachePutEntry(struct SRaftEntryHashCache* pCache, SSyncRaftEntry* pEntry);
int32_t raftCacheGetEntry(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheGetEntryP(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheDelEntry(struct SRaftEntryHashCache* pCache, SyncIndex index);
int32_t raftCacheGetAndDel(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheClear(struct SRaftEntryHashCache* pCache);
cJSON* raftCache2Json(SRaftEntryHashCache* pObj);
char* raftCache2Str(SRaftEntryHashCache* pObj);
void raftCachePrint(SRaftEntryHashCache* pObj);
void raftCachePrint2(char* s, SRaftEntryHashCache* pObj);
void raftCacheLog(SRaftEntryHashCache* pObj);
void raftCacheLog2(char* s, SRaftEntryHashCache* pObj);
//-----------------------------------
typedef struct SRaftEntryCache {
SSkipList* pSkipList;
int32_t maxCount;
int32_t currentCount;
TdThreadMutex mutex;
SSyncNode* pSyncNode;
} SRaftEntryCache; } SRaftEntryCache;
SRaftEntryCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount); SRaftEntryCache* raftEntryCacheCreate(SSyncNode* pSyncNode, int32_t maxCount);
void raftCacheDestroy(SRaftEntryCache* pCache); void raftEntryCacheDestroy(SRaftEntryCache* pCache);
int32_t raftCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry); int32_t raftEntryCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry);
int32_t raftCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry); int32_t raftEntryCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry); int32_t raftEntryCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheDelEntry(struct SRaftEntryCache* pCache, SyncIndex index); int32_t raftEntryCacheClear(struct SRaftEntryCache* pCache, int32_t count);
int32_t raftCacheGetAndDel(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheClear(struct SRaftEntryCache* pCache);
cJSON* raftCache2Json(SRaftEntryCache* pObj); cJSON* raftEntryCache2Json(SRaftEntryCache* pObj);
char* raftCache2Str(SRaftEntryCache* pObj); char* raftEntryCache2Str(SRaftEntryCache* pObj);
void raftCachePrint(SRaftEntryCache* pObj); void raftEntryCachePrint(SRaftEntryCache* pObj);
void raftCachePrint2(char* s, SRaftEntryCache* pObj); void raftEntryCachePrint2(char* s, SRaftEntryCache* pObj);
void raftCacheLog(SRaftEntryCache* pObj); void raftEntryCacheLog(SRaftEntryCache* pObj);
void raftCacheLog2(char* s, SRaftEntryCache* pObj); void raftEntryCacheLog2(char* s, SRaftEntryCache* pObj);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/libs/sync/src/syncMain.c
浏览文件 @ d0e83605
...@@ -1055,19 +1055,12 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) { ...@@ -1055,19 +1055,12 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) {
} }
// tools // tools
pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, 0); pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS);
ASSERT(pSyncNode->pSyncRespMgr != NULL); ASSERT(pSyncNode->pSyncRespMgr != NULL);
// restore state // restore state
pSyncNode->restoreFinish = false; pSyncNode->restoreFinish = false;
// pSyncNode->pSnapshot = NULL;
// if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
// pSyncNode->pSnapshot = taosMemoryMalloc(sizeof(SSnapshot));
// pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, pSyncNode->pSnapshot);
// }
// tsem_init(&(pSyncNode->restoreSem), 0, 0);
// snapshot senders // snapshot senders
for (int i = 0; i < TSDB_MAX_REPLICA; ++i) { for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, i); SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, i);
......
source/libs/sync/src/syncRaftEntry.c
浏览文件 @ d0e83605
...@@ -198,8 +198,8 @@ void syncEntryLog2(char* s, const SSyncRaftEntry* pObj) { ...@@ -198,8 +198,8 @@ void syncEntryLog2(char* s, const SSyncRaftEntry* pObj) {
} }
//----------------------------------- //-----------------------------------
SRaftEntryCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) { SRaftEntryHashCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) {
SRaftEntryCache* pCache = taosMemoryMalloc(sizeof(SRaftEntryCache)); SRaftEntryHashCache* pCache = taosMemoryMalloc(sizeof(SRaftEntryHashCache));
if (pCache == NULL) { if (pCache == NULL) {
sError("vgId:%d raft cache create error", pSyncNode->vgId); sError("vgId:%d raft cache create error", pSyncNode->vgId);
return NULL; return NULL;
...@@ -220,7 +220,7 @@ SRaftEntryCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) { ...@@ -220,7 +220,7 @@ SRaftEntryCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) {
return pCache; return pCache;
} }
void raftCacheDestroy(SRaftEntryCache* pCache) { void raftCacheDestroy(SRaftEntryHashCache* pCache) {
if (pCache != NULL) { if (pCache != NULL) {
taosThreadMutexLock(&(pCache->mutex)); taosThreadMutexLock(&(pCache->mutex));
taosHashCleanup(pCache->pEntryHash); taosHashCleanup(pCache->pEntryHash);
...@@ -233,7 +233,7 @@ void raftCacheDestroy(SRaftEntryCache* pCache) { ...@@ -233,7 +233,7 @@ void raftCacheDestroy(SRaftEntryCache* pCache) {
// success, return 1 // success, return 1
// max count, return 0 // max count, return 0
// error, return -1 // error, return -1
int32_t raftCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry) { int32_t raftCachePutEntry(struct SRaftEntryHashCache* pCache, SSyncRaftEntry* pEntry) {
taosThreadMutexLock(&(pCache->mutex)); taosThreadMutexLock(&(pCache->mutex));
if (pCache->currentCount >= pCache->maxCount) { if (pCache->currentCount >= pCache->maxCount) {
...@@ -259,7 +259,7 @@ int32_t raftCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry ...@@ -259,7 +259,7 @@ int32_t raftCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry
// success, return 0 // success, return 0
// error, return -1 // error, return -1
// not exist, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST // not exist, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
int32_t raftCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) { int32_t raftCacheGetEntry(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
if (ppEntry == NULL) { if (ppEntry == NULL) {
return -1; return -1;
} }
...@@ -292,7 +292,7 @@ int32_t raftCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index, SSync ...@@ -292,7 +292,7 @@ int32_t raftCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index, SSync
// success, return 0 // success, return 0
// error, return -1 // error, return -1
// not exist, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST // not exist, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
int32_t raftCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) { int32_t raftCacheGetEntryP(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
if (ppEntry == NULL) { if (ppEntry == NULL) {
return -1; return -1;
} }
...@@ -321,7 +321,7 @@ int32_t raftCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyn ...@@ -321,7 +321,7 @@ int32_t raftCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyn
return -1; return -1;
} }
int32_t raftCacheDelEntry(struct SRaftEntryCache* pCache, SyncIndex index) { int32_t raftCacheDelEntry(struct SRaftEntryHashCache* pCache, SyncIndex index) {
taosThreadMutexLock(&(pCache->mutex)); taosThreadMutexLock(&(pCache->mutex));
taosHashRemove(pCache->pEntryHash, &index, sizeof(index)); taosHashRemove(pCache->pEntryHash, &index, sizeof(index));
--(pCache->currentCount); --(pCache->currentCount);
...@@ -329,7 +329,7 @@ int32_t raftCacheDelEntry(struct SRaftEntryCache* pCache, SyncIndex index) { ...@@ -329,7 +329,7 @@ int32_t raftCacheDelEntry(struct SRaftEntryCache* pCache, SyncIndex index) {
return 0; return 0;
} }
int32_t raftCacheGetAndDel(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) { int32_t raftCacheGetAndDel(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
if (ppEntry == NULL) { if (ppEntry == NULL) {
return -1; return -1;
} }
...@@ -362,7 +362,7 @@ int32_t raftCacheGetAndDel(struct SRaftEntryCache* pCache, SyncIndex index, SSyn ...@@ -362,7 +362,7 @@ int32_t raftCacheGetAndDel(struct SRaftEntryCache* pCache, SyncIndex index, SSyn
return -1; return -1;
} }
int32_t raftCacheClear(struct SRaftEntryCache* pCache) { int32_t raftCacheClear(struct SRaftEntryHashCache* pCache) {
taosThreadMutexLock(&(pCache->mutex)); taosThreadMutexLock(&(pCache->mutex));
taosHashClear(pCache->pEntryHash); taosHashClear(pCache->pEntryHash);
pCache->currentCount = 0; pCache->currentCount = 0;
...@@ -371,7 +371,7 @@ int32_t raftCacheClear(struct SRaftEntryCache* pCache) { ...@@ -371,7 +371,7 @@ int32_t raftCacheClear(struct SRaftEntryCache* pCache) {
} }
//----------------------------------- //-----------------------------------
cJSON* raftCache2Json(SRaftEntryCache* pCache) { cJSON* raftCache2Json(SRaftEntryHashCache* pCache) {
char u64buf[128] = {0}; char u64buf[128] = {0};
cJSON* pRoot = cJSON_CreateObject(); cJSON* pRoot = cJSON_CreateObject();
...@@ -402,41 +402,283 @@ cJSON* raftCache2Json(SRaftEntryCache* pCache) { ...@@ -402,41 +402,283 @@ cJSON* raftCache2Json(SRaftEntryCache* pCache) {
} }
cJSON* pJson = cJSON_CreateObject(); cJSON* pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SRaftEntryCache", pRoot); cJSON_AddItemToObject(pJson, "SRaftEntryHashCache", pRoot);
return pJson; return pJson;
} }
char* raftCache2Str(SRaftEntryCache* pCache) { char* raftCache2Str(SRaftEntryHashCache* pCache) {
cJSON* pJson = raftCache2Json(pCache); cJSON* pJson = raftCache2Json(pCache);
char* serialized = cJSON_Print(pJson); char* serialized = cJSON_Print(pJson);
cJSON_Delete(pJson); cJSON_Delete(pJson);
return serialized; return serialized;
} }
void raftCachePrint(SRaftEntryCache* pCache) { void raftCachePrint(SRaftEntryHashCache* pCache) {
char* serialized = raftCache2Str(pCache); char* serialized = raftCache2Str(pCache);
printf("raftCachePrint | len:%" PRIu64 " | %s \n", strlen(serialized), serialized); printf("raftCachePrint | len:%" PRIu64 " | %s \n", strlen(serialized), serialized);
fflush(NULL); fflush(NULL);
taosMemoryFree(serialized); taosMemoryFree(serialized);
} }
void raftCachePrint2(char* s, SRaftEntryCache* pCache) { void raftCachePrint2(char* s, SRaftEntryHashCache* pCache) {
char* serialized = raftCache2Str(pCache); char* serialized = raftCache2Str(pCache);
printf("raftCachePrint2 | len:%" PRIu64 " | %s | %s \n", strlen(serialized), s, serialized); printf("raftCachePrint2 | len:%" PRIu64 " | %s | %s \n", strlen(serialized), s, serialized);
fflush(NULL); fflush(NULL);
taosMemoryFree(serialized); taosMemoryFree(serialized);
} }
void raftCacheLog(SRaftEntryCache* pCache) { void raftCacheLog(SRaftEntryHashCache* pCache) {
char* serialized = raftCache2Str(pCache); char* serialized = raftCache2Str(pCache);
sTrace("raftCacheLog | len:%" PRIu64 " | %s", strlen(serialized), serialized); sTrace("raftCacheLog | len:%" PRIu64 " | %s", strlen(serialized), serialized);
taosMemoryFree(serialized); taosMemoryFree(serialized);
} }
void raftCacheLog2(char* s, SRaftEntryCache* pCache) { void raftCacheLog2(char* s, SRaftEntryHashCache* pCache) {
if (gRaftDetailLog) { if (gRaftDetailLog) {
char* serialized = raftCache2Str(pCache); char* serialized = raftCache2Str(pCache);
sTraceLong("raftCacheLog2 | len:%" PRIu64 " | %s | %s", strlen(serialized), s, serialized); sTraceLong("raftCacheLog2 | len:%" PRIu64 " | %s | %s", strlen(serialized), s, serialized);
taosMemoryFree(serialized); taosMemoryFree(serialized);
} }
} }
//-----------------------------------
static char* keyFn(const void* pData) {
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)pData;
return (char*)(&(pEntry->index));
}
static int cmpFn(const void* p1, const void* p2) { return memcmp(p1, p2, sizeof(SyncIndex)); }
SRaftEntryCache* raftEntryCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) {
SRaftEntryCache* pCache = taosMemoryMalloc(sizeof(SRaftEntryCache));
if (pCache == NULL) {
sError("vgId:%d raft cache create error", pSyncNode->vgId);
return NULL;
}
pCache->pSkipList =
tSkipListCreate(MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, sizeof(SyncIndex), cmpFn, SL_ALLOW_DUP_KEY, keyFn);
if (pCache->pSkipList == NULL) {
sError("vgId:%d raft cache create hash error", pSyncNode->vgId);
return NULL;
}
taosThreadMutexInit(&(pCache->mutex), NULL);
pCache->maxCount = maxCount;
pCache->currentCount = 0;
pCache->pSyncNode = pSyncNode;
return pCache;
}
void raftEntryCacheDestroy(SRaftEntryCache* pCache) {
if (pCache != NULL) {
taosThreadMutexLock(&(pCache->mutex));
tSkipListDestroy(pCache->pSkipList);
taosThreadMutexUnlock(&(pCache->mutex));
taosThreadMutexDestroy(&(pCache->mutex));
taosMemoryFree(pCache);
}
}
// success, return 1
// max count, return 0
// error, return -1
int32_t raftEntryCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry) {
taosThreadMutexLock(&(pCache->mutex));
if (pCache->currentCount >= pCache->maxCount) {
taosThreadMutexUnlock(&(pCache->mutex));
return 0;
}
SSkipListNode* pSkipListNode = tSkipListPut(pCache->pSkipList, pEntry);
ASSERT(pSkipListNode != NULL);
++(pCache->currentCount);
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "raft cache add, type:%s,%d, type2:%s,%d, index:%" PRId64 ", bytes:%d",
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType,
pEntry->index, pEntry->bytes);
syncNodeEventLog(pCache->pSyncNode, eventLog);
} while (0);
taosThreadMutexUnlock(&(pCache->mutex));
return 1;
}
// find one, return 1
// not found, return 0
// error, return -1
int32_t raftEntryCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
ASSERT(ppEntry != NULL);
SSyncRaftEntry* pEntry = NULL;
int32_t code = raftEntryCacheGetEntryP(pCache, index, &pEntry);
if (code == 1) {
*ppEntry = taosMemoryMalloc(pEntry->bytes);
memcpy(*ppEntry, pEntry, pEntry->bytes);
} else {
*ppEntry = NULL;
}
return code;
}
// find one, return 1
// not found, return 0
// error, return -1
int32_t raftEntryCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
taosThreadMutexLock(&(pCache->mutex));
SyncIndex index2 = index;
int32_t code = 0;
SArray* entryPArray = tSkipListGet(pCache->pSkipList, (char*)(&index2));
int32_t arraySize = taosArrayGetSize(entryPArray);
if (arraySize == 1) {
SSkipListNode** ppNode = (SSkipListNode**)taosArrayGet(entryPArray, 0);
ASSERT(*ppNode != NULL);
*ppEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(*ppNode);
code = 1;
} else if (arraySize == 0) {
code = 0;
} else {
ASSERT(0);
code = -1;
}
taosArrayDestroy(entryPArray);
taosThreadMutexUnlock(&(pCache->mutex));
return code;
}
// count = -1, clear all
// count >= 0, clear count
// return -1, error
// return delete count
int32_t raftEntryCacheClear(struct SRaftEntryCache* pCache, int32_t count) {
taosThreadMutexLock(&(pCache->mutex));
int32_t returnCnt = 0;
if (count == -1) {
// clear all
SSkipListIterator* pIter = tSkipListCreateIter(pCache->pSkipList);
while (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
syncEntryDestory(pEntry);
++returnCnt;
}
tSkipListDestroyIter(pIter);
tSkipListDestroy(pCache->pSkipList);
pCache->pSkipList =
tSkipListCreate(MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, sizeof(SyncIndex), cmpFn, SL_ALLOW_DUP_KEY, keyFn);
ASSERT(pCache->pSkipList != NULL);
} else {
// clear count
int i = 0;
SSkipListIterator* pIter = tSkipListCreateIter(pCache->pSkipList);
SArray* delNodeArray = taosArrayInit(0, sizeof(SSkipListNode*));
// free entry
while (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
if (i++ >= count) {
break;
}
// sDebug("push pNode:%p", pNode);
taosArrayPush(delNodeArray, &pNode);
++returnCnt;
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
syncEntryDestory(pEntry);
}
tSkipListDestroyIter(pIter);
// delete skiplist node
int32_t arraySize = taosArrayGetSize(delNodeArray);
for (int32_t i = 0; i < arraySize; ++i) {
SSkipListNode** ppNode = taosArrayGet(delNodeArray, i);
// sDebug("get pNode:%p", *ppNode);
tSkipListRemoveNode(pCache->pSkipList, *ppNode);
}
taosArrayDestroy(delNodeArray);
}
pCache->currentCount -= returnCnt;
taosThreadMutexUnlock(&(pCache->mutex));
return returnCnt;
}
cJSON* raftEntryCache2Json(SRaftEntryCache* pCache) {
char u64buf[128] = {0};
cJSON* pRoot = cJSON_CreateObject();
if (pCache != NULL) {
taosThreadMutexLock(&(pCache->mutex));
snprintf(u64buf, sizeof(u64buf), "%p", pCache->pSyncNode);
cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
cJSON_AddNumberToObject(pRoot, "currentCount", pCache->currentCount);
cJSON_AddNumberToObject(pRoot, "maxCount", pCache->maxCount);
cJSON* pEntries = cJSON_CreateArray();
cJSON_AddItemToObject(pRoot, "entries", pEntries);
SSkipListIterator* pIter = tSkipListCreateIter(pCache->pSkipList);
while (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
}
tSkipListDestroyIter(pIter);
taosThreadMutexUnlock(&(pCache->mutex));
}
cJSON* pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SRaftEntryCache", pRoot);
return pJson;
}
char* raftEntryCache2Str(SRaftEntryCache* pObj) {
cJSON* pJson = raftEntryCache2Json(pObj);
char* serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
void raftEntryCachePrint(SRaftEntryCache* pObj) {
char* serialized = raftEntryCache2Str(pObj);
printf("raftEntryCachePrint | len:%" PRIu64 " | %s \n", strlen(serialized), serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftEntryCachePrint2(char* s, SRaftEntryCache* pObj) {
char* serialized = raftEntryCache2Str(pObj);
printf("raftEntryCachePrint2 | len:%" PRIu64 " | %s | %s \n", strlen(serialized), s, serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftEntryCacheLog(SRaftEntryCache* pObj) {
char* serialized = raftEntryCache2Str(pObj);
sTrace("raftEntryCacheLog | len:%" PRIu64 " | %s", strlen(serialized), serialized);
taosMemoryFree(serialized);
}
void raftEntryCacheLog2(char* s, SRaftEntryCache* pObj) {
if (gRaftDetailLog) {
char* serialized = raftEntryCache2Str(pObj);
sTraceLong("raftEntryCacheLog2 | len:%" PRIu64 " | %s | %s", strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}
}
\ No newline at end of file
source/libs/sync/src/syncRespMgr.c
浏览文件 @ d0e83605
...@@ -122,54 +122,45 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) { ...@@ -122,54 +122,45 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) {
int cnt = 0; int cnt = 0;
SSyncNode *pSyncNode = pObj->data; SSyncNode *pSyncNode = pObj->data;
SArray *delIndexArray = taosArrayInit(0, sizeof(SyncIndex)); SArray *delIndexArray = taosArrayInit(0, sizeof(uint64_t));
ASSERT(delIndexArray != NULL); ASSERT(delIndexArray != NULL);
while (pStub) { while (pStub) {
size_t len; size_t len;
void * key = taosHashGetKey(pStub, &len); void * key = taosHashGetKey(pStub, &len);
SyncIndex *pIndex = (SyncIndex *)key; uint64_t *pSeqNum = (uint64_t *)key;
int64_t nowMS = taosGetTimestampMs(); int64_t nowMS = taosGetTimestampMs();
if (nowMS - pStub->createTime > ttl) { if (nowMS - pStub->createTime > ttl) {
taosArrayPush(delIndexArray, pIndex); taosArrayPush(delIndexArray, pSeqNum);
cnt++; cnt++;
SSyncRaftEntry *pEntry = NULL;
int32_t code = 0;
if (pSyncNode->pLogStore != NULL) {
code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, *pIndex, &pEntry);
if (code == 0 && pEntry != NULL) {
SFsmCbMeta cbMeta = {0}; SFsmCbMeta cbMeta = {0};
cbMeta.index = pEntry->index; cbMeta.index = SYNC_INDEX_INVALID;
cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(pSyncNode, cbMeta.index); cbMeta.lastConfigIndex = SYNC_INDEX_INVALID;
cbMeta.isWeak = pEntry->isWeak; cbMeta.isWeak = false;
cbMeta.code = TSDB_CODE_SYN_TIMEOUT; cbMeta.code = TSDB_CODE_SYN_TIMEOUT;
cbMeta.state = pSyncNode->state; cbMeta.state = pSyncNode->state;
cbMeta.seqNum = pEntry->seqNum; cbMeta.seqNum = *pSeqNum;
cbMeta.term = pEntry->term; cbMeta.term = SYNC_TERM_INVALID;
cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm; cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;
cbMeta.flag = 0; cbMeta.flag = 0;
SRpcMsg rpcMsg = pStub->rpcMsg; pStub->rpcMsg.pCont = NULL;
rpcMsg.pCont = rpcMallocCont(pEntry->dataLen); pStub->rpcMsg.contLen = 0;
memcpy(rpcMsg.pCont, pEntry->data, pEntry->dataLen); pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &(pStub->rpcMsg), cbMeta);
pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, cbMeta);
syncEntryDestory(pEntry);
}
}
} }
pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, pStub); pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, pStub);
} }
int32_t arraySize = taosArrayGetSize(delIndexArray); int32_t arraySize = taosArrayGetSize(delIndexArray);
sDebug("vgId:%d, resp clean by ttl, cnt:%d, array-size:%d", pSyncNode->vgId, cnt, arraySize); sDebug("vgId:%d, resp mgr clean by ttl, cnt:%d, array-size:%d", pSyncNode->vgId, cnt, arraySize);
for (int32_t i = 0; i < arraySize; ++i) { for (int32_t i = 0; i < arraySize; ++i) {
SyncIndex *pIndex = taosArrayGet(delIndexArray, i); uint64_t *pSeqNum = taosArrayGet(delIndexArray, i);
taosHashRemove(pObj->pRespHash, pIndex, sizeof(SyncIndex)); taosHashRemove(pObj->pRespHash, pSeqNum, sizeof(uint64_t));
sDebug("vgId:%d, resp mgr clean by ttl, seq:%d", pSyncNode->vgId, *pSeqNum);
} }
taosArrayDestroy(delIndexArray); taosArrayDestroy(delIndexArray);
} }
source/libs/sync/src/syncTimeout.c
浏览文件 @ d0e83605
...@@ -16,9 +16,14 @@ ...@@ -16,9 +16,14 @@
#include "syncTimeout.h" #include "syncTimeout.h"
#include "syncElection.h" #include "syncElection.h"
#include "syncReplication.h" #include "syncReplication.h"
#include "syncRespMgr.h"
int32_t syncNodeTimerRoutine(SSyncNode* ths) { int32_t syncNodeTimerRoutine(SSyncNode* ths) {
syncNodeEventLog(ths, "timer routines ... "); syncNodeEventLog(ths, "timer routines ... ");
if (ths->vgId != 1) {
syncRespClean(ths->pSyncRespMgr);
}
return 0; return 0;
} }
......
source/libs/sync/test/CMakeLists.txt
浏览文件 @ d0e83605
...@@ -18,6 +18,7 @@ add_executable(syncIndexMgrTest "") ...@@ -18,6 +18,7 @@ add_executable(syncIndexMgrTest "")
add_executable(syncLogStoreTest "") add_executable(syncLogStoreTest "")
add_executable(syncEntryTest "") add_executable(syncEntryTest "")
add_executable(syncEntryCacheTest "") add_executable(syncEntryCacheTest "")
add_executable(syncHashCacheTest "")
add_executable(syncRequestVoteTest "") add_executable(syncRequestVoteTest "")
add_executable(syncRequestVoteReplyTest "") add_executable(syncRequestVoteReplyTest "")
add_executable(syncAppendEntriesTest "") add_executable(syncAppendEntriesTest "")
...@@ -137,6 +138,10 @@ target_sources(syncEntryCacheTest ...@@ -137,6 +138,10 @@ target_sources(syncEntryCacheTest
PRIVATE PRIVATE
"syncEntryCacheTest.cpp" "syncEntryCacheTest.cpp"
) )
target_sources(syncHashCacheTest
PRIVATE
"syncHashCacheTest.cpp"
)
target_sources(syncRequestVoteTest target_sources(syncRequestVoteTest
PRIVATE PRIVATE
"syncRequestVoteTest.cpp" "syncRequestVoteTest.cpp"
...@@ -387,6 +392,11 @@ target_include_directories(syncEntryCacheTest ...@@ -387,6 +392,11 @@ target_include_directories(syncEntryCacheTest
"${TD_SOURCE_DIR}/include/libs/sync" "${TD_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc" "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
) )
target_include_directories(syncHashCacheTest
PUBLIC
"${TD_SOURCE_DIR}/include/libs/sync"
"${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
target_include_directories(syncRequestVoteTest target_include_directories(syncRequestVoteTest
PUBLIC PUBLIC
"${TD_SOURCE_DIR}/include/libs/sync" "${TD_SOURCE_DIR}/include/libs/sync"
...@@ -654,6 +664,10 @@ target_link_libraries(syncEntryCacheTest ...@@ -654,6 +664,10 @@ target_link_libraries(syncEntryCacheTest
sync sync
gtest_main gtest_main
) )
target_link_libraries(syncHashCacheTest
sync
gtest_main
)
target_link_libraries(syncRequestVoteTest target_link_libraries(syncRequestVoteTest
sync sync
gtest_main gtest_main
......
source/libs/sync/test/syncEntryCacheTest.cpp
浏览文件 @ d0e83605
...@@ -43,7 +43,7 @@ SRaftEntryCache* createCache(int maxCount) { ...@@ -43,7 +43,7 @@ SRaftEntryCache* createCache(int maxCount) {
SSyncNode* pSyncNode = createFakeNode(); SSyncNode* pSyncNode = createFakeNode();
ASSERT(pSyncNode != NULL); ASSERT(pSyncNode != NULL);
SRaftEntryCache* pCache = raftCacheCreate(pSyncNode, maxCount); SRaftEntryCache* pCache = raftEntryCacheCreate(pSyncNode, maxCount);
ASSERT(pCache != NULL); ASSERT(pCache != NULL);
return pCache; return pCache;
...@@ -52,213 +52,73 @@ SRaftEntryCache* createCache(int maxCount) { ...@@ -52,213 +52,73 @@ SRaftEntryCache* createCache(int maxCount) {
void test1() { void test1() {
int32_t code = 0; int32_t code = 0;
SRaftEntryCache* pCache = createCache(5); SRaftEntryCache* pCache = createCache(5);
for (int i = 0; i < 5; ++i) { for (int i = 0; i < 10; ++i) {
SSyncRaftEntry* pEntry = createEntry(i); SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry); code = raftEntryCachePutEntry(pCache, pEntry);
ASSERT(code == 1); sTrace("put entry code:%d, pEntry:%p", code, pEntry);
syncEntryDestory(pEntry);
} }
raftCacheLog2((char*)"==test1 write 5 entries==", pCache); raftEntryCacheLog2((char*)"==test1 write 5 entries==", pCache);
SyncIndex index; raftEntryCacheClear(pCache, 3);
index = 1; raftEntryCacheLog2((char*)"==test1 evict 3 entries==", pCache);
code = raftCacheDelEntry(pCache, index);
ASSERT(code == 0);
index = 3;
code = raftCacheDelEntry(pCache, index);
ASSERT(code == 0);
raftCacheLog2((char*)"==test1 delete 1,3==", pCache);
code = raftCacheClear(pCache); raftEntryCacheClear(pCache, -1);
ASSERT(code == 0); raftEntryCacheLog2((char*)"==test1 evict -1(all) entries==", pCache);
raftCacheLog2((char*)"==clear all==", pCache);
} }
void test2() { void test2() {
int32_t code = 0; int32_t code = 0;
SRaftEntryCache* pCache = createCache(5); SRaftEntryCache* pCache = createCache(5);
for (int i = 0; i < 5; ++i) { for (int i = 0; i < 10; ++i) {
SSyncRaftEntry* pEntry = createEntry(i); SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry); code = raftEntryCachePutEntry(pCache, pEntry);
ASSERT(code == 1); sTrace("put entry code:%d, pEntry:%p", code, pEntry);
syncEntryDestory(pEntry);
} }
raftCacheLog2((char*)"==test2 write 5 entries==", pCache); raftEntryCacheLog2((char*)"==test1 write 5 entries==", pCache);
SyncIndex index; SyncIndex index = 2;
index = 1; SSyncRaftEntry* pEntry = NULL;
SSyncRaftEntry* pEntry;
code = raftCacheGetEntry(pCache, index, &pEntry);
ASSERT(code == 0);
syncEntryDestory(pEntry);
syncEntryLog2((char*)"==test2 get entry 1==", pEntry);
index = 2; code = raftEntryCacheGetEntryP(pCache, index, &pEntry);
code = raftCacheGetEntryP(pCache, index, &pEntry); ASSERT(code == 1 && index == pEntry->index);
ASSERT(code == 0); sTrace("get entry:%p for %ld", pEntry, index);
syncEntryLog2((char*)"==test2 get entry pointer 2==", pEntry); syncEntryLog2((char*)"==test2 get entry pointer 2==", pEntry);
code = raftEntryCacheGetEntry(pCache, index, &pEntry);
ASSERT(code == 1 && index == pEntry->index);
sTrace("get entry:%p for %ld", pEntry, index);
syncEntryLog2((char*)"==test2 get entry 2==", pEntry);
syncEntryDestory(pEntry);
// not found // not found
index = 8; index = 8;
code = raftCacheGetEntry(pCache, index, &pEntry); code = raftEntryCacheGetEntry(pCache, index, &pEntry);
ASSERT(code == -1 && terrno == TSDB_CODE_WAL_LOG_NOT_EXIST); ASSERT(code == 0);
sTrace("get entry:%p for %ld", pEntry, index);
sTrace("==test2 get entry 8 not found=="); sTrace("==test2 get entry 8 not found==");
// not found // not found
index = 9; index = 9;
code = raftCacheGetEntryP(pCache, index, &pEntry); code = raftEntryCacheGetEntry(pCache, index, &pEntry);
ASSERT(code == -1 && terrno == TSDB_CODE_WAL_LOG_NOT_EXIST);
sTrace("==test2 get entry pointer 9 not found==");
}
void test3() {
int32_t code = 0;
SRaftEntryCache* pCache = createCache(5);
for (int i = 0; i < 5; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry);
ASSERT(code == 1);
syncEntryDestory(pEntry);
}
for (int i = 6; i < 10; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry);
ASSERT(code == 0); ASSERT(code == 0);
syncEntryDestory(pEntry); sTrace("get entry:%p for %ld", pEntry, index);
} sTrace("==test2 get entry 9 not found==");
raftCacheLog2((char*)"==test3 write 10 entries, max count is 5==", pCache);
} }
void test4() { void test3() {
int32_t code = 0; int32_t code = 0;
SRaftEntryCache* pCache = createCache(5); SRaftEntryCache* pCache = createCache(20);
for (int i = 0; i < 5; ++i) { for (int i = 0; i <= 4; ++i) {
SSyncRaftEntry* pEntry = createEntry(i); SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry); code = raftEntryCachePutEntry(pCache, pEntry);
ASSERT(code == 1); sTrace("put entry code:%d, pEntry:%p", code, pEntry);
syncEntryDestory(pEntry);
}
raftCacheLog2((char*)"==test4 write 5 entries==", pCache);
SyncIndex index;
index = 3;
SSyncRaftEntry* pEntry;
code = raftCacheGetAndDel(pCache, index, &pEntry);
ASSERT(code == 0);
syncEntryLog2((char*)"==test4 get-and-del entry 3==", pEntry);
raftCacheLog2((char*)"==test4 after get-and-del entry 3==", pCache);
}
static char* keyFn(const void* pData) {
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)pData;
return (char*)(&(pEntry->index));
}
static int cmpFn(const void* p1, const void* p2) { return memcmp(p1, p2, sizeof(SyncIndex)); }
void printSkipList(SSkipList* pSkipList) {
ASSERT(pSkipList != NULL);
SSkipListIterator* pIter = tSkipListCreateIter(pSkipList);
while (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
syncEntryPrint2((char*)"", pEntry);
} }
}
void delSkipListFirst(SSkipList* pSkipList, int n) {
ASSERT(pSkipList != NULL);
sTrace("delete first %d -------------", n);
SSkipListIterator* pIter = tSkipListCreateIter(pSkipList);
for (int i = 0; i < n; ++i) {
tSkipListIterNext(pIter);
SSkipListNode* pNode = tSkipListIterGet(pIter);
tSkipListRemoveNode(pSkipList, pNode);
}
}
SSyncRaftEntry* getLogEntry2(SSkipList* pSkipList, SyncIndex index) {
SyncIndex index2 = index;
SSyncRaftEntry *pEntry = NULL;
int arraySize = 0;
SArray* entryPArray = tSkipListGet(pSkipList, (char*)(&index2));
arraySize = taosArrayGetSize(entryPArray);
if (arraySize > 0) {
SSkipListNode** ppNode = (SSkipListNode**)taosArrayGet(entryPArray, 0);
ASSERT(*ppNode != NULL);
pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(*ppNode);
}
taosArrayDestroy(entryPArray);
sTrace("get index2: %ld, arraySize:%d -------------", index, arraySize);
syncEntryLog2((char*)"getLogEntry2", pEntry);
return pEntry;
}
SSyncRaftEntry* getLogEntry(SSkipList* pSkipList, SyncIndex index) {
sTrace("get index: %ld -------------", index);
SyncIndex index2 = index;
SSyncRaftEntry *pEntry = NULL;
SSkipListIterator* pIter = tSkipListCreateIterFromVal(pSkipList, (const char *)&index2, TSDB_DATA_TYPE_BINARY, TSDB_ORDER_ASC);
if (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
}
syncEntryLog2((char*)"getLogEntry", pEntry);
return pEntry;
}
void test5() {
SSkipList* pSkipList =
tSkipListCreate(MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, sizeof(SyncIndex), cmpFn, SL_ALLOW_DUP_KEY, keyFn);
ASSERT(pSkipList != NULL);
sTrace("insert 9 - 5");
for (int i = 9; i >= 5; --i) { for (int i = 9; i >= 5; --i) {
SSyncRaftEntry* pEntry = createEntry(i); SSyncRaftEntry* pEntry = createEntry(i);
SSkipListNode* pSkipListNode = tSkipListPut(pSkipList, pEntry); code = raftEntryCachePutEntry(pCache, pEntry);
} sTrace("put entry code:%d, pEntry:%p", code, pEntry);
sTrace("insert 0 - 4");
for (int i = 0; i <= 4; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
SSkipListNode* pSkipListNode = tSkipListPut(pSkipList, pEntry);
}
sTrace("insert 7 7 7 7 7");
for (int i = 0; i <= 4; ++i) {
SSyncRaftEntry* pEntry = createEntry(7);
SSkipListNode* pSkipListNode = tSkipListPut(pSkipList, pEntry);
} }
raftEntryCacheLog2((char*)"==test3 write 10 entries==", pCache);
sTrace("print: -------------");
printSkipList(pSkipList);
delSkipListFirst(pSkipList, 3);
sTrace("print: -------------");
printSkipList(pSkipList);
getLogEntry(pSkipList, 2);
getLogEntry(pSkipList, 5);
getLogEntry(pSkipList, 7);
getLogEntry(pSkipList, 7);
getLogEntry2(pSkipList, 2);
getLogEntry2(pSkipList, 5);
getLogEntry2(pSkipList, 7);
getLogEntry2(pSkipList, 7);
tSkipListDestroy(pSkipList);
} }
int main(int argc, char** argv) { int main(int argc, char** argv) {
...@@ -266,14 +126,9 @@ int main(int argc, char** argv) { ...@@ -266,14 +126,9 @@ int main(int argc, char** argv) {
tsAsyncLog = 0; tsAsyncLog = 0;
sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE + DEBUG_DEBUG; sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE + DEBUG_DEBUG;
/*
test1(); test1();
test2(); test2();
test3(); test3();
test4();
*/
test5();
return 0; return 0;
} }
source/libs/sync/test/syncHashCacheTest.cpp 0 → 100644
浏览文件 @ d0e83605
#include <stdio.h>
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncRaftLog.h"
#include "syncRaftStore.h"
#include "syncUtil.h"
#include "tskiplist.h"
void logTest() {
sTrace("--- sync log test: trace");
sDebug("--- sync log test: debug");
sInfo("--- sync log test: info");
sWarn("--- sync log test: warn");
sError("--- sync log test: error");
sFatal("--- sync log test: fatal");
}
SSyncRaftEntry* createEntry(int i) {
int32_t dataLen = 20;
SSyncRaftEntry* pEntry = syncEntryBuild(dataLen);
assert(pEntry != NULL);
pEntry->msgType = 88;
pEntry->originalRpcType = 99;
pEntry->seqNum = 3;
pEntry->isWeak = true;
pEntry->term = 100 + i;
pEntry->index = i;
snprintf(pEntry->data, dataLen, "value%d", i);
return pEntry;
}
SSyncNode* createFakeNode() {
SSyncNode* pSyncNode = (SSyncNode*)taosMemoryMalloc(sizeof(SSyncNode));
ASSERT(pSyncNode != NULL);
memset(pSyncNode, 0, sizeof(SSyncNode));
return pSyncNode;
}
SRaftEntryHashCache* createCache(int maxCount) {
SSyncNode* pSyncNode = createFakeNode();
ASSERT(pSyncNode != NULL);
SRaftEntryHashCache* pCache = raftCacheCreate(pSyncNode, maxCount);
ASSERT(pCache != NULL);
return pCache;
}
void test1() {
int32_t code = 0;
SRaftEntryHashCache* pCache = createCache(5);
for (int i = 0; i < 5; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry);
ASSERT(code == 1);
syncEntryDestory(pEntry);
}
raftCacheLog2((char*)"==test1 write 5 entries==", pCache);
SyncIndex index;
index = 1;
code = raftCacheDelEntry(pCache, index);
ASSERT(code == 0);
index = 3;
code = raftCacheDelEntry(pCache, index);
ASSERT(code == 0);
raftCacheLog2((char*)"==test1 delete 1,3==", pCache);
code = raftCacheClear(pCache);
ASSERT(code == 0);
raftCacheLog2((char*)"==clear all==", pCache);
}
void test2() {
int32_t code = 0;
SRaftEntryHashCache* pCache = createCache(5);
for (int i = 0; i < 5; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry);
ASSERT(code == 1);
syncEntryDestory(pEntry);
}
raftCacheLog2((char*)"==test2 write 5 entries==", pCache);
SyncIndex index;
index = 1;
SSyncRaftEntry* pEntry;
code = raftCacheGetEntry(pCache, index, &pEntry);
ASSERT(code == 0);
syncEntryDestory(pEntry);
syncEntryLog2((char*)"==test2 get entry 1==", pEntry);
index = 2;
code = raftCacheGetEntryP(pCache, index, &pEntry);
ASSERT(code == 0);
syncEntryLog2((char*)"==test2 get entry pointer 2==", pEntry);
// not found
index = 8;
code = raftCacheGetEntry(pCache, index, &pEntry);
ASSERT(code == -1 && terrno == TSDB_CODE_WAL_LOG_NOT_EXIST);
sTrace("==test2 get entry 8 not found==");
// not found
index = 9;
code = raftCacheGetEntryP(pCache, index, &pEntry);
ASSERT(code == -1 && terrno == TSDB_CODE_WAL_LOG_NOT_EXIST);
sTrace("==test2 get entry pointer 9 not found==");
}
void test3() {
int32_t code = 0;
SRaftEntryHashCache* pCache = createCache(5);
for (int i = 0; i < 5; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry);
ASSERT(code == 1);
syncEntryDestory(pEntry);
}
for (int i = 6; i < 10; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry);
ASSERT(code == 0);
syncEntryDestory(pEntry);
}
raftCacheLog2((char*)"==test3 write 10 entries, max count is 5==", pCache);
}
void test4() {
int32_t code = 0;
SRaftEntryHashCache* pCache = createCache(5);
for (int i = 0; i < 5; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
code = raftCachePutEntry(pCache, pEntry);
ASSERT(code == 1);
syncEntryDestory(pEntry);
}
raftCacheLog2((char*)"==test4 write 5 entries==", pCache);
SyncIndex index;
index = 3;
SSyncRaftEntry* pEntry;
code = raftCacheGetAndDel(pCache, index, &pEntry);
ASSERT(code == 0);
syncEntryLog2((char*)"==test4 get-and-del entry 3==", pEntry);
raftCacheLog2((char*)"==test4 after get-and-del entry 3==", pCache);
}
static char* keyFn(const void* pData) {
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)pData;
return (char*)(&(pEntry->index));
}
static int cmpFn(const void* p1, const void* p2) { return memcmp(p1, p2, sizeof(SyncIndex)); }
void printSkipList(SSkipList* pSkipList) {
ASSERT(pSkipList != NULL);
SSkipListIterator* pIter = tSkipListCreateIter(pSkipList);
while (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
syncEntryPrint2((char*)"", pEntry);
}
}
void delSkipListFirst(SSkipList* pSkipList, int n) {
ASSERT(pSkipList != NULL);
sTrace("delete first %d -------------", n);
SSkipListIterator* pIter = tSkipListCreateIter(pSkipList);
for (int i = 0; i < n; ++i) {
tSkipListIterNext(pIter);
SSkipListNode* pNode = tSkipListIterGet(pIter);
tSkipListRemoveNode(pSkipList, pNode);
}
}
SSyncRaftEntry* getLogEntry2(SSkipList* pSkipList, SyncIndex index) {
SyncIndex index2 = index;
SSyncRaftEntry* pEntry = NULL;
int arraySize = 0;
SArray* entryPArray = tSkipListGet(pSkipList, (char*)(&index2));
arraySize = taosArrayGetSize(entryPArray);
if (arraySize > 0) {
SSkipListNode** ppNode = (SSkipListNode**)taosArrayGet(entryPArray, 0);
ASSERT(*ppNode != NULL);
pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(*ppNode);
}
taosArrayDestroy(entryPArray);
sTrace("get index2: %ld, arraySize:%d -------------", index, arraySize);
syncEntryLog2((char*)"getLogEntry2", pEntry);
return pEntry;
}
SSyncRaftEntry* getLogEntry(SSkipList* pSkipList, SyncIndex index) {
sTrace("get index: %ld -------------", index);
SyncIndex index2 = index;
SSyncRaftEntry* pEntry = NULL;
SSkipListIterator* pIter =
tSkipListCreateIterFromVal(pSkipList, (const char*)&index2, TSDB_DATA_TYPE_BINARY, TSDB_ORDER_ASC);
if (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
}
syncEntryLog2((char*)"getLogEntry", pEntry);
return pEntry;
}
void test5() {
SSkipList* pSkipList =
tSkipListCreate(MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, sizeof(SyncIndex), cmpFn, SL_ALLOW_DUP_KEY, keyFn);
ASSERT(pSkipList != NULL);
sTrace("insert 9 - 5");
for (int i = 9; i >= 5; --i) {
SSyncRaftEntry* pEntry = createEntry(i);
SSkipListNode* pSkipListNode = tSkipListPut(pSkipList, pEntry);
}
sTrace("insert 0 - 4");
for (int i = 0; i <= 4; ++i) {
SSyncRaftEntry* pEntry = createEntry(i);
SSkipListNode* pSkipListNode = tSkipListPut(pSkipList, pEntry);
}
sTrace("insert 7 7 7 7 7");
for (int i = 0; i <= 4; ++i) {
SSyncRaftEntry* pEntry = createEntry(7);
SSkipListNode* pSkipListNode = tSkipListPut(pSkipList, pEntry);
}
sTrace("print: -------------");
printSkipList(pSkipList);
delSkipListFirst(pSkipList, 3);
sTrace("print: -------------");
printSkipList(pSkipList);
getLogEntry(pSkipList, 2);
getLogEntry(pSkipList, 5);
getLogEntry(pSkipList, 7);
getLogEntry(pSkipList, 7);
getLogEntry2(pSkipList, 2);
getLogEntry2(pSkipList, 5);
getLogEntry2(pSkipList, 7);
getLogEntry2(pSkipList, 7);
tSkipListDestroy(pSkipList);
}
int main(int argc, char** argv) {
gRaftDetailLog = true;
tsAsyncLog = 0;
sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE + DEBUG_DEBUG;
/*
test1();
test2();
test3();
test4();
*/
test5();
return 0;
}
tests/pytest/util/sqlset.py
浏览文件 @ d0e83605
...@@ -41,7 +41,7 @@ class TDSetSql: ...@@ -41,7 +41,7 @@ class TDSetSql:
create_stb_sql = f'create table {stbname} ({column_sql[:-1]}) tags({tag_sql[:-1]})' create_stb_sql = f'create table {stbname} ({column_sql[:-1]}) tags({tag_sql[:-1]})'
return create_stb_sql return create_stb_sql
def set_insertsql(self,column_dict,tbname,binary_str,nchar_str): def set_insertsql(self,column_dict,tbname,binary_str=None,nchar_str=None):
sql = '' sql = ''
for k, v in column_dict.items(): for k, v in column_dict.items():
if v.lower() == 'timestamp' or v.lower() == 'tinyint' or v.lower() == 'smallint' or v.lower() == 'int' or v.lower() == 'bigint' or \ if v.lower() == 'timestamp' or v.lower() == 'tinyint' or v.lower() == 'smallint' or v.lower() == 'int' or v.lower() == 'bigint' or \
......
tests/system-test/0-others/sysinfo.py 0 → 100644
浏览文件 @ d0e83605
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
from util.log import *
from util.cases import *
from util.sql import *
import subprocess
from util.common import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor())
self.dbname = 'db'
def get_database_info(self):
tdSql.query('select database()')
tdSql.checkData(0,0,None)
tdSql.execute(f'create database if not exists {self.dbname}')
tdSql.execute(f'use {self.dbname}')
tdSql.query('select database()')
tdSql.checkData(0,0,self.dbname)
tdSql.execute(f'drop database {self.dbname}')
def check_version(self):
taos_list = ['server','client']
for i in taos_list:
tdSql.query(f'select {i}_version()')
version_info = str(subprocess.run('cat ../../source/util/src/version.c |grep "char version"', shell=True,capture_output=True).stdout.decode('utf8')).split('"')[1]
tdSql.checkData(0,0,version_info)
def get_server_status(self):
tdSql.query('select server_status()')
tdSql.checkData(0,0,1)
tdDnodes.stoptaosd(1)
tdSql.query('select server_status()')
print(tdSql.queryResult)
def run(self):
self.get_database_info()
self.check_version()
# self.get_server_status()
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file
tests/system-test/2-query/count.py
浏览文件 @ d0e83605
from util.log import * from util.log import *
from util.sql import * from util.sql import *
from util.cases import * from util.cases import *
from util.sqlset import *
class TDTestCase: class TDTestCase:
def init(self, conn, logSql): def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__) tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor()) tdSql.init(conn.cursor(),logSql)
self.setsql = TDSetSql()
self.rowNum = 10 self.rowNum = 10
self.ts = 1537146000000 self.ts = 1537146000000
def run(self): self.ntbname = 'ntb'
self.stbname = 'stb'
self.column_dict = {
'ts':'timestamp',
'c1':'int',
'c2':'float',
'c3':'double',
'c4':'timestamp'
}
self.tag_dict = {
't0':'int'
}
# The number of tag_values should be same as tbnum
self.tbnum = 2
self.tag_values = [
f'10',
f'100'
]
def query_stb(self,k,stbname,tbnum,rownum):
tdSql.query(f'select count({k}) from {stbname}')
tdSql.checkEqual(tdSql.queryResult[0][0],tbnum*rownum)
tdSql.query(f'select count({k}) from {stbname} where ts <={self.ts+self.rowNum-1}')
tdSql.checkEqual(tdSql.queryResult[0][0],tbnum*rownum)
tdSql.query(f'select count({k}) from {stbname} where ts <={self.ts+self.rowNum-2}')
tdSql.checkEqual(tdSql.queryResult[0][0],tbnum*(rownum-1))
def query_ctb(self,k,i,stbname,rownum):
tdSql.query(f'select count({k}) from {stbname}_{i}')
tdSql.checkEqual(tdSql.queryResult[0][0],rownum)
tdSql.query(f'select count({k}) from {stbname}_{i} where ts <={self.ts+self.rowNum-1}')
tdSql.checkEqual(tdSql.queryResult[0][0],rownum)
tdSql.query(f'select count({k}) from {stbname}_{i} where ts <={self.ts+self.rowNum-2}')
tdSql.checkEqual(tdSql.queryResult[0][0],rownum-1)
def query_ntb(self,k,ntbname,rownum):
tdSql.query(f'select count({k}) from {ntbname}')
tdSql.checkEqual(tdSql.queryResult[0][0],rownum)
tdSql.query(f'select count({k}) from {ntbname} where ts <={self.ts+self.rowNum-1}')
tdSql.checkEqual(tdSql.queryResult[0][0],rownum)
tdSql.query(f'select count({k}) from {ntbname} where ts <={self.ts+self.rowNum-2}')
tdSql.checkEqual(tdSql.queryResult[0][0],rownum-1)
def count_query_stb(self,column_dict,tag_dict,stbname,tbnum,rownum):
tdSql.query(f'select count(tbname) from {stbname}')
tdSql.checkEqual(tdSql.queryResult[0][0],tbnum*rownum)
tdSql.query(f'SELECT count(*) from (select distinct tbname from {stbname})')
tdSql.checkEqual(tdSql.queryResult[0][0],tbnum)
for k in column_dict.keys():
self.query_stb(k,stbname,tbnum,rownum)
for k in tag_dict.keys():
self.query_stb(k,stbname,tbnum,rownum)
def count_query_ctb(self,column_dict,tag_dict,stbname,tbnum,rownum):
for i in range(tbnum):
tdSql.query(f'select count(tbname) from {stbname}_{i}')
tdSql.checkEqual(tdSql.queryResult[0][0],rownum)
for k in column_dict.keys():
self.query_ctb(k,i,stbname,rownum)
for k in tag_dict.keys():
self.query_ctb(k,i,stbname,rownum)
def count_query_ntb(self,column_dict,ntbname,rownum):
tdSql.query(f'select count(tbname) from {ntbname}')
tdSql.checkEqual(tdSql.queryResult[0][0],rownum)
for k in column_dict.keys():
self.query_ntb(k,ntbname,rownum)
def insert_data(self,column_dict,tbname,row_num):
insert_sql = self.setsql.set_insertsql(column_dict,tbname)
for i in range(row_num):
insert_list = []
self.setsql.insert_values(column_dict,i,insert_sql,insert_list,self.ts)
def check_ntb(self):
tdSql.prepare() tdSql.prepare()
tdSql.execute(self.setsql.set_create_normaltable_sql(self.ntbname,self.column_dict))
tdSql.execute('''create table stb(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double, tdSql.query(f'select count(tbname) from {self.ntbname}')
col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned) tags(loc nchar(20))''') tdSql.checkRows(0)
tdSql.execute("create table stb_1 using stb tags('beijing')") tdSql.execute('flush database db')
tdSql.execute("create table stb_2 using stb tags('shanghai')") tdSql.query(f'select count(tbname) from {self.ntbname}')
tdSql.checkRows(0)
tdSql.execute('''create table ntb(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double, self.insert_data(self.column_dict,self.ntbname,self.rowNum)
col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned)''') self.count_query_ntb(self.column_dict,self.ntbname,self.rowNum)
tdSql.execute('flush database db')
for i in range(self.rowNum): self.count_query_ntb(self.column_dict,self.ntbname,self.rowNum)
tdSql.execute("insert into stb_1 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)" tdSql.execute('drop database db')
% (self.ts + i, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i + 1, i + 1, i + 1, i + 1)) def check_stb(self):
tdSql.execute("insert into stb_2 values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)" tdSql.prepare()
% (self.ts + i, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i + 1, i + 1, i + 1, i + 1)) tdSql.execute(self.setsql.set_create_stable_sql(self.stbname,self.column_dict,self.tag_dict))
for i in range(self.tbnum):
for i in range(self.rowNum): tdSql.execute(f'create table {self.stbname}_{i} using {self.stbname} tags({self.tag_values[i]})')
tdSql.execute("insert into ntb values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)" #!TODO
% (self.ts + i, i + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i + 1, i + 1, i + 1, i + 1)) # tdSql.query(f'SELECT count(*) from (select distinct tbname from {self.stbname})')
# tdSql.checkEqual(tdSql.queryResult[0][0],self.tbnum)
tdSql.query("select count(*) from stb") tdSql.query(f'select count(tbname) from {self.stbname}')
tdSql.checkData(0,0,20) tdSql.checkRows(0)
tdSql.query("select count(*) from db.stb") tdSql.execute('flush database db')
tdSql.checkData(0,0,20) tdSql.query(f'select count(tbname) from {self.stbname}')
tdSql.query("select count(ts) from stb") tdSql.checkRows(0)
tdSql.checkData(0,0,20) #!TODO
tdSql.query("select count(ts) from db.stb") # tdSql.query(f'SELECT count(*) from (select distinct tbname from {self.stbname})')
tdSql.checkData(0,0,20) # tdSql.checkEqual(tdSql.queryResult[0][0],self.tbnum)
tdSql.query("select count(col1) from stb") for i in range(self.tbnum):
tdSql.checkData(0,0,20) self.insert_data(self.column_dict,f'{self.stbname}_{i}',self.rowNum)
tdSql.query("select count(col1) from db.stb") self.count_query_stb(self.column_dict,self.tag_dict,self.stbname,self.tbnum,self.rowNum)
tdSql.checkData(0,0,20) self.count_query_ctb(self.column_dict,self.tag_dict,self.stbname,self.tbnum,self.rowNum)
tdSql.query("select count(col2) from stb") tdSql.execute('flush database db')
tdSql.checkData(0,0,20) self.count_query_stb(self.column_dict,self.tag_dict,self.stbname,self.tbnum,self.rowNum)
tdSql.query("select count(col2) from db.stb") self.count_query_ctb(self.column_dict,self.tag_dict,self.stbname,self.tbnum,self.rowNum)
tdSql.checkData(0,0,20) tdSql.execute('drop database db')
tdSql.query("select count(col3) from stb") def run(self):
tdSql.checkData(0,0,20) self.check_stb()
tdSql.query("select count(col3) from db.stb") self.check_ntb()
tdSql.checkData(0,0,20)
tdSql.query("select count(col4) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col4) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col5) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col5) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col6) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col6) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col7) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col7) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col8) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col8) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col9) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col9) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col11) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col11) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col12) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col12) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col13) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col13) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col14) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col14) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(ts) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(ts) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col1) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col1) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col2) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col2) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col3) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col3) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col4) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col4) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col5) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col5) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col6) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col6) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col7) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col7) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col8) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col8) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col9) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col9) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col11) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col11) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col12) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col12) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col13) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col13) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col14) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col14) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col1) from stb_1 group by col7")
tdSql.checkRows(2)
tdSql.execute("insert into stb_1 values(now,null,null,null,null,null,null,null,null,null,null,null,null,null)")
tdSql.query("select count(col1) from stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col1),count(ts) from stb_1")
tdSql.checkData(0,0,10)
tdSql.checkData(0,1,11)
tdSql.query("select count(col1) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.query("select count(col1),count(ts) from db.stb_1")
tdSql.checkData(0,0,10)
tdSql.checkData(0,1,11)
tdSql.query("select count(col1) from stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col1),count(ts) from stb")
tdSql.checkData(0,0,20)
tdSql.checkData(0,1,21)
tdSql.query("select count(col1) from db.stb")
tdSql.checkData(0,0,20)
tdSql.query("select count(col1),count(ts) from db.stb")
tdSql.checkData(0,0,20)
tdSql.checkData(0,1,21)
tdSql.query("select count(col1) from stb_1 group by col7")
tdSql.checkRows(3)
tdSql.query("select count(col1) from stb_2 group by col7")
tdSql.checkRows(2)
tdSql.query("select count(col1) from stb group by col7")
tdSql.checkRows(3)
def stop(self): def stop(self):
tdSql.close() tdSql.close()
tdLog.success("%s successfully executed" % __file__) tdLog.success("%s successfully executed" % __file__)
......
tests/system-test/7-tmq/tmqCommon.py
浏览文件 @ d0e83605
...@@ -371,7 +371,7 @@ class TMQCom: ...@@ -371,7 +371,7 @@ class TMQCom:
elif (i % 3 == 0): elif (i % 3 == 0):
tagBinaryValue = 'changsha' tagBinaryValue = 'changsha'
sql += " %s.%s_%d using %s.%s tags (%d, %d, %d, '%s', '%s') values "%(dbName,ctbPrefix,i+ctbStartIdx,dbName,stbName,i+ctbStartIdx,i+ctbStartIdx,i+ctbStartIdx,tagBinaryValue,tagBinaryValue) sql += " %s.%s%d using %s.%s tags (%d, %d, %d, '%s', '%s') values "%(dbName,ctbPrefix,i+ctbStartIdx,dbName,stbName,i+ctbStartIdx,i+ctbStartIdx,i+ctbStartIdx,tagBinaryValue,tagBinaryValue)
for j in range(rowsPerTbl): for j in range(rowsPerTbl):
sql += "(%d, %d, %d, %d, 'binary_%d', 'nchar_%d', now) "%(startTs+j, j,j, j,i+ctbStartIdx,rowsBatched) sql += "(%d, %d, %d, %d, 'binary_%d', 'nchar_%d', now) "%(startTs+j, j,j, j,i+ctbStartIdx,rowsBatched)
rowsBatched += 1 rowsBatched += 1
...@@ -379,7 +379,7 @@ class TMQCom: ...@@ -379,7 +379,7 @@ class TMQCom:
tsql.execute(sql) tsql.execute(sql)
rowsBatched = 0 rowsBatched = 0
if j < rowsPerTbl - 1: if j < rowsPerTbl - 1:
sql = "insert into %s.%s_%d using %s.%s tags (%d, %d, %d, '%s', '%s') values " %(dbName,ctbPrefix,i+ctbStartIdx,dbName,stbName,i+ctbStartIdx,i+ctbStartIdx,i+ctbStartIdx,tagBinaryValue,tagBinaryValue) sql = "insert into %s.%s%d using %s.%s tags (%d, %d, %d, '%s', '%s') values " %(dbName,ctbPrefix,i+ctbStartIdx,dbName,stbName,i+ctbStartIdx,i+ctbStartIdx,i+ctbStartIdx,tagBinaryValue,tagBinaryValue)
else: else:
sql = "insert into " sql = "insert into "
#end sql #end sql
......
tests/system-test/7-tmq/tmqUpdate-1ctb.py
浏览文件 @ d0e83605
...@@ -17,7 +17,7 @@ from tmqCommon import * ...@@ -17,7 +17,7 @@ from tmqCommon import *
class TDTestCase: class TDTestCase:
def __init__(self): def __init__(self):
self.snapshot = 0 self.snapshot = 0
self.vgroups = 2 self.vgroups = 4
self.ctbNum = 1 self.ctbNum = 1
self.rowsPerTbl = 100000 self.rowsPerTbl = 100000
...@@ -235,18 +235,18 @@ class TDTestCase: ...@@ -235,18 +235,18 @@ class TDTestCase:
def run(self): def run(self):
tdSql.prepare() tdSql.prepare()
self.prepareTestEnv() # self.prepareTestEnv()
tdLog.printNoPrefix("=============================================") # tdLog.printNoPrefix("=============================================")
tdLog.printNoPrefix("======== snapshot is 0: only consume from wal") # tdLog.printNoPrefix("======== snapshot is 0: only consume from wal")
self.tmqCase1() # self.tmqCase1()
self.tmqCase2() # self.tmqCase2()
self.prepareTestEnv() self.prepareTestEnv()
tdLog.printNoPrefix("====================================================================") tdLog.printNoPrefix("====================================================================")
tdLog.printNoPrefix("======== snapshot is 1: firstly consume from tsbs, and then from wal") tdLog.printNoPrefix("======== snapshot is 1: firstly consume from tsbs, and then from wal")
self.snapshot = 1 self.snapshot = 1
self.tmqCase1() self.tmqCase1()
self.tmqCase2() # self.tmqCase2()
def stop(self): def stop(self):
......
tests/system-test/7-tmq/tmqUpdate-multiCtb.py
浏览文件 @ d0e83605
...@@ -17,9 +17,10 @@ from tmqCommon import * ...@@ -17,9 +17,10 @@ from tmqCommon import *
class TDTestCase: class TDTestCase:
def __init__(self): def __init__(self):
self.snapshot = 0 self.snapshot = 0
self.vgroups = 2 self.vgroups = 4
self.ctbNum = 1 self.ctbNum = 100
self.rowsPerTbl = 100000 self.rowsPerTbl = 1000
self.autoCtbPrefix = 'aCtb'
def init(self, conn, logSql): def init(self, conn, logSql):
tdLog.debug(f"start to excute {__file__}") tdLog.debug(f"start to excute {__file__}")
...@@ -38,9 +39,9 @@ class TDTestCase: ...@@ -38,9 +39,9 @@ class TDTestCase:
'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
'ctbPrefix': 'ctb', 'ctbPrefix': 'ctb',
'ctbStartIdx': 0, 'ctbStartIdx': 0,
'ctbNum': 1, 'ctbNum': 1000,
'rowsPerTbl': 100000, 'rowsPerTbl': 1000,
'batchNum': 1200, 'batchNum': 10000,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000 'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 3, 'pollDelay': 3,
'showMsg': 1, 'showMsg': 1,
...@@ -62,9 +63,9 @@ class TDTestCase: ...@@ -62,9 +63,9 @@ class TDTestCase:
tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
# tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix="ctbx", tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=self.autoCtbPrefix,
# ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
# startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
# tdLog.info("restart taosd to ensure that the data falls into the disk") # tdLog.info("restart taosd to ensure that the data falls into the disk")
# tdSql.query("flush database %s"%(paraDict['dbName'])) # tdSql.query("flush database %s"%(paraDict['dbName']))
...@@ -84,9 +85,9 @@ class TDTestCase: ...@@ -84,9 +85,9 @@ class TDTestCase:
'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
'ctbPrefix': 'ctb', 'ctbPrefix': 'ctb',
'ctbStartIdx': 0, 'ctbStartIdx': 0,
'ctbNum': 1, 'ctbNum': 1000,
'rowsPerTbl': 100000, 'rowsPerTbl': 1000,
'batchNum': 3000, 'batchNum': 10000,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000 'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 5, 'pollDelay': 5,
'showMsg': 1, 'showMsg': 1,
...@@ -98,10 +99,11 @@ class TDTestCase: ...@@ -98,10 +99,11 @@ class TDTestCase:
paraDict['rowsPerTbl'] = self.rowsPerTbl paraDict['rowsPerTbl'] = self.rowsPerTbl
# update to half tables # update to half tables
paraDict['ctbNum'] = int(self.ctbNum/2)
paraDict['rowsPerTbl'] = int(self.rowsPerTbl / 2) paraDict['rowsPerTbl'] = int(self.rowsPerTbl / 2)
# tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix="ctbx", tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=self.autoCtbPrefix,
# ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
# startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx'])
...@@ -113,10 +115,14 @@ class TDTestCase: ...@@ -113,10 +115,14 @@ class TDTestCase:
tdLog.info("create topic sql: %s"%sqlString) tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString) tdSql.execute(sqlString)
# paraDict['ctbNum'] = self.ctbNum paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl paraDict['rowsPerTbl'] = self.rowsPerTbl
consumerId = 0 consumerId = 0
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 3/2) if self.snapshot == 0:
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (2 + 1/2*1/2*2))
elif self.snapshot == 1:
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (2))
topicList = topicFromStb1 topicList = topicFromStb1
ifcheckdata = 1 ifcheckdata = 1
ifManualCommit = 1 ifManualCommit = 1
...@@ -129,7 +135,7 @@ class TDTestCase: ...@@ -129,7 +135,7 @@ class TDTestCase:
tdLog.info("start consume processor") tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])
tdLog.info("insert process end, and start to check consume result") tdLog.info("start to check consume result")
expectRows = 1 expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows) resultList = tmqCom.selectConsumeResult(expectRows)
totalConsumeRows = 0 totalConsumeRows = 0
...@@ -143,7 +149,7 @@ class TDTestCase: ...@@ -143,7 +149,7 @@ class TDTestCase:
if totalConsumeRows != expectrowcnt: if totalConsumeRows != expectrowcnt:
tdLog.exit("tmq consume rows error!") tdLog.exit("tmq consume rows error!")
tmqCom.checkFileContent(consumerId, queryString) # tmqCom.checkFileContent(consumerId, queryString)
tdSql.query("drop topic %s"%topicFromStb1) tdSql.query("drop topic %s"%topicFromStb1)
tdLog.printNoPrefix("======== test case 1 end ...... ") tdLog.printNoPrefix("======== test case 1 end ...... ")
...@@ -161,9 +167,9 @@ class TDTestCase: ...@@ -161,9 +167,9 @@ class TDTestCase:
'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}],
'ctbPrefix': 'ctb', 'ctbPrefix': 'ctb',
'ctbStartIdx': 0, 'ctbStartIdx': 0,
'ctbNum': 1, 'ctbNum': 1000,
'rowsPerTbl': 10000, 'rowsPerTbl': 1000,
'batchNum': 5000, 'batchNum': 10000,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000 'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 5, 'pollDelay': 5,
'showMsg': 1, 'showMsg': 1,
...@@ -179,14 +185,20 @@ class TDTestCase: ...@@ -179,14 +185,20 @@ class TDTestCase:
tdSql.query("flush database %s"%(paraDict['dbName'])) tdSql.query("flush database %s"%(paraDict['dbName']))
# update to half tables # update to half tables
paraDict['startTs'] = paraDict['startTs'] + int(self.rowsPerTbl / 2) paraDict['ctbNum'] = int(self.ctbNum/2)
paraDict['rowsPerTbl'] = int(self.rowsPerTbl / 2) paraDict['rowsPerTbl'] = int(self.rowsPerTbl / 2)
tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict["ctbPrefix"], paraDict['startTs'] = paraDict['startTs'] + int(self.rowsPerTbl / 2)
tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=self.autoCtbPrefix,
ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']+int(self.ctbNum/2))
# tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
# ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix="aCtby",
# startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']+int(self.ctbNum/2))
tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']+int(self.ctbNum/2))
tmqCom.initConsumerTable() tmqCom.initConsumerTable()
tdLog.info("create topics from stb1") tdLog.info("create topics from stb1")
...@@ -197,9 +209,14 @@ class TDTestCase: ...@@ -197,9 +209,14 @@ class TDTestCase:
tdSql.execute(sqlString) tdSql.execute(sqlString)
# paraDict['ctbNum'] = self.ctbNum # paraDict['ctbNum'] = self.ctbNum
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl paraDict['rowsPerTbl'] = self.rowsPerTbl
consumerId = 1 consumerId = 1
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2) if self.snapshot == 0:
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (2 + 1/2*1/2*2 + 1/2*1/2))
elif self.snapshot == 1:
expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (2 + 1/2*1/2))
topicList = topicFromStb1 topicList = topicFromStb1
ifcheckdata = 1 ifcheckdata = 1
ifManualCommit = 1 ifManualCommit = 1
...@@ -212,7 +229,7 @@ class TDTestCase: ...@@ -212,7 +229,7 @@ class TDTestCase:
tdLog.info("start consume processor") tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])
tdLog.info("insert process end, and start to check consume result") tdLog.info("start to check consume result")
expectRows = 1 expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows) resultList = tmqCom.selectConsumeResult(expectRows)
totalConsumeRows = 0 totalConsumeRows = 0
......
tests/system-test/fulltest.sh
浏览文件 @ d0e83605
...@@ -11,7 +11,7 @@ python3 ./test.py -f 0-others/udfTest.py ...@@ -11,7 +11,7 @@ python3 ./test.py -f 0-others/udfTest.py
python3 ./test.py -f 0-others/udf_create.py python3 ./test.py -f 0-others/udf_create.py
python3 ./test.py -f 0-others/udf_restart_taosd.py python3 ./test.py -f 0-others/udf_restart_taosd.py
python3 ./test.py -f 0-others/cachelast.py python3 ./test.py -f 0-others/cachelast.py
python3 ./test.py -f 0-others/sysinfo.py
python3 ./test.py -f 0-others/user_control.py python3 ./test.py -f 0-others/user_control.py
python3 ./test.py -f 0-others/fsync.py python3 ./test.py -f 0-others/fsync.py
...@@ -178,6 +178,8 @@ python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py ...@@ -178,6 +178,8 @@ python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py
python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb.py python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb.py
python3 ./test.py -f 7-tmq/tmqAutoCreateTbl.py python3 ./test.py -f 7-tmq/tmqAutoCreateTbl.py
#python3 ./test.py -f 7-tmq/tmqDnodeRestart.py #python3 ./test.py -f 7-tmq/tmqDnodeRestart.py
#python3 ./test.py -f 7-tmq/tmqUpdate-1ctb.py
python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb.py
#------------querPolicy 2----------- #------------querPolicy 2-----------
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或