tq.c 30.4 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 19 20 21
int32_t tqInit() {
  //
  return 0;
}
L
Liu Jicong 已提交
22

L
Liu Jicong 已提交
23
void tqCleanUp() {}
L
Liu Jicong 已提交
24

L
Liu Jicong 已提交
25
STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) {
wafwerar's avatar
wafwerar 已提交
26
  STQ* pTq = taosMemoryMalloc(sizeof(STQ));
L
Liu Jicong 已提交
27
  if (pTq == NULL) {
L
Liu Jicong 已提交
28
    terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
L
Liu Jicong 已提交
29 30
    return NULL;
  }
H
Hongze Cheng 已提交
31
  pTq->path = strdup(path);
L
Liu Jicong 已提交
32
  pTq->pVnode = pVnode;
L
Liu Jicong 已提交
33
  pTq->pWal = pWal;
34

L
Liu Jicong 已提交
35
#if 0
L
Liu Jicong 已提交
36 37
  pTq->tqMeta = tqStoreOpen(pTq, path, (FTqSerialize)tqSerializeConsumer, (FTqDeserialize)tqDeserializeConsumer,
                            (FTqDelete)taosMemoryFree, 0);
L
Liu Jicong 已提交
38
  if (pTq->tqMeta == NULL) {
wafwerar's avatar
wafwerar 已提交
39
    taosMemoryFree(pTq);
L
Liu Jicong 已提交
40 41
    return NULL;
  }
L
Liu Jicong 已提交
42
#endif
L
Liu Jicong 已提交
43

L
Liu Jicong 已提交
44
  pTq->execs = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
45

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

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

L
Liu Jicong 已提交
50 51
  return pTq;
}
L
Liu Jicong 已提交
52

L
Liu Jicong 已提交
53
void tqClose(STQ* pTq) {
H
Hongze Cheng 已提交
54
  if (pTq) {
wafwerar's avatar
wafwerar 已提交
55 56
    taosMemoryFreeClear(pTq->path);
    taosMemoryFree(pTq);
H
Hongze Cheng 已提交
57
  }
L
Liu Jicong 已提交
58 59
  // TODO
}
L
Liu Jicong 已提交
60

L
Liu Jicong 已提交
61 62 63 64 65 66 67 68 69 70 71
int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
  if (msgType != TDMT_VND_SUBMIT) return 0;
  void*       pIter = NULL;
  STqExec*    pExec = NULL;
  SSubmitReq* pReq = (SSubmitReq*)msg;
  int32_t     workerId = 4;
  int64_t     fetchOffset = ver;

  while (1) {
    pIter = taosHashIterate(pTq->pushMgr, pIter);
    if (pIter == NULL) break;
L
Liu Jicong 已提交
72
    pExec = *(STqExec**)pIter;
L
Liu Jicong 已提交
73 74 75 76 77 78 79 80

    taosWLockLatch(&pExec->pushHandle.lock);

    SRpcMsg* pMsg = atomic_load_ptr(&pExec->pushHandle.handle);
    ASSERT(pMsg);

    SMqDataBlkRsp rsp = {0};
    rsp.reqOffset = pExec->pushHandle.reqOffset;
81
    rsp.blockData = taosArrayInit(0, sizeof(void*));
L
Liu Jicong 已提交
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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178
    rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));

    if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
      qTaskInfo_t task = pExec->task[workerId];
      ASSERT(task);
      qSetStreamInput(task, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK);
      while (1) {
        SSDataBlock* pDataBlock = NULL;
        uint64_t     ts = 0;
        if (qExecTask(task, &pDataBlock, &ts) < 0) {
          ASSERT(0);
        }
        if (pDataBlock == NULL) break;

        ASSERT(pDataBlock->info.rows != 0);
        ASSERT(pDataBlock->info.numOfCols != 0);

        int32_t            dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pDataBlock);
        void*              buf = taosMemoryCalloc(1, dataStrLen);
        SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
        pRetrieve->useconds = ts;
        pRetrieve->precision = TSDB_DEFAULT_PRECISION;
        pRetrieve->compressed = 0;
        pRetrieve->completed = 1;
        pRetrieve->numOfRows = htonl(pDataBlock->info.rows);

        // TODO enable compress
        int32_t actualLen = 0;
        blockCompressEncode(pDataBlock, pRetrieve->data, &actualLen, pDataBlock->info.numOfCols, false);
        actualLen += sizeof(SRetrieveTableRsp);
        ASSERT(actualLen <= dataStrLen);
        taosArrayPush(rsp.blockDataLen, &actualLen);
        taosArrayPush(rsp.blockData, &buf);
        rsp.blockNum++;
      }
    } else if (pExec->subType == TOPIC_SUB_TYPE__DB) {
      STqReadHandle* pReader = pExec->pExecReader[workerId];
      tqReadHandleSetMsg(pReader, pReq, 0);
      while (tqNextDataBlock(pReader)) {
        SSDataBlock block = {0};
        if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.rows,
                                &block.info.numOfCols) < 0) {
          ASSERT(0);
        }
        int32_t            dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(&block);
        void*              buf = taosMemoryCalloc(1, dataStrLen);
        SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
        /*pRetrieve->useconds = 0;*/
        pRetrieve->precision = TSDB_DEFAULT_PRECISION;
        pRetrieve->compressed = 0;
        pRetrieve->completed = 1;
        pRetrieve->numOfRows = htonl(block.info.rows);

        // TODO enable compress
        int32_t actualLen = 0;
        blockCompressEncode(&block, pRetrieve->data, &actualLen, block.info.numOfCols, false);
        actualLen += sizeof(SRetrieveTableRsp);
        ASSERT(actualLen <= dataStrLen);
        taosArrayPush(rsp.blockDataLen, &actualLen);
        taosArrayPush(rsp.blockData, &buf);
        rsp.blockNum++;
      }
    } else {
      ASSERT(0);
    }

    if (rsp.blockNum == 0) {
      taosWUnLockLatch(&pExec->pushHandle.lock);
      continue;
    }

    ASSERT(taosArrayGetSize(rsp.blockData) == rsp.blockNum);
    ASSERT(taosArrayGetSize(rsp.blockDataLen) == rsp.blockNum);

    rsp.rspOffset = fetchOffset;

    int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqDataBlkRsp(NULL, &rsp);
    void*   buf = rpcMallocCont(tlen);
    if (buf == NULL) {
      pMsg->code = -1;
      return -1;
    }

    ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
    ((SMqRspHead*)buf)->epoch = pExec->pushHandle.epoch;
    ((SMqRspHead*)buf)->consumerId = pExec->pushHandle.consumerId;

    void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
    tEncodeSMqDataBlkRsp(&abuf, &rsp);
    pMsg->pCont = buf;
    pMsg->contLen = tlen;
    pMsg->code = 0;
    tmsgSendRsp(pMsg);

    atomic_store_ptr(&pExec->pushHandle.handle, NULL);
    taosWUnLockLatch(&pExec->pushHandle.lock);

H
Hongze Cheng 已提交
179 180 181
    tqDebug("vg %d offset %ld from consumer %ld (epoch %d) send rsp, block num: %d, reqOffset: %ld, rspOffset: %ld",
            TD_VID(pTq->pVnode), fetchOffset, pExec->pushHandle.consumerId, pExec->pushHandle.epoch, rsp.blockNum,
            rsp.reqOffset, rsp.rspOffset);
L
Liu Jicong 已提交
182 183 184 185 186 187 188 189 190

    // TODO destroy
    taosArrayDestroy(rsp.blockData);
    taosArrayDestroy(rsp.blockDataLen);
  }

  return 0;
}

L
Liu Jicong 已提交
191
int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
L
Liu Jicong 已提交
192
  if (msgType != TDMT_VND_SUBMIT) return 0;
L
Liu Jicong 已提交
193

wafwerar's avatar
wafwerar 已提交
194
  void* data = taosMemoryMalloc(msgLen);
L
Liu Jicong 已提交
195 196
  if (data == NULL) {
    return -1;
L
Liu Jicong 已提交
197
  }
L
Liu Jicong 已提交
198
  memcpy(data, msg, msgLen);
C
Cary Xu 已提交
199

C
Cary Xu 已提交
200 201 202
  // make sure msgType == TDMT_VND_SUBMIT
  if (tsdbUpdateSmaWindow(pTq->pVnode->pTsdb, msg, ver) != 0) {
    return -1;
C
Cary Xu 已提交
203 204
  }

L
Liu Jicong 已提交
205 206 207 208 209
  SRpcMsg req = {
      .msgType = TDMT_VND_STREAM_TRIGGER,
      .pCont = data,
      .contLen = msgLen,
  };
L
Liu Jicong 已提交
210

L
Liu Jicong 已提交
211
  tmsgPutToQueue(&pTq->pVnode->msgCb, FETCH_QUEUE, &req);
L
Liu Jicong 已提交
212

L
Liu Jicong 已提交
213 214 215
  return 0;
}

L
Liu Jicong 已提交
216 217 218 219 220
int tqCommit(STQ* pTq) {
  // do nothing
  /*return tqStorePersist(pTq->tqMeta);*/
  return 0;
}
L
Liu Jicong 已提交
221 222

int32_t tqGetTopicHandleSize(const STqTopic* pTopic) {
L
Liu Jicong 已提交
223 224
  return strlen(pTopic->topicName) + strlen(pTopic->sql) + strlen(pTopic->physicalPlan) + strlen(pTopic->qmsg) +
         sizeof(int64_t) * 3;
L
Liu Jicong 已提交
225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242
}

int32_t tqGetConsumerHandleSize(const STqConsumer* pConsumer) {
  int     num = taosArrayGetSize(pConsumer->topics);
  int32_t sz = 0;
  for (int i = 0; i < num; i++) {
    STqTopic* pTopic = taosArrayGet(pConsumer->topics, i);
    sz += tqGetTopicHandleSize(pTopic);
  }
  return sz;
}

static FORCE_INLINE int32_t tEncodeSTqTopic(void** buf, const STqTopic* pTopic) {
  int32_t tlen = 0;
  tlen += taosEncodeString(buf, pTopic->topicName);
  /*tlen += taosEncodeString(buf, pTopic->sql);*/
  /*tlen += taosEncodeString(buf, pTopic->physicalPlan);*/
  tlen += taosEncodeString(buf, pTopic->qmsg);
L
Liu Jicong 已提交
243 244 245
  /*tlen += taosEncodeFixedI64(buf, pTopic->persistedOffset);*/
  /*tlen += taosEncodeFixedI64(buf, pTopic->committedOffset);*/
  /*tlen += taosEncodeFixedI64(buf, pTopic->currentOffset);*/
L
Liu Jicong 已提交
246 247 248 249 250 251 252 253
  return tlen;
}

static FORCE_INLINE const void* tDecodeSTqTopic(const void* buf, STqTopic* pTopic) {
  buf = taosDecodeStringTo(buf, pTopic->topicName);
  /*buf = taosDecodeString(buf, &pTopic->sql);*/
  /*buf = taosDecodeString(buf, &pTopic->physicalPlan);*/
  buf = taosDecodeString(buf, &pTopic->qmsg);
L
Liu Jicong 已提交
254 255 256
  /*buf = taosDecodeFixedI64(buf, &pTopic->persistedOffset);*/
  /*buf = taosDecodeFixedI64(buf, &pTopic->committedOffset);*/
  /*buf = taosDecodeFixedI64(buf, &pTopic->currentOffset);*/
L
Liu Jicong 已提交
257 258 259 260 261 262 263 264
  return buf;
}

static FORCE_INLINE int32_t tEncodeSTqConsumer(void** buf, const STqConsumer* pConsumer) {
  int32_t sz;

  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);
L
Liu Jicong 已提交
265
  tlen += taosEncodeFixedI32(buf, pConsumer->epoch);
L
Liu Jicong 已提交
266 267 268 269 270 271 272 273
  tlen += taosEncodeString(buf, pConsumer->cgroup);
  sz = taosArrayGetSize(pConsumer->topics);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    STqTopic* pTopic = taosArrayGet(pConsumer->topics, i);
    tlen += tEncodeSTqTopic(buf, pTopic);
  }
  return tlen;
L
Liu Jicong 已提交
274 275
}

L
Liu Jicong 已提交
276 277 278 279
static FORCE_INLINE const void* tDecodeSTqConsumer(const void* buf, STqConsumer* pConsumer) {
  int32_t sz;

  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);
L
Liu Jicong 已提交
280
  buf = taosDecodeFixedI32(buf, &pConsumer->epoch);
L
Liu Jicong 已提交
281 282 283 284 285 286 287 288 289 290 291 292 293 294 295
  buf = taosDecodeStringTo(buf, pConsumer->cgroup);
  buf = taosDecodeFixedI32(buf, &sz);
  pConsumer->topics = taosArrayInit(sz, sizeof(STqTopic));
  if (pConsumer->topics == NULL) return NULL;
  for (int32_t i = 0; i < sz; i++) {
    STqTopic pTopic;
    buf = tDecodeSTqTopic(buf, &pTopic);
    taosArrayPush(pConsumer->topics, &pTopic);
  }
  return buf;
}

int tqSerializeConsumer(const STqConsumer* pConsumer, STqSerializedHead** ppHead) {
  int32_t sz = tEncodeSTqConsumer(NULL, pConsumer);

L
Liu Jicong 已提交
296
  if (sz > (*ppHead)->ssize) {
wafwerar's avatar
wafwerar 已提交
297
    void* tmpPtr = taosMemoryRealloc(*ppHead, sizeof(STqSerializedHead) + sz);
L
Liu Jicong 已提交
298
    if (tmpPtr == NULL) {
wafwerar's avatar
wafwerar 已提交
299
      taosMemoryFree(*ppHead);
L
Liu Jicong 已提交
300
      terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
L
Liu Jicong 已提交
301 302 303 304 305 306 307
      return -1;
    }
    *ppHead = tmpPtr;
    (*ppHead)->ssize = sz;
  }

  void* ptr = (*ppHead)->content;
L
Liu Jicong 已提交
308 309
  void* abuf = ptr;
  tEncodeSTqConsumer(&abuf, pConsumer);
L
Liu Jicong 已提交
310

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

L
Liu Jicong 已提交
314 315
int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsumer** ppConsumer) {
  const void* str = pHead->content;
wafwerar's avatar
wafwerar 已提交
316
  *ppConsumer = taosMemoryCalloc(1, sizeof(STqConsumer));
L
Liu Jicong 已提交
317 318 319 320 321 322 323 324 325 326
  if (*ppConsumer == NULL) {
    terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
    return -1;
  }
  if (tDecodeSTqConsumer(str, *ppConsumer) == NULL) {
    terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
    return -1;
  }
  STqConsumer* pConsumer = *ppConsumer;
  int32_t      sz = taosArrayGetSize(pConsumer->topics);
L
Liu Jicong 已提交
327
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
328 329 330 331 332
    STqTopic* pTopic = taosArrayGet(pConsumer->topics, i);
    pTopic->pReadhandle = walOpenReadHandle(pTq->pWal);
    if (pTopic->pReadhandle == NULL) {
      ASSERT(false);
    }
L
Liu Jicong 已提交
333 334
    for (int j = 0; j < TQ_BUFFER_SIZE; j++) {
      pTopic->buffer.output[j].status = 0;
L
Liu Jicong 已提交
335
      STqReadHandle* pReadHandle = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
L
Liu Jicong 已提交
336 337
      SReadHandle    handle = {
             .reader = pReadHandle,
L
Liu Jicong 已提交
338
             .meta = pTq->pVnode->pMeta,
L
Liu Jicong 已提交
339
      };
L
Liu Jicong 已提交
340 341
      pTopic->buffer.output[j].pReadHandle = pReadHandle;
      pTopic->buffer.output[j].task = qCreateStreamExecTaskInfo(pTopic->qmsg, &handle);
L
Liu Jicong 已提交
342
    }
L
Liu Jicong 已提交
343
  }
L
Liu Jicong 已提交
344 345

  return 0;
L
Liu Jicong 已提交
346
}
L
Liu Jicong 已提交
347

L
fix  
Liu Jicong 已提交
348
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
L
Liu Jicong 已提交
349 350 351 352 353
  SMqPollReq* pReq = pMsg->pCont;
  int64_t     consumerId = pReq->consumerId;
  int64_t     waitTime = pReq->waitTime;
  int32_t     reqEpoch = pReq->epoch;
  int64_t     fetchOffset;
354

L
Liu Jicong 已提交
355
  // get offset to fetch message
L
Liu Jicong 已提交
356
  if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__EARLIEAST) {
L
Liu Jicong 已提交
357
    fetchOffset = walGetFirstVer(pTq->pWal);
L
Liu Jicong 已提交
358 359 360 361 362 363
  } else if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__LATEST) {
    fetchOffset = walGetLastVer(pTq->pWal);
  } else {
    fetchOffset = pReq->currentOffset + 1;
  }

H
Hongze Cheng 已提交
364 365
  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 已提交
366

L
Liu Jicong 已提交
367
  STqExec* pExec = taosHashGet(pTq->execs, pReq->subKey, strlen(pReq->subKey));
L
Liu Jicong 已提交
368
  ASSERT(pExec);
L
Liu Jicong 已提交
369

L
Liu Jicong 已提交
370
  int32_t consumerEpoch = atomic_load_32(&pExec->epoch);
L
Liu Jicong 已提交
371
  while (consumerEpoch < reqEpoch) {
L
Liu Jicong 已提交
372
    consumerEpoch = atomic_val_compare_exchange_32(&pExec->epoch, consumerEpoch, reqEpoch);
L
Liu Jicong 已提交
373 374
  }

375 376 377 378 379 380 381
  SWalHead* pHeadWithCkSum = taosMemoryMalloc(sizeof(SWalHead) + 2048);
  if (pHeadWithCkSum == NULL) {
    return -1;
  }

  walSetReaderCapacity(pExec->pWalReader, 2048);

L
Liu Jicong 已提交
382
  SMqDataBlkRsp rsp = {0};
L
Liu Jicong 已提交
383
  rsp.reqOffset = pReq->currentOffset;
L
Liu Jicong 已提交
384
  rsp.withSchema = pExec->withSchema;
L
Liu Jicong 已提交
385
  rsp.blockData = taosArrayInit(0, sizeof(void*));
L
Liu Jicong 已提交
386
  rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
L
Liu Jicong 已提交
387
  rsp.blockSchema = taosArrayInit(0, sizeof(void*));
388 389

  while (1) {
L
Liu Jicong 已提交
390
    consumerEpoch = atomic_load_32(&pExec->epoch);
L
Liu Jicong 已提交
391
    if (consumerEpoch > reqEpoch) {
H
Hongze Cheng 已提交
392 393
      tqDebug("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 已提交
394 395
      break;
    }
L
Liu Jicong 已提交
396

397 398 399
    taosThreadMutexLock(&pExec->pWalReader->mutex);

    if (walFetchHead(pExec->pWalReader, fetchOffset, pHeadWithCkSum) < 0) {
H
Hongze Cheng 已提交
400 401
      tqDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, no more log to return", consumerId, pReq->epoch,
              TD_VID(pTq->pVnode), fetchOffset);
402 403 404 405 406 407 408 409 410 411 412 413 414 415 416
      taosThreadMutexUnlock(&pExec->pWalReader->mutex);
      break;
    }

    if (pHeadWithCkSum->head.msgType != TDMT_VND_SUBMIT) {
      walSkipFetchBody(pExec->pWalReader, pHeadWithCkSum);
    } else {
      walFetchBody(pExec->pWalReader, &pHeadWithCkSum);
    }

    SWalReadHead* pHead = &pHeadWithCkSum->head;

    taosThreadMutexUnlock(&pExec->pWalReader->mutex);

#if 0
L
fix  
Liu Jicong 已提交
417
    SWalReadHead* pHead;
L
Liu Jicong 已提交
418
    if (walReadWithHandle_s(pExec->pWalReader, fetchOffset, &pHead) < 0) {
L
Liu Jicong 已提交
419 420
      // TODO: no more log, set timer to wait blocking time
      // if data inserted during waiting, launch query and
L
Liu Jicong 已提交
421
      // response to user
H
Hongze Cheng 已提交
422
      tqDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, no more log to return", consumerId, pReq->epoch,
H
refact  
Hongze Cheng 已提交
423
             TD_VID(pTq->pVnode), fetchOffset);
L
Liu Jicong 已提交
424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445

#if 0
      // add to pushMgr
      taosWLockLatch(&pExec->pushHandle.lock);

      pExec->pushHandle.consumerId = consumerId;
      pExec->pushHandle.epoch = reqEpoch;
      pExec->pushHandle.reqOffset = rsp.reqOffset;
      pExec->pushHandle.skipLogNum = rsp.skipLogNum;
      pExec->pushHandle.handle = pMsg;

      taosWUnLockLatch(&pExec->pushHandle.lock);

      // TODO add timer

      // TODO: the pointer will always be valid?
      taosHashPut(pTq->pushMgr, &consumerId, sizeof(int64_t), &pExec, sizeof(void*));
      taosArrayDestroy(rsp.blockData);
      taosArrayDestroy(rsp.blockDataLen);
      return 0;
#endif

446 447 448
    break;
  }
#endif
L
Liu Jicong 已提交
449

H
Hongze Cheng 已提交
450 451
    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 已提交
452

L
fix  
Liu Jicong 已提交
453 454
    if (pHead->msgType == TDMT_VND_SUBMIT) {
      SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
455
      // table subscribe
L
Liu Jicong 已提交
456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486
      if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
        qTaskInfo_t task = pExec->task[workerId];
        ASSERT(task);
        qSetStreamInput(task, pCont, STREAM_DATA_TYPE_SUBMIT_BLOCK);
        while (1) {
          SSDataBlock* pDataBlock = NULL;
          uint64_t     ts = 0;
          if (qExecTask(task, &pDataBlock, &ts) < 0) {
            ASSERT(0);
          }
          if (pDataBlock == NULL) break;

          ASSERT(pDataBlock->info.rows != 0);
          ASSERT(pDataBlock->info.numOfCols != 0);

          int32_t            dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pDataBlock);
          void*              buf = taosMemoryCalloc(1, dataStrLen);
          SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
          pRetrieve->useconds = ts;
          pRetrieve->precision = TSDB_DEFAULT_PRECISION;
          pRetrieve->compressed = 0;
          pRetrieve->completed = 1;
          pRetrieve->numOfRows = htonl(pDataBlock->info.rows);

          // TODO enable compress
          int32_t actualLen = 0;
          blockCompressEncode(pDataBlock, pRetrieve->data, &actualLen, pDataBlock->info.numOfCols, false);
          actualLen += sizeof(SRetrieveTableRsp);
          ASSERT(actualLen <= dataStrLen);
          taosArrayPush(rsp.blockDataLen, &actualLen);
          taosArrayPush(rsp.blockData, &buf);
L
Liu Jicong 已提交
487 488 489 490 491 492

          if (pExec->withSchema) {
            SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
            taosArrayPush(rsp.blockSchema, &pSW);
          }

L
Liu Jicong 已提交
493
          rsp.blockNum++;
494
        }
495
        // db subscribe
L
Liu Jicong 已提交
496
      } else if (pExec->subType == TOPIC_SUB_TYPE__DB) {
L
Liu Jicong 已提交
497
        STqReadHandle* pReader = pExec->pExecReader[workerId];
L
Liu Jicong 已提交
498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520
        tqReadHandleSetMsg(pReader, pCont, 0);
        while (tqNextDataBlock(pReader)) {
          SSDataBlock block = {0};
          if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.rows,
                                  &block.info.numOfCols) < 0) {
            ASSERT(0);
          }
          int32_t            dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(&block);
          void*              buf = taosMemoryCalloc(1, dataStrLen);
          SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
          /*pRetrieve->useconds = 0;*/
          pRetrieve->precision = TSDB_DEFAULT_PRECISION;
          pRetrieve->compressed = 0;
          pRetrieve->completed = 1;
          pRetrieve->numOfRows = htonl(block.info.rows);

          // TODO enable compress
          int32_t actualLen = 0;
          blockCompressEncode(&block, pRetrieve->data, &actualLen, block.info.numOfCols, false);
          actualLen += sizeof(SRetrieveTableRsp);
          ASSERT(actualLen <= dataStrLen);
          taosArrayPush(rsp.blockDataLen, &actualLen);
          taosArrayPush(rsp.blockData, &buf);
L
Liu Jicong 已提交
521 522 523 524

          SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
          taosArrayPush(rsp.blockSchema, &pSW);

L
Liu Jicong 已提交
525 526 527 528
          rsp.blockNum++;
        }
      } else {
        ASSERT(0);
L
fix  
Liu Jicong 已提交
529
      }
L
Liu Jicong 已提交
530
    }
531

L
Liu Jicong 已提交
532 533
    // TODO batch optimization:
    // TODO continue scan until meeting batch requirement
L
Liu Jicong 已提交
534 535 536 537
    if (rsp.blockNum != 0) break;
    rsp.skipLogNum++;
    fetchOffset++;
  }
538

L
Liu Jicong 已提交
539 540
  ASSERT(taosArrayGetSize(rsp.blockData) == rsp.blockNum);
  ASSERT(taosArrayGetSize(rsp.blockDataLen) == rsp.blockNum);
L
fix  
Liu Jicong 已提交
541

L
Liu Jicong 已提交
542 543 544 545
  if (rsp.blockNum != 0)
    rsp.rspOffset = fetchOffset;
  else
    rsp.rspOffset = fetchOffset - 1;
546

L
Liu Jicong 已提交
547
  int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqDataBlkRsp(NULL, &rsp);
548 549 550 551 552
  void*   buf = rpcMallocCont(tlen);
  if (buf == NULL) {
    pMsg->code = -1;
    return -1;
  }
L
Liu Jicong 已提交
553

L
Liu Jicong 已提交
554 555
  ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
  ((SMqRspHead*)buf)->epoch = pReq->epoch;
L
Liu Jicong 已提交
556
  ((SMqRspHead*)buf)->consumerId = consumerId;
557

L
Liu Jicong 已提交
558
  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
L
Liu Jicong 已提交
559
  tEncodeSMqDataBlkRsp(&abuf, &rsp);
560 561 562
  pMsg->pCont = buf;
  pMsg->contLen = tlen;
  pMsg->code = 0;
S
shm  
Shengliang Guan 已提交
563
  tmsgSendRsp(pMsg);
L
Liu Jicong 已提交
564

H
Hongze Cheng 已提交
565 566
  tqDebug("vg %d offset %ld from consumer %ld (epoch %d) send rsp, block num: %d, reqOffset: %ld, rspOffset: %ld",
          TD_VID(pTq->pVnode), fetchOffset, consumerId, pReq->epoch, rsp.blockNum, rsp.reqOffset, rsp.rspOffset);
L
Liu Jicong 已提交
567 568

  // TODO destroy
L
Liu Jicong 已提交
569 570
  taosArrayDestroy(rsp.blockData);
  taosArrayDestroy(rsp.blockDataLen);
L
Liu Jicong 已提交
571 572
  taosArrayDestroyP(rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);

573 574
  return 0;
}
L
Liu Jicong 已提交
575

L
Liu Jicong 已提交
576
#if 0
L
Liu Jicong 已提交
577 578 579 580 581 582 583 584
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
  SMqPollReq* pReq = pMsg->pCont;
  int64_t     consumerId = pReq->consumerId;
  int64_t     fetchOffset;
  int64_t     blockingTime = pReq->blockingTime;
  int32_t     reqEpoch = pReq->epoch;

  if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__EARLIEAST) {
L
Liu Jicong 已提交
585
    fetchOffset = walGetFirstVer(pTq->pWal);
L
Liu Jicong 已提交
586 587 588 589 590 591
  } else if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__LATEST) {
    fetchOffset = walGetLastVer(pTq->pWal);
  } else {
    fetchOffset = pReq->currentOffset + 1;
  }

H
Hongze Cheng 已提交
592
  tqDebug("tmq poll: consumer %ld (epoch %d) recv poll req in vg %d, req %ld %ld", consumerId, pReq->epoch,
H
refact  
Hongze Cheng 已提交
593
         TD_VID(pTq->pVnode), pReq->currentOffset, fetchOffset);
L
Liu Jicong 已提交
594 595 596 597 598 599

  SMqPollRspV2 rspV2 = {0};
  rspV2.dataLen = 0;

  STqConsumer* pConsumer = tqHandleGet(pTq->tqMeta, consumerId);
  if (pConsumer == NULL) {
H
refact  
Hongze Cheng 已提交
600
    vWarn("tmq poll: consumer %ld (epoch %d) not found in vg %d", consumerId, pReq->epoch, TD_VID(pTq->pVnode));
L
Liu Jicong 已提交
601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625
    pMsg->pCont = NULL;
    pMsg->contLen = 0;
    pMsg->code = -1;
    tmsgSendRsp(pMsg);
    return 0;
  }

  int32_t consumerEpoch = atomic_load_32(&pConsumer->epoch);
  while (consumerEpoch < reqEpoch) {
    consumerEpoch = atomic_val_compare_exchange_32(&pConsumer->epoch, consumerEpoch, reqEpoch);
  }

  STqTopic* pTopic = NULL;
  int32_t   topicSz = taosArrayGetSize(pConsumer->topics);
  for (int32_t i = 0; i < topicSz; i++) {
    STqTopic* topic = taosArrayGet(pConsumer->topics, i);
    // TODO race condition
    ASSERT(pConsumer->consumerId == consumerId);
    if (strcmp(topic->topicName, pReq->topic) == 0) {
      pTopic = topic;
      break;
    }
  }
  if (pTopic == NULL) {
    vWarn("tmq poll: consumer %ld (epoch %d) topic %s not found in vg %d", consumerId, pReq->epoch, pReq->topic,
H
refact  
Hongze Cheng 已提交
626
          TD_VID(pTq->pVnode));
L
Liu Jicong 已提交
627 628 629 630 631 632 633
    pMsg->pCont = NULL;
    pMsg->contLen = 0;
    pMsg->code = -1;
    tmsgSendRsp(pMsg);
    return 0;
  }

H
Hongze Cheng 已提交
634
  tqDebug("poll topic %s from consumer %ld (epoch %d) vg %d", pTopic->topicName, consumerId, pReq->epoch,
H
refact  
Hongze Cheng 已提交
635
         TD_VID(pTq->pVnode));
L
Liu Jicong 已提交
636 637 638 639 640 641 642 643 644

  rspV2.reqOffset = pReq->currentOffset;
  rspV2.skipLogNum = 0;

  while (1) {
    /*if (fetchOffset > walGetLastVer(pTq->pWal) || walReadWithHandle(pTopic->pReadhandle, fetchOffset) < 0) {*/
    // TODO
    consumerEpoch = atomic_load_32(&pConsumer->epoch);
    if (consumerEpoch > reqEpoch) {
H
Hongze Cheng 已提交
645
      tqDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, found new consumer epoch %d discard req epoch %d",
H
refact  
Hongze Cheng 已提交
646
             consumerId, pReq->epoch, TD_VID(pTq->pVnode), fetchOffset, consumerEpoch, reqEpoch);
L
Liu Jicong 已提交
647 648 649 650 651 652 653
      break;
    }
    SWalReadHead* pHead;
    if (walReadWithHandle_s(pTopic->pReadhandle, fetchOffset, &pHead) < 0) {
      // TODO: no more log, set timer to wait blocking time
      // if data inserted during waiting, launch query and
      // response to user
H
Hongze Cheng 已提交
654
      tqDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, no more log to return", consumerId, pReq->epoch,
H
refact  
Hongze Cheng 已提交
655
             TD_VID(pTq->pVnode), fetchOffset);
L
Liu Jicong 已提交
656 657
      break;
    }
H
Hongze Cheng 已提交
658
    tqDebug("tmq poll: consumer %ld (epoch %d) iter log, vg %d offset %ld msgType %d", consumerId, pReq->epoch,
H
refact  
Hongze Cheng 已提交
659
           TD_VID(pTq->pVnode), fetchOffset, pHead->msgType);
L
Liu Jicong 已提交
660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682
    /*int8_t pos = fetchOffset % TQ_BUFFER_SIZE;*/
    /*pHead = pTopic->pReadhandle->pHead;*/
    if (pHead->msgType == TDMT_VND_SUBMIT) {
      SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
      qTaskInfo_t task = pTopic->buffer.output[workerId].task;
      ASSERT(task);
      qSetStreamInput(task, pCont, STREAM_DATA_TYPE_SUBMIT_BLOCK);
      SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
      while (1) {
        SSDataBlock* pDataBlock = NULL;
        uint64_t     ts;
        if (qExecTask(task, &pDataBlock, &ts) < 0) {
          ASSERT(false);
        }
        if (pDataBlock == NULL) {
          /*pos = fetchOffset % TQ_BUFFER_SIZE;*/
          break;
        }

        taosArrayPush(pRes, pDataBlock);
      }

      if (taosArrayGetSize(pRes) == 0) {
H
Hongze Cheng 已提交
683
        tqDebug("tmq poll: consumer %ld (epoch %d) iter log, vg %d skip log %ld since not wanted", consumerId,
H
refact  
Hongze Cheng 已提交
684
               pReq->epoch, TD_VID(pTq->pVnode), fetchOffset);
L
Liu Jicong 已提交
685 686 687 688 689 690 691 692
        fetchOffset++;
        rspV2.skipLogNum++;
        taosArrayDestroy(pRes);
        continue;
      }
      rspV2.rspOffset = fetchOffset;

      int32_t blockSz = taosArrayGetSize(pRes);
L
Liu Jicong 已提交
693
      int32_t dataBlockStrLen = 0;
L
Liu Jicong 已提交
694 695
      for (int32_t i = 0; i < blockSz; i++) {
        SSDataBlock* pBlock = taosArrayGet(pRes, i);
L
Liu Jicong 已提交
696
        dataBlockStrLen += sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
L
Liu Jicong 已提交
697 698
      }

L
Liu Jicong 已提交
699 700
      void* dataBlockBuf = taosMemoryMalloc(dataBlockStrLen);
      if (dataBlockBuf == NULL) {
L
Liu Jicong 已提交
701 702 703 704
        pMsg->code = -1;
        taosMemoryFree(pHead);
      }

L
Liu Jicong 已提交
705
      rspV2.blockData = dataBlockBuf;
L
Liu Jicong 已提交
706 707

      int32_t pos;
L
Liu Jicong 已提交
708
      rspV2.blockPos = taosArrayInit(blockSz, sizeof(int32_t));
L
Liu Jicong 已提交
709 710
      for (int32_t i = 0; i < blockSz; i++) {
        pos = 0;
L
Liu Jicong 已提交
711 712 713 714 715 716 717 718
        SSDataBlock*       pBlock = taosArrayGet(pRes, i);
        SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)dataBlockBuf;
        pRetrieve->useconds = 0;
        pRetrieve->precision = 0;
        pRetrieve->compressed = 0;
        pRetrieve->completed = 1;
        pRetrieve->numOfRows = htonl(pBlock->info.rows);
        blockCompressEncode(pBlock, pRetrieve->data, &pos, pBlock->info.numOfCols, false);
L
Liu Jicong 已提交
719
        taosArrayPush(rspV2.blockPos, &rspV2.dataLen);
L
Liu Jicong 已提交
720 721 722 723 724

        int32_t totLen = sizeof(SRetrieveTableRsp) + pos;
        pRetrieve->compLen = htonl(totLen);
        rspV2.dataLen += totLen;
        dataBlockBuf = POINTER_SHIFT(dataBlockBuf, totLen);
L
Liu Jicong 已提交
725
      }
L
Liu Jicong 已提交
726
      ASSERT(POINTER_DISTANCE(dataBlockBuf, rspV2.blockData) <= dataBlockStrLen);
L
Liu Jicong 已提交
727 728 729 730 731 732 733 734 735 736 737 738 739 740 741

      int32_t msgLen = sizeof(SMqRspHead) + tEncodeSMqPollRspV2(NULL, &rspV2);
      void*   buf = rpcMallocCont(msgLen);

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

      void* msgBodyBuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
      tEncodeSMqPollRspV2(&msgBodyBuf, &rspV2);

      /*rsp.pBlockData = pRes;*/

      /*taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);*/
      pMsg->pCont = buf;
L
Liu Jicong 已提交
742
      pMsg->contLen = msgLen;
L
Liu Jicong 已提交
743
      pMsg->code = 0;
H
Hongze Cheng 已提交
744
      tqDebug("vg %d offset %ld msgType %d from consumer %ld (epoch %d) actual rsp", TD_VID(pTq->pVnode), fetchOffset,
L
Liu Jicong 已提交
745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777
             pHead->msgType, consumerId, pReq->epoch);
      tmsgSendRsp(pMsg);
      taosMemoryFree(pHead);
      return 0;
    } else {
      taosMemoryFree(pHead);
      fetchOffset++;
      rspV2.skipLogNum++;
    }
  }

  /*if (blockingTime != 0) {*/
  /*tqAddClientPusher(pTq->tqPushMgr, pMsg, consumerId, blockingTime);*/
  /*} else {*/

  rspV2.rspOffset = fetchOffset - 1;

  int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqPollRspV2(NULL, &rspV2);
  void*   buf = rpcMallocCont(tlen);
  if (buf == NULL) {
    pMsg->code = -1;
    return -1;
  }
  ((SMqRspHead*)buf)->mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
  ((SMqRspHead*)buf)->epoch = pReq->epoch;
  ((SMqRspHead*)buf)->consumerId = consumerId;

  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
  tEncodeSMqPollRspV2(&abuf, &rspV2);
  pMsg->pCont = buf;
  pMsg->contLen = tlen;
  pMsg->code = 0;
  tmsgSendRsp(pMsg);
H
Hongze Cheng 已提交
778
  tqDebug("vg %d offset %ld from consumer %ld (epoch %d) not rsp", TD_VID(pTq->pVnode), fetchOffset, consumerId,
L
Liu Jicong 已提交
779 780 781 782 783
         pReq->epoch);
  /*}*/

  return 0;
}
L
Liu Jicong 已提交
784 785 786 787
#endif

// TODO: persist meta into tdb
int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
L
Liu Jicong 已提交
788
  SMqRebVgReq req = {0};
L
Liu Jicong 已提交
789 790
  tDecodeSMqRebVgReq(msg, &req);
  // todo lock
L
Liu Jicong 已提交
791
  STqExec* pExec = taosHashGet(pTq->execs, req.subKey, strlen(req.subKey));
L
Liu Jicong 已提交
792 793 794 795 796 797 798 799 800 801
  if (pExec == NULL) {
    ASSERT(req.oldConsumerId == -1);
    ASSERT(req.newConsumerId != -1);
    STqExec exec = {0};
    pExec = &exec;
    /*taosInitRWLatch(&pExec->lock);*/

    memcpy(pExec->subKey, req.subKey, TSDB_SUBSCRIBE_KEY_LEN);
    pExec->consumerId = req.newConsumerId;
    pExec->epoch = -1;
L
Liu Jicong 已提交
802 803 804 805 806 807

    pExec->subType = req.subType;
    pExec->withTbName = req.withTbName;
    pExec->withSchema = req.withSchema;
    pExec->withTag = req.withTag;

L
Liu Jicong 已提交
808 809
    pExec->qmsg = req.qmsg;
    req.qmsg = NULL;
L
Liu Jicong 已提交
810 811

    pExec->pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
L
Liu Jicong 已提交
812 813
    for (int32_t i = 0; i < 5; i++) {
      pExec->pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
L
Liu Jicong 已提交
814 815 816 817 818 819 820 821 822 823
      if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
        SReadHandle handle = {
            .reader = pExec->pExecReader[i],
            .meta = pTq->pVnode->pMeta,
        };
        pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);
        ASSERT(pExec->task[i]);
      } else {
        pExec->task[i] = NULL;
      }
L
Liu Jicong 已提交
824
    }
L
Liu Jicong 已提交
825
    taosHashPut(pTq->execs, req.subKey, strlen(req.subKey), pExec, sizeof(STqExec));
L
Liu Jicong 已提交
826 827
    return 0;
  } else {
L
Liu Jicong 已提交
828 829 830 831 832 833 834 835 836 837 838 839 840 841
    /*if (req.newConsumerId != -1) {*/
    /*taosWLockLatch(&pExec->lock);*/
    ASSERT(pExec->consumerId == req.oldConsumerId);
    // TODO handle qmsg and exec modification
    atomic_store_32(&pExec->epoch, -1);
    atomic_store_64(&pExec->consumerId, req.newConsumerId);
    atomic_add_fetch_32(&pExec->epoch, 1);
    /*taosWUnLockLatch(&pExec->lock);*/
    return 0;
    /*} else {*/
    // TODO
    /*taosHashRemove(pTq->tqMetaNew, req.subKey, strlen(req.subKey));*/
    /*return 0;*/
    /*}*/
L
Liu Jicong 已提交
842 843
  }
}
844

L
Liu Jicong 已提交
845
int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int32_t parallel) {
L
Liu Jicong 已提交
846 847 848
  if (pTask->execType == TASK_EXEC__NONE) return 0;

  pTask->exec.numOfRunners = parallel;
L
Liu Jicong 已提交
849
  pTask->exec.runners = taosMemoryCalloc(parallel, sizeof(SStreamRunner));
L
Liu Jicong 已提交
850 851 852
  if (pTask->exec.runners == NULL) {
    return -1;
  }
L
Liu Jicong 已提交
853
  for (int32_t i = 0; i < parallel; i++) {
L
Liu Jicong 已提交
854
    STqReadHandle* pStreamReader = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
L
Liu Jicong 已提交
855
    SReadHandle    handle = {
L
Liu Jicong 已提交
856 857
           .reader = pStreamReader,
           .meta = pTq->pVnode->pMeta,
L
Liu Jicong 已提交
858
    };
L
Liu Jicong 已提交
859
    pTask->exec.runners[i].inputHandle = pStreamReader;
L
Liu Jicong 已提交
860
    pTask->exec.runners[i].executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);
L
Liu Jicong 已提交
861
    ASSERT(pTask->exec.runners[i].executor);
L
Liu Jicong 已提交
862 863 864 865
  }
  return 0;
}

L
Liu Jicong 已提交
866
int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
L
Liu Jicong 已提交
867
  SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
L
Liu Jicong 已提交
868 869 870 871 872
  if (pTask == NULL) {
    return -1;
  }
  SCoder decoder;
  tCoderInit(&decoder, TD_LITTLE_ENDIAN, (uint8_t*)msg, msgLen, TD_DECODER);
L
Liu Jicong 已提交
873 874 875
  if (tDecodeSStreamTask(&decoder, pTask) < 0) {
    ASSERT(0);
  }
L
Liu Jicong 已提交
876 877
  tCoderClear(&decoder);

L
Liu Jicong 已提交
878
  // exec
L
Liu Jicong 已提交
879 880 881
  if (tqExpandTask(pTq, pTask, 4) < 0) {
    ASSERT(0);
  }
L
Liu Jicong 已提交
882 883

  // sink
L
Liu Jicong 已提交
884
  pTask->ahandle = pTq->pVnode;
L
Liu Jicong 已提交
885 886 887
  if (pTask->sinkType == TASK_SINK__SMA) {
    pTask->smaSink.smaHandle = smaHandleRes;
  }
L
Liu Jicong 已提交
888

L
Liu Jicong 已提交
889 890 891 892
  taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), pTask, sizeof(SStreamTask));

  return 0;
}
L
Liu Jicong 已提交
893

L
Liu Jicong 已提交
894
int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen, int32_t workerId) {
L
Liu Jicong 已提交
895 896 897 898 899 900 901
  void* pIter = NULL;

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

L
Liu Jicong 已提交
902
    if (streamExecTask(pTask, &pTq->pVnode->msgCb, data, STREAM_DATA_TYPE_SUBMIT_BLOCK, workerId) < 0) {
L
Liu Jicong 已提交
903
      // TODO
L
Liu Jicong 已提交
904 905 906 907 908
    }
  }
  return 0;
}

L
Liu Jicong 已提交
909
int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId) {
L
Liu Jicong 已提交
910
  SStreamTaskExecReq req;
L
Liu Jicong 已提交
911
  tDecodeSStreamTaskExecReq(msg, &req);
L
Liu Jicong 已提交
912

L
Liu Jicong 已提交
913 914 915
  int32_t taskId = req.taskId;
  ASSERT(taskId);

L
Liu Jicong 已提交
916
  SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
L
Liu Jicong 已提交
917
  ASSERT(pTask);
L
Liu Jicong 已提交
918

L
Liu Jicong 已提交
919
  if (streamExecTask(pTask, &pTq->pVnode->msgCb, req.data, STREAM_DATA_TYPE_SSDATA_BLOCK, workerId) < 0) {
L
Liu Jicong 已提交
920
    // TODO
L
Liu Jicong 已提交
921
  }
L
Liu Jicong 已提交
922 923
  return 0;
}