tdbPager.c 25.2 KB
Newer Older
H
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
/*
 * 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/>.
H
refact  
Hongze Cheng 已提交
14 15
 */

H
Hongze Cheng 已提交
16 17
#include "tdbInt.h"

H
Hongze Cheng 已提交
18
#pragma pack(push, 1)
wafwerar's avatar
wafwerar 已提交
19
typedef struct {
H
Hongze Cheng 已提交
20 21 22 23 24
  u8    hdrString[16];
  u16   pageSize;
  SPgno freePage;
  u32   nFreePages;
  u8    reserved[102];
H
Hongze Cheng 已提交
25
} SFileHdr;
wafwerar's avatar
wafwerar 已提交
26
#pragma pack(pop)
H
Hongze Cheng 已提交
27

H
Hongze Cheng 已提交
28 29
TDB_STATIC_ASSERT(sizeof(SFileHdr) == 128, "Size of file header is not correct");

30
struct hashset_st {
M
Minglei Jin 已提交
31 32 33
  size_t  nbits;
  size_t  mask;
  size_t  capacity;
34
  size_t *items;
M
Minglei Jin 已提交
35 36
  size_t  nitems;
  double  load_factor;
37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70
};

static const unsigned int prime = 39;
static const unsigned int prime2 = 5009;

hashset_t hashset_create(void) {
  hashset_t set = tdbOsCalloc(1, sizeof(struct hashset_st));
  if (!set) {
    return NULL;
  }

  set->nbits = 4;
  set->capacity = (size_t)(1 << set->nbits);
  set->items = tdbOsCalloc(set->capacity, sizeof(size_t));
  if (!set->items) {
    tdbOsFree(set);
    return NULL;
  }
  set->mask = set->capacity - 1;
  set->nitems = 0;

  set->load_factor = 0.75;

  return set;
}

void hashset_destroy(hashset_t set) {
  if (set) {
    tdbOsFree(set->items);
    tdbOsFree(set);
  }
}

int hashset_add_member(hashset_t set, void *item) {
M
Minglei Jin 已提交
71
  size_t value = (size_t)item;
72 73 74
  size_t h;

  if (value == 0) {
M
Minglei Jin 已提交
75
    return -1;
76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105
  }

  for (h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
    if (set->items[h] == value) {
      return 0;
    }
  }

  set->items[h] = value;
  ++set->nitems;
  return 1;
}

int hashset_add(hashset_t set, void *item) {
  int ret = hashset_add_member(set, item);

  size_t old_capacity = set->capacity;
  if (set->nitems >= (double)old_capacity * set->load_factor) {
    size_t *old_items = set->items;
    ++set->nbits;
    set->capacity = (size_t)(1 << set->nbits);
    set->mask = set->capacity - 1;

    set->items = tdbOsCalloc(set->capacity, sizeof(size_t));
    if (!set->items) {
      return -1;
    }

    set->nitems = 0;
    for (size_t i = 0; i < old_capacity; ++i) {
M
Minglei Jin 已提交
106
      hashset_add_member(set, (void *)old_items[i]);
107 108 109 110 111 112 113 114
    }
    tdbOsFree(old_items);
  }

  return ret;
}

int hashset_remove(hashset_t set, void *item) {
M
Minglei Jin 已提交
115
  size_t value = (size_t)item;
116 117 118 119 120 121 122 123 124 125 126 127 128

  for (size_t h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
    if (set->items[h] == value) {
      set->items[h] = 0;
      --set->nitems;
      return 1;
    }
  }

  return 0;
}

int hashset_contains(hashset_t set, void *item) {
M
Minglei Jin 已提交
129
  size_t value = (size_t)item;
130 131 132 133 134 135 136 137 138 139

  for (size_t h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
    if (set->items[h] == value) {
      return 1;
    }
  }

  return 0;
}

H
Hongze Cheng 已提交
140 141
#define TDB_PAGE_INITIALIZED(pPage) ((pPage)->pPager != NULL)

H
Hongze Cheng 已提交
142 143
static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage *, void *, int), void *arg,
                            u8 loadPage);
H
Hongze Cheng 已提交
144
static int tdbPagerWritePageToJournal(SPager *pPager, SPage *pPage);
145
static int tdbPagerPWritePageToDB(SPager *pPager, SPage *pPage);
H
Hongze Cheng 已提交
146

H
Hongze Cheng 已提交
147 148 149
static FORCE_INLINE int32_t pageCmpFn(const SRBTreeNode *lhs, const SRBTreeNode *rhs) {
  SPage *pPageL = (SPage *)(((uint8_t *)lhs) - offsetof(SPage, node));
  SPage *pPageR = (SPage *)(((uint8_t *)rhs) - offsetof(SPage, node));
150 151 152 153 154 155 156 157 158 159 160 161 162

  SPgno pgnoL = TDB_PAGE_PGNO(pPageL);
  SPgno pgnoR = TDB_PAGE_PGNO(pPageR);

  if (pgnoL < pgnoR) {
    return -1;
  } else if (pgnoL > pgnoR) {
    return 1;
  } else {
    return 0;
  }
}

H
refact  
Hongze Cheng 已提交
163
int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
H
more  
Hongze Cheng 已提交
164
  uint8_t *pPtr;
H
Hongze Cheng 已提交
165
  SPager  *pPager;
