tstream.h 18.3 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

16
#include "os.h"
17
#include "streamState.h"
L
Liu Jicong 已提交
18
#include "tdatablock.h"
L
Liu Jicong 已提交
19
#include "tdbInt.h"
L
Liu Jicong 已提交
20 21
#include "tmsg.h"
#include "tmsgcb.h"
L
Liu Jicong 已提交
22
#include "tqueue.h"
L
Liu Jicong 已提交
23 24 25 26 27

#ifdef __cplusplus
extern "C" {
#endif

L
Liu Jicong 已提交
28 29
#ifndef _STREAM_H_
#define _STREAM_H_
L
Liu Jicong 已提交
30

L
Liu Jicong 已提交
31 32
typedef struct SStreamTask SStreamTask;

33 34
enum {
  STREAM_STATUS__NORMAL = 0,
L
Liu Jicong 已提交
35
  STREAM_STATUS__STOP,
L
Liu Jicong 已提交
36
  STREAM_STATUS__INIT,
L
Liu Jicong 已提交
37
  STREAM_STATUS__FAILED,
38
  STREAM_STATUS__RECOVER,
5
54liuyao 已提交
39
  STREAM_STATUS__PAUSE,
40 41
};

L
Liu Jicong 已提交
42
enum {
L
Liu Jicong 已提交
43 44
  TASK_STATUS__NORMAL = 0,
  TASK_STATUS__DROPPING,
L
Liu Jicong 已提交
45 46
  TASK_STATUS__FAIL,
  TASK_STATUS__STOP,
Y
yihaoDeng 已提交
47 48
  TASK_STATUS__SCAN_HISTORY,  // stream task scan history data by using tsdbread in the stream scanner
  TASK_STATUS__HALT,          // stream task will handle all data in the input queue, and then paused
L
liuyao 已提交
49
  TASK_STATUS__PAUSE,
L
Liu Jicong 已提交
50 51 52
};

enum {
L
Liu Jicong 已提交
53 54 55 56
  TASK_SCHED_STATUS__INACTIVE = 1,
  TASK_SCHED_STATUS__WAITING,
  TASK_SCHED_STATUS__ACTIVE,
  TASK_SCHED_STATUS__FAILED,
L
Liu Jicong 已提交
57
  TASK_SCHED_STATUS__DROPPING,
L
Liu Jicong 已提交
58 59 60 61 62 63 64
};

enum {
  TASK_INPUT_STATUS__NORMAL = 1,
  TASK_INPUT_STATUS__BLOCKED,
  TASK_INPUT_STATUS__RECOVER,
  TASK_INPUT_STATUS__STOP,
L
Liu Jicong 已提交
65
  TASK_INPUT_STATUS__FAILED,
L
Liu Jicong 已提交
66 67 68 69 70 71
};

enum {
  TASK_OUTPUT_STATUS__NORMAL = 1,
  TASK_OUTPUT_STATUS__WAIT,
  TASK_OUTPUT_STATUS__BLOCKED,
L
Liu Jicong 已提交
72 73
};

74 75 76 77 78
enum {
  TASK_TRIGGER_STATUS__INACTIVE = 1,
  TASK_TRIGGER_STATUS__ACTIVE,
};

79
typedef enum {
80 81 82
  TASK_LEVEL__SOURCE = 1,
  TASK_LEVEL__AGG,
  TASK_LEVEL__SINK,
83
} ETASK_LEVEL;
84 85 86 87 88 89 90 91 92

enum {
  TASK_OUTPUT__FIXED_DISPATCH = 1,
  TASK_OUTPUT__SHUFFLE_DISPATCH,
  TASK_OUTPUT__TABLE,
  TASK_OUTPUT__SMA,
  TASK_OUTPUT__FETCH,
};

L
Liu Jicong 已提交
93 94 95 96 97 98
enum {
  STREAM_QUEUE__SUCESS = 1,
  STREAM_QUEUE__FAILED,
  STREAM_QUEUE__PROCESSING,
};

L
Liu Jicong 已提交
99 100 101 102
typedef struct {
  int8_t type;
} SStreamQueueItem;

103 104
typedef void    FTbSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
typedef int32_t FTaskExpand(void* ahandle, SStreamTask* pTask, int64_t ver);
L
Liu Jicong 已提交
105 106

typedef struct {
L
Liu Jicong 已提交
107 108 109 110
  int8_t      type;
  int64_t     ver;
  int32_t*    dataRef;
  SPackedData submit;
111
} SStreamDataSubmit;
L
Liu Jicong 已提交
112 113 114 115 116 117

typedef struct {
  int8_t  type;
  int64_t ver;
  SArray* dataRefs;  // SArray<int32_t*>
  SArray* submits;   // SArray<SPackedSubmit>
118
} SStreamMergedSubmit;
119

L
Liu Jicong 已提交
120 121 122
typedef struct {
  int8_t type;

L
Liu Jicong 已提交
123
  int32_t srcVgId;
L
Liu Jicong 已提交
124
  int32_t childId;
L
Liu Jicong 已提交
125
  int64_t sourceVer;
L
Liu Jicong 已提交
126
  int64_t reqId;
L
Liu Jicong 已提交
127

L
Liu Jicong 已提交
128
  SArray* blocks;  // SArray<SSDataBlock>
L
Liu Jicong 已提交
129 130
} SStreamDataBlock;

L
Liu Jicong 已提交
131 132 133 134 135 136
// ref data block, for delete
typedef struct {
  int8_t       type;
  SSDataBlock* pBlock;
} SStreamRefDataBlock;

L
Liu Jicong 已提交
137 138 139 140
typedef struct {
  int8_t type;
} SStreamCheckpoint;

141 142 143 144 145
typedef struct {
  int8_t       type;
  SSDataBlock* pBlock;
} SStreamTrigger;

L
Liu Jicong 已提交
146 147 148 149 150 151 152 153 154 155 156 157 158 159
typedef struct SStreamQueueNode SStreamQueueNode;

struct SStreamQueueNode {
  SStreamQueueItem* item;
  SStreamQueueNode* next;
};

typedef struct {
  SStreamQueueNode* head;
  int64_t           size;
} SStreamQueueRes;

void streamFreeQitem(SStreamQueueItem* data);

5
54liuyao 已提交
160
#if 0
L
Liu Jicong 已提交
161 162 163 164 165 166
bool              streamQueueResEmpty(const SStreamQueueRes* pRes);
int64_t           streamQueueResSize(const SStreamQueueRes* pRes);
SStreamQueueNode* streamQueueResFront(SStreamQueueRes* pRes);
SStreamQueueNode* streamQueueResPop(SStreamQueueRes* pRes);
void              streamQueueResClear(SStreamQueueRes* pRes);
SStreamQueueRes   streamQueueBuildRes(SStreamQueueNode* pNode);
5
54liuyao 已提交
167
#endif
L
Liu Jicong 已提交
168 169 170 171 172

typedef struct {
  SStreamQueueNode* pHead;
} SStreamQueue1;

5
54liuyao 已提交
173
#if 0
L
Liu Jicong 已提交
174 175 176
bool            streamQueueHasTask(const SStreamQueue1* pQueue);
int32_t         streamQueuePush(SStreamQueue1* pQueue, SStreamQueueItem* pItem);
SStreamQueueRes streamQueueGetRes(SStreamQueue1* pQueue);
5
54liuyao 已提交
177
#endif
L
Liu Jicong 已提交
178

L
Liu Jicong 已提交
179 180 181 182
typedef struct {
  STaosQueue* queue;
  STaosQall*  qall;
  void*       qItem;
L
Liu Jicong 已提交
183 184
  int8_t      status;
} SStreamQueue;
L
Liu Jicong 已提交
185

186 187 188
int32_t streamInit();
void    streamCleanUp();

dengyihao's avatar
dengyihao 已提交
189
SStreamQueue* streamQueueOpen(int64_t cap);
L
Liu Jicong 已提交
190 191 192
void          streamQueueClose(SStreamQueue* queue);

static FORCE_INLINE void streamQueueProcessSuccess(SStreamQueue* queue) {
193
  ASSERT(atomic_load_8(&queue->status) == STREAM_QUEUE__PROCESSING);
L
Liu Jicong 已提交
194 195
  queue->qItem = NULL;
  atomic_store_8(&queue->status, STREAM_QUEUE__SUCESS);
L
Liu Jicong 已提交
196 197
}

L
Liu Jicong 已提交
198
static FORCE_INLINE void streamQueueProcessFail(SStreamQueue* queue) {
199
  ASSERT(atomic_load_8(&queue->status) == STREAM_QUEUE__PROCESSING);
L
Liu Jicong 已提交
200 201 202
  atomic_store_8(&queue->status, STREAM_QUEUE__FAILED);
}

203
void* streamQueueNextItem(SStreamQueue* pQueue);
L
Liu Jicong 已提交
204

205
SStreamDataSubmit* streamDataSubmitNew(SPackedData* pData, int32_t type);
206
void               streamDataSubmitDestroy(SStreamDataSubmit* pDataSubmit);
L
Liu Jicong 已提交
207

L
Liu Jicong 已提交
208
typedef struct {
209 210 211
  char*              qmsg;
  void*              pExecutor;   // not applicable to encoder and decoder
  struct SWalReader* pWalReader;  // not applicable to encoder and decoder
L
Liu Jicong 已提交
212 213 214
} STaskExec;

typedef struct {
L
Liu Jicong 已提交
215
  int32_t taskId;
L
Liu Jicong 已提交
216 217 218 219 220
  int32_t nodeId;
  SEpSet  epSet;
} STaskDispatcherFixedEp;

typedef struct {
L
Liu Jicong 已提交
221
  char      stbFullName[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
222
  int32_t   waitingRspCnt;
L
Liu Jicong 已提交
223
  SUseDbRsp dbInfo;
L
Liu Jicong 已提交
224 225 226
} STaskDispatcherShuffle;

typedef struct {
L
Liu Jicong 已提交
227
  int64_t         stbUid;
L
Liu Jicong 已提交
228
  char            stbFullName[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
229
  SSchemaWrapper* pSchemaWrapper;
230 231 232
  void*           vnode;  // not available to encoder and decoder
  FTbSink*        tbSinkFunc;
  STSchema*       pTSchema;
L
liuyao 已提交
233
  SSHashObj*      pTblInfo;
L
Liu Jicong 已提交
234 235
} STaskSinkTb;

L
Liu Jicong 已提交
236
typedef void FSmaSink(void* vnode, int64_t smaId, const SArray* data);
L
Liu Jicong 已提交
237

L
Liu Jicong 已提交
238
typedef struct {
L
Liu Jicong 已提交
239 240
  int64_t smaId;
  // following are not applicable to encoder and decoder
L
Liu Jicong 已提交
241
  void*     vnode;
L
Liu Jicong 已提交
242
  FSmaSink* smaSink;
L
Liu Jicong 已提交
243 244 245 246 247 248
} STaskSinkSma;

typedef struct {
  int8_t reserved;
} STaskSinkFetch;

249
typedef struct SStreamChildEpInfo {
L
Liu Jicong 已提交
250 251 252
  int32_t nodeId;
  int32_t childId;
  int32_t taskId;
L
Liu Jicong 已提交
253
  SEpSet  epSet;
L
Liu Jicong 已提交
254 255
} SStreamChildEpInfo;

256 257 258 259 260
typedef struct SStreamId {
  int64_t     streamId;
  int32_t     taskId;
  const char* idStr;
} SStreamId;
L
Liu Jicong 已提交
261

262 263
typedef struct SCheckpointInfo {
  int64_t id;
264 265
  int64_t version;     // offset in WAL
  int64_t currentVer;  // current offset in WAL, not serialize it
266 267
} SCheckpointInfo;

268
typedef struct SStreamStatus {
Y
yihaoDeng 已提交
269 270 271 272 273
  int8_t        taskStatus;
  int8_t        checkDownstream;
  int8_t        schedStatus;
  int8_t        keepTaskStatus;
  bool          transferState;
274
  TdThreadMutex lock;
275
} SStreamStatus;
L
Liu Jicong 已提交
276

277
typedef struct SHistDataRange {
278 279
  SVersionRange range;
  STimeWindow   window;
280
} SHistDataRange;
281

282
typedef struct SSTaskBasicInfo {
Y
yihaoDeng 已提交
283 284 285 286 287 288
  int32_t nodeId;  // vgroup id or snode id
  SEpSet  epSet;
  int32_t selfChildId;
  int32_t totalLevel;
  int8_t  taskLevel;
  int8_t  fillHistory;  // is fill history task or not
289
} SSTaskBasicInfo;
290

291
typedef struct SDispatchMsgInfo {
Y
yihaoDeng 已提交
292 293 294 295
  void*   pData;       // current dispatch data
  int16_t msgType;     // dispatch msg type
  int32_t retryCount;  // retry send data count
  int64_t blockingTs;  // output blocking timestamp
296
} SDispatchMsgInfo;
297

298 299 300 301 302 303
typedef struct {
  int8_t        outputType;
  int8_t        outputStatus;
  SStreamQueue* outputQueue;
} SSTaskOutputInfo;

304 305 306 307 308 309 310 311 312 313
struct SStreamTask {
  SStreamId        id;
  SSTaskBasicInfo  info;
  int8_t           outputType;
  SDispatchMsgInfo msgInfo;
  SStreamStatus    status;
  SCheckpointInfo  chkInfo;
  STaskExec        exec;
  SHistDataRange   dataRange;
  SStreamId        historyTaskId;
H
Haojun Liao 已提交
314
  SStreamId        streamTaskId;
315 316 317
  SArray*          pUpstreamEpInfoList;  // SArray<SStreamChildEpInfo*>, // children info
  int32_t          nextCheckId;
  SArray*          checkpointInfo;  // SArray<SStreamCheckpointInfo>
L
Liu Jicong 已提交
318

319
  // output
L
Liu Jicong 已提交
320 321 322
  union {
    STaskDispatcherFixedEp fixedEpDispatcher;
    STaskDispatcherShuffle shuffleDispatcher;
323 324 325
    STaskSinkTb            tbSink;
    STaskSinkSma           smaSink;
    STaskSinkFetch         fetchSink;
L
Liu Jicong 已提交
326 327
  };

328 329
  int8_t        inputStatus;
  int8_t        outputStatus;
L
Liu Jicong 已提交
330 331
  SStreamQueue* inputQueue;
  SStreamQueue* outputQueue;
L
Liu Jicong 已提交
332

333
  // trigger
334 335 336 337 338
  int8_t        triggerStatus;
  int64_t       triggerParam;
  void*         timer;
  SMsgCb*       pMsgCb;  // msg handle
  SStreamState* pState;  // state backend
339

340
  // the followings attributes don't be serialized
341
  int32_t             notReadyTasks;
342
  int32_t             numOfWaitingUpstream;
343 344 345 346 347 348
  int64_t             checkReqId;
  SArray*             checkReqIds;  // shuffle
  int32_t             refCnt;
  int64_t             checkpointingId;
  int32_t             checkpointAlignCnt;
  struct SStreamMeta* pMeta;
L
liuyao 已提交
349
  SSHashObj*          pNameMap;
350
};
L
Liu Jicong 已提交
351

352 353
// meta
typedef struct SStreamMeta {
Y
yihaoDeng 已提交
354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369
  char*         path;
  TDB*          db;
  TTB*          pTaskDb;
  TTB*          pCheckpointDb;
  SHashObj*     pTasks;
  SArray*       pTaskList;  // SArray<task_id*>
  void*         ahandle;
  TXN*          txn;
  FTaskExpand*  expandFunc;
  int32_t       vgId;
  SRWLatch      lock;
  int32_t       walScanCounter;
  void*         streamBackend;
  int64_t       streamBackendRid;
  SHashObj*     pTaskBackendUnique;
  TdThreadMutex backendMutex;
370 371
} SStreamMeta;

L
Liu Jicong 已提交
372 373 374
int32_t tEncodeStreamEpInfo(SEncoder* pEncoder, const SStreamChildEpInfo* pInfo);
int32_t tDecodeStreamEpInfo(SDecoder* pDecoder, SStreamChildEpInfo* pInfo);

375 376
SStreamTask* tNewStreamTask(int64_t streamId, int8_t taskLevel, int8_t fillHistory, int64_t triggerParam,
                            SArray* pTaskList);
377 378
int32_t      tEncodeStreamTask(SEncoder* pEncoder, const SStreamTask* pTask);
int32_t      tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask);
379
void         tFreeStreamTask(SStreamTask* pTask);
380
int32_t      tAppendDataToInputQueue(SStreamTask* pTask, SStreamQueueItem* pItem);
381
bool         tInputQueueIsFull(const SStreamTask* pTask);
L
Liu Jicong 已提交
382 383 384 385 386 387 388

typedef struct {
  SMsgHead head;
  int64_t  streamId;
  int32_t  taskId;
} SStreamTaskRunReq;

L
Liu Jicong 已提交
389 390 391
typedef struct {
  int64_t streamId;
  int32_t taskId;
L
Liu Jicong 已提交
392 393 394
  int32_t dataSrcVgId;
  int32_t upstreamTaskId;
  int32_t upstreamChildId;
L
Liu Jicong 已提交
395
  int32_t upstreamNodeId;
L
Liu Jicong 已提交
396
  int32_t blockNum;
397
  int64_t totalLen;
L
Liu Jicong 已提交
398 399
  SArray* dataLen;  // SArray<int32_t>
  SArray* data;     // SArray<SRetrieveTableRsp*>
L
Liu Jicong 已提交
400 401 402 403
} SStreamDispatchReq;

typedef struct {
  int64_t streamId;
404 405 406 407
  int32_t upstreamNodeId;
  int32_t upstreamTaskId;
  int32_t downstreamNodeId;
  int32_t downstreamTaskId;
L
Liu Jicong 已提交
408 409 410
  int8_t  inputStatus;
} SStreamDispatchRsp;

L
Liu Jicong 已提交
411 412
typedef struct {
  int64_t            streamId;
L
Liu Jicong 已提交
413
  int64_t            reqId;
L
Liu Jicong 已提交
414 415 416 417 418 419 420 421 422 423 424 425 426 427 428
  int32_t            srcTaskId;
  int32_t            srcNodeId;
  int32_t            dstTaskId;
  int32_t            dstNodeId;
  int32_t            retrieveLen;
  SRetrieveTableRsp* pRetrieve;
} SStreamRetrieveReq;

typedef struct {
  int64_t streamId;
  int32_t childId;
  int32_t rspFromTaskId;
  int32_t rspToTaskId;
} SStreamRetrieveRsp;

429
typedef struct {
430
  int64_t reqId;
431
  int64_t streamId;
432 433 434 435
  int32_t upstreamNodeId;
  int32_t upstreamTaskId;
  int32_t downstreamNodeId;
  int32_t downstreamTaskId;
436
  int32_t childId;
437
} SStreamTaskCheckReq;
438

L
Liu Jicong 已提交
439
typedef struct {
440
  int64_t reqId;
L
Liu Jicong 已提交
441
  int64_t streamId;
L
Liu Jicong 已提交
442
  int32_t upstreamNodeId;
443 444 445 446 447 448
  int32_t upstreamTaskId;
  int32_t downstreamNodeId;
  int32_t downstreamTaskId;
  int32_t childId;
  int8_t  status;
} SStreamTaskCheckRsp;
L
Liu Jicong 已提交
449 450

typedef struct {
451 452 453
  SMsgHead msgHead;
  int64_t  streamId;
  int32_t  taskId;
L
liuyao 已提交
454 455
  int8_t   igUntreated;
} SStreamScanHistoryReq;
L
Liu Jicong 已提交
456 457 458 459

typedef struct {
  int64_t streamId;
  int32_t taskId;
460
  int32_t childId;
H
Haojun Liao 已提交
461
} SStreamRecoverFinishReq, SStreamTransferReq;
L
Liu Jicong 已提交
462

H
Haojun Liao 已提交
463 464
int32_t tEncodeStreamRecoverFinishReq(SEncoder* pEncoder, const SStreamRecoverFinishReq* pReq);
int32_t tDecodeStreamRecoverFinishReq(SDecoder* pDecoder, SStreamRecoverFinishReq* pReq);
L
Liu Jicong 已提交
465

466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519
typedef struct {
  int64_t streamId;
  int64_t checkpointId;
  int32_t taskId;
  int32_t nodeId;
  int64_t expireTime;
} SStreamCheckpointSourceReq;

typedef struct {
  int64_t streamId;
  int64_t checkpointId;
  int32_t taskId;
  int32_t nodeId;
  int64_t expireTime;
} SStreamCheckpointSourceRsp;

int32_t tEncodeSStreamCheckpointSourceReq(SEncoder* pEncoder, const SStreamCheckpointSourceReq* pReq);
int32_t tDecodeSStreamCheckpointSourceReq(SDecoder* pDecoder, SStreamCheckpointSourceReq* pReq);

int32_t tEncodeSStreamCheckpointSourceRsp(SEncoder* pEncoder, const SStreamCheckpointSourceRsp* pRsp);
int32_t tDecodeSStreamCheckpointSourceRsp(SDecoder* pDecoder, SStreamCheckpointSourceRsp* pRsp);

typedef struct {
  SMsgHead msgHead;
  int64_t  streamId;
  int64_t  checkpointId;
  int32_t  downstreamTaskId;
  int32_t  downstreamNodeId;
  int32_t  upstreamTaskId;
  int32_t  upstreamNodeId;
  int32_t  childId;
  int64_t  expireTime;
  int8_t   taskLevel;
} SStreamCheckpointReq;

typedef struct {
  SMsgHead msgHead;
  int64_t  streamId;
  int64_t  checkpointId;
  int32_t  downstreamTaskId;
  int32_t  downstreamNodeId;
  int32_t  upstreamTaskId;
  int32_t  upstreamNodeId;
  int32_t  childId;
  int64_t  expireTime;
  int8_t   taskLevel;
} SStreamCheckpointRsp;

int32_t tEncodeSStreamCheckpointReq(SEncoder* pEncoder, const SStreamCheckpointReq* pReq);
int32_t tDecodeSStreamCheckpointReq(SDecoder* pDecoder, SStreamCheckpointReq* pReq);

int32_t tEncodeSStreamCheckpointRsp(SEncoder* pEncoder, const SStreamCheckpointRsp* pRsp);
int32_t tDecodeSStreamCheckpointRsp(SDecoder* pDecoder, SStreamCheckpointRsp* pRsp);

L
Liu Jicong 已提交
520 521
typedef struct {
  int64_t streamId;
L
Liu Jicong 已提交
522 523 524
  int32_t downstreamTaskId;
  int32_t taskId;
} SStreamRecoverDownstreamReq;
L
Liu Jicong 已提交
525 526

typedef struct {
L
Liu Jicong 已提交
527 528 529 530 531 532
  int64_t streamId;
  int32_t downstreamTaskId;
  int32_t taskId;
  SArray* checkpointVer;  // SArray<SStreamCheckpointInfo>
} SStreamRecoverDownstreamRsp;

533 534
int32_t tEncodeStreamTaskCheckReq(SEncoder* pEncoder, const SStreamTaskCheckReq* pReq);
int32_t tDecodeStreamTaskCheckReq(SDecoder* pDecoder, SStreamTaskCheckReq* pReq);
535

536 537
int32_t tEncodeStreamTaskCheckRsp(SEncoder* pEncoder, const SStreamTaskCheckRsp* pRsp);
int32_t tDecodeStreamTaskCheckRsp(SDecoder* pDecoder, SStreamTaskCheckRsp* pRsp);
538

L
Liu Jicong 已提交
539
int32_t tEncodeSStreamTaskRecoverReq(SEncoder* pEncoder, const SStreamRecoverDownstreamReq* pReq);
L
Liu Jicong 已提交
540 541 542 543 544
int32_t tDecodeSStreamTaskRecoverReq(SDecoder* pDecoder, SStreamRecoverDownstreamReq* pReq);

int32_t tEncodeSStreamTaskRecoverRsp(SEncoder* pEncoder, const SStreamRecoverDownstreamRsp* pRsp);
int32_t tDecodeSStreamTaskRecoverRsp(SDecoder* pDecoder, SStreamRecoverDownstreamRsp* pRsp);

545
int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq);
L
Liu Jicong 已提交
546
int32_t tDecodeStreamRetrieveReq(SDecoder* pDecoder, SStreamRetrieveReq* pReq);
L
Liu Jicong 已提交
547 548
void    tDeleteStreamRetrieveReq(SStreamRetrieveReq* pReq);

549 550 551
int32_t tInitStreamDispatchReq(SStreamDispatchReq* pReq, const SStreamTask* pTask, int32_t vgId, int32_t numOfBlocks,
                               int64_t dstTaskId);
void    tDeleteStreamDispatchReq(SStreamDispatchReq* pReq);
552

553
int32_t streamSetupTrigger(SStreamTask* pTask);
L
Liu Jicong 已提交
554

L
Liu Jicong 已提交
555
int32_t streamProcessRunReq(SStreamTask* pTask);
556
int32_t streamProcessDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pReq, SRpcMsg* pMsg, bool exec);
557
int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, int32_t code);
L
Liu Jicong 已提交
558

