#4991

Committed 17 Mar 2026 07:57AM UTC coverage: 69.756% (+0.4%) from 69.348%

Build # #4991

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch #34807

Run Details

14 of 16 new or added lines in 5 files covered. (87.5%)

3928 existing lines in 138 files now uncovered.

192146 of 275455 relevant lines covered (69.76%)

137208686.18 hits per line

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

45.27

/source/dnode/vnode/src/tsdb/tsdbMigrate.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 "tss.h"
#include "tsdb.h"
#include "tsdbFS2.h"
#include "tsdbFSet2.h"
#include "vnd.h"
#include "tsdbInt.h"

extern int32_t tsdbAsyncCompact(STsdb* tsdb, const STimeWindow* tw, ETsdbOpType type, bool force);

// migrate monitor related functions
typedef struct SSsMigrateMonitor {
  TdThreadCond  stateChanged;
  int32_t ssMigrateId;
  int32_t fid;
  int32_t state;
  int64_t startTimeSec;
} SSsMigrateMonitor;


static int32_t getSsMigrateId(STsdb* tsdb) {
  return tsdb->pSsMigrateMonitor->ssMigrateId;
}


int32_t tsdbOpenSsMigrateMonitor(STsdb *tsdb) {
  SSsMigrateMonitor* pmm = (SSsMigrateMonitor*)taosMemCalloc(1, sizeof(SSsMigrateMonitor));
  if (pmm == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  TAOS_UNUSED(taosThreadCondInit(&pmm->stateChanged, NULL));
  tsdb->pSsMigrateMonitor = pmm;
  return 0;
}


void tsdbCloseSsMigrateMonitor(STsdb *tsdb) {
  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;
  if (pmm == NULL) {
    return;
  }

  TAOS_UNUSED(taosThreadCondDestroy(&pmm->stateChanged));
  taosMemoryFree(tsdb->pSsMigrateMonitor);
  tsdb->pSsMigrateMonitor = NULL;
}


static void setMigrationState(STsdb* tsdb, int32_t state) {
  (void)taosThreadMutexLock(&tsdb->mutex);
  tsdb->pSsMigrateMonitor->state = state;
  (void)taosThreadMutexUnlock(&tsdb->mutex);
}


int32_t tsdbQuerySsMigrateProgress(STsdb *tsdb, SSsMigrateProgress *pProgress) {
  int32_t code = 0, vid = TD_VID(tsdb->pVnode);

  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;

  (void)taosThreadMutexLock(&tsdb->mutex);

  if (pmm->ssMigrateId != pProgress->ssMigrateId) {
    tsdbError("vgId:%d, ssMigrateId:%d, fid:%d, migrate id mismatch in query progress, actual %d",
              vid, pProgress->ssMigrateId, pProgress->fid, pmm->ssMigrateId);
    code = TSDB_CODE_INVALID_MSG;
  } else if (pmm->fid != pProgress->fid) {
    tsdbError("vgId:%d, ssMigrateId:%d, fid:%d, file set id mismatch in query progress, actual %d",
              vid, pProgress->ssMigrateId, pProgress->fid, pmm->fid);
    code = TSDB_CODE_INVALID_MSG;
  } else {
    pProgress->state = pmm->state;
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);

  return code;
}



int32_t tsdbUpdateSsMigrateProgress(STsdb* tsdb, SSsMigrateProgress* pProgress) {
  int32_t vid = TD_VID(tsdb->pVnode), code = 0;
  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;

  // the state was generated by this vnode, so no need to process it
  if (pProgress->nodeId == vnodeNodeId(tsdb->pVnode)) {
    tsdbDebug("vgId:%d, skip migration progress update since it was generated by this vnode", vid);
    return 0;
  }

  (void)taosThreadMutexLock(&tsdb->mutex);

  if (pmm->ssMigrateId != pProgress->ssMigrateId) {
    tsdbError("vgId:%d, ssMigrateId:%d, fid:%d, migrate id mismatch in update progress, actual %d",
              vid, pProgress->ssMigrateId, pProgress->fid, pmm->ssMigrateId);
    code = TSDB_CODE_INVALID_MSG;
  } else if (pmm->fid != pProgress->fid) {
    tsdbError("vgId:%d, ssMigrateId:%d, fid:%d, file set id mismatch in update progress, actual %d",
              vid, pProgress->ssMigrateId, pProgress->fid, pmm->fid);
    code = TSDB_CODE_INVALID_MSG;
  } else {
    // update the state, and broadcast state change message, to avoid the timeout of
    // the waiting thread, we should always broadcast the message even if the state
    // is not changed actually.
    pmm->state = pProgress->state;
    (void)taosThreadCondBroadcast(&pmm->stateChanged);
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);

  return code;
}



// migrate file related functions
int32_t tsdbSsFidLevel(int32_t fid, STsdbKeepCfg *pKeepCfg, int32_t ssKeepLocal, int64_t nowSec) {
  int32_t localFid;
  TSKEY   key;

  if (pKeepCfg->precision == TSDB_TIME_PRECISION_MILLI) {
    nowSec = nowSec * 1000;
  } else if (pKeepCfg->precision == TSDB_TIME_PRECISION_MICRO) {
    nowSec = nowSec * 1000000l;
  } else if (pKeepCfg->precision == TSDB_TIME_PRECISION_NANO) {
    nowSec = nowSec * 1000000000l;
  }

  nowSec = nowSec - pKeepCfg->keepTimeOffset * tsTickPerHour[pKeepCfg->precision];

  key = nowSec - ssKeepLocal * tsTickPerMin[pKeepCfg->precision];
  localFid = tsdbKeyFid(key, pKeepCfg->days, pKeepCfg->precision);

  return fid >= localFid ? 0 : 1;
}


static int32_t downloadManifest(SVnode* pVnode, int32_t fid, STFileSet** ppFileSet) {
  int32_t code = 0, vid = TD_VID(pVnode);

  char path[64];
  snprintf(path, sizeof(path), "vnode%d/f%d/manifests.json", vid, fid);
  int64_t size = 0;
  code = tssGetFileSizeOfDefault(path, &size);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to get manifests size since %s", vid, fid, tstrerror(code));
    return code;
  }

  char* buf = taosMemoryMalloc(size + 1);
  code = tssReadFileFromDefault(path, 0, buf, &size);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to read manifest from shared storage since %s", vid, fid, tstrerror(code));
    taosMemoryFree(buf);
    return code;
  }
  buf[size] = 0;

  cJSON* json = cJSON_Parse(buf);
  taosMemoryFree(buf);
  if (json == NULL) {
    tsdbError("vgId:%d, fid:%d, failed to parse manifest json since %s", vid, fid, tstrerror(code));
    return TSDB_CODE_FILE_CORRUPTED;
  }
  
  code = tsdbJsonToTFileSet(pVnode->pTsdb, json, ppFileSet);
  cJSON_Delete(json);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to parse manifest since %s", vid, fid, tstrerror(code));
    return code;
  }

  STFileSet* fset = *ppFileSet;
  if (fset->fid != fid) {
    tsdbError("vgId:%d, fid:%d, mismatch fid, manifest fid is %d", vid, fid, fset->fid);
    tsdbTFileSetClear(ppFileSet);
    return TSDB_CODE_FILE_CORRUPTED;
  }
  if (fset->farr[TSDB_FTYPE_DATA] == NULL) {
    tsdbError("vgId:%d, fid:%d, data file not found in manifest", vid, fid);
    tsdbTFileSetClear(ppFileSet);
    return TSDB_CODE_FILE_CORRUPTED;
  }

  return code;
}


