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

32

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

×

33

  int32_t childId = taosArrayGetSize(pArray);

×

34

  pTask->info.selfChildId = childId;

×

35

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

×

36

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

×

39

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

×

40

  int32_t code = 0;

×

41

  char    buf[512] = {0};

×

43

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

×

44

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

×

45

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

×

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;

×

51

    if (!isEqual) {

×

52

      (*pUpdated) = true;

×

53

      char tmp[512] = {0};

×

54

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

×

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;

×

60

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

×

61

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

×

63

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

×

68

  int32_t level = pTask->info.taskLevel;

×

69

  if (level == TASK_LEVEL__SOURCE) {

×

70

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

×

71

  } else if (level == TASK_LEVEL__AGG) {

×

72

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

×

73

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

×

75

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

×

78

  return code;

×

81

static void freeItem(void* p) {

×

82

  SStreamContinueExecInfo* pInfo = p;

×

83

  rpcFreeCont(pInfo->msg.pCont);

×

84

×

86

static void freeUpstreamItem(void* p) {

×

87

  SStreamUpstreamEpInfo** pInfo = p;

×

88

  taosMemoryFree(*pInfo);

×

89

×

91

static SStreamUpstreamEpInfo* createStreamTaskEpInfo(const SStreamTask* pTask) {

×

92

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

×

93

  if (pEpInfo == NULL) {

×

94

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

95

    return NULL;

×

98

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

×

99

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

×

100

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

×

101

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

×

102

  pEpInfo->stage = -1;

×

104

  return pEpInfo;

×

107

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

×

110

  *p = NULL;

×

112

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

×

113

  if (pTask == NULL) {

×

114

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

×

116

    return terrno;

×

119

  pTask->ver = SSTREAM_TASK_VER;

×

120

  pTask->id.taskId = tGenIdPI32();

×

121

  pTask->id.streamId = streamId;

×

123

  pTask->info.taskLevel = taskLevel;

×

124

  pTask->info.fillHistory = fillHistory;

×

125

  pTask->info.trigger = trigger;

×

126

  pTask->info.delaySchedParam = triggerParam;

×

127

  pTask->subtableWithoutMd5 = subtableWithoutMd5;

×

129

  int32_t code = streamCreateStateMachine(pTask);

×

130

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

×

131

    taosMemoryFreeClear(pTask);

×

132

    return code;

×

135

  char    buf[128] = {0};

×

136

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

×

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;

×

142

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

×

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;

×

148

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

149

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

×

150

  pTask->inputq.status = TASK_INPUT_STATUS__NORMAL;

×

151

  pTask->outputq.status = TASK_OUTPUT_STATUS__NORMAL;

×

153

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

×

154

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

×

155

  if (code) {

×

156

    return code;

×

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;

×

164

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

×

166

  code = addToTaskset(pTaskList, pTask);

×

167

  *p = pTask;

×

169

  return code;

×

172

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

×

179

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

×

180

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

×

183

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

×

184

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

×

185

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

×

186

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

×

187

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

×

188

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

×

190

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

×

191

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

×

193

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

×

194

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

×

195

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

×

196

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

×

198

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

×

199

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

×

201

  tEndDecode(pDecoder);

×

202

  return 0;

×

205

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

×

207

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

×

208

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

×

209

  if (ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1;

×

211

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

×

213

  int32_t taskId = 0;

×

214

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

×

216

  pTaskId->taskId = taskId;

×

217

  tEndDecode(pDecoder);

×

218

  return 0;

×

221

void tFreeStreamTask(void* pParam) {

×

222

  char*        p = NULL;

×

223

  SStreamTask* pTask = pParam;

×

224

  int32_t      taskId = pTask->id.taskId;

×

226

  STaskExecStatisInfo* pStatis = &pTask->execInfo;

×

228

  ETaskStatus status1 = TASK_STATUS__UNINIT;

×

229

  streamMutexLock(&pTask->lock);

×

230

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

×

231

    SStreamTaskState status = streamTaskGetStatus(pTask);

×

232

    p = status.name;

×

233

    status1 = status.state;

×

235

  streamMutexUnlock(&pTask->lock);

×

237

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

×

239

  SCheckpointInfo* pCkInfo = &pTask->chkInfo;

×

240

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

×

246

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

×

247

    streamTmrStop(pTask->schedInfo.pDelayTimer);

×

248

    pTask->schedInfo.pDelayTimer = NULL;

×

251

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

×

252

    streamTmrStop(pTask->hTaskInfo.pTimer);

×

253

    pTask->hTaskInfo.pTimer = NULL;

×

256

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

×

257

    streamTmrStop(pTask->msgInfo.pRetryTmr);

×

258

    pTask->msgInfo.pRetryTmr = NULL;

×

261

  if (pTask->inputq.queue) {

×

262

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

×

263

    pTask->inputq.queue = NULL;

×

266

  if (pTask->outputq.queue) {

×

267

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

×

268

    pTask->outputq.queue = NULL;

×

271

  if (pTask->exec.qmsg) {

×

272

    taosMemoryFree(pTask->exec.qmsg);

×

275

  if (pTask->exec.pExecutor) {

×

276

    qDestroyTask(pTask->exec.pExecutor);

×

277

    pTask->exec.pExecutor = NULL;

×

280

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

×

281

    walCloseReader(pTask->exec.pWalReader);

×

282

    pTask->exec.pWalReader = NULL;

×

285

  streamClearChkptReadyMsg(pTask->chkInfo.pActiveInfo);

×

287

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

×

288

    clearBufferedDispatchMsg(pTask);

×

291

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

×

292

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

×

293

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

×

294

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

×

295

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

×

296

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

×

297

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

×

300

  streamTaskCleanupCheckInfo(&pTask->taskCheckInfo);

×

301

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

×

303

  if (pTask->pNameMap) {

×

304

    tSimpleHashCleanup(pTask->pNameMap);

×

307

  streamDestroyStateMachine(pTask->status.pSM);

×

308

  pTask->status.pSM = NULL;

×

310

  streamTaskDestroyUpstreamInfo(&pTask->upstreamInfo);

×

312

  taosMemoryFree(pTask->outputInfo.pTokenBucket);

×

313

  streamMutexDestroy(&pTask->lock);

×

315

  taosArrayDestroy(pTask->msgInfo.pSendInfo);

×

316

  pTask->msgInfo.pSendInfo = NULL;

×

317

  streamMutexDestroy(&pTask->msgInfo.lock);

×

319

  taosArrayDestroy(pTask->outputInfo.pNodeEpsetUpdateList);

×

320

  pTask->outputInfo.pNodeEpsetUpdateList = NULL;

×

322

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

×

323

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

×

326

  streamTaskDestroyActiveChkptInfo(pTask->chkInfo.pActiveInfo);

×

327

  pTask->chkInfo.pActiveInfo = NULL;

×

329

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

×

330

  taosMemoryFreeClear(pTask->notifyInfo.streamName);

×

331

  taosMemoryFreeClear(pTask->notifyInfo.stbFullName);

×

332

  tDeleteSchemaWrapper(pTask->notifyInfo.pSchemaWrapper);

×

334

  taosMemoryFree(pTask);

×

335

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

×

336

×

338

void streamFreeTaskState(SStreamTask* pTask, int8_t remove) {

×

339

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

×

340

  if (pTask->pState != NULL) {

×

341

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

×

342

    streamStateClose(pTask->pState, remove);

×

344

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

×

345

    taskDbRemoveRef(pTask->pBackend);

×

346

    pTask->pBackend = NULL;

×

347

    pTask->pState = NULL;

×

349

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

×

351

    if (remove) {

×

352

      if (pTask->backendPath != NULL) {

×

353

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

×

354

        taosRemoveDir(pTask->backendPath);

×

359

  if (pTask->backendPath != NULL) {

×

360

    taosMemoryFree(pTask->backendPath);

×

361

    pTask->backendPath = NULL;

×

363

×

365

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

×

366

  SCheckpointInfo* pChkInfo = &pTask->chkInfo;

×

367

  SDataRange*      pRange = &pTask->dataRange;

×

370

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

×

371

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

×

372

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

×

373

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

×

375

    pRange->range.maxVer = ver;

×

376

    pRange->range.minVer = ver;

×

380

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

×

381

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

×

382

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

×

383

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

×

385

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

×

386

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

×

387

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

×

390

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

×

391

          pChkInfo->checkpointVer = 0;

×

392

          pChkInfo->processedVer = 0;

×

393

          pChkInfo->nextProcessVer = 1;

×

394

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

×

399

×

401

int32_t streamTaskSetBackendPath(SStreamTask* pTask) {

×

402

  int64_t streamId = 0;

×

403

  int32_t taskId = 0;

×

405

  if (pTask->info.fillHistory) {

×

406

    streamId = pTask->streamTaskId.streamId;

×

407

    taskId = pTask->streamTaskId.taskId;

×

409

    streamId = pTask->id.streamId;

×

410

    taskId = pTask->id.taskId;

×

413

  char    id[128] = {0};

×

414

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

×

415

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

×

416

    return TSDB_CODE_OUT_OF_BUFFER;

×

419

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

×

420

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

×

421

  if (pTask->backendPath == NULL) {

×

422

    return terrno;

×

425

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

×

426

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

×

427

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

×

428

    return TSDB_CODE_OUT_OF_BUFFER;

×

430

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

×

431

    return 0;

×

435

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

×

436

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

×

437

  if (code) {

×

438

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

×

439

    return code;

×

442

  pTask->id.refId = 0;

×

443

  pTask->inputq.status = TASK_INPUT_STATUS__NORMAL;

×

444

  pTask->outputq.status = TASK_OUTPUT_STATUS__NORMAL;

×

446

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

×

447

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

×

448

  if (code1 || code2) {

×

449

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

×

450

    return TSDB_CODE_OUT_OF_MEMORY;

×

453

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

455

  code = streamCreateStateMachine(pTask);

×

456

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

×

457

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

×

459

    return code;

×

462

  pTask->execInfo.created = taosGetTimestampMs();

×

463

  setInitialVersionInfo(pTask, ver);

×

465

  pTask->pMeta = pMeta;

×

466

  pTask->pMsgCb = pMsgCb;

×

467

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

×

468

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

×

469

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

×

470

    return terrno;

×

473

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

×

474

  if (code) {

×

475

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

×

476

    return code;

×

479

  TdThreadMutexAttr attr = {0};

×

480

  code = taosThreadMutexAttrInit(&attr);

×

481

  if (code != 0) {

×

482

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

×

483

    return code;

×

486

  code = taosThreadMutexAttrSetType(&attr, PTHREAD_MUTEX_RECURSIVE);

×

487

  if (code != 0) {

×

488

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

×

489

    return code;

×

492

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

×

493

  if (code) {

×

494

    return code;

×

497

  code = taosThreadMutexAttrDestroy(&attr);

×

498

  if (code) {

×

499

    return code;

×

502

  streamTaskOpenAllUpstreamInput(pTask);

×

504

  STaskOutputInfo* pOutputInfo = &pTask->outputInfo;

×

505

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

×

506

  if (pOutputInfo->pTokenBucket == NULL) {

×

507

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

×

508

    return terrno;

×

513

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

×

514

  if (code) {

×

515

    return code;

×

518

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

×

519

  if (pOutputInfo->pNodeEpsetUpdateList == NULL) {

×

520

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

×

521

    return terrno;

×

524

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

×

525

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

×

526

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

×

527

    return terrno;

×

530

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

×

531

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

×

532

    if (code) {

×

533

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

×

534

      return code;

×

538

  return streamTaskSetBackendPath(pTask);

×

541

int32_t streamTaskGetNumOfDownstream(const SStreamTask* pTask) {

×

542

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

×

543

    return 0;

×

546

  int32_t type = pTask->outputInfo.type;

×

547

  if (type == TASK_OUTPUT__TABLE) {

×

548

    return 0;

×

549

  } else if (type == TASK_OUTPUT__FIXED_DISPATCH) {

×

550

    return 1;

×

552

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

×

553

    return taosArrayGetSize(vgInfo);

×

557

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

×

559

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

×

560

  SStreamUpstreamEpInfo* pEpInfo = createStreamTaskEpInfo(pUpstreamTask);

×

561

  if (pEpInfo == NULL) {

×

562

    return terrno;

×

565

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

×

566

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

×

569

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

×

570

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

×

573

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

×

574

  int32_t code = 0;

×

575

  char    buf[512] = {0};

×

576

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

×

577

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

×

578

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

×

579

    return code;

×

582

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

×

583

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

×

584

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

×

585

    if (pInfo->nodeId == nodeId) {

×

586

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

×

587

      if (!equal) {

×

588

        *pUpdated = true;

×

590

        char tmp[512] = {0};

×

591

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

×

592

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

×

593

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

×

594

          return code;

×

597

        epsetAssign(&pInfo->epSet, pEpSet);

×

598

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

×

601

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

×

605

      break;

×

609

  return code;

×

612

void streamTaskDestroyUpstreamInfo(SUpstreamInfo* pUpstreamInfo) {

×

613

  if (pUpstreamInfo->pList != NULL) {

×

614

    taosArrayDestroyEx(pUpstreamInfo->pList, freeUpstreamItem);

×

615

    pUpstreamInfo->numOfClosed = 0;

×

616

    pUpstreamInfo->pList = NULL;

×

618

×

620

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

×

621

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

×

622

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

×

623

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

×

624

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

×

626

  pTask->outputInfo.type = TASK_OUTPUT__FIXED_DISPATCH;

×

627

  pTask->msgInfo.msgType = TDMT_STREAM_TASK_DISPATCH;

×

628

×

630

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

×

631

  char    buf[512] = {0};

×

632

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

×

633

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

×

634

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

×

635

    return code;

×

638

  int32_t id = pTask->id.taskId;

×

639

  int8_t  type = pTask->outputInfo.type;

×

641

  if (type == TASK_OUTPUT__SHUFFLE_DISPATCH) {

×

642

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

×

644

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

×

645

      SVgroupInfo* pVgInfo = taosArrayGet(pVgs, i);

×

646

      if (pVgInfo == NULL) {

×

647

        continue;

×

650

      if (pVgInfo->vgId == nodeId) {

×

651

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

×

652

        if (!isEqual) {

×

653

          *pUpdated = true;

×

655

          char tmp[512] = {0};

×

656

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

×

657

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

×

658

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

×

659

            return code;

×

662

          epsetAssign(&pVgInfo->epSet, pEpSet);

×

663

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

×

666

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

×

669

        break;

×

672

  } else if (type == TASK_OUTPUT__FIXED_DISPATCH) {

×

673

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

×

674

    if (pDispatcher->nodeId == nodeId) {

×

675

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

×

676

      if (!equal) {

×

677

        *pUpdated = true;

×

679

        char tmp[512] = {0};

×

680

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

×

681

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

×

682

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

×

683

          return code;

×

686

        epsetAssign(&pDispatcher->epSet, pEpSet);

×

687

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

×

690

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

×

696

  return code;

×

699

int32_t streamTaskStop(SStreamTask* pTask) {

×

700

  int32_t     vgId = pTask->pMeta->vgId;

×

701

  int64_t     st = taosGetTimestampMs();

×

702

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

×

704

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

×

705

  if (code) {

×

706

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

×

707

    return code;

×

710

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

×

711

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

×

712

    if (code != TSDB_CODE_SUCCESS) {

×

713

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

×

717

  while (!streamTaskIsIdle(pTask)) {

×

718

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

×

720

    taosMsleep(100);

×

723

  int64_t el = taosGetTimestampMs() - st;

×

724

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

×

725

  return code;

×

728

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

×

729

  STaskExecStatisInfo* p = &pTask->execInfo;

×

731

  int32_t numOfNodes = taosArrayGetSize(pNodeList);

×

732

  int64_t prevTs = p->latestUpdateTs;

×

734

  p->latestUpdateTs = taosGetTimestampMs();

×

735

  p->updateCount += 1;

×

736

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

×

739

  bool updated = false;

×

740

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

×

741

    SNodeUpdateInfo* pInfo = taosArrayGet(pNodeList, i);

×

742

    if (pInfo == NULL) {

×

743

      continue;

×

746

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

×

747

    if (code) {

×

748

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

×

752

  return updated;

×

755

void streamTaskResetUpstreamStageInfo(SStreamTask* pTask) {

×

756

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

×

757

    return;

×

760

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

×

761

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

×

762

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

×

763

    pInfo->stage = -1;

×

766

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

×

769

void streamTaskOpenAllUpstreamInput(SStreamTask* pTask) {

×

770

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

×

771

  if (num == 0) {

×

772

    return;

×

775

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

×

776

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

×

777

    pInfo->dataAllowed = true;

×

780

  pTask->upstreamInfo.numOfClosed = 0;

×

781

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

×

784

void streamTaskCloseUpstreamInput(SStreamTask* pTask, int32_t taskId) {

×

785

  SStreamUpstreamEpInfo* pInfo = NULL;

×

786

  streamTaskGetUpstreamTaskEpInfo(pTask, taskId, &pInfo);

×

788

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

×

789

    pInfo->dataAllowed = false;

×

790

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

×

791

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

×

793

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

×

796

×

798

void streamTaskOpenUpstreamInput(SStreamTask* pTask, int32_t taskId) {

×

799

  SStreamUpstreamEpInfo* pInfo = NULL;

×

800

  streamTaskGetUpstreamTaskEpInfo(pTask, taskId, &pInfo);

×

802

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

×

803

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

×

804

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

×

805

    pInfo->dataAllowed = true;

×

807

×

809

bool streamTaskIsAllUpstreamClosed(SStreamTask* pTask) {

×

810

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

×

813

bool streamTaskSetSchedStatusWait(SStreamTask* pTask) {

×

814

  bool ret = false;

×

816

  streamMutexLock(&pTask->lock);

×

817

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

×

818

    pTask->status.schedStatus = TASK_SCHED_STATUS__WAITING;

×

819

    ret = true;

×

822

  streamMutexUnlock(&pTask->lock);

×

823

  return ret;

×

826

int8_t streamTaskSetSchedStatusActive(SStreamTask* pTask) {

×

827

  streamMutexLock(&pTask->lock);

×

828

  int8_t status = pTask->status.schedStatus;

×

829

  if (status == TASK_SCHED_STATUS__WAITING) {

×

830

    pTask->status.schedStatus = TASK_SCHED_STATUS__ACTIVE;

×

832

  streamMutexUnlock(&pTask->lock);

×

834

  return status;

×

837

int8_t streamTaskSetSchedStatusInactive(SStreamTask* pTask) {

×

838

  streamMutexLock(&pTask->lock);

×

839

  int8_t status = pTask->status.schedStatus;

×

840

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

841

  streamMutexUnlock(&pTask->lock);

×

843

  return status;

×

846

int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t resetRelHalt) {

×

847

  int32_t      code = 0;

×

848

  SStreamMeta* pMeta = pTask->pMeta;

×

849

  SStreamTask* pStreamTask = NULL;

×

851

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

×

852

    return code;

×

855

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

×

856

  if (code == 0) {

×

857

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

×

860

    streamMutexLock(&(pStreamTask->lock));

×

861

    CLEAR_RELATED_FILLHISTORY_TASK(pStreamTask);

×

863

    if (resetRelHalt) {

×

864

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

×

867

      pStreamTask->status.taskStatus = TASK_STATUS__READY;

×

870

    code = streamMetaSaveTask(pMeta, pStreamTask);

×

871

    streamMutexUnlock(&(pStreamTask->lock));

×

873

    streamMetaReleaseTask(pMeta, pStreamTask);

×

876

  return code;

×

879

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

×

880

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

×

881

  if (pReq == NULL) {

×

882

    return terrno;

×

885

  pReq->head.vgId = vgId;

×

886

  pReq->taskId = pTaskId->taskId;

×

887

  pReq->streamId = pTaskId->streamId;

×

888

  pReq->resetRelHalt = resetRelHalt;

×

890

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

×

891

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

×

892

  if (code != TSDB_CODE_SUCCESS) {

×

893

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

×

895

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

×

898

  return code;

×

901

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

×

902

  int32_t                code = 0;

×

903

  int32_t                tlen = 0;

×

904

  int32_t                vgId = pTask->pMeta->vgId;

×

905

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

×

906

  SActiveCheckpointInfo* pActive = pCheckpointInfo->pActiveInfo;

×

908

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

×

909

                           .taskId = pTask->id.taskId,

×

912

                           .transId = pActive->transId,

×

913

                           .checkpointId = pActive->activeId,

×

914

                           .checkpointVer = pCheckpointInfo->processedVer,

×

915

                           .checkpointTs = pCheckpointInfo->startTs};

×

917

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

×

918

  if (code < 0) {

×

919

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

×

920

    return -1;

×

923

  void* buf = rpcMallocCont(tlen);

×

924

  if (buf == NULL) {

×

925

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

×

927

    return -1;

×

931

  tEncoderInit(&encoder, buf, tlen);

×

932

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

×

933

    rpcFreeCont(buf);

×

934

    tEncoderClear(&encoder);

×

935

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

×

936

    return -1;

×

938

  tEncoderClear(&encoder);

×

940

  SRpcMsg msg = {0};

×

941

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

×

942

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

×

944

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

×

947

STaskId streamTaskGetTaskId(const SStreamTask* pTask) {

×

948

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

×

949

  return id;

×

952

void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo) {

×

953

  pInfo->waitInterval = LAUNCH_HTASK_INTERVAL;

×

954

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

×

955

  pInfo->retryTimes = 0;

×

956

×

958

void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo) {

×

959

  pInfo->waitInterval *= RETRY_LAUNCH_INTERVAL_INC_RATE;

×

960

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

×

961

  pInfo->retryTimes += 1;

×

962

×

964

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

×

965

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

×

966

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

×

967

  pEntry->stage = -1;

×

968

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

×

969

  pEntry->status = TASK_STATUS__STOP;

×

970

×

972

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

×

973

  pDst->stage = pSrc->stage;

×

974

  pDst->inputQUsed = pSrc->inputQUsed;

×

975

  pDst->inputRate = pSrc->inputRate;

×

976

  pDst->procsTotal = pSrc->procsTotal;

×

977

  pDst->procsThroughput = pSrc->procsThroughput;

×

978

  pDst->outputTotal = pSrc->outputTotal;

×

979

  pDst->outputThroughput = pSrc->outputThroughput;

×

980

  pDst->processedVer = pSrc->processedVer;

×

981

  pDst->verRange = pSrc->verRange;

×

982

  pDst->sinkQuota = pSrc->sinkQuota;

×

983

  pDst->sinkDataSize = pSrc->sinkDataSize;

×

984

  pDst->checkpointInfo = pSrc->checkpointInfo;

×

985

  pDst->startCheckpointId = pSrc->startCheckpointId;

×

986

  pDst->startCheckpointVer = pSrc->startCheckpointVer;

×

987

  pDst->status = pSrc->status;

×

989

  pDst->startTime = pSrc->startTime;

×

990

  pDst->hTaskId = pSrc->hTaskId;

×

991

×

993

STaskStatusEntry streamTaskGetStatusEntry(SStreamTask* pTask) {

×

994

  SStreamMeta*         pMeta = pTask->pMeta;

×

995

  STaskExecStatisInfo* pExecInfo = &pTask->execInfo;

×

997

  STaskStatusEntry entry = {

×

998

      .id = streamTaskGetTaskId(pTask),

×

999

      .status = streamTaskGetStatus(pTask).state,

×

1000

      .nodeId = pMeta->vgId,

×

1001

      .stage = pMeta->stage,

×

1003

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

×

1004

      .startTime = pExecInfo->readyTs,

×

1005

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

×

1006

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

×

1007

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

×

1012

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

×

1013

      .procsTotal = SIZE_IN_MiB(pExecInfo->inputDataSize),

×

1014

      .outputTotal = SIZE_IN_MiB(pExecInfo->outputDataSize),

×

1015

      .procsThroughput = SIZE_IN_KiB(pExecInfo->procsThroughput),

×

1016

      .outputThroughput = SIZE_IN_KiB(pExecInfo->outputThroughput),

×

1017

      .startCheckpointId = pExecInfo->startCheckpointId,

×

1018

      .startCheckpointVer = pExecInfo->startCheckpointVer,

×

1020

  return entry;

×

1023

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

×

1024

  SStreamMeta* pMeta = pTask->pMeta;

×

1025

  int32_t      code = 0;

×

1027

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

×

1028

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

×

1031

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

×

1032

    void* pExecutor = pTask->exec.pExecutor;

×

1033

    code = qKillTask(pExecutor, TSDB_CODE_SUCCESS);

×

1036

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

×

1037

  return code;

×

1040

void streamTaskPause(SStreamTask* pTask) {

×

1041

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

×

1042

  if (code) {

×

1043

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

×

1045

×

1047

void streamTaskResume(SStreamTask* pTask) {

×

1048

  SStreamTaskState prevState = streamTaskGetStatus(pTask);

×

1050

  SStreamMeta* pMeta = pTask->pMeta;

×

1051

  int32_t      code = streamTaskRestoreStatus(pTask);

×

1052

  if (code == TSDB_CODE_SUCCESS) {

×

1053

    char*   pNew = streamTaskGetStatus(pTask).name;

×

1054

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

×

1055

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

×

1057

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

×

1060

×

1062

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

×

1065

int32_t streamTaskSendCheckpointReq(SStreamTask* pTask) {

×

1067

  int32_t     tlen = 0;

×

1068

  int32_t     vgId = pTask->pMeta->vgId;

×

1069

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

×

1071

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

×

1072

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

×

1073

  if (code < 0) {

×

1074

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

×

1075

    return TSDB_CODE_INVALID_MSG;

×

1078

  void* buf = rpcMallocCont(tlen);

×

1079

  if (buf == NULL) {

×

1080

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

×

1081

    return terrno;

×

1085

  tEncoderInit(&encoder, buf, tlen);

×

1086

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

×

1087

    rpcFreeCont(buf);

×

1088

    tEncoderClear(&encoder);

×

1089

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

×

1090

    return code;

×

1093

  tEncoderClear(&encoder);

×

1095

  SRpcMsg msg = {0};

×

1096

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

×

1097

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

×

1099

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

×

1102

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

×

1103

  *pEpInfo = NULL;

×

1105

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

×

1106

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

×

1107

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

×

1108

    if (pInfo == NULL) {

×

1109

      return;

×

1112

    if (pInfo->taskId == taskId) {

×

1113

      *pEpInfo = pInfo;

×

1114

      return;

×

1118

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

×

1121

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

×

1122

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

×

1123

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

×

1124

      return &pTask->outputInfo.fixedDispatcher.epSet;

×

1126

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

×

1127

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

×

1128

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

×

1129

      SVgroupInfo* pVgInfo = taosArrayGet(pList, i);

×

1130

      if (pVgInfo == NULL) {

×

1131

        continue;

×

1134

      if (pVgInfo->taskId == taskId) {

×

1135

        return &pVgInfo->epSet;

×

1140

  return NULL;

×

1143

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

×

1144

  char buf[128] = {0};

×

1145

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

×

1146

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

×

1147

    return TSDB_CODE_OUT_OF_BUFFER;

×

1150

  *pId = taosStrdup(buf);

×

1152

  if (*pId == NULL) {

×

1153

    return terrno;

×

1155

    return TSDB_CODE_SUCCESS;

×

1159

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

×

1163

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

×

1164

  if (code) {

×

1165

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

×

1166

    return terrno = code;

×

1169

  pData->type = STREAM_INPUT__DATA_RETRIEVE;

×

1170

  pData->srcVgId = 0;

×

1172

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

×

1173

  if (code != TSDB_CODE_SUCCESS) {

×

1174

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

×

1175

    taosFreeQitem(pData);

×

1176

    return code;

×

1179

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

×

1180

  if (code != TSDB_CODE_SUCCESS) {

×

1181

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

×

1185

  return code;

×

1188

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

×

1189

  int32_t code = streamTaskEnqueueRetrieve(pTask, pReq);

×

1190

  if (code != 0) {

×

1191

    return code;

×

1193

  return streamTrySchedExec(pTask);

×

1196

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

×

1198

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

×

1199

  if (pTransId != NULL) {

×

1200

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

×

1203

  if (pCheckpointId != NULL) {

×

1204

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

×

1206

×

1208

int32_t streamTaskSetActiveCheckpointInfo(SStreamTask* pTask, int64_t activeCheckpointId) {

×

1209

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

×

1210

  return TSDB_CODE_SUCCESS;

×

1213

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

×

1214

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

×

1215

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

×

1216

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

×

1217

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

×

1218

×

1220

int32_t streamTaskCreateActiveChkptInfo(SActiveCheckpointInfo** pRes) {

×

1221

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

×

1222

  if (pInfo == NULL) {

×

1223

    return terrno;

×

1226

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

×

1227

  if (code != TSDB_CODE_SUCCESS) {

×

1228

    return code;

×

1231

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

×

1232

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

×

1233

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

×

1235

  *pRes = pInfo;

×

1236

  return code;

×

1239

void streamTaskDestroyActiveChkptInfo(SActiveCheckpointInfo* pInfo) {

×

1240

  if (pInfo == NULL) {

×

1241

    return;

×

1244

  streamMutexDestroy(&pInfo->lock);

×

1245

  taosArrayDestroy(pInfo->pDispatchTriggerList);

×

1246

  pInfo->pDispatchTriggerList = NULL;

×

1247

  taosArrayDestroy(pInfo->pReadyMsgList);

×

1248

  pInfo->pReadyMsgList = NULL;

×

1249

  taosArrayDestroy(pInfo->pCheckpointReadyRecvList);

×

1250

  pInfo->pCheckpointReadyRecvList = NULL;

×

1252

  SStreamTmrInfo* pTriggerTmr = &pInfo->chkptTriggerMsgTmr;

×

1253

  if (pTriggerTmr->tmrHandle != NULL) {

×

1254

    streamTmrStop(pTriggerTmr->tmrHandle);

×

1255

    pTriggerTmr->tmrHandle = NULL;

×

1258

  SStreamTmrInfo* pReadyTmr = &pInfo->chkptReadyMsgTmr;

×

1259

  if (pReadyTmr->tmrHandle != NULL) {

×

1260

    streamTmrStop(pReadyTmr->tmrHandle);

×

1261

    pReadyTmr->tmrHandle = NULL;

×

1264

  taosMemoryFree(pInfo);

×

1269

void streamTaskClearActiveInfo(SActiveCheckpointInfo* pInfo) {

×

1270

  pInfo->activeId = 0;

×

1271

  pInfo->transId = 0;

×

1272

  pInfo->allUpstreamTriggerRecv = 0;

×

1273

  pInfo->dispatchTrigger = false;

×

1275

  taosArrayClear(pInfo->pDispatchTriggerList);

×

1276

  taosArrayClear(pInfo->pCheckpointReadyRecvList);

×

1277

×

1279

const char* streamTaskGetExecType(int32_t type) {

×

1280

  switch (type) {

×

1281

    case STREAM_EXEC_T_EXTRACT_WAL_DATA:

×

1282

      return "scan-wal-file";

×

1283

    case STREAM_EXEC_T_START_ALL_TASKS:

×

1284

      return "start-all-tasks";

×

1285

    case STREAM_EXEC_T_START_ONE_TASK:

×

1286

      return "start-one-task";

×

1287

    case STREAM_EXEC_T_RESTART_ALL_TASKS:

×

1288

      return "restart-all-tasks";

×

1289

    case STREAM_EXEC_T_STOP_ALL_TASKS:

×

1290

      return "stop-all-tasks";

×

1291

    case STREAM_EXEC_T_RESUME_TASK:

×

1292

      return "resume-task-from-idle";

×

1293

    case STREAM_EXEC_T_ADD_FAILED_TASK:

×

1294

      return "record-start-failed-task";

×

1295

    case 0:

×

1296

      return "exec-all-tasks";

×

1297

    default:

×

1298

      return "invalid-exec-type";

×

1302

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

×

1303

  *pRefId = taosMemoryMalloc(sizeof(int64_t));

×

1304

  if (*pRefId != NULL) {

×

1305

    **pRefId = pTask->id.refId;

×

1306

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

×

1307

    if (code != 0) {

×

1308

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

×

1311

    return code;

×

1313

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

×

1314

    return terrno;

×

1318

void streamTaskFreeRefId(int64_t* pRefId) {

×

1319

  if (pRefId == NULL) {

×

1320

    return;

×

1323

  metaRefMgtRemove(pRefId);

×

1326

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

×

1327

  int32_t code = TSDB_CODE_SUCCESS;

×

1328

  int32_t lino = 0;

×

1330

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

×

1331

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

×

1333

  int32_t addrSize = taosArrayGetSize(info->pNotifyAddrUrls);

×

1334

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

×

1335

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

×

1336

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

×

1337

    TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, url));

×

1339

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

×

1340

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

×

1341

  if (addrSize > 0) {

×

1342

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

×

1343

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

×

1344

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

×

1347

_exit:

×

1348

  if (code != TSDB_CODE_SUCCESS) {

×

1349

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

×

1351

  return code;

×

1354

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

×

1355

  int32_t code = TSDB_CODE_SUCCESS;

×

1356

  int32_t lino = 0;

×

1358

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

×

1359

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

×

1361

  int32_t addrSize = 0;

×

1362

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

×

1363

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

×

1364

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

×

1365

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

×

1366

    char *url = NULL;

×

1367

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &url));

×

1368

    url = taosStrndup(url, TSDB_STREAM_NOTIFY_URL_LEN);

×

1369

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

×

1370

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

×

1371

      taosMemoryFree(url);

×

1372

      TAOS_CHECK_EXIT(terrno);

×

1375

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

×

1376

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

×

1377

  if (addrSize > 0) {

×

1378

    char* name = NULL;

×

1379

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &name));

×

1380

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

×

1381

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

×

1382

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &name));

×

1383

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

×

1384

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

×

1385

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

×

1386

    if (info->pSchemaWrapper == NULL) {

×

1387

      TAOS_CHECK_EXIT(terrno);

×

1389

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

×

1392

_exit:

×

1393

  if (code != TSDB_CODE_SUCCESS) {

×

1394

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

×

1396

  return code;

×

1399

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

×

1400

  int32_t code = 0;

×

1403

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

×

1404

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

×

1405

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

×

1406

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

×

1407

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

×

1408

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

×

1409

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

×

1410

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

×

1412

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

×

1413

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

×

1415

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

×

1416

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

×

1417

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

×

1418

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

×

1420

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

×

1421

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

×

1422

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

×

1424

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

×

1425

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

×

1426

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, taskId));

×

1428

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

×

1429

  taskId = pTask->streamTaskId.taskId;

×

1430

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, taskId));

×

1432

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

×

1433

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

×

1434

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

×

1435

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

×

1437

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

×

1438

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, epSz));

×

1439

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

×

1440

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

×

1441

    TAOS_CHECK_EXIT(tEncodeStreamEpInfo(pEncoder, pInfo));

×

1444

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

×

1445

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

×

1448

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

×

1449

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

×

1450

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

×

1451

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

×

1452

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

×

1453

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

×

1454

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

×

1455

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

×

1456

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

×

1457

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

×

1458

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

×

1459

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

×

1460

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

×

1461

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

×

1462

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

×

1464

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

×

1465

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

×

1466

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

×

1468

  if (pTask->ver >= SSTREAM_TASK_ADD_NOTIFY_VER) {

×

1469

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

×

1472

  tEndEncode(pEncoder);

×

1473

_exit:

×

1474

  return code;

×

1477

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

×

1478

  int32_t taskId = 0;

×

1479

  int32_t code = 0;

×

1482

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

×

1483

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

×

1484

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

×

1485

    TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);

×

1488

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

×

1489

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

×

1490

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

×

1491

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

×

1492

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

×

1493

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

×

1495

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

×

1496

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

×

1498

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

×

1499

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

×

1500

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

×

1501

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

×

1503

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

×

1504

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

×

1505

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

×

1507

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

×

1508

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));

×

1509

  pTask->hTaskInfo.id.taskId = taskId;

×

1511

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

×

1512

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));

×

1513

  pTask->streamTaskId.taskId = taskId;

×

1515

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

×

1516

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

×

1517

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

×

1518

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

×

1520

  int32_t epSz = -1;

×

1521

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

×

1523

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

×

1524

    TAOS_CHECK_EXIT(terrno);

×

1526

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

×

1527

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

×

1528

    if (pInfo == NULL) {

×

1529

      TAOS_CHECK_EXIT(terrno);

×

1531

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

×

1532

      taosMemoryFreeClear(pInfo);

×

1533

      goto _exit;

×

1535

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

×

1536

      TAOS_CHECK_EXIT(terrno);

×

1540

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

×

1541

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

×

1544

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

×

1545

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

×

1546

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

×

1547

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

×

1548

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

×

1549

      TAOS_CHECK_EXIT(terrno);

×

1551

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

×

1552

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

×

1553

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

×

1554

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

×

1555

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

×

1556

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

×

1557

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

×

1558

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

×

1559

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

×

1560

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

×

1561

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

×

1562

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

×

1564

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

×

1565

  if (pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER) {

×

1566

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

×

1568

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

×

1570

  if (pTask->ver >= SSTREAM_TASK_ADD_NOTIFY_VER) {

×

1571

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

×

1574

  tEndDecode(pDecoder);

×

1576

_exit:

×

1577

  return code;

×

taosdata / TDengine / #3599

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

taosdata / TDengine / #3599

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