#3530

Committed 16 Nov 2024 07:44AM UTC coverage: 60.219% (-0.7%) from 60.888%

Build # #3530

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Update 03-ad.md

Run Details

118417 of 252124 branches covered (46.97%)

Branch coverage included in aggregate %.

198982 of 274951 relevant lines covered (72.37%)

6072359.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

30.0

/source/dnode/mnode/impl/src/mndStreamHb.c

/*
 * 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/>.
 */

#include "mndStream.h"
#include "mndTrans.h"

typedef struct SFailedCheckpointInfo {
  int64_t streamUid;
  int64_t checkpointId;
  int32_t transId;
} SFailedCheckpointInfo;

static int32_t mndStreamSendUpdateChkptInfoMsg(SMnode *pMnode);
static int32_t mndSendDropOrphanTasksMsg(SMnode *pMnode, SArray *pList);
static int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t transId);
static void    updateStageInfo(STaskStatusEntry *pTaskEntry, int64_t stage);
static void    addIntoFailedChkptList(SArray *pList, const SFailedCheckpointInfo *pInfo);
static int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList);
static int32_t suspendAllStreams(SMnode *pMnode, SRpcHandleInfo *info);
static bool    validateHbMsg(const SArray *pNodeList, int32_t vgId);
static void    cleanupAfterProcessHbMsg(SStreamHbMsg *pReq, SArray *pFailedChkptList, SArray *pOrphanTasks);
static void    doSendHbMsgRsp(int32_t code, SRpcHandleInfo *pRpcInfo, int32_t vgId, int32_t msgId);
static void    checkforOrphanTask(SMnode* pMnode, STaskStatusEntry* p, SArray* pOrphanTasks);

void updateStageInfo(STaskStatusEntry *pTaskEntry, int64_t stage) {
  int32_t numOfNodes = taosArrayGetSize(execInfo.pNodeList);
  for (int32_t j = 0; j < numOfNodes; ++j) {
    SNodeEntry *pNodeEntry = taosArrayGet(execInfo.pNodeList, j);
    if (pNodeEntry == NULL) {
      continue;
    }

    if (pNodeEntry->nodeId == pTaskEntry->nodeId) {
      mInfo("vgId:%d stage updated from %" PRId64 " to %" PRId64 ", nodeUpdate trigger by s-task:0x%" PRIx64,
            pTaskEntry->nodeId, pTaskEntry->stage, stage, pTaskEntry->id.taskId);

      pNodeEntry->stageUpdated = true;
      pTaskEntry->stage = stage;
      break;
    }
  }
}

void addIntoFailedChkptList(SArray *pList, const SFailedCheckpointInfo *pInfo) {
  int32_t num = taosArrayGetSize(pList);
  for (int32_t i = 0; i < num; ++i) {
    SFailedCheckpointInfo *p = taosArrayGet(pList, i);
    if (p && (p->transId == pInfo->transId)) {
      return;
    }
  }

  void* p = taosArrayPush(pList, pInfo);
  if (p == NULL) {
    mError("failed to push failed checkpoint info checkpointId:%" PRId64 " in list", pInfo->checkpointId);
  }
}

int32_t mndCreateStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream) {
  STrans *pTrans = NULL;
  int32_t code = doCreateTrans(pMnode, pStream, NULL, TRN_CONFLICT_NOTHING, MND_STREAM_TASK_RESET_NAME,
                               " reset from failed checkpoint", &pTrans);
  if (pTrans == NULL || code) {
    sdbRelease(pMnode->pSdb, pStream);
    return terrno;
  }

  code = mndStreamRegisterTrans(pTrans, MND_STREAM_TASK_RESET_NAME, pStream->uid);
  if (code) {
    sdbRelease(pMnode->pSdb, pStream);
    mndTransDrop(pTrans);
    return code;
  }

  code = mndStreamSetResetTaskAction(pMnode, pTrans, pStream);
  if (code) {
    sdbRelease(pMnode->pSdb, pStream);
    mndTransDrop(pTrans);
    return code;
  }

  code = mndPersistTransLog(pStream, pTrans, SDB_STATUS_READY);
  if (code != TSDB_CODE_SUCCESS) {
    sdbRelease(pMnode->pSdb, pStream);
    mndTransDrop(pTrans);
    return code;
  }

  code = mndTransPrepare(pMnode, pTrans);
  if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_ACTION_IN_PROGRESS) {
    mError("trans:%d, failed to prepare update stream trans since %s", pTrans->id, tstrerror(code));
    sdbRelease(pMnode->pSdb, pStream);
    mndTransDrop(pTrans);
    return code;
  }

  sdbRelease(pMnode->pSdb, pStream);
  mndTransDrop(pTrans);

  if (code == 0) {
    code = TSDB_CODE_ACTION_IN_PROGRESS;
  }
  return code;
}

int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t transId) {
  int32_t size = sizeof(SStreamTaskResetMsg);

  int32_t num = taosArrayGetSize(execInfo.pKilledChkptTrans);
  for(int32_t i = 0; i < num; ++i) {
    SStreamTaskResetMsg* p = taosArrayGet(execInfo.pKilledChkptTrans, i);
    if (p == NULL) {
      continue;
    }

    if (p->transId == transId && p->streamId == streamId) {
      mDebug("already reset stream:0x%" PRIx64 ", not send reset-msg again for transId:%d", streamId, transId);
      return TSDB_CODE_SUCCESS;
    }
  }

  if (num >= 10) {
    taosArrayRemove(execInfo.pKilledChkptTrans, 0);  // remove this first, append new reset trans in the tail
  }

  SStreamTaskResetMsg p = {.streamId = streamId, .transId = transId};

  void *px = taosArrayPush(execInfo.pKilledChkptTrans, &p);
  if (px == NULL) {
    mError("failed to push reset-msg trans:%d into the killed chkpt trans list, size:%d", transId, num - 1);
    return terrno;
  }

  SStreamTaskResetMsg *pReq = rpcMallocCont(size);
  if (pReq == NULL) {
    return terrno;
  }

  pReq->streamId = streamId;
  pReq->transId = transId;

  SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_TASK_RESET, .pCont = pReq, .contLen = size};
  int32_t code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
  if (code) {
    mError("failed to put reset-task msg into write queue, code:%s", tstrerror(code));
  } else {
    mDebug("send reset task status msg for transId:%d succ", transId);
  }

  return code;
}

int32_t mndStreamSendUpdateChkptInfoMsg(SMnode *pMnode) {  // here reuse the doCheckpointmsg
  int32_t size = sizeof(SMStreamDoCheckpointMsg);
  void   *pMsg = rpcMallocCont(size);
  if (pMsg == NULL) {
    return terrno;
  }

  SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_UPDATE_CHKPT_EVT, .pCont = pMsg, .contLen = size};
  int32_t code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
  if (code) {
    mError("failed to put update-checkpoint-info msg into write queue, code:%s", tstrerror(code));
  } else {
    mDebug("send update checkpoint-info msg succ");
  }

  return code;
}

int32_t mndSendDropOrphanTasksMsg(SMnode *pMnode, SArray *pList) {
  SMStreamDropOrphanMsg msg = {.pList = pList};

  int32_t num = taosArrayGetSize(pList);
  int32_t contLen = tSerializeDropOrphanTaskMsg(NULL, 0, &msg);
  if (contLen <= 0) {
    return terrno;
  }

  void *pReq = rpcMallocCont(contLen);
  if (pReq == NULL) {
    return terrno;
  }

  int32_t code = tSerializeDropOrphanTaskMsg(pReq, contLen, &msg);
  if (code <= 0) {
    mError("failed to serialize the drop orphan task msg, code:%s", tstrerror(code));
  }

  SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_DROP_ORPHANTASKS, .pCont = pReq, .contLen = contLen};
  code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
  if (code) {
    mError("failed to put drop-orphan task msg into write queue, code:%s", tstrerror(code));
  } else {
    mDebug("send drop %d orphan tasks msg succ", num);
  }

  return code;
}

int32_t mndProcessResetStatusReq(SRpcMsg *pReq) {
  SMnode     *pMnode = pReq->info.node;
  int32_t     code = TSDB_CODE_SUCCESS;
  SStreamObj *pStream = NULL;

  SStreamTaskResetMsg* pMsg = pReq->pCont;
  mndKillTransImpl(pMnode, pMsg->transId, "");

  streamMutexLock(&execInfo.lock);
  code = mndResetChkptReportInfo(execInfo.pChkptStreams, pMsg->streamId);   // do thing if failed
  streamMutexUnlock(&execInfo.lock);

  code = mndGetStreamObj(pMnode, pMsg->streamId, &pStream);
  if (pStream == NULL || code != 0) {
    code = TSDB_CODE_STREAM_TASK_NOT_EXIST;
    mError("failed to acquire the streamObj:0x%" PRIx64 " to reset checkpoint, may have been dropped", pStream->uid);
  } else {
    code = mndStreamTransConflictCheck(pMnode, pStream->uid, MND_STREAM_TASK_RESET_NAME, false);
    if (code) {
      mError("stream:%s other trans exists in DB:%s, dstTable:%s failed to start reset-status trans", pStream->name,
             pStream->sourceDb, pStream->targetSTbName);
    } else {
      mDebug("stream:%s (0x%" PRIx64 ") reset checkpoint procedure, transId:%d, create reset trans", pStream->name,
             pStream->uid, pMsg->transId);
      code = mndCreateStreamResetStatusTrans(pMnode, pStream);
    }
  }

  mndReleaseStream(pMnode, pStream);
  return code;
}

int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList) {
  int32_t num = taosArrayGetSize(pNodeList);
  mInfo("set node expired for %d nodes", num);

  for (int k = 0; k < num; ++k) {
    int32_t *pVgId = taosArrayGet(pNodeList, k);
    if (pVgId == NULL) {
      continue;
    }

    mInfo("set node expired for nodeId:%d, total:%d", *pVgId, num);

    bool    setFlag = false;
    int32_t numOfNodes = taosArrayGetSize(execInfo.pNodeList);

    for (int i = 0; i < numOfNodes; ++i) {
      SNodeEntry *pNodeEntry = taosArrayGet(execInfo.pNodeList, i);
      if ((pNodeEntry) && (pNodeEntry->nodeId == *pVgId)) {
        mInfo("vgId:%d expired for some stream tasks, needs update nodeEp", *pVgId);
        pNodeEntry->stageUpdated = true;
        setFlag = true;
        break;
      }
    }

    if (!setFlag) {
      mError("failed to set nodeUpdate flag, nodeId:%d not exists in nodelist", *pVgId);
      return TSDB_CODE_FAILED;
    }
  }
  return TSDB_CODE_SUCCESS;
}

int32_t suspendAllStreams(SMnode *pMnode, SRpcHandleInfo *info) {
  SSdb       *pSdb = pMnode->pSdb;
  SStreamObj *pStream = NULL;
  void       *pIter = NULL;
  int32_t     code = 0;

  while (1) {
    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
    if (pIter == NULL) break;

    if (pStream->status != STREAM_STATUS__PAUSE) {
      SMPauseStreamReq reqPause = {0};
      strcpy(reqPause.name, pStream->name);
      reqPause.igNotExists = 1;

      int32_t contLen = tSerializeSMPauseStreamReq(NULL, 0, &reqPause);
      void   *pHead = rpcMallocCont(contLen);
      if (pHead == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        sdbRelease(pSdb, pStream);
        continue;
      }

      code = tSerializeSMPauseStreamReq(pHead, contLen, &reqPause);
      if (code) {
        sdbRelease(pSdb, pStream);
        continue;
      }

      SRpcMsg rpcMsg = {
          .msgType = TDMT_MND_PAUSE_STREAM,
          .pCont = pHead,
          .contLen = contLen,
          .info = *info,
      };

      code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
      mInfo("receive pause stream:%s, %s, %" PRId64 ", because grant expired, code:%s", pStream->name, reqPause.name,
            pStream->uid, tstrerror(code));
    }

    sdbRelease(pSdb, pStream);
  }
  return code;
}

