tq.c 14.3 KB
Newer Older
H
refact  
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
/*
 * 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/>.
S
Shengliang Guan 已提交
14 15
 */

H
Hongze Cheng 已提交
16
#include "tq.h"
S
Shengliang Guan 已提交
17

L
Liu Jicong 已提交
18
int32_t tqInit() {
L
Liu Jicong 已提交
19 20 21 22 23 24
  int8_t old;
  while (1) {
    old = atomic_val_compare_exchange_8(&tqMgmt.inited, 0, 2);
    if (old != 2) break;
  }

25 26 27 28 29 30
  if (old == 0) {
    tqMgmt.timer = taosTmrInit(10000, 100, 10000, "TQ");
    if (tqMgmt.timer == NULL) {
      atomic_store_8(&tqMgmt.inited, 0);
      return -1;
    }
L
Liu Jicong 已提交
31
    atomic_store_8(&tqMgmt.inited, 1);
32
  }
L
Liu Jicong 已提交
33 34
  return 0;
}
L
Liu Jicong 已提交
35

36
void tqCleanUp() {
L
Liu Jicong 已提交
37 38 39 40 41 42 43 44 45 46
  int8_t old;
  while (1) {
    old = atomic_val_compare_exchange_8(&tqMgmt.inited, 1, 2);
    if (old != 2) break;
  }

  if (old == 1) {
    taosTmrCleanUp(tqMgmt.timer);
    atomic_store_8(&tqMgmt.inited, 0);
  }
47
}
L
Liu Jicong 已提交
48

L
Liu Jicong 已提交
49
STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) {
wafwerar's avatar
wafwerar 已提交
50
  STQ* pTq = taosMemoryMalloc(sizeof(STQ));
L
Liu Jicong 已提交
51
  if (pTq == NULL) {
L
Liu Jicong 已提交
52
    terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
L
Liu Jicong 已提交
53 54
    return NULL;
  }
H
Hongze Cheng 已提交
55
  pTq->path = strdup(path);
L
Liu Jicong 已提交
56
  pTq->pVnode = pVnode;
L
Liu Jicong 已提交
57
  pTq->pWal = pWal;
58

L
Liu Jicong 已提交
59
  pTq->handles = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
60

L
Liu Jicong 已提交
61 62
  pTq->pStreamTasks = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);

L
Liu Jicong 已提交
63 64
  pTq->pushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);

L
Liu Jicong 已提交
65
  if (tqMetaOpen(pTq) < 0) {
66 67 68
    ASSERT(0);
  }

L
Liu Jicong 已提交
69 70
  return pTq;
}
L
Liu Jicong 已提交
71

L
Liu Jicong 已提交
72
void tqClose(STQ* pTq) {
H
Hongze Cheng 已提交
73
  if (pTq) {
wafwerar's avatar
wafwerar 已提交
74
    taosMemoryFreeClear(pTq->path);
L
Liu Jicong 已提交
75
    taosHashCleanup(pTq->handles);
L
Liu Jicong 已提交
76 77
    taosHashCleanup(pTq->pStreamTasks);
    taosHashCleanup(pTq->pushMgr);
L
Liu Jicong 已提交
78
    tqMetaClose(pTq);
wafwerar's avatar
wafwerar 已提交
79
    taosMemoryFree(pTq);
H
Hongze Cheng 已提交
80
  }
L
Liu Jicong 已提交
81 82
  // TODO
}
L
Liu Jicong 已提交
83

L
Liu Jicong 已提交
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 110 111
int32_t tqSendPollRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataBlkRsp* pRsp) {
  int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqDataBlkRsp(NULL, pRsp);
  void*   buf = rpcMallocCont(tlen);
  if (buf == NULL) {
    return -1;
  }

  ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
  ((SMqRspHead*)buf)->epoch = pReq->epoch;
  ((SMqRspHead*)buf)->consumerId = pReq->consumerId;

  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
  tEncodeSMqDataBlkRsp(&abuf, pRsp);

  SRpcMsg resp = {
      .info = pMsg->info,
      .pCont = buf,
      .contLen = tlen,
      .code = 0,
  };
  tmsgSendRsp(&resp);

  tqDebug("vg %d from consumer %ld (epoch %d) send rsp, block num: %d, reqOffset: %ld, rspOffset: %ld",
          TD_VID(pTq->pVnode), pReq->consumerId, pReq->epoch, pRsp->blockNum, pRsp->reqOffset, pRsp->rspOffset);

  return 0;
}