L
Liu Jicong 已提交
559
int32_t streamProcessRetrieveReq(SStreamTask* pTask, SStreamRetrieveReq* pReq, SRpcMsg* pMsg);
5
54liuyao 已提交
560
// int32_t streamProcessRetrieveRsp(SStreamTask* pTask, SStreamRetrieveRsp* pRsp);
L
Liu Jicong 已提交
561

562
void    streamTaskInputFail(SStreamTask* pTask);
L
Liu Jicong 已提交
563 564
int32_t streamTryExec(SStreamTask* pTask);
int32_t streamSchedExec(SStreamTask* pTask);
565
int32_t streamTaskOutputResultBlock(SStreamTask* pTask, SStreamDataBlock* pBlock);
566
bool    streamTaskShouldStop(const SStreamStatus* pStatus);
L
liuyao 已提交
567
bool    streamTaskShouldPause(const SStreamStatus* pStatus);
568
bool    streamTaskIsIdle(const SStreamTask* pTask);
L
Liu Jicong 已提交
569

570 571
int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz);

572 573
char*   createStreamTaskIdStr(int64_t streamId, int32_t taskId);

574
// recover and fill history
575
void    streamPrepareNdoCheckDownstream(SStreamTask* pTask);
576
int32_t streamTaskCheckDownstreamTasks(SStreamTask* pTask);
577
int32_t streamTaskLaunchScanHistory(SStreamTask* pTask);
578
int32_t streamTaskCheckStatus(SStreamTask* pTask);
579
int32_t streamProcessCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp* pRsp);
580
int32_t streamCheckHistoryTaskDownstrem(SStreamTask* pTask);
581
int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask);
L
liuyao 已提交
582
int32_t streamStartRecoverTask(SStreamTask* pTask, int8_t igUntreated);
583