H
more  
Hongze Cheng 已提交
166 167
  int      fsize;
  int      zsize;
H
Hongze Cheng 已提交
168
  int      ret;
H
more  
Hongze Cheng 已提交
169

H
refact  
Hongze Cheng 已提交
170
  *ppPager = NULL;
H
more  
Hongze Cheng 已提交
171 172

  fsize = strlen(fileName);
H
refact  
Hongze Cheng 已提交
173
  zsize = sizeof(*pPager)  /* SPager */
H
more  
Hongze Cheng 已提交
174 175
          + fsize + 1      /* dbFileName */
          + fsize + 8 + 1; /* jFileName */
H
Hongze Cheng 已提交
176
  pPtr = (uint8_t *)tdbOsCalloc(1, zsize);
H
more  
Hongze Cheng 已提交
177 178 179 180
  if (pPtr == NULL) {
    return -1;
  }

H
refact  
Hongze Cheng 已提交
181 182 183 184 185 186
  pPager = (SPager *)pPtr;
  pPtr += sizeof(*pPager);
  // pPager->dbFileName
  pPager->dbFileName = (char *)pPtr;
  memcpy(pPager->dbFileName, fileName, fsize);
  pPager->dbFileName[fsize] = '\0';
H
more  
Hongze Cheng 已提交
187
  pPtr += fsize + 1;
H
refact  
Hongze Cheng 已提交
188 189 190 191 192 193 194 195
  // pPager->jFileName
  pPager->jFileName = (char *)pPtr;
  memcpy(pPager->jFileName, fileName, fsize);
  memcpy(pPager->jFileName + fsize, "-journal", 8);
  pPager->jFileName[fsize + 8] = '\0';
  // pPager->pCache
  pPager->pCache = pCache;

H
Hongze Cheng 已提交
196
  pPager->fd = tdbOsOpen(pPager->dbFileName, TDB_O_CREAT | TDB_O_RDWR, 0755);
M
Minglei Jin 已提交
197 198
  if (TDB_FD_INVALID(pPager->fd)) {
    // if (pPager->fd < 0) {
H
more  
Hongze Cheng 已提交
199 200 201
    return -1;
  }

wafwerar's avatar
wafwerar 已提交
202
  ret = tdbGnrtFileID(pPager->fd, pPager->fid, false);
H
Hongze Cheng 已提交
203 204 205 206
  if (ret < 0) {
    return -1;
  }

H
Hongze Cheng 已提交
207
  // pPager->jfd = -1;
H
Hongze Cheng 已提交
208
  pPager->pageSize = tdbPCacheGetPageSize(pCache);
H
Hongze Cheng 已提交
209 210
  // pPager->dbOrigSize
  ret = tdbGetFileSize(pPager->fd, pPager->pageSize, &(pPager->dbOrigSize));
H
Hongze Cheng 已提交
211
  pPager->dbFileSize = pPager->dbOrigSize;
H
more  
Hongze Cheng 已提交
212

213
  tdbTrace("pager/open reset dirty tree: %p", &pPager->rbt);
214 215
  tRBTreeCreate(&pPager->rbt, pageCmpFn);

H
refact  
Hongze Cheng 已提交
216
  *ppPager = pPager;
H
more  
Hongze Cheng 已提交
217 218 219
  return 0;
}

H
refact  
Hongze Cheng 已提交
220
int tdbPagerClose(SPager *pPager) {
H
Hongze Cheng 已提交
221
  if (pPager) {
M
Minglei Jin 已提交
222
    /*
H
Hongze Cheng 已提交
223 224 225
    if (pPager->inTran) {
      tdbOsClose(pPager->jfd);
    }
M
Minglei Jin 已提交
226
    */
H
Hongze Cheng 已提交
227 228 229
    tdbOsClose(pPager->fd);
    tdbOsFree(pPager);
  }
H
more  
Hongze Cheng 已提交
230 231
  return 0;
}
H
Hongze Cheng 已提交
232

H
refact  
Hongze Cheng 已提交
233
int tdbPagerWrite(SPager *pPager, SPage *pPage) {
H
Hongze Cheng 已提交
234 235
  int     ret;
  SPage **ppPage;
H
more  
Hongze Cheng 已提交
236

237
  // ASSERT(pPager->inTran);
H
Hongze Cheng 已提交
238
  if (pPage->isDirty) return 0;
H
more  
Hongze Cheng 已提交
239

H
Hongze Cheng 已提交
240
  // ref page one more time so the page will not be release
H
Hongze Cheng 已提交
241
  tdbRefPage(pPage);
242
  tdbTrace("pager/mdirty page %p/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id);
243

H
Hongze Cheng 已提交
244 245
  // Set page as dirty
  pPage->isDirty = 1;
246
  /*
H
Hongze Cheng 已提交
247
  // Add page to dirty list(TODO: NOT use O(n^2) algorithm)
H
Hongze Cheng 已提交
248 249 250
  for (ppPage = &pPager->pDirty; (*ppPage) && TDB_PAGE_PGNO(*ppPage) < TDB_PAGE_PGNO(pPage);
       ppPage = &((*ppPage)->pDirtyNext)) {
  }
251 252 253 254 255 256 257

  if (*ppPage && TDB_PAGE_PGNO(*ppPage) == TDB_PAGE_PGNO(pPage)) {
    tdbUnrefPage(pPage);

    return 0;
  }

258
  ASSERT(*ppPage == NULL || TDB_PAGE_PGNO(*ppPage) > TDB_PAGE_PGNO(pPage));
H
Hongze Cheng 已提交
259 260
  pPage->pDirtyNext = *ppPage;
  *ppPage = pPage;
261
  */
262
  tdbTrace("put page: %p %d to dirty tree: %p", pPage, TDB_PAGE_PGNO(pPage), &pPager->rbt);
263
  tRBTreePut(&pPager->rbt, (SRBTreeNode *)pPage);
H
Hongze Cheng 已提交
264

H
Hongze Cheng 已提交
265
  // Write page to journal if neccessary
M
Minglei Jin 已提交
266
  if (TDB_PAGE_PGNO(pPage) <= pPager->dbOrigSize &&
M
Minglei Jin 已提交
267 268
      (pPager->pActiveTxn->jPageSet == NULL ||
       !hashset_contains(pPager->pActiveTxn->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage))))) {
H
Hongze Cheng 已提交
269 270
    ret = tdbPagerWritePageToJournal(pPager, pPage);
    if (ret < 0) {
271
      tdbError("failed to write page to journal since %s", tstrerror(terrno));
H
Hongze Cheng 已提交
272
      return -1;
H
Hongze Cheng 已提交
273
    }
274

M
Minglei Jin 已提交
275 276
    if (pPager->pActiveTxn->jPageSet) {
      hashset_add(pPager->pActiveTxn->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage)));
277
    }
H
more  
Hongze Cheng 已提交
278
  }
H
Hongze Cheng 已提交
279

H
Hongze Cheng 已提交
280 281 282
  return 0;
}

H
Hongze Cheng 已提交
283
int tdbPagerBegin(SPager *pPager, TXN *pTxn) {
M
Minglei Jin 已提交
284
  /*
H
refact  
Hongze Cheng 已提交
285
  if (pPager->inTran) {
H
more  
Hongze Cheng 已提交
286 287
    return 0;
  }
M
Minglei Jin 已提交
288
  */
H
Hongze Cheng 已提交
289
  // Open the journal
M
Minglei Jin 已提交
290 291 292 293
  char jTxnFileName[TDB_FILENAME_LEN];
  sprintf(jTxnFileName, "%s.%" PRId64, pPager->jFileName, pTxn->txnId);
  pTxn->jfd = tdbOsOpen(jTxnFileName, TDB_O_CREAT | TDB_O_RDWR, 0755);
  if (TDB_FD_INVALID(pTxn->jfd)) {
294 295
    tdbError("failed to open file due to %s. jFileName:%s", strerror(errno), pPager->jFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
296 297 298
    return -1;
  }

M
Minglei Jin 已提交
299
  pTxn->jPageSet = hashset_create();
M
Minglei Jin 已提交
300

M
Minglei Jin 已提交
301
  pPager->pActiveTxn = pTxn;
H
Hongze Cheng 已提交
302
  // TODO: write the size of the file
M
Minglei Jin 已提交
303
  /*
H
refact  
Hongze Cheng 已提交
304
  pPager->inTran = 1;
M
Minglei Jin 已提交
305
  */
H
more  
Hongze Cheng 已提交
306 307 308
  return 0;
}

H
Hongze Cheng 已提交
309
int tdbPagerCommit(SPager *pPager, TXN *pTxn) {
H
Hongze Cheng 已提交
310 311 312
  SPage *pPage;
  int    ret;

H
Hongze Cheng 已提交
313
  // sync the journal file
M
Minglei Jin 已提交
314
  ret = tdbOsFSync(pTxn->jfd);
H
Hongze Cheng 已提交
315
  if (ret < 0) {
M
Minglei Jin 已提交
316
    tdbError("failed to fsync: %s. jFileName:%s, %" PRId64, strerror(errno), pPager->jFileName, pTxn->txnId);
317 318
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
H
Hongze Cheng 已提交
319 320
  }

321
  // loop to write the dirty pages to file
322 323 324 325
  SRBTreeIter  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  SRBTreeNode *pNode = NULL;
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
326

327
    ASSERT(pPage->nOverflow == 0);
328
    ret = tdbPagerPWritePageToDB(pPager, pPage);
329
    if (ret < 0) {
330
      tdbError("failed to write page to db since %s", tstrerror(terrno));
331 332 333 334
      return -1;
    }
  }

335 336
  tdbTrace("tdbttl commit:%p, %d/%d", pPager, pPager->dbOrigSize, pPager->dbFileSize);
  pPager->dbOrigSize = pPager->dbFileSize;
H
Hongze Cheng 已提交
337

H
Hongze Cheng 已提交
338
  // release the page
339 340 341
  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
H
Hongze Cheng 已提交
342 343 344

    pPage->isDirty = 0;

345
    tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
M
Minglei Jin 已提交
346 347
    if (pTxn->jPageSet) {
      hashset_remove(pTxn->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage)));
348
    }
H
Hongze Cheng 已提交
349
    tdbPCacheRelease(pPager->pCache, pPage, pTxn);
H
Hongze Cheng 已提交
350
  }
351

352
  tdbTrace("pager/commit reset dirty tree: %p", &pPager->rbt);
353
  tRBTreeCreate(&pPager->rbt, pageCmpFn);
H
Hongze Cheng 已提交
354 355

  // sync the db file
