smaFS.c 18.4 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
/*
 * 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 "sma.h"

// =================================================================================================

K
kailixu 已提交
20
// static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **output);
21
static int32_t tdQTaskInfCmprFn1(const void *p1, const void *p2);
K
kailixu 已提交
22 23 24 25 26 27 28 29

static FORCE_INLINE int32_t tPutQTaskF(uint8_t *p, SQTaskFile *pFile) {
  int32_t n = 0;

  n += tPutI8(p ? p + n : p, pFile->level);
  n += tPutI64v(p ? p + n : p, pFile->size);
  n += tPutI64v(p ? p + n : p, pFile->suid);
  n += tPutI64v(p ? p + n : p, pFile->version);
30
  n += tPutI64v(p ? p + n : p, pFile->mtime);
K
kailixu 已提交
31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57

  return n;
}

static int32_t tdRSmaFSToBinary(uint8_t *p, SRSmaFS *pFS) {
  int32_t  n = 0;
  uint32_t size = taosArrayGetSize(pFS->aQTaskInf);

  // version
  n += tPutI8(p ? p + n : p, 0);

  // SArray<SQTaskFile>
  n += tPutU32v(p ? p + n : p, size);
  for (uint32_t i = 0; i < size; ++i) {
    n += tPutQTaskF(p ? p + n : p, taosArrayGet(pFS->aQTaskInf, i));
  }

  return n;
}

int32_t tdRSmaGetQTaskF(uint8_t *p, SQTaskFile *pFile) {
  int32_t n = 0;

  n += tGetI8(p + n, &pFile->level);
  n += tGetI64v(p + n, &pFile->size);
  n += tGetI64v(p + n, &pFile->suid);
  n += tGetI64v(p + n, &pFile->version);
58
  n += tGetI64v(p + n, &pFile->mtime);
K
kailixu 已提交
59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79

  return n;
}

static int32_t tsdbBinaryToFS(uint8_t *pData, int64_t nData, SRSmaFS *pFS) {
  int32_t code = 0;
  int32_t n = 0;
  int8_t  version = 0;

  // version
  n += tGetI8(pData + n, &version);

  // SArray<SQTaskFile>
  taosArrayClear(pFS->aQTaskInf);
  uint32_t size = 0;
  n += tGetU32v(pData + n, &size);
  for (uint32_t i = 0; i < size; ++i) {
    SQTaskFile qTaskF = {0};

    int32_t nt = tdRSmaGetQTaskF(pData + n, &qTaskF);
    if (nt < 0) {
K
kailixu 已提交
80
      code = TSDB_CODE_FILE_CORRUPTED;
K
kailixu 已提交
81 82 83 84 85 86 87 88 89 90
      goto _exit;
    }

    n += nt;
    if (taosArrayPush(pFS->aQTaskInf, &qTaskF) == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      goto _exit;
    }
  }

K
kailixu 已提交
91 92
  if (ASSERTS(n + sizeof(TSCKSUM) == nData, "n:%d + sizeof(TSCKSUM):%d != nData:%d", n, (int32_t)sizeof(TSCKSUM),
              nData)) {
K
kailixu 已提交
93 94 95
    code = TSDB_CODE_FILE_CORRUPTED;
    goto _exit;
  }
K
kailixu 已提交
96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144

_exit:
  return code;
}

static int32_t tdRSmaSaveFSToFile(SRSmaFS *pFS, const char *fname) {
  int32_t code = 0;
  int32_t lino = 0;

  // encode to binary
  int32_t  size = tdRSmaFSToBinary(NULL, pFS) + sizeof(TSCKSUM);
  uint8_t *pData = taosMemoryMalloc(size);
  if (pData == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  tdRSmaFSToBinary(pData, pFS);
  taosCalcChecksumAppend(0, pData, size);

  // save to file
  TdFilePtr pFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (pFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  int64_t n = taosWriteFile(pFD, pData, size);
  if (n < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (taosFsyncFile(pFD) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  taosCloseFile(&pFD);

_exit:
  if (pData) taosMemoryFree(pData);
  if (code) {
    smaError("%s failed at line %d since %s, fname:%s", __func__, lino, tstrerror(code), fname);
  }
  return code;
}

K
kailixu 已提交
145
static int32_t tdRSmaFSCreate(SRSmaFS *pFS, int32_t size) {
K
kailixu 已提交
146
  int32_t code = 0;
147

K
kailixu 已提交
148
  pFS->aQTaskInf = taosArrayInit(size, sizeof(SQTaskFile));
K
kailixu 已提交
149 150 151
  if (pFS->aQTaskInf == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
152
  }
K
kailixu 已提交
153 154 155

_exit:
  return code;
156 157
}

K
kailixu 已提交
158 159 160 161
static void tdRSmaGetCurrentFName(SSma *pSma, char *current, char *current_t) {
  SVnode *pVnode = pSma->pVnode;
  if (pVnode->pTfs) {
    if (current) {
K
kailixu 已提交
162 163
      snprintf(current, TSDB_FILENAME_LEN - 1, "%s%svnode%svnode%d%srsma%sPRESENT", tfsGetPrimaryPath(pVnode->pTfs),
               TD_DIRSEP, TD_DIRSEP, TD_VID(pVnode), TD_DIRSEP, TD_DIRSEP);
K
kailixu 已提交
164 165
    }
    if (current_t) {
166
      snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%svnode%svnode%d%srsma%sPRESENT.t", tfsGetPrimaryPath(pVnode->pTfs),
K
kailixu 已提交
167
               TD_DIRSEP, TD_DIRSEP, TD_VID(pVnode), TD_DIRSEP, TD_DIRSEP);
K
kailixu 已提交
168 169 170 171
    }
  } else {
#if 0
    if (current) {
K
kailixu 已提交
172
      snprintf(current, TSDB_FILENAME_LEN - 1, "%s%sPRESENT", pTsdb->path, TD_DIRSEP);
K
kailixu 已提交
173 174
    }
    if (current_t) {
K
kailixu 已提交
175
      snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%sPRESENT.t", pTsdb->path, TD_DIRSEP);
K
kailixu 已提交
176 177 178 179
    }
#endif
  }
}
180

K
kailixu 已提交
181 182 183 184 185 186 187 188 189 190
static int32_t tdRSmaLoadFSFromFile(const char *fname, SRSmaFS *pFS) {
  int32_t  code = 0;
  int32_t  lino = 0;
  uint8_t *pData = NULL;

  // load binary
  TdFilePtr pFD = taosOpenFile(fname, TD_FILE_READ);
  if (pFD == NULL) {
    code = TAOS_SYSTEM_ERROR(errno);
    TSDB_CHECK_CODE(code, lino, _exit);
191
  }
K
kailixu 已提交
192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230

  int64_t size;
  if (taosFStatFile(pFD, &size, NULL) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  pData = taosMemoryMalloc(size);
  if (pData == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (taosReadFile(pFD, pData, size) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (!taosCheckChecksumWhole(pData, size)) {
    code = TSDB_CODE_FILE_CORRUPTED;
    taosCloseFile(&pFD);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  taosCloseFile(&pFD);

  // decode binary
  code = tsdbBinaryToFS(pData, size, pFS);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (pData) taosMemoryFree(pData);
  if (code) {
    smaError("%s failed at line %d since %s, fname:%s", __func__, lino, tstrerror(code), fname);
  }
  return code;
231 232
}

K
kailixu 已提交
233 234 235 236 237 238 239 240
static int32_t tdQTaskInfCmprFn1(const void *p1, const void *p2) {
  const SQTaskFile *q1 = (const SQTaskFile *)p1;
  const SQTaskFile *q2 = (const SQTaskFile *)p2;

  if (q1->suid < q2->suid) {
    return -1;
  } else if (q1->suid > q2->suid) {
    return 1;
241
  }
K
kailixu 已提交
242 243 244 245 246 247 248 249

  if (q1->level < q2->level) {
    return -1;
  } else if (q1->level > q2->level) {
    return 1;
  }

  if (q1->version < q2->version) {
K
kailixu 已提交
250
    return -2;
K
kailixu 已提交
251 252 253 254 255
  } else if (q1->version > q2->version) {
    return 1;
  }

  return 0;
256 257
}

258
static int32_t tdRSmaFSApplyChange(SSma *pSma, SRSmaFS *pFSNew) {
K
kailixu 已提交
259 260
  int32_t    code = 0;
  int32_t    lino = 0;
K
kailixu 已提交
261
  int32_t    nRef = 0;
K
kailixu 已提交
262 263 264
  SVnode    *pVnode = pSma->pVnode;
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
K
kailixu 已提交
265 266 267
  SRSmaFS   *pFSOld = RSMA_FS(pStat);
  int64_t    version = pStat->commitAppliedVer;
  char       fname[TSDB_FILENAME_LEN] = {0};
K
kailixu 已提交
268 269

  // SQTaskFile
270
  int32_t nNew = taosArrayGetSize(pFSNew->aQTaskInf);
K
kailixu 已提交
271
  int32_t iNew = 0;
K
kailixu 已提交
272
  while (iNew < nNew) {
273
    SQTaskFile *pQTaskFNew = TARRAY_GET_ELEM(pFSNew->aQTaskInf, iNew++);
K
kailixu 已提交
274

K
kailixu 已提交
275
    int32_t idx = taosArraySearchIdx(pFSOld->aQTaskInf, pQTaskFNew, tdQTaskInfCmprFn1, TD_GE);
K
kailixu 已提交
276

K
kailixu 已提交
277 278 279
    if (idx < 0) {
      idx = taosArrayGetSize(pFSOld->aQTaskInf);
      pQTaskFNew->nRef = 1;
K
kailixu 已提交
280
    } else {
K
kailixu 已提交
281 282 283
      SQTaskFile *pTaskF = TARRAY_GET_ELEM(pFSOld->aQTaskInf, idx);
      int32_t     c1 = tdQTaskInfCmprFn1(pQTaskFNew, pTaskF);
      if (c1 == 0) {
284
        // utilize the item in pFSOld->qQTaskInf, instead of pFSNew
K
kailixu 已提交
285
        continue;
K
kailixu 已提交
286
      } else if (c1 < 0) {
287
        // NOTHING TODO
K
kailixu 已提交
288
      } else {
K
kailixu 已提交
289 290
        code = TSDB_CODE_RSMA_FS_UPDATE;
        TSDB_CHECK_CODE(code, lino, _exit);
K
kailixu 已提交
291
      }
K
kailixu 已提交
292
    }
K
kailixu 已提交
293

294
    if (taosArrayInsert(pFSOld->aQTaskInf, idx, pQTaskFNew) == NULL) {
K
kailixu 已提交
295
      code = TSDB_CODE_OUT_OF_MEMORY;
K
kailixu 已提交
296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    // remove previous version
    while (--idx >= 0) {
      SQTaskFile *preTaskF = TARRAY_GET_ELEM(pFSOld->aQTaskInf, idx);
      int32_t     c2 = tdQTaskInfCmprFn1(preTaskF, pQTaskFNew);
      if (c2 == 0) {
        code = TSDB_CODE_RSMA_FS_UPDATE;
        TSDB_CHECK_CODE(code, lino, _exit);
      } else if (c2 != -2) {
        break;
      }

      nRef = atomic_sub_fetch_32(&preTaskF->nRef, 1);
      if (nRef <= 0) {
        tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), preTaskF->suid, preTaskF->level, preTaskF->version,
                                   tfsGetPrimaryPath(pVnode->pTfs), fname);
        (void)taosRemoveFile(fname);
        taosArrayRemove(pFSOld->aQTaskInf, idx);
      }
317 318
    }
  }
K
kailixu 已提交
319 320 321

_exit:
  if (code) {
K
kailixu 已提交
322
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
K
kailixu 已提交
323 324
  }
  return code;
325 326
}

K
kailixu 已提交
327
static int32_t tdRSmaFSScanAndTryFix(SSma *pSma) {
X
Xiaoyu Wang 已提交
328
  int32_t code = 0;
K
kailixu 已提交
329
#if 0
K
kailixu 已提交
330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371
  int32_t    lino = 0;
  SVnode    *pVnode = pSma->pVnode;
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  SRSmaFS   *pFS = RSMA_FS(pStat);
  char       fname[TSDB_FILENAME_LEN] = {0};
  char       fnameVer[TSDB_FILENAME_LEN] = {0};

  // SArray<SQTaskFile>
  int32_t size = taosArrayGetSize(pFS->aQTaskInf);
  for (int32_t i = 0; i < size; ++i) {
    SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, i);

    // main.tdb =========
    tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, pTaskF->version,
                               tfsGetPrimaryPath(pVnode->pTfs), fnameVer);
    tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, -1, tfsGetPrimaryPath(pVnode->pTfs), fname);

    if (taosCheckExistFile(fnameVer)) {
      if (taosRenameFile(fnameVer, fname) < 0) {
        code = TAOS_SYSTEM_ERROR(errno);
        TSDB_CHECK_CODE(code, lino, _exit);
      }
      smaDebug("vgId:%d, %s:%d succeed to to rename %s to %s", TD_VID(pVnode), __func__, lino, fnameVer, fname);
    } else if (taosCheckExistFile(fname)) {
      if (taosRemoveFile(fname) < 0) {
        code = TAOS_SYSTEM_ERROR(errno);
        TSDB_CHECK_CODE(code, lino, _exit);
      }
      smaDebug("vgId:%d, %s:%d succeed to to remove %s", TD_VID(pVnode), __func__, lino, fname);
    }
  }

  {
    // remove those invalid files (todo)
    // main.tdb-journal.5 // TDB should handle its clear for kill -9
  }

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  }
K
kailixu 已提交
372
#endif
K
kailixu 已提交
373 374 375
  return code;
}

K
kailixu 已提交
376
// EXPOSED APIS ====================================================================================
K
kailixu 已提交
377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407

int32_t tdRSmaFSOpen(SSma *pSma, int64_t version, int8_t rollback) {
  int32_t    code = 0;
  int32_t    lino = 0;
  SVnode    *pVnode = pSma->pVnode;
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);

  // open handle
  code = tdRSmaFSCreate(RSMA_FS(pStat), 0);
  TSDB_CHECK_CODE(code, lino, _exit);

  // open impl
  char current[TSDB_FILENAME_LEN] = {0};
  char current_t[TSDB_FILENAME_LEN] = {0};
  tdRSmaGetCurrentFName(pSma, current, current_t);

  if (taosCheckExistFile(current)) {
    code = tdRSmaLoadFSFromFile(current, RSMA_FS(pStat));
    TSDB_CHECK_CODE(code, lino, _exit);

    if (taosCheckExistFile(current_t)) {
      if (rollback) {
        code = tdRSmaFSRollback(pSma);
        TSDB_CHECK_CODE(code, lino, _exit);
      } else {
        code = tdRSmaFSCommit(pSma);
        TSDB_CHECK_CODE(code, lino, _exit);
      }
    }
  } else {
K
kailixu 已提交
408
    // 1st time open with empty current/qTaskInfoFile
K
kailixu 已提交
409 410 411 412 413 414 415 416 417 418 419 420 421 422 423
    code = tdRSmaSaveFSToFile(RSMA_FS(pStat), current);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  // scan and try fix(remove main.db/main.db.xxx and use the one with version)
  code = tdRSmaFSScanAndTryFix(pSma);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

K
kailixu 已提交
424
void tdRSmaFSClose(SRSmaFS *pFS) { pFS->aQTaskInf = taosArrayDestroy(pFS->aQTaskInf); }
K
kailixu 已提交
425 426 427 428 429 430 431 432

int32_t tdRSmaFSPrepareCommit(SSma *pSma, SRSmaFS *pFSNew) {
  int32_t code = 0;
  int32_t lino = 0;
  char    tfname[TSDB_FILENAME_LEN];

  tdRSmaGetCurrentFName(pSma, NULL, tfname);

433
  // generate PRESENT.t
K
kailixu 已提交
434 435 436 437 438 439 440 441 442 443
  code = tdRSmaSaveFSToFile(pFSNew, tfname);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

K
kailixu 已提交
444 445 446 447 448 449 450 451 452
int32_t tdRSmaFSCommit(SSma *pSma) {
  int32_t code = 0;
  int32_t lino = 0;
  SRSmaFS fs = {0};

  char current[TSDB_FILENAME_LEN] = {0};
  char current_t[TSDB_FILENAME_LEN] = {0};
  tdRSmaGetCurrentFName(pSma, current, current_t);

K
kailixu 已提交
453 454 455
  if (!taosCheckExistFile(current_t)) {
    goto _exit;
  }
K
kailixu 已提交
456 457 458 459 460

  // rename the file
  if (taosRenameFile(current_t, current) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    TSDB_CHECK_CODE(code, lino, _exit);
461 462
  }

K
kailixu 已提交
463
  // load the new FS
K
kailixu 已提交
464
  code = tdRSmaFSCreate(&fs, 1);
K
kailixu 已提交
465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481
  TSDB_CHECK_CODE(code, lino, _exit);

  code = tdRSmaLoadFSFromFile(current, &fs);
  TSDB_CHECK_CODE(code, lino, _exit);

  // apply file change
  code = tdRSmaFSApplyChange(pSma, &fs);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  tdRSmaFSClose(&fs);
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pSma), __func__, lino, tstrerror(code));
  }
  return code;
}

K
kailixu 已提交
482 483 484 485 486 487 488
int32_t tdRSmaFSFinishCommit(SSma *pSma) {
  int32_t    code = 0;
  int32_t    lino = 0;
  SSmaEnv   *pSmaEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv);

  taosWLockLatch(RSMA_FS_LOCK(pStat));
K
kailixu 已提交
489 490
  code = tdRSmaFSCommit(pSma);
  TSDB_CHECK_CODE(code, lino, _exit);
K
kailixu 已提交
491 492

_exit:
K
kailixu 已提交
493
  taosWUnLockLatch(RSMA_FS_LOCK(pStat));
K
kailixu 已提交
494 495 496 497 498 499 500 501
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pSma), __func__, lino, tstrerror(code));
  } else {
    smaInfo("vgId:%d, rsmaFS finish commit", SMA_VID(pSma));
  }
  return code;
}

K
kailixu 已提交
502 503 504 505 506 507 508 509 510 511 512 513 514
int32_t tdRSmaFSRollback(SSma *pSma) {
  int32_t code = 0;
  int32_t lino = 0;

  char current_t[TSDB_FILENAME_LEN] = {0};
  tdRSmaGetCurrentFName(pSma, NULL, current_t);
  (void)taosRemoveFile(current_t);

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", SMA_VID(pSma), __func__, lino, tstrerror(errno));
  }
  return code;
515 516
}

K
kailixu 已提交
517
int32_t tdRSmaFSUpsertQTaskFile(SSma *pSma, SRSmaFS *pFS, SQTaskFile *qTaskFile, int32_t nSize) {
K
kailixu 已提交
518
  int32_t code = 0;
K
kailixu 已提交
519

K
kailixu 已提交
520
  for (int32_t i = 0; i < nSize; ++i) {
K
kailixu 已提交
521
    SQTaskFile *qTaskF = qTaskFile + i;
K
kailixu 已提交
522

K
kailixu 已提交
523
    int32_t idx = taosArraySearchIdx(pFS->aQTaskInf, qTaskF, tdQTaskInfCmprFn1, TD_GE);
K
kailixu 已提交
524

K
kailixu 已提交
525 526 527 528 529 530
    if (idx < 0) {
      idx = taosArrayGetSize(pFS->aQTaskInf);
    } else {
      SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, idx);
      int32_t     c = tdQTaskInfCmprFn1(pTaskF, qTaskF);
      if (c == 0) {
K
kailixu 已提交
531 532 533 534 535 536 537 538
        if (pTaskF->size != qTaskF->size) {
          code = TSDB_CODE_RSMA_FS_UPDATE;
          smaError("vgId:%d, %s failed at line %d since %s, level:%" PRIi8 ", suid:%" PRIi64 ", version:%" PRIi64
                   ", size:%" PRIi64 " != %" PRIi64,
                   SMA_VID(pSma), __func__, __LINE__, tstrerror(code), pTaskF->level, pTaskF->suid, pTaskF->version,
                   pTaskF->size, qTaskF->size);
          goto _exit;
        }
539
        continue;
K
kailixu 已提交
540 541 542
      }
    }

543
    if (!taosArrayInsert(pFS->aQTaskInf, idx, qTaskF)) {
K
kailixu 已提交
544
      code = TSDB_CODE_OUT_OF_MEMORY;
545 546 547 548 549 550
      goto _exit;
    }
  }

_exit:
  return code;
K
kailixu 已提交
551
}
K
kailixu 已提交
552 553 554 555 556 557 558 559 560 561

int32_t tdRSmaFSRef(SSma *pSma, SRSmaFS *pFS) {
  int32_t    code = 0;
  int32_t    lino = 0;
  int32_t    nRef = 0;
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  SRSmaFS   *qFS = RSMA_FS(pStat);
  int32_t    size = taosArrayGetSize(qFS->aQTaskInf);

X
Xiaoyu Wang 已提交
562
  pFS->aQTaskInf = taosArrayInit_s(size, sizeof(SQTaskFile), size);
K
kailixu 已提交
563 564 565 566 567 568 569 570
  if (pFS->aQTaskInf == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  for (int32_t i = 0; i < size; ++i) {
    SQTaskFile *qTaskF = (SQTaskFile *)taosArrayGet(qFS->aQTaskInf, i);
    nRef = atomic_fetch_add_32(&qTaskF->nRef, 1);
K
kailixu 已提交
571
    if (nRef <= 0) {
K
kailixu 已提交
572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613
      code = TSDB_CODE_RSMA_FS_REF;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
  }

  memcpy(pFS->aQTaskInf->pData, qFS->aQTaskInf->pData, size * sizeof(SQTaskFile));

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s, nRef %d", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code),
             nRef);
  }
  return code;
}

void tdRSmaFSUnRef(SSma *pSma, SRSmaFS *pFS) {
  int32_t    nRef = 0;
  char       fname[TSDB_FILENAME_LEN];
  SVnode    *pVnode = pSma->pVnode;
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  int32_t    size = taosArrayGetSize(pFS->aQTaskInf);

  for (int32_t i = 0; i < size; ++i) {
    SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, i);

    nRef = atomic_sub_fetch_32(&pTaskF->nRef, 1);
    if (nRef == 0) {
      tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->suid, pTaskF->level, pTaskF->version,
                                 tfsGetPrimaryPath(pVnode->pTfs), fname);
      if (taosRemoveFile(fname) < 0) {
        smaWarn("vgId:%d, failed to remove %s since %s", TD_VID(pVnode), fname, tstrerror(TAOS_SYSTEM_ERROR(errno)));
      } else {
        smaDebug("vgId:%d, success to remove %s", TD_VID(pVnode), fname);
      }
    } else if (nRef < 0) {
      smaWarn("vgId:%d, abnormal unref %s since %s", TD_VID(pVnode), fname, tstrerror(TSDB_CODE_RSMA_FS_REF));
    }
  }

  taosArrayDestroy(pFS->aQTaskInf);
}
K
kailixu 已提交
614

K
kailixu 已提交
615
int32_t tdRSmaFSTakeSnapshot(SSma *pSma, SRSmaFS *pFS) {
K
kailixu 已提交
616 617
  int32_t    code = 0;
  int32_t    lino = 0;
K
kailixu 已提交
618 619 620 621
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);

  taosRLockLatch(RSMA_FS_LOCK(pStat));
K
kailixu 已提交
622
  code = tdRSmaFSRef(pSma, pFS);
K
kailixu 已提交
623
  TSDB_CHECK_CODE(code, lino, _exit);
K
kailixu 已提交
624
_exit:
K
kailixu 已提交
625
  taosRUnLockLatch(RSMA_FS_LOCK(pStat));
K
kailixu 已提交
626 627 628 629 630 631
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

K
kailixu 已提交
632
int32_t tdRSmaFSCopy(SSma *pSma, SRSmaFS *pFS) {
K
kailixu 已提交
633 634 635 636
  int32_t    code = 0;
  int32_t    lino = 0;
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
K
kailixu 已提交
637 638
  SRSmaFS   *qFS = RSMA_FS(pStat);
  int32_t    size = taosArrayGetSize(qFS->aQTaskInf);
K
kailixu 已提交
639

K
kailixu 已提交
640
  code = tdRSmaFSCreate(pFS, size);
K
kailixu 已提交
641
  TSDB_CHECK_CODE(code, lino, _exit);
X
Xiaoyu Wang 已提交
642
  taosArrayAddBatch(pFS->aQTaskInf, qFS->aQTaskInf->pData, size);
K
kailixu 已提交
643 644 645 646 647 648 649

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pSma->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}