584
// common
Y
yihaoDeng 已提交
585 586 587
int32_t     streamSetParamForScanHistoryData(SStreamTask* pTask);
int32_t     streamRestoreParam(SStreamTask* pTask);
int32_t     streamSetStatusNormal(SStreamTask* pTask);
588 589
const char* streamGetTaskStatusStr(int32_t status);

590
// source level
Y
yihaoDeng 已提交
591
int32_t streamSetParamForStreamScanner(SStreamTask* pTask, SVersionRange* pVerRange, STimeWindow* pWindow);
L
liuyao 已提交
592
int32_t streamBuildSourceRecover1Req(SStreamTask* pTask, SStreamScanHistoryReq* pReq, int8_t igUntreated);
593
int32_t streamSourceScanHistoryData(SStreamTask* pTask);
Y
yihaoDeng 已提交
594
// int32_t streamSourceRecoverScanStep2(SStreamTask* pTask, int64_t ver);
595
int32_t streamDispatchScanHistoryFinishMsg(SStreamTask* pTask);
H
Haojun Liao 已提交
596 597 598

int32_t streamDispatchTransferStateMsg(SStreamTask* pTask);

599 600 601 602
// agg level
int32_t streamAggRecoverPrepare(SStreamTask* pTask);
int32_t streamProcessRecoverFinishReq(SStreamTask* pTask, int32_t childId);