static int32_t uploadManifest(int32_t dnode, int32_t vnode, STFileSet* fset, int32_t mid) {
  int32_t code = 0;

  cJSON* json = cJSON_CreateObject();
  if (json == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  // update migration id for all files in the file set
  STFileObj* fobj = fset->farr[TSDB_FTYPE_HEAD];
  fobj->f->mid = mid;
  fobj = fset->farr[TSDB_FTYPE_SMA];
  fobj->f->mid = mid;
  fobj = fset->farr[TSDB_FTYPE_DATA];
  fobj->f->mid = mid;
  fobj = fset->farr[TSDB_FTYPE_TOMB];
  if (fobj != NULL) {
    fobj->f->mid = mid;
  }

  SSttLvl* lvl;
  TARRAY2_FOREACH(fset->lvlArr, lvl) {
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      fobj->f->mid = mid;
    }
  }
  
  if (cJSON_AddNumberToObject(json, "fmtv", 1) == NULL) {
    code = terrno;
    cJSON_Delete(json);
    return code;
  }
  if (cJSON_AddNumberToObject(json, "dnode", dnode) == NULL) {
    code = terrno;
    cJSON_Delete(json);
    return code;
  }
  if (cJSON_AddNumberToObject(json, "vnode", vnode) == NULL) {
    code = terrno;
    cJSON_Delete(json);
    return code;
  }

  code = tsdbTFileSetToJson(fset, json);
  if (code != TSDB_CODE_SUCCESS) {
    cJSON_Delete(json);
    return code;
  }
  char* buf = cJSON_PrintUnformatted(json);
  cJSON_Delete(json);

  char path[64];
  snprintf(path, sizeof(path), "vnode%d/f%d/manifests.json", vnode, fset->fid);
  code = tssUploadToDefault(path, buf, strlen(buf));
  taosMemoryFree(buf);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to upload manifest since %s", vnode, fset->fid, tstrerror(code));
    return code;
  }

  return code;
}


// upload local files to shared storage
//
// local file name is like:
//     [base]/vnode2/f1736/v2f1736ver16.head
// or
//     [base]/vnode2/f1736/v2f1736ver16.m334233.head
//
// remote file name is like:
//     vnode2/f1736/v2f1736ver16.m13552343.head
//
// NOTE: the migration id is always included in the remote file name, because
// the commit id may be different between the vnodes of the same vgroup,
// that's an interrupted migration may overwrite some of the remote files
// while leaving the others intact if we don't include the migration id in
// the remote file name.
static int32_t uploadFile(SRTNer* rtner, STFileObj* fobj) {
  if (fobj == NULL) {
    return TSDB_CODE_SUCCESS;
  }

  const char* ext = strrchr(fobj->fname, '.');
  int32_t vid = TD_VID(rtner->tsdb->pVnode), mid = getSsMigrateId(rtner->tsdb);
  STFile* f = fobj->f;
  
  char path[TSDB_FILENAME_LEN];
  snprintf(path, sizeof(path), "vnode%d/f%d/v%df%dver%" PRId64 ".m%d%s", vid, f->fid, vid, f->fid, f->cid, mid, ext);

  int code = tssUploadFileToDefault(path, fobj->fname, 0, -1);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to upload file %s since %s", vid, f->fid, fobj->fname, tstrerror(code));
    return code;
  }

  return 0;
}



static int32_t downloadFile(SRTNer* rtner, STFileObj* fobj) {
  if (fobj == NULL) {
    return TSDB_CODE_SUCCESS;
  }

  const char* fname = strrchr(fobj->fname, TD_DIRSEP_CHAR) + 1;
  int32_t vid = TD_VID(rtner->tsdb->pVnode);
  STFile* f = fobj->f;
  
  char path[TSDB_FILENAME_LEN];
  snprintf(path, sizeof(path), "vnode%d/f%d/%s", vid, f->fid, fname);
  int code = tssDownloadFileFromDefault(path, fobj->fname, 0, -1);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to download file %s since %s", vid, f->fid, path, tstrerror(code));
    return code;
  }

  return 0;
}


