tstream.h 7.5 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/>.
 */

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

#ifdef __cplusplus
extern "C" {
#endif

#ifndef _TSTREAM_H_
#define _TSTREAM_H_

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

L
Liu Jicong 已提交
32
enum {
L
Liu Jicong 已提交
33 34 35 36 37 38 39 40 41 42
  TASK_STATUS__IDLE = 1,
  TASK_STATUS__EXECUTING,
  TASK_STATUS__CLOSING,
};

enum {
  TASK_INPUT_STATUS__NORMAL = 1,
  TASK_INPUT_STATUS__BLOCKED,
  TASK_INPUT_STATUS__RECOVER,
  TASK_INPUT_STATUS__STOP,
L
Liu Jicong 已提交
43
  TASK_INPUT_STATUS__FAILED,
L
Liu Jicong 已提交
44 45 46 47 48 49
};

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

L
Liu Jicong 已提交
52 53 54 55 56
enum {
  STREAM_CREATED_BY__USER = 1,
  STREAM_CREATED_BY__SMA,
};

L
Liu Jicong 已提交
57 58 59 60 61 62
enum {
  STREAM_INPUT__DATA_SUBMIT = 1,
  STREAM_INPUT__DATA_BLOCK,
  STREAM_INPUT__CHECKPOINT,
};

L
Liu Jicong 已提交
63
typedef struct {
L
Liu Jicong 已提交
64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109
  int8_t type;

  int32_t sourceVg;
  int64_t sourceVer;

  int32_t*    dataRef;
  SSubmitReq* data;
} SStreamDataSubmit;

typedef struct {
  int8_t type;

  int32_t sourceVg;
  int64_t sourceVer;

  SArray* blocks;  // SArray<SSDataBlock*>
} SStreamDataBlock;

typedef struct {
  int8_t type;
} SStreamCheckpoint;

static FORCE_INLINE SStreamDataSubmit* streamDataSubmitNew(SSubmitReq* pReq) {
  SStreamDataSubmit* pDataSubmit = (SStreamDataSubmit*)taosMemoryCalloc(1, sizeof(SStreamDataSubmit));
  if (pDataSubmit == NULL) return NULL;
  pDataSubmit->data = pReq;
  pDataSubmit->dataRef = (int32_t*)taosMemoryMalloc(sizeof(int32_t));
  if (pDataSubmit->data == NULL) goto FAIL;
  *pDataSubmit->dataRef = 1;
  return pDataSubmit;
FAIL:
  taosMemoryFree(pDataSubmit);
  return NULL;
}

static FORCE_INLINE void streamDataSubmitRefInc(SStreamDataSubmit* pDataSubmit) {
  //
  atomic_add_fetch_32(pDataSubmit->dataRef, 1);
}

static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit) {
  int32_t ref = atomic_sub_fetch_32(pDataSubmit->dataRef, 1);
  ASSERT(ref >= 0);
  if (ref == 0) {
    taosMemoryFree(pDataSubmit->data);
    taosMemoryFree(pDataSubmit->dataRef);
L
Liu Jicong 已提交
110
    // taosFreeQitem(pDataSubmit);
L
Liu Jicong 已提交
111 112 113 114 115
  }
}

int32_t streamDataBlockEncode(void** buf, const SStreamDataBlock* pOutput);
void*   streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput);
L
Liu Jicong 已提交
116 117 118 119 120 121 122 123 124 125 126 127 128 129 130

typedef struct {
  void* inputHandle;
  void* executor;
} SStreamRunner;

typedef struct {
  int8_t parallelizable;
  char*  qmsg;
  // followings are not applicable to encoder and decoder
  int8_t         numOfRunners;
  SStreamRunner* runners;
} STaskExec;

typedef struct {
L
Liu Jicong 已提交
131
  int32_t taskId;
L
Liu Jicong 已提交
132 133 134
} STaskDispatcherInplace;

typedef struct {
L
Liu Jicong 已提交
135
  int32_t taskId;
L
Liu Jicong 已提交
136 137 138 139 140
  int32_t nodeId;
  SEpSet  epSet;
} STaskDispatcherFixedEp;

typedef struct {
L
Liu Jicong 已提交
141 142 143
  // int8_t  hashMethod;
  char      stbFullName[TSDB_TABLE_FNAME_LEN];
  SUseDbRsp dbInfo;
L
Liu Jicong 已提交
144 145
} STaskDispatcherShuffle;

L
Liu Jicong 已提交
146
typedef void FTbSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
L
Liu Jicong 已提交
147

L
Liu Jicong 已提交
148
typedef struct {
L
Liu Jicong 已提交
149
  int64_t         stbUid;
L
Liu Jicong 已提交
150
  SSchemaWrapper* pSchemaWrapper;
L
Liu Jicong 已提交
151
  // not applicable to encoder and decoder
L
Liu Jicong 已提交
152 153
  void*     vnode;
  FTbSink*  tbSinkFunc;
L
Liu Jicong 已提交
154
  STSchema* pTSchema;
L
Liu Jicong 已提交
155 156 157
  SHashObj* pHash;  // groupId to tbuid
} STaskSinkTb;

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

L
Liu Jicong 已提交
160
typedef struct {
L
Liu Jicong 已提交
161 162
  int64_t smaId;
  // following are not applicable to encoder and decoder
L
Liu Jicong 已提交
163
  FSmaSink* smaSink;
L
Liu Jicong 已提交
164 165 166 167 168 169 170 171
} STaskSinkSma;

typedef struct {
  int8_t reserved;
} STaskSinkFetch;

enum {
  TASK_SOURCE__SCAN = 1,
L
Liu Jicong 已提交
172 173
  TASK_SOURCE__PIPE,
  TASK_SOURCE__MERGE,
L
Liu Jicong 已提交
174 175 176 177
};

enum {
  TASK_EXEC__NONE = 1,
L
Liu Jicong 已提交
178 179
  TASK_EXEC__PIPE,
  TASK_EXEC__MERGE,
L
Liu Jicong 已提交
180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195
};

enum {
  TASK_DISPATCH__NONE = 1,
  TASK_DISPATCH__INPLACE,
  TASK_DISPATCH__FIXED,
  TASK_DISPATCH__SHUFFLE,
};

enum {
  TASK_SINK__NONE = 1,
  TASK_SINK__TABLE,
  TASK_SINK__SMA,
  TASK_SINK__FETCH,
};

L
Liu Jicong 已提交
196 197 198 199 200
enum {
  TASK_INPUT_TYPE__SUMBIT_BLOCK = 1,
  TASK_INPUT_TYPE__DATA_BLOCK,
};

L
Liu Jicong 已提交
201
struct SStreamTask {
L
Liu Jicong 已提交
202 203
  int64_t streamId;
  int32_t taskId;
L
Liu Jicong 已提交
204
  int8_t  inputType;
L
Liu Jicong 已提交
205 206 207 208 209 210 211 212
  int8_t  status;

  int8_t  sourceType;
  int8_t  execType;
  int8_t  sinkType;
  int8_t  dispatchType;
  int16_t dispatchMsgType;

L
Liu Jicong 已提交
213 214 215
  int32_t nodeId;
  SEpSet  epSet;

L
Liu Jicong 已提交
216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234
  // source preprocess

  // exec
  STaskExec exec;

  // local sink
  union {
    STaskSinkTb    tbSink;
    STaskSinkSma   smaSink;
    STaskSinkFetch fetchSink;
  };

  // dispatch
  union {
    STaskDispatcherInplace inplaceDispatcher;
    STaskDispatcherFixedEp fixedEpDispatcher;
    STaskDispatcherShuffle shuffleDispatcher;
  };

L
Liu Jicong 已提交
235 236 237
  int8_t inputStatus;
  int8_t outputStatus;

L
Liu Jicong 已提交
238
  STaosQueue* inputQ;
L
Liu Jicong 已提交
239
  STaosQall*  inputQAll;
L
Liu Jicong 已提交
240
  STaosQueue* outputQ;
L
Liu Jicong 已提交
241
  STaosQall*  outputQAll;
L
Liu Jicong 已提交
242

L
Liu Jicong 已提交
243 244
  // application storage
  void* ahandle;
L
Liu Jicong 已提交
245
};
L
Liu Jicong 已提交
246 247

SStreamTask* tNewSStreamTask(int64_t streamId);
H
Hongze Cheng 已提交
248 249
int32_t      tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask);
int32_t      tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask);
L
Liu Jicong 已提交
250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282
void         tFreeSStreamTask(SStreamTask* pTask);

typedef struct {
  // SMsgHead     head;
  SStreamTask* task;
} SStreamTaskDeployReq;

typedef struct {
  int32_t reserved;
} SStreamTaskDeployRsp;

typedef struct {
  // SMsgHead head;
  int64_t streamId;
  int32_t taskId;
  SArray* data;  // SArray<SSDataBlock>
} SStreamTaskExecReq;

int32_t tEncodeSStreamTaskExecReq(void** buf, const SStreamTaskExecReq* pReq);
void*   tDecodeSStreamTaskExecReq(const void* buf, SStreamTaskExecReq* pReq);
void    tFreeSStreamTaskExecReq(SStreamTaskExecReq* pReq);

typedef struct {
  int32_t reserved;
} SStreamTaskExecRsp;

typedef struct {
  // SMsgHead head;
  int64_t streamId;
  int64_t version;
  SArray* res;  // SArray<SSDataBlock>
} SStreamSinkReq;

L
Liu Jicong 已提交
283 284 285 286 287 288
typedef struct {
  SMsgHead head;
  int64_t  streamId;
  int32_t  taskId;
} SStreamTaskRunReq;

L
Liu Jicong 已提交
289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318
typedef struct {
  int64_t streamId;
  int32_t taskId;
  int32_t sourceTaskId;
  int32_t sourceVg;
#if 0
  int64_t sourceVer;
#endif
  SArray* data;  // SArray<SSDataBlock>
} SStreamDispatchReq;

typedef struct {
  int64_t streamId;
  int32_t taskId;
  int8_t  inputStatus;
} SStreamDispatchRsp;

typedef struct {
  int64_t streamId;
  int32_t taskId;
  int32_t sourceTaskId;
  int32_t sourceVg;
} SStreamTaskRecoverReq;

typedef struct {
  int64_t streamId;
  int32_t taskId;
  int8_t  inputStatus;
} SStreamTaskRecoverRsp;

L
Liu Jicong 已提交
319 320 321
int32_t streamEnqueueDataSubmit(SStreamTask* pTask, SStreamDataSubmit* input);
int32_t streamEnqueueDataBlk(SStreamTask* pTask, SStreamDataBlock* input);
int32_t streamDequeueOutput(SStreamTask* pTask, void** output);
L
Liu Jicong 已提交
322

L
Liu Jicong 已提交
323 324
int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId);

L
Liu Jicong 已提交
325
int32_t streamTaskRun(SStreamTask* pTask);
L
Liu Jicong 已提交
326 327 328

int32_t streamTaskHandleInput(SStreamTask* pTask, void* data);

L
Liu Jicong 已提交
329 330 331 332 333 334
int32_t streamTaskProcessRunReq(SStreamTask* pTask, SMsgCb* pMsgCb);
int32_t streamTaskProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pMsg);
int32_t streamTaskProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchRsp* pRsp);
int32_t streamTaskProcessRecoverReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamTaskRecoverReq* pReq, SRpcMsg* pMsg);
int32_t streamTaskProcessRecoverRsp(SStreamTask* pTask, SStreamTaskRecoverRsp* pRsp);

L
Liu Jicong 已提交
335 336 337 338 339
#ifdef __cplusplus
}
#endif

#endif /* ifndef _TSTREAM_H_ */