356 357 358 359 360
  if (tdbOsFSync(pPager->fd) < 0) {
    tdbError("failed to fsync fd due to %s. file:%s", strerror(errno), pPager->dbFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }
H
Hongze Cheng 已提交
361

362 363 364 365
  return 0;
}

int tdbPagerPostCommit(SPager *pPager, TXN *pTxn) {
M
Minglei Jin 已提交
366 367 368
  char jTxnFileName[TDB_FILENAME_LEN];
  sprintf(jTxnFileName, "%s.%" PRId64, pPager->jFileName, pTxn->txnId);

369
  // remove the journal file
M
Minglei Jin 已提交
370 371
  if (tdbOsClose(pTxn->jfd) < 0) {
    tdbError("failed to close jfd: %s. file:%s, %" PRId64, strerror(errno), pPager->jFileName, pTxn->txnId);
372 373 374 375
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

M
Minglei Jin 已提交
376 377
  if (tdbOsRemove(jTxnFileName) < 0 && errno != ENOENT) {
    tdbError("failed to remove file due to %s. file:%s", strerror(errno), jTxnFileName);
378 379 380 381
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

M
Minglei Jin 已提交
382
  // pPager->inTran = 0;
383 384 385 386

  return 0;
}

387 388
int tdbPagerPrepareAsyncCommit(SPager *pPager, TXN *pTxn) {
  SPage *pPage;
M
Minglei Jin 已提交
389
  SPgno  maxPgno = pPager->dbOrigSize;
390 391 392
  int    ret;

  // sync the journal file
M
Minglei Jin 已提交
393
  ret = tdbOsFSync(pTxn->jfd);
394
  if (ret < 0) {
M
Minglei Jin 已提交
395
    tdbError("failed to fsync jfd: %s. jfile:%s, %" PRId64, strerror(errno), pPager->jFileName, pTxn->txnId);
396 397 398 399
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

400 401 402 403 404 405
  // loop to write the dirty pages to file
  SRBTreeIter  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  SRBTreeNode *pNode = NULL;
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
    if (pPage->isLocal) continue;
M
Minglei Jin 已提交
406 407 408 409 410

    SPgno pgno = TDB_PAGE_PGNO(pPage);
    if (pgno > maxPgno) {
      maxPgno = pgno;
    }
411
    ret = tdbPagerPWritePageToDB(pPager, pPage);
412 413 414 415 416 417 418
    if (ret < 0) {
      tdbError("failed to write page to db since %s", tstrerror(terrno));
      return -1;
    }
  }

  tdbTrace("tdbttl commit:%p, %d/%d", pPager, pPager->dbOrigSize, pPager->dbFileSize);
M
Minglei Jin 已提交
419 420
  pPager->dbOrigSize = maxPgno;
  //  pPager->dbOrigSize = pPager->dbFileSize;
421 422 423 424 425 426 427 428 429 430 431

  // release the page
  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
    if (pPage->isLocal) continue;
    pPage->isDirty = 0;

    tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
    tdbPCacheRelease(pPager->pCache, pPage, pTxn);
  }
M
Minglei Jin 已提交
432

433
  /*
434
  tdbTrace("reset dirty tree: %p", &pPager->rbt);
435
  tRBTreeCreate(&pPager->rbt, pageCmpFn);
436

437 438 439 440 441 442 443
  // sync the db file
  if (tdbOsFSync(pPager->fd) < 0) {
    tdbError("failed to fsync fd due to %s. file:%s", strerror(errno), pPager->dbFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }
  */
444 445 446
  return 0;
}

447 448 449 450 451 452 453
// recovery dirty pages
int tdbPagerAbort(SPager *pPager, TXN *pTxn) {
  SPage *pPage;
  int    pgIdx;
  SPgno  journalSize = 0;
  int    ret;

M
Minglei Jin 已提交
454 455
  // sync the journal file
  ret = tdbOsFSync(pTxn->jfd);
456
  if (ret < 0) {
M
Minglei Jin 已提交
457
    tdbError("failed to fsync jfd: %s. jfile:%s, %" PRId64, strerror(errno), pPager->jFileName, pTxn->txnId);
458 459
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
460 461
  }

M
Minglei Jin 已提交
462
  tdb_fd_t jfd = pTxn->jfd;
463 464 465 466 467 468

  ret = tdbGetFileSize(jfd, pPager->pageSize, &journalSize);
  if (ret < 0) {
    return -1;
  }

469 470 471 472 473 474
  if (tdbOsLSeek(jfd, 0L, SEEK_SET) < 0) {
    tdbError("failed to lseek jfd due to %s. file:%s, offset:0", strerror(errno), pPager->dbFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

M
Minglei Jin 已提交
475
  u8 *pageBuf = tdbOsCalloc(1, pPager->pageSize);
476 477 478
  if (pageBuf == NULL) {
    return -1;
  }
479

480
  tdbDebug("tdb/abort: pager:%p,", pPager);
481

482
  for (int pgIndex = 0; pgIndex < journalSize; ++pgIndex) {
483 484 485 486 487
    // read pgno & the page from journal
    SPgno pgno;

    int ret = tdbOsRead(jfd, &pgno, sizeof(pgno));
    if (ret < 0) {
488
      tdbOsFree(pageBuf);
489 490 491
      return -1;
    }

492 493
    tdbTrace("tdb/abort: pgno:%d,", pgno);

494 495
    ret = tdbOsRead(jfd, pageBuf, pPager->pageSize);
    if (ret < 0) {
496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513
      tdbOsFree(pageBuf);
      return -1;
    }

    i64 offset = pPager->pageSize * (pgno - 1);
    if (tdbOsLSeek(pPager->fd, offset, SEEK_SET) < 0) {
      tdbError("failed to lseek fd due to %s. file:%s, offset:%" PRId64, strerror(errno), pPager->dbFileName, offset);
      terrno = TAOS_SYSTEM_ERROR(errno);
      tdbOsFree(pageBuf);
      return -1;
    }

    ret = tdbOsWrite(pPager->fd, pageBuf, pPager->pageSize);
    if (ret < 0) {
      tdbError("failed to write buf due to %s. file: %s, bufsize:%d", strerror(errno), pPager->dbFileName,
               pPager->pageSize);
      terrno = TAOS_SYSTEM_ERROR(errno);
      tdbOsFree(pageBuf);
514 515 516
      return -1;
    }
  }
517 518 519 520 521 522 523 524 525 526

  if (tdbOsFSync(pPager->fd) < 0) {
    tdbError("failed to fsync fd due to %s. dbfile:%s", strerror(errno), pPager->dbFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
    tdbOsFree(pageBuf);
    return -1;
  }

  tdbOsFree(pageBuf);

527
  // 3, release the dirty pages
528 529 530 531
  SRBTreeIter  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  SRBTreeNode *pNode = NULL;
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
532 533 534

    pPage->isDirty = 0;

535
    tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
M
Minglei Jin 已提交
536
    hashset_remove(pTxn->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage)));
M
Minglei Jin 已提交
537
    tdbPCacheMarkFree(pPager->pCache, pPage);
538 539 540
    tdbPCacheRelease(pPager->pCache, pPage, pTxn);
  }

541
  tdbTrace("reset dirty tree: %p", &pPager->rbt);
542 543
  tRBTreeCreate(&pPager->rbt, pageCmpFn);

544
  // 4, remove the journal file
M
Minglei Jin 已提交
545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560
  if (tdbOsClose(pTxn->jfd) < 0) {
    tdbError("failed to close jfd: %s. file:%s, %" PRId64, strerror(errno), pPager->jFileName, pTxn->txnId);
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

  char jTxnFileName[TDB_FILENAME_LEN];
  sprintf(jTxnFileName, "%s.%" PRId64, pPager->jFileName, pTxn->txnId);

  if (tdbOsRemove(jTxnFileName) < 0 && errno != ENOENT) {
    tdbError("failed to remove file due to %s. file:%s", strerror(errno), jTxnFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

  // pPager->inTran = 0;
H
Hongze Cheng 已提交
561

H
more  
Hongze Cheng 已提交
562 563 564
  return 0;
}

565 566
int tdbPagerFlushPage(SPager *pPager, TXN *pTxn) {
  SPage *pPage;
567 568
  i32    nRef;
  SPgno  maxPgno = pPager->dbOrigSize;
569 570 571 572 573 574 575
  int    ret;

  // loop to write the dirty pages to file
  SRBTreeIter  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  SRBTreeNode *pNode = NULL;
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
576 577 578 579 580 581 582 583 584
    nRef = tdbGetPageRef(pPage);
    if (nRef > 1) {
      continue;
    }

    SPgno pgno = TDB_PAGE_PGNO(pPage);
    if (pgno > maxPgno) {
      maxPgno = pgno;
    }
585
    ret = tdbPagerPWritePageToDB(pPager, pPage);
586 587 588 589 590
    if (ret < 0) {
      tdbError("failed to write page to db since %s", tstrerror(terrno));
      return -1;
    }

591
    tdbTrace("tdb/flush:%p, pgno:%d, %d/%d/%d", pPager, pgno, pPager->dbOrigSize, pPager->dbFileSize, maxPgno);
592 593 594 595
    pPager->dbOrigSize = maxPgno;

    pPage->isDirty = 0;

596
    tdbTrace("pager/flush drop page: %p, pgno:%d, from dirty tree: %p", pPage, TDB_PAGE_PGNO(pPage), &pPager->rbt);
597 598 599 600 601 602 603 604
    tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
    tdbPCacheRelease(pPager->pCache, pPage, pTxn);

    break;
  }
  /*
  tdbTrace("tdb/flush:%p, %d/%d/%d", pPager, pPager->dbOrigSize, pPager->dbFileSize, maxPgno);
  pPager->dbOrigSize = maxPgno;
605 606 607 608 609

  // release the page
  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
610 611 612 613
    nRef = tdbGetPageRef(pPage);
    if (nRef > 1) {
      continue;
    }
614 615 616

    pPage->isDirty = 0;

617
    tdbTrace("pager/flush drop page: %p %d from dirty tree: %p", pPage, TDB_PAGE_PGNO(pPage), &pPager->rbt);
618 619 620
    tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
    tdbPCacheRelease(pPager->pCache, pPage, pTxn);
  }
621
  */
622 623 624
  return 0;
}

H
Hongze Cheng 已提交
625
int tdbPagerFetchPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage)(SPage *, void *, int), void *arg,
H
Hongze Cheng 已提交
626
                      TXN *pTxn) {
H
more  
Hongze Cheng 已提交
627 628 629
  SPage *pPage;
  SPgid  pgid;
  int    ret;
H
Hongze Cheng 已提交
630 631
  SPgno  pgno;
  u8     loadPage;
H
more  
Hongze Cheng 已提交
632

H
Hongze Cheng 已提交
633 634
  pgno = *ppgno;
  loadPage = 1;
H
more  
Hongze Cheng 已提交
635

H
Hongze Cheng 已提交
636 637 638 639
  // alloc new page
  if (pgno == 0) {
    loadPage = 0;
    ret = tdbPagerAllocPage(pPager, &pgno);
H
Hongze Cheng 已提交
640
    if (ret < 0) {
641
      ASSERT(0);
H
Hongze Cheng 已提交
642
      return -1;
H
more  
Hongze Cheng 已提交
643
    }
H
more  
Hongze Cheng 已提交
644 645
  }

646
  ASSERT(pgno > 0);
H
Hongze Cheng 已提交
647

H
Hongze Cheng 已提交
648
  // fetch a page container
H
more  
Hongze Cheng 已提交
649
  memcpy(&pgid, pPager->fid, TDB_FILE_ID_LEN);
H
Hongze Cheng 已提交
650
  pgid.pgno = pgno;
651 652
  while ((pPage = tdbPCacheFetch(pPager->pCache, &pgid, pTxn)) == NULL) {
    tdbPagerFlushPage(pPager, pTxn);
H
more  
Hongze Cheng 已提交
653 654
  }

655
  tdbTrace("tdbttl fetch pager:%p", pPage->pPager);
H
Hongze Cheng 已提交
656 657 658 659
  // init page if need
  if (!TDB_PAGE_INITIALIZED(pPage)) {
    ret = tdbPagerInitPage(pPager, pPage, initPage, arg, loadPage);
    if (ret < 0) {
660
      ASSERT(0);
H
Hongze Cheng 已提交
661 662
      return -1;
    }
H
Hongze Cheng 已提交
663
  }
H
more  
Hongze Cheng 已提交
664

H
Hongze Cheng 已提交
665 666 667
  // printf("thread %" PRId64 " pager fetch page %d pgno %d ppage %p\n", taosGetSelfPthreadId(), pPage->id,
  //        TDB_PAGE_PGNO(pPage), pPage);

668 669
  ASSERT(TDB_PAGE_INITIALIZED(pPage));
  ASSERT(pPage->pPager == pPager);
H
more  
Hongze Cheng 已提交
670

H
Hongze Cheng 已提交
671
  *ppgno = pgno;
H
more  
Hongze Cheng 已提交
672 673 674 675
  *ppPage = pPage;
  return 0;
}

H
Hongze Cheng 已提交
676 677 678 679 680
void tdbPagerReturnPage(SPager *pPager, SPage *pPage, TXN *pTxn) {
  tdbPCacheRelease(pPager->pCache, pPage, pTxn);
  // printf("thread %" PRId64 " pager retun page %d pgno %d ppage %p\n", taosGetSelfPthreadId(), pPage->id,
  //        TDB_PAGE_PGNO(pPage), pPage);
}
H
Hongze Cheng 已提交
681

H
more  
Hongze Cheng 已提交
682 683 684 685 686 687 688 689 690 691
static int tdbPagerAllocFreePage(SPager *pPager, SPgno *ppgno) {
  // TODO: Allocate a page from the free list
  return 0;
}

static int tdbPagerAllocNewPage(SPager *pPager, SPgno *ppgno) {
  *ppgno = ++pPager->dbFileSize;
  return 0;
}

H
Hongze Cheng 已提交
692
int tdbPagerAllocPage(SPager *pPager, SPgno *ppgno) {
H
more  
Hongze Cheng 已提交
693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710
  int ret;

  *ppgno = 0;

  // Try to allocate from the free list of the pager
  ret = tdbPagerAllocFreePage(pPager, ppgno);
  if (ret < 0) {
    return -1;
  }

  if (*ppgno != 0) return 0;

  // Allocate the page by extending the pager
  ret = tdbPagerAllocNewPage(pPager, ppgno);
  if (ret < 0) {
    return -1;
  }

711
  ASSERT(*ppgno != 0);
H
more  
Hongze Cheng 已提交
712

H
Hongze Cheng 已提交
713 714 715
  return 0;
}

H
Hongze Cheng 已提交
716 717 718 719 720 721 722 723
static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage *, void *, int), void *arg,
                            u8 loadPage) {
  int   ret;
  int   lcode;
  int   nLoops;
  i64   nRead;
  SPgno pgno;
  int   init = 0;
H
Hongze Cheng 已提交
724

H
Hongze Cheng 已提交
725 726
  lcode = TDB_TRY_LOCK_PAGE(pPage);
  if (lcode == P_LOCK_SUCC) {
H
Hongze Cheng 已提交
727 728 729 730 731
    if (TDB_PAGE_INITIALIZED(pPage)) {
      TDB_UNLOCK_PAGE(pPage);
      return 0;
    }

H
Hongze Cheng 已提交
732 733
    pgno = TDB_PAGE_PGNO(pPage);

734
    tdbTrace("tdb/pager:%p, pgno:%d, loadPage:%d, size:%d", pPager, pgno, loadPage, pPager->dbOrigSize);
H
Hongze Cheng 已提交
735 736 737
    if (loadPage && pgno <= pPager->dbOrigSize) {
      init = 1;

H
Hongze Cheng 已提交
738
      nRead = tdbOsPRead(pPager->fd, pPage->pData, pPage->pageSize, ((i64)pPage->pageSize) * (pgno - 1));
739
      tdbTrace("tdb/pager:%p, pgno:%d, nRead:%" PRId64, pPager, pgno, nRead);
H
Hongze Cheng 已提交
740
      if (nRead < pPage->pageSize) {
741
        ASSERT(0);
742 743
        tdbError("tdb/pager:%p, pgno:%d, nRead:%" PRId64 "pgSize:%" PRId32, pPager, pgno, nRead, pPage->pageSize);
        TDB_UNLOCK_PAGE(pPage);
H
Hongze Cheng 已提交
744 745
        return -1;
      }
H
Hongze Cheng 已提交
746 747
    } else {
      init = 0;
H
Hongze Cheng 已提交
748 749
    }

H
Hongze Cheng 已提交
750
    ret = (*initPage)(pPage, arg, init);
H
Hongze Cheng 已提交
751
    if (ret < 0) {
752
      ASSERT(0);
H
Hongze Cheng 已提交
753 754 755 756 757 758 759
      TDB_UNLOCK_PAGE(pPage);
      return -1;
    }

    pPage->pPager = pPager;

    TDB_UNLOCK_PAGE(pPage);
H
Hongze Cheng 已提交
760
  } else if (lcode == P_LOCK_BUSY) {
H
Hongze Cheng 已提交
761 762 763 764 765 766 767 768 769
    nLoops = 0;
    for (;;) {
      if (TDB_PAGE_INITIALIZED(pPage)) break;
      nLoops++;
      if (nLoops > 1000) {
        sched_yield();
        nLoops = 0;
      }
    }
H
Hongze Cheng 已提交
770
  } else {
771
    ASSERT(0);
H
Hongze Cheng 已提交
772
    return -1;
H
Hongze Cheng 已提交
773 774
  }

H
Hongze Cheng 已提交
775 776 777 778 779 780 781 782 783 784
  return 0;
}

// ---------------------------- Journal manipulation
static int tdbPagerWritePageToJournal(SPager *pPager, SPage *pPage) {
  int   ret;
  SPgno pgno;

  pgno = TDB_PAGE_PGNO(pPage);

M
Minglei Jin 已提交
785
  ret = tdbOsWrite(pPager->pActiveTxn->jfd, &pgno, sizeof(pgno));
H
Hongze Cheng 已提交
786
  if (ret < 0) {
M
Minglei Jin 已提交
787 788
    tdbError("failed to write pgno due to %s. file:%s, pgno:%u, txnId:%" PRId64, strerror(errno), pPager->jFileName,
             pgno, pPager->pActiveTxn->txnId);
789
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
790 791 792
    return -1;
  }

M
Minglei Jin 已提交
793
  ret = tdbOsWrite(pPager->pActiveTxn->jfd, pPage->pData, pPage->pageSize);
H
Hongze Cheng 已提交
794
  if (ret < 0) {
M
Minglei Jin 已提交
795 796
    tdbError("failed to write page data due to %s. file:%s, pageSize:%d, txnId:%" PRId64, strerror(errno),
             pPager->jFileName, pPage->pageSize, pPager->pActiveTxn->txnId);
797
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
798 799 800 801 802
    return -1;
  }

  return 0;
}
803
/*
H
Hongze Cheng 已提交
804 805 806 807
static int tdbPagerWritePageToDB(SPager *pPager, SPage *pPage) {
  i64 offset;
  int ret;

808
  offset = (i64)pPage->pageSize * (TDB_PAGE_PGNO(pPage) - 1);
H
Hongze Cheng 已提交
809
  if (tdbOsLSeek(pPager->fd, offset, SEEK_SET) < 0) {
810
    tdbError("failed to lseek due to %s. file:%s, offset:%" PRId64, strerror(errno), pPager->dbFileName, offset);
811
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
812 813 814
    return -1;
  }

H
Hongze Cheng 已提交
815
  ret = tdbOsWrite(pPager->fd, pPage->pData, pPage->pageSize);
H
Hongze Cheng 已提交
816
  if (ret < 0) {
M
Minglei Jin 已提交
817
    tdbError("failed to write page data due to %s. file:%s, pageSize:%d", strerror(errno), pPager->dbFileName,
M
Minglei Jin 已提交
818
             pPage->pageSize);
819
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
820 821 822
    return -1;
  }

823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839
  return 0;
}
*/
static int tdbPagerPWritePageToDB(SPager *pPager, SPage *pPage) {
  i64 offset;
  int ret;

  offset = (i64)pPage->pageSize * (TDB_PAGE_PGNO(pPage) - 1);

  ret = tdbOsPWrite(pPager->fd, pPage->pData, pPage->pageSize, offset);
  if (ret < 0) {
    tdbError("failed to pwrite page data due to %s. file:%s, pageSize:%d", strerror(errno), pPager->dbFileName,
             pPage->pageSize);
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

H
refact  
Hongze Cheng 已提交
840
  return 0;
841 842
}

843
static int tdbPagerRestore(SPager *pPager, const char *jFileName) {
844
  int   ret = 0;
845
  SPgno journalSize = 0;
846
  u8   *pageBuf = NULL;
847

848
  tdb_fd_t jfd = tdbOsOpen(jFileName, TDB_O_RDWR, 0755);
849
  if (jfd == NULL) {
850 851 852 853 854 855 856 857 858 859 860 861 862 863 864
    return 0;
  }

  ret = tdbGetFileSize(jfd, pPager->pageSize, &journalSize);
  if (ret < 0) {
    return -1;
  }

  pageBuf = tdbOsCalloc(1, pPager->pageSize);
  if (pageBuf == NULL) {
    return -1;
  }

  for (int pgIndex = 0; pgIndex < journalSize; ++pgIndex) {
    // read pgno & the page from journal
865
    SPgno pgno;
866 867 868

    int ret = tdbOsRead(jfd, &pgno, sizeof(pgno));
    if (ret < 0) {
M
Minglei Jin 已提交
869
      tdbOsFree(pageBuf);
870 871 872 873 874
      return -1;
    }

    ret = tdbOsRead(jfd, pageBuf, pPager->pageSize);
    if (ret < 0) {
M
Minglei Jin 已提交
875
      tdbOsFree(pageBuf);
876 877 878
      return -1;
    }

M
Minglei Jin 已提交
879 880
    i64 offset = pPager->pageSize * (pgno - 1);
    if (tdbOsLSeek(pPager->fd, offset, SEEK_SET) < 0) {
881
      tdbError("failed to lseek fd due to %s. file:%s, offset:%" PRId64, strerror(errno), pPager->dbFileName, offset);
882
      terrno = TAOS_SYSTEM_ERROR(errno);
M
Minglei Jin 已提交
883
      tdbOsFree(pageBuf);
M
Minglei Jin 已提交
884 885 886 887 888
      return -1;
    }

    ret = tdbOsWrite(pPager->fd, pageBuf, pPager->pageSize);
    if (ret < 0) {
M
Minglei Jin 已提交
889 890
      tdbError("failed to write buf due to %s. file: %s, bufsize:%d", strerror(errno), pPager->dbFileName,
               pPager->pageSize);
891
      terrno = TAOS_SYSTEM_ERROR(errno);
M
Minglei Jin 已提交
892
      tdbOsFree(pageBuf);
M
Minglei Jin 已提交
893 894
      return -1;
    }
895 896
  }

897 898 899
  if (tdbOsFSync(pPager->fd) < 0) {
    tdbError("failed to fsync fd due to %s. dbfile:%s", strerror(errno), pPager->dbFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
M
Minglei Jin 已提交
900
    tdbOsFree(pageBuf);
901 902
    return -1;
  }
903 904 905

  tdbOsFree(pageBuf);

906 907 908 909 910 911 912 913 914 915 916
  if (tdbOsClose(jfd) < 0) {
    tdbError("failed to close jfd due to %s. jFileName:%s", strerror(errno), pPager->jFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }

  if (tdbOsRemove(pPager->jFileName) < 0 && errno != ENOENT) {
    tdbError("failed to remove file due to %s. jFileName:%s", strerror(errno), pPager->jFileName);
    terrno = TAOS_SYSTEM_ERROR(errno);
    return -1;
  }
917 918 919

  return 0;
}
920

921
int tdbPagerRestoreJournals(SPager *pPager) {
922 923 924 925 926 927 928 929 930 931
  tdbDirEntryPtr pDirEntry;
  tdbDirPtr      pDir = taosOpenDir(pPager->pEnv->dbName);
  if (pDir == NULL) {
    tdbError("failed to open %s since %s", pPager->pEnv->dbName, strerror(errno));
    return -1;
  }

  while ((pDirEntry = tdbReadDir(pDir)) != NULL) {
    char *name = tdbDirEntryBaseName(tdbGetDirEntryName(pDirEntry));
    if (strncmp(TDB_MAINDB_NAME "-journal", name, 16) == 0) {
932
      if (tdbPagerRestore(pPager, name) < 0) {
M
Minglei Jin 已提交
933 934
        tdbCloseDir(&pDir);

935 936 937 938 939 940 941 942 943 944 945
        tdbError("failed to restore file due to %s. jFileName:%s", strerror(errno), name);
        return -1;
      }
    }
  }

  tdbCloseDir(&pDir);

  return 0;
}

946
int tdbPagerRollback(SPager *pPager) {
947 948 949 950
  tdbDirEntryPtr pDirEntry;
  tdbDirPtr      pDir = taosOpenDir(pPager->pEnv->dbName);
  if (pDir == NULL) {
    tdbError("failed to open %s since %s", pPager->pEnv->dbName, strerror(errno));
951 952 953
    return -1;
  }

954 955 956 957 958
  while ((pDirEntry = tdbReadDir(pDir)) != NULL) {
    char *name = tdbDirEntryBaseName(tdbGetDirEntryName(pDirEntry));

    if (strncmp(TDB_MAINDB_NAME "-journal", name, 16) == 0) {
      if (tdbOsRemove(name) < 0 && errno != ENOENT) {
M
Minglei Jin 已提交
959 960
        tdbCloseDir(&pDir);

961 962 963 964 965 966 967 968 969
        tdbError("failed to remove file due to %s. jFileName:%s", strerror(errno), name);
        terrno = TAOS_SYSTEM_ERROR(errno);
        return -1;
      }
    }
  }

  tdbCloseDir(&pDir);

970 971
  return 0;
}