dengyihao's avatar
dengyihao 已提交
603 604
void         streamMetaInit();
void         streamMetaCleanup();
605
SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandFunc, int32_t vgId);
L
Liu Jicong 已提交
606 607
void         streamMetaClose(SStreamMeta* streamMeta);

608 609 610 611
int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask);
int32_t streamMetaAddDeployedTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask);
int32_t streamMetaAddSerializedTask(SStreamMeta* pMeta, int64_t checkpointVer, char* msg, int32_t msgLen);
int32_t streamMetaGetNumOfTasks(const SStreamMeta* pMeta);
L
Liu Jicong 已提交
612

L
Liu Jicong 已提交
613 614
SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int32_t taskId);
void         streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask);
615
void         streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId);
L
Liu Jicong 已提交
616

617 618 619 620
int32_t streamMetaBegin(SStreamMeta* pMeta);
int32_t streamMetaCommit(SStreamMeta* pMeta);
int32_t streamMetaRollBack(SStreamMeta* pMeta);
int32_t streamLoadTasks(SStreamMeta* pMeta, int64_t ver);
L
Liu Jicong 已提交
621

622 623 624 625 626
// checkpoint
int32_t streamProcessCheckpointSourceReq(SStreamMeta* pMeta, SStreamTask* pTask, SStreamCheckpointSourceReq* pReq);
int32_t streamProcessCheckpointReq(SStreamMeta* pMeta, SStreamTask* pTask, SStreamCheckpointReq* pReq);
int32_t streamProcessCheckpointRsp(SStreamMeta* pMeta, SStreamTask* pTask, SStreamCheckpointRsp* pRsp);

L
liuyao 已提交
627 628 629
int32_t streamTaskReleaseState(SStreamTask* pTask);
int32_t streamTaskReloadState(SStreamTask* pTask);

L
Liu Jicong 已提交
630 631 632 633
#ifdef __cplusplus
}
#endif

L
Liu Jicong 已提交
634
#endif /* ifndef _STREAM_H_ */