L
fix  
Liu Jicong 已提交
112
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
L
Liu Jicong 已提交
113 114
  SMqPollReq* pReq = pMsg->pCont;
  int64_t     consumerId = pReq->consumerId;
L
Liu Jicong 已提交
115
  int64_t     timeout = pReq->timeout;
L
Liu Jicong 已提交
116 117
  int32_t     reqEpoch = pReq->epoch;
  int64_t     fetchOffset;
L
Liu Jicong 已提交
118
  int32_t     code = 0;
119

L
Liu Jicong 已提交
120
  // get offset to fetch message
L
Liu Jicong 已提交
121
  if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__EARLIEAST) {
L
Liu Jicong 已提交
122
    fetchOffset = walGetFirstVer(pTq->pWal);
L
Liu Jicong 已提交
123
  } else if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__LATEST) {
L
Liu Jicong 已提交
124
    fetchOffset = walGetCommittedVer(pTq->pWal);
L
Liu Jicong 已提交
125 126 127 128
  } else {
    fetchOffset = pReq->currentOffset + 1;
  }

H
Hongze Cheng 已提交
129 130
  tqDebug("tmq poll: consumer %ld (epoch %d) recv poll req in vg %d, req %ld %ld", consumerId, pReq->epoch,
          TD_VID(pTq->pVnode), pReq->currentOffset, fetchOffset);
L
fix  
Liu Jicong 已提交
131

L
Liu Jicong 已提交
132 133
  STqHandle* pHandle = taosHashGet(pTq->handles, pReq->subKey, strlen(pReq->subKey));
  ASSERT(pHandle);
L
Liu Jicong 已提交
134

L
Liu Jicong 已提交
135
  int32_t consumerEpoch = atomic_load_32(&pHandle->epoch);
L
Liu Jicong 已提交
136
  while (consumerEpoch < reqEpoch) {
L
Liu Jicong 已提交
137
    consumerEpoch = atomic_val_compare_exchange_32(&pHandle->epoch, consumerEpoch, reqEpoch);
L
Liu Jicong 已提交
138 139
  }

140 141 142 143 144
  SWalHead* pHeadWithCkSum = taosMemoryMalloc(sizeof(SWalHead) + 2048);
  if (pHeadWithCkSum == NULL) {
    return -1;
  }

L
Liu Jicong 已提交
145
  walSetReaderCapacity(pHandle->pWalReader, 2048);
146

L
Liu Jicong 已提交
147
  SMqDataBlkRsp rsp = {0};
L
Liu Jicong 已提交
148
  rsp.reqOffset = pReq->currentOffset;
L
Liu Jicong 已提交
149

L
Liu Jicong 已提交
150
  rsp.blockData = taosArrayInit(0, sizeof(void*));
L
Liu Jicong 已提交
151
  rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
152

L
Liu Jicong 已提交
153
  rsp.withTbName = pReq->withTbName;
154 155 156
  if (rsp.withTbName) {
    rsp.blockTbName = taosArrayInit(0, sizeof(void*));
  }
L
Liu Jicong 已提交
157 158 159 160 161 162
  if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
    rsp.withSchema = false;
    rsp.withTag = false;
  } else {
    rsp.withSchema = true;
    rsp.withTag = false;
L
Liu Jicong 已提交
163
    rsp.blockSchema = taosArrayInit(0, sizeof(void*));
164
  }
L
Liu Jicong 已提交
165

166
  while (1) {
L
Liu Jicong 已提交
167
    consumerEpoch = atomic_load_32(&pHandle->epoch);
L
Liu Jicong 已提交
168
    if (consumerEpoch > reqEpoch) {
L
Liu Jicong 已提交
169 170
      tqWarn("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, found new consumer epoch %d, discard req epoch %d",
             consumerId, pReq->epoch, TD_VID(pTq->pVnode), fetchOffset, consumerEpoch, reqEpoch);
L
Liu Jicong 已提交
171 172
      break;
    }
L
Liu Jicong 已提交
173

L
Liu Jicong 已提交
174 175
    if (tqFetchLog(pTq, pHandle, &fetchOffset, &pHeadWithCkSum) < 0) {
      // TODO add push mgr
176 177 178 179 180
      break;
    }

    SWalReadHead* pHead = &pHeadWithCkSum->head;

H
Hongze Cheng 已提交
181 182
    tqDebug("tmq poll: consumer %ld (epoch %d) iter log, vg %d offset %ld msgType %d", consumerId, pReq->epoch,
            TD_VID(pTq->pVnode), fetchOffset, pHead->msgType);
L
Liu Jicong 已提交
183

L
fix  
Liu Jicong 已提交
184 185
    if (pHead->msgType == TDMT_VND_SUBMIT) {
      SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
L
Liu Jicong 已提交
186

L
Liu Jicong 已提交
187 188 189
      if (tqDataExec(pTq, &pHandle->execHandle, pCont, &rsp, workerId) < 0) {
        /*ASSERT(0);*/
      }
L
Liu Jicong 已提交
190 191 192
    } else {
      // TODO
      ASSERT(0);
L
Liu Jicong 已提交
193
    }
194

L
Liu Jicong 已提交
195 196
    // TODO batch optimization:
    // TODO continue scan until meeting batch requirement
L
Liu Jicong 已提交
197 198
    if (rsp.blockNum > 0 /* threshold */) {
      break;
L
Liu Jicong 已提交
199 200
    } else {
      fetchOffset++;
L
Liu Jicong 已提交
201
    }
L
Liu Jicong 已提交
202
  }
203

L
Liu Jicong 已提交
204
  taosMemoryFree(pHeadWithCkSum);
L
Liu Jicong 已提交
205

L
Liu Jicong 已提交
206 207
  ASSERT(taosArrayGetSize(rsp.blockData) == rsp.blockNum);
  ASSERT(taosArrayGetSize(rsp.blockDataLen) == rsp.blockNum);
L
Liu Jicong 已提交
208 209 210
  if (rsp.withSchema) {
    ASSERT(taosArrayGetSize(rsp.blockSchema) == rsp.blockNum);
  }
L
fix  
Liu Jicong 已提交
211

L
Liu Jicong 已提交
212
  rsp.rspOffset = fetchOffset;
213

L
Liu Jicong 已提交
214 215
  if (tqSendPollRsp(pTq, pMsg, pReq, &rsp) < 0) {
    code = -1;
216
  }
L
Liu Jicong 已提交
217

L
Liu Jicong 已提交
218
  // TODO wrap in destroy func
L
Liu Jicong 已提交
219
  taosArrayDestroy(rsp.blockDataLen);
220
  taosArrayDestroyP(rsp.blockData, (FDelete)taosMemoryFree);
221 222 223 224 225 226 227 228

  if (rsp.withSchema) {
    taosArrayDestroyP(rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
  }

  if (rsp.withTbName) {
    taosArrayDestroyP(rsp.blockTbName, (FDelete)taosMemoryFree);
  }
L
Liu Jicong 已提交
229

L
Liu Jicong 已提交
230
  return code;
231
}
L
Liu Jicong 已提交
232

L
Liu Jicong 已提交
233 234
int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen) {
  SMqVDeleteReq* pReq = (SMqVDeleteReq*)msg;
L
Liu Jicong 已提交
235

L
Liu Jicong 已提交
236
  int32_t code = taosHashRemove(pTq->handles, pReq->subKey, strlen(pReq->subKey));
L
Liu Jicong 已提交
237
  ASSERT(code == 0);
238

L
Liu Jicong 已提交
239
  if (tqMetaDeleteHandle(pTq, pReq->subKey) < 0) {
240 241
    ASSERT(0);
  }
L
Liu Jicong 已提交
242
  return 0;
L
Liu Jicong 已提交
243 244
}

L
Liu Jicong 已提交
245 246
// TODO: persist meta into tdb
int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
L
Liu Jicong 已提交
247
  SMqRebVgReq req = {0};
L
Liu Jicong 已提交
248 249
  tDecodeSMqRebVgReq(msg, &req);
  // todo lock