int32_t mndProcessStreamHb(SRpcMsg *pReq) {
  SMnode      *pMnode = pReq->info.node;
  SStreamHbMsg req = {0};
  SArray      *pFailedChkpt = NULL;
  SArray      *pOrphanTasks = NULL;
  int32_t      code = 0;

  if ((code = grantCheckExpire(TSDB_GRANT_STREAMS)) < 0) {
    if (suspendAllStreams(pMnode, &pReq->info) < 0) {
      return code;
    }
  }

  SDecoder decoder = {0};
  tDecoderInit(&decoder, pReq->pCont, pReq->contLen);

  if (tDecodeStreamHbMsg(&decoder, &req) < 0) {
    tCleanupStreamHbMsg(&req);
    tDecoderClear(&decoder);
    TAOS_RETURN(TSDB_CODE_INVALID_MSG);
  }
  tDecoderClear(&decoder);

  mDebug("receive stream-meta hb from vgId:%d, active numOfTasks:%d, HbMsgId:%d, HbMsgTs:%" PRId64, req.vgId,
         req.numOfTasks, req.msgId, req.ts);

  pFailedChkpt = taosArrayInit(4, sizeof(SFailedCheckpointInfo));
  pOrphanTasks = taosArrayInit(4, sizeof(SOrphanTask));
  if (pFailedChkpt == NULL || pOrphanTasks == NULL) {
    taosArrayDestroy(pFailedChkpt);
    taosArrayDestroy(pOrphanTasks);
    TAOS_RETURN(terrno);
  }

  streamMutexLock(&execInfo.lock);

  mndInitStreamExecInfo(pMnode, &execInfo);
  if (!validateHbMsg(execInfo.pNodeList, req.vgId)) {
    mError("vgId:%d not exists in nodeList buf, discarded", req.vgId);

    doSendHbMsgRsp(terrno, &pReq->info, req.vgId, req.msgId);

    streamMutexUnlock(&execInfo.lock);
    cleanupAfterProcessHbMsg(&req, pFailedChkpt, pOrphanTasks);
    TAOS_RETURN(TSDB_CODE_INVALID_MSG);
  }

  for(int32_t i = 0; i < taosArrayGetSize(execInfo.pNodeList); ++i) {
    SNodeEntry* pEntry = taosArrayGet(execInfo.pNodeList, i);
    if (pEntry == NULL) {
      continue;
    }

    if (pEntry->nodeId != req.vgId) {
      continue;
    }

    if ((pEntry->lastHbMsgId == req.msgId) && (pEntry->lastHbMsgTs == req.ts)) {
      mError("vgId:%d HbMsgId:%d already handled, bh msg discard", pEntry->nodeId, req.msgId);

      terrno = TSDB_CODE_INVALID_MSG;
      doSendHbMsgRsp(terrno, &pReq->info, req.vgId, req.msgId);

      streamMutexUnlock(&execInfo.lock);
      cleanupAfterProcessHbMsg(&req, pFailedChkpt, pOrphanTasks);
      return terrno;
    } else {
      pEntry->lastHbMsgId = req.msgId;
      pEntry->lastHbMsgTs = req.ts;
    }
  }

  int32_t numOfUpdated = taosArrayGetSize(req.pUpdateNodes);
  if (numOfUpdated > 0) {
    mDebug("%d stream node(s) need updated from hbMsg(vgId:%d)", numOfUpdated, req.vgId);
    int32_t unused = setNodeEpsetExpiredFlag(req.pUpdateNodes);
  }

  bool snodeChanged = false;
  for (int32_t i = 0; i < req.numOfTasks; ++i) {
    STaskStatusEntry *p = taosArrayGet(req.pTaskStatus, i);
    if (p == NULL) {
      continue;
    }

    STaskStatusEntry *pTaskEntry = taosHashGet(execInfo.pTaskMap, &p->id, sizeof(p->id));
    if (pTaskEntry == NULL) {
      checkforOrphanTask(pMnode, p, pOrphanTasks);
      continue;
    }

    STaskCkptInfo *pChkInfo = &p->checkpointInfo;
    if (pChkInfo->consensusChkptId != 0) {
      SRestoreCheckpointInfo cp = {
          .streamId = p->id.streamId,
          .taskId = p->id.taskId,
          .checkpointId = p->checkpointInfo.latestId,
          .startTs = pChkInfo->consensusTs,
      };

      SStreamObj *pStream = NULL;
      code = mndGetStreamObj(pMnode, p->id.streamId, &pStream);
      if (code) {
        mError("stream:0x%" PRIx64 " not exist, failed to handle consensus checkpoint-info req for task:0x%x, code:%s",
               p->id.streamId, (int32_t)p->id.taskId, tstrerror(code));
        continue;
      }

      int32_t numOfTasks = mndGetNumOfStreamTasks(pStream);

      SCheckpointConsensusInfo *pInfo = NULL;
      code = mndGetConsensusInfo(execInfo.pStreamConsensus, p->id.streamId, numOfTasks, &pInfo);
      if (code == 0) {
        mndAddConsensusTasks(pInfo, &cp);
      } else {
        mError("failed to get consensus checkpoint-info for stream:0x%" PRIx64, p->id.streamId);
      }

      mndReleaseStream(pMnode, pStream);
    }

    if (pTaskEntry->stage != p->stage && pTaskEntry->stage != -1) {
      updateStageInfo(pTaskEntry, p->stage);
      if (pTaskEntry->nodeId == SNODE_HANDLE) {
        snodeChanged = true;
      }
    } else {
      streamTaskStatusCopy(pTaskEntry, p);

      if ((pChkInfo->activeId != 0) && pChkInfo->failed) {
        mError("stream task:0x%" PRIx64 " checkpointId:%" PRId64 " transId:%d failed, kill it", p->id.taskId,
               pChkInfo->activeId, pChkInfo->activeTransId);

        SFailedCheckpointInfo info = {
            .transId = pChkInfo->activeTransId, .checkpointId = pChkInfo->activeId, .streamUid = p->id.streamId};
        addIntoFailedChkptList(pFailedChkpt, &info);

        // remove failed trans from pChkptStreams
        code = mndResetChkptReportInfo(execInfo.pChkptStreams, p->id.streamId);
        if (code) {
          mError("failed to remove stream:0x%"PRIx64" in checkpoint stream list", p->id.streamId);
        }
      }
    }

    if (p->status != TASK_STATUS__READY) {
      mDebug("received s-task:0x%" PRIx64 " not in ready status:%s", p->id.taskId, streamTaskGetStatusStr(p->status));
    }
  }

  // current checkpoint is failed, rollback from the checkpoint trans
  // kill the checkpoint trans and then set all tasks status to be normal
  if (taosArrayGetSize(pFailedChkpt) > 0) {
    bool allReady = true;

    if (pMnode != NULL) {
      SArray *p = NULL;
      code = mndTakeVgroupSnapshot(pMnode, &allReady, &p);
      taosArrayDestroy(p);
      if (code) {
        mError("failed to get the vgroup snapshot, ignore it and continue");
      }
    } else {
      allReady = false;
    }

    if (allReady || snodeChanged) {
      // if the execInfo.activeCheckpoint == 0, the checkpoint is restoring from wal
      for (int32_t i = 0; i < taosArrayGetSize(pFailedChkpt); ++i) {
        SFailedCheckpointInfo *pInfo = taosArrayGet(pFailedChkpt, i);
        if (pInfo == NULL) {
          continue;
        }

        mInfo("checkpointId:%" PRId64 " transId:%d failed, issue task-reset trans to reset all tasks status",
              pInfo->checkpointId, pInfo->transId);

        code = mndSendResetFromCheckpointMsg(pMnode, pInfo->streamUid, pInfo->transId);
        if (code) {
          mError("failed to create reset task trans, code:%s", tstrerror(code));
        }
      }
    } else {
      mInfo("not all vgroups are ready, wait for next HB from stream tasks to reset the task status");
    }
  }

  // handle the orphan tasks that are invalid but not removed in some vnodes or snode due to some unknown errors.
  if (taosArrayGetSize(pOrphanTasks) > 0) {
    code = mndSendDropOrphanTasksMsg(pMnode, pOrphanTasks);
    if (code) {
      mError("failed to send drop orphan tasks msg, code:%s, try next time", tstrerror(code));
    }
  }

  if (pMnode != NULL) {  // make sure that the unit test case can work
    code = mndStreamSendUpdateChkptInfoMsg(pMnode);
    if (code) {
      mError("failed to send update checkpointInfo msg, code:%s, try next time", tstrerror(code));
    }
  }

  streamMutexUnlock(&execInfo.lock);

  doSendHbMsgRsp(TSDB_CODE_SUCCESS, &pReq->info, req.vgId, req.msgId);
  cleanupAfterProcessHbMsg(&req, pFailedChkpt, pOrphanTasks);

  return code;
}

bool validateHbMsg(const SArray *pNodeList, int32_t vgId) {
  for (int32_t i = 0; i < taosArrayGetSize(pNodeList); ++i) {
    SNodeEntry *pEntry = taosArrayGet(pNodeList, i);
    if ((pEntry) && (pEntry->nodeId == vgId)) {
      return true;
    }
  }

  return false;
}

void cleanupAfterProcessHbMsg(SStreamHbMsg *pReq, SArray *pFailedChkptList, SArray *pOrphanTasks) {
  tCleanupStreamHbMsg(pReq);
  taosArrayDestroy(pFailedChkptList);
  taosArrayDestroy(pOrphanTasks);
}

void doSendHbMsgRsp(int32_t code, SRpcHandleInfo *pRpcInfo, int32_t vgId, int32_t msgId) {
  SRpcMsg rsp = {.code = code, .info = *pRpcInfo, .contLen = sizeof(SMStreamHbRspMsg)};
  rsp.pCont = rpcMallocCont(rsp.contLen);

  SMStreamHbRspMsg *pMsg = rsp.pCont;
  pMsg->head.vgId = htonl(vgId);
  pMsg->msgId = msgId;

  tmsgSendRsp(&rsp);
  pRpcInfo->handle = NULL;  // disable auto rsp
}

void checkforOrphanTask(SMnode* pMnode, STaskStatusEntry* p, SArray* pOrphanTasks) {
  SStreamObj *pStream = NULL;

  int32_t code = mndGetStreamObj(pMnode, p->id.streamId, &pStream);
  if (code) {
    mError("stream:0x%" PRIx64 " not exists, s-task:0x%" PRIx64 " not found in task list, add into orphan list",
           p->id.streamId, p->id.taskId);

    SOrphanTask oTask = {.streamId = p->id.streamId, .taskId = p->id.taskId, .nodeId = p->nodeId};
    void       *px = taosArrayPush(pOrphanTasks, &oTask);
    if (px == NULL) {
      mError("failed to put task into orphan list, taskId:0x%" PRIx64", code:%s", p->id.taskId, tstrerror(terrno));
    }
  } else {
    if (pStream != NULL) {
      mndReleaseStream(pMnode, pStream);
    }

    mError("s-task:0x%" PRIx64 " not found in task list but exists in mnode meta, data inconsistent, not drop yet",
           p->id.taskId);
  }
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "mndStream.h"
17	#include "mndTrans.h"
18
19	typedef struct SFailedCheckpointInfo {
20	int64_t streamUid;
21	int64_t checkpointId;
22	int32_t transId;
23	} SFailedCheckpointInfo;
24
25	static int32_t mndStreamSendUpdateChkptInfoMsg(SMnode *pMnode);
26	static int32_t mndSendDropOrphanTasksMsg(SMnode pMnode, SArray pList);
27	static int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t transId);
28	static void updateStageInfo(STaskStatusEntry *pTaskEntry, int64_t stage);
29	static void addIntoFailedChkptList(SArray pList, const SFailedCheckpointInfo pInfo);
30	static int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList);
31	static int32_t suspendAllStreams(SMnode pMnode, SRpcHandleInfo info);
32	static bool validateHbMsg(const SArray *pNodeList, int32_t vgId);
33	static void cleanupAfterProcessHbMsg(SStreamHbMsg pReq, SArray pFailedChkptList, SArray *pOrphanTasks);
34	static void doSendHbMsgRsp(int32_t code, SRpcHandleInfo *pRpcInfo, int32_t vgId, int32_t msgId);
35	static void checkforOrphanTask(SMnode* pMnode, STaskStatusEntry* p, SArray* pOrphanTasks);
36
37	void updateStageInfo(STaskStatusEntry *pTaskEntry, int64_t stage) {	7✔
38	int32_t numOfNodes = taosArrayGetSize(execInfo.pNodeList);	7✔
39	for (int32_t j = 0; j < numOfNodes; ++j) {	11!
40	SNodeEntry *pNodeEntry = taosArrayGet(execInfo.pNodeList, j);	11✔
41	if (pNodeEntry == NULL) {	11!
42	continue;	×
43	}
44
45	if (pNodeEntry->nodeId == pTaskEntry->nodeId) {	11✔
46	mInfo("vgId:%d stage updated from %" PRId64 " to %" PRId64 ", nodeUpdate trigger by s-task:0x%" PRIx64,	7!
47	pTaskEntry->nodeId, pTaskEntry->stage, stage, pTaskEntry->id.taskId);
48
49	pNodeEntry->stageUpdated = true;	7✔
50	pTaskEntry->stage = stage;	7✔
51	break;	7✔
52	}
53	}
54	}	7✔
55
56	void addIntoFailedChkptList(SArray pList, const SFailedCheckpointInfo pInfo) {	×
57	int32_t num = taosArrayGetSize(pList);	×
58	for (int32_t i = 0; i < num; ++i) {	×
59	SFailedCheckpointInfo *p = taosArrayGet(pList, i);	×
60	if (p && (p->transId == pInfo->transId)) {	×
61	return;	×
62	}
63	}
64
65	void* p = taosArrayPush(pList, pInfo);	×
66	if (p == NULL) {	×
67	mError("failed to push failed checkpoint info checkpointId:%" PRId64 " in list", pInfo->checkpointId);	×
68	}
69	}
70
71	int32_t mndCreateStreamResetStatusTrans(SMnode pMnode, SStreamObj pStream) {	×
72	STrans *pTrans = NULL;	×
73	int32_t code = doCreateTrans(pMnode, pStream, NULL, TRN_CONFLICT_NOTHING, MND_STREAM_TASK_RESET_NAME,	×
74	" reset from failed checkpoint", &pTrans);
75	if (pTrans == NULL \|\| code) {	×
76	sdbRelease(pMnode->pSdb, pStream);	×
77	return terrno;	×
78	}
79
80	code = mndStreamRegisterTrans(pTrans, MND_STREAM_TASK_RESET_NAME, pStream->uid);	×
81	if (code) {	×
82	sdbRelease(pMnode->pSdb, pStream);	×
83	mndTransDrop(pTrans);	×
84	return code;	×
85	}
86
87	code = mndStreamSetResetTaskAction(pMnode, pTrans, pStream);	×
88	if (code) {	×
89	sdbRelease(pMnode->pSdb, pStream);	×
90	mndTransDrop(pTrans);	×
91	return code;	×
92	}
93
94	code = mndPersistTransLog(pStream, pTrans, SDB_STATUS_READY);	×
95	if (code != TSDB_CODE_SUCCESS) {	×
96	sdbRelease(pMnode->pSdb, pStream);	×
97	mndTransDrop(pTrans);	×
98	return code;	×
99	}
100
101	code = mndTransPrepare(pMnode, pTrans);	×
102	if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_ACTION_IN_PROGRESS) {	×
103	mError("trans:%d, failed to prepare update stream trans since %s", pTrans->id, tstrerror(code));	×
104	sdbRelease(pMnode->pSdb, pStream);	×
105	mndTransDrop(pTrans);	×
106	return code;	×
107	}
108
109	sdbRelease(pMnode->pSdb, pStream);	×
110	mndTransDrop(pTrans);	×
111
112	if (code == 0) {	×
113	code = TSDB_CODE_ACTION_IN_PROGRESS;	×
114	}
115	return code;	×
116	}
117
118	int32_t mndSendResetFromCheckpointMsg(SMnode *pMnode, int64_t streamId, int32_t transId) {	×
119	int32_t size = sizeof(SStreamTaskResetMsg);	×
120
121	int32_t num = taosArrayGetSize(execInfo.pKilledChkptTrans);	×
122	for(int32_t i = 0; i < num; ++i) {	×
123	SStreamTaskResetMsg* p = taosArrayGet(execInfo.pKilledChkptTrans, i);	×
124	if (p == NULL) {	×
125	continue;	×
126	}
127
128	if (p->transId == transId && p->streamId == streamId) {	×
129	mDebug("already reset stream:0x%" PRIx64 ", not send reset-msg again for transId:%d", streamId, transId);	×
130	return TSDB_CODE_SUCCESS;	×
131	}
132	}
133
134	if (num >= 10) {	×
135	taosArrayRemove(execInfo.pKilledChkptTrans, 0); // remove this first, append new reset trans in the tail	×
136	}
137
138	SStreamTaskResetMsg p = {.streamId = streamId, .transId = transId};	×
139
140	void *px = taosArrayPush(execInfo.pKilledChkptTrans, &p);	×
141	if (px == NULL) {	×
142	mError("failed to push reset-msg trans:%d into the killed chkpt trans list, size:%d", transId, num - 1);	×
143	return terrno;	×
144	}
145
146	SStreamTaskResetMsg *pReq = rpcMallocCont(size);	×
147	if (pReq == NULL) {	×
148	return terrno;	×
149	}
150
151	pReq->streamId = streamId;	×
152	pReq->transId = transId;	×
153
154	SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_TASK_RESET, .pCont = pReq, .contLen = size};	×
155	int32_t code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);	×
156	if (code) {	×
157	mError("failed to put reset-task msg into write queue, code:%s", tstrerror(code));	×
158	} else {
159	mDebug("send reset task status msg for transId:%d succ", transId);	×
160	}
161
162	return code;	×
163	}
164
165	int32_t mndStreamSendUpdateChkptInfoMsg(SMnode *pMnode) { // here reuse the doCheckpointmsg	4,272✔
166	int32_t size = sizeof(SMStreamDoCheckpointMsg);	4,272✔
167	void *pMsg = rpcMallocCont(size);	4,272✔
168	if (pMsg == NULL) {	4,272!
169	return terrno;	×
170	}
171
172	SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_UPDATE_CHKPT_EVT, .pCont = pMsg, .contLen = size};	4,272✔
173	int32_t code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);	4,272✔
174	if (code) {	4,272!
175	mError("failed to put update-checkpoint-info msg into write queue, code:%s", tstrerror(code));	×
176	} else {
177	mDebug("send update checkpoint-info msg succ");	4,272✔
178	}
179
180	return code;	4,272✔
181	}
182
183	int32_t mndSendDropOrphanTasksMsg(SMnode pMnode, SArray pList) {	×
184	SMStreamDropOrphanMsg msg = {.pList = pList};	×
185
186	int32_t num = taosArrayGetSize(pList);	×
187	int32_t contLen = tSerializeDropOrphanTaskMsg(NULL, 0, &msg);	×
188	if (contLen <= 0) {	×
189	return terrno;	×
190	}
191
192	void *pReq = rpcMallocCont(contLen);	×
193	if (pReq == NULL) {	×
194	return terrno;	×
195	}
196
197	int32_t code = tSerializeDropOrphanTaskMsg(pReq, contLen, &msg);	×
198	if (code <= 0) {	×
199	mError("failed to serialize the drop orphan task msg, code:%s", tstrerror(code));	×
200	}
201
202	SRpcMsg rpcMsg = {.msgType = TDMT_MND_STREAM_DROP_ORPHANTASKS, .pCont = pReq, .contLen = contLen};	×
203	code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);	×
204	if (code) {	×
205	mError("failed to put drop-orphan task msg into write queue, code:%s", tstrerror(code));	×
206	} else {
207	mDebug("send drop %d orphan tasks msg succ", num);	×
208	}
209
210	return code;	×
211	}
212
213	int32_t mndProcessResetStatusReq(SRpcMsg *pReq) {	×
214	SMnode *pMnode = pReq->info.node;	×
215	int32_t code = TSDB_CODE_SUCCESS;	×
216	SStreamObj *pStream = NULL;	×
217
218	SStreamTaskResetMsg* pMsg = pReq->pCont;	×
219	mndKillTransImpl(pMnode, pMsg->transId, "");	×
220
221	streamMutexLock(&execInfo.lock);	×
222	code = mndResetChkptReportInfo(execInfo.pChkptStreams, pMsg->streamId); // do thing if failed	×
223	streamMutexUnlock(&execInfo.lock);	×
224
225	code = mndGetStreamObj(pMnode, pMsg->streamId, &pStream);	×
226	if (pStream == NULL \|\| code != 0) {	×
227	code = TSDB_CODE_STREAM_TASK_NOT_EXIST;	×
228	mError("failed to acquire the streamObj:0x%" PRIx64 " to reset checkpoint, may have been dropped", pStream->uid);	×
229	} else {
230	code = mndStreamTransConflictCheck(pMnode, pStream->uid, MND_STREAM_TASK_RESET_NAME, false);	×
231	if (code) {	×
232	mError("stream:%s other trans exists in DB:%s, dstTable:%s failed to start reset-status trans", pStream->name,	×
233	pStream->sourceDb, pStream->targetSTbName);
234	} else {
235	mDebug("stream:%s (0x%" PRIx64 ") reset checkpoint procedure, transId:%d, create reset trans", pStream->name,	×
236	pStream->uid, pMsg->transId);
237	code = mndCreateStreamResetStatusTrans(pMnode, pStream);	×
238	}
239	}
240
241	mndReleaseStream(pMnode, pStream);	×
242	return code;	×
243	}
244
245	int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList) {	×
246	int32_t num = taosArrayGetSize(pNodeList);	×
247	mInfo("set node expired for %d nodes", num);	×
248
249	for (int k = 0; k < num; ++k) {	×
250	int32_t *pVgId = taosArrayGet(pNodeList, k);	×
251	if (pVgId == NULL) {	×
252	continue;	×
253	}
254
255	mInfo("set node expired for nodeId:%d, total:%d", *pVgId, num);	×
256
257	bool setFlag = false;	×
258	int32_t numOfNodes = taosArrayGetSize(execInfo.pNodeList);	×
259
260	for (int i = 0; i < numOfNodes; ++i) {	×
261	SNodeEntry *pNodeEntry = taosArrayGet(execInfo.pNodeList, i);	×
262	if ((pNodeEntry) && (pNodeEntry->nodeId == *pVgId)) {	×
263	mInfo("vgId:%d expired for some stream tasks, needs update nodeEp", *pVgId);	×
264	pNodeEntry->stageUpdated = true;	×
265	setFlag = true;	×
266	break;	×
267	}
268	}
269
270	if (!setFlag) {	×
271	mError("failed to set nodeUpdate flag, nodeId:%d not exists in nodelist", *pVgId);	×
272	return TSDB_CODE_FAILED;	×
273	}
274	}
275	return TSDB_CODE_SUCCESS;	×
276	}
277
278	int32_t suspendAllStreams(SMnode pMnode, SRpcHandleInfo info) {	×
279	SSdb *pSdb = pMnode->pSdb;	×
280	SStreamObj *pStream = NULL;	×
281	void *pIter = NULL;	×
282	int32_t code = 0;	×
283
284	while (1) {
285	pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);	×
286	if (pIter == NULL) break;	×
287
288	if (pStream->status != STREAM_STATUS__PAUSE) {	×
289	SMPauseStreamReq reqPause = {0};	×
290	strcpy(reqPause.name, pStream->name);	×
291	reqPause.igNotExists = 1;	×
292
293	int32_t contLen = tSerializeSMPauseStreamReq(NULL, 0, &reqPause);	×
294	void *pHead = rpcMallocCont(contLen);	×
295	if (pHead == NULL) {	×
296	code = TSDB_CODE_OUT_OF_MEMORY;	×
297	sdbRelease(pSdb, pStream);	×
298	continue;	×
299	}
300
301	code = tSerializeSMPauseStreamReq(pHead, contLen, &reqPause);	×
302	if (code) {	×
303	sdbRelease(pSdb, pStream);	×
304	continue;	×
305	}
306
307	SRpcMsg rpcMsg = {	×
308	.msgType = TDMT_MND_PAUSE_STREAM,
309	.pCont = pHead,
310	.contLen = contLen,
311	.info = *info,
312	};
313
314	code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);	×
315	mInfo("receive pause stream:%s, %s, %" PRId64 ", because grant expired, code:%s", pStream->name, reqPause.name,	×
316	pStream->uid, tstrerror(code));
317	}
318
319	sdbRelease(pSdb, pStream);	×
320	}
321	return code;	×
322	}
323
324	int32_t mndProcessStreamHb(SRpcMsg *pReq) {	4,272✔
325	SMnode *pMnode = pReq->info.node;	4,272✔
326	SStreamHbMsg req = {0};	4,272✔
327	SArray *pFailedChkpt = NULL;	4,272✔
328	SArray *pOrphanTasks = NULL;	4,272✔
329	int32_t code = 0;	4,272✔
330
331	if ((code = grantCheckExpire(TSDB_GRANT_STREAMS)) < 0) {	4,272!
332	if (suspendAllStreams(pMnode, &pReq->info) < 0) {	×
333	return code;	×
334	}
335	}
336
337	SDecoder decoder = {0};	4,272✔
338	tDecoderInit(&decoder, pReq->pCont, pReq->contLen);	4,272✔
339
340	if (tDecodeStreamHbMsg(&decoder, &req) < 0) {	4,272!
341	tCleanupStreamHbMsg(&req);	×
342	tDecoderClear(&decoder);	×
343	TAOS_RETURN(TSDB_CODE_INVALID_MSG);	×
344	}
345	tDecoderClear(&decoder);	4,272✔
346
347	mDebug("receive stream-meta hb from vgId:%d, active numOfTasks:%d, HbMsgId:%d, HbMsgTs:%" PRId64, req.vgId,	4,272✔
348	req.numOfTasks, req.msgId, req.ts);
349
350	pFailedChkpt = taosArrayInit(4, sizeof(SFailedCheckpointInfo));	4,272✔
351	pOrphanTasks = taosArrayInit(4, sizeof(SOrphanTask));	4,272✔
352	if (pFailedChkpt == NULL \|\| pOrphanTasks == NULL) {	4,271!
353	taosArrayDestroy(pFailedChkpt);	×
354	taosArrayDestroy(pOrphanTasks);	×
355	TAOS_RETURN(terrno);	×
356	}
357
358	streamMutexLock(&execInfo.lock);	4,271✔
359
360	mndInitStreamExecInfo(pMnode, &execInfo);	4,272✔
361	if (!validateHbMsg(execInfo.pNodeList, req.vgId)) {	4,272!
362	mError("vgId:%d not exists in nodeList buf, discarded", req.vgId);	×
363
364	doSendHbMsgRsp(terrno, &pReq->info, req.vgId, req.msgId);	×
365
366	streamMutexUnlock(&execInfo.lock);	×
367	cleanupAfterProcessHbMsg(&req, pFailedChkpt, pOrphanTasks);	×
368	TAOS_RETURN(TSDB_CODE_INVALID_MSG);	×
369	}
370
371	for(int32_t i = 0; i < taosArrayGetSize(execInfo.pNodeList); ++i) {	35,299✔
372	SNodeEntry* pEntry = taosArrayGet(execInfo.pNodeList, i);	31,027✔
373	if (pEntry == NULL) {	31,027!
374	continue;	×
375	}
376
377	if (pEntry->nodeId != req.vgId) {	31,027✔
378	continue;	26,755✔
379	}
380
381	if ((pEntry->lastHbMsgId == req.msgId) && (pEntry->lastHbMsgTs == req.ts)) {	4,272!
382	mError("vgId:%d HbMsgId:%d already handled, bh msg discard", pEntry->nodeId, req.msgId);	×
383
384	terrno = TSDB_CODE_INVALID_MSG;	×
385	doSendHbMsgRsp(terrno, &pReq->info, req.vgId, req.msgId);	×
386
387	streamMutexUnlock(&execInfo.lock);	×
388	cleanupAfterProcessHbMsg(&req, pFailedChkpt, pOrphanTasks);	×
389	return terrno;	×
390	} else {
391	pEntry->lastHbMsgId = req.msgId;	4,272✔
392	pEntry->lastHbMsgTs = req.ts;	4,272✔
393	}
394	}
395
396	int32_t numOfUpdated = taosArrayGetSize(req.pUpdateNodes);	4,272✔
397	if (numOfUpdated > 0) {	4,272!
398	mDebug("%d stream node(s) need updated from hbMsg(vgId:%d)", numOfUpdated, req.vgId);	×
399	int32_t unused = setNodeEpsetExpiredFlag(req.pUpdateNodes);	×
400	}
401
402	bool snodeChanged = false;	4,272✔
403	for (int32_t i = 0; i < req.numOfTasks; ++i) {	18,651✔
404	STaskStatusEntry *p = taosArrayGet(req.pTaskStatus, i);	14,379✔
405	if (p == NULL) {	14,379!
406	continue;	×
407	}
408
409	STaskStatusEntry *pTaskEntry = taosHashGet(execInfo.pTaskMap, &p->id, sizeof(p->id));	14,379✔
410	if (pTaskEntry == NULL) {	14,379✔
411	checkforOrphanTask(pMnode, p, pOrphanTasks);	44✔
412	continue;	44✔
413	}
414
415	STaskCkptInfo *pChkInfo = &p->checkpointInfo;	14,335✔
416	if (pChkInfo->consensusChkptId != 0) {	14,335✔
417	SRestoreCheckpointInfo cp = {	21✔
418	.streamId = p->id.streamId,	21✔
419	.taskId = p->id.taskId,	21✔
420	.checkpointId = p->checkpointInfo.latestId,	21✔
421	.startTs = pChkInfo->consensusTs,	21✔
422	};
423
424	SStreamObj *pStream = NULL;	21✔
425	code = mndGetStreamObj(pMnode, p->id.streamId, &pStream);	21✔
426	if (code) {	21!
427	mError("stream:0x%" PRIx64 " not exist, failed to handle consensus checkpoint-info req for task:0x%x, code:%s",	×
428	p->id.streamId, (int32_t)p->id.taskId, tstrerror(code));
429	continue;	×
430	}
431
432	int32_t numOfTasks = mndGetNumOfStreamTasks(pStream);	21✔
433
434	SCheckpointConsensusInfo *pInfo = NULL;	21✔
435	code = mndGetConsensusInfo(execInfo.pStreamConsensus, p->id.streamId, numOfTasks, &pInfo);	21✔
436	if (code == 0) {	21!
437	mndAddConsensusTasks(pInfo, &cp);	21✔
438	} else {
439	mError("failed to get consensus checkpoint-info for stream:0x%" PRIx64, p->id.streamId);	×
440	}
441
442	mndReleaseStream(pMnode, pStream);	21✔
443	}
444
445	if (pTaskEntry->stage != p->stage && pTaskEntry->stage != -1) {	14,335✔
446	updateStageInfo(pTaskEntry, p->stage);	7✔
447	if (pTaskEntry->nodeId == SNODE_HANDLE) {	7✔
448	snodeChanged = true;	1✔
449	}
450	} else {
451	streamTaskStatusCopy(pTaskEntry, p);	14,328✔
452
453	if ((pChkInfo->activeId != 0) && pChkInfo->failed) {	14,328!
454	mError("stream task:0x%" PRIx64 " checkpointId:%" PRId64 " transId:%d failed, kill it", p->id.taskId,	×
455	pChkInfo->activeId, pChkInfo->activeTransId);
456
457	SFailedCheckpointInfo info = {	×
458	.transId = pChkInfo->activeTransId, .checkpointId = pChkInfo->activeId, .streamUid = p->id.streamId};	×
459	addIntoFailedChkptList(pFailedChkpt, &info);	×
460
461	// remove failed trans from pChkptStreams
462	code = mndResetChkptReportInfo(execInfo.pChkptStreams, p->id.streamId);	×
463	if (code) {	×
464	mError("failed to remove stream:0x%"PRIx64" in checkpoint stream list", p->id.streamId);	×
465	}
466	}
467	}
468
469	if (p->status != TASK_STATUS__READY) {	14,335✔
470	mDebug("received s-task:0x%" PRIx64 " not in ready status:%s", p->id.taskId, streamTaskGetStatusStr(p->status));	1,908✔
471	}
472	}
473
474	// current checkpoint is failed, rollback from the checkpoint trans
475	// kill the checkpoint trans and then set all tasks status to be normal
476	if (taosArrayGetSize(pFailedChkpt) > 0) {	4,272!
477	bool allReady = true;	×
478
479	if (pMnode != NULL) {	×
480	SArray *p = NULL;	×
481	code = mndTakeVgroupSnapshot(pMnode, &allReady, &p);	×
482	taosArrayDestroy(p);	×
483	if (code) {	×
484	mError("failed to get the vgroup snapshot, ignore it and continue");	×
485	}
486	} else {
487	allReady = false;	×
488	}
489
490	if (allReady \|\| snodeChanged) {	×
491	// if the execInfo.activeCheckpoint == 0, the checkpoint is restoring from wal
492	for (int32_t i = 0; i < taosArrayGetSize(pFailedChkpt); ++i) {	×
493	SFailedCheckpointInfo *pInfo = taosArrayGet(pFailedChkpt, i);	×
494	if (pInfo == NULL) {	×
495	continue;	×
496	}
497
498	mInfo("checkpointId:%" PRId64 " transId:%d failed, issue task-reset trans to reset all tasks status",	×
499	pInfo->checkpointId, pInfo->transId);
500
501	code = mndSendResetFromCheckpointMsg(pMnode, pInfo->streamUid, pInfo->transId);	×
502	if (code) {	×
503	mError("failed to create reset task trans, code:%s", tstrerror(code));	×
504	}
505	}
506	} else {
507	mInfo("not all vgroups are ready, wait for next HB from stream tasks to reset the task status");	×
508	}
509	}
510
511	// handle the orphan tasks that are invalid but not removed in some vnodes or snode due to some unknown errors.
512	if (taosArrayGetSize(pOrphanTasks) > 0) {	4,272!
513	code = mndSendDropOrphanTasksMsg(pMnode, pOrphanTasks);	×
514	if (code) {	×
515	mError("failed to send drop orphan tasks msg, code:%s, try next time", tstrerror(code));	×
516	}
517	}
518
519	if (pMnode != NULL) { // make sure that the unit test case can work	4,272!
520	code = mndStreamSendUpdateChkptInfoMsg(pMnode);	4,272✔
521	if (code) {	4,272!
522	mError("failed to send update checkpointInfo msg, code:%s, try next time", tstrerror(code));	×
523	}
524	}
525
526	streamMutexUnlock(&execInfo.lock);	4,272✔
527
528	doSendHbMsgRsp(TSDB_CODE_SUCCESS, &pReq->info, req.vgId, req.msgId);	4,272✔
529	cleanupAfterProcessHbMsg(&req, pFailedChkpt, pOrphanTasks);	4,272✔
530
531	return code;	4,272✔
532	}
533
534	bool validateHbMsg(const SArray *pNodeList, int32_t vgId) {	4,272✔
535	for (int32_t i = 0; i < taosArrayGetSize(pNodeList); ++i) {	18,602!
536	SNodeEntry *pEntry = taosArrayGet(pNodeList, i);	18,602✔
537	if ((pEntry) && (pEntry->nodeId == vgId)) {	18,602!
538	return true;	4,272✔
539	}
540	}
541
542	return false;	×
543	}
544
545	void cleanupAfterProcessHbMsg(SStreamHbMsg pReq, SArray pFailedChkptList, SArray *pOrphanTasks) {	4,272✔
546	tCleanupStreamHbMsg(pReq);	4,272✔
547	taosArrayDestroy(pFailedChkptList);	4,272✔
548	taosArrayDestroy(pOrphanTasks);	4,272✔
549	}	4,272✔
550
551	void doSendHbMsgRsp(int32_t code, SRpcHandleInfo *pRpcInfo, int32_t vgId, int32_t msgId) {	4,272✔
552	SRpcMsg rsp = {.code = code, .info = *pRpcInfo, .contLen = sizeof(SMStreamHbRspMsg)};	4,272✔
553	rsp.pCont = rpcMallocCont(rsp.contLen);	4,272✔
554
555	SMStreamHbRspMsg *pMsg = rsp.pCont;	4,272✔
556	pMsg->head.vgId = htonl(vgId);	4,272✔
557	pMsg->msgId = msgId;	4,272✔
558
559	tmsgSendRsp(&rsp);	4,272✔
560	pRpcInfo->handle = NULL; // disable auto rsp	4,272✔
561	}	4,272✔
562
563	void checkforOrphanTask(SMnode* pMnode, STaskStatusEntry* p, SArray* pOrphanTasks) {	44✔
564	SStreamObj *pStream = NULL;	44✔
565
566	int32_t code = mndGetStreamObj(pMnode, p->id.streamId, &pStream);	44✔
567	if (code) {	44!
568	mError("stream:0x%" PRIx64 " not exists, s-task:0x%" PRIx64 " not found in task list, add into orphan list",	×
569	p->id.streamId, p->id.taskId);
570
571	SOrphanTask oTask = {.streamId = p->id.streamId, .taskId = p->id.taskId, .nodeId = p->nodeId};	×
572	void *px = taosArrayPush(pOrphanTasks, &oTask);	×
573	if (px == NULL) {	×
574	mError("failed to put task into orphan list, taskId:0x%" PRIx64", code:%s", p->id.taskId, tstrerror(terrno));	×
575	}
576	} else {
577	if (pStream != NULL) {	44!
578	mndReleaseStream(pMnode, pStream);	44✔
579	}
580
581	mError("s-task:0x%" PRIx64 " not found in task list but exists in mnode meta, data inconsistent, not drop yet",	44!
582	p->id.taskId);
583	}
584	}	44✔

taosdata / TDengine / #3530

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous