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

Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/row_refact

This commit is contained in:
Hongze Cheng 2022-06-08 01:58:10 +00:00
parent e907d58d78 6a5604fca6
commit 70c819ae76
9 changed files with 59 additions and 325 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
include/libs/stream/tstream.h
View File

@ -131,14 +131,7 @@ static FORCE_INLINE void* streamQueueNextItem(SStreamQueue* queue) {
SStreamDataSubmit* streamDataSubmitNew(SSubmitReq* pReq); SStreamDataSubmit* streamDataSubmitNew(SSubmitReq* pReq);
static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit) { void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit);
int32_t ref = atomic_sub_fetch_32(pDataSubmit->dataRef, 1);
ASSERT(ref >= 0);
if (ref == 0) {
taosMemoryFree(pDataSubmit->data);
taosMemoryFree(pDataSubmit->dataRef);
}
}
SStreamDataSubmit* streamSubmitRefClone(SStreamDataSubmit* pSubmit); SStreamDataSubmit* streamSubmitRefClone(SStreamDataSubmit* pSubmit);
@ -189,6 +182,7 @@ typedef void FSmaSink(void* vnode, int64_t smaId, const SArray* data);
typedef struct { typedef struct {
int64_t smaId; int64_t smaId;
// following are not applicable to encoder and decoder // following are not applicable to encoder and decoder
void* vnode;
FSmaSink* smaSink; FSmaSink* smaSink;
} STaskSinkSma; } STaskSinkSma;
@ -270,7 +264,7 @@ struct SStreamTask {
SStreamQueue* outputQueue; SStreamQueue* outputQueue;
// application storage // application storage
void* ahandle; // void* ahandle;
}; };
SStreamTask* tNewSStreamTask(int64_t streamId, int32_t childId); SStreamTask* tNewSStreamTask(int64_t streamId, int32_t childId);
@ -311,7 +305,16 @@ static FORCE_INLINE void streamTaskInputFail(SStreamTask* pTask) {
} }
static FORCE_INLINE int32_t streamTaskOutput(SStreamTask* pTask, SStreamDataBlock* pBlock) { static FORCE_INLINE int32_t streamTaskOutput(SStreamTask* pTask, SStreamDataBlock* pBlock) {
taosWriteQitem(pTask->outputQueue->queue, pBlock); if (pTask->sinkType == TASK_SINK__TABLE) {
ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE);
pTask->tbSink.tbSinkFunc(pTask, pTask->tbSink.vnode, 0, pBlock->blocks);
} else if (pTask->sinkType == TASK_SINK__SMA) {
ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE);
pTask->smaSink.smaSink(pTask->smaSink.vnode, pTask->smaSink.smaId, pBlock->blocks);
} else {
ASSERT(pTask->dispatchType != TASK_DISPATCH__NONE);
taosWriteQitem(pTask->outputQueue->queue, pBlock);
}
return 0; return 0;
} }
@ -319,26 +322,11 @@ typedef struct {
int32_t reserved; int32_t reserved;
} SStreamTaskDeployRsp; } SStreamTaskDeployRsp;
typedef struct {
// SMsgHead head;
int64_t streamId;
int32_t taskId;
SArray* data; // SArray<SSDataBlock>
} SStreamTaskExecReq;
typedef struct { typedef struct {
// SMsgHead head; // SMsgHead head;
SStreamTask* task; SStreamTask* task;
} SStreamTaskDeployReq; } SStreamTaskDeployReq;
int32_t tEncodeSStreamTaskExecReq(void** buf, const SStreamTaskExecReq* pReq);
void* tDecodeSStreamTaskExecReq(const void* buf, SStreamTaskExecReq* pReq);
void tFreeSStreamTaskExecReq(SStreamTaskExecReq* pReq);
typedef struct {
int32_t reserved;
} SStreamTaskExecRsp;
typedef struct { typedef struct {
SMsgHead head; SMsgHead head;
int64_t streamId; int64_t streamId;

3
source/dnode/vnode/src/tq/tq.c
View File

@ -349,8 +349,9 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
} }
// sink // sink
pTask->ahandle = pTq->pVnode; /*pTask->ahandle = pTq->pVnode;*/
if (pTask->sinkType == TASK_SINK__SMA) { if (pTask->sinkType == TASK_SINK__SMA) {
pTask->smaSink.vnode = pTq->pVnode;
pTask->smaSink.smaSink = smaHandleRes; pTask->smaSink.smaSink = smaHandleRes;
} else if (pTask->sinkType == TASK_SINK__TABLE) { } else if (pTask->sinkType == TASK_SINK__TABLE) {
pTask->tbSink.vnode = pTq->pVnode; pTask->tbSink.vnode = pTq->pVnode;

14
source/libs/executor/src/timewindowoperator.c
View File

@ -1,3 +1,17 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "executorimpl.h" #include "executorimpl.h"
#include "function.h" #include "function.h"
#include "functionMgt.h" #include "functionMgt.h"

5
source/libs/stream/inc/streamInc.h
View File

@ -24,10 +24,9 @@ extern "C" {
#endif #endif
int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb); int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb);
int32_t streamSink1(SStreamTask* pTask, SMsgCb* pMsgCb); int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb);
int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDataBlock* data);
int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock* pData); int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock* pData);
int32_t streamBuildDispatchMsg(SStreamTask* pTask, SStreamDataBlock* data, SRpcMsg* pMsg, SEpSet** ppEpSet);
#ifdef __cplusplus #ifdef __cplusplus
} }