static void tsdbRemoveSsGarbageFiles(int32_t vid, STFileSet* fset) {
  char prefix[TSDB_FILENAME_LEN];
  snprintf(prefix, sizeof(prefix), "vnode%d/f%d/", vid, fset->fid);

  SArray* paths = taosArrayInit(10, sizeof(char*));
  int32_t code = tssListFileOfDefault(prefix, paths);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to list files in shared storage since %s", vid, fset->fid, tstrerror(code));
    taosArrayDestroy(paths);
    return;
  }

  for(int i = 0; i < taosArrayGetSize(paths); i++) {
      char* p = *(char**)taosArrayGet(paths, i);
      const char* ext = strrchr(p, '.');
      const char* rname = strrchr(p, '/') + 1;
      bool remove = true;
      int32_t vgId = 0, fid = 0, mid = 0, cn = 0;
      int64_t cid = 0;
  
      // NOTE: when compare file name, don't use strcmp(fobj->fname, rname) because 'fobj->fname' may not
      // contain the migration id. that's why we use sscanf to parse the file name.

      if (ext == NULL) {
        // no extension, remove
      } else if (taosStrcasecmp(ext, ".head") == 0) {
        STFileObj* fobj = fset->farr[TSDB_FTYPE_HEAD];
        int n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.head", &vgId, &fid, &cid, &mid);
        remove = (n != 4 || vgId != vid || fid != fset->fid || cid != fobj->f->cid || mid != fobj->f->mid);
      } else if (taosStrcasecmp(ext, ".sma") == 0) {
        STFileObj* fobj = fset->farr[TSDB_FTYPE_SMA];
        int n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.sma", &vgId, &fid, &cid, &mid);
        remove = (n != 4 || vgId != vid || fid != fset->fid || cid != fobj->f->cid || mid != fobj->f->mid);
      } else if (taosStrcasecmp(ext, ".tomb") == 0) {
        STFileObj* fobj = fset->farr[TSDB_FTYPE_TOMB];
        if (fobj) {
          int n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.tomb", &vgId, &fid, &cid, &mid);
          remove = (n != 4 || vgId != vid || fid != fset->fid || cid != fobj->f->cid || mid != fobj->f->mid);
        }
      } else if (taosStrcasecmp(ext, ".stt") == 0) {
        SSttLvl* lvl = NULL;
        TARRAY2_FOREACH(fset->lvlArr, lvl) {
          STFileObj* fobj;
          TARRAY2_FOREACH(lvl->fobjArr, fobj) {
            int n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.stt", &vgId, &fid, &cid, &mid);
            if (n == 4 && vgId == vid && fid == fset->fid && cid == fobj->f->cid && mid == fobj->f->mid) {
              remove = false;
              break;
            }
          }
          if (!remove) {
            break;
          }
        }
      } else if (taosStrcasecmp(ext, ".data") == 0) {
        STFileObj* fobj = fset->farr[TSDB_FTYPE_DATA];
        int n = sscanf(rname, "v%df%dver%" PRId64 ".%d.data", &vgId, &fid, &cid, &cn);
        if (n == 4) {
          if (vgId == vid && fid == fset->fid && cid == fobj->f->cid && cn >= 1 && cn < fobj->f->lcn) {
            remove = false; // not the last chunk, keep it
          }
        } else {
          n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.%d.data", &vgId, &fid, &cid, &mid, &cn);
          remove = (n != 5 || vgId != vid || fid != fset->fid || cid != fobj->f->cid || mid != fobj->f->mid || cn != fobj->f->lcn);
        }
      } else {
        remove = (taosStrcasecmp(rname, "manifests.json") != 0); // keep manifest, remove all other files
      }

      if (remove) {
        int32_t code = tssDeleteFileFromDefault(p);
        if (code != TSDB_CODE_SUCCESS) {
          tsdbError("vgId:%d, fid:%d, failed to remove garbage file %s from shared storage since %s", vid, fset->fid, p, tstrerror(code));
        } else {
          tsdbInfo("vgId:%d, fid:%d, garbage file %s is removed from shared storage", vid, fset->fid, p);
        }
      }

      taosMemoryFree(p);
  }

  taosArrayDestroy(paths);
}


// download the last chunk of a data file
// remote file name is like:
//      vnode2/f1736/v2f1736ver16.m13552343.4.data
static int32_t downloadDataFileLastChunk(SRTNer* rtner, STFileObj* fobj) {
  int32_t code = 0;
  int32_t vid = TD_VID(rtner->tsdb->pVnode);
  SVnodeCfg *pCfg = &rtner->tsdb->pVnode->config;
  STFile *f = fobj->f;

  char lpath[TSDB_FILENAME_LEN], rpath[TSDB_FILENAME_LEN];
  tsdbTFileLastChunkName(rtner->tsdb, f, lpath);
  char* fname = strrchr(lpath, TD_DIRSEP_CHAR) + 1;

  snprintf(rpath, sizeof(rpath), "vnode%d/f%d/%s", vid, f->fid, fname);

  code = tssDownloadFileFromDefault(rpath, lpath, 0, -1);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to download data file %s since %s", vid, f->fid, rpath, tstrerror(code));
    return code;
  }

  return code;
}