L
Liu Jicong 已提交
250 251
  STqHandle* pHandle = taosHashGet(pTq->handles, req.subKey, strlen(req.subKey));
  if (pHandle == NULL) {
L
Liu Jicong 已提交
252 253
    ASSERT(req.oldConsumerId == -1);
    ASSERT(req.newConsumerId != -1);
L
Liu Jicong 已提交
254 255
    STqHandle tqHandle = {0};
    pHandle = &tqHandle;
L
Liu Jicong 已提交
256 257
    /*taosInitRWLatch(&pExec->lock);*/

L
Liu Jicong 已提交
258 259 260
    memcpy(pHandle->subKey, req.subKey, TSDB_SUBSCRIBE_KEY_LEN);
    pHandle->consumerId = req.newConsumerId;
    pHandle->epoch = -1;
L
Liu Jicong 已提交
261

L
Liu Jicong 已提交
262
    pHandle->execHandle.subType = req.subType;
L
Liu Jicong 已提交
263

L
Liu Jicong 已提交
264 265 266 267 268
    pHandle->pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
    for (int32_t i = 0; i < 5; i++) {
      pHandle->execHandle.pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
    }
    if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
L
Liu Jicong 已提交
269
      pHandle->execHandle.execCol.qmsg = req.qmsg;
L
Liu Jicong 已提交
270
      req.qmsg = NULL;
271
      for (int32_t i = 0; i < 5; i++) {
L
Liu Jicong 已提交
272
        SReadHandle handle = {
L
Liu Jicong 已提交
273
            .reader = pHandle->execHandle.pExecReader[i],
L
Liu Jicong 已提交
274
            .meta = pTq->pVnode->pMeta,
275
            .pMsgCb = &pTq->pVnode->msgCb,
L
Liu Jicong 已提交
276
        };
L
Liu Jicong 已提交
277 278
        pHandle->execHandle.execCol.task[i] = qCreateStreamExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle);
        ASSERT(pHandle->execHandle.execCol.task[i]);
279
      }
L
Liu Jicong 已提交
280
    } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
L
Liu Jicong 已提交
281
      pHandle->execHandle.execDb.pFilterOutTbUid =
L
Liu Jicong 已提交
282 283
          taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
    } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
L
Liu Jicong 已提交
284
      pHandle->execHandle.execTb.suid = req.suid;
L
Liu Jicong 已提交
285 286 287 288 289 290 291
      SArray* tbUidList = taosArrayInit(0, sizeof(int64_t));
      tsdbGetCtbIdList(pTq->pVnode->pMeta, req.suid, tbUidList);
      tqDebug("vg %d, tq try get suid: %ld", pTq->pVnode->config.vgId, req.suid);
      for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
        int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i);
        tqDebug("vg %d, idx %d, uid: %ld", pTq->pVnode->config.vgId, i, tbUid);
      }
L
Liu Jicong 已提交
292 293 294
      for (int32_t i = 0; i < 5; i++) {
        tqReadHandleSetTbUidList(pHandle->execHandle.pExecReader[i], tbUidList);
      }
L
Liu Jicong 已提交
295
      taosArrayDestroy(tbUidList);
L
Liu Jicong 已提交
296
    }
L
Liu Jicong 已提交
297
    taosHashPut(pTq->handles, req.subKey, strlen(req.subKey), pHandle, sizeof(STqHandle));
L
Liu Jicong 已提交
298 299 300
    if (tqMetaSaveHandle(pTq, req.subKey, pHandle) < 0) {
      // TODO
    }
L
Liu Jicong 已提交
301
  } else {
L
Liu Jicong 已提交
302
    /*ASSERT(pExec->consumerId == req.oldConsumerId);*/
L
Liu Jicong 已提交
303
    // TODO handle qmsg and exec modification
L
Liu Jicong 已提交
304 305 306
    atomic_store_32(&pHandle->epoch, -1);
    atomic_store_64(&pHandle->consumerId, req.newConsumerId);
    atomic_add_fetch_32(&pHandle->epoch, 1);
L
Liu Jicong 已提交
307 308 309
    if (tqMetaSaveHandle(pTq, req.subKey, pHandle) < 0) {
      // TODO
    }
L
Liu Jicong 已提交
310
  }