12
source/libs/stream/src/stream.c
View File

@ -83,7 +83,9 @@ int32_t streamProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDisp
// 3. handle output // 3. handle output
// 3.1 check and set status // 3.1 check and set status
// 3.2 dispatch / sink // 3.2 dispatch / sink
streamSink1(pTask, pMsgCb); if (pTask->dispatchType != TASK_DISPATCH__NONE) {
streamDispatch(pTask, pMsgCb);
}
return 0; return 0;
} }
@ -97,13 +99,17 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDisp
return 0; return 0;
} }
// continue dispatch // continue dispatch
streamSink1(pTask, pMsgCb); if (pTask->dispatchType != TASK_DISPATCH__NONE) {
streamDispatch(pTask, pMsgCb);
}
return 0; return 0;
} }
int32_t streamTaskProcessRunReq(SStreamTask* pTask, SMsgCb* pMsgCb) { int32_t streamTaskProcessRunReq(SStreamTask* pTask, SMsgCb* pMsgCb) {
streamExec(pTask, pMsgCb); streamExec(pTask, pMsgCb);
streamSink1(pTask, pMsgCb); if (pTask->dispatchType != TASK_DISPATCH__NONE) {
streamDispatch(pTask, pMsgCb);
}
return 0; return 0;
} }

19
source/libs/stream/src/streamData.c
View File

@ -86,22 +86,3 @@ SStreamDataSubmit* streamSubmitRefClone(SStreamDataSubmit* pSubmit) {
memcpy(pSubmitClone, pSubmit, sizeof(SStreamDataSubmit)); memcpy(pSubmitClone, pSubmit, sizeof(SStreamDataSubmit));
return pSubmitClone; return pSubmitClone;
} }
#if 0
int32_t tEncodeSStreamTaskExecReq(void** buf, const SStreamTaskExecReq* pReq) {
int32_t tlen = 0;
tlen += taosEncodeFixedI64(buf, pReq->streamId);
tlen += taosEncodeFixedI32(buf, pReq->taskId);
tlen += tEncodeDataBlocks(buf, pReq->data);
return tlen;
}
void* tDecodeSStreamTaskExecReq(const void* buf, SStreamTaskExecReq* pReq) {
buf = taosDecodeFixedI64(buf, &pReq->streamId);
buf = taosDecodeFixedI32(buf, &pReq->taskId);
buf = tDecodeDataBlocks(buf, &pReq->data);
return (void*)buf;
}
void tFreeSStreamTaskExecReq(SStreamTaskExecReq* pReq) { taosArrayDestroy(pReq->data); }
#endif

125
source/libs/stream/src/streamMsg.c → source/libs/stream/src/streamDispatch.c
View File

@ -174,131 +174,26 @@ FAIL:
return code; return code;
} }
int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDataBlock* data) { int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb) {
#if 0 #if 1
int8_t old = int8_t old =
atomic_val_compare_exchange_8(&pTask->outputStatus, TASK_OUTPUT_STATUS__NORMAL, TASK_OUTPUT_STATUS__WAIT); atomic_val_compare_exchange_8(&pTask->outputStatus, TASK_OUTPUT_STATUS__NORMAL, TASK_OUTPUT_STATUS__WAIT);
if (old != TASK_OUTPUT_STATUS__NORMAL) { if (old != TASK_OUTPUT_STATUS__NORMAL) {
return 0; return 0;
} }
#endif #endif
if (pTask->dispatchType == TASK_DISPATCH__INPLACE) {
SRpcMsg dispatchMsg = {0};
if (streamBuildDispatchMsg(pTask, data, &dispatchMsg, NULL) < 0) {
ASSERT(0);
return -1;
}
int32_t qType; SStreamDataBlock* pBlock = streamQueueNextItem(pTask->outputQueue);
if (pTask->dispatchMsgType == TDMT_STREAM_TASK_DISPATCH) { if (pBlock == NULL) return 0;
qType = FETCH_QUEUE; ASSERT(pBlock->type == STREAM_DATA_TYPE_SSDATA_BLOCK);
} else if (pTask->dispatchMsgType == TDMT_VND_STREAM_DISPATCH_WRITE) {
qType = WRITE_QUEUE;
} else {
ASSERT(0);
}
tmsgPutToQueue(pMsgCb, qType, &dispatchMsg);
} else if (pTask->dispatchType == TASK_DISPATCH__FIXED) {
SRpcMsg dispatchMsg = {0};
SEpSet* pEpSet = NULL;
if (streamBuildDispatchMsg(pTask, data, &dispatchMsg, &pEpSet) < 0) {
ASSERT(0);
return -1;
}
tmsgSendReq(pEpSet, &dispatchMsg); SRpcMsg dispatchMsg = {0};
} else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) { SEpSet* pEpSet = NULL;
SRpcMsg dispatchMsg = {0}; if (streamBuildDispatchMsg(pTask, pBlock, &dispatchMsg, &pEpSet) < 0) {
SEpSet* pEpSet = NULL; ASSERT(0);
if (streamBuildDispatchMsg(pTask, data, &dispatchMsg, &pEpSet) < 0) {
ASSERT(0);
return -1;
}
tmsgSendReq(pEpSet, &dispatchMsg);
}
return 0;
}
#if 0
static int32_t streamBuildExecMsg(SStreamTask* pTask, SArray* data, SRpcMsg* pMsg, SEpSet** ppEpSet) {
SStreamTaskExecReq req = {
.streamId = pTask->streamId,
.data = data,
};
int32_t tlen = sizeof(SMsgHead) + tEncodeSStreamTaskExecReq(NULL, &req);
void* buf = rpcMallocCont(tlen);
if (buf == NULL) {
return -1; return -1;
} }
if (pTask->dispatchType == TASK_DISPATCH__INPLACE) { tmsgSendReq(pEpSet, &dispatchMsg);
((SMsgHead*)buf)->vgId = 0;
req.taskId = pTask->inplaceDispatcher.taskId;
} else if (pTask->dispatchType == TASK_DISPATCH__FIXED) {
((SMsgHead*)buf)->vgId = htonl(pTask->fixedEpDispatcher.nodeId);
*ppEpSet = &pTask->fixedEpDispatcher.epSet;
req.taskId = pTask->fixedEpDispatcher.taskId;
} else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) {
// TODO use general name rule of schemaless
char ctbName[TSDB_TABLE_FNAME_LEN + 22] = {0};
// all groupId must be the same in an array
SSDataBlock* pBlock = taosArrayGet(data, 0);
sprintf(ctbName, "%s:%ld", pTask->shuffleDispatcher.stbFullName, pBlock->info.groupId);
// TODO: get hash function by hashMethod
// get groupId, compute hash value
uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));
// get node
// TODO: optimize search process
SArray* vgInfo = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
int32_t sz = taosArrayGetSize(vgInfo);
int32_t nodeId = 0;
for (int32_t i = 0; i < sz; i++) {
SVgroupInfo* pVgInfo = taosArrayGet(vgInfo, i);
if (hashValue >= pVgInfo->hashBegin && hashValue <= pVgInfo->hashEnd) {
nodeId = pVgInfo->vgId;
req.taskId = pVgInfo->taskId;
*ppEpSet = &pVgInfo->epSet;
break;
}
}
ASSERT(nodeId != 0);
((SMsgHead*)buf)->vgId = htonl(nodeId);
}
void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
tEncodeSStreamTaskExecReq(&abuf, &req);
pMsg->pCont = buf;
pMsg->contLen = tlen;
pMsg->code = 0;
pMsg->msgType = pTask->dispatchMsgType;
pMsg->info.noResp = 1;
return 0; return 0;
} }
static int32_t streamShuffleDispatch(SStreamTask* pTask, SMsgCb* pMsgCb, SHashObj* data) {
void* pIter = NULL;
while (1) {
pIter = taosHashIterate(data, pIter);
if (pIter == NULL) return 0;
SArray* pData = *(SArray**)pIter;
SRpcMsg dispatchMsg = {0};
SEpSet* pEpSet;
if (streamBuildExecMsg(pTask, pData, &dispatchMsg, &pEpSet) < 0) {
ASSERT(0);
return -1;
}
tmsgSendReq(pEpSet, &dispatchMsg);
}
return 0;
}
#endif