// while other files all include the migration id in the remote file name, only the last
// chunk of a data file does the same. this is ok because:
// 1. without a compaction, data file is always uploaded chunk by chunk, only the last
//    chunk may be modified.
// 2. after a compaction, all of the data is downloaded to local, so overwriting remote
//    data chunks won't cause any problem. (this is not likely to happen because we will
//    cancel the migration in this case, refer comment in function shouldMigrate).
static int32_t uploadDataFile(SRTNer* rtner, STFileObj* fobj) {
  int32_t code = 0;
  int32_t vid = TD_VID(rtner->tsdb->pVnode), mid = getSsMigrateId(rtner->tsdb);
  SVnodeCfg *pCfg = &rtner->tsdb->pVnode->config;
  int64_t szFile = 0, szChunk = (int64_t)pCfg->tsdbPageSize * pCfg->ssChunkSize;
  STFile *f = fobj->f;
  STFileOp op = {.optype = TSDB_FOP_MODIFY, .fid = f->fid, .of = *f};

  char path[TSDB_FILENAME_LEN];
  if (f->lcn <= 1) {
    tstrncpy(path, fobj->fname, sizeof(path));
  } else {
    tsdbTFileLastChunkName(rtner->tsdb, f, path);
  }

  code = taosStatFile(path, &szFile, NULL, NULL);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d failed to stat file %s since %s", vid, f->fid, path, tstrerror(code));
    return false;
  }

  if (f->lcn > 1) {
    szFile += szChunk * (f->lcn - 1); // add the size of migrated chunks
  }

  int totalChunks = szFile / szChunk;
  if (szFile % szChunk) {
    totalChunks++;
  }

  int lcn = f->lcn < 1 ? 1 : f->lcn;

  // upload chunks one by one, the first chunk may already been uploaded, but may be
  // modified thereafter, so we need to upload it again
  for (int i = lcn; i <= totalChunks; ++i) {
    int64_t offset = (int64_t)(i - lcn) * szChunk;
    int64_t size = szChunk;
    if (i == totalChunks && szFile % szChunk) {
        size = szFile % szChunk;
    }

    // only include the migration id in the last chunk filename
    char rpath[TSDB_FILENAME_LEN];
    if (i == totalChunks) {
        snprintf(rpath, sizeof(rpath), "vnode%d/f%d/v%df%dver%" PRId64 ".m%d.%d.data", vid, f->fid, vid, f->fid, f->cid,
                 mid, i);
    } else {
        snprintf(rpath, sizeof(rpath), "vnode%d/f%d/v%df%dver%" PRId64 ".%d.data", vid, f->fid, vid, f->fid, f->cid, i);
    }

    code = tssUploadFileToDefault(rpath, path, offset, size);
    if (code != TSDB_CODE_SUCCESS) {
      tsdbError("vgId:%d, fid:%d, failed to migrate data file since %s", vid, f->fid, tstrerror(code));
      return code;
    }
  }

  f->lcn = totalChunks;
  op.nf = *f;
  code = TARRAY2_APPEND(&rtner->fopArr, op);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to append file operation since %s", vid, f->fid, tstrerror(code));
    return code;
  }

  // manifest must be uploaded before copy last chunk, otherwise, failed to upload manifest
  // will result in a broken migration
  tsdbInfo("vgId:%d, fid:%d, data file migrated, begin generate & upload manifest", vid, f->fid);

  // manifest, this also commit the migration
  code = uploadManifest(vnodeNodeId(rtner->tsdb->pVnode), vid, rtner->fset, getSsMigrateId(rtner->tsdb));
  if (code != TSDB_CODE_SUCCESS) {
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, manifest uploaded, begin remove garbage files", vid, f->fid);
  tsdbRemoveSsGarbageFiles(vid, rtner->fset);

  setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_SUCCEEDED);
  tsdbInfo("vgId:%d, fid:%d, leader migration succeeded", vid, f->fid);

  // no new chunks generated, no need to copy the last chunk
  if (totalChunks == lcn) {
    return 0;
  }

  // copy the last chunk to the new file
  char newPath[TSDB_FILENAME_LEN];
  tsdbTFileLastChunkName(rtner->tsdb, &op.nf, newPath);

  int64_t offset = (int64_t)(totalChunks - lcn) * szChunk;
  int64_t size = szChunk;
  if (szFile % szChunk) {
    size = szFile % szChunk;
  }

  TdFilePtr fdFrom = taosOpenFile(path, TD_FILE_READ);
  if (fdFrom == NULL) {
    code = terrno;
    tsdbError("vgId:%d, fid:%d, failed to open source file %s since %s", vid, f->fid, path, tstrerror(code));
    return code;
  }

  TdFilePtr fdTo = taosOpenFile(newPath, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (fdTo == NULL) {
    code = terrno;
    tsdbError("vgId:%d, fid:%d, failed to open target file %s since %s", vid, f->fid, newPath, tstrerror(code));
    TAOS_UNUSED(taosCloseFile(&fdFrom));
    return code;
  }

  int64_t n = taosFSendFile(fdTo, fdFrom, &offset, size);
  if (n < 0) {
    code = terrno;
    tsdbError("vgId:%d, fid:%d, failed to copy file %s to %s since %s", vid, f->fid, path, newPath, tstrerror(code));
  }
  TAOS_UNUSED(taosCloseFile(&fdFrom));
  TAOS_UNUSED(taosCloseFile(&fdTo));

  return code;
}


static bool shouldMigrate(SRTNer *rtner, int32_t *pCode) {
  int32_t vid = TD_VID(rtner->tsdb->pVnode);
  SVnodeCfg *pCfg = &rtner->tsdb->pVnode->config;
  STFileSet *pLocalFset = rtner->fset;
  STFileObj *flocal = pLocalFset->farr[TSDB_FTYPE_DATA];

  *pCode = 0;
  if (!flocal) {
    tsdbInfo("vgId:%d, fid:%d, migration cancelled, local data file not exist", vid, pLocalFset->fid);
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_SKIPPED);
    return false;
  }

  if (rtner->lastCommit != pLocalFset->lastCommit) {
    tsdbInfo("vgId:%d, fid:%d, migration cancelled, there are new commits after migration task is scheduled", vid, pLocalFset->fid);
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_SKIPPED);
    return false;
  }

  if (pCfg->ssCompact && flocal->f->lcn < 0) {
    int32_t     lcn = flocal->f->lcn;
    STimeWindow win = {0};
    tsdbFidKeyRange(pLocalFset->fid, rtner->tsdb->keepCfg.days, rtner->tsdb->keepCfg.precision, &win.skey, &win.ekey);
    *pCode = tsdbAsyncCompact(rtner->tsdb, &win, TSDB_OPTR_SSMIGRATE, false);
    tsdbInfo("vgId:%d, fid:%d, migration cancelled, fileset need compact, lcn: %d", vid, pLocalFset->fid, lcn);
    if (*pCode) {
      setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    } else {
      setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_COMPACT);
    }
    return false; // compact in progress
  }

  char path[TSDB_FILENAME_LEN];
  if (flocal->f->lcn <= 1) {
    tstrncpy(path, flocal->fname, sizeof(path));
  } else {
    tsdbTFileLastChunkName(rtner->tsdb, flocal->f, path);
  }

  int64_t size = 0, mtime = 0;
  *pCode = taosStatFile(path, &size, &mtime, NULL);
  if (*pCode != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, migration cancelled, failed to stat file %s since %s", vid, pLocalFset->fid, path, tstrerror(*pCode));
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return false;
  }

  // file may become smaller after a compaction, especially after a RSMA compaction
  if (flocal->f->lcn < 1 && size <= (int64_t)pCfg->tsdbPageSize * pCfg->ssChunkSize) {
    tsdbInfo("vgId:%d, fid:%d, migration skipped, data file is too small, size: %" PRId64 " bytes", vid, flocal->f->fid, size);
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_SKIPPED);
    return false;
  }

  // 'mtime >= rtner->tw.ekey - tsSsUploadDelaySec' means the file is active writing, and we should skip
  // the migration. However, this may also be a result of the [ssCompact] option, which should not
  // be skipped, so we also check 'mtime > pLocalFset->lastCompact / 1000 || !pCfg->ssCompact', note
  // this is not an acurate condition, but is simple and good enough.
  if (mtime >= rtner->tw.ekey - tsSsUploadDelaySec && (mtime > pLocalFset->lastCompact / 1000 || !pCfg->ssCompact)) {
    tsdbInfo("vgId:%d, fid:%d, migration skipped, data file is active writting, modified at %" PRId64, vid, pLocalFset->fid, mtime);
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_SKIPPED);
    return false; // still active writing, postpone migration
  }

  // download manifest from shared storage
  STFileSet *pRemoteFset = NULL;
  *pCode = downloadManifest(rtner->tsdb->pVnode, pLocalFset->fid, &pRemoteFset);
  if (*pCode == TSDB_CODE_SUCCESS) {
    // remote file exists but local file has not been migrated, there are two possibilities:
    // 1. there's a compact after the last migration, this is a normal case, we can discard
    //    the remote files and continue the migration;
    // 2. in the last migration, this node was a follower, the leader did its job successfully,
    //    but this node did not, continue the migration may overwrite remote file and result in
    //    data corruption on other nodes.
    // however, it is hard to distinguish them, so just treat both as a migration error. hope
    // the user could do something to recover, such as remove remote files.
    if (flocal->f->lcn < 1) {
      tsdbTFileSetClear(&pRemoteFset);
      tsdbError("vgId:%d, fid:%d, migration cancelled, remote manifest found but local lcn < 1", vid, pLocalFset->fid);
      *pCode = TSDB_CODE_FILE_CORRUPTED;
      setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
      return false;
    }

  } else if (*pCode == TSDB_CODE_NOT_FOUND) {
    if (flocal->f->lcn >= 1) {
      tsdbError("vgId:%d, fid:%d, migration cancelled, remote manifest not found but local lcn >= 1", vid, pLocalFset->fid);
      *pCode = TSDB_CODE_FILE_CORRUPTED;
      setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
      return false;
    }

    // this is the first migration, we should do it.
    *pCode = TSDB_CODE_SUCCESS;
    return true;

  } else {
    tsdbError("vgId:%d, fid:%d, migration cancelled, failed to download manifest, code: %d", vid, pLocalFset->fid, *pCode);
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return false;
  }

  STFileObj *fremote = pRemoteFset->farr[TSDB_FTYPE_DATA];
  if (fremote == NULL) {
    tsdbError("vgId:%d, fid:%d, migration cancelled, cannot find data file information from remote manifest", vid, pLocalFset->fid);
    tsdbTFileSetClear(&pRemoteFset);
    *pCode = TSDB_CODE_FILE_CORRUPTED;
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return false;
  }

  if (fremote->f->lcn != flocal->f->lcn) {
    tsdbError("vgId:%d, fid:%d, migration cancelled, remote and local data file information mismatch", vid, pLocalFset->fid);
    tsdbTFileSetClear(&pRemoteFset);
    *pCode = TSDB_CODE_FILE_CORRUPTED;
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return false;
  }
  
  if (fremote->f->maxVer == flocal->f->maxVer) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbInfo("vgId:%d, fid:%d, migration skipped, no new data", vid, pLocalFset->fid);
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_SKIPPED);
    return false; // no new data
  }

  tsdbTFileSetClear(&pRemoteFset); // we use the local file set information for migration
  tsdbInfo("vgId:%d, fid:%d, file set will be migrated", vid, pLocalFset->fid);
  return true;
}