L
Liu Jicong 已提交
311

L
Liu Jicong 已提交
312
  return 0;
L
Liu Jicong 已提交
313
}
314

L
Liu Jicong 已提交
315 316 317 318 319 320 321 322 323 324 325 326
int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
  SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
  if (pTask == NULL) {
    return -1;
  }
  SDecoder decoder;
  tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
  if (tDecodeSStreamTask(&decoder, pTask) < 0) {
    ASSERT(0);
  }
  tDecoderClear(&decoder);

L
Liu Jicong 已提交
327
  pTask->status = TASK_STATUS__IDLE;
L
Liu Jicong 已提交
328 329 330

  pTask->inputQueue = streamQueueOpen();
  pTask->outputQueue = streamQueueOpen();
L
Liu Jicong 已提交
331 332 333
  pTask->inputStatus = TASK_INPUT_STATUS__NORMAL;
  pTask->outputStatus = TASK_OUTPUT_STATUS__NORMAL;

L
Liu Jicong 已提交
334
  if (pTask->inputQueue == NULL || pTask->outputQueue == NULL) goto FAIL;
L
Liu Jicong 已提交
335

L
Liu Jicong 已提交
336
  // exec
L
Liu Jicong 已提交
337 338
  if (pTask->execType != TASK_EXEC__NONE) {
    // expand runners
L
Liu Jicong 已提交
339 340 341 342 343 344 345 346 347 348
    STqReadHandle* pStreamReader = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
    SReadHandle    handle = {
           .reader = pStreamReader,
           .meta = pTq->pVnode->pMeta,
           .pMsgCb = &pTq->pVnode->msgCb,
           .vnode = pTq->pVnode,
    };
    pTask->exec.inputHandle = pStreamReader;
    pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
    ASSERT(pTask->exec.executor);
L
Liu Jicong 已提交
349
  }
L
Liu Jicong 已提交
350 351

  // sink
L
Liu Jicong 已提交
352
  /*pTask->ahandle = pTq->pVnode;*/
L
Liu Jicong 已提交
353
  if (pTask->sinkType == TASK_SINK__SMA) {
L
Liu Jicong 已提交
354
    pTask->smaSink.vnode = pTq->pVnode;
L
Liu Jicong 已提交
355
    pTask->smaSink.smaSink = smaHandleRes;
L
Liu Jicong 已提交
356
  } else if (pTask->sinkType == TASK_SINK__TABLE) {
L
Liu Jicong 已提交
357 358 359
    pTask->tbSink.vnode = pTq->pVnode;
    pTask->tbSink.tbSinkFunc = tqTableSink;

L
Liu Jicong 已提交
360 361
    ASSERT(pTask->tbSink.pSchemaWrapper);
    ASSERT(pTask->tbSink.pSchemaWrapper->pSchema);
L
Liu Jicong 已提交
362

L
Liu Jicong 已提交
363 364 365
    pTask->tbSink.pTSchema =
        tdGetSTSChemaFromSSChema(&pTask->tbSink.pSchemaWrapper->pSchema, pTask->tbSink.pSchemaWrapper->nCols);
    ASSERT(pTask->tbSink.pTSchema);
L
Liu Jicong 已提交
366
  }
L
Liu Jicong 已提交
367
  tqInfo("deploy stream task id %d child id %d on vg %d", pTask->taskId, pTask->childId, pTq->pVnode->config.vgId);
L
Liu Jicong 已提交
368

L
Liu Jicong 已提交
369 370
  taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), pTask, sizeof(SStreamTask));

L
Liu Jicong 已提交
371 372
  return 0;
FAIL:
L
Liu Jicong 已提交
373 374
  if (pTask->inputQueue) streamQueueClose(pTask->inputQueue);
  if (pTask->outputQueue) streamQueueClose(pTask->outputQueue);
L
Liu Jicong 已提交
375
  if (pTask) taosMemoryFree(pTask);
L
Liu Jicong 已提交
376 377
  return -1;
}
L
Liu Jicong 已提交
378

