tdbPager.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "tdbInt.h"

struct SPager {
  char *   dbFileName;
  char *   jFileName;
  int      pageSize;
  uint8_t  fid[TDB_FILE_ID_LEN];
  int      fd;
  int      jfd;
  SPCache *pCache;
  SPgno    dbFileSize;
  SPgno    dbOrigSize;
  int      nDirty;
  SPage *  pDirty;
  SPage *  pDirtyTail;
  u8       inTran;
};

typedef struct __attribute__((__packed__)) {
  u8    hdrString[16];
  u16   pageSize;
  SPgno freePage;
  u32   nFreePages;
  u8    reserved[102];
} SFileHdr;

TDB_STATIC_ASSERT(sizeof(SFileHdr) == 128, "Size of file header is not correct");

static int tdbPagerReadPage(SPager *pPager, SPage *pPage);

int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
  uint8_t *pPtr;
  SPager * pPager;
  int      fsize;
  int      zsize;
  int      ret;

  *ppPager = NULL;

  fsize = strlen(fileName);
  zsize = sizeof(*pPager)  /* SPager */
          + fsize + 1      /* dbFileName */
          + fsize + 8 + 1; /* jFileName */
  pPtr = (uint8_t *)calloc(1, zsize);
  if (pPtr == NULL) {
    return -1;
  }

  pPager = (SPager *)pPtr;
  pPtr += sizeof(*pPager);
  // pPager->dbFileName
  pPager->dbFileName = (char *)pPtr;
  memcpy(pPager->dbFileName, fileName, fsize);
  pPager->dbFileName[fsize] = '\0';
  pPtr += fsize + 1;
  // 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;

  pPager->fd = open(pPager->dbFileName, O_RDWR | O_CREAT, 0755);
  if (pPager->fd < 0) {
    return -1;
  }

  ret = tdbGnrtFileID(pPager->dbFileName, pPager->fid, false);
  if (ret < 0) {
    return -1;
  }

  pPager->jfd = -1;

  *ppPager = pPager;
  return 0;
}

int tdbPagerClose(SPager *pPager) {
  // TODO
  return 0;
}

int tdbPagerOpenDB(SPager *pPager, SPgno *ppgno, bool toCreate) {
  SPgno  pgno;
  SPage *pPage;
  int    ret;

  {
    // TODO: try to search the main DB to get the page number
    pgno = 0;
  }

  if (pgno == 0 && toCreate) {
    ret = tdbPagerAllocPage(pPager, &pPage, &pgno);
    if (ret < 0) {
      return -1;
    }

    // tdbpPagerZeroPage(pPage);
    ret = tdbPagerWrite(pPager, pPage);
    if (ret < 0) {
      return -1;
    }
  }

  *ppgno = pgno;
  return 0;
}

SPage *tdbPagerGet(SPager *pPager, SPgno pgno) {
  SPgid  pgid;
  SPage *pPage;

  memcpy(pgid.fileid, pPager->fid, TDB_FILE_ID_LEN);
  pgid.pgno = pgno;

  pPage = tdbPCacheFetch(pPager->pCache, &pgid, 1);
  if (pPage == NULL) {
    // TODO: handle error
    return NULL;
  }
  tdbPCacheFetchFinish(pPager->pCache, pPage);

  return pPage;
}

int tdbPagerWrite(SPager *pPager, SPage *pPage) {
  int ret;

  if (pPager->inTran == 0) {
    ret = tdbPagerBegin(pPager);
    if (ret < 0) {
      return -1;
    }
  }

  if (pPage->isDirty == 0) {
    pPage->isDirty = 1;
    // TODO: add the page to the dirty list

    // TODO: write the page to the journal
    if (1 /*actually load from the file*/) {
    }
  }
  return 0;
}

int tdbPagerAllocPage(SPager *pPager, SPage **ppPage, SPgno *ppgno) {
  SPage *pPage;
  SPgno  pgno;

  if (1 /*TODO: no free page*/) {
    pgno = ++pPager->dbFileSize;
    pPage = tdbPagerGet(pPager, pgno);
    ASSERT(pPage != NULL);
  } else {
    /* TODO: allocate from the free list */
    ASSERT(0);
  }

  *ppPage = pPage;
  *ppgno = pgno;
  return 0;
}

int tdbPagerBegin(SPager *pPager) {
  if (pPager->inTran) {
    return 0;
  }

  // Open the journal
  pPager->jfd = open(pPager->jFileName, O_RDWR | O_CREAT, 0755);
  if (pPager->jfd < 0) {
    return -1;
  }

  // TODO: write the size of the file

  pPager->inTran = 1;

  return 0;
}

int tdbPagerCommit(SPager *pPager) {
  // TODO
  return 0;
}

static int tdbPagerReadPage(SPager *pPager, SPage *pPage) {
  i64 offset;
  int ret;

  ASSERT(memcmp(pPager->fid, pPage->pgid.fileid, TDB_FILE_ID_LEN) == 0);

  offset = (pPage->pgid.pgno - 1) * (i64)(pPager->pageSize);
  ret = tdbPRead(pPager->fd, pPage->pData, pPager->pageSize, offset);
  if (ret < 0) {
    // TODO: handle error
    return -1;
  }
  return 0;
}