tqPush.c 15.0 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14
/*
 * 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 已提交
15 16

#include "tq.h"
17
#include "vnd.h"
L
Liu Jicong 已提交
18

L
Liu Jicong 已提交
19
#if 0
L
Liu Jicong 已提交
20 21 22 23 24
void tqTmrRspFunc(void* param, void* tmrId) {
  STqHandle* pHandle = (STqHandle*)param;
  atomic_store_8(&pHandle->pushHandle.tmrStopped, 1);
}

L
Liu Jicong 已提交
25
static int32_t tqLoopExecFromQueue(STQ* pTq, STqHandle* pHandle, SStreamDataSubmit** ppSubmit, SMqDataRsp* pRsp) {
26 27
  SStreamDataSubmit* pSubmit = *ppSubmit;
  while (pSubmit != NULL) {
L
Liu Jicong 已提交
28
    if (tqLogScanExec(pTq, &pHandle->execHandle, pSubmit->data, pRsp, 0) < 0) {
29 30 31
    }
    // update processed
    atomic_store_64(&pHandle->pushHandle.processedVer, pSubmit->ver);
L
Liu Jicong 已提交
32
    streamQueueProcessSuccess(&pHandle->pushHandle.inputQ);
33
    streamDataSubmitDestroy(pSubmit);
34 35 36 37
    if (pRsp->blockNum > 0) {
      *ppSubmit = pSubmit;
      return 0;
    } else {
L
Liu Jicong 已提交
38
      pSubmit = streamQueueNextItem(&pHandle->pushHandle.inputQ);
39 40 41 42 43 44
    }
  }
  *ppSubmit = pSubmit;
  return -1;
}

L
Liu Jicong 已提交
45
int32_t tqExecFromInputQ(STQ* pTq, STqHandle* pHandle) {
L
Liu Jicong 已提交
46
  SMqDataRsp rsp = {0};
L
Liu Jicong 已提交
47
  // 1. guard and set status executing
L
Liu Jicong 已提交
48 49 50
  int8_t execStatus = atomic_val_compare_exchange_8(&pHandle->pushHandle.execStatus, TASK_EXEC_STATUS__IDLE,
                                                    TASK_EXEC_STATUS__EXECUTING);
  if (execStatus == TASK_EXEC_STATUS__IDLE) {
L
Liu Jicong 已提交
51 52 53 54
    SStreamDataSubmit* pSubmit = NULL;
    // 2. check processedVer
    // 2.1. if not missed, get msg from queue
    // 2.2. if missed, scan wal
L
Liu Jicong 已提交
55
    pSubmit = streamQueueNextItem(&pHandle->pushHandle.inputQ);
L
Liu Jicong 已提交
56 57 58 59
    while (pHandle->pushHandle.processedVer <= pSubmit->ver) {
      // read from wal
    }
    while (pHandle->pushHandle.processedVer > pSubmit->ver + 1) {
L
Liu Jicong 已提交
60
      streamQueueProcessSuccess(&pHandle->pushHandle.inputQ);
61
      streamDataSubmitDestroy(pSubmit);
L
Liu Jicong 已提交
62
      pSubmit = streamQueueNextItem(&pHandle->pushHandle.inputQ);
L
Liu Jicong 已提交
63 64 65 66
      if (pSubmit == NULL) break;
    }
    // 3. exec, after each success, update processed ver
    // first run
67 68
    if (tqLoopExecFromQueue(pTq, pHandle, &pSubmit, &rsp) == 0) {
      goto SEND_RSP;
L
Liu Jicong 已提交
69 70
    }
    // set exec status closing
L
Liu Jicong 已提交
71
    atomic_store_8(&pHandle->pushHandle.execStatus, TASK_EXEC_STATUS__CLOSING);
L
Liu Jicong 已提交
72
    // second run
73 74
    if (tqLoopExecFromQueue(pTq, pHandle, &pSubmit, &rsp) == 0) {
      goto SEND_RSP;
L
Liu Jicong 已提交
75 76
    }
    // set exec status idle
L
Liu Jicong 已提交
77
    atomic_store_8(&pHandle->pushHandle.execStatus, TASK_EXEC_STATUS__IDLE);
L
Liu Jicong 已提交
78 79
  }
SEND_RSP:
L
Liu Jicong 已提交
80 81
  // 4. if get result
  // 4.1 set exec input status blocked and exec status idle
L
Liu Jicong 已提交
82
  atomic_store_8(&pHandle->pushHandle.execStatus, TASK_EXEC_STATUS__IDLE);
L
Liu Jicong 已提交
83
  // 4.2 rpc send
L
Liu Jicong 已提交
84 85 86 87
  rsp.rspOffset = pHandle->pushHandle.processedVer;
  /*if (tqSendPollRsp(pTq, pMsg, pReq, &rsp) < 0) {*/
  /*return -1;*/
  /*}*/
L
Liu Jicong 已提交
88
  // 4.3 clear rpc info
L
Liu Jicong 已提交
89
  memset(&pHandle->pushHandle.rpcInfo, 0, sizeof(SRpcHandleInfo));
L
Liu Jicong 已提交
90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108
  return 0;
}

int32_t tqOpenPushHandle(STQ* pTq, STqHandle* pHandle) {
  memset(&pHandle->pushHandle, 0, sizeof(STqPushHandle));
  pHandle->pushHandle.inputQ.queue = taosOpenQueue();
  pHandle->pushHandle.inputQ.qall = taosAllocateQall();
  if (pHandle->pushHandle.inputQ.queue == NULL || pHandle->pushHandle.inputQ.qall == NULL) {
    if (pHandle->pushHandle.inputQ.queue) {
      taosCloseQueue(pHandle->pushHandle.inputQ.queue);
    }
    if (pHandle->pushHandle.inputQ.qall) {
      taosFreeQall(pHandle->pushHandle.inputQ.qall);
    }
    return -1;
  }
  return 0;
}

L
Liu Jicong 已提交
109 110
int32_t tqPreparePush(STQ* pTq, STqHandle* pHandle, int64_t reqId, const SRpcHandleInfo* pInfo, int64_t processedVer,
                      int64_t timeout) {
L
Liu Jicong 已提交
111 112 113 114
  memcpy(&pHandle->pushHandle.rpcInfo, pInfo, sizeof(SRpcHandleInfo));
  atomic_store_64(&pHandle->pushHandle.reqId, reqId);
  atomic_store_64(&pHandle->pushHandle.processedVer, processedVer);
  atomic_store_8(&pHandle->pushHandle.inputStatus, TASK_INPUT_STATUS__NORMAL);
L
Liu Jicong 已提交
115 116 117
  atomic_store_8(&pHandle->pushHandle.tmrStopped, 0);
  taosTmrReset(tqTmrRspFunc, (int32_t)timeout, pHandle, tqMgmt.timer, &pHandle->pushHandle.timerId);
  return 0;
L
Liu Jicong 已提交
118 119 120 121 122
}

int32_t tqEnqueue(STqHandle* pHandle, SStreamDataSubmit* pSubmit) {
  int8_t inputStatus = atomic_load_8(&pHandle->pushHandle.inputStatus);
  if (inputStatus == TASK_INPUT_STATUS__NORMAL) {
123
    SStreamDataSubmit* pSubmitClone = streamSubmitBlockClone(pSubmit);
L
Liu Jicong 已提交
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
    if (pSubmitClone == NULL) {
      return -1;
    }
    taosWriteQitem(pHandle->pushHandle.inputQ.queue, pSubmitClone);
    return 0;
  }
  return -1;
}

int32_t tqSendExecReq(STQ* pTq, STqHandle* pHandle) {
  //
  return 0;
}

int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver, SRpcHandleInfo handleInfo) {
  if (msgType != TDMT_VND_SUBMIT) return 0;
  void*       pIter = NULL;
  STqHandle*  pHandle = NULL;
  SSubmitReq* pReq = (SSubmitReq*)msg;
  int32_t     workerId = 4;
  int64_t     fetchOffset = ver;

  while (1) {
    pIter = taosHashIterate(pTq->pushMgr, pIter);
    if (pIter == NULL) break;
    pHandle = *(STqHandle**)pIter;

    taosWLockLatch(&pHandle->pushHandle.lock);

L
Liu Jicong 已提交
153
    SMqDataRsp rsp = {0};
L
Liu Jicong 已提交
154 155 156 157 158
    rsp.reqOffset = pHandle->pushHandle.reqOffset;
    rsp.blockData = taosArrayInit(0, sizeof(void*));
    rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));

    if (msgType == TDMT_VND_SUBMIT) {
L
Liu Jicong 已提交
159
      tqLogScanExec(pTq, &pHandle->execHandle, pReq, &rsp, workerId);
L
Liu Jicong 已提交
160
    } else {
L
Liu Jicong 已提交
161
      tqError("tq push unexpected msg type %d", msgType);
L
Liu Jicong 已提交
162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195
    }

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

    rsp.rspOffset = fetchOffset;

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

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

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

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

    memset(&pHandle->pushHandle.rpcInfo, 0, sizeof(SRpcHandleInfo));
    taosWUnLockLatch(&pHandle->pushHandle.lock);

196
    tqDebug("vgId:%d offset %" PRId64 " from consumer:%" PRId64 ", (epoch %d) send rsp, block num: %d, req:%" PRId64 ", rsp:%" PRId64,
L
Liu Jicong 已提交
197 198 199 200 201 202 203 204 205 206
            TD_VID(pTq->pVnode), fetchOffset, pHandle->pushHandle.consumerId, pHandle->pushHandle.epoch, rsp.blockNum,
            rsp.reqOffset, rsp.rspOffset);

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

  return 0;
}
L
Liu Jicong 已提交
207
#endif
L
Liu Jicong 已提交
208

209
typedef struct {
210
  void*   pKey;
211 212 213 214
  int64_t keyLen;
} SItem;

static void recordPushedEntry(SArray* cachedKey, void* pIter);
215
static void doRemovePushedEntry(SArray* pCachedKeys, STQ* pTq);
216 217

static void freeItem(void* param) {
X
Xiaoyu Wang 已提交
218
  SItem* p = (SItem*)param;
219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235
  taosMemoryFree(p->pKey);
}

static void doPushDataForEntry(void* pIter, STqExecHandle* pExec, STQ* pTq, int64_t ver, int32_t vgId, char* pData,
                               int32_t dataLen, SArray* pCachedKey) {
  STqPushEntry* pPushEntry = *(STqPushEntry**)pIter;

  SMqDataRsp* pRsp = pPushEntry->pDataRsp;
  if (pRsp->reqOffset.version >= ver) {
    tqDebug("vgId:%d, push entry req version %" PRId64 ", while push version %" PRId64 ", skip", vgId,
            pRsp->reqOffset.version, ver);
    return;
  }

  qTaskInfo_t pTaskInfo = pExec->task;

  // prepare scan mem data
236
  SPackedData submit = {.msgStr = pData, .msgLen = dataLen, .ver = ver};
237

238
  if (qStreamSetScanMemData(pTaskInfo, submit) != 0) {
239 240
    return;
  }
241
  qStreamSetOpen(pTaskInfo);
242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261
  // here start to scan submit block to extract the subscribed data
  int32_t totalRows = 0;

  while (1) {
    SSDataBlock* pDataBlock = NULL;
    uint64_t     ts = 0;
    if (qExecTask(pTaskInfo, &pDataBlock, &ts) < 0) {
      tqDebug("vgId:%d, tq exec error since %s", vgId, terrstr());
    }

    if (pDataBlock == NULL) {
      break;
    }

    tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);
    pRsp->blockNum++;
    totalRows += pDataBlock->info.rows;
  }

  tqDebug("vgId:%d, tq handle push, subkey:%s, block num:%d, rows:%d", vgId, pPushEntry->subKey, pRsp->blockNum,
X
Xiaoyu Wang 已提交
262
          totalRows);
263 264 265

  if (pRsp->blockNum > 0) {
    tqOffsetResetToLog(&pRsp->rspOffset, ver);
266
    tqPushDataRsp(pPushEntry, vgId);
267 268 269 270
    recordPushedEntry(pCachedKey, pIter);
  }
}

271
int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
272 273
  void*   pReq = POINTER_SHIFT(msg, sizeof(SSubmitReq2Msg));
  int32_t len = msgLen - sizeof(SSubmitReq2Msg);
H
Haojun Liao 已提交
274
  int32_t vgId = TD_VID(pTq->pVnode);
L
Liu Jicong 已提交
275

L
Liu Jicong 已提交
276 277
  if (msgType == TDMT_VND_SUBMIT) {
    // lock push mgr to avoid potential msg lost
278
    taosWLockLatch(&pTq->lock);
279

280 281 282
    int32_t numOfRegisteredPush = taosHashGetSize(pTq->pPushMgr);
    if (numOfRegisteredPush > 0) {
      tqDebug("vgId:%d tq push msg version:%" PRId64 " type:%s, head:%p, body:%p len:%d, numOfPushed consumers:%d",
X
Xiaoyu Wang 已提交
283
              vgId, ver, TMSG_INFO(msgType), msg, pReq, len, numOfRegisteredPush);
284

285
      void* data = taosMemoryMalloc(len);
L
Liu Jicong 已提交
286 287
      if (data == NULL) {
        terrno = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
288
        tqError("failed to copy data for stream since out of memory, vgId:%d", vgId);
289
        taosWUnLockLatch(&pTq->lock);
L
Liu Jicong 已提交
290 291
        return -1;
      }
292

L
Liu Jicong 已提交
293
      memcpy(data, pReq, len);
L
Liu Jicong 已提交
294

295 296 297
      SArray* cachedKey = taosArrayInit(0, sizeof(SItem));
      void*   pIter = NULL;

L
Liu Jicong 已提交
298 299
      while (1) {
        pIter = taosHashIterate(pTq->pPushMgr, pIter);
H
Haojun Liao 已提交
300 301 302 303
        if (pIter == NULL) {
          break;
        }

L
Liu Jicong 已提交
304 305 306 307
        STqPushEntry* pPushEntry = *(STqPushEntry**)pIter;

        STqHandle* pHandle = taosHashGet(pTq->pHandle, pPushEntry->subKey, strlen(pPushEntry->subKey));
        if (pHandle == NULL) {
X
Xiaoyu Wang 已提交
308 309
          tqDebug("vgId:%d, failed to find handle %s in pushing data to consumer, ignore", pTq->pVnode->config.vgId,
                  pPushEntry->subKey);
L
Liu Jicong 已提交
310 311
          continue;
        }
312

L
Liu Jicong 已提交
313
        STqExecHandle* pExec = &pHandle->execHandle;
314
        doPushDataForEntry(pIter, pExec, pTq, ver, vgId, data, len, cachedKey);
L
Liu Jicong 已提交
315
      }
H
Haojun Liao 已提交
316

317 318
      doRemovePushedEntry(cachedKey, pTq);
      taosArrayDestroyEx(cachedKey, freeItem);
L
Liu Jicong 已提交
319
      taosMemoryFree(data);
L
Liu Jicong 已提交
320
    }
321

L
Liu Jicong 已提交
322
    // unlock
323
    taosWUnLockLatch(&pTq->lock);
L
Liu Jicong 已提交
324 325
  }

326 327
  tqDebug("handle submit, restore:%d, size:%d", pTq->pVnode->restored, (int)taosHashGetSize(pTq->pStreamMeta->pTasks));

328
  // push data for stream processing:
329
  // 1. the vnode has already been restored.
330 331
  // 2. the vnode should be the leader.
  // 3. the stream is not suspended yet.
332
  if (!tsDisableStream && vnodeIsRoleLeader(pTq->pVnode) && pTq->pVnode->restored) {
333 334 335 336
    if (taosHashGetSize(pTq->pStreamMeta->pTasks) == 0) {
      return 0;
    }

L
Liu Jicong 已提交
337
    if (msgType == TDMT_VND_SUBMIT) {
338
#if 0
L
Liu Jicong 已提交
339
      void* data = taosMemoryMalloc(len);
L
Liu Jicong 已提交
340
      if (data == NULL) {
341
        // todo: for all stream in this vnode, keep this offset in the offset files, and wait for a moment, and then retry
L
Liu Jicong 已提交
342
        terrno = TSDB_CODE_OUT_OF_MEMORY;
343
        tqError("vgId:%d, failed to copy submit data for stream processing, since out of memory", vgId);
L
Liu Jicong 已提交
344 345
        return -1;
      }
346

L
Liu Jicong 已提交
347
      memcpy(data, pReq, len);
348
      SPackedData submit = {.msgStr = data, .msgLen = len, .ver = ver};
L
Liu Jicong 已提交
349

350 351 352 353
      tqDebug("vgId:%d tq copy submit msg:%p len:%d ver:%" PRId64 " from %p for stream", vgId, data, len, ver, pReq);
      tqProcessSubmitReq(pTq, submit);
#endif
      SPackedData submit = {0};
L
Liu Jicong 已提交
354
      tqProcessSubmitReq(pTq, submit);
L
Liu Jicong 已提交
355
    }
H
Haojun Liao 已提交
356

L
Liu Jicong 已提交
357 358
    if (msgType == TDMT_VND_DELETE) {
      tqProcessDelReq(pTq, POINTER_SHIFT(msg, sizeof(SMsgHead)), msgLen - sizeof(SMsgHead), ver);
L
Liu Jicong 已提交
359 360 361 362 363
    }
  }

  return 0;
}
364

H
Haojun Liao 已提交
365
int32_t tqRegisterPushHandle(STQ* pTq, void* pHandle, const SMqPollReq* pRequest, SRpcMsg* pRpcMsg, SMqDataRsp* pDataRsp,
X
Xiaoyu Wang 已提交
366
                            int32_t type) {
367 368 369 370 371 372 373 374 375 376 377 378
  uint64_t   consumerId = pRequest->consumerId;
  int32_t    vgId = TD_VID(pTq->pVnode);
  STqHandle* pTqHandle = pHandle;

  STqPushEntry* pPushEntry = taosMemoryCalloc(1, sizeof(STqPushEntry));
  if (pPushEntry == NULL) {
    tqDebug("tmq poll: consumer:0x%" PRIx64 ", vgId:%d failed to malloc, size:%d", consumerId, vgId,
            (int32_t)sizeof(STqPushEntry));
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

379
  pPushEntry->pHandle = pHandle;
H
Haojun Liao 已提交
380
  pPushEntry->info = pRpcMsg->info;
381 382
  memcpy(pPushEntry->subKey, pTqHandle->subKey, TSDB_SUBSCRIBE_KEY_LEN);

H
Haojun Liao 已提交
383
  if (type == TMQ_MSG_TYPE__TAOSX_RSP) {
X
Xiaoyu Wang 已提交
384
    pPushEntry->pDataRsp = taosMemoryCalloc(1, sizeof(STaosxRsp));
H
Haojun Liao 已提交
385 386 387 388 389 390 391
    memcpy(pPushEntry->pDataRsp, pDataRsp, sizeof(STaosxRsp));
  } else if (type == TMQ_MSG_TYPE__POLL_RSP) {
    pPushEntry->pDataRsp = taosMemoryCalloc(1, sizeof(SMqDataRsp));
    memcpy(pPushEntry->pDataRsp, pDataRsp, sizeof(SMqDataRsp));
  }

  SMqRspHead* pHead = &pPushEntry->pDataRsp->head;
392

H
Haojun Liao 已提交
393 394 395 396
  pHead->consumerId = consumerId;
  pHead->epoch = pRequest->epoch;
  pHead->mqMsgType = type;

397 398
  taosHashPut(pTq->pPushMgr, pTqHandle->subKey, strlen(pTqHandle->subKey), &pPushEntry, sizeof(void*));

X
Xiaoyu Wang 已提交
399 400
  tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s offset:%" PRId64 ", vgId:%d save handle to push mgr, total:%d",
          consumerId, pTqHandle->subKey, pDataRsp->reqOffset.version, vgId, taosHashGetSize(pTq->pPushMgr));
401 402 403
  return 0;
}

404
int32_t tqUnregisterPushHandle(STQ* pTq, const char* pKey, int32_t keyLen, uint64_t consumerId, bool rspConsumer) {
405
  int32_t        vgId = TD_VID(pTq->pVnode);
406
  STqPushEntry** pEntry = taosHashGet(pTq->pPushMgr, pKey, keyLen);
H
Haojun Liao 已提交
407

408
  if (pEntry != NULL) {
H
Haojun Liao 已提交
409
    uint64_t cId = (*pEntry)->pDataRsp->head.consumerId;
410 411 412 413
    ASSERT(consumerId == cId);

    tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s vgId:%d remove from push mgr, remains:%d", consumerId,
            (*pEntry)->subKey, vgId, taosHashGetSize(pTq->pPushMgr) - 1);
H
Haojun Liao 已提交
414

X
Xiaoyu Wang 已提交
415
    if (rspConsumer) {  // rsp the old consumer with empty block.
416
      tqPushDataRsp(*pEntry, vgId);
417
    }
418 419

    taosHashRemove(pTq->pPushMgr, pKey, keyLen);
420 421 422 423
  }

  return 0;
}
424 425 426 427

void recordPushedEntry(SArray* cachedKey, void* pIter) {
  size_t kLen = 0;
  void*  key = taosHashGetKey(pIter, &kLen);
X
Xiaoyu Wang 已提交
428
  SItem  item = {.pKey = strndup(key, kLen), .keyLen = kLen};
429 430 431 432 433
  taosArrayPush(cachedKey, &item);
}

void doRemovePushedEntry(SArray* pCachedKeys, STQ* pTq) {
  int32_t vgId = TD_VID(pTq->pVnode);
X
Xiaoyu Wang 已提交
434
  int32_t numOfKeys = (int32_t)taosArrayGetSize(pCachedKeys);
435 436 437 438

  for (int32_t i = 0; i < numOfKeys; i++) {
    SItem* pItem = taosArrayGet(pCachedKeys, i);
    if (taosHashRemove(pTq->pPushMgr, pItem->pKey, pItem->keyLen) != 0) {
X
Xiaoyu Wang 已提交
439
      tqError("vgId:%d, tq push hash remove key error, key: %s", vgId, (char*)pItem->pKey);
440 441 442 443 444 445 446
    }
  }

  if (numOfKeys > 0) {
    tqDebug("vgId:%d, pushed %d items and remain:%d", vgId, numOfKeys, (int32_t)taosHashGetSize(pTq->pPushMgr));
  }
}