L
Liu Jicong 已提交
379
int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
L
Liu Jicong 已提交
380 381 382
  void*              pIter = NULL;
  bool               failed = false;
  SStreamDataSubmit* pSubmit = NULL;
L
Liu Jicong 已提交
383

L
Liu Jicong 已提交
384
  pSubmit = streamDataSubmitNew(pReq);
L
Liu Jicong 已提交
385 386 387 388 389 390 391 392 393 394
  if (pSubmit == NULL) {
    failed = true;
  }

  while (1) {
    pIter = taosHashIterate(pTq->pStreamTasks, pIter);
    if (pIter == NULL) break;
    SStreamTask* pTask = (SStreamTask*)pIter;
    if (pTask->inputType != STREAM_INPUT__DATA_SUBMIT) continue;

L
Liu Jicong 已提交
395 396
    if (!failed) {
      if (streamTaskInput(pTask, (SStreamQueueItem*)pSubmit) < 0) {
L
Liu Jicong 已提交
397 398 399
        continue;
      }

L
Liu Jicong 已提交
400
      if (streamTriggerByWrite(pTask, pTq->pVnode->config.vgId, &pTq->pVnode->msgCb) < 0) {
L
Liu Jicong 已提交
401 402
        continue;
      }
L
Liu Jicong 已提交
403
    } else {
L
Liu Jicong 已提交
404
      streamTaskInputFail(pTask);
L
Liu Jicong 已提交
405 406 407
    }
  }

L
Liu Jicong 已提交
408
  if (pSubmit) {
L
Liu Jicong 已提交
409
    streamDataSubmitRefDec(pSubmit);
L
Liu Jicong 已提交
410
    taosFreeQitem(pSubmit);
L
Liu Jicong 已提交
411
  }
L
Liu Jicong 已提交
412 413

  return failed ? -1 : 0;
L
Liu Jicong 已提交
414 415
}

L
Liu Jicong 已提交
416
int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {
L
Liu Jicong 已提交
417
  //
L
Liu Jicong 已提交
418 419 420 421 422 423 424 425
  SStreamTaskRunReq* pReq = pMsg->pCont;
  int32_t            taskId = pReq->taskId;
  SStreamTask*       pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
  streamTaskProcessRunReq(pTask, &pTq->pVnode->msgCb);
  return 0;
}

int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg) {
426 427 428 429 430 431 432 433 434 435 436 437 438 439
  char*              msgStr = pMsg->pCont;
  char*              msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead));
  int32_t            msgLen = pMsg->contLen - sizeof(SMsgHead);
  SStreamDispatchReq req;
  SDecoder           decoder;
  tDecoderInit(&decoder, msgBody, msgLen);
  tDecodeStreamDispatchReq(&decoder, &req);
  int32_t      taskId = req.taskId;
  SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
  SRpcMsg      rsp = {
           .info = pMsg->info,
           .code = 0,
  };
  streamProcessDispatchReq(pTask, &pTq->pVnode->msgCb, &req, &rsp);
L
Liu Jicong 已提交
440 441 442 443 444 445 446
  return 0;
}

int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg) {
  SStreamTaskRecoverReq* pReq = pMsg->pCont;
  int32_t                taskId = pReq->taskId;
  SStreamTask*           pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
L
Liu Jicong 已提交
447
  streamProcessRecoverReq(pTask, &pTq->pVnode->msgCb, pReq, pMsg);
L
Liu Jicong 已提交
448 449 450 451
  return 0;
}

int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
L
Liu Jicong 已提交
452
  SStreamDispatchRsp* pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
L
Liu Jicong 已提交
453 454
  int32_t             taskId = pRsp->taskId;
  SStreamTask*        pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
L
Liu Jicong 已提交
455
  streamProcessDispatchRsp(pTask, &pTq->pVnode->msgCb, pRsp);
L
Liu Jicong 已提交
456 457 458 459 460 461 462
  return 0;
}

int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg) {
  SStreamTaskRecoverRsp* pRsp = pMsg->pCont;
  int32_t                taskId = pRsp->taskId;
  SStreamTask*           pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
L
Liu Jicong 已提交
463
  streamProcessRecoverRsp(pTask, pRsp);
L
Liu Jicong 已提交
464 465
  return 0;
}