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

feature/qnode

This commit is contained in:
dapan1121 2022-01-12 09:22:48 +08:00
parent 6677dfba94
commit 876942b4f2
3 changed files with 75 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
source/libs/scheduler/inc/schedulerInt.h
View File

@ -36,11 +36,31 @@ enum {
SCH_WRITE, SCH_WRITE,
}; };
typedef struct SSchApiStat {
} SSchApiStat;
typedef struct SSchRuntimeStat {
} SSchRuntimeStat;
typedef struct SSchJobStat {
} SSchJobStat;
typedef struct SSchedulerStat {
SSchApiStat api;
SSchRuntimeStat runtime;
SSchJobStat job;
} SSchedulerStat;
typedef struct SSchedulerMgmt { typedef struct SSchedulerMgmt {
uint64_t taskId; // sequential taksId uint64_t taskId; // sequential taksId
uint64_t sId; // schedulerId uint64_t sId; // schedulerId
SSchedulerCfg cfg; SSchedulerCfg cfg;
SHashObj *jobs; // key: queryId, value: SQueryJob* SHashObj *jobs; // key: queryId, value: SQueryJob*
SSchedulerStat stat;
} SSchedulerMgmt; } SSchedulerMgmt;
typedef struct SSchCallbackParam { typedef struct SSchCallbackParam {

55
source/libs/scheduler/src/scheduler.c
View File

@ -1462,35 +1462,38 @@ void scheduleFreeJob(void *job) {
} }
SSchJob *pJob = job; SSchJob *pJob = job;
uint64_t queryId = pJob->queryId;
if (0 != taosHashRemove(schMgmt.jobs, &pJob->queryId, sizeof(pJob->queryId))) { if (SCH_GET_JOB_STATUS(pJob) > 0) {
SCH_JOB_ELOG("taosHashRemove job from list failed, may already freed, pJob:%p", pJob); if (0 != taosHashRemove(schMgmt.jobs, &pJob->queryId, sizeof(pJob->queryId))) {
return; SCH_JOB_ELOG("taosHashRemove job from list failed, may already freed, pJob:%p", pJob);
} return;
schCheckAndUpdateJobStatus(pJob, JOB_TASK_STATUS_DROPPING);
SCH_JOB_DLOG("job removed from list, no further ref, ref:%d", atomic_load_32(&pJob->ref));
while (true) {
int32_t ref = atomic_load_32(&pJob->ref);
if (0 == ref) {
break;
} else if (ref > 0) {
usleep(1);
} else {
assert(0);
} }
schCheckAndUpdateJobStatus(pJob, JOB_TASK_STATUS_DROPPING);
SCH_JOB_DLOG("job removed from list, no further ref, ref:%d", atomic_load_32(&pJob->ref));
while (true) {
int32_t ref = atomic_load_32(&pJob->ref);
if (0 == ref) {
break;
} else if (ref > 0) {
usleep(1);
} else {
assert(0);
}
}
SCH_JOB_DLOG("job no ref now, status:%d", SCH_GET_JOB_STATUS(pJob));
if (pJob->status == JOB_TASK_STATUS_EXECUTING) {
schCancelJob(pJob);
}
schDropJobAllTasks(pJob);
} }
SCH_JOB_DLOG("job no ref now, status:%d", SCH_GET_JOB_STATUS(pJob));
if (pJob->status == JOB_TASK_STATUS_EXECUTING) {
schCancelJob(pJob);
}
schDropJobAllTasks(pJob);
pJob->subPlans = NULL; // it is a reference to pDag->pSubplans pJob->subPlans = NULL; // it is a reference to pDag->pSubplans
int32_t numOfLevels = taosArrayGetSize(pJob->levels); int32_t numOfLevels = taosArrayGetSize(pJob->levels);
@ -1515,6 +1518,8 @@ void scheduleFreeJob(void *job) {
tfree(pJob->res); tfree(pJob->res);
tfree(pJob); tfree(pJob);
qDebug("QID:%"PRIx64" job freed", queryId);
} }
void schedulerDestroy(void) { void schedulerDestroy(void) {

21
source/libs/scheduler/test/schedulerTests.cpp
View File

@ -79,6 +79,7 @@ void schtBuildQueryDag(SQueryDag *dag) {
scanPlan->level = 1; scanPlan->level = 1;
scanPlan->pParents = taosArrayInit(1, POINTER_BYTES); scanPlan->pParents = taosArrayInit(1, POINTER_BYTES);
scanPlan->pNode = (SPhyNode*)calloc(1, sizeof(SPhyNode)); scanPlan->pNode = (SPhyNode*)calloc(1, sizeof(SPhyNode));
scanPlan->msgType = TDMT_VND_QUERY;
mergePlan->id.queryId = qId; mergePlan->id.queryId = qId;
mergePlan->id.templateId = 0x4444444444; mergePlan->id.templateId = 0x4444444444;
@ -89,6 +90,7 @@ void schtBuildQueryDag(SQueryDag *dag) {
mergePlan->pChildren = taosArrayInit(1, POINTER_BYTES); mergePlan->pChildren = taosArrayInit(1, POINTER_BYTES);
mergePlan->pParents = NULL; mergePlan->pParents = NULL;
mergePlan->pNode = (SPhyNode*)calloc(1, sizeof(SPhyNode)); mergePlan->pNode = (SPhyNode*)calloc(1, sizeof(SPhyNode));
mergePlan->msgType = TDMT_VND_QUERY;
SSubplan *mergePointer = (SSubplan *)taosArrayPush(merge, &mergePlan); SSubplan *mergePointer = (SSubplan *)taosArrayPush(merge, &mergePlan);
SSubplan *scanPointer = (SSubplan *)taosArrayPush(scan, &scanPlan); SSubplan *scanPointer = (SSubplan *)taosArrayPush(scan, &scanPlan);
@ -163,6 +165,11 @@ void schtExecNode(SSubplan* subplan, uint64_t templateId, SQueryNodeAddr* ep) {
} }
void schtRpcSendRequest(void *shandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t *pRid) {
}
void schtSetPlanToString() { void schtSetPlanToString() {
static Stub stub; static Stub stub;
@ -190,6 +197,20 @@ void schtSetExecNode() {
} }
} }
void schtSetRpcSendRequest() {
static Stub stub;
stub.set(rpcSendRequest, schtRpcSendRequest);
{
AddrAny any("libtransport.so");
std::map<std::string,void*> result;
any.get_global_func_addr_dynsym("^rpcSendRequest$", result);
for (const auto& f : result) {
stub.set(f.second, schtRpcSendRequest);
}
}
}
void *schtSendRsp(void *param) { void *schtSendRsp(void *param) {
SSchJob *job = NULL; SSchJob *job = NULL;
int32_t code = 0; int32_t code = 0;