static int32_t tsdbFollowerDoSsMigrate(SRTNer *rtner) {
  int32_t code = 0, vid = TD_VID(rtner->tsdb->pVnode);
  STFileSet *fset = rtner->fset;
  SSsMigrateMonitor* pmm = rtner->tsdb->pSsMigrateMonitor;
  int32_t fsIdx = 0;

  // though we make this check in the leader node, we should do this in the follower nodes too.
  // because there may be a leader change and the execution order of async tasks may result in
  // different commit time. if we don't do this, we may corrupt the follower data.
  if (rtner->lastCommit != fset->lastCommit) {
    tsdbInfo("vgId:%d, fid:%d, follower migration cancelled, there are new commits after migration is scheduled", vid, fset->fid);
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_SKIPPED);
    return 0;
  }

  tsdbInfo("vgId:%d, fid:%d, vnode is follower, waiting leader on node %d to upload.", vid, fset->fid, rtner->nodeId);

  (void)taosThreadMutexLock(&rtner->tsdb->mutex);

  while(pmm->state == SSMIGRATE_FILESET_STATE_IN_PROGRESS) {
    struct timespec ts;
    if ((code = taosClockGetTime(CLOCK_REALTIME, &ts)) != TSDB_CODE_SUCCESS) {
      tsdbError("vgId:%d, fid:%d, failed to get current time since %s", vid, fset->fid, tstrerror(code));
      TAOS_UNUSED(taosThreadMutexUnlock(&rtner->tsdb->mutex));
      return code;
    }
    ts.tv_sec += 30; // TODO: make it configurable
    code = taosThreadCondTimedWait(&pmm->stateChanged, &rtner->tsdb->mutex, &ts);
    if (code == TSDB_CODE_TIMEOUT_ERROR) {
      tsdbError("vgId:%d, fid:%d, waiting leader migration timed out", vid, fset->fid);
      pmm->state = SSMIGRATE_FILESET_STATE_FAILED;
    }
  }

  if (pmm->state != SSMIGRATE_FILESET_STATE_SUCCEEDED) {
    TAOS_UNUSED(taosThreadMutexUnlock(&rtner->tsdb->mutex));
    tsdbInfo("vgId:%d, fid:%d, follower migration skipped because leader migration skipped or failed", vid, fset->fid);
    return 0;
  }

  TAOS_UNUSED(taosThreadMutexUnlock(&rtner->tsdb->mutex));

  // NOTE: After the leader node finished processing the current file set and mnode sent the final
  // follower migrate request, mnode waits at least 30 seconds before triggering the processing of
  // the next file set. Because all file sets of a vgroup shares the same [pmm], from now on, the
  // follower should not access [pmm->state] anymore.
  // refer comments in mndUpdateSsMigrateProgress for more details.

  tsdbInfo("vgId:%d, fid:%d, follower migration started, begin downloading manifest...", vid, fset->fid);
  STFileSet *pRemoteFset = NULL;
  code = downloadManifest(rtner->tsdb->pVnode, fset->fid, &pRemoteFset);
  if (code == TSDB_CODE_NOT_FOUND) {
    tsdbTFileSetClear(&pRemoteFset);
    return TSDB_CODE_FILE_CORRUPTED;
  }

  // this often happens in the catch up process of a new node, it is ok to continue, but will
  // result in download same files more than once, which is a waste of time and bandwidth.
  if (pRemoteFset->farr[TSDB_FTYPE_HEAD]->f->mid != getSsMigrateId(rtner->tsdb)) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbWarn("vgId:%d, fid:%d, follower migration cancelled, migration id mismatch", vid, fset->fid);
    return 0;
  }

  tsdbInfo("vgId:%d, fid:%d, manifest downloaded, begin downloading head file", vid, fset->fid);
  code = downloadFile(rtner, pRemoteFset->farr[TSDB_FTYPE_HEAD]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    return code;
  }
  STFileOp op = {.optype = TSDB_FOP_MODIFY, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_HEAD]->f, .nf = *pRemoteFset->farr[TSDB_FTYPE_HEAD]->f};
  code = TARRAY2_APPEND(&rtner->fopArr, op);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, failed to append head file operation since %s", vid, fset->fid, tstrerror(code));
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, head file downloaded, begin downloading sma file", vid, fset->fid);
  code = downloadFile(rtner, pRemoteFset->farr[TSDB_FTYPE_SMA]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    return code;
  }
  op = (STFileOp) {.optype = TSDB_FOP_MODIFY, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_SMA]->f, .nf = *pRemoteFset->farr[TSDB_FTYPE_SMA]->f};
  code = TARRAY2_APPEND(&rtner->fopArr, op);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, failed to append sma file operation since %s", vid, fset->fid, tstrerror(code));
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, sma file downloaded, begin downloading tomb file", vid, fset->fid);
  code = downloadFile(rtner, pRemoteFset->farr[TSDB_FTYPE_TOMB]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    return code;
  }
  if (fset->farr[TSDB_FTYPE_TOMB] != NULL && pRemoteFset->farr[TSDB_FTYPE_TOMB] != NULL) {
    op = (STFileOp) {.optype = TSDB_FOP_MODIFY, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_TOMB]->f, .nf = *pRemoteFset->farr[TSDB_FTYPE_TOMB]->f};
    code = TARRAY2_APPEND(&rtner->fopArr, op);
  } else if (fset->farr[TSDB_FTYPE_TOMB] != NULL) {
    // the remote tomb file is not found, but local tomb file exists, we should remove it
    op = (STFileOp) {.optype = TSDB_FOP_REMOVE, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_TOMB]->f};
    code = TARRAY2_APPEND(&rtner->fopArr, op);
  } else if (pRemoteFset->farr[TSDB_FTYPE_TOMB] != NULL) {
    op = (STFileOp) {.optype = TSDB_FOP_CREATE, .fid = fset->fid, .nf = *pRemoteFset->farr[TSDB_FTYPE_TOMB]->f};
    code = TARRAY2_APPEND(&rtner->fopArr, op);
  }
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, failed to append tomb file operation since %s", vid, fset->fid, tstrerror(code));
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, tomb file downloaded, begin downloading stt files", vid, fset->fid);
  SSttLvl* lvl;
  TARRAY2_FOREACH(fset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      op = (STFileOp) {.optype = TSDB_FOP_REMOVE, .fid = fset->fid, .of = *fobj->f};
      code = TARRAY2_APPEND(&rtner->fopArr, op);
      if (code != TSDB_CODE_SUCCESS) {
        tsdbTFileSetClear(&pRemoteFset);
        tsdbError("vgId:%d, fid:%d, failed to append stt file remove operation since %s", vid, fset->fid, tstrerror(code));
        return code;
      }
    }
  }
  TARRAY2_FOREACH(pRemoteFset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      code = downloadFile(rtner, fobj);
      if (code != TSDB_CODE_SUCCESS) {
        tsdbTFileSetClear(&pRemoteFset);
        return code;
      }
      op = (STFileOp) {.optype = TSDB_FOP_CREATE, .fid = fset->fid, .nf = *fobj->f};
      code = TARRAY2_APPEND(&rtner->fopArr, op);
      if (code != TSDB_CODE_SUCCESS) {
        tsdbTFileSetClear(&pRemoteFset);
        tsdbError("vgId:%d, fid:%d, failed to append stt file create operation since %s", vid, fset->fid, tstrerror(code));
        return code;
      }
    }
  }

  tsdbInfo("vgId:%d, fid:%d, stt files downloaded, begin downloading data file", vid, fset->fid);
  code = downloadDataFileLastChunk(rtner, pRemoteFset->farr[TSDB_FTYPE_DATA]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    return code;
  }
  op = (STFileOp) {.optype = TSDB_FOP_MODIFY, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_DATA]->f, .nf = *pRemoteFset->farr[TSDB_FTYPE_DATA]->f};
  code = TARRAY2_APPEND(&rtner->fopArr, op);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, failed to append data file operation since %s", vid, fset->fid, tstrerror(code));
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, data file downloaded", vid, fset->fid);
  tsdbTFileSetClear(&pRemoteFset);
  return 0;
}


