tdbPager.c 26.0 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;
302 303 304

  tdbDebug("pager/begin: %p, %d/%d, txnId:%" PRId64, pPager, pPager->dbOrigSize, pPager->dbFileSize, pTxn->txnId);

H
Hongze Cheng 已提交
305
  // TODO: write the size of the file
M
Minglei Jin 已提交
306
  /*
H
refact  
Hongze Cheng 已提交
307
  pPager->inTran = 1;
M
Minglei Jin 已提交
308
  */
H
more  
Hongze Cheng 已提交
309 310 311
  return 0;
}

H
Hongze Cheng 已提交
312
int tdbPagerCommit(SPager *pPager, TXN *pTxn) {
H
Hongze Cheng 已提交
313 314 315
  SPage *pPage;
  int    ret;

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

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

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

338 339
  tdbDebug("pager/commit: %p, %d/%d, txnId:%" PRId64, pPager, pPager->dbOrigSize, pPager->dbFileSize, pTxn->txnId);

340
  pPager->dbOrigSize = pPager->dbFileSize;
H
Hongze Cheng 已提交
341

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

    pPage->isDirty = 0;

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

356
  tdbTrace("pager/commit reset dirty tree: %p", &pPager->rbt);
357
  tRBTreeCreate(&pPager->rbt, pageCmpFn);
H
Hongze Cheng 已提交
358 359

  // sync the db file
360 361 362 363 364
  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 已提交
365

366 367 368 369
  return 0;
}

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

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

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

M
Minglei Jin 已提交
386
  // pPager->inTran = 0;
387

388 389
  tdbDebug("pager/post-commit:%p, %d/%d", pPager, pPager->dbOrigSize, pPager->dbFileSize);

390 391 392
  return 0;
}

393 394
int tdbPagerPrepareAsyncCommit(SPager *pPager, TXN *pTxn) {
  SPage *pPage;
M
Minglei Jin 已提交
395
  SPgno  maxPgno = pPager->dbOrigSize;
396 397 398
  int    ret;

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

406 407 408 409 410 411
  // 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 已提交
412 413 414 415 416

    SPgno pgno = TDB_PAGE_PGNO(pPage);
    if (pgno > maxPgno) {
      maxPgno = pgno;
    }
417
    ret = tdbPagerPWritePageToDB(pPager, pPage);
418 419 420 421 422 423 424
    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 已提交
425 426
  pPager->dbOrigSize = maxPgno;
  //  pPager->dbOrigSize = pPager->dbFileSize;
427 428 429 430 431 432 433 434 435 436 437

  // 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 已提交
438

439
  /*
440
  tdbTrace("reset dirty tree: %p", &pPager->rbt);
441
  tRBTreeCreate(&pPager->rbt, pageCmpFn);
442

443 444 445 446 447 448 449
  // 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;
  }
  */
450 451 452
  return 0;
}

453 454 455 456 457 458 459
// recovery dirty pages
int tdbPagerAbort(SPager *pPager, TXN *pTxn) {
  SPage *pPage;
  int    pgIdx;
  SPgno  journalSize = 0;
  int    ret;

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

M
Minglei Jin 已提交
468
  tdb_fd_t jfd = pTxn->jfd;
469 470 471 472 473 474

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

475 476 477 478 479 480
  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 已提交
481
  u8 *pageBuf = tdbOsCalloc(1, pPager->pageSize);
482 483 484
  if (pageBuf == NULL) {
    return -1;
  }
485

486
  tdbDebug("pager/abort: %p, %d/%d, txnId:%" PRId64, pPager, pPager->dbOrigSize, pPager->dbFileSize, pTxn->txnId);
487

488
  for (int pgIndex = 0; pgIndex < journalSize; ++pgIndex) {
489 490 491 492 493
    // read pgno & the page from journal
    SPgno pgno;

    int ret = tdbOsRead(jfd, &pgno, sizeof(pgno));
    if (ret < 0) {
494
      tdbOsFree(pageBuf);
495 496 497
      return -1;
    }

498 499 500
    tdbTrace("pager/abort: restore pgno:%d,", pgno);

    tdbPCacheInvalidatePage(pPager->pCache, pPager, pgno);
501

502 503
    ret = tdbOsRead(jfd, pageBuf, pPager->pageSize);
    if (ret < 0) {
504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521
      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);
522 523 524
      return -1;
    }
  }
525 526 527 528 529 530 531 532 533 534

  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);

535
  // 3, release the dirty pages
536 537 538 539
  SRBTreeIter  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  SRBTreeNode *pNode = NULL;
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
540 541 542
    SPgno pgno = TDB_PAGE_PGNO(pPage);

    tdbTrace("pager/abort: drop dirty pgno:%d,", pgno);
543 544 545

    pPage->isDirty = 0;

546
    tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
M
Minglei Jin 已提交
547
    hashset_remove(pTxn->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage)));
M
Minglei Jin 已提交
548
    tdbPCacheMarkFree(pPager->pCache, pPage);
549 550 551
    tdbPCacheRelease(pPager->pCache, pPage, pTxn);
  }

552
  tdbTrace("pager/abort: reset dirty tree: %p", &pPager->rbt);
553 554
  tRBTreeCreate(&pPager->rbt, pageCmpFn);

555
  // 4, remove the journal file
M
Minglei Jin 已提交
556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571
  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 已提交
572

H
more  
Hongze Cheng 已提交
573 574 575
  return 0;
}

576 577
int tdbPagerFlushPage(SPager *pPager, TXN *pTxn) {
  SPage *pPage;
578 579
  i32    nRef;
  SPgno  maxPgno = pPager->dbOrigSize;
580 581 582 583 584 585 586
  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;
587 588 589 590 591 592 593 594 595
    nRef = tdbGetPageRef(pPage);
    if (nRef > 1) {
      continue;
    }

    SPgno pgno = TDB_PAGE_PGNO(pPage);
    if (pgno > maxPgno) {
      maxPgno = pgno;
    }
596
    ret = tdbPagerPWritePageToDB(pPager, pPage);
597 598 599 600 601
    if (ret < 0) {
      tdbError("failed to write page to db since %s", tstrerror(terrno));
      return -1;
    }

602
    tdbTrace("tdb/flush:%p, pgno:%d, %d/%d/%d", pPager, pgno, pPager->dbOrigSize, pPager->dbFileSize, maxPgno);
603 604 605 606
    pPager->dbOrigSize = maxPgno;

    pPage->isDirty = 0;

607
    tdbTrace("pager/flush drop page: %p, pgno:%d, from dirty tree: %p", pPage, TDB_PAGE_PGNO(pPage), &pPager->rbt);
608 609 610 611 612
    tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
    tdbPCacheRelease(pPager->pCache, pPage, pTxn);

    break;
  }
613 614 615

  tdbDebug("pager/flush: %p, %d/%d, txnId:%" PRId64, pPager, pPager->dbOrigSize, pPager->dbFileSize, pTxn->txnId);

616 617 618
  /*
  tdbTrace("tdb/flush:%p, %d/%d/%d", pPager, pPager->dbOrigSize, pPager->dbFileSize, maxPgno);
  pPager->dbOrigSize = maxPgno;
619 620 621 622 623

  // release the page
  iter = tRBTreeIterCreate(&pPager->rbt, 1);
  while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
    pPage = (SPage *)pNode;
624 625 626 627
    nRef = tdbGetPageRef(pPage);
    if (nRef > 1) {
      continue;
    }
628 629 630

    pPage->isDirty = 0;

631
    tdbTrace("pager/flush drop page: %p %d from dirty tree: %p", pPage, TDB_PAGE_PGNO(pPage), &pPager->rbt);
632 633 634
    tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
    tdbPCacheRelease(pPager->pCache, pPage, pTxn);
  }
635
  */
636 637 638
  return 0;
}

H
Hongze Cheng 已提交
639
int tdbPagerFetchPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage)(SPage *, void *, int), void *arg,
H
Hongze Cheng 已提交
640
                      TXN *pTxn) {
H
more  
Hongze Cheng 已提交
641 642 643
  SPage *pPage;
  SPgid  pgid;
  int    ret;
H
Hongze Cheng 已提交
644 645
  SPgno  pgno;
  u8     loadPage;
H
more  
Hongze Cheng 已提交
646

H
Hongze Cheng 已提交
647 648
  pgno = *ppgno;
  loadPage = 1;
H
more  
Hongze Cheng 已提交
649

H
Hongze Cheng 已提交
650 651 652 653
  // alloc new page
  if (pgno == 0) {
    loadPage = 0;
    ret = tdbPagerAllocPage(pPager, &pgno);
H
Hongze Cheng 已提交
654
    if (ret < 0) {
655
      ASSERT(0);
H
Hongze Cheng 已提交
656
      return -1;
H
more  
Hongze Cheng 已提交
657
    }
H
more  
Hongze Cheng 已提交
658 659
  }

660
  ASSERT(pgno > 0);
H
Hongze Cheng 已提交
661

H
Hongze Cheng 已提交
662
  // fetch a page container
H
more  
Hongze Cheng 已提交
663
  memcpy(&pgid, pPager->fid, TDB_FILE_ID_LEN);
H
Hongze Cheng 已提交
664
  pgid.pgno = pgno;
665 666
  while ((pPage = tdbPCacheFetch(pPager->pCache, &pgid, pTxn)) == NULL) {
    tdbPagerFlushPage(pPager, pTxn);
H
more  
Hongze Cheng 已提交
667 668
  }

669
  tdbTrace("tdbttl fetch pager:%p", pPage->pPager);
H
Hongze Cheng 已提交
670 671 672 673
  // init page if need
  if (!TDB_PAGE_INITIALIZED(pPage)) {
    ret = tdbPagerInitPage(pPager, pPage, initPage, arg, loadPage);
    if (ret < 0) {
674
      ASSERT(0);
H
Hongze Cheng 已提交
675 676
      return -1;
    }
H
Hongze Cheng 已提交
677
  }
H
more  
Hongze Cheng 已提交
678

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

682 683
  ASSERT(TDB_PAGE_INITIALIZED(pPage));
  ASSERT(pPage->pPager == pPager);
H
more  
Hongze Cheng 已提交
684

H
Hongze Cheng 已提交
685
  *ppgno = pgno;
H
more  
Hongze Cheng 已提交
686 687 688 689
  *ppPage = pPage;
  return 0;
}

H
Hongze Cheng 已提交
690 691 692 693 694
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 已提交
695

H
more  
Hongze Cheng 已提交
696 697 698 699 700 701 702 703 704 705
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 已提交
706
int tdbPagerAllocPage(SPager *pPager, SPgno *ppgno) {
H
more  
Hongze Cheng 已提交
707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724
  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;
  }

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

H
Hongze Cheng 已提交
727 728 729
  return 0;
}

H
Hongze Cheng 已提交
730 731 732 733 734 735 736 737
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 已提交
738

H
Hongze Cheng 已提交
739 740
  lcode = TDB_TRY_LOCK_PAGE(pPage);
  if (lcode == P_LOCK_SUCC) {
H
Hongze Cheng 已提交
741 742 743 744 745
    if (TDB_PAGE_INITIALIZED(pPage)) {
      TDB_UNLOCK_PAGE(pPage);
      return 0;
    }

H
Hongze Cheng 已提交
746 747
    pgno = TDB_PAGE_PGNO(pPage);

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

H
Hongze Cheng 已提交
752
      nRead = tdbOsPRead(pPager->fd, pPage->pData, pPage->pageSize, ((i64)pPage->pageSize) * (pgno - 1));
753
      tdbTrace("tdb/pager:%p, pgno:%d, nRead:%" PRId64, pPager, pgno, nRead);
H
Hongze Cheng 已提交
754
      if (nRead < pPage->pageSize) {
755
        ASSERT(0);
756 757
        tdbError("tdb/pager:%p, pgno:%d, nRead:%" PRId64 "pgSize:%" PRId32, pPager, pgno, nRead, pPage->pageSize);
        TDB_UNLOCK_PAGE(pPage);
H
Hongze Cheng 已提交
758 759
        return -1;
      }
H
Hongze Cheng 已提交
760 761
    } else {
      init = 0;
H
Hongze Cheng 已提交
762 763
    }

H
Hongze Cheng 已提交
764
    ret = (*initPage)(pPage, arg, init);
H
Hongze Cheng 已提交
765
    if (ret < 0) {
766
      ASSERT(0);
H
Hongze Cheng 已提交
767 768 769 770 771 772 773
      TDB_UNLOCK_PAGE(pPage);
      return -1;
    }

    pPage->pPager = pPager;

    TDB_UNLOCK_PAGE(pPage);
H
Hongze Cheng 已提交
774
  } else if (lcode == P_LOCK_BUSY) {
H
Hongze Cheng 已提交
775 776 777 778 779 780 781 782 783
    nLoops = 0;
    for (;;) {
      if (TDB_PAGE_INITIALIZED(pPage)) break;
      nLoops++;
      if (nLoops > 1000) {
        sched_yield();
        nLoops = 0;
      }
    }
H
Hongze Cheng 已提交
784
  } else {
785
    ASSERT(0);
H
Hongze Cheng 已提交
786
    return -1;
H
Hongze Cheng 已提交
787 788
  }

H
Hongze Cheng 已提交
789 790 791 792 793 794 795 796 797 798
  return 0;
}

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

  pgno = TDB_PAGE_PGNO(pPage);

M
Minglei Jin 已提交
799
  ret = tdbOsWrite(pPager->pActiveTxn->jfd, &pgno, sizeof(pgno));
H
Hongze Cheng 已提交
800
  if (ret < 0) {
M
Minglei Jin 已提交
801 802
    tdbError("failed to write pgno due to %s. file:%s, pgno:%u, txnId:%" PRId64, strerror(errno), pPager->jFileName,
             pgno, pPager->pActiveTxn->txnId);
803
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
804 805 806
    return -1;
  }

M
Minglei Jin 已提交
807
  ret = tdbOsWrite(pPager->pActiveTxn->jfd, pPage->pData, pPage->pageSize);
H
Hongze Cheng 已提交
808
  if (ret < 0) {
M
Minglei Jin 已提交
809 810
    tdbError("failed to write page data due to %s. file:%s, pageSize:%d, txnId:%" PRId64, strerror(errno),
             pPager->jFileName, pPage->pageSize, pPager->pActiveTxn->txnId);
811
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
812 813 814 815 816
    return -1;
  }

  return 0;
}
817
/*
H
Hongze Cheng 已提交
818 819 820 821
static int tdbPagerWritePageToDB(SPager *pPager, SPage *pPage) {
  i64 offset;
  int ret;

822
  offset = (i64)pPage->pageSize * (TDB_PAGE_PGNO(pPage) - 1);
H
Hongze Cheng 已提交
823
  if (tdbOsLSeek(pPager->fd, offset, SEEK_SET) < 0) {
824
    tdbError("failed to lseek due to %s. file:%s, offset:%" PRId64, strerror(errno), pPager->dbFileName, offset);
825
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
826 827 828
    return -1;
  }

H
Hongze Cheng 已提交
829
  ret = tdbOsWrite(pPager->fd, pPage->pData, pPage->pageSize);
H
Hongze Cheng 已提交
830
  if (ret < 0) {
M
Minglei Jin 已提交
831
    tdbError("failed to write page data due to %s. file:%s, pageSize:%d", strerror(errno), pPager->dbFileName,
M
Minglei Jin 已提交
832
             pPage->pageSize);
833
    terrno = TAOS_SYSTEM_ERROR(errno);
H
Hongze Cheng 已提交
834 835 836
    return -1;
  }

837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853
  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 已提交
854
  return 0;
855 856
}

857
static int tdbPagerRestore(SPager *pPager, const char *jFileName) {
858
  int   ret = 0;
859
  SPgno journalSize = 0;
860
  u8   *pageBuf = NULL;
861

862
  tdb_fd_t jfd = tdbOsOpen(jFileName, TDB_O_RDWR, 0755);
863
  if (jfd == NULL) {
864 865 866 867 868 869 870 871
    return 0;
  }

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

M
Minglei Jin 已提交
872 873 874 875 876 877
  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;
  }

878 879 880 881 882 883 884
  pageBuf = tdbOsCalloc(1, pPager->pageSize);
  if (pageBuf == NULL) {
    return -1;
  }

  for (int pgIndex = 0; pgIndex < journalSize; ++pgIndex) {
    // read pgno & the page from journal
885
    SPgno pgno;
886 887 888

    int ret = tdbOsRead(jfd, &pgno, sizeof(pgno));
    if (ret < 0) {
M
Minglei Jin 已提交
889
      tdbOsFree(pageBuf);
890 891 892 893 894
      return -1;
    }

    ret = tdbOsRead(jfd, pageBuf, pPager->pageSize);
    if (ret < 0) {
M
Minglei Jin 已提交
895
      tdbOsFree(pageBuf);
896 897 898
      return -1;
    }

M
Minglei Jin 已提交
899 900
    i64 offset = pPager->pageSize * (pgno - 1);
    if (tdbOsLSeek(pPager->fd, offset, SEEK_SET) < 0) {
901
      tdbError("failed to lseek fd due to %s. file:%s, offset:%" PRId64, strerror(errno), pPager->dbFileName, offset);
902
      terrno = TAOS_SYSTEM_ERROR(errno);
M
Minglei Jin 已提交
903
      tdbOsFree(pageBuf);
M
Minglei Jin 已提交
904 905 906 907 908
      return -1;
    }

    ret = tdbOsWrite(pPager->fd, pageBuf, pPager->pageSize);
    if (ret < 0) {
M
Minglei Jin 已提交
909 910
      tdbError("failed to write buf due to %s. file: %s, bufsize:%d", strerror(errno), pPager->dbFileName,
               pPager->pageSize);
911
      terrno = TAOS_SYSTEM_ERROR(errno);
M
Minglei Jin 已提交
912
      tdbOsFree(pageBuf);
M
Minglei Jin 已提交
913 914
      return -1;
    }
915 916
  }

917 918 919
  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 已提交
920
    tdbOsFree(pageBuf);
921 922
    return -1;
  }
923 924 925

  tdbOsFree(pageBuf);

926 927 928 929 930 931 932 933 934 935 936
  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;
  }
937 938 939

  return 0;
}
940

941
int tdbPagerRestoreJournals(SPager *pPager) {
942 943 944 945 946 947 948 949 950 951
  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) {
952
      if (tdbPagerRestore(pPager, name) < 0) {
M
Minglei Jin 已提交
953 954
        tdbCloseDir(&pDir);

955 956 957 958 959 960 961 962 963 964 965
        tdbError("failed to restore file due to %s. jFileName:%s", strerror(errno), name);
        return -1;
      }
    }
  }

  tdbCloseDir(&pDir);

  return 0;
}

966
int tdbPagerRollback(SPager *pPager) {
967 968 969 970
  tdbDirEntryPtr pDirEntry;
  tdbDirPtr      pDir = taosOpenDir(pPager->pEnv->dbName);
  if (pDir == NULL) {
    tdbError("failed to open %s since %s", pPager->pEnv->dbName, strerror(errno));
971 972 973
    return -1;
  }

974 975 976 977 978
  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 已提交
979 980
        tdbCloseDir(&pDir);

981 982 983 984 985 986 987 988 989
        tdbError("failed to remove file due to %s. jFileName:%s", strerror(errno), name);
        terrno = TAOS_SYSTEM_ERROR(errno);
        return -1;
      }
    }
  }

  tdbCloseDir(&pDir);

990 991
  return 0;
}