taosdata/TDengine | Build 3798 | 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;

×

103

  pEpInfo->lastMsgId = -1;

×

105

  return pEpInfo;

×

108

int32_t tNewStreamTask(int64_t streamId, int8_t taskLevel, SEpSet* pEpset, EStreamTaskType type, int32_t trigger,

×

111

  *p = NULL;

×

113

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

×

114

  if (pTask == NULL) {

×

115

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

×

117

    return terrno;

×

120

  pTask->ver = SSTREAM_TASK_VER;

×

121

  pTask->id.taskId = tGenIdPI32();

×

122

  pTask->id.streamId = streamId;

×

124

  pTask->info.taskLevel = taskLevel;

×

125

  pTask->info.fillHistory = type;

×

126

  pTask->info.trigger = trigger;

×

127

  pTask->info.hasAggTasks = hasAggTasks;

×

128

  pTask->info.delaySchedParam = triggerParam;

×

129

  pTask->subtableWithoutMd5 = subtableWithoutMd5;

×

131

  int32_t code = streamCreateStateMachine(pTask);

×

132

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

×

133

    taosMemoryFreeClear(pTask);

×

134

    return code;

×

137

  char    buf[128] = {0};

×

138

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

×

139

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

×

140

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

×

141

    return TSDB_CODE_OUT_OF_BUFFER;

×

144

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

×

145

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

×

146

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

×

147

    return terrno;

×

150

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

151

  pTask->status.taskStatus =

×

152

      (pTask->info.fillHistory == STREAM_HISTORY_TASK) ? TASK_STATUS__SCAN_HISTORY : TASK_STATUS__READY;

×

153

  pTask->inputq.status = TASK_INPUT_STATUS__NORMAL;

×

154

  pTask->outputq.status = TASK_OUTPUT_STATUS__NORMAL;

×

156

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

×

158

  if ((pTask->info.fillHistory == STREAM_HISTORY_TASK) && !hasFillhistory) {

×

159

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

×

160

    return TSDB_CODE_INVALID_PARA;

×

163

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

×

165

  code = addToTaskset(pTaskList, pTask);

×

166

  *p = pTask;

×

168

  return code;

×

171

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

×

178

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

×

179

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

×

182

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

×

183

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

×

184

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

×

185

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

×

186

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

×

187

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

×

189

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

×

190

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

×

192

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

×

193

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

×

194

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

×

195

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

×

197

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

×

198

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

×

200

  tEndDecode(pDecoder);

×

201

  return 0;

×

204

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

×

206

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

×

207

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

×

208

  if (ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1;

×

210

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

×

212

  int32_t taskId = 0;

×

213

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

×

215

  pTaskId->taskId = taskId;

×

216

  tEndDecode(pDecoder);

×

217

  return 0;

×

220

void tFreeStreamTask(void* pParam) {

×

221

  char*        p = NULL;

×

222

  SStreamTask* pTask = pParam;

×

223

  int32_t      taskId = pTask->id.taskId;

×

225

  STaskExecStatisInfo* pStatis = &pTask->execInfo;

×

227

  ETaskStatus status1 = TASK_STATUS__UNINIT;

×

228

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

×

229

    streamMutexLock(&pTask->lock);

×

230

    SStreamTaskState status = streamTaskGetStatus(pTask);

×

231

    p = status.name;

×

232

    status1 = status.state;

×

233

    streamMutexUnlock(&pTask->lock);

×

236

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

×

238

  SCheckpointInfo* pCkInfo = &pTask->chkInfo;

×

239

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

×

245

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

×

246

    streamTmrStop(pTask->schedInfo.pDelayTimer);

×

247

    pTask->schedInfo.pDelayTimer = NULL;

×

250

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

×

251

    streamTmrStop(pTask->hTaskInfo.pTimer);

×

252

    pTask->hTaskInfo.pTimer = NULL;

×

255

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

×

256

    streamTmrStop(pTask->msgInfo.pRetryTmr);

×

257

    pTask->msgInfo.pRetryTmr = NULL;

×

260

  if (pTask->inputq.queue) {

×

261

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

×

262

    pTask->inputq.queue = NULL;

×

265

  if (pTask->outputq.queue) {

×

266

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

×

267

    pTask->outputq.queue = NULL;

×

270

  if (pTask->exec.qmsg) {

×

271

    taosMemoryFree(pTask->exec.qmsg);

×

274

  if (pTask->exec.pExecutor) {

×

275

    qDestroyTask(pTask->exec.pExecutor);

×

276

    pTask->exec.pExecutor = NULL;

×

279

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

×

280

    walCloseReader(pTask->exec.pWalReader);

×

281

    pTask->exec.pWalReader = NULL;

×

284

  streamClearChkptReadyMsg(pTask->chkInfo.pActiveInfo);

×

286

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

×

287

    clearBufferedDispatchMsg(pTask);

×

290

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

×

291

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

×

292

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

×

293

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

×

294

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

×

295

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

×

296

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

×

297

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

×

298

    taosArrayDestroy(pTask->outputInfo.vtableMapDispatcher.taskInfos);

×

299

    tSimpleHashCleanup(pTask->outputInfo.vtableMapDispatcher.vtableMap);

×

302

  streamTaskCleanupCheckInfo(&pTask->taskCheckInfo);

×

303

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

×

305

  if (pTask->pNameMap) {

×

306

    tSimpleHashCleanup(pTask->pNameMap);

×

309

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

×

310

    streamMutexDestroy(&pTask->lock);

×

311

    streamMutexDestroy(&pTask->msgInfo.lock);

×

312

    streamMutexDestroy(&pTask->taskCheckInfo.checkInfoLock);

×

315

  taosArrayDestroy(pTask->pVTables);

×

316

  pTask->pVTables = NULL;

×

318

  streamDestroyStateMachine(pTask->status.pSM);

×

319

  pTask->status.pSM = NULL;

×

321

  streamTaskDestroyUpstreamInfo(&pTask->upstreamInfo);

×

323

  taosMemoryFree(pTask->outputInfo.pTokenBucket);

×

325

  taosArrayDestroy(pTask->msgInfo.pSendInfo);

×

326

  pTask->msgInfo.pSendInfo = NULL;

×

328

  taosArrayDestroy(pTask->outputInfo.pNodeEpsetUpdateList);

×

329

  pTask->outputInfo.pNodeEpsetUpdateList = NULL;

×

331

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

×

332

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

×

335

  streamTaskDestroyActiveChkptInfo(pTask->chkInfo.pActiveInfo);

×

336

  pTask->chkInfo.pActiveInfo = NULL;

×

338

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

×

339

  taosMemoryFreeClear(pTask->notifyInfo.streamName);

×

340

  taosMemoryFreeClear(pTask->notifyInfo.stbFullName);

×

341

  tDeleteSchemaWrapper(pTask->notifyInfo.pSchemaWrapper);

×

343

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

×

345

  taosMemoryFree(pTask);

×

346

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

×

347

×

349

void streamFreeTaskState(SStreamTask* pTask, int8_t remove) {

×

350

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

×

351

  if (pTask->pState != NULL) {

×

352

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

×

353

    streamStateClose(pTask->pState, remove);

×

355

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

×

356

    taskDbRemoveRef(pTask->pBackend);

×

357

    pTask->pBackend = NULL;

×

358

    pTask->pState = NULL;

×

361

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

×

363

    if (remove) {

×

364

      if (pTask->backendPath != NULL) {

×

365

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

×

366

        taosRemoveDir(pTask->backendPath);

×

371

  if (pTask->backendPath != NULL) {

×

372

    taosMemoryFree(pTask->backendPath);

×

373

    pTask->backendPath = NULL;

×

377

  if (pTask->pRecalState != NULL) {

×

378

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

×

379

    streamStateClose(pTask->pRecalState, remove);

×

381

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

×

382

    taskDbRemoveRef(pTask->pRecalBackend);

×

383

    pTask->pRecalBackend = NULL;

×

384

    pTask->pRecalState = NULL;

×

400

×

402

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

×

403

  SCheckpointInfo* pChkInfo = &pTask->chkInfo;

×

404

  SDataRange*      pRange = &pTask->dataRange;

×

407

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

×

408

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

×

409

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

×

410

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

×

412

    pRange->range.maxVer = ver;

×

413

    pRange->range.minVer = ver;

×

417

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

×

418

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

×

419

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

×

420

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

×

422

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

×

423

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

×

424

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

×

427

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

×

428

          pChkInfo->checkpointVer = 0;

×

429

          pChkInfo->processedVer = 0;

×

430

          pChkInfo->nextProcessVer = 1;

×

431

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

×

436

×

438

int32_t streamTaskSetBackendPath(SStreamTask* pTask) {

×

439

  int64_t streamId = 0;

×

440

  int32_t taskId = 0;

×

442

  if (pTask->info.fillHistory) {

×

443

    streamId = pTask->streamTaskId.streamId;

×

444

    taskId = pTask->streamTaskId.taskId;

×

446

    streamId = pTask->id.streamId;

×

447

    taskId = pTask->id.taskId;

×

450

  char    id[128] = {0};

×

451

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

×

452

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

×

453

    return TSDB_CODE_OUT_OF_BUFFER;

×

456

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

×

457

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

×

458

  if (pTask->backendPath == NULL) {

×

459

    return terrno;

×

462

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

×

463

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

×

464

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

×

465

    return TSDB_CODE_OUT_OF_BUFFER;

×

467

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

×

468

    return 0;

×

472

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

×

473

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

×

474

  if (code) {

×

475

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

×

476

    return code;

×

479

  pTask->id.refId = 0;

×

480

  pTask->inputq.status = TASK_INPUT_STATUS__NORMAL;

×

481

  pTask->outputq.status = TASK_OUTPUT_STATUS__NORMAL;

×

483

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

×

484

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

×

485

  if (code1 || code2) {

×

486

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

×

487

    return TSDB_CODE_OUT_OF_MEMORY;

×

490

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

492

  code = streamCreateStateMachine(pTask);

×

493

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

×

494

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

×

496

    return code;

×

499

  pTask->execInfo.created = taosGetTimestampMs();

×

500

  setInitialVersionInfo(pTask, ver);

×

502

  pTask->pMeta = pMeta;

×

503

  pTask->pMsgCb = pMsgCb;

×

504

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

×

505

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

×

506

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

×

507

    return terrno;

×

510

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

×

511

  if (code) {

×

512

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

×

513

    return code;

×

516

  TdThreadMutexAttr attr = {0};

×

517

  code = taosThreadMutexAttrInit(&attr);

×

518

  if (code != 0) {

×

519

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

×

520

    return code;

×

523

  code = taosThreadMutexAttrSetType(&attr, PTHREAD_MUTEX_RECURSIVE);

×

524

  if (code != 0) {

×

525

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

×

526

    return code;

×

529

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

×

530

  if (code) {

×

531

    return code;

×

534

  code = taosThreadMutexAttrDestroy(&attr);

×

535

  if (code) {

×

536

    return code;

×

539

  streamTaskOpenAllUpstreamInput(pTask);

×

541

  STaskOutputInfo* pOutputInfo = &pTask->outputInfo;

×

542

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

×

543

  if (pOutputInfo->pTokenBucket == NULL) {

×

544

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

×

545

    return terrno;

×

550

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

×

551

  if (code) {

×

552

    return code;

×

555

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

×

556

  if (pOutputInfo->pNodeEpsetUpdateList == NULL) {

×

557

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

×

558

    return terrno;

×

561

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

×

562

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

×

563

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

×

564

    return terrno;

×

567

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

×

568

  if (code) {

×

569

    return code;

×

572

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

×

573

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

×

574

    if (code) {

×

575

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

×

576

      return code;

×

580

  return streamTaskSetBackendPath(pTask);

×

583

int32_t streamTaskGetNumOfDownstream(const SStreamTask* pTask) {

×

584

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

×

585

    return 0;

×

588

  int32_t type = pTask->outputInfo.type;

×

589

  if (type == TASK_OUTPUT__TABLE) {

×

590

    return 0;

×

591

  } else if (type == TASK_OUTPUT__FIXED_DISPATCH) {

×

592

    return 1;

×

593

  } else if (type == TASK_OUTPUT__VTABLE_MAP) {

×

594

    SArray* pTaskInfos = pTask->outputInfo.vtableMapDispatcher.taskInfos;

×

595

    return taosArrayGetSize(pTaskInfos);

×

597

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

×

598

    return taosArrayGetSize(vgInfo);

×

602

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

×

604

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

×

605

  SStreamUpstreamEpInfo* pEpInfo = createStreamTaskEpInfo(pUpstreamTask);

×

606

  if (pEpInfo == NULL) {

×

607

    return terrno;

×

610

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

×

611

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

×

614

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

×

615

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

×

618

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

×

619

  int32_t code = 0;

×

620

  char    buf[512] = {0};

×

621

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

×

622

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

×

623

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

×

624

    return code;

×

627

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

×

628

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

×

629

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

×

630

    if (pInfo->nodeId == nodeId) {

×

631

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

×

632

      if (!equal) {

×

633

        *pUpdated = true;

×

635

        char tmp[512] = {0};

×

636

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

×

637

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

×

638

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

×

639

          return code;

×

642

        epsetAssign(&pInfo->epSet, pEpSet);

×

643

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

×

646

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

×

650

      break;

×

654

  return code;

×

657

void streamTaskDestroyUpstreamInfo(SUpstreamInfo* pUpstreamInfo) {

×

658

  if (pUpstreamInfo->pList != NULL) {

×

659

    taosArrayDestroyEx(pUpstreamInfo->pList, freeUpstreamItem);

×

660

    pUpstreamInfo->numOfClosed = 0;

×

661

    pUpstreamInfo->pList = NULL;

×

663

×

665

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

×

666

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

×

667

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

×

668

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

×

669

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

×

671

  pTask->outputInfo.type = TASK_OUTPUT__FIXED_DISPATCH;

×

672

  pTask->msgInfo.msgType = TDMT_STREAM_TASK_DISPATCH;

×

673

×

675

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

×

676

  char    buf[512] = {0};

×

677

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

×

678

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

×

679

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

×

680

    return code;

×

683

  int32_t id = pTask->id.taskId;

×

684

  int8_t  type = pTask->outputInfo.type;

×

686

  if (type == TASK_OUTPUT__SHUFFLE_DISPATCH) {

×

687

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

×

689

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

×

690

      SVgroupInfo* pVgInfo = taosArrayGet(pVgs, i);

×

691

      if (pVgInfo == NULL) {

×

692

        continue;

×

695

      if (pVgInfo->vgId == nodeId) {

×

696

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

×

697

        if (!isEqual) {

×

698

          *pUpdated = true;

×

700

          char tmp[512] = {0};

×

701

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

×

702

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

×

703

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

×

704

            return code;

×

707

          epsetAssign(&pVgInfo->epSet, pEpSet);

×

708

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

×

711

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

×

714

        break;

×

717

  } else if (type == TASK_OUTPUT__FIXED_DISPATCH) {

×

718

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

×

719

    if (pDispatcher->nodeId == nodeId) {

×

720

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

×

721

      if (!equal) {

×

722

        *pUpdated = true;

×

724

        char tmp[512] = {0};

×

725

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

×

726

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

×

727

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

×

728

          return code;

×

731

        epsetAssign(&pDispatcher->epSet, pEpSet);

×

732

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

×

735

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

×

739

  } else if (type == TASK_OUTPUT__VTABLE_MAP) {

×

740

    SArray* pTaskInfos = pTask->outputInfo.vtableMapDispatcher.taskInfos;

×

742

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

×

743

      STaskDispatcherFixed* pAddr = taosArrayGet(pTaskInfos, i);

×

744

      if (pAddr == NULL) {

×

745

        continue;

×

748

      if (pAddr->nodeId == nodeId) {

×

749

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

×

750

        if (!isEqual) {

×

751

          *pUpdated = true;

×

753

          char tmp[512] = {0};

×

754

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

×

755

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

×

756

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

×

757

            return code;

×

760

          epsetAssign(&pAddr->epSet, pEpSet);

×

761

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

×

764

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

×

767

        break;

×

772

  return code;

×

775

int32_t streamTaskStop(SStreamTask* pTask) {

×

776

  int32_t     vgId = pTask->pMeta->vgId;

×

777

  int64_t     st = taosGetTimestampMs();

×

778

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

×

780

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

×

781

  if (code) {

×

782

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

×

783

    return code;

×

786

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

×

787

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

×

788

    if (code != TSDB_CODE_SUCCESS) {

×

789

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

×

793

  while (!streamTaskIsIdle(pTask)) {

×

794

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

×

796

    taosMsleep(100);

×

799

  int64_t el = taosGetTimestampMs() - st;

×

800

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

×

801

  return code;

×

804

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

×

805

  STaskExecStatisInfo* p = &pTask->execInfo;

×

807

  int32_t numOfNodes = taosArrayGetSize(pNodeList);

×

808

  int64_t prevTs = p->latestUpdateTs;

×

810

  p->latestUpdateTs = taosGetTimestampMs();

×

811

  p->updateCount += 1;

×

812

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

×

815

  bool updated = false;

×

816

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

×

817

    SNodeUpdateInfo* pInfo = taosArrayGet(pNodeList, i);

×

818

    if (pInfo == NULL) {

×

819

      continue;

×

822

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

×

823

    if (code) {

×

824

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

×

828

  return updated;

×

831

void streamTaskResetUpstreamStageInfo(SStreamTask* pTask) {

×

832

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

×

833

    return;

×

836

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

×

837

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

×

838

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

×

839

    pInfo->stage = -1;

×

842

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

×

845

void streamTaskOpenAllUpstreamInput(SStreamTask* pTask) {

×

846

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

×

847

  if (num == 0) {

×

848

    return;

×

851

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

×

852

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

×

853

    pInfo->dataAllowed = true;

×

856

  pTask->upstreamInfo.numOfClosed = 0;

×

857

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

×

860

void streamTaskCloseUpstreamInput(SStreamTask* pTask, int32_t taskId) {

×

861

  SStreamUpstreamEpInfo* pInfo = NULL;

×

862

  streamTaskGetUpstreamTaskEpInfo(pTask, taskId, &pInfo);

×

864

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

×

865

    pInfo->dataAllowed = false;

×

866

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

×

867

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

×

869

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

×

872

×

874

void streamTaskOpenUpstreamInput(SStreamTask* pTask, int32_t taskId) {

×

875

  SStreamUpstreamEpInfo* pInfo = NULL;

×

876

  streamTaskGetUpstreamTaskEpInfo(pTask, taskId, &pInfo);

×

878

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

×

879

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

×

880

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

×

881

    pInfo->dataAllowed = true;

×

883

×

885

bool streamTaskIsAllUpstreamClosed(SStreamTask* pTask) {

×

886

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

×

889

bool streamTaskSetSchedStatusWait(SStreamTask* pTask) {

×

890

  bool ret = false;

×

892

  streamMutexLock(&pTask->lock);

×

893

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

×

894

    pTask->status.schedStatus = TASK_SCHED_STATUS__WAITING;

×

895

    ret = true;

×

898

  streamMutexUnlock(&pTask->lock);

×

899

  return ret;

×

902

int8_t streamTaskSetSchedStatusActive(SStreamTask* pTask) {

×

903

  streamMutexLock(&pTask->lock);

×

904

  int8_t status = pTask->status.schedStatus;

×

905

  if (status == TASK_SCHED_STATUS__WAITING) {

×

906

    pTask->status.schedStatus = TASK_SCHED_STATUS__ACTIVE;

×

908

  streamMutexUnlock(&pTask->lock);

×

910

  return status;

×

913

int8_t streamTaskSetSchedStatusInactive(SStreamTask* pTask) {

×

914

  streamMutexLock(&pTask->lock);

×

915

  int8_t status = pTask->status.schedStatus;

×

916

  pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;

×

917

  streamMutexUnlock(&pTask->lock);

×

919

  return status;

×

922

int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t resetRelHalt) {

×

923

  int32_t      code = 0;

×

924

  SStreamMeta* pMeta = pTask->pMeta;

×

925

  SStreamTask* pStreamTask = NULL;

×

927

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

×

928

    return code;

×

931

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

×

932

  if (code == 0) {

×

933

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

×

936

    streamMutexLock(&(pStreamTask->lock));

×

937

    CLEAR_RELATED_FILLHISTORY_TASK(pStreamTask);

×

939

    if (resetRelHalt) {

×

940

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

×

943

      pStreamTask->status.taskStatus = TASK_STATUS__READY;

×

946

    code = streamMetaSaveTaskInMeta(pMeta, pStreamTask);

×

947

    streamMutexUnlock(&(pStreamTask->lock));

×

949

    streamMetaReleaseTask(pMeta, pStreamTask);

×

952

  return code;

×

955

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

×

956

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

×

957

  if (pReq == NULL) {

×

958

    return terrno;

×

961

  pReq->head.vgId = vgId;

×

962

  pReq->taskId = pTaskId->taskId;

×

963

  pReq->streamId = pTaskId->streamId;

×

964

  pReq->resetRelHalt = resetRelHalt;

×

966

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

×

967

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

×

968

  if (code != TSDB_CODE_SUCCESS) {

×

969

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

×

971

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

×

974

  return code;

×

977

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

×

978

  int32_t                code = 0;

×

979

  int32_t                tlen = 0;

×

980

  int32_t                vgId = pTask->pMeta->vgId;

×

981

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

×

982

  SActiveCheckpointInfo* pActive = pCheckpointInfo->pActiveInfo;

×

984

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

×

985

                           .taskId = pTask->id.taskId,

×

988

                           .transId = pActive->transId,

×

989

                           .checkpointId = pActive->activeId,

×

990

                           .checkpointVer = pCheckpointInfo->processedVer,

×

991

                           .checkpointTs = pCheckpointInfo->startTs};

×

993

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

×

994

  if (code < 0) {

×

995

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

×

996

    return -1;

×

999

  void* buf = rpcMallocCont(tlen);

×

1000

  if (buf == NULL) {

×

1001

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

×

1003

    return -1;

×

1007

  tEncoderInit(&encoder, buf, tlen);

×

1008

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

×

1009

    rpcFreeCont(buf);

×

1010

    tEncoderClear(&encoder);

×

1011

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

×

1012

    return -1;

×

1014

  tEncoderClear(&encoder);

×

1016

  SRpcMsg msg = {0};

×

1017

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

×

1018

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

×

1020

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

×

1023

STaskId streamTaskGetTaskId(const SStreamTask* pTask) {

×

1024

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

×

1025

  return id;

×

1028

void streamTaskInitForLaunchHTask(SHistoryTaskInfo* pInfo) {

×

1029

  pInfo->waitInterval = LAUNCH_HTASK_INTERVAL;

×

1030

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

×

1031

  pInfo->retryTimes = 0;

×

1032

×

1034

void streamTaskSetRetryInfoForLaunch(SHistoryTaskInfo* pInfo) {

×

1035

  pInfo->waitInterval *= RETRY_LAUNCH_INTERVAL_INC_RATE;

×

1036

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

×

1037

  pInfo->retryTimes += 1;

×

1038

×

1040

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

×

1041

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

×

1042

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

×

1043

  pEntry->stage = -1;

×

1044

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

×

1045

  pEntry->status = TASK_STATUS__STOP;

×

1046

×

1048

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

×

1049

  pDst->stage = pSrc->stage;

×

1050

  pDst->inputQUsed = pSrc->inputQUsed;

×

1051

  pDst->inputRate = pSrc->inputRate;

×

1052

  pDst->procsTotal = pSrc->procsTotal;

×

1053

  pDst->procsThroughput = pSrc->procsThroughput;

×

1054

  pDst->outputTotal = pSrc->outputTotal;

×

1055

  pDst->outputThroughput = pSrc->outputThroughput;

×

1056

  pDst->processedVer = pSrc->processedVer;

×

1057

  pDst->verRange = pSrc->verRange;

×

1058

  pDst->sinkQuota = pSrc->sinkQuota;

×

1059

  pDst->sinkDataSize = pSrc->sinkDataSize;

×

1060

  pDst->checkpointInfo = pSrc->checkpointInfo;

×

1061

  pDst->startCheckpointId = pSrc->startCheckpointId;

×

1062

  pDst->startCheckpointVer = pSrc->startCheckpointVer;

×

1063

  pDst->status = pSrc->status;

×

1065

  pDst->startTime = pSrc->startTime;

×

1066

  pDst->hTaskId = pSrc->hTaskId;

×

1067

  pDst->notifyEventStat = pSrc->notifyEventStat;

×

1068

×

1070

STaskStatusEntry streamTaskGetStatusEntry(SStreamTask* pTask) {

×

1071

  SStreamMeta*         pMeta = pTask->pMeta;

×

1072

  STaskExecStatisInfo* pExecInfo = &pTask->execInfo;

×

1074

  STaskStatusEntry entry = {

×

1075

      .id = streamTaskGetTaskId(pTask),

×

1076

      .status = streamTaskGetStatus(pTask).state,

×

1077

      .nodeId = pMeta->vgId,

×

1078

      .stage = pMeta->stage,

×

1080

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

×

1081

      .startTime = pExecInfo->readyTs,

×

1082

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

×

1083

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

×

1084

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

×

1089

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

×

1090

      .procsTotal = SIZE_IN_MiB(pExecInfo->inputDataSize),

×

1091

      .outputTotal = SIZE_IN_MiB(pExecInfo->outputDataSize),

×

1092

      .procsThroughput = SIZE_IN_KiB(pExecInfo->procsThroughput),

×

1093

      .outputThroughput = SIZE_IN_KiB(pExecInfo->outputThroughput),

×

1094

      .startCheckpointId = pExecInfo->startCheckpointId,

×

1095

      .startCheckpointVer = pExecInfo->startCheckpointVer,

×

1098

  return entry;

×

1101

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

×

1102

  SStreamMeta* pMeta = pTask->pMeta;

×

1103

  int32_t      code = 0;

×

1105

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

×

1106

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

×

1109

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

×

1110

    void* pExecutor = pTask->exec.pExecutor;

×

1111

    code = qKillTask(pExecutor, TSDB_CODE_SUCCESS, 10000);

×

1114

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

×

1115

  return code;

×

1118

void streamTaskPause(SStreamTask* pTask) {

×

1119

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

×

1120

  if (code) {

×

1121

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

×

1123

×

1125

void streamTaskResume(SStreamTask* pTask) {

×

1126

  SStreamTaskState prevState = streamTaskGetStatus(pTask);

×

1128

  SStreamMeta* pMeta = pTask->pMeta;

×

1129

  int32_t      code = streamTaskRestoreStatus(pTask);

×

1130

  if (code == TSDB_CODE_SUCCESS) {

×

1131

    char*   pNew = streamTaskGetStatus(pTask).name;

×

1132

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

×

1133

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

×

1135

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

×

1138

×

1140

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

×

1143

int32_t streamTaskSendCheckpointReq(SStreamTask* pTask) {

×

1145

  int32_t     tlen = 0;

×

1146

  int32_t     vgId = pTask->pMeta->vgId;

×

1147

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

×

1149

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

×

1150

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

×

1151

  if (code < 0) {

×

1152

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

×

1153

    return TSDB_CODE_INVALID_MSG;

×

1156

  void* buf = rpcMallocCont(tlen);

×

1157

  if (buf == NULL) {

×

1158

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

×

1159

    return terrno;

×

1163

  tEncoderInit(&encoder, buf, tlen);

×

1164

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

×

1165

    rpcFreeCont(buf);

×

1166

    tEncoderClear(&encoder);

×

1167

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

×

1168

    return code;

×

1171

  tEncoderClear(&encoder);

×

1173

  SRpcMsg msg = {0};

×

1174

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

×

1175

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

×

1177

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

×

1180

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

×

1181

  *pEpInfo = NULL;

×

1183

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

×

1184

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

×

1185

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

×

1186

    if (pInfo == NULL) {

×

1187

      return;

×

1190

    if (pInfo->taskId == taskId) {

×

1191

      *pEpInfo = pInfo;

×

1192

      return;

×

1196

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

×

1199

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

×

1200

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

×

1201

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

×

1202

      return &pTask->outputInfo.fixedDispatcher.epSet;

×

1204

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

×

1205

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

×

1206

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

×

1207

      SVgroupInfo* pVgInfo = taosArrayGet(pList, i);

×

1208

      if (pVgInfo == NULL) {

×

1209

        continue;

×

1212

      if (pVgInfo->taskId == taskId) {

×

1213

        return &pVgInfo->epSet;

×

1216

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

×

1217

    SArray* pTaskInfos = pTask->outputInfo.vtableMapDispatcher.taskInfos;

×

1218

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

×

1219

      STaskDispatcherFixed* pAddr = taosArrayGet(pTaskInfos, i);

×

1220

      if (pAddr == NULL) {

×

1221

        continue;

×

1224

      if (pAddr->taskId == taskId) {

×

1225

        return &pAddr->epSet;

×

1230

  return NULL;

×

1233

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

×

1234

  char    buf[128] = {0};

×

1235

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

×

1236

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

×

1237

    return TSDB_CODE_OUT_OF_BUFFER;

×

1240

  *pId = taosStrdup(buf);

×

1242

  if (*pId == NULL) {

×

1243

    return terrno;

×

1245

    return TSDB_CODE_SUCCESS;

×

1249

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

×

1253

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

×

1254

  if (code) {

×

1255

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

×

1256

    return terrno = code;

×

1259

  pData->type = STREAM_INPUT__DATA_RETRIEVE;

×

1260

  pData->srcVgId = 0;

×

1262

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

×

1263

  if (code != TSDB_CODE_SUCCESS) {

×

1264

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

×

1265

    taosFreeQitem(pData);

×

1266

    return code;

×

1269

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

×

1270

  if (code != TSDB_CODE_SUCCESS) {

×

1271

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

×

1275

  return code;

×

1278

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

×

1279

  int32_t code = streamTaskEnqueueRetrieve(pTask, pReq);

×

1280

  if (code != 0) {

×

1281

    return code;

×

1283

  return streamTrySchedExec(pTask, false);

×

1286

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

×

1288

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

×

1289

  if (pTransId != NULL) {

×

1290

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

×

1293

  if (pCheckpointId != NULL) {

×

1294

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

×

1296

×

1298

int32_t streamTaskSetActiveCheckpointInfo(SStreamTask* pTask, int64_t activeCheckpointId) {

×

1299

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

×

1300

  return TSDB_CODE_SUCCESS;

×

1303

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

×

1304

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

×

1305

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

×

1306

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

×

1307

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

×

1308

×

1310

int32_t streamTaskCreateActiveChkptInfo(SActiveCheckpointInfo** pRes) {

×

1311

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

×

1312

  if (pInfo == NULL) {

×

1313

    return terrno;

×

1316

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

×

1317

  if (code != TSDB_CODE_SUCCESS) {

×

1318

    return code;

×

1321

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

×

1322

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

×

1323

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

×

1325

  *pRes = pInfo;

×

1326

  return code;

×

1329

void streamTaskDestroyActiveChkptInfo(SActiveCheckpointInfo* pInfo) {

×

1330

  if (pInfo == NULL) {

×

1331

    return;

×

1334

  streamMutexDestroy(&pInfo->lock);

×

1335

  taosArrayDestroy(pInfo->pDispatchTriggerList);

×

1336

  pInfo->pDispatchTriggerList = NULL;

×

1337

  taosArrayDestroy(pInfo->pReadyMsgList);

×

1338

  pInfo->pReadyMsgList = NULL;

×

1339

  taosArrayDestroy(pInfo->pCheckpointReadyRecvList);

×

1340

  pInfo->pCheckpointReadyRecvList = NULL;

×

1342

  SStreamTmrInfo* pTriggerTmr = &pInfo->chkptTriggerMsgTmr;

×

1343

  if (pTriggerTmr->tmrHandle != NULL) {

×

1344

    streamTmrStop(pTriggerTmr->tmrHandle);

×

1345

    pTriggerTmr->tmrHandle = NULL;

×

1348

  SStreamTmrInfo* pReadyTmr = &pInfo->chkptReadyMsgTmr;

×

1349

  if (pReadyTmr->tmrHandle != NULL) {

×

1350

    streamTmrStop(pReadyTmr->tmrHandle);

×

1351

    pReadyTmr->tmrHandle = NULL;

×

1354

  taosMemoryFree(pInfo);

×

1359

void streamTaskClearActiveInfo(SActiveCheckpointInfo* pInfo) {

×

1360

  pInfo->activeId = 0;

×

1361

  pInfo->transId = 0;

×

1362

  pInfo->allUpstreamTriggerRecv = 0;

×

1363

  pInfo->dispatchTrigger = false;

×

1365

  taosArrayClear(pInfo->pDispatchTriggerList);

×

1366

  taosArrayClear(pInfo->pCheckpointReadyRecvList);

×

1367

×

1369

const char* streamTaskGetExecType(int32_t type) {

×

1370

  switch (type) {

×

1371

    case STREAM_EXEC_T_EXTRACT_WAL_DATA:

×

1372

      return "scan-wal-file";

×

1373

    case STREAM_EXEC_T_START_ALL_TASKS:

×

1374

      return "start-all-tasks";

×

1375

    case STREAM_EXEC_T_START_ONE_TASK:

×

1376

      return "start-one-task";

×

1377

    case STREAM_EXEC_T_RESTART_ALL_TASKS:

×

1378

      return "restart-all-tasks";

×

1379

    case STREAM_EXEC_T_STOP_ALL_TASKS:

×

1380

      return "stop-all-tasks";

×

1381

    case STREAM_EXEC_T_RESUME_TASK:

×

1382

      return "resume-task-from-idle";

×

1383

    case STREAM_EXEC_T_ADD_FAILED_TASK:

×

1384

      return "record-start-failed-task";

×

1385

    case STREAM_EXEC_T_STOP_ONE_TASK:

×

1386

      return "stop-one-task";

×

1387

    case 0:

×

1388

      return "exec-all-tasks";

×

1389

    default:

×

1390

      return "invalid-exec-type";

×

1394

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

×

1395

  *pRefId = taosMemoryMalloc(sizeof(int64_t));

×

1396

  if (*pRefId != NULL) {

×

1397

    **pRefId = pTask->id.refId;

×

1398

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

×

1399

    if (code != 0) {

×

1400

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

×

1403

    return code;

×

1405

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

×

1406

    return terrno;

×

1410

void streamTaskFreeRefId(int64_t* pRefId) {

×

1411

  if (pRefId == NULL) {

×

1412

    return;

×

1415

  metaRefMgtRemove(pRefId);

×

1418

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

×

1419

  int32_t code = TSDB_CODE_SUCCESS;

×

1420

  int32_t lino = 0;

×

1422

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

×

1423

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

×

1425

  int32_t addrSize = taosArrayGetSize(info->pNotifyAddrUrls);

×

1426

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, addrSize));

×

1427

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

×

1428

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

×

1429

    TAOS_CHECK_EXIT(tEncodeCStr(pEncoder, url));

×

1431

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

×

1432

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

×

1433

  if (addrSize > 0) {

×

1434

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

×

1435

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

×

1436

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

×

1439

_exit:

×

1440

  if (code != TSDB_CODE_SUCCESS) {

×

1441

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

×

1443

  return code;

×

1446

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

×

1447

  int32_t code = TSDB_CODE_SUCCESS;

×

1448

  int32_t lino = 0;

×

1450

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

×

1451

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

×

1453

  int32_t addrSize = 0;

×

1454

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addrSize));

×

1455

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

×

1456

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

×

1457

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

×

1458

    char* url = NULL;

×

1459

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &url));

×

1460

    url = taosStrndup(url, TSDB_STREAM_NOTIFY_URL_LEN);

×

1461

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

×

1462

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

×

1463

      taosMemoryFree(url);

×

1464

      TAOS_CHECK_EXIT(terrno);

×

1467

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

×

1468

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

×

1469

  if (addrSize > 0) {

×

1470

    char* name = NULL;

×

1471

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &name));

×

1472

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

×

1473

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

×

1474

    TAOS_CHECK_EXIT(tDecodeCStr(pDecoder, &name));

×

1475

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

×

1476

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

×

1477

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

×

1478

    if (info->pSchemaWrapper == NULL) {

×

1479

      TAOS_CHECK_EXIT(terrno);

×

1481

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

×

1484

_exit:

×

1485

  if (code != TSDB_CODE_SUCCESS) {

×

1486

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

×

1488

  return code;

×

1491

int32_t tEncodeVTablesInfo(SEncoder* pEncoder, SArray* pVTables) {

×

1492

  int32_t code = TSDB_CODE_SUCCESS;

×

1493

  int32_t lino = 0;

×

1494

  SVCTableMergeInfo* pMergeInfo = NULL;

×

1495

  int32_t mergeNum = taosArrayGetSize(pVTables);

×

1496

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, mergeNum));

×

1497

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

×

1498

    pMergeInfo = (SVCTableMergeInfo*)taosArrayGet(pVTables, i);

×

1499

    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, pMergeInfo->uid));

×

1500

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pMergeInfo->numOfSrcTbls));

×

1503

_exit:

×

1504

  if (code != TSDB_CODE_SUCCESS) {

×

1505

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

×

1507

  return code;

×

1510

static int32_t tDecodeVTablesInfo(SDecoder* pDecoder, SArray** pTables) {

×

1511

  int32_t code = TSDB_CODE_SUCCESS;

×

1512

  int32_t lino = 0;

×

1514

  int32_t mergeNum = 0;

×

1515

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &mergeNum));

×

1516

  if (mergeNum > 0) {

×

1517

    *pTables = taosArrayInit(mergeNum, sizeof(SVCTableMergeInfo));

×

1518

    QUERY_CHECK_NULL(*pTables, code, lino, _exit, terrno);

×

1522

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

×

1523

    TAOS_CHECK_EXIT(tDecodeI64(pDecoder, &mergeInfo.uid));

×

1524

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &mergeInfo.numOfSrcTbls));

×

1525

    if (taosArrayPush(*pTables, &mergeInfo) == NULL) {

×

1526

      TAOS_CHECK_EXIT(terrno);

×

1530

_exit:

×

1531

  if (code != TSDB_CODE_SUCCESS) {

×

1532

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

×

1534

  return code;

×

1537

int32_t tSerializeDispatcherTaskInfo(SEncoder* pEncoder, const SArray* pTaskInfos) {

×

1538

  int32_t code = TSDB_CODE_SUCCESS;

×

1539

  int32_t lino = 0;

×

1540

  int32_t nTasks = taosArrayGetSize(pTaskInfos);

×

1541

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, nTasks));

×

1542

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

×

1543

    STaskDispatcherFixed* pAddr = taosArrayGet(pTaskInfos, i);

×

1544

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pAddr->taskId));

×

1545

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, pAddr->nodeId));

×

1546

    TAOS_CHECK_EXIT(tEncodeSEpSet(pEncoder, &pAddr->epSet));

×

1549

_exit:

×

1550

  if (code != TSDB_CODE_SUCCESS) {

×

1551

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

×

1553

  return code;

×

1556

int32_t tSerializeDispatcherVtableMap(SEncoder* pEncoder, const SSHashObj* pVtables) {

×

1557

  int32_t code = TSDB_CODE_SUCCESS;

×

1558

  int32_t lino = 0;

×

1559

  int32_t tbNum = tSimpleHashGetSize(pVtables);

×

1560

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, tbNum));

×

1561

  int32_t iter = 0;

×

1562

  void*   p = tSimpleHashIterate(pVtables, NULL, &iter);

×

1563

  while (p != NULL) {

×

1564

    int64_t* pUid = tSimpleHashGetKey(p, NULL);

×

1565

    int32_t* pIdx = p;

×

1566

    TAOS_CHECK_EXIT(tEncodeI64(pEncoder, *pUid));

×

1567

    TAOS_CHECK_EXIT(tEncodeI32(pEncoder, *pIdx));

×

1568

    p = tSimpleHashIterate(pVtables, p, &iter);

×

1571

_exit:

×

1572

  if (code != TSDB_CODE_SUCCESS) {

×

1573

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

×

1575

  return code;

×

1578

int32_t tDeserializeDispatcherTaskInfo(SDecoder* pDecoder, SArray** ppTaskInfos) {

×

1579

  int32_t code = TSDB_CODE_SUCCESS;

×

1580

  int32_t lino = 0;

×

1581

  int32_t nTasks = 0;

×

1582

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &nTasks));

×

1583

  if (nTasks <= 0) {

×

1584

    return code;

×

1587

  *ppTaskInfos = taosArrayInit(nTasks, sizeof(STaskDispatcherFixed));

×

1588

  QUERY_CHECK_NULL(*ppTaskInfos, code, lino, _exit, terrno);

×

1591

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

×

1592

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addr.taskId));

×

1593

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &addr.nodeId));

×

1594

    TAOS_CHECK_EXIT(tDecodeSEpSet(pDecoder, &addr.epSet));

×

1595

    void* px = taosArrayPush(*ppTaskInfos, &addr);

×

1596

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

×

1599

_exit:

×

1600

  if (code != TSDB_CODE_SUCCESS) {

×

1601

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

×

1603

  return code;

×

1606

int32_t tDeserializeDispatcherVtableMap(SDecoder* pDecoder, SSHashObj** ppVtables) {

×

1607

  int32_t code = TSDB_CODE_SUCCESS;

×

1608

  int32_t lino = 0;

×

1609

  int32_t tbNum = 0;

×

1610

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &tbNum));

×

1611

  if (tbNum <= 0) {

×

1612

    return code;

×

1615

  *ppVtables = tSimpleHashInit(tbNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

×

1616

  QUERY_CHECK_NULL(*ppVtables, code, lino, _exit, terrno);

×

1618

  uint64_t uid = 0;

×

1619

  int32_t  idx = 0;

×

1620

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

×

1621

    TAOS_CHECK_EXIT(tDecodeU64(pDecoder, &uid));

×

1622

    TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &idx));

×

1623

    TAOS_CHECK_EXIT(tSimpleHashPut(*ppVtables, &uid, sizeof(uid), &idx, sizeof(idx)));

×

1626

_exit:

×

1628

  if (code != TSDB_CODE_SUCCESS) {

×

1629

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

×

1631

  return code;

×

1634

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

×

1635

  int32_t code = 0;

×

1638

  TAOS_CHECK_EXIT(tStartEncode(pEncoder));

×

1639

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

×

1640

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

×

1641

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

×

1642

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

×

1643

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

×

1644

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

×

1645

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

×

1647

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

×

1648

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

×

1650

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

×

1651

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

×

1652

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

×

1653

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

×

1655

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

×

1656

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

×

1657

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

×

1659

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

×

1660

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

×

1661

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, taskId));

×

1663

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

×

1664

  taskId = pTask->streamTaskId.taskId;

×

1665

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, taskId));

×

1667

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

×

1668

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

×

1669

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

×

1670

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

×

1672

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

×

1673

  TAOS_CHECK_EXIT(tEncodeI32(pEncoder, epSz));

×

1674

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

×

1675

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

×

1676

    TAOS_CHECK_EXIT(tEncodeStreamEpInfo(pEncoder, pInfo));

×

1679

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

×

1680

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

×

1683

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

×

1684

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

×

1685

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

×

1686

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

×

1687

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

×

1688

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

×

1689

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

×

1690

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

×

1691

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

×

1692

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

×

1693

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

×

1694

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

×

1695

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

×

1696

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

×

1697

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

×

1698

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

×

1699

    TAOS_CHECK_EXIT(tSerializeDispatcherTaskInfo(pEncoder, pTask->outputInfo.vtableMapDispatcher.taskInfos));

×

1700

    TAOS_CHECK_EXIT(tSerializeDispatcherVtableMap(pEncoder, pTask->outputInfo.vtableMapDispatcher.vtableMap));

×

1702

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

×

1703

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

×

1704

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

×

1706

  if (pTask->ver >= SSTREAM_TASK_ADD_NOTIFY_VER) {

×

1707

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

×

1708

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

×

1709

    TAOS_CHECK_EXIT(tEncodeVTablesInfo(pEncoder, pTask->pVTables));

×

1712

  tEndEncode(pEncoder);

×

1713

_exit:

×

1714

  return code;

×

1717

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

×

1718

  int32_t taskId = 0;

×

1719

  int32_t code = 0;

×

1722

  TAOS_CHECK_EXIT(tStartDecode(pDecoder));

×

1723

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

×

1724

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

×

1725

    TAOS_CHECK_EXIT(TSDB_CODE_INVALID_MSG);

×

1728

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

×

1729

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

×

1730

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

×

1731

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

×

1732

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

×

1733

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

×

1735

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

×

1736

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

×

1738

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

×

1739

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

×

1740

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

×

1741

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

×

1743

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

×

1744

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

×

1745

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

×

1747

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

×

1748

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));

×

1749

  pTask->hTaskInfo.id.taskId = taskId;

×

1751

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

×

1752

  TAOS_CHECK_EXIT(tDecodeI32(pDecoder, &taskId));

×

1753

  pTask->streamTaskId.taskId = taskId;

×

1755

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

×

1756

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

×

1757

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

×

1758

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

×

1760

  int32_t epSz = -1;

×

1761

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

×

1763

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

×

1764

    TAOS_CHECK_EXIT(terrno);

×

1766

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

×

1767

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

×

1768

    if (pInfo == NULL) {

×

1769

      TAOS_CHECK_EXIT(terrno);

×

1771

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

×

1772

      taosMemoryFreeClear(pInfo);

×

1773

      goto _exit;

×

1775

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

×

1776

      TAOS_CHECK_EXIT(terrno);

×

1780

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

×

1781

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

×

1784

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

×

1785

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

×

1786

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

×

1787

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

×

1788

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

×

1789

      TAOS_CHECK_EXIT(terrno);

×

1791

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

×

1792

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

×

1793

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

×

1794

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

×

1795

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

×

1796

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

×

1797

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

×

1798

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

×

1799

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

×

1800

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

×

1801

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

×

1802

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

×

1803

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

×

1804

    TAOS_CHECK_EXIT(tDeserializeDispatcherTaskInfo(pDecoder, &pTask->outputInfo.vtableMapDispatcher.taskInfos));

×

1805

    TAOS_CHECK_EXIT(tDeserializeDispatcherVtableMap(pDecoder, &pTask->outputInfo.vtableMapDispatcher.vtableMap));

×

1807

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

×

1808

  if (pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER) {

×

1809

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

×

1811

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

×

1813

  if (pTask->ver >= SSTREAM_TASK_ADD_NOTIFY_VER) {

×

1814

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

×

1815

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

×

1816

    TAOS_CHECK_EXIT(tDecodeVTablesInfo(pDecoder, &pTask->pVTables));

×

1819

  tEndDecode(pDecoder);

×

1821

_exit:

×

1822

  return code;

×

taosdata / TDengine / #3798

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

taosdata / TDengine / #3798

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