taosdata/TDengine | Build 3621 | source/libs/stream/src/streamTask.c | Coveralls

UNCOV

static int32_t addToTaskset(SArray* pArray, SStreamTask* pTask) {

×

UNCOV

33

  int32_t childId = taosArrayGetSize(pArray);

×

UNCOV

34

  pTask->info.selfChildId = childId;

×

UNCOV

35

  void* p = taosArrayPush(pArray, &pTask);

×

UNCOV

36

  return (p == NULL) ? terrno : TSDB_CODE_SUCCESS;

×

UNCOV

39

static int32_t doUpdateTaskEpset(SStreamTask* pTask, int32_t nodeId, SEpSet* pEpSet, bool* pUpdated) {

×

UNCOV

40

  int32_t code = 0;

×

UNCOV

41

  char    buf[512] = {0};

×

UNCOV

43

  if (pTask->info.nodeId == nodeId) {  // execution task should be moved away

×

UNCOV

44

    bool isEqual = isEpsetEqual(&pTask->info.epSet, pEpSet);

×

UNCOV

45

    code = epsetToStr(pEpSet, buf, tListLen(buf));

×

UNCOV

46

    if (code) { // print error and continue

×

47

      stError("%s failed to convert epset to str, code:%s", pTask->id.idStr, tstrerror(code));

×

48

      return code;

×

UNCOV

51

    if (!isEqual) {

×

UNCOV

52

      (*pUpdated) = true;

×

UNCOV

53

      char tmp[512] = {0};

×

UNCOV

54

      code = epsetToStr(&pTask->info.epSet, tmp, tListLen(tmp));  // only for log file, ignore errors

×

UNCOV

55

      if (code) { // print error and continue

×

56

        stError("%s failed to convert epset to str, code:%s", pTask->id.idStr, tstrerror(code));

×

57

        return code;

×

UNCOV

60

      epsetAssign(&pTask->info.epSet, pEpSet);

×

UNCOV

61

      stDebug("s-task:0x%x (vgId:%d) self node epset is updated %s, old:%s", pTask->id.taskId, nodeId, buf, tmp);

×

UNCOV

63

      stDebug("s-task:0x%x (vgId:%d) not updated task epset, since epset identical, %s", pTask->id.taskId, nodeId, buf);

×

UNCOV

68

  int32_t level = pTask->info.taskLevel;

×

UNCOV

69

  if (level == TASK_LEVEL__SOURCE) {

×

UNCOV

70

    code = streamTaskUpdateDownstreamInfo(pTask, nodeId, pEpSet, pUpdated);

×

UNCOV

71

  } else if (level == TASK_LEVEL__AGG) {

×

UNCOV

72

    code = streamTaskUpdateUpstreamInfo(pTask, nodeId, pEpSet, pUpdated);

×

UNCOV

73

    code = streamTaskUpdateDownstreamInfo(pTask, nodeId, pEpSet, pUpdated);

×

UNCOV

75

    code = streamTaskUpdateUpstreamInfo(pTask, nodeId, pEpSet, pUpdated);

×

UNCOV

78

  return code;

×

81

static void freeItem(void* p) {

×

82

  SStreamContinueExecInfo* pInfo = p;

×

83

  rpcFreeCont(pInfo->msg.pCont);

×

84

×

UNCOV

86

static void freeUpstreamItem(void* p) {

×

UNCOV

87

  SStreamUpstreamEpInfo** pInfo = p;

×

UNCOV

88

  taosMemoryFree(*pInfo);

×

UNCOV

89

×

UNCOV

91

static SStreamUpstreamEpInfo* createStreamTaskEpInfo(const SStreamTask* pTask) {

×

UNCOV

92

  SStreamUpstreamEpInfo* pEpInfo = taosMemoryMalloc(sizeof(SStreamUpstreamEpInfo));

×

UNCOV

93

  if (pEpInfo == NULL) {

×

94

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

95

    return NULL;

×

UNCOV

98

  pEpInfo->childId = pTask->info.selfChildId;

×

UNCOV

99

  pEpInfo->epSet = pTask->info.epSet;

×

UNCOV

100

  pEpInfo->nodeId = pTask->info.nodeId;

×

UNCOV

101

  pEpInfo->taskId = pTask->id.taskId;

×

UNCOV

102

  pEpInfo->stage = -1;

×

UNCOV

104

  return pEpInfo;

×

UNCOV

107

int32_t tNewStreamTask(int64_t streamId, int8_t taskLevel, SEpSet* pEpset, bool fillHistory, int32_t trigger,

×

UNCOV

110

  *p = NULL;

×

UNCOV

112

  SStreamTask* pTask = (SStreamTask*)taosMemoryCalloc(1, sizeof(SStreamTask));

×

UNCOV

113

  if (pTask == NULL) {

×

114

    stError("s-task:0x%" PRIx64 " failed malloc new stream task, size:%d, code:%s", streamId,

×

116

    return terrno;

×

UNCOV

119

  pTask->ver = SSTREAM_TASK_VER;

×

UNCOV

120

  pTask->id.taskId = tGenIdPI32();

×

UNCOV

121

  pTask->id.streamId = streamId;

×

UNCOV

123

  pTask->info.taskLevel = taskLevel;

×

UNCOV

124

  pTask->info.fillHistory = fillHistory;

×

UNCOV

125

  pTask->info.trigger = trigger;

×

UNCOV

126

  pTask->info.delaySchedParam = triggerParam;

×

UNCOV

127

  pTask->subtableWithoutMd5 = subtableWithoutMd5;

×

UNCOV

129

  int32_t code = streamCreateStateMachine(pTask);

×

UNCOV

130

  if (pTask->status.pSM == NULL || code != TSDB_CODE_SUCCESS) {

×

131

    taosMemoryFreeClear(pTask);

×

132

    return code;

×

UNCOV

135

  char    buf[128] = {0};

×

UNCOV

136

  int32_t ret = snprintf(buf, tListLen(buf), "0x%" PRIx64 "-0x%x", pTask->id.streamId, pTask->id.taskId);

×

UNCOV

137

  if (ret < 0 || ret >= tListLen(buf)) {

×

138

    stError("s-task:0x%x failed to set the taskIdstr, code: out of buffer", pTask->id.taskId);

×

139

    return TSDB_CODE_OUT_OF_BUFFER;

×

UNCOV

142

  pTask->id.idStr = taosStrdup(buf);

×

UNCOV

143

  if (pTask->id.idStr == NULL) {

×

144

    stError("s-task:0x%x failed to build task id, code: out of memory", pTask->id.taskId);

×

145

    return terrno;

×

UNCOV

148

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

UNCOV

149

  pTask->status.taskStatus = fillHistory ? TASK_STATUS__SCAN_HISTORY : TASK_STATUS__READY;

×

UNCOV

150

  pTask->inputq.status = TASK_INPUT_STATUS__NORMAL;

×

UNCOV

151

  pTask->outputq.status = TASK_OUTPUT_STATUS__NORMAL;

×

UNCOV

153

  pTask->taskCheckInfo.pList = taosArrayInit(4, sizeof(SDownstreamStatusInfo));

×

UNCOV

154

  code = taosThreadMutexInit(&pTask->taskCheckInfo.checkInfoLock, NULL);

×

UNCOV

155

  if (code) {

×

156

    return code;

×

UNCOV

159

  if (fillHistory && !hasFillhistory) {

×

160

    stError("s-task:0x%x create task failed, due to inconsistent fill-history flag", pTask->id.taskId);

×

161

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

164

  epsetAssign(&(pTask->info.mnodeEpset), pEpset);

×

UNCOV

166

  code = addToTaskset(pTaskList, pTask);

×

UNCOV

167

  *p = pTask;

×

UNCOV

169

  return code;

×

UNCOV

172

int32_t tDecodeStreamTaskChkInfo(SDecoder* pDecoder, SCheckpointInfo* pChkpInfo) {

×

UNCOV

179

  if (tStartDecode(pDecoder) < 0) return -1;

×

UNCOV

180

  if (tDecodeI64(pDecoder, &pChkpInfo->msgVer) < 0) return -1;

×

UNCOV

183

  if (tDecodeI64(pDecoder, &skip64) < 0) return -1;

×

UNCOV

184

  if (tDecodeI32(pDecoder, &skip32) < 0) return -1;

×

UNCOV

185

  if (tDecodeI32(pDecoder, &skip32) < 0) return -1;

×

UNCOV

186

  if (tDecodeI8(pDecoder, &skip8) < 0) return -1;

×

UNCOV

187

  if (tDecodeI8(pDecoder, &skip8) < 0) return -1;

×

UNCOV

188

  if (tDecodeI16(pDecoder, &skip16) < 0) return -1;

×

UNCOV

190

  if (tDecodeI8(pDecoder, &skip8) < 0) return -1;

×

UNCOV

191

  if (tDecodeI8(pDecoder, &skip8) < 0) return -1;

×

UNCOV

193

  if (tDecodeI32(pDecoder, &skip32) < 0) return -1;

×

UNCOV

194

  if (tDecodeI32(pDecoder, &skip32) < 0) return -1;

×

UNCOV

195

  if (tDecodeSEpSet(pDecoder, &epSet) < 0) return -1;

×

UNCOV

196

  if (tDecodeSEpSet(pDecoder, &epSet) < 0) return -1;

×

UNCOV

198

  if (tDecodeI64(pDecoder, &pChkpInfo->checkpointId) < 0) return -1;

×

UNCOV

199

  if (tDecodeI64(pDecoder, &pChkpInfo->checkpointVer) < 0) return -1;

×

UNCOV

201

  tEndDecode(pDecoder);

×

UNCOV

202

  return 0;

×

UNCOV

205

int32_t tDecodeStreamTaskId(SDecoder* pDecoder, STaskId* pTaskId) {

×

UNCOV

207

  if (tStartDecode(pDecoder) < 0) return -1;

×

UNCOV

208

  if (tDecodeI64(pDecoder, &ver) < 0) return -1;

×

UNCOV

209

  if (ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1;

×

UNCOV

211

  if (tDecodeI64(pDecoder, &pTaskId->streamId) < 0) return -1;

×

UNCOV

213

  int32_t taskId = 0;

×

UNCOV

214

  if (tDecodeI32(pDecoder, &taskId) < 0) return -1;

×

UNCOV

216

  pTaskId->taskId = taskId;

×

UNCOV

217

  tEndDecode(pDecoder);

×

UNCOV

218

  return 0;

×

UNCOV

221

void tFreeStreamTask(void* pParam) {

×

UNCOV

222

  char*        p = NULL;

×

UNCOV

223

  SStreamTask* pTask = pParam;

×

UNCOV

224

  int32_t      taskId = pTask->id.taskId;

×

UNCOV

226

  STaskExecStatisInfo* pStatis = &pTask->execInfo;

×

UNCOV

228

  ETaskStatus status1 = TASK_STATUS__UNINIT;

×

UNCOV

229

  streamMutexLock(&pTask->lock);

×

UNCOV

230

  if (pTask->status.pSM != NULL) {

×

UNCOV

231

    SStreamTaskState status = streamTaskGetStatus(pTask);

×

UNCOV

232

    p = status.name;

×

UNCOV

233

    status1 = status.state;

×

UNCOV

235

  streamMutexUnlock(&pTask->lock);

×

UNCOV

237

  stDebug("start to free s-task:0x%x %p, state:%s, refId:%" PRId64, taskId, pTask, p, pTask->id.refId);

×

UNCOV

239

  SCheckpointInfo* pCkInfo = &pTask->chkInfo;

×

UNCOV

240

  stDebug("s-task:0x%x task exec summary: create:%" PRId64 ", init:%" PRId64 ", start:%" PRId64

×

UNCOV

246

  if (pTask->schedInfo.pDelayTimer != NULL) {

×

UNCOV

247

    streamTmrStop(pTask->schedInfo.pDelayTimer);

×

UNCOV

248

    pTask->schedInfo.pDelayTimer = NULL;

×

UNCOV

251

  if (pTask->hTaskInfo.pTimer != NULL) {

×

UNCOV

252

    streamTmrStop(pTask->hTaskInfo.pTimer);

×

UNCOV

253

    pTask->hTaskInfo.pTimer = NULL;

×

UNCOV

256

  if (pTask->msgInfo.pRetryTmr != NULL) {

×

UNCOV

257

    streamTmrStop(pTask->msgInfo.pRetryTmr);

×

UNCOV

258

    pTask->msgInfo.pRetryTmr = NULL;

×

UNCOV

261

  if (pTask->inputq.queue) {

×

UNCOV

262

    streamQueueClose(pTask->inputq.queue, pTask->id.taskId);

×

UNCOV

263

    pTask->inputq.queue = NULL;

×

UNCOV

266

  if (pTask->outputq.queue) {

×

UNCOV

267

    streamQueueClose(pTask->outputq.queue, pTask->id.taskId);

×

UNCOV

268

    pTask->outputq.queue = NULL;

×

UNCOV

271

  if (pTask->exec.qmsg) {

×

UNCOV

272

    taosMemoryFree(pTask->exec.qmsg);

×

UNCOV

275

  if (pTask->exec.pExecutor) {

×

UNCOV

276

    qDestroyTask(pTask->exec.pExecutor);

×

UNCOV

277

    pTask->exec.pExecutor = NULL;

×

UNCOV

280

  if (pTask->exec.pWalReader != NULL) {

×

UNCOV

281

    walCloseReader(pTask->exec.pWalReader);

×

UNCOV

282

    pTask->exec.pWalReader = NULL;

×

UNCOV

285

  streamClearChkptReadyMsg(pTask->chkInfo.pActiveInfo);

×

UNCOV

287

  if (pTask->msgInfo.pData != NULL) {

×

UNCOV

288

    clearBufferedDispatchMsg(pTask);

×

UNCOV

291

  if (pTask->outputInfo.type == TASK_OUTPUT__TABLE) {

×

UNCOV

292

    tDeleteSchemaWrapper(pTask->outputInfo.tbSink.pSchemaWrapper);

×

UNCOV

293

    taosMemoryFree(pTask->outputInfo.tbSink.pTSchema);

×

UNCOV

294

    tSimpleHashCleanup(pTask->outputInfo.tbSink.pTbInfo);

×

UNCOV

295

    tDeleteSchemaWrapper(pTask->outputInfo.tbSink.pTagSchema);

×

UNCOV

296

  } else if (pTask->outputInfo.type == TASK_OUTPUT__SHUFFLE_DISPATCH) {

×

UNCOV

297

    taosArrayDestroy(pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos);

×

UNCOV

300

  streamTaskCleanupCheckInfo(&pTask->taskCheckInfo);

×

UNCOV

301

  streamFreeTaskState(pTask, pTask->status.removeBackendFiles ? 1 : 0);

×

UNCOV

303

  if (pTask->pNameMap) {

×

UNCOV

304

    tSimpleHashCleanup(pTask->pNameMap);

×

UNCOV

307

  streamDestroyStateMachine(pTask->status.pSM);

×

UNCOV

308

  pTask->status.pSM = NULL;

×

UNCOV

310

  streamTaskDestroyUpstreamInfo(&pTask->upstreamInfo);

×

UNCOV

312

  taosMemoryFree(pTask->outputInfo.pTokenBucket);

×

UNCOV

313

  streamMutexDestroy(&pTask->lock);

×

UNCOV

315

  taosArrayDestroy(pTask->msgInfo.pSendInfo);

×

UNCOV

316

  pTask->msgInfo.pSendInfo = NULL;

×

UNCOV

317

  streamMutexDestroy(&pTask->msgInfo.lock);

×

UNCOV

319

  taosArrayDestroy(pTask->outputInfo.pNodeEpsetUpdateList);

×

UNCOV

320

  pTask->outputInfo.pNodeEpsetUpdateList = NULL;

×

UNCOV

322

  if (pTask->id.idStr != NULL) {

×

UNCOV

323

    taosMemoryFree((void*)pTask->id.idStr);

×

UNCOV

326

  streamTaskDestroyActiveChkptInfo(pTask->chkInfo.pActiveInfo);

×

UNCOV

327

  pTask->chkInfo.pActiveInfo = NULL;

×

UNCOV

329

  taosArrayDestroyP(pTask->notifyInfo.pNotifyAddrUrls, NULL);

×

UNCOV

330

  taosMemoryFreeClear(pTask->notifyInfo.streamName);

×

UNCOV

331

  taosMemoryFreeClear(pTask->notifyInfo.stbFullName);

×

UNCOV

332

  tDeleteSchemaWrapper(pTask->notifyInfo.pSchemaWrapper);

×

UNCOV

334

  pTask->notifyEventStat = (STaskNotifyEventStat){0};

×

UNCOV

336

  taosMemoryFree(pTask);

×

UNCOV

337

  stDebug("s-task:0x%x free task completed", taskId);

×

UNCOV

338

×

UNCOV

340

void streamFreeTaskState(SStreamTask* pTask, int8_t remove) {

×

UNCOV

341

  stDebug("s-task:0x%x start to free task state/backend", pTask->id.taskId);

×

UNCOV

342

  if (pTask->pState != NULL) {

×

UNCOV

343

    stDebug("s-task:0x%x start to free task state", pTask->id.taskId);

×

UNCOV

344

    streamStateClose(pTask->pState, remove);

×

UNCOV

346

    if (remove) taskDbSetClearFileFlag(pTask->pBackend);

×

UNCOV

347

    taskDbRemoveRef(pTask->pBackend);

×

UNCOV

348

    pTask->pBackend = NULL;

×

UNCOV

349

    pTask->pState = NULL;

×

UNCOV

351

    stDebug("s-task:0x%x task state is NULL, may del backend:%s", pTask->id.taskId,

×

UNCOV

353

    if (remove) {

×

UNCOV

354

      if (pTask->backendPath != NULL) {

×

UNCOV

355

        stDebug("s-task:0x%x task state is NULL, do del backend:%s", pTask->id.taskId, pTask->backendPath);

×

UNCOV

356

        taosRemoveDir(pTask->backendPath);

×

UNCOV

361

  if (pTask->backendPath != NULL) {

×

UNCOV

362

    taosMemoryFree(pTask->backendPath);

×

UNCOV

363

    pTask->backendPath = NULL;

×

UNCOV

365

×

UNCOV

367

static void setInitialVersionInfo(SStreamTask* pTask, int64_t ver) {

×

UNCOV

368

  SCheckpointInfo* pChkInfo = &pTask->chkInfo;

×

UNCOV

369

  SDataRange*      pRange = &pTask->dataRange;

×

UNCOV

372

  if ((pTask->info.fillHistory == 0) && (!HAS_RELATED_FILLHISTORY_TASK(pTask))) {

×

UNCOV

373

    pChkInfo->checkpointVer = ver - 1;  // only update when generating checkpoint

×

UNCOV

374

    pChkInfo->processedVer = ver - 1;   // already processed version

×

UNCOV

375

    pChkInfo->nextProcessVer = ver;     // next processed version

×

UNCOV

377

    pRange->range.maxVer = ver;

×

UNCOV

378

    pRange->range.minVer = ver;

×

UNCOV

382

    if (pTask->info.fillHistory == 1) {

×

UNCOV

383

      pChkInfo->checkpointVer = pRange->range.maxVer;

×

UNCOV

384

      pChkInfo->processedVer = pRange->range.maxVer;

×

UNCOV

385

      pChkInfo->nextProcessVer = pRange->range.maxVer + 1;

×

UNCOV

387

      pChkInfo->checkpointVer = pRange->range.minVer - 1;

×

UNCOV

388

      pChkInfo->processedVer = pRange->range.minVer - 1;

×

UNCOV

389

      pChkInfo->nextProcessVer = pRange->range.minVer;

×

UNCOV

392

        if (pRange->range.minVer == 0) {

×

UNCOV

393

          pChkInfo->checkpointVer = 0;

×

UNCOV

394

          pChkInfo->processedVer = 0;

×

UNCOV

395

          pChkInfo->nextProcessVer = 1;

×

UNCOV

396

          stDebug("s-task:%s update the processedVer to 0 from -1 due to compatible purpose", pTask->id.idStr);

×

UNCOV

401

×

UNCOV

403

int32_t streamTaskSetBackendPath(SStreamTask* pTask) {

×

UNCOV

404

  int64_t streamId = 0;

×

UNCOV

405

  int32_t taskId = 0;

×

UNCOV

407

  if (pTask->info.fillHistory) {

×

UNCOV

408

    streamId = pTask->streamTaskId.streamId;

×

UNCOV

409

    taskId = pTask->streamTaskId.taskId;

×

UNCOV

411

    streamId = pTask->id.streamId;

×

UNCOV

412

    taskId = pTask->id.taskId;

×

UNCOV

415

  char    id[128] = {0};

×

UNCOV

416

  int32_t nBytes = snprintf(id, tListLen(id), "0x%" PRIx64 "-0x%x", streamId, taskId);

×

UNCOV

417

  if (nBytes < 0 || nBytes >= sizeof(id)) {

×

UNCOV

418

    return TSDB_CODE_OUT_OF_BUFFER;

×

UNCOV

421

  int32_t len = strlen(pTask->pMeta->path);

×

UNCOV

422

  pTask->backendPath = (char*)taosMemoryMalloc(len + nBytes + 2);

×

UNCOV

423

  if (pTask->backendPath == NULL) {

×

424

    return terrno;

×

UNCOV

427

  int32_t code = snprintf(pTask->backendPath, len + nBytes + 2, "%s%s%s", pTask->pMeta->path, TD_DIRSEP, id);

×

UNCOV

428

  if (code < 0 || code >= len + nBytes + 2) {

×

UNCOV

429

    stError("s-task:%s failed to set backend path:%s, code: out of buffer", pTask->id.idStr, pTask->backendPath);

×

430

    return TSDB_CODE_OUT_OF_BUFFER;

×

UNCOV

432

    stDebug("s-task:%s set backend path:%s", pTask->id.idStr, pTask->backendPath);

×

UNCOV

433

    return 0;

×

UNCOV

437

int32_t streamTaskInit(SStreamTask* pTask, SStreamMeta* pMeta, SMsgCb* pMsgCb, int64_t ver) {

×

UNCOV

438

  int32_t code = createStreamTaskIdStr(pTask->id.streamId, pTask->id.taskId, &pTask->id.idStr);

×

UNCOV

439

  if (code) {

×

440

    stError("0x%x failed create stream task id str, code:%s", pTask->id.taskId, tstrerror(code));

×

441

    return code;

×

UNCOV

444

  pTask->id.refId = 0;

×

UNCOV

445

  pTask->inputq.status = TASK_INPUT_STATUS__NORMAL;

×

UNCOV

446

  pTask->outputq.status = TASK_OUTPUT_STATUS__NORMAL;

×

UNCOV

448

  int32_t code1 = streamQueueOpen(512 << 10, &pTask->inputq.queue);

×

UNCOV

449

  int32_t code2 = streamQueueOpen(512 << 10, &pTask->outputq.queue);

×

UNCOV

450

  if (code1 || code2) {

×

UNCOV

451

    stError("s-task:%s failed to prepare the input/output queue, initialize task failed", pTask->id.idStr);

×

452

    return TSDB_CODE_OUT_OF_MEMORY;

×

UNCOV

455

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

UNCOV

457

  code = streamCreateStateMachine(pTask);

×

UNCOV

458

  if (pTask->status.pSM == NULL || code != TSDB_CODE_SUCCESS) {

×

459

    stError("s-task:%s failed create state-machine for stream task, initialization failed, code:%s", pTask->id.idStr,

×

461

    return code;

×

UNCOV

464

  pTask->execInfo.created = taosGetTimestampMs();

×

UNCOV

465

  setInitialVersionInfo(pTask, ver);

×

UNCOV

467

  pTask->pMeta = pMeta;

×

UNCOV

468

  pTask->pMsgCb = pMsgCb;

×

UNCOV

469

  pTask->msgInfo.pSendInfo = taosArrayInit(4, sizeof(SDispatchEntry));

×

UNCOV

470

  if (pTask->msgInfo.pSendInfo == NULL) {

×

471

    stError("s-task:%s failed to create sendInfo struct for stream task, code:Out of memory", pTask->id.idStr);

×

472

    return terrno;

×

UNCOV

475

  code = taosThreadMutexInit(&pTask->msgInfo.lock, NULL);

×

UNCOV

476

  if (code) {

×

477

    stError("s-task:0x%x failed to init msgInfo mutex, code:%s", pTask->id.taskId, tstrerror(code));

×

478

    return code;

×

UNCOV

481

  TdThreadMutexAttr attr = {0};

×

UNCOV

482

  code = taosThreadMutexAttrInit(&attr);

×

UNCOV

483

  if (code != 0) {

×

484

    stError("s-task:%s initElapsed mutex attr failed, code:%s", pTask->id.idStr, tstrerror(code));

×

485

    return code;

×

UNCOV

488

  code = taosThreadMutexAttrSetType(&attr, PTHREAD_MUTEX_RECURSIVE);

×

UNCOV

489

  if (code != 0) {

×

490

    stError("s-task:%s set mutex attr recursive, code:%s", pTask->id.idStr, tstrerror(code));

×

491

    return code;

×

UNCOV

494

  code = taosThreadMutexInit(&pTask->lock, &attr);

×

UNCOV

495

  if (code) {

×

496

    return code;

×

UNCOV

499

  code = taosThreadMutexAttrDestroy(&attr);

×

UNCOV

500

  if (code) {

×

501

    return code;

×

UNCOV

504

  streamTaskOpenAllUpstreamInput(pTask);

×

UNCOV

506

  STaskOutputInfo* pOutputInfo = &pTask->outputInfo;

×

UNCOV

507

  pOutputInfo->pTokenBucket = taosMemoryCalloc(1, sizeof(STokenBucket));

×

UNCOV

508

  if (pOutputInfo->pTokenBucket == NULL) {

×

509

    stError("s-task:%s failed to prepare the tokenBucket, code:%s", pTask->id.idStr, tstrerror(terrno));

×

510

    return terrno;

×

UNCOV

515

  code = streamTaskInitTokenBucket(pOutputInfo->pTokenBucket, 35, 35, tsSinkDataRate, pTask->id.idStr);

×

UNCOV

516

  if (code) {

×

517

    return code;

×

UNCOV

520

  pOutputInfo->pNodeEpsetUpdateList = taosArrayInit(4, sizeof(SDownstreamTaskEpset));

×

UNCOV

521

  if (pOutputInfo->pNodeEpsetUpdateList == NULL) {

×

522

    stError("s-task:%s failed to prepare downstreamUpdateList, code:%s", pTask->id.idStr, tstrerror(terrno));

×

523

    return terrno;

×

UNCOV

526

  pTask->taskCheckInfo.pList = taosArrayInit(4, sizeof(SDownstreamStatusInfo));

×

UNCOV

527

  if (pTask->taskCheckInfo.pList == NULL) {

×

528

    stError("s-task:%s failed to prepare taskCheckInfo list, code:%s", pTask->id.idStr, tstrerror(terrno));

×

529

    return terrno;

×

UNCOV

532

  if (pTask->chkInfo.pActiveInfo == NULL) {

×

UNCOV

533

    code = streamTaskCreateActiveChkptInfo(&pTask->chkInfo.pActiveInfo);

×

UNCOV

534

    if (code) {

×

535

      stError("s-task:%s failed to create active checkpoint info, code:%s", pTask->id.idStr, tstrerror(code));

×

536

      return code;

×

UNCOV

540

  return streamTaskSetBackendPath(pTask);

×

UNCOV

543

int32_t streamTaskGetNumOfDownstream(const SStreamTask* pTask) {

×

UNCOV

544

  if (pTask->info.taskLevel == TASK_LEVEL__SINK) {

×

UNCOV

545

    return 0;

×

UNCOV

548

  int32_t type = pTask->outputInfo.type;

×

UNCOV

549

  if (type == TASK_OUTPUT__TABLE) {

×

UNCOV

550

    return 0;

×

UNCOV

551

  } else if (type == TASK_OUTPUT__FIXED_DISPATCH) {

×

UNCOV

552

    return 1;

×

UNCOV

554

    SArray* vgInfo = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos;

×

UNCOV

555

    return taosArrayGetSize(vgInfo);

×

UNCOV

559

int32_t streamTaskGetNumOfUpstream(const SStreamTask* pTask) { return taosArrayGetSize(pTask->upstreamInfo.pList); }

×

UNCOV

561

int32_t streamTaskSetUpstreamInfo(SStreamTask* pTask, const SStreamTask* pUpstreamTask) {

×

UNCOV

562

  SStreamUpstreamEpInfo* pEpInfo = createStreamTaskEpInfo(pUpstreamTask);

×

UNCOV

563

  if (pEpInfo == NULL) {

×

564

    return terrno;

×

UNCOV

567

  if (pTask->upstreamInfo.pList == NULL) {

×

UNCOV

568

    pTask->upstreamInfo.pList = taosArrayInit(4, POINTER_BYTES);

×

UNCOV

571

  void* p = taosArrayPush(pTask->upstreamInfo.pList, &pEpInfo);

×

UNCOV

572

  return (p == NULL) ? terrno : TSDB_CODE_SUCCESS;

×

UNCOV

575

int32_t streamTaskUpdateUpstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet, bool* pUpdated) {

×

UNCOV

576

  int32_t code = 0;

×

UNCOV

577

  char    buf[512] = {0};

×

UNCOV

578

  code = epsetToStr(pEpSet, buf, tListLen(buf));  // ignore error since it is only for log file.

×

UNCOV

579

  if (code != 0) {  // print error and continue

×

580

    stError("%s failed to convert epset to str, code:%s", pTask->id.idStr, tstrerror(code));

×

581

    return code;

×

UNCOV

584

  int32_t numOfUpstream = taosArrayGetSize(pTask->upstreamInfo.pList);

×

UNCOV

585

  for (int32_t i = 0; i < numOfUpstream; ++i) {

×

UNCOV

586

    SStreamUpstreamEpInfo* pInfo = taosArrayGetP(pTask->upstreamInfo.pList, i);

×

UNCOV

587

    if (pInfo->nodeId == nodeId) {

×

UNCOV

588

      bool equal = isEpsetEqual(&pInfo->epSet, pEpSet);

×

UNCOV

589

      if (!equal) {

×

UNCOV

590

        *pUpdated = true;

×

UNCOV

592

        char tmp[512] = {0};

×

UNCOV

593

        code = epsetToStr(&pInfo->epSet, tmp, tListLen(tmp));

×

UNCOV

594

        if (code != 0) {  // print error and continue

×

595

          stError("%s failed to convert epset to str, code:%s", pTask->id.idStr, tstrerror(code));

×

596

          return code;

×

UNCOV

599

        epsetAssign(&pInfo->epSet, pEpSet);

×

UNCOV

600

        stDebug("s-task:0x%x update the upstreamInfo taskId:0x%x(nodeId:%d) newEpset:%s old:%s", pTask->id.taskId,

×

UNCOV

603

        stDebug("s-task:0x%x not update upstreamInfo, since identical, task:0x%x(nodeId:%d) epset:%s", pTask->id.taskId,

×

UNCOV

607

      break;

×

UNCOV

611

  return code;

×

UNCOV

614

void streamTaskDestroyUpstreamInfo(SUpstreamInfo* pUpstreamInfo) {

×

UNCOV

615

  if (pUpstreamInfo->pList != NULL) {

×

UNCOV

616

    taosArrayDestroyEx(pUpstreamInfo->pList, freeUpstreamItem);

×

UNCOV

617

    pUpstreamInfo->numOfClosed = 0;

×

UNCOV

618

    pUpstreamInfo->pList = NULL;

×

UNCOV

620

×

UNCOV

622

void streamTaskSetFixedDownstreamInfo(SStreamTask* pTask, const SStreamTask* pDownstreamTask) {

×

UNCOV

623

  STaskDispatcherFixed* pDispatcher = &pTask->outputInfo.fixedDispatcher;

×

UNCOV

624

  pDispatcher->taskId = pDownstreamTask->id.taskId;

×

UNCOV

625

  pDispatcher->nodeId = pDownstreamTask->info.nodeId;

×

UNCOV

626

  pDispatcher->epSet = pDownstreamTask->info.epSet;

×

UNCOV

628

  pTask->outputInfo.type = TASK_OUTPUT__FIXED_DISPATCH;

×

UNCOV

629

  pTask->msgInfo.msgType = TDMT_STREAM_TASK_DISPATCH;

×

UNCOV

630

×

UNCOV

632

int32_t streamTaskUpdateDownstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet, bool* pUpdated) {

×

UNCOV

633

  char    buf[512] = {0};

×

UNCOV

634

  int32_t code = epsetToStr(pEpSet, buf, tListLen(buf));  // ignore the error since only for log files.

×

UNCOV

635

  if (code != 0) {                                        // print error and continue

×

636

    stError("%s failed to convert epset to str, code:%s", pTask->id.idStr, tstrerror(code));

×

637

    return code;

×

UNCOV

640

  int32_t id = pTask->id.taskId;

×

UNCOV

641

  int8_t  type = pTask->outputInfo.type;

×

UNCOV

643

  if (type == TASK_OUTPUT__SHUFFLE_DISPATCH) {

×

UNCOV

644

    SArray* pVgs = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos;

×

UNCOV

646

    for (int32_t i = 0; i < taosArrayGetSize(pVgs); i++) {

×

UNCOV

647

      SVgroupInfo* pVgInfo = taosArrayGet(pVgs, i);

×

UNCOV

648

      if (pVgInfo == NULL) {

×

649

        continue;

×

UNCOV

652

      if (pVgInfo->vgId == nodeId) {

×

UNCOV

653

        bool isEqual = isEpsetEqual(&pVgInfo->epSet, pEpSet);

×

UNCOV

654

        if (!isEqual) {

×

UNCOV

655

          *pUpdated = true;

×

UNCOV

657

          char tmp[512] = {0};

×

UNCOV

658

          code = epsetToStr(&pVgInfo->epSet, tmp, tListLen(tmp));

×

UNCOV

659

          if (code != 0) {  // print error and continue

×

660

            stError("%s failed to convert epset to str, code:%s", pTask->id.idStr, tstrerror(code));

×

661

            return code;

×

UNCOV

664

          epsetAssign(&pVgInfo->epSet, pEpSet);

×

UNCOV

665

          stDebug("s-task:0x%x update dispatch info, task:0x%x(nodeId:%d) newEpset:%s old:%s", id, pVgInfo->taskId,

×

UNCOV

668

          stDebug("s-task:0x%x not update dispatch info, since identical, task:0x%x(nodeId:%d) epset:%s", id,

×

UNCOV

671

        break;

×

UNCOV

674

  } else if (type == TASK_OUTPUT__FIXED_DISPATCH) {

×

UNCOV

675

    STaskDispatcherFixed* pDispatcher = &pTask->outputInfo.fixedDispatcher;

×

UNCOV

676

    if (pDispatcher->nodeId == nodeId) {

×

UNCOV

677

      bool equal = isEpsetEqual(&pDispatcher->epSet, pEpSet);

×

UNCOV

678

      if (!equal) {

×

679

        *pUpdated = true;

×

681

        char tmp[512] = {0};

×

682

        code = epsetToStr(&pDispatcher->epSet, tmp, tListLen(tmp));

×

683

        if (code != 0) {  // print error and continue

×

684

          stError("%s failed to convert epset to str, code:%s", pTask->id.idStr, tstrerror(code));

×

685

          return code;

×

688

        epsetAssign(&pDispatcher->epSet, pEpSet);

×

689

        stDebug("s-task:0x%x update dispatch info, task:0x%x(nodeId:%d) newEpset:%s old:%s", id, pDispatcher->taskId,

×

UNCOV

692

        stDebug("s-task:0x%x not update dispatch info, since identical, task:0x%x(nodeId:%d) epset:%s", id,

×

UNCOV

698

  return code;

×

UNCOV

701

int32_t streamTaskStop(SStreamTask* pTask) {

×

UNCOV

702

  int32_t     vgId = pTask->pMeta->vgId;

×

UNCOV

703

  int64_t     st = taosGetTimestampMs();

×

UNCOV

704

  const char* id = pTask->id.idStr;

×

UNCOV

706

  int32_t code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_STOP);

×

UNCOV

707

  if (code) {

×

708

    stError("failed to handle STOP event, s-task:%s, code:%s", id, tstrerror(code));

×

709

    return code;

×

UNCOV

712

  if (pTask->info.taskLevel != TASK_LEVEL__SINK && pTask->exec.pExecutor != NULL) {

×

NEW

713

    code = qKillTask(pTask->exec.pExecutor, TSDB_CODE_SUCCESS, 5000);

×

UNCOV

714

    if (code != TSDB_CODE_SUCCESS) {

×

715

      stError("s-task:%s failed to kill task related query handle, code:%s", id, tstrerror(code));

×

UNCOV

719

  while (!streamTaskIsIdle(pTask)) {

×

720

    stDebug("s-task:%s level:%d wait for task to be idle and then close, check again in 100ms", id,

×

722

    taosMsleep(100);

×

UNCOV

725

  int64_t el = taosGetTimestampMs() - st;

×

UNCOV

726

  stDebug("vgId:%d s-task:%s is closed in %" PRId64 " ms", vgId, id, el);

×

UNCOV

727

  return code;

×

UNCOV

730

bool streamTaskUpdateEpsetInfo(SStreamTask* pTask, SArray* pNodeList) {

×

UNCOV

731

  STaskExecStatisInfo* p = &pTask->execInfo;

×

UNCOV

733

  int32_t numOfNodes = taosArrayGetSize(pNodeList);

×

UNCOV

734

  int64_t prevTs = p->latestUpdateTs;

×

UNCOV

736

  p->latestUpdateTs = taosGetTimestampMs();

×

UNCOV

737

  p->updateCount += 1;

×

UNCOV

738

  stDebug("s-task:0x%x update task nodeEp epset, updatedNodes:%d, updateCount:%d, prevTs:%" PRId64, pTask->id.taskId,

×

UNCOV

741

  bool updated = false;

×

UNCOV

742

  for (int32_t i = 0; i < numOfNodes; ++i) {

×

UNCOV

743

    SNodeUpdateInfo* pInfo = taosArrayGet(pNodeList, i);

×

UNCOV

744

    if (pInfo == NULL) {

×

745

      continue;

×

UNCOV

748

    int32_t code = doUpdateTaskEpset(pTask, pInfo->nodeId, &pInfo->newEp, &updated);

×

UNCOV

749

    if (code) {

×

750

      stError("s-task:0x%x failed to update the task nodeEp epset, code:%s", pTask->id.taskId, tstrerror(code));

×

UNCOV

754

  return updated;

×

UNCOV

757

void streamTaskResetUpstreamStageInfo(SStreamTask* pTask) {

×

UNCOV

758

  if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {

×

UNCOV

759

    return;

×

UNCOV

762

  int32_t size = taosArrayGetSize(pTask->upstreamInfo.pList);

×

UNCOV

763

  for (int32_t i = 0; i < size; ++i) {

×

UNCOV

764

    SStreamUpstreamEpInfo* pInfo = taosArrayGetP(pTask->upstreamInfo.pList, i);

×

UNCOV

765

    pInfo->stage = -1;

×

UNCOV

768

  stDebug("s-task:%s reset all upstream tasks stage info", pTask->id.idStr);

×

UNCOV

771

void streamTaskOpenAllUpstreamInput(SStreamTask* pTask) {

×

UNCOV

772

  int32_t num = taosArrayGetSize(pTask->upstreamInfo.pList);

×

UNCOV

773

  if (num == 0) {

×

UNCOV

774

    return;

×

UNCOV

777

  for (int32_t i = 0; i < num; ++i) {

×

UNCOV

778

    SStreamUpstreamEpInfo* pInfo = taosArrayGetP(pTask->upstreamInfo.pList, i);

×

UNCOV

779

    pInfo->dataAllowed = true;

×

UNCOV

782

  pTask->upstreamInfo.numOfClosed = 0;

×

UNCOV

783

  stDebug("s-task:%s opening up inputQ for %d upstream tasks", pTask->id.idStr, num);

×

UNCOV

786

void streamTaskCloseUpstreamInput(SStreamTask* pTask, int32_t taskId) {

×

UNCOV

787

  SStreamUpstreamEpInfo* pInfo = NULL;

×

UNCOV

788

  streamTaskGetUpstreamTaskEpInfo(pTask, taskId, &pInfo);

×

UNCOV

790

  if ((pInfo != NULL) && pInfo->dataAllowed) {

×

UNCOV

791

    pInfo->dataAllowed = false;

×

UNCOV

792

    if (pTask->upstreamInfo.numOfClosed < streamTaskGetNumOfUpstream(pTask)) {

×

UNCOV

793

      int32_t t = atomic_add_fetch_32(&pTask->upstreamInfo.numOfClosed, 1);

×

795

      stError("s-task:%s not inc closed input, since they have been all closed already", pTask->id.idStr);

×

UNCOV

798

×

UNCOV

800

void streamTaskOpenUpstreamInput(SStreamTask* pTask, int32_t taskId) {

×

UNCOV

801

  SStreamUpstreamEpInfo* pInfo = NULL;

×

UNCOV

802

  streamTaskGetUpstreamTaskEpInfo(pTask, taskId, &pInfo);

×

UNCOV

804

  if (pInfo != NULL && (!pInfo->dataAllowed)) {

×

UNCOV

805

    int32_t t = atomic_sub_fetch_32(&pTask->upstreamInfo.numOfClosed, 1);

×

UNCOV

806

    stDebug("s-task:%s open inputQ for upstream:0x%x, remain closed:%d", pTask->id.idStr, taskId, t);

×

UNCOV

807

    pInfo->dataAllowed = true;

×

UNCOV

809

×

811

bool streamTaskIsAllUpstreamClosed(SStreamTask* pTask) {

×

812

  return pTask->upstreamInfo.numOfClosed == taosArrayGetSize(pTask->upstreamInfo.pList);

×

UNCOV

815

bool streamTaskSetSchedStatusWait(SStreamTask* pTask) {

×

UNCOV

816

  bool ret = false;

×

UNCOV

818

  streamMutexLock(&pTask->lock);

×

UNCOV

819

  if (pTask->status.schedStatus == TASK_SCHED_STATUS__INACTIVE) {

×

UNCOV

820

    pTask->status.schedStatus = TASK_SCHED_STATUS__WAITING;

×

UNCOV

821

    ret = true;

×

UNCOV

824

  streamMutexUnlock(&pTask->lock);

×

UNCOV

825

  return ret;

×

UNCOV

828

int8_t streamTaskSetSchedStatusActive(SStreamTask* pTask) {

×

UNCOV

829

  streamMutexLock(&pTask->lock);

×

UNCOV

830

  int8_t status = pTask->status.schedStatus;

×

UNCOV

831

  if (status == TASK_SCHED_STATUS__WAITING) {

×

UNCOV

832

    pTask->status.schedStatus = TASK_SCHED_STATUS__ACTIVE;

×

UNCOV

834

  streamMutexUnlock(&pTask->lock);

×

UNCOV

836

  return status;

×

UNCOV

839

int8_t streamTaskSetSchedStatusInactive(SStreamTask* pTask) {

×

UNCOV

840

  streamMutexLock(&pTask->lock);

×

UNCOV

841

  int8_t status = pTask->status.schedStatus;

×

UNCOV

842

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

UNCOV

843

  streamMutexUnlock(&pTask->lock);

×

UNCOV

845

  return status;

×

UNCOV

848

int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t resetRelHalt) {

×

UNCOV

849

  int32_t      code = 0;

×

UNCOV

850

  SStreamMeta* pMeta = pTask->pMeta;

×

UNCOV

851

  SStreamTask* pStreamTask = NULL;

×

UNCOV

853

  if (pTask->info.fillHistory == 0) {

×

UNCOV

854

    return code;

×

857

  code = streamMetaAcquireTaskUnsafe(pMeta, &pTask->streamTaskId, &pStreamTask);

×

UNCOV

858

  if (code == 0) {

×

859

    stDebug("s-task:%s clear the related stream task:0x%x attr to fill-history task", pTask->id.idStr,

×

862

    streamMutexLock(&(pStreamTask->lock));

×

863

    CLEAR_RELATED_FILLHISTORY_TASK(pStreamTask);

×

865

    if (resetRelHalt) {

×

866

      stDebug("s-task:0x%" PRIx64 " set the persistent status attr to be ready, prev:%s, status in sm:%s",

×

869

      pStreamTask->status.taskStatus = TASK_STATUS__READY;

×

NEW

872

    code = streamMetaSaveTaskInMeta(pMeta, pStreamTask);

×

873

    streamMutexUnlock(&(pStreamTask->lock));

×

875

    streamMetaReleaseTask(pMeta, pStreamTask);

×

UNCOV

878

  return code;

×

UNCOV

881

int32_t streamBuildAndSendDropTaskMsg(SMsgCb* pMsgCb, int32_t vgId, SStreamTaskId* pTaskId, int64_t resetRelHalt) {

×

UNCOV

882

  SVDropStreamTaskReq* pReq = rpcMallocCont(sizeof(SVDropStreamTaskReq));

×

UNCOV

883

  if (pReq == NULL) {

×

884

    return terrno;

×

UNCOV

887

  pReq->head.vgId = vgId;

×

UNCOV

888

  pReq->taskId = pTaskId->taskId;

×

UNCOV

889

  pReq->streamId = pTaskId->streamId;

×

UNCOV

890

  pReq->resetRelHalt = resetRelHalt;

×

UNCOV

892

  SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_DROP, .pCont = pReq, .contLen = sizeof(SVDropStreamTaskReq)};

×

UNCOV

893

  int32_t code = tmsgPutToQueue(pMsgCb, WRITE_QUEUE, &msg);

×

UNCOV

894

  if (code != TSDB_CODE_SUCCESS) {

×

895

    stError("vgId:%d failed to send drop task:0x%x msg, code:%s", vgId, pTaskId->taskId, tstrerror(code));

×

UNCOV

897

    stDebug("vgId:%d build and send drop task:0x%x msg", vgId, pTaskId->taskId);

×

UNCOV

900

  return code;

×

UNCOV

903

int32_t streamSendChkptReportMsg(SStreamTask* pTask, SCheckpointInfo* pCheckpointInfo, int8_t dropRelHTask) {

×

UNCOV

904

  int32_t                code = 0;

×

UNCOV

905

  int32_t                tlen = 0;

×

UNCOV

906

  int32_t                vgId = pTask->pMeta->vgId;

×

UNCOV

907

  const char*            id = pTask->id.idStr;

×

UNCOV

908

  SActiveCheckpointInfo* pActive = pCheckpointInfo->pActiveInfo;

×

UNCOV

910

  SCheckpointReport req = {.streamId = pTask->id.streamId,

×

UNCOV

911

                           .taskId = pTask->id.taskId,

×

UNCOV

914

                           .transId = pActive->transId,

×

UNCOV

915

                           .checkpointId = pActive->activeId,

×

UNCOV

916

                           .checkpointVer = pCheckpointInfo->processedVer,

×

UNCOV

917

                           .checkpointTs = pCheckpointInfo->startTs};

×

UNCOV

919

  tEncodeSize(tEncodeStreamTaskChkptReport, &req, tlen, code);

×

UNCOV

920

  if (code < 0) {

×

921

    stError("s-task:%s vgId:%d encode stream task checkpoint-report failed, code:%s", id, vgId, tstrerror(code));

×

922

    return -1;

×

UNCOV

925

  void* buf = rpcMallocCont(tlen);

×

UNCOV

926

  if (buf == NULL) {

×

927

    stError("s-task:%s vgId:%d encode stream task checkpoint-report msg failed, code:%s", id, vgId,

×

929

    return -1;

×

UNCOV

933

  tEncoderInit(&encoder, buf, tlen);

×

UNCOV

934

  if ((code = tEncodeStreamTaskChkptReport(&encoder, &req)) < 0) {

×

935

    rpcFreeCont(buf);

×

936

    tEncoderClear(&encoder);

×

937

    stError("s-task:%s vgId:%d encode stream task checkpoint-report msg failed, code:%s", id, vgId, tstrerror(code));

×

938

    return -1;

×

UNCOV

940

  tEncoderClear(&encoder);

×

UNCOV

942

  SRpcMsg msg = {0};

×

UNCOV

943

  initRpcMsg(&msg, TDMT_MND_STREAM_CHKPT_REPORT, buf, tlen);

×

UNCOV

944

  stDebug("s-task:%s vgId:%d build and send task checkpoint-report to mnode", id, vgId);

×

UNCOV

946

  return tmsgSendReq(&pTask->info.mnodeEpset, &msg);

×

UNCOV

949

STaskId streamTaskGetTaskId(const SStreamTask* pTask) {

×

UNCOV

950

  STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId};

×

UNCOV

951

  return id;

×

UNCOV

954

void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo) {

×

UNCOV

955

  pInfo->waitInterval = LAUNCH_HTASK_INTERVAL;

×

UNCOV

956

  pInfo->tickCount = ceil(LAUNCH_HTASK_INTERVAL / WAIT_FOR_MINIMAL_INTERVAL);

×

UNCOV

957

  pInfo->retryTimes = 0;

×

UNCOV

958

×

UNCOV

960

void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo) {

×

UNCOV

961

  pInfo->waitInterval *= RETRY_LAUNCH_INTERVAL_INC_RATE;

×

UNCOV

962

  pInfo->tickCount = ceil(pInfo->waitInterval / WAIT_FOR_MINIMAL_INTERVAL);

×

UNCOV

963

  pInfo->retryTimes += 1;

×

UNCOV

964

×

UNCOV

966

void streamTaskStatusInit(STaskStatusEntry* pEntry, const SStreamTask* pTask) {

×

UNCOV

967

  pEntry->id.streamId = pTask->id.streamId;

×

UNCOV

968

  pEntry->id.taskId = pTask->id.taskId;

×

UNCOV

969

  pEntry->stage = -1;

×

UNCOV

970

  pEntry->nodeId = pTask->info.nodeId;

×

UNCOV

971

  pEntry->status = TASK_STATUS__STOP;

×

UNCOV

972

×

UNCOV

974

void streamTaskStatusCopy(STaskStatusEntry* pDst, const STaskStatusEntry* pSrc) {

×

UNCOV

975

  pDst->stage = pSrc->stage;

×

UNCOV

976

  pDst->inputQUsed = pSrc->inputQUsed;

×

UNCOV

977

  pDst->inputRate = pSrc->inputRate;

×

UNCOV

978

  pDst->procsTotal = pSrc->procsTotal;

×

UNCOV

979

  pDst->procsThroughput = pSrc->procsThroughput;

×

UNCOV

980

  pDst->outputTotal = pSrc->outputTotal;

×

UNCOV

981

  pDst->outputThroughput = pSrc->outputThroughput;

×

UNCOV

982

  pDst->processedVer = pSrc->processedVer;

×

UNCOV

983

  pDst->verRange = pSrc->verRange;

×

UNCOV

984

  pDst->sinkQuota = pSrc->sinkQuota;

×

UNCOV

985

  pDst->sinkDataSize = pSrc->sinkDataSize;

×

UNCOV

986

  pDst->checkpointInfo = pSrc->checkpointInfo;

×

UNCOV

987

  pDst->startCheckpointId = pSrc->startCheckpointId;

×

UNCOV

988

  pDst->startCheckpointVer = pSrc->startCheckpointVer;

×

UNCOV

989

  pDst->status = pSrc->status;

×

UNCOV

991

  pDst->startTime = pSrc->startTime;

×

UNCOV

992

  pDst->hTaskId = pSrc->hTaskId;

×

UNCOV

993

  pDst->notifyEventStat = pSrc->notifyEventStat;

×

UNCOV

994

×

UNCOV

996

STaskStatusEntry streamTaskGetStatusEntry(SStreamTask* pTask) {

×

UNCOV

997

  SStreamMeta*         pMeta = pTask->pMeta;

×

UNCOV

998

  STaskExecStatisInfo* pExecInfo = &pTask->execInfo;

×

UNCOV

1000

  STaskStatusEntry entry = {

×

UNCOV

1001

      .id = streamTaskGetTaskId(pTask),

×

UNCOV

1002

      .status = streamTaskGetStatus(pTask).state,

×

UNCOV

1003

      .nodeId = pMeta->vgId,

×

UNCOV

1004

      .stage = pMeta->stage,

×

UNCOV

1006

      .inputQUsed = SIZE_IN_MiB(streamQueueGetItemSize(pTask->inputq.queue)),

×

UNCOV

1007

      .startTime = pExecInfo->readyTs,

×

UNCOV

1008

      .checkpointInfo.latestId = pTask->chkInfo.checkpointId,

×

UNCOV

1009

      .checkpointInfo.latestVer = pTask->chkInfo.checkpointVer,

×

UNCOV

1010

      .checkpointInfo.latestTime = pTask->chkInfo.checkpointTime,

×

UNCOV

1015

      .hTaskId = pTask->hTaskInfo.id.taskId,

×

UNCOV

1016

      .procsTotal = SIZE_IN_MiB(pExecInfo->inputDataSize),

×

UNCOV

1017

      .outputTotal = SIZE_IN_MiB(pExecInfo->outputDataSize),

×

UNCOV

1018

      .procsThroughput = SIZE_IN_KiB(pExecInfo->procsThroughput),

×

UNCOV

1019

      .outputThroughput = SIZE_IN_KiB(pExecInfo->outputThroughput),

×

UNCOV

1020

      .startCheckpointId = pExecInfo->startCheckpointId,

×

UNCOV

1021

      .startCheckpointVer = pExecInfo->startCheckpointVer,

×

UNCOV

1024

  return entry;

×

UNCOV

1027

static int32_t taskPauseCallback(SStreamTask* pTask, void* param) {

×

UNCOV

1028

  SStreamMeta* pMeta = pTask->pMeta;

×

UNCOV

1029

  int32_t      code = 0;

×

UNCOV

1031

  int32_t num = atomic_add_fetch_32(&pMeta->numOfPausedTasks, 1);

×

UNCOV

1032

  stInfo("vgId:%d s-task:%s pause stream task. paused task num:%d", pMeta->vgId, pTask->id.idStr, num);

×

UNCOV

1035

  if (pTask->info.fillHistory == 1) {

×

UNCOV

1036

    void* pExecutor = pTask->exec.pExecutor;

×

NEW

1037

    code = qKillTask(pExecutor, TSDB_CODE_SUCCESS, 10000);

×

UNCOV

1040

  stDebug("vgId:%d s-task:%s set pause flag and pause task", pMeta->vgId, pTask->id.idStr);

×

UNCOV

1041

  return code;

×

UNCOV

1044

void streamTaskPause(SStreamTask* pTask) {

×

UNCOV

1045

  int32_t code = streamTaskHandleEventAsync(pTask->status.pSM, TASK_EVENT_PAUSE, taskPauseCallback, NULL);

×

UNCOV

1046

  if (code) {

×

1047

    stError("s-task:%s failed handle pause event async, code:%s", pTask->id.idStr, tstrerror(code));

×

UNCOV

1049

×

UNCOV

1051

void streamTaskResume(SStreamTask* pTask) {

×

UNCOV

1052

  SStreamTaskState prevState = streamTaskGetStatus(pTask);

×

UNCOV

1054

  SStreamMeta* pMeta = pTask->pMeta;

×

UNCOV

1055

  int32_t      code = streamTaskRestoreStatus(pTask);

×

UNCOV

1056

  if (code == TSDB_CODE_SUCCESS) {

×

UNCOV

1057

    char*   pNew = streamTaskGetStatus(pTask).name;

×

UNCOV

1058

    int32_t num = atomic_sub_fetch_32(&pMeta->numOfPausedTasks, 1);

×

UNCOV

1059

    stInfo("s-task:%s status:%s resume from %s, paused task(s):%d", pTask->id.idStr, pNew, prevState.name, num);

×

UNCOV

1061

    stInfo("s-task:%s status:%s no need to resume, paused task(s):%d", pTask->id.idStr, prevState.name,

×

UNCOV

1064

×

UNCOV

1066

bool streamTaskIsSinkTask(const SStreamTask* pTask) { return pTask->info.taskLevel == TASK_LEVEL__SINK; }

×

UNCOV

1069

int32_t streamTaskSendCheckpointReq(SStreamTask* pTask) {

×

UNCOV

1071

  int32_t     tlen = 0;

×

UNCOV

1072

  int32_t     vgId = pTask->pMeta->vgId;

×

UNCOV

1073

  const char* id = pTask->id.idStr;

×

UNCOV

1075

  SStreamTaskCheckpointReq req = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId, .nodeId = vgId};

×

UNCOV

1076

  tEncodeSize(tEncodeStreamTaskCheckpointReq, &req, tlen, code);

×

UNCOV

1077

  if (code < 0) {

×

1078

    stError("s-task:%s vgId:%d encode stream task req checkpoint failed, code:%s", id, vgId, tstrerror(code));

×

1079

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

1082

  void* buf = rpcMallocCont(tlen);

×

UNCOV

1083

  if (buf == NULL) {

×

1084

    stError("s-task:%s vgId:%d encode stream task req checkpoint msg failed, code:Out of memory", id, vgId);

×

1085

    return terrno;

×

UNCOV

1089

  tEncoderInit(&encoder, buf, tlen);

×

UNCOV

1090

  if ((code = tEncodeStreamTaskCheckpointReq(&encoder, &req)) < 0) {

×

1091

    rpcFreeCont(buf);

×

1092

    tEncoderClear(&encoder);

×

1093

    stError("s-task:%s vgId:%d encode stream task req checkpoint msg failed, code:%s", id, vgId, tstrerror(code));

×

1094

    return code;

×

UNCOV

1097

  tEncoderClear(&encoder);

×

UNCOV

1099

  SRpcMsg msg = {0};

×

UNCOV

1100

  initRpcMsg(&msg, TDMT_MND_STREAM_REQ_CHKPT, buf, tlen);

×

UNCOV

1101

  stDebug("s-task:%s vgId:%d build and send task checkpoint req", id, vgId);

×

UNCOV

1103

  return tmsgSendReq(&pTask->info.mnodeEpset, &msg);

×

UNCOV

1106

void streamTaskGetUpstreamTaskEpInfo(SStreamTask* pTask, int32_t taskId, SStreamUpstreamEpInfo** pEpInfo) {

×

UNCOV

1107

  *pEpInfo = NULL;

×

UNCOV

1109

  int32_t num = taosArrayGetSize(pTask->upstreamInfo.pList);

×

UNCOV

1110

  for (int32_t i = 0; i < num; ++i) {

×

UNCOV

1111

    SStreamUpstreamEpInfo* pInfo = taosArrayGetP(pTask->upstreamInfo.pList, i);

×

UNCOV

1112

    if (pInfo == NULL) {

×

1113

      return;

×

UNCOV

1116

    if (pInfo->taskId == taskId) {

×

UNCOV

1117

      *pEpInfo = pInfo;

×

UNCOV

1118

      return;

×

1122

  stError("s-task:%s failed to find upstream task:0x%x", pTask->id.idStr, taskId);

×

1125

SEpSet* streamTaskGetDownstreamEpInfo(SStreamTask* pTask, int32_t taskId) {

×

1126

  if (pTask->info.taskLevel == TASK_OUTPUT__FIXED_DISPATCH) {

×

1127

    if (pTask->outputInfo.fixedDispatcher.taskId == taskId) {

×

1128

      return &pTask->outputInfo.fixedDispatcher.epSet;

×

1130

  } else if (pTask->info.taskLevel == TASK_OUTPUT__SHUFFLE_DISPATCH) {

×

1131

    SArray* pList = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos;

×

1132

    for (int32_t i = 0; i < taosArrayGetSize(pList); ++i) {

×

1133

      SVgroupInfo* pVgInfo = taosArrayGet(pList, i);

×

1134

      if (pVgInfo == NULL) {

×

1135

        continue;

×

1138

      if (pVgInfo->taskId == taskId) {

×

1139

        return &pVgInfo->epSet;

×

1144

  return NULL;

×

UNCOV

1147

int32_t createStreamTaskIdStr(int64_t streamId, int32_t taskId, const char** pId) {

×

UNCOV

1148

  char buf[128] = {0};

×

UNCOV

1149

  int32_t code = snprintf(buf, tListLen(buf),"0x%" PRIx64 "-0x%x", streamId, taskId);

×

UNCOV

1150

  if (code < 0 || code >= tListLen(buf)) {

×

1151

    return TSDB_CODE_OUT_OF_BUFFER;

×

UNCOV

1154

  *pId = taosStrdup(buf);

×

UNCOV

1156

  if (*pId == NULL) {

×

1157

    return terrno;

×

UNCOV

1159

    return TSDB_CODE_SUCCESS;

×

UNCOV

1163

static int32_t streamTaskEnqueueRetrieve(SStreamTask* pTask, SStreamRetrieveReq* pReq) {

×

UNCOV

1167

  code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SStreamDataBlock), (void**)&pData);

×

UNCOV

1168

  if (code) {

×

1169

    stError("s-task:%s failed to allocated retrieve-block", pTask->id.idStr);

×

1170

    return terrno = code;

×

UNCOV

1173

  pData->type = STREAM_INPUT__DATA_RETRIEVE;

×

UNCOV

1174

  pData->srcVgId = 0;

×

UNCOV

1176

  code = streamRetrieveReqToData(pReq, pData, pTask->id.idStr);

×

UNCOV

1177

  if (code != TSDB_CODE_SUCCESS) {

×

1178

    stError("s-task:%s failed to convert retrieve-data to block, code:%s", pTask->id.idStr, tstrerror(code));

×

1179

    taosFreeQitem(pData);

×

1180

    return code;

×

UNCOV

1183

  code = streamTaskPutDataIntoInputQ(pTask, (SStreamQueueItem*)pData);

×

UNCOV

1184

  if (code != TSDB_CODE_SUCCESS) {

×

1185

    stError("s-task:%s failed to put retrieve-block into inputQ, inputQ is full, discard the retrieve msg",

×

UNCOV

1189

  return code;

×

UNCOV

1192

int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq) {

×

UNCOV

1193

  int32_t code = streamTaskEnqueueRetrieve(pTask, pReq);

×

UNCOV

1194

  if (code != 0) {

×

1195

    return code;

×

UNCOV

1197

  return streamTrySchedExec(pTask);

×

UNCOV

1200

void streamTaskSetRemoveBackendFiles(SStreamTask* pTask) { pTask->status.removeBackendFiles = true; }

×

1202

void streamTaskGetActiveCheckpointInfo(const SStreamTask* pTask, int32_t* pTransId, int64_t* pCheckpointId) {

×

1203

  if (pTransId != NULL) {

×

1204

    *pTransId = pTask->chkInfo.pActiveInfo->transId;

×

1207

  if (pCheckpointId != NULL) {

×

1208

    *pCheckpointId = pTask->chkInfo.pActiveInfo->activeId;

×

1210

×

UNCOV

1212

int32_t streamTaskSetActiveCheckpointInfo(SStreamTask* pTask, int64_t activeCheckpointId) {

×

UNCOV

1213

  pTask->chkInfo.pActiveInfo->activeId = activeCheckpointId;

×

UNCOV

1214

  return TSDB_CODE_SUCCESS;

×

UNCOV

1217

void streamTaskSetFailedChkptInfo(SStreamTask* pTask, int32_t transId, int64_t checkpointId) {

×

UNCOV

1218

  pTask->chkInfo.pActiveInfo->transId = transId;

×

UNCOV

1219

  pTask->chkInfo.pActiveInfo->activeId = checkpointId;

×

UNCOV

1220

  pTask->chkInfo.pActiveInfo->failedId = checkpointId;

×

UNCOV

1221

  stDebug("s-task:%s set failed checkpointId:%"PRId64, pTask->id.idStr, checkpointId);

×

UNCOV

1222

×

UNCOV

1224

int32_t streamTaskCreateActiveChkptInfo(SActiveCheckpointInfo** pRes) {

×

UNCOV

1225

  SActiveCheckpointInfo* pInfo = taosMemoryCalloc(1, sizeof(SActiveCheckpointInfo));

×

UNCOV

1226

  if (pInfo == NULL) {

×

1227

    return terrno;

×

UNCOV

1230

  int32_t code = taosThreadMutexInit(&pInfo->lock, NULL);

×

UNCOV

1231

  if (code != TSDB_CODE_SUCCESS) {

×

1232

    return code;

×

UNCOV

1235

  pInfo->pDispatchTriggerList = taosArrayInit(4, sizeof(STaskTriggerSendInfo));

×

UNCOV

1236

  pInfo->pReadyMsgList = taosArrayInit(4, sizeof(STaskCheckpointReadyInfo));

×

UNCOV

1237

  pInfo->pCheckpointReadyRecvList = taosArrayInit(4, sizeof(STaskDownstreamReadyInfo));

×

UNCOV

1239

  *pRes = pInfo;

×

UNCOV

1240

  return code;

×

UNCOV

1243

void streamTaskDestroyActiveChkptInfo(SActiveCheckpointInfo* pInfo) {

×

UNCOV

1244

  if (pInfo == NULL) {

×

UNCOV

1245

    return;

×

UNCOV

1248

  streamMutexDestroy(&pInfo->lock);

×

UNCOV

1249

  taosArrayDestroy(pInfo->pDispatchTriggerList);

×

UNCOV

1250

  pInfo->pDispatchTriggerList = NULL;

×

UNCOV

1251

  taosArrayDestroy(pInfo->pReadyMsgList);

×

UNCOV

1252

  pInfo->pReadyMsgList = NULL;

×

UNCOV

1253

  taosArrayDestroy(pInfo->pCheckpointReadyRecvList);

×

UNCOV

1254

  pInfo->pCheckpointReadyRecvList = NULL;

×

UNCOV

1256

  SStreamTmrInfo* pTriggerTmr = &pInfo->chkptTriggerMsgTmr;

×

UNCOV

1257

  if (pTriggerTmr->tmrHandle != NULL) {

×

UNCOV

1258

    streamTmrStop(pTriggerTmr->tmrHandle);

×

UNCOV

1259

    pTriggerTmr->tmrHandle = NULL;

×

UNCOV

1262

  SStreamTmrInfo* pReadyTmr = &pInfo->chkptReadyMsgTmr;

×

UNCOV

1263

  if (pReadyTmr->tmrHandle != NULL) {

×

UNCOV

1264

    streamTmrStop(pReadyTmr->tmrHandle);

×

UNCOV

1265

    pReadyTmr->tmrHandle = NULL;

×

UNCOV

1268

  taosMemoryFree(pInfo);

×

UNCOV

1273

void streamTaskClearActiveInfo(SActiveCheckpointInfo* pInfo) {

×

UNCOV

1274

  pInfo->activeId = 0;

×

UNCOV

1275

  pInfo->transId = 0;

×

UNCOV

1276

  pInfo->allUpstreamTriggerRecv = 0;

×

UNCOV

1277

  pInfo->dispatchTrigger = false;

×

UNCOV

1279

  taosArrayClear(pInfo->pDispatchTriggerList);

×

UNCOV

1280

  taosArrayClear(pInfo->pCheckpointReadyRecvList);

×

UNCOV

1281

×

UNCOV

1283

const char* streamTaskGetExecType(int32_t type) {

×

UNCOV

1284

  switch (type) {

×

UNCOV

1285

    case STREAM_EXEC_T_EXTRACT_WAL_DATA:

×

UNCOV

1286

      return "scan-wal-file";

×

UNCOV

1287

    case STREAM_EXEC_T_START_ALL_TASKS:

×

UNCOV

1288

      return "start-all-tasks";

×

UNCOV

1289

    case STREAM_EXEC_T_START_ONE_TASK:

×

UNCOV

1290

      return "start-one-task";

×

UNCOV

1291

    case STREAM_EXEC_T_RESTART_ALL_TASKS:

×

UNCOV

1292

      return "restart-all-tasks";

×

UNCOV

1293

    case STREAM_EXEC_T_STOP_ALL_TASKS:

×

UNCOV

1294

      return "stop-all-tasks";

×

UNCOV

1295

    case STREAM_EXEC_T_RESUME_TASK:

×

UNCOV

1296

      return "resume-task-from-idle";

×

1297

    case STREAM_EXEC_T_ADD_FAILED_TASK:

×

1298

      return "record-start-failed-task";

×

NEW

1299

    case STREAM_EXEC_T_STOP_ONE_TASK:

×

NEW

1300

      return "stop-one-task";

×

UNCOV

1301

    case 0:

×

UNCOV

1302

      return "exec-all-tasks";

×

1303

    default:

×

1304

      return "invalid-exec-type";

×

UNCOV

1308

int32_t streamTaskAllocRefId(SStreamTask* pTask, int64_t** pRefId) {

×

UNCOV

1309

  *pRefId = taosMemoryMalloc(sizeof(int64_t));

×

UNCOV

1310

  if (*pRefId != NULL) {

×

UNCOV

1311

    **pRefId = pTask->id.refId;

×

UNCOV

1312

    int32_t code = metaRefMgtAdd(pTask->pMeta->vgId, *pRefId);

×

UNCOV

1313

    if (code != 0) {

×

1314

      stError("s-task:%s failed to add refId:%" PRId64 " into refId-mgmt, code:%s", pTask->id.idStr, pTask->id.refId,

×

UNCOV

1317

    return code;

×

1319

    stError("s-task:%s failed to alloc new ref id, code:%s", pTask->id.idStr, tstrerror(terrno));

×

1320

    return terrno;

×

UNCOV

1324

void streamTaskFreeRefId(int64_t* pRefId) {

×

UNCOV

1325

  if (pRefId == NULL) {

×

UNCOV

1326

    return;

×

UNCOV

1329

  metaRefMgtRemove(pRefId);

×

UNCOV

1332

static int32_t tEncodeStreamNotifyInfo(SEncoder* pEncoder, const SNotifyInfo* info) {

×

UNCOV

1333

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1334

  int32_t lino = 0;

×

UNCOV

1336

  QUERY_CHECK_NULL(pEncoder, code, lino, _exit, TSDB_CODE_INVALID_PARA);

×

UNCOV

1337

  QUERY_CHECK_NULL(info, code, lino, _exit, TSDB_CODE_INVALID_PARA);

×

UNCOV

1339

  int32_t addrSize = taosArrayGetSize(info->pNotifyAddrUrls);

×

UNCOV

1340

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

×

UNCOV

1341

  for (int32_t i = 0; i < addrSize; ++i) {

×

1342

    const char* url = taosArrayGetP(info->pNotifyAddrUrls, i);

×

1343

    TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, url));

×

UNCOV

1345

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, info->notifyEventTypes));

×

UNCOV

1346

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, info->notifyErrorHandle));

×

UNCOV

1347

  if (addrSize > 0) {

×

1348

    TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, info->streamName));

×

1349

    TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, info->stbFullName));

×

1350

    TAOS_CHECK_EXIT(tEncodeSSchemaWrapper(pEncoder, info->pSchemaWrapper));

×

UNCOV

1353

_exit:

×

UNCOV

1354

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

1355

    stError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1357

  return code;

×

UNCOV

1360

static int32_t tDecodeStreamNotifyInfo(SDecoder* pDecoder, SNotifyInfo* info) {

×

UNCOV

1361

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1362

  int32_t lino = 0;

×

UNCOV

1364

  QUERY_CHECK_NULL(pDecoder, code, lino, _exit, TSDB_CODE_INVALID_PARA);

×

UNCOV

1365

  QUERY_CHECK_NULL(info, code, lino, _exit, TSDB_CODE_INVALID_PARA);

×

UNCOV

1367

  int32_t addrSize = 0;

×

UNCOV

1368

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

×

UNCOV

1369

  info->pNotifyAddrUrls = taosArrayInit(addrSize, POINTER_BYTES);

×

UNCOV

1370

  QUERY_CHECK_NULL(info->pNotifyAddrUrls, code, lino, _exit, terrno);

×

UNCOV

1371

  for (int32_t i = 0; i < addrSize; ++i) {

×

1372

    char *url = NULL;

×

1373

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &url));

×

1374

    url = taosStrndup(url, TSDB_STREAM_NOTIFY_URL_LEN);

×

1375

    QUERY_CHECK_NULL(url, code, lino, _exit, terrno);

×

1376

    if (taosArrayPush(info->pNotifyAddrUrls, &url) == NULL) {

×

1377

      taosMemoryFree(url);

×

1378

      TAOS_CHECK_EXIT(terrno);

×

UNCOV

1381

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &info->notifyEventTypes));

×

UNCOV

1382

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &info->notifyErrorHandle));

×

UNCOV

1383

  if (addrSize > 0) {

×

1384

    char* name = NULL;

×

1385

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &name));

×

1386

    info->streamName = taosStrndup(name, TSDB_STREAM_FNAME_LEN + 1);

×

1387

    QUERY_CHECK_NULL(info->streamName, code, lino, _exit, terrno);

×

1388

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &name));

×

1389

    info->stbFullName = taosStrndup(name, TSDB_STREAM_FNAME_LEN + 1);

×

1390

    QUERY_CHECK_NULL(info->stbFullName, code, lino, _exit, terrno);

×

1391

    info->pSchemaWrapper = taosMemoryCalloc(1, sizeof(SSchemaWrapper));

×

1392

    if (info->pSchemaWrapper == NULL) {

×

1393

      TAOS_CHECK_EXIT(terrno);

×

1395

    TAOS_CHECK_EXIT(tDecodeSSchemaWrapper(pDecoder, info->pSchemaWrapper));

×

UNCOV

1398

_exit:

×

UNCOV

1399

  if (code != TSDB_CODE_SUCCESS) {

×

1400

    stError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1402

  return code;

×

UNCOV

1405

int32_t tEncodeStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {

×

UNCOV

1406

  int32_t code = 0;

×

UNCOV

1409

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

×

UNCOV

1410

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->ver));

×

UNCOV

1411

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->id.streamId));

×

UNCOV

1412

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->id.taskId));

×

UNCOV

1413

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->info.trigger));

×

UNCOV

1414

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pTask->info.taskLevel));

×

UNCOV

1415

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pTask->outputInfo.type));

×

UNCOV

1416

  TAOS_CHECK_EXIT(tEncodeI16(pEncoder, pTask->msgInfo.msgType));

×

UNCOV

1418

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pTask->status.taskStatus));

×

UNCOV

1419

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pTask->status.schedStatus));

×

UNCOV

1421

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->info.selfChildId));

×

UNCOV

1422

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->info.nodeId));

×

UNCOV

1423

  TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pTask->info.epSet));

×

UNCOV

1424

  TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pTask->info.mnodeEpset));

×

UNCOV

1426

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->chkInfo.checkpointId));

×

UNCOV

1427

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->chkInfo.checkpointVer));

×

UNCOV

1428

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pTask->info.fillHistory));

×

UNCOV

1430

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->hTaskInfo.id.streamId));

×

UNCOV

1431

  int32_t taskId = pTask->hTaskInfo.id.taskId;

×

UNCOV

1432

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, taskId));

×

UNCOV

1434

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->streamTaskId.streamId));

×

UNCOV

1435

  taskId = pTask->streamTaskId.taskId;

×

UNCOV

1436

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, taskId));

×

UNCOV

1438

  TAOS_CHECK_EXIT(tEncodeU64(pEncoder, pTask->dataRange.range.minVer));

×

UNCOV

1439

  TAOS_CHECK_EXIT(tEncodeU64(pEncoder, pTask->dataRange.range.maxVer));

×

UNCOV

1440

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->dataRange.window.skey));

×

UNCOV

1441

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->dataRange.window.ekey));

×

UNCOV

1443

  int32_t epSz = taosArrayGetSize(pTask->upstreamInfo.pList);

×

UNCOV

1444

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, epSz));

×

UNCOV

1445

  for (int32_t i = 0; i < epSz; i++) {

×

UNCOV

1446

    SStreamUpstreamEpInfo* pInfo = taosArrayGetP(pTask->upstreamInfo.pList, i);

×

UNCOV

1447

    TAOS_CHECK_EXIT(tEncodeStreamEpInfo(pEncoder, pInfo));

×

UNCOV

1450

  if (pTask->info.taskLevel != TASK_LEVEL__SINK) {

×

UNCOV

1451

    TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, pTask->exec.qmsg));

×

UNCOV

1454

  if (pTask->outputInfo.type == TASK_OUTPUT__TABLE) {

×

UNCOV

1455

    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->outputInfo.tbSink.stbUid));

×

UNCOV

1456

    TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, pTask->outputInfo.tbSink.stbFullName));

×

UNCOV

1457

    TAOS_CHECK_EXIT(tEncodeSSchemaWrapper(pEncoder, pTask->outputInfo.tbSink.pSchemaWrapper));

×

UNCOV

1458

  } else if (pTask->outputInfo.type == TASK_OUTPUT__SMA) {

×

UNCOV

1459

    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->outputInfo.smaSink.smaId));

×

UNCOV

1460

  } else if (pTask->outputInfo.type == TASK_OUTPUT__FETCH) {

×

1461

    TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pTask->outputInfo.fetchSink.reserved));

×

UNCOV

1462

  } else if (pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH) {

×

UNCOV

1463

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->outputInfo.fixedDispatcher.taskId));

×

UNCOV

1464

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pTask->outputInfo.fixedDispatcher.nodeId));

×

UNCOV

1465

    TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pTask->outputInfo.fixedDispatcher.epSet));

×

UNCOV

1466

  } else if (pTask->outputInfo.type == TASK_OUTPUT__SHUFFLE_DISPATCH) {

×

UNCOV

1467

    TAOS_CHECK_EXIT(tSerializeSUseDbRspImp(pEncoder, &pTask->outputInfo.shuffleDispatcher.dbInfo));

×

UNCOV

1468

    TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, pTask->outputInfo.shuffleDispatcher.stbFullName));

×

UNCOV

1470

  TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pTask->info.delaySchedParam));

×

UNCOV

1471

  TAOS_CHECK_EXIT(tEncodeI8(pEncoder, pTask->subtableWithoutMd5));

×

UNCOV

1472

  TAOS_CHECK_EXIT(tEncodeCStrWithLen(pEncoder, pTask->reserve, sizeof(pTask->reserve) - 1));

×

UNCOV

1474

  if (pTask->ver >= SSTREAM_TASK_ADD_NOTIFY_VER) {

×

UNCOV

1475

    TAOS_CHECK_EXIT(tEncodeStreamNotifyInfo(pEncoder, &pTask->notifyInfo));

×

UNCOV

1478

  tEndEncode(pEncoder);

×

UNCOV

1479

_exit:

×

UNCOV

1480

  return code;

×

UNCOV

1483

int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {

×

UNCOV

1484

  int32_t taskId = 0;

×

UNCOV

1485

  int32_t code = 0;

×

UNCOV

1488

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

×

UNCOV

1489

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->ver));

×

UNCOV

1490

  if (pTask->ver <= SSTREAM_TASK_INCOMPATIBLE_VER || pTask->ver > SSTREAM_TASK_VER) {

×

UNCOV

1491

    TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);

×

UNCOV

1494

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->id.streamId));

×

UNCOV

1495

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->id.taskId));

×

UNCOV

1496

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->info.trigger));

×

UNCOV

1497

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pTask->info.taskLevel));

×

UNCOV

1498

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pTask->outputInfo.type));

×

UNCOV

1499

  TAOS_CHECK_EXIT(tDecodeI16(pDecoder, &pTask->msgInfo.msgType));

×

UNCOV

1501

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pTask->status.taskStatus));

×

UNCOV

1502

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pTask->status.schedStatus));

×

UNCOV

1504

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->info.selfChildId));

×

UNCOV

1505

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->info.nodeId));

×

UNCOV

1506

  TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &pTask->info.epSet));

×

UNCOV

1507

  TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &pTask->info.mnodeEpset));

×

UNCOV

1509

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->chkInfo.checkpointId));

×

UNCOV

1510

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->chkInfo.checkpointVer));

×

UNCOV

1511

  TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pTask->info.fillHistory));

×

UNCOV

1513

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->hTaskInfo.id.streamId));

×

UNCOV

1514

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));

×

UNCOV

1515

  pTask->hTaskInfo.id.taskId = taskId;

×

UNCOV

1517

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->streamTaskId.streamId));

×

UNCOV

1518

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));

×

UNCOV

1519

  pTask->streamTaskId.taskId = taskId;

×

UNCOV

1521

  TAOS_CHECK_EXIT(tDecodeU64(pDecoder, (uint64_t*)&pTask->dataRange.range.minVer));

×

UNCOV

1522

  TAOS_CHECK_EXIT(tDecodeU64(pDecoder, (uint64_t*)&pTask->dataRange.range.maxVer));

×

UNCOV

1523

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->dataRange.window.skey));

×

UNCOV

1524

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->dataRange.window.ekey));

×

UNCOV

1526

  int32_t epSz = -1;

×

UNCOV

1527

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &epSz) < 0);

×

UNCOV

1529

  if ((pTask->upstreamInfo.pList = taosArrayInit(epSz, POINTER_BYTES)) == NULL) {

×

1530

    TAOS_CHECK_EXIT(terrno);

×

UNCOV

1532

  for (int32_t i = 0; i < epSz; i++) {

×

UNCOV

1533

    SStreamUpstreamEpInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamUpstreamEpInfo));

×

UNCOV

1534

    if (pInfo == NULL) {

×

1535

      TAOS_CHECK_EXIT(terrno);

×

UNCOV

1537

    if ((code = tDecodeStreamEpInfo(pDecoder, pInfo)) < 0) {

×

1538

      taosMemoryFreeClear(pInfo);

×

1539

      goto _exit;

×

UNCOV

1541

    if (taosArrayPush(pTask->upstreamInfo.pList, &pInfo) == NULL) {

×

1542

      TAOS_CHECK_EXIT(terrno);

×

UNCOV

1546

  if (pTask->info.taskLevel != TASK_LEVEL__SINK) {

×

UNCOV

1547

    TAOS_CHECK_EXIT(tDecodeCStrAlloc(pDecoder, &pTask->exec.qmsg));

×

UNCOV

1550

  if (pTask->outputInfo.type == TASK_OUTPUT__TABLE) {

×

UNCOV

1551

    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->outputInfo.tbSink.stbUid));

×

UNCOV

1552

    TAOS_CHECK_EXIT(tDecodeCStrTo(pDecoder, pTask->outputInfo.tbSink.stbFullName));

×

UNCOV

1553

    pTask->outputInfo.tbSink.pSchemaWrapper = taosMemoryCalloc(1, sizeof(SSchemaWrapper));

×

UNCOV

1554

    if (pTask->outputInfo.tbSink.pSchemaWrapper == NULL) {

×

1555

      TAOS_CHECK_EXIT(terrno);

×

UNCOV

1557

    TAOS_CHECK_EXIT(tDecodeSSchemaWrapper(pDecoder, pTask->outputInfo.tbSink.pSchemaWrapper));

×

UNCOV

1558

  } else if (pTask->outputInfo.type == TASK_OUTPUT__SMA) {

×

UNCOV

1559

    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->outputInfo.smaSink.smaId));

×

UNCOV

1560

  } else if (pTask->outputInfo.type == TASK_OUTPUT__FETCH) {

×

1561

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pTask->outputInfo.fetchSink.reserved));

×

UNCOV

1562

  } else if (pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH) {

×

UNCOV

1563

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->outputInfo.fixedDispatcher.taskId));

×

UNCOV

1564

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &pTask->outputInfo.fixedDispatcher.nodeId));

×

UNCOV

1565

    TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &pTask->outputInfo.fixedDispatcher.epSet));

×

UNCOV

1566

  } else if (pTask->outputInfo.type == TASK_OUTPUT__SHUFFLE_DISPATCH) {

×

UNCOV

1567

    TAOS_CHECK_EXIT(tDeserializeSUseDbRspImp(pDecoder, &pTask->outputInfo.shuffleDispatcher.dbInfo));

×

UNCOV

1568

    TAOS_CHECK_EXIT(tDecodeCStrTo(pDecoder, pTask->outputInfo.shuffleDispatcher.stbFullName));

×

UNCOV

1570

  TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &pTask->info.delaySchedParam));

×

UNCOV

1571

  if (pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER) {

×

UNCOV

1572

    TAOS_CHECK_EXIT(tDecodeI8(pDecoder, &pTask->subtableWithoutMd5));

×

UNCOV

1574

  TAOS_CHECK_EXIT(tDecodeCStrTo(pDecoder, pTask->reserve));

×

UNCOV

1576

  if (pTask->ver >= SSTREAM_TASK_ADD_NOTIFY_VER) {

×

UNCOV

1577

    TAOS_CHECK_EXIT(tDecodeStreamNotifyInfo(pDecoder, &pTask->notifyInfo));

×

UNCOV

1580

  tEndDecode(pDecoder);

×

UNCOV

1582

_exit:

×

UNCOV

1583

  return code;

×

taosdata / TDengine / #3621

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #3621

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous