taosdata/TDengine | Build 3798 | source/dnode/mnode/sdb/src/sdbFile.c | Coveralls

31

static int32_t sdbDeployData(SSdb *pSdb) {

×

32

  int32_t code = 0;

×

33

  mInfo("start to deploy sdb");

×

35

  for (int32_t i = SDB_MAX - 1; i >= 0; --i) {

×

36

    SdbDeployFp fp = pSdb->deployFps[i];

×

37

    if (fp == NULL) continue;

×

39

    mInfo("start to deploy sdb:%s", sdbTableName(i));

×

40

    code = (*fp)(pSdb->pMnode);

×

41

    if (code != 0) {

×

42

      mError("failed to deploy sdb:%s since %s", sdbTableName(i), tstrerror(code));

×

43

      return -1;

×

47

  mInfo("sdb deploy success");

×

48

  return 0;

×

51

static int32_t sdbAfterRestoredData(SSdb *pSdb) {

×

52

  int32_t code = 0;

×

53

  mInfo("start to prepare sdb");

×

55

  for (int32_t i = SDB_MAX - 1; i >= 0; --i) {

×

56

    SdbAfterRestoredFp fp = pSdb->afterRestoredFps[i];

×

57

    if (fp == NULL) continue;

×

59

    mInfo("start to prepare sdb:%s", sdbTableName(i));

×

60

    code = (*fp)(pSdb->pMnode);

×

61

    if (code != 0) {

×

62

      mError("failed to prepare sdb:%s since %s", sdbTableName(i), tstrerror(code));

×

63

      return -1;

×

67

  mInfo("sdb prepare success");

×

68

  return 0;

×

71

static void sdbResetData(SSdb *pSdb) {

×

72

  mInfo("start to reset sdb");

×

74

  for (ESdbType i = 0; i < SDB_MAX; ++i) {

×

75

    SHashObj *hash = pSdb->hashObjs[i];

×

76

    if (hash == NULL) continue;

×

78

    sdbWriteLock(pSdb, i);

×

80

    SSdbRow **ppRow = taosHashIterate(hash, NULL);

×

81

    while (ppRow != NULL) {

×

82

      SSdbRow *pRow = *ppRow;

×

83

      if (pRow == NULL) continue;

×

85

      sdbFreeRow(pSdb, pRow, true);

×

86

      ppRow = taosHashIterate(hash, ppRow);

×

89

    taosHashClear(pSdb->hashObjs[i]);

×

90

    pSdb->tableVer[i] = 0;

×

91

    pSdb->maxId[i] = 0;

×

93

    sdbUnLock(pSdb, i);

×

95

    mInfo("sdb:%s is reset", sdbTableName(i));

×

98

  pSdb->applyIndex = -1;

×

99

  pSdb->applyTerm = -1;

×

100

  pSdb->applyConfig = -1;

×

101

  pSdb->commitIndex = -1;

×

102

  pSdb->commitTerm = -1;

×

103

  pSdb->commitConfig = -1;

×

104

  mInfo("sdb reset success");

×

105

×

107

static int32_t sdbReadFileHead(SSdb *pSdb, TdFilePtr pFile) {

×

108

  int32_t code = 0;

×

109

  int64_t sver = 0;

×

110

  int32_t ret = taosReadFile(pFile, &sver, sizeof(int64_t));

×

111

  if (ret < 0) {

×

112

    return terrno;

×

114

  if (ret != sizeof(int64_t)) {

×

115

    code = TSDB_CODE_FILE_CORRUPTED;

×

116

    TAOS_RETURN(code);

×

118

  if (sver != SDB_FILE_VER) {

×

119

    code = TSDB_CODE_FILE_CORRUPTED;

×

120

    TAOS_RETURN(code);

×

123

  ret = taosReadFile(pFile, &pSdb->applyIndex, sizeof(int64_t));

×

124

  if (ret < 0) {

×

125

    return terrno;

×

127

  if (ret != sizeof(int64_t)) {

×

128

    code = TSDB_CODE_FILE_CORRUPTED;

×

129

    TAOS_RETURN(code);

×

132

  ret = taosReadFile(pFile, &pSdb->applyTerm, sizeof(int64_t));

×

133

  if (ret < 0) {

×

134

    return terrno;

×

136

  if (ret != sizeof(int64_t)) {

×

137

    code = TSDB_CODE_FILE_CORRUPTED;

×

138

    TAOS_RETURN(code);

×

141

  ret = taosReadFile(pFile, &pSdb->applyConfig, sizeof(int64_t));

×

142

  if (ret < 0) {

×

143

    return terrno;

×

145

  if (ret != sizeof(int64_t)) {

×

146

    code = TSDB_CODE_FILE_CORRUPTED;

×

147

    TAOS_RETURN(code);

×

150

  for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {

×

151

    int64_t maxId = 0;

×

152

    ret = taosReadFile(pFile, &maxId, sizeof(int64_t));

×

153

    if (ret < 0) {

×

154

      return terrno;

×

156

    if (ret != sizeof(int64_t)) {

×

157

      code = TSDB_CODE_FILE_CORRUPTED;

×

158

      TAOS_RETURN(code);

×

160

    if (i < SDB_MAX) {

×

161

      pSdb->maxId[i] = maxId;

×

165

  for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {

×

166

    int64_t ver = 0;

×

167

    ret = taosReadFile(pFile, &ver, sizeof(int64_t));

×

168

    if (ret < 0) {

×

169

      return terrno;

×

171

    if (ret != sizeof(int64_t)) {

×

172

      code = TSDB_CODE_FILE_CORRUPTED;

×

173

      TAOS_RETURN(code);

×

175

    if (i < SDB_MAX) {

×

176

      pSdb->tableVer[i] = ver;

×

181

  for (int32_t i = SDB_TABLE_SIZE; i < SDB_TABLE_SIZE_EXTRA; ++i) {

×

182

    int64_t maxId = 0;

×

183

    ret = taosReadFile(pFile, &maxId, sizeof(int64_t));

×

184

    if (ret < 0) {

×

185

      code = TAOS_SYSTEM_ERROR(ERRNO);

×

186

      TAOS_RETURN(code);

×

188

    if (ret != sizeof(int64_t)) {

×

189

      code = TSDB_CODE_FILE_CORRUPTED;

×

190

      TAOS_RETURN(code);

×

192

    if (i < SDB_MAX) {

×

193

      pSdb->maxId[i] = maxId;

×

196

    int64_t ver = 0;

×

197

    ret = taosReadFile(pFile, &ver, sizeof(int64_t));

×

198

    if (ret < 0) {

×

199

      code = TAOS_SYSTEM_ERROR(ERRNO);

×

200

      TAOS_RETURN(code);

×

202

    if (ret != sizeof(int64_t)) {

×

203

      code = TSDB_CODE_FILE_CORRUPTED;

×

204

      TAOS_RETURN(code);

×

206

    if (i < SDB_MAX) {

×

207

      pSdb->tableVer[i] = ver;

×

211

  char reserve[SDB_RESERVE_SIZE_EXTRA] = {0};

×

212

  ret = taosReadFile(pFile, reserve, sizeof(reserve));

×

213

  if (ret < 0) {

×

214

    return terrno;

×

216

  if (ret != sizeof(reserve)) {

×

217

    code = TSDB_CODE_FILE_CORRUPTED;

×

218

    TAOS_RETURN(code);

×

221

  return 0;

×

224

static int32_t sdbWriteFileHead(SSdb *pSdb, TdFilePtr pFile) {

×

225

  int64_t sver = SDB_FILE_VER;

×

226

  if (taosWriteFile(pFile, &sver, sizeof(int64_t)) != sizeof(int64_t)) {

×

227

    return terrno;

×

230

  mInfo("vgId:1, write sdb file with sdb applyIndex:%" PRId64 " term:%" PRId64 " config:%" PRId64, pSdb->applyIndex,

×

232

  if (taosWriteFile(pFile, &pSdb->applyIndex, sizeof(int64_t)) != sizeof(int64_t)) {

×

233

    return terrno;

×

236

  if (taosWriteFile(pFile, &pSdb->applyTerm, sizeof(int64_t)) != sizeof(int64_t)) {

×

237

    return terrno;

×

240

  if (taosWriteFile(pFile, &pSdb->applyConfig, sizeof(int64_t)) != sizeof(int64_t)) {

×

241

    return terrno;

×

244

  for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {

×

245

    int64_t maxId = 0;

×

246

    if (i < SDB_MAX) {

×

247

      maxId = pSdb->maxId[i];

×

249

    if (taosWriteFile(pFile, &maxId, sizeof(int64_t)) != sizeof(int64_t)) {

×

250

      return terrno;

×

254

  for (int32_t i = 0; i < SDB_TABLE_SIZE; ++i) {

×

255

    int64_t ver = 0;

×

256

    if (i < SDB_MAX) {

×

257

      ver = pSdb->tableVer[i];

×

259

    if (taosWriteFile(pFile, &ver, sizeof(int64_t)) != sizeof(int64_t)) {

×

260

      return terrno;

×

265

  for (int32_t i = SDB_TABLE_SIZE; i < SDB_TABLE_SIZE_EXTRA; ++i) {

×

266

    int64_t maxId = 0;

×

267

    if (i < SDB_MAX) {

×

268

      maxId = pSdb->maxId[i];

×

270

    if (taosWriteFile(pFile, &maxId, sizeof(int64_t)) != sizeof(int64_t)) {

×

271

      return terrno;

×

274

    int64_t ver = 0;

×

275

    if (i < SDB_MAX) {

×

276

      ver = pSdb->tableVer[i];

×

278

    if (taosWriteFile(pFile, &ver, sizeof(int64_t)) != sizeof(int64_t)) {

×

279

      return terrno;

×

283

  char reserve[SDB_RESERVE_SIZE_EXTRA] = {0};

×

284

  if (taosWriteFile(pFile, reserve, sizeof(reserve)) != sizeof(reserve)) {

×

285

    return terrno;

×

288

  return 0;

×

291

static int32_t sdbReadFileImp(SSdb *pSdb) {

×

292

  int64_t offset = 0;

×

293

  int32_t code = 0;

×

294

  int32_t readLen = 0;

×

295

  int64_t ret = 0;

×

296

  char    file[PATH_MAX] = {0};

×

297

  int32_t bufLen = TSDB_MAX_MSG_SIZE;

×

299

  snprintf(file, sizeof(file), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);

×

300

  mInfo("start to read sdb file:%s", file);

×

302

  SSdbRaw *pRaw = taosMemoryMalloc(bufLen + 100);

×

303

  if (pRaw == NULL) {

×

304

    code = terrno;

×

305

    mError("failed read sdb file since %s", tstrerror(code));

×

306

    TAOS_RETURN(code);

×

309

  TdFilePtr pFile = taosOpenFile(file, TD_FILE_READ);

×

310

  if (pFile == NULL) {

×

311

    taosMemoryFree(pRaw);

×

312

    code = terrno;

×

313

    mInfo("read sdb file:%s finished since %s", file, tstrerror(code));

×

314

    return 0;

×

317

  code = sdbReadFileHead(pSdb, pFile);

×

318

  if (code != 0) {

×

319

    mError("failed to read sdb file:%s head since %s", file, tstrerror(code));

×

320

    taosMemoryFree(pRaw);

×

321

    int32_t ret = 0;

×

322

    if ((ret = taosCloseFile(&pFile)) != 0) {

×

323

      mError("failed to close sdb file:%s since %s", file, tstrerror(ret));

×

325

    return code;

×

328

  int64_t tableVer[SDB_MAX] = {0};

×

329

  memcpy(tableVer, pSdb->tableVer, sizeof(tableVer));

×

331

  while (1) {

×

332

    readLen = sizeof(SSdbRaw);

×

333

    ret = taosReadFile(pFile, pRaw, readLen);

×

334

    if (ret == 0) break;

×

336

    if (ret < 0) {

×

337

      code = terrno;

×

338

      mError("failed to read sdb file:%s since %s", file, tstrerror(code));

×

339

      goto _OVER;

×

342

    if (ret != readLen) {

×

343

      code = TSDB_CODE_FILE_CORRUPTED;

×

344

      mError("failed to read sdb file:%s since %s, ret:%" PRId64 " != readLen:%d", file, tstrerror(code), ret, readLen);

×

345

      goto _OVER;

×

348

    readLen = pRaw->dataLen + sizeof(int32_t);

×

349

    if (tsiEncryptAlgorithm == DND_CA_SM4 && (tsiEncryptScope & DND_CS_SDB) == DND_CS_SDB) {

×

350

      readLen = ENCRYPTED_LEN(pRaw->dataLen) + sizeof(int32_t);

×

352

    if (readLen >= bufLen) {

×

353

      bufLen = pRaw->dataLen * 2;

×

354

      SSdbRaw *pNewRaw = taosMemoryMalloc(bufLen + 100);

×

355

      if (pNewRaw == NULL) {

×

356

        code = terrno;

×

357

        mError("failed read sdb file since malloc new sdbRaw size:%d failed", bufLen);

×

358

        goto _OVER;

×

360

      mInfo("malloc new sdb raw size:%d, type:%d", bufLen, pRaw->type);

×

361

      memcpy(pNewRaw, pRaw, sizeof(SSdbRaw));

×

362

      sdbFreeRaw(pRaw);

×

363

      pRaw = pNewRaw;

×

366

    ret = taosReadFile(pFile, pRaw->pData, readLen);

×

367

    if (ret < 0) {

×

368

      code = terrno;

×

369

      mError("failed to read sdb file:%s since %s, ret:%" PRId64 " readLen:%d", file, tstrerror(code), ret, readLen);

×

370

      goto _OVER;

×

373

    if (ret != readLen) {

×

374

      code = TSDB_CODE_FILE_CORRUPTED;

×

375

      mError("failed to read sdb file:%s since %s, ret:%" PRId64 " != readLen:%d", file, tstrerror(code), ret, readLen);

×

376

      goto _OVER;

×

379

    if (tsiEncryptAlgorithm == DND_CA_SM4 && (tsiEncryptScope & DND_CS_SDB) == DND_CS_SDB) {

×

380

      int32_t count = 0;

×

382

      char *plantContent = taosMemoryMalloc(ENCRYPTED_LEN(pRaw->dataLen));

×

383

      if (plantContent == NULL) {

×

384

        code = terrno;

×

385

        goto _OVER;

×

389

      opts.len = ENCRYPTED_LEN(pRaw->dataLen);

×

390

      opts.source = pRaw->pData;

×

391

      opts.result = plantContent;

×

392

      opts.unitLen = 16;

×

393

      tstrncpy(opts.key, tsEncryptKey, ENCRYPT_KEY_LEN + 1);

×

395

      count = CBC_Decrypt(&opts);

×

399

      memcpy(pRaw->pData, plantContent, pRaw->dataLen);

×

400

      taosMemoryFree(plantContent);

×

401

      memcpy(pRaw->pData + pRaw->dataLen, &pRaw->pData[ENCRYPTED_LEN(pRaw->dataLen)], sizeof(int32_t));

×

404

    int32_t totalLen = sizeof(SSdbRaw) + pRaw->dataLen + sizeof(int32_t);

×

405

    if ((!taosCheckChecksumWhole((const uint8_t *)pRaw, totalLen)) != 0) {

×

406

      code = TSDB_CODE_CHECKSUM_ERROR;

×

407

      mError("failed to read sdb file:%s since %s, readLen:%d", file, tstrerror(code), readLen);

×

408

      goto _OVER;

×

411

    if (pRaw->type >= SDB_MAX) {

×

412

      mInfo("skip sdb raw type:%d since it is not supported", pRaw->type);

×

413

      continue;

×

416

    code = sdbWriteWithoutFree(pSdb, pRaw);

×

417

    if (code != 0) {

×

418

      mError("failed to exec sdbWrite while read sdb file:%s since %s", file, terrstr());

×

419

      goto _OVER;

×

423

  code = 0;

×

424

  pSdb->commitIndex = pSdb->applyIndex;

×

425

  pSdb->commitTerm = pSdb->applyTerm;

×

426

  pSdb->commitConfig = pSdb->applyConfig;

×

427

  memcpy(pSdb->tableVer, tableVer, sizeof(tableVer));

×

428

  mInfo("vgId:1, trans:0, read sdb file:%s success, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64, file,

×

431

_OVER:

×

432

  if ((ret = taosCloseFile(&pFile)) != 0) {

×

433

    mError("failed to close sdb file:%s since %s", file, tstrerror(ret));

×

435

  sdbFreeRaw(pRaw);

×

437

  TAOS_RETURN(code);

×

440

int32_t sdbReadFile(SSdb *pSdb) {

×

441

  (void)taosThreadMutexLock(&pSdb->filelock);

×

443

  sdbResetData(pSdb);

×

444

  int32_t code = sdbReadFileImp(pSdb);

×

445

  if (code != 0) {

×

446

    mError("failed to read sdb file since %s", tstrerror(code));

×

447

    sdbResetData(pSdb);

×

450

  (void)taosThreadMutexUnlock(&pSdb->filelock);

×

451

  return code;

×

454

static int32_t sdbWriteFileImp(SSdb *pSdb, int32_t skip_type) {

×

455

  int32_t code = 0;

×

457

  char tmpfile[PATH_MAX] = {0};

×

458

  snprintf(tmpfile, sizeof(tmpfile), "%s%ssdb.data", pSdb->tmpDir, TD_DIRSEP);

×

459

  char curfile[PATH_MAX] = {0};

×

460

  snprintf(curfile, sizeof(curfile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);

×

462

  mInfo("start to write sdb file, apply index:%" PRId64 " term:%" PRId64 " config:%" PRId64 ", commit index:%" PRId64

×

467

  TdFilePtr pFile = taosOpenFile(tmpfile, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);

×

468

  if (pFile == NULL) {

×

469

    code = terrno;

×

470

    mError("failed to open sdb file:%s for write since %s", tmpfile, tstrerror(code));

×

471

    TAOS_RETURN(code);

×

474

  code = sdbWriteFileHead(pSdb, pFile);

×

475

  if (code != 0) {

×

476

    mError("failed to write sdb file:%s head since %s", tmpfile, tstrerror(code));

×

477

    int32_t ret = 0;

×

478

    if ((ret = taosCloseFile(&pFile)) != 0) {

×

479

      mError("failed to close sdb file:%s since %s", tmpfile, tstrerror(ret));

×

481

    return code;

×

484

  for (int32_t i = SDB_MAX - 1; i >= 0; --i) {

×

485

    if (i == skip_type) continue;

×

486

    SdbEncodeFp encodeFp = pSdb->encodeFps[i];

×

487

    if (encodeFp == NULL) continue;

×

489

    mInfo("write %s to sdb file, total %d rows", sdbTableName(i), sdbGetSize(pSdb, i));

×

491

    SHashObj *hash = pSdb->hashObjs[i];

×

492

    sdbWriteLock(pSdb, i);

×

494

    SSdbRow **ppRow = taosHashIterate(hash, NULL);

×

495

    while (ppRow != NULL) {

×

496

      SSdbRow *pRow = *ppRow;

×

497

      if (pRow == NULL) {

×

498

        ppRow = taosHashIterate(hash, ppRow);

×

499

        continue;

×

502

      if (pRow->status != SDB_STATUS_READY && pRow->status != SDB_STATUS_DROPPING) {

×

503

        sdbPrintOper(pSdb, pRow, "not-write");

×

504

        ppRow = taosHashIterate(hash, ppRow);

×

505

        continue;

×

508

      sdbPrintOper(pSdb, pRow, "write");

×

510

      SSdbRaw *pRaw = (*encodeFp)(pRow->pObj);

×

511

      if (pRaw != NULL) {

×

512

        pRaw->status = pRow->status;

×

514

        if (taosWriteFile(pFile, pRaw, sizeof(SSdbRaw)) != sizeof(SSdbRaw)) {

×

515

          code = terrno;

×

516

          taosHashCancelIterate(hash, ppRow);

×

517

          sdbFreeRaw(pRaw);

×

518

          break;

×

521

        int32_t newDataLen = pRaw->dataLen;

×

522

        char   *newData = pRaw->pData;

×

523

        if (tsiEncryptAlgorithm == DND_CA_SM4 && (tsiEncryptScope & DND_CS_SDB) == DND_CS_SDB) {

×

524

          newDataLen = ENCRYPTED_LEN(pRaw->dataLen);

×

525

          newData = taosMemoryMalloc(newDataLen);

×

526

          if (newData == NULL) {

×

527

            code = terrno;

×

528

            taosHashCancelIterate(hash, ppRow);

×

529

            sdbFreeRaw(pRaw);

×

530

            break;

×

534

          opts.len = newDataLen;

×

535

          opts.source = pRaw->pData;

×

536

          opts.result = newData;

×

537

          opts.unitLen = 16;

×

538

          tstrncpy(opts.key, tsEncryptKey, ENCRYPT_KEY_LEN + 1);

×

540

          int32_t count = CBC_Encrypt(&opts);

×

546

        if (taosWriteFile(pFile, newData, newDataLen) != newDataLen) {

×

547

          code = terrno;

×

548

          taosHashCancelIterate(hash, ppRow);

×

549

          sdbFreeRaw(pRaw);

×

550

          break;

×

553

        if (tsiEncryptAlgorithm == DND_CA_SM4 && (tsiEncryptScope & DND_CS_SDB) == DND_CS_SDB) {

×

554

          taosMemoryFree(newData);

×

557

        int32_t cksum = taosCalcChecksum(0, (const uint8_t *)pRaw, sizeof(SSdbRaw) + pRaw->dataLen);

×

558

        if (taosWriteFile(pFile, &cksum, sizeof(int32_t)) != sizeof(int32_t)) {

×

559

          code = terrno;

×

560

          taosHashCancelIterate(hash, ppRow);

×

561

          sdbFreeRaw(pRaw);

×

562

          break;

×

565

        code = TSDB_CODE_APP_ERROR;

×

566

        taosHashCancelIterate(hash, ppRow);

×

567

        break;

×

570

      sdbFreeRaw(pRaw);

×

571

      ppRow = taosHashIterate(hash, ppRow);

×

573

    sdbUnLock(pSdb, i);

×

576

  if (code == 0) {

×

577

    code = taosFsyncFile(pFile);

×

578

    if (code != 0) {

×

579

      code = TAOS_SYSTEM_ERROR(ERRNO);

×

580

      mError("failed to sync sdb file:%s since %s", tmpfile, tstrerror(code));

×

584

  if (taosCloseFile(&pFile) != 0) {

×

585

    code = taosRenameFile(tmpfile, curfile);

×

588

  if (code == 0) {

×

589

    code = taosRenameFile(tmpfile, curfile);

×

590

    if (code != 0) {

×

591

      mError("failed to write sdb file:%s since %s", curfile, tstrerror(code));

×

595

  if (code != 0) {

×

596

    mError("failed to write sdb file:%s since %s", curfile, tstrerror(code));

×

598

    pSdb->commitIndex = pSdb->applyIndex;

×

599

    pSdb->commitTerm = pSdb->applyTerm;

×

600

    pSdb->commitConfig = pSdb->applyConfig;

×

601

    mInfo("vgId:1, trans:0, write sdb file success, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64

×

606

  terrno = code;

×

607

  return code;

×

610

int32_t sdbWriteFile(SSdb *pSdb, int32_t delta) {

×

611

  int32_t code = 0;

×

612

  if (pSdb->applyIndex == pSdb->commitIndex) {

×

613

    return 0;

×

616

  if (pSdb->applyIndex - pSdb->commitIndex < delta) {

×

617

    return 0;

×

620

  (void)taosThreadMutexLock(&pSdb->filelock);

×

621

  if (pSdb->pWal != NULL) {

×

622

    if (pSdb->sync > 0) {

×

623

      code = syncBeginSnapshot(pSdb->sync, pSdb->applyIndex);

×

626

  if (code == 0) {

×

627

    code = sdbWriteFileImp(pSdb, -1);

×

629

  if (code == 0) {

×

630

    if (pSdb->pWal != NULL) {

×

631

      if (pSdb->sync > 0) {

×

632

        code = syncEndSnapshot(pSdb->sync);

×

636

  if (code != 0) {

×

637

    mError("failed to write sdb file since %s", tstrerror(code));

×

639

    mInfo("vgId:1, trans:0, write sdb file success, apply index:%" PRId64 ", term:%" PRId64 ", config:%" PRId64,

×

642

  (void)taosThreadMutexUnlock(&pSdb->filelock);

×

643

  return code;

×

646

int32_t sdbWriteFileForDump(SSdb *pSdb) {

×

647

  int32_t code = 0;

×

649

  code = sdbWriteFileImp(pSdb, 0);

×

651

  return code;

×

654

int32_t sdbDeploy(SSdb *pSdb) {

×

655

  int32_t code = 0;

×

656

  code = sdbDeployData(pSdb);

×

657

  if (code != 0) {

×

658

    TAOS_RETURN(code);

×

661

  code = sdbWriteFile(pSdb, 0);

×

662

  if (code != 0) {

×

663

    TAOS_RETURN(code);

×

666

  return 0;

×

669

int32_t sdbAfterRestored(SSdb *pSdb) {

×

670

  int32_t code = 0;

×

671

  code = sdbAfterRestoredData(pSdb);

×

672

  if (code != 0) {

×

673

    TAOS_RETURN(code);

×

675

  return 0;

×

678

static SSdbIter *sdbCreateIter(SSdb *pSdb) {

×

679

  SSdbIter *pIter = taosMemoryCalloc(1, sizeof(SSdbIter));

×

680

  if (pIter == NULL) {

×

681

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

682

    return NULL;

×

685

  char name[PATH_MAX + 100] = {0};

×

686

  snprintf(name, sizeof(name), "%s%ssdb.data.%" PRIu64, pSdb->tmpDir, TD_DIRSEP, (uint64_t)pIter);

×

687

  pIter->name = taosStrdup(name);

×

688

  if (pIter->name == NULL) {

×

689

    taosMemoryFree(pIter);

×

690

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

691

    return NULL;

×

694

  return pIter;

×

697

static void sdbCloseIter(SSdbIter *pIter) {

×

698

  if (pIter == NULL) return;

×

700

  if (pIter->file != NULL) {

×

701

    int32_t ret = 0;

×

702

    if ((ret = taosCloseFile(&pIter->file)) != 0) {

×

703

      mError("failed to close sdb file since %s", tstrerror(ret));

×

705

    pIter->file = NULL;

×

708

  if (pIter->name != NULL) {

×

709

    int32_t ret = 0;

×

710

    if ((ret = taosRemoveFile(pIter->name)) != 0) {

×

711

      mError("failed to remove sdb file:%s since %s", pIter->name, tstrerror(ret));

×

713

    taosMemoryFree(pIter->name);

×

714

    pIter->name = NULL;

×

717

  mInfo("sdbiter:%p, is closed, total:%" PRId64, pIter, pIter->total);

×

718

  taosMemoryFree(pIter);

×

721

int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter, int64_t *index, int64_t *term, int64_t *config) {

×

722

  int32_t   code = 0;

×

723

  SSdbIter *pIter = sdbCreateIter(pSdb);

×

724

  if (pIter == NULL) return -1;

×

726

  char datafile[PATH_MAX] = {0};

×

727

  snprintf(datafile, sizeof(datafile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);

×

729

  (void)taosThreadMutexLock(&pSdb->filelock);

×

730

  int64_t commitIndex = pSdb->commitIndex;

×

731

  int64_t commitTerm = pSdb->commitTerm;

×

732

  int64_t commitConfig = pSdb->commitConfig;

×

733

  if (taosCopyFile(datafile, pIter->name) < 0) {

×

734

    code = terrno;

×

735

    (void)taosThreadMutexUnlock(&pSdb->filelock);

×

736

    mError("failed to copy sdb file %s to %s since %s", datafile, pIter->name, tstrerror(code));

×

737

    sdbCloseIter(pIter);

×

738

    TAOS_RETURN(code);

×

740

  (void)taosThreadMutexUnlock(&pSdb->filelock);

×

742

  pIter->file = taosOpenFile(pIter->name, TD_FILE_READ);

×

743

  if (pIter->file == NULL) {

×

744

    code = terrno;

×

745

    mError("failed to open sdb file:%s since %s", pIter->name, tstrerror(code));

×

746

    sdbCloseIter(pIter);

×

747

    TAOS_RETURN(code);

×

750

  *ppIter = pIter;

×

751

  if (index != NULL) *index = commitIndex;

×

752

  if (term != NULL) *term = commitTerm;

×

753

  if (config != NULL) *config = commitConfig;

×

755

  mInfo("sdbiter:%p, is created to read snapshot, commit index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " file:%s",

×

757

  return 0;

×

760

void sdbStopRead(SSdb *pSdb, SSdbIter *pIter) { sdbCloseIter(pIter); }

×

762

int32_t sdbDoRead(SSdb *pSdb, SSdbIter *pIter, void **ppBuf, int32_t *len) {

×

763

  int32_t code = 0;

×

764

  int32_t maxlen = 4096;

×

765

  void   *pBuf = taosMemoryCalloc(1, maxlen);

×

766

  if (pBuf == NULL) {

×

767

    code = terrno;

×

768

    TAOS_RETURN(code);

×

771

  int32_t readlen = taosReadFile(pIter->file, pBuf, maxlen);

×

772

  if (readlen < 0 || readlen > maxlen) {

×

773

    code = terrno;

×

774

    mError("sdbiter:%p, failed to read snapshot since %s, total:%" PRId64, pIter, tstrerror(code), pIter->total);

×

775

    *ppBuf = NULL;

×

776

    *len = 0;

×

777

    taosMemoryFree(pBuf);

×

778

    TAOS_RETURN(code);

×

779

  } else if (readlen == 0) {

×

780

    mInfo("sdbiter:%p, read snapshot to the end, total:%" PRId64, pIter, pIter->total);

×

781

    *ppBuf = NULL;

×

782

    *len = 0;

×

783

    taosMemoryFree(pBuf);

×

784

    return 0;

×

786

    pIter->total += readlen;

×

787

    mInfo("sdbiter:%p, read:%d bytes from snapshot, total:%" PRId64, pIter, readlen, pIter->total);

×

788

    *ppBuf = pBuf;

×

789

    *len = readlen;

×

790

    return 0;

×

794

int32_t sdbStartWrite(SSdb *pSdb, SSdbIter **ppIter) {

×

795

  int32_t   code = 0;

×

796

  SSdbIter *pIter = sdbCreateIter(pSdb);

×

797

  if (pIter == NULL) {

×

798

    code = terrno;

×

799

    TAOS_RETURN(code);

×

802

  pIter->file = taosOpenFile(pIter->name, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);

×

803

  if (pIter->file == NULL) {

×

804

    code = terrno;

×

805

    mError("failed to open %s since %s", pIter->name, tstrerror(code));

×

806

    sdbCloseIter(pIter);

×

807

    TAOS_RETURN(code);

×

810

  *ppIter = pIter;

×

811

  mInfo("sdbiter:%p, is created to write snapshot, file:%s", pIter, pIter->name);

×

812

  return 0;

×

815

int32_t sdbStopWrite(SSdb *pSdb, SSdbIter *pIter, bool isApply, int64_t index, int64_t term, int64_t config) {

×

816

  int32_t code = -1;

×

818

  if (!isApply) {

×

819

    mInfo("sdbiter:%p, not apply to sdb", pIter);

×

820

    code = 0;

×

821

    goto _OVER;

×

824

  if (taosFsyncFile(pIter->file) != 0) {

×

825

    code = TAOS_SYSTEM_ERROR(ERRNO);

×

826

    mError("sdbiter:%p, failed to fasync file %s since %s", pIter, pIter->name, tstrerror(code));

×

827

    goto _OVER;

×

830

  if (taosCloseFile(&pIter->file) != 0) {

×

831

    code = TAOS_SYSTEM_ERROR(ERRNO);

×

832

    goto _OVER;

×

834

  pIter->file = NULL;

×

836

  char datafile[PATH_MAX] = {0};

×

837

  snprintf(datafile, sizeof(datafile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);

×

838

  code = taosRenameFile(pIter->name, datafile);

×

839

  if (code != 0) {

×

840

    mError("sdbiter:%p, failed to rename file %s to %s since %s", pIter, pIter->name, datafile, tstrerror(code));

×

841

    goto _OVER;

×

844

  code = sdbReadFile(pSdb);

×

845

  if (code != 0) {

×

846

    mError("sdbiter:%p, failed to read from %s since %s", pIter, datafile, tstrerror(code));

×

847

    goto _OVER;

×

850

  if (config > 0) {

×

851

    pSdb->commitConfig = config;

×

853

  if (term > 0) {

×

854

    pSdb->commitTerm = term;

×

856

  if (index > 0) {

×

857

    pSdb->commitIndex = index;

×

860

  mInfo("sdbiter:%p, success applyed to sdb", pIter);

×

861

  code = 0;

×

863

_OVER:

×

864

  sdbCloseIter(pIter);

×

865

  return code;

×

868

int32_t sdbDoWrite(SSdb *pSdb, SSdbIter *pIter, void *pBuf, int32_t len) {

×

869

  int32_t code = 0;

×

870

  int32_t writelen = taosWriteFile(pIter->file, pBuf, len);

×

871

  if (writelen != len) {

×

872

    code = terrno;

×

873

    mError("failed to write len:%d since %s, total:%" PRId64, len, tstrerror(code), pIter->total);

×

874

    TAOS_RETURN(code);

×

877

  pIter->total += writelen;

×

878

  mInfo("sdbiter:%p, write:%d bytes to snapshot, total:%" PRId64, pIter, writelen, pIter->total);

×

879

  return 0;

×

taosdata / TDengine / #3798

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

taosdata / TDengine / #3798

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