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

Merge pull request #13676 from taosdata/feature/stream 

fix(stream): set child id
This commit is contained in:
Liu Jicong 2022-06-10 12:57:34 +08:00 committed by GitHub
parent 3ae24b38a0 73b8f02ac7
commit e7f3d74187
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 14 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -141,8 +141,7 @@ void* streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput);
#endif
typedef struct {
int8_t parallelizable;
char* qmsg;
char* qmsg;
// followings are not applicable to encoder and decoder
void* inputHandle;
void* executor;
@ -267,7 +266,7 @@ struct SStreamTask {
// void* ahandle;
};
SStreamTask* tNewSStreamTask(int64_t streamId, int32_t childId);
SStreamTask* tNewSStreamTask(int64_t streamId);
int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask);
int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask);
void tFreeSStreamTask(SStreamTask* pTask);

19
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -197,7 +197,7 @@ int32_t mndAddShuffledSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* p
sdbRelease(pSdb, pVgroup);
continue;
}
SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0);
SStreamTask* pTask = tNewSStreamTask(pStream->uid);
if (pTask == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
@ -237,7 +237,7 @@ int32_t mndAddShuffledSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* p
int32_t mndAddFixedSinkToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
ASSERT(pStream->fixedSinkVgId != 0);
SArray* tasks = taosArrayGetP(pStream->tasks, 0);
SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0);
SStreamTask* pTask = tNewSStreamTask(pStream->uid);
if (pTask == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
@ -329,7 +329,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
sdbRelease(pSdb, pVgroup);
continue;
}
SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0);
SStreamTask* pTask = tNewSStreamTask(pStream->uid);
mndAddTaskToTaskSet(taskOneLevel, pTask);
// source part
pTask->sourceType = TASK_SOURCE__SCAN;
pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK;
@ -378,14 +379,12 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
// exec part
pTask->execType = TASK_EXEC__PIPE;
pTask->exec.parallelizable = 1;
if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) {
sdbRelease(pSdb, pVgroup);
qDestroyQueryPlan(pPlan);
return -1;
}
sdbRelease(pSdb, pVgroup);
mndAddTaskToTaskSet(taskOneLevel, pTask);
}
} else {
// merge plan
@ -394,7 +393,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
// else, assign to vnode
ASSERT(plan->subplanType == SUBPLAN_TYPE_MERGE);
SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0);
SStreamTask* pTask = tNewSStreamTask(pStream->uid);
mndAddTaskToTaskSet(taskOneLevel, pTask);
// source part, currently only support multi source
pTask->sourceType = TASK_SOURCE__PIPE;
@ -456,7 +456,6 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
// exec part
pTask->execType = TASK_EXEC__MERGE;
pTask->exec.parallelizable = 0;
SVgObj* pVgroup = mndSchedFetchOneVg(pMnode, pStream->dbUid);
ASSERT(pVgroup);
if (mndAssignTaskToVg(pMnode, pTrans, pTask, plan, pVgroup) < 0) {
@ -465,12 +464,12 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
return -1;
}
sdbRelease(pSdb, pVgroup);
taosArrayPush(taskOneLevel, &pTask);
}
taosArrayPush(pStream->tasks, &taskOneLevel);
}
#if 0
if (totLevel == 2) {
void* pIter = NULL;
while (1) {
@ -481,7 +480,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
sdbRelease(pSdb, pVgroup);
continue;
}
SStreamTask* pTask = tNewSStreamTask(pStream->uid, 0);
SStreamTask* pTask = tNewSStreamTask(pStream->uid);
// source part
pTask->sourceType = TASK_SOURCE__MERGE;
@ -495,9 +494,9 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
// exec part
pTask->execType = TASK_EXEC__NONE;
pTask->exec.parallelizable = 0;
}
}
#endif
// free memory
qDestroyQueryPlan(pPlan);

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

@ -364,6 +364,7 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
tdGetSTSChemaFromSSChema(&pTask->tbSink.pSchemaWrapper->pSchema, pTask->tbSink.pSchemaWrapper->nCols);
ASSERT(pTask->tbSink.pTSchema);
}
tqInfo("deploy stream task id %d child id %d on vg %d", pTask->taskId, pTask->childId, pTq->pVnode->config.vgId);
taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), pTask, sizeof(SStreamTask));

1
source/libs/stream/src/streamDispatch.c
View File

@ -100,6 +100,7 @@ int32_t streamBuildDispatchMsg(SStreamTask* pTask, SStreamDataBlock* data, SRpcM
.upstreamNodeId = pTask->nodeId,
.blockNum = blockNum,
};
qInfo("dispatch from task %d (child id %d)", pTask->taskId, pTask->childId);
req.data = taosArrayInit(blockNum, sizeof(void*));
req.dataLen = taosArrayInit(blockNum, sizeof(int32_t));

5
source/libs/stream/src/streamTask.c
View File

@ -16,14 +16,13 @@
#include "executor.h"
#include "tstream.h"
SStreamTask* tNewSStreamTask(int64_t streamId, int32_t childId) {
SStreamTask* tNewSStreamTask(int64_t streamId) {
SStreamTask* pTask = (SStreamTask*)taosMemoryCalloc(1, sizeof(SStreamTask));
if (pTask == NULL) {
return NULL;
}
pTask->taskId = tGenIdPI32();
pTask->streamId = streamId;
pTask->childId = childId;
pTask->status = TASK_STATUS__IDLE;
pTask->inputStatus = TASK_INPUT_STATUS__NORMAL;
pTask->outputStatus = TASK_OUTPUT_STATUS__NORMAL;
@ -48,7 +47,6 @@ int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {
if (tEncodeSEpSet(pEncoder, &pTask->epSet) < 0) return -1;
if (pTask->execType != TASK_EXEC__NONE) {
if (tEncodeI8(pEncoder, pTask->exec.parallelizable) < 0) return -1;
if (tEncodeCStr(pEncoder, pTask->exec.qmsg) < 0) return -1;
}
@ -96,7 +94,6 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
if (tDecodeSEpSet(pDecoder, &pTask->epSet) < 0) return -1;
if (pTask->execType != TASK_EXEC__NONE) {
if (tDecodeI8(pDecoder, &pTask->exec.parallelizable) < 0) return -1;
if (tDecodeCStrAlloc(pDecoder, &pTask->exec.qmsg) < 0) return -1;
}