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

enh(stream): sm for stream tasks.

This commit is contained in:
Haojun Liao 2023-10-17 13:47:55 +08:00
parent 7b7ce10475
commit 64ed7e49a2
2 changed files with 19 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/stream/inc/streamsm.h
View File

@ -55,6 +55,7 @@ struct SStreamTaskSM {
SArray* pTransList; // SArray<STaskStateTrans> SArray* pTransList; // SArray<STaskStateTrans>
int64_t stateTs; int64_t stateTs;
SStreamTask* pTask; SStreamTask* pTask;
STaskStateTrans* pActiveTrans;
}; };
typedef struct SStreamEventInfo { typedef struct SStreamEventInfo {

23
source/libs/stream/src/streamTaskSm.c
View File

@ -38,6 +38,7 @@ static int32_t initStateTransferTable(SStreamTaskSM* pSM);
static int32_t dummyFn(SStreamTask* UNUSED_PARAM(p)) { return 0; } static int32_t dummyFn(SStreamTask* UNUSED_PARAM(p)) { return 0; }
static int32_t streamTaskStartCheckDownstream(SStreamTask* pTask) { static int32_t streamTaskStartCheckDownstream(SStreamTask* pTask) {
stDebug("s-task:%s start to check downstream tasks", pTask->id.idStr); stDebug("s-task:%s start to check downstream tasks", pTask->id.idStr);
streamTaskCheckDownstream(pTask);
return 0; return 0;
} }
static int32_t streamTaskDoPause(SStreamTask* pTask) { static int32_t streamTaskDoPause(SStreamTask* pTask) {
@ -84,21 +85,33 @@ SStreamTaskSM* streamCreateStateMachine(SStreamTask* pTask) {
taosMemoryFree(pSM); taosMemoryFree(pSM);
return NULL; return NULL;
} }
return pSM; return pSM;
} }
int32_t taskSMHandleEvent(SStreamTaskSM* pSM, const EStreamTaskEvent* pEvent) { int32_t taskSMHandleEvent(SStreamTaskSM* pSM, const EStreamTaskEvent* pEvent) {
STaskStateTrans* pTrans = streamTaskFindTransform(pSM, pEvent); STaskStateTrans* pTrans = streamTaskFindTransform(pSM, pEvent);
qDebug("start to handle event:%d", *pEvent); ASSERT(pSM->pActiveTrans == NULL);
pSM->current = pTrans->next; stDebug("start to handle event:%d, state:%s", *pEvent, pSM->current.name);
pSM->pActiveTrans = pTrans;
pSM->stateTs = taosGetTimestampMs(); pSM->stateTs = taosGetTimestampMs();
qDebug("new state:%s from %s", pTrans->next.name, pSM->current.name);
return pTrans->pAction(pSM->pTask); return pTrans->pAction(pSM->pTask);
} }
int32_t taskSMOnHandleEventSuccess(SStreamTaskSM* pSM) {
STaskStateTrans* pTrans = pSM->pActiveTrans;
EStreamTaskEvent* pEvent = &pTrans->event;
int64_t el = (taosGetTimestampMs() - pSM->stateTs);
stDebug("handle event:%d completed, elapsd time:%" PRId64 "ms new state:%s from %s", *pEvent, el, pTrans->next.name,
pSM->current.name);
// todo: add lock
pSM->current = pTrans->next;
}
STaskStateTrans createStateTransform(ETaskStatus current, ETaskStatus next, EStreamTaskEvent event, __state_trans_fn fn) { STaskStateTrans createStateTransform(ETaskStatus current, ETaskStatus next, EStreamTaskEvent event, __state_trans_fn fn) {
STaskStateTrans trans = {0}; STaskStateTrans trans = {0};
trans.state = StreamTaskStatusList[current]; trans.state = StreamTaskStatusList[current];