9
source/libs/stream/src/streamQueue.c
View File

@ -41,3 +41,12 @@ void streamQueueClose(SStreamQueue* queue) {
return; return;
} }
} }
void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit) {
int32_t ref = atomic_sub_fetch_32(pDataSubmit->dataRef, 1);
ASSERT(ref >= 0);
if (ref == 0) {
taosMemoryFree(pDataSubmit->data);
taosMemoryFree(pDataSubmit->dataRef);
}
}

159
source/libs/stream/src/streamSink.c
View File

@ -1,159 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "streamInc.h"
int32_t streamSink1(SStreamTask* pTask, SMsgCb* pMsgCb) {
SStreamQueue* queue;
if (pTask->execType == TASK_EXEC__NONE) {
queue = pTask->inputQueue;
} else {
queue = pTask->outputQueue;
}
/*if (streamDequeueBegin(queue) == true) {*/
/*return -1;*/
/*}*/
if (pTask->sinkType == TASK_SINK__TABLE || pTask->sinkType == TASK_SINK__SMA ||
pTask->dispatchType != TASK_DISPATCH__NONE) {
while (1) {
SStreamDataBlock* pBlock = streamQueueNextItem(queue);
if (pBlock == NULL) break;
ASSERT(pBlock->type == STREAM_DATA_TYPE_SSDATA_BLOCK);
// local sink
if (pTask->sinkType == TASK_SINK__TABLE) {
ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE);
pTask->tbSink.tbSinkFunc(pTask, pTask->tbSink.vnode, 0, pBlock->blocks);
} else if (pTask->sinkType == TASK_SINK__SMA) {
ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE);
pTask->smaSink.smaSink(pTask->ahandle, pTask->smaSink.smaId, pBlock->blocks);
}
// TODO: sink and dispatch should be only one
if (pTask->dispatchType != TASK_DISPATCH__NONE) {
ASSERT(queue == pTask->outputQueue);
ASSERT(pTask->sinkType == TASK_SINK__NONE);
streamDispatch(pTask, pMsgCb, pBlock);
}
streamQueueProcessSuccess(queue);
}
}
return 0;
}
#if 0
int32_t streamSink(SStreamTask* pTask, SMsgCb* pMsgCb) {
bool firstRun = 1;
while (1) {
SStreamDataBlock* pBlock = NULL;
if (!firstRun) {
taosReadAllQitems(pTask->outputQ, pTask->outputQAll);
}
taosGetQitem(pTask->outputQAll, (void**)&pBlock);
if (pBlock == NULL) {
if (firstRun) {
firstRun = 0;
continue;
} else {
break;
}
}
SArray* pRes = pBlock->blocks;
// sink
if (pTask->sinkType == TASK_SINK__TABLE) {
// blockDebugShowData(pRes);
pTask->tbSink.tbSinkFunc(pTask, pTask->tbSink.vnode, 0, pRes);
} else if (pTask->sinkType == TASK_SINK__SMA) {
pTask->smaSink.smaSink(pTask->ahandle, pTask->smaSink.smaId, pRes);
//
} else if (pTask->sinkType == TASK_SINK__FETCH) {
//
} else {
ASSERT(pTask->sinkType == TASK_SINK__NONE);
}
// dispatch
// TODO dispatch guard
int8_t outputStatus = atomic_load_8(&pTask->outputStatus);
if (outputStatus == TASK_OUTPUT_STATUS__NORMAL) {
if (pTask->dispatchType == TASK_DISPATCH__INPLACE) {
SRpcMsg dispatchMsg = {0};
if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, NULL) < 0) {
ASSERT(0);
return -1;
}
int32_t qType;
if (pTask->dispatchMsgType == TDMT_VND_TASK_DISPATCH || pTask->dispatchMsgType == TDMT_SND_TASK_DISPATCH) {
qType = FETCH_QUEUE;
/*} else if (pTask->dispatchMsgType == TDMT_VND_TASK_MERGE_EXEC ||*/
/*pTask->dispatchMsgType == TDMT_SND_TASK_MERGE_EXEC) {*/
/*qType = MERGE_QUEUE;*/
/*} else if (pTask->dispatchMsgType == TDMT_VND_TASK_WRITE_EXEC) {*/
/*qType = WRITE_QUEUE;*/
} else {
ASSERT(0);
}
tmsgPutToQueue(pMsgCb, qType, &dispatchMsg);
} else if (pTask->dispatchType == TASK_DISPATCH__FIXED) {
SRpcMsg dispatchMsg = {0};
SEpSet* pEpSet = NULL;
if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, &pEpSet) < 0) {
ASSERT(0);
return -1;
}
tmsgSendReq(pEpSet, &dispatchMsg);
} else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) {
SHashObj* pShuffleRes = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
if (pShuffleRes == NULL) {
return -1;
}
int32_t sz = taosArrayGetSize(pRes);
for (int32_t i = 0; i < sz; i++) {
SSDataBlock* pDataBlock = taosArrayGet(pRes, i);
SArray* pArray = taosHashGet(pShuffleRes, &pDataBlock->info.groupId, sizeof(int64_t));
if (pArray == NULL) {
pArray = taosArrayInit(0, sizeof(SSDataBlock));
if (pArray == NULL) {
return -1;
}
taosHashPut(pShuffleRes, &pDataBlock->info.groupId, sizeof(int64_t), &pArray, sizeof(void*));
}
taosArrayPush(pArray, pDataBlock);
}
if (streamShuffleDispatch(pTask, pMsgCb, pShuffleRes) < 0) {
return -1;
}
} else {
ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE);
}
}
}
return 0;
}
#endif