static int32_t tsdbLeaderDoSsMigrate(SRTNer *rtner) {
  int32_t code = 0, vid = TD_VID(rtner->tsdb->pVnode);
  SVnodeCfg *pCfg = &rtner->tsdb->pVnode->config;
  STFileSet *fset = rtner->fset;

  tsdbInfo("vgId:%d, fid:%d, vnode is leader, migration started", vid, fset->fid);

  if (!shouldMigrate(rtner, &code)) {
    return code;
  }

  // head file
  tsdbInfo("vgId:%d, fid:%d, begin migrate head file", vid, fset->fid);
  code = uploadFile(rtner, fset->farr[TSDB_FTYPE_HEAD]);
  if (code != TSDB_CODE_SUCCESS) {
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, head file migrated, begin migrate sma file", vid, fset->fid);

  // sma file
  code = uploadFile(rtner, fset->farr[TSDB_FTYPE_SMA]);
  if (code != TSDB_CODE_SUCCESS) {
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, sma file migrated, begin migrate tomb file", vid, fset->fid);

  // tomb file
  code = uploadFile(rtner, fset->farr[TSDB_FTYPE_TOMB]);
  if (code != TSDB_CODE_SUCCESS) {
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, tomb file migrated, begin migrate stt files", vid, fset->fid);

  // stt files
  SSttLvl* lvl;
  TARRAY2_FOREACH(fset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      code = uploadFile(rtner, fobj);
      if (code != TSDB_CODE_SUCCESS) {
        setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
        return code;
      }
    }
  }
  
  tsdbInfo("vgId:%d, fid:%d, stt files migrated, begin migrate data file", vid, fset->fid);

  // data file
  code = uploadDataFile(rtner, fset->farr[TSDB_FTYPE_DATA]);
  if (code != TSDB_CODE_SUCCESS) {
    setMigrationState(rtner->tsdb, SSMIGRATE_FILESET_STATE_FAILED);
    return code;
  }

  return TSDB_CODE_SUCCESS;
}


int32_t tsdbDoSsMigrate(SRTNer *rtner) {
  // note: leader is decided when the task is scheduled, the actual leader may change after that,
  // but this is ok.
  if (rtner->nodeId == vnodeNodeId(rtner->tsdb->pVnode)) {
    return tsdbLeaderDoSsMigrate(rtner);
  }
  return tsdbFollowerDoSsMigrate(rtner);
}


#ifdef USE_SHARED_STORAGE

int32_t tsdbListSsMigrateFileSets(STsdb *tsdb, SArray* fidArr) {
  int32_t vid = TD_VID(tsdb->pVnode), code = 0;
  int64_t now = taosGetTimestampSec();
  SVnodeCfg *pCfg = &tsdb->pVnode->config;

  (void)taosThreadMutexLock(&tsdb->mutex);

  STFileSet *fset;
  TARRAY2_FOREACH(tsdb->pFS->fSetArr, fset) {
    if (tsdbFidLevel(fset->fid, &tsdb->keepCfg, now) < 0) {
      tsdbInfo("vgId:%d, fid:%d, migration skipped, file set is expired", vid, fset->fid);
      continue;
    }

    if (tsdbSsFidLevel(fset->fid, &tsdb->keepCfg, tsdb->pVnode->config.ssKeepLocal, now) < 1) {
      tsdbInfo("vgId:%d, fid:%d, migration skipped, keep local file set", vid, fset->fid);
      continue;
    }

    STFileObj *fdata = fset->farr[TSDB_FTYPE_DATA];
    if (fdata == NULL) {
      tsdbInfo("vgId:%d, fid:%d, migration skipped, no data file", vid, fset->fid);
      continue;
    }

    char path[TSDB_FILENAME_LEN];
    if (fdata->f->lcn <= 1) {
      tstrncpy(path, fdata->fname, sizeof(path));
    } else {
      tsdbTFileLastChunkName(tsdb, fdata->f, path);
    }

    int64_t size = 0;
    int32_t code = taosStatFile(path, &size, NULL, NULL);
    if (code != TSDB_CODE_SUCCESS) {
      tsdbError("vgId:%d, fid:%d, migration skipped, failed to stat file since %s", vid, fset->fid, tstrerror(code));
      continue;
    }

    if (size <= (int64_t)pCfg->tsdbPageSize * pCfg->ssChunkSize) {
      tsdbInfo("vgId:%d, fid:%d, migration skipped, data file is too small, size: %" PRId64 " bytes", vid, fset->fid, size);
      continue;
    }

    if (taosArrayPush(fidArr, &fset->fid) == NULL) {
      code = terrno;
      tsdbError("vgId:%d, failed to push file set id %d to array since %s", vid, fset->fid, tstrerror(code));
      break;
    }
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);
  return code;
}


static int32_t tsdbAsyncMigrateFileSetImpl(STsdb *tsdb,  const SSsMigrateFileSetReq *pReq) {
  int32_t vid = TD_VID(tsdb->pVnode), code = 0;

  // check if background task is disabled
  if (tsdb->bgTaskDisabled) {
    tsdbInfo("vgId:%d, ssMigrateId:%d, background task is disabled, skip", vid, pReq->ssMigrateId);
    return TSDB_CODE_FAILED;
  }

  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;
  if (pmm->fid != 0 && pmm->state == SSMIGRATE_FILESET_STATE_IN_PROGRESS) {
    tsdbError("vgId:%d, fid:%d, ssMigrateId:%d, failed to monitor since previous migration is still in progress",
              vid, pReq->fid, pReq->ssMigrateId);
    return TSDB_CODE_FAILED;
  }

  STFileSet *fset = NULL, *fset1;
  TARRAY2_FOREACH(tsdb->pFS->fSetArr, fset1) {
    if (fset1->fid == pReq->fid) {
      fset = fset1;
      break;
    }
  }

  if (fset == NULL) {
    tsdbError("vgId:%d, fid:%d, ssMigrateId:%d, file set not found", vid, pReq->fid, pReq->ssMigrateId);
    return TSDB_CODE_NOT_FOUND;
  }

  if (fset->lastMigrate/1000 >= pReq->startTimeSec) {
    tsdbInfo("vgId:%d, fid:%d, ssMigrate:%d, start time < last migration time, skip", vid, fset->fid, pReq->ssMigrateId);
    return TSDB_CODE_FAILED;
  }

  SRtnArg *arg = taosMemoryMalloc(sizeof(*arg));
  if (arg == NULL) {
    code = terrno;
    tsdbError("vgId:%d, fid:%d, ssMigrateId:%d, memory allocation failed", vid, pReq->fid, pReq->ssMigrateId);
    return code;
  }

  arg->tsdb = tsdb;
  arg->tw.skey = INT64_MIN;
  arg->tw.ekey = pReq->startTimeSec;
  arg->fid = fset->fid;
  arg->nodeId = pReq->nodeId;
  arg->optrType = TSDB_OPTR_SSMIGRATE;
  arg->lastCommit = fset->lastCommit;

  pmm->ssMigrateId = pReq->ssMigrateId;
  pmm->fid = fset->fid;
  pmm->state = SSMIGRATE_FILESET_STATE_IN_PROGRESS;
  pmm->startTimeSec = pReq->startTimeSec;

  code = vnodeAsync(RETENTION_TASK_ASYNC, EVA_PRIORITY_LOW, tsdbRetention, tsdbRetentionCancel, arg, &fset->migrateTask);
  if (code) {
    tsdbError("vgId:%d, fid:%d, ssMigrateId:%d, schedule async task failed", vid, pReq->fid, pReq->ssMigrateId);
    taosMemoryFree(arg);
    pmm->state = SSMIGRATE_FILESET_STATE_FAILED;
  }

  return code;
}


int32_t tsdbAsyncSsMigrateFileSet(STsdb *tsdb, SSsMigrateFileSetReq *pReq) {
  int32_t code = 0;

  (void)taosThreadMutexLock(&tsdb->mutex);
  code = tsdbAsyncMigrateFileSetImpl(tsdb, pReq);
  (void)taosThreadMutexUnlock(&tsdb->mutex);

  return code;
}


void tsdbStopSsMigrateTask(STsdb* tsdb, int32_t ssMigrateId) {
  (void)taosThreadMutexLock(&tsdb->mutex);

  if (tsdb->pSsMigrateMonitor == NULL) {
    (void)taosThreadMutexUnlock(&tsdb->mutex);
    return;
  }

  if (tsdb->pSsMigrateMonitor->ssMigrateId != ssMigrateId) {
    tsdbInfo("vgId:%d, ssMigrateId:%d, migration task not found", TD_VID(tsdb->pVnode), ssMigrateId);
    (void)taosThreadMutexUnlock(&tsdb->mutex);
    return;
  }

  if (tsdb->pSsMigrateMonitor->state != SSMIGRATE_FILESET_STATE_IN_PROGRESS) {
    (void)taosThreadMutexUnlock(&tsdb->mutex);
    return;
  }

  STFileSet *fset;
  TARRAY2_FOREACH(tsdb->pFS->fSetArr, fset) {
    if (fset->fid == tsdb->pSsMigrateMonitor->fid) {
      (void)vnodeACancel(&fset->migrateTask);
      break;
    }
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);
}

#endif

taosdata / TDengine / #4991

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