tdbInt.h

/*
 * 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/>.
 */

#ifndef _TD_TDB_INTERNAL_H_
#define _TD_TDB_INTERNAL_H_

#include "tdb.h"

#include "tlog.h"

#ifdef __cplusplus
extern "C" {
#endif

// clang-format off
extern int32_t tdbDebugFlag;

#define tdbFatal(...) do { if (tdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}     while(0)
#define tdbError(...) do { if (tdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}     while(0)
#define tdbWarn(...)  do { if (tdbDebugFlag & DEBUG_WARN)  { taosPrintLog("TDB WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}       while(0)
#define tdbInfo(...)  do { if (tdbDebugFlag & DEBUG_INFO)  { taosPrintLog("TDB ", DEBUG_INFO, 255, __VA_ARGS__); }}            while(0)
#define tdbDebug(...) do { if (tdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", DEBUG_DEBUG, tdbDebugFlag, __VA_ARGS__); }} while(0)
#define tdbTrace(...) do { if (tdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", DEBUG_TRACE, tdbDebugFlag, __VA_ARGS__); }} while(0)
// clang-format on

typedef int8_t   i8;
typedef int16_t  i16;
typedef int32_t  i32;
typedef int64_t  i64;
typedef uint8_t  u8;
typedef uint16_t u16;
typedef uint32_t u32;
typedef uint64_t u64;

// SPgno
typedef u32 SPgno;

#include "tdbOs.h"
#include "tdbUtil.h"

// p must be u8 *
#define TDB_GET_U24(p) ((p)[0] * 65536 + *(u16 *)((p) + 1))
#define TDB_PUT_U24(p, v)       \
  do {                          \
    int tv = (v);               \
    (p)[2] = tv & 0xff;         \
    (p)[1] = (tv >> 8) & 0xff;  \
    (p)[0] = (tv >> 16) & 0xff; \
  } while (0)

// fileid
#define TDB_FILE_ID_LEN 24

// SPgid
typedef struct {
  uint8_t fileid[TDB_FILE_ID_LEN];
  SPgno   pgno;
} SPgid;

// pgsz_t
#define TDB_MIN_PGSIZE       512       // 512B
#define TDB_MAX_PGSIZE       16777216  // 16M
#define TDB_DEFAULT_PGSIZE   4096
#define TDB_IS_PGSIZE_VLD(s) (((s) >= TDB_MIN_PGSIZE) && ((s) <= TDB_MAX_PGSIZE))

// dbname
#define TDB_MAX_DBNAME_LEN 24

#define TDB_VARIANT_LEN ((int)-1)

#define TDB_JOURNAL_NAME "tdb.journal"

#define TDB_FILENAME_LEN 128

#define BTREE_MAX_DEPTH 20

#define TDB_FLAG_IS(flags, flag)     ((flags) == (flag))
#define TDB_FLAG_HAS(flags, flag)    (((flags) & (flag)) != 0)
#define TDB_FLAG_NO(flags, flag)     ((flags) & (flag) == 0)
#define TDB_FLAG_ADD(flags, flag)    ((flags) | (flag))
#define TDB_FLAG_REMOVE(flags, flag) ((flags) & (~(flag)))

typedef struct SPager  SPager;
typedef struct SPCache SPCache;
typedef struct SPage   SPage;

// transaction

#define TDB_TXN_IS_WRITE(PTXN)            ((PTXN)->flags & TDB_TXN_WRITE)
#define TDB_TXN_IS_READ(PTXN)             (!TDB_TXN_IS_WRITE(PTXN))
#define TDB_TXN_IS_READ_UNCOMMITTED(PTXN) ((PTXN)->flags & TDB_TXN_READ_UNCOMMITTED)

// tdbEnv.c ====================================
void    tdbEnvAddPager(TDB *pEnv, SPager *pPager);
void    tdbEnvRemovePager(TDB *pEnv, SPager *pPager);
SPager *tdbEnvGetPager(TDB *pEnv, const char *fname);

// tdbBtree.c ====================================
typedef struct SBTree SBTree;
typedef struct SBTC   SBTC;
typedef struct SBtInfo {
  SPgno root;
  int   nLevel;
  int   nData;
} SBtInfo;

typedef struct {
  int       kLen;
  const u8 *pKey;
  int       vLen;
  const u8 *pVal;
  SPgno     pgno;
  u8       *pBuf;
} SCellDecoder;

struct SBTC {
  SBTree      *pBt;
  i8           iPage;
  SPage       *pPage;
  int          idx;
  int          idxStack[BTREE_MAX_DEPTH + 1];
  SPage       *pgStack[BTREE_MAX_DEPTH + 1];
  SCellDecoder coder;
  TXN         *pTxn;
  TXN          txn;
};

// SBTree
int tdbBtreeOpen(int keyLen, int valLen, SPager *pFile, tdb_cmpr_fn_t kcmpr, SBTree **ppBt);
int tdbBtreeClose(SBTree *pBt);
int tdbBtreeInsert(SBTree *pBt, const void *pKey, int kLen, const void *pVal, int vLen, TXN *pTxn);
int tdbBtreeDelete(SBTree *pBt, const void *pKey, int kLen, TXN *pTxn);
int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, int nData, TXN *pTxn);
int tdbBtreeGet(SBTree *pBt, const void *pKey, int kLen, void **ppVal, int *vLen);
int tdbBtreePGet(SBTree *pBt, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen);

// SBTC
int tdbBtcOpen(SBTC *pBtc, SBTree *pBt, TXN *pTxn);
int tdbBtcClose(SBTC *pBtc);
int tdbBtcIsValid(SBTC *pBtc);
int tdbBtcMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst);
int tdbBtcMoveToFirst(SBTC *pBtc);
int tdbBtcMoveToLast(SBTC *pBtc);
int tdbBtcMoveToNext(SBTC *pBtc);
int tdbBtcMoveToPrev(SBTC *pBtc);
int tdbBtreeNext(SBTC *pBtc, void **ppKey, int *kLen, void **ppVal, int *vLen);
int tdbBtcGet(SBTC *pBtc, const void **ppKey, int *kLen, const void **ppVal, int *vLen);
int tdbBtcDelete(SBTC *pBtc);
int tdbBtcUpsert(SBTC *pBtc, const void *pKey, int kLen, const void *pData, int nData, int insert);

// tdbPager.c ====================================

int  tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager);
int  tdbPagerClose(SPager *pPager);
int  tdbPagerOpenDB(SPager *pPager, SPgno *ppgno, bool toCreate);
int  tdbPagerWrite(SPager *pPager, SPage *pPage);
int  tdbPagerBegin(SPager *pPager, TXN *pTxn);
int  tdbPagerCommit(SPager *pPager, TXN *pTxn);
int  tdbPagerFetchPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage)(SPage *, void *, int), void *arg,
                       TXN *pTxn);
void tdbPagerReturnPage(SPager *pPager, SPage *pPage, TXN *pTxn);
int  tdbPagerAllocPage(SPager *pPager, SPgno *ppgno);

// tdbPCache.c ====================================
#define TDB_PCACHE_PAGE    \
  u8           isAnchor;   \
  u8           isLocal;    \
  u8           isDirty;    \
  volatile i32 nRef;       \
  i32          id;         \
  SPage       *pFreeNext;  \
  SPage       *pHashNext;  \
  SPage       *pLruNext;   \
  SPage       *pLruPrev;   \
  SPage       *pDirtyNext; \
  SPager      *pPager;     \
  SPgid        pgid;

// For page ref

int    tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache);
int    tdbPCacheClose(SPCache *pCache);
SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn);
void   tdbPCacheRelease(SPCache *pCache, SPage *pPage, TXN *pTxn);
int    tdbPCacheGetPageSize(SPCache *pCache);

// tdbPage.c ====================================
typedef u8 SCell;

// PAGE APIS implemented
typedef struct {
  int szOffset;
  int szPageHdr;
  int szFreeCell;
  // cell number
  int (*getCellNum)(SPage *);
  void (*setCellNum)(SPage *, int);
  // cell content offset
  int (*getCellBody)(SPage *);
  void (*setCellBody)(SPage *, int);
  // first free cell offset (0 means no free cells)
  int (*getCellFree)(SPage *);
  void (*setCellFree)(SPage *, int);
  // total free bytes
  int (*getFreeBytes)(SPage *);
  void (*setFreeBytes)(SPage *, int);
  // cell offset at idx
  int (*getCellOffset)(SPage *, int);
  void (*setCellOffset)(SPage *, int, int);
  // free cell info
  void (*getFreeCellInfo)(SCell *pCell, int *szCell, int *nxOffset);
  void (*setFreeCellInfo)(SCell *pCell, int szCell, int nxOffset);
} SPageMethods;

#pragma pack(push, 1)

// Page footer
typedef struct {
  u8 cksm[4];
} SPageFtr;
#pragma pack(pop)

struct SPage {
  tdb_spinlock_t lock;
  int            pageSize;
  u8            *pData;
  SPageMethods  *pPageMethods;
  // Fields below used by pager and am
  u8       *pPageHdr;
  u8       *pCellIdx;
  u8       *pFreeStart;
  u8       *pFreeEnd;
  SPageFtr *pPageFtr;
  int       nOverflow;
  SCell    *apOvfl[4];
  int       aiOvfl[4];
  int       kLen;  // key length of the page, -1 for unknown
  int       vLen;  // value length of the page, -1 for unknown
  int       maxLocal;
  int       minLocal;
  int (*xCellSize)(const SPage *, SCell *);
  // Fields used by SPCache
  TDB_PCACHE_PAGE
};

static inline i32 tdbRefPage(SPage *pPage) {
  i32 nRef = atomic_add_fetch_32(&((pPage)->nRef), 1);
  tdbTrace("ref page %d, nRef %d", pPage->id, nRef);
  return nRef;
}

static inline i32 tdbUnrefPage(SPage *pPage) {
  i32 nRef = atomic_sub_fetch_32(&((pPage)->nRef), 1);
  tdbTrace("unref page %d, nRef %d", pPage->id, nRef);
  return nRef;
}

#define tdbGetPageRef(pPage) atomic_load_32(&((pPage)->nRef))

// For page lock
#define P_LOCK_SUCC 0
#define P_LOCK_BUSY 1
#define P_LOCK_FAIL -1

static inline int tdbTryLockPage(tdb_spinlock_t *pLock) {
  int ret;
  if (tdbSpinlockTrylock(pLock) == 0) {
    ret = P_LOCK_SUCC;
  } else if (errno == EBUSY) {
    ret = P_LOCK_BUSY;
  } else {
    ret = P_LOCK_FAIL;
  }
  return ret;
}

#define TDB_INIT_PAGE_LOCK(pPage)    tdbSpinlockInit(&((pPage)->lock), 0)
#define TDB_DESTROY_PAGE_LOCK(pPage) tdbSpinlockDestroy(&((pPage)->lock))
#define TDB_LOCK_PAGE(pPage)         tdbSpinlockLock(&((pPage)->lock))
#define TDB_UNLOCK_PAGE(pPage)       tdbSpinlockUnlock(&((pPage)->lock))
#define TDB_TRY_LOCK_PAGE(pPage)     tdbTryLockPage(&((pPage)->lock))

// APIs
#define TDB_PAGE_TOTAL_CELLS(pPage)        ((pPage)->nOverflow + (pPage)->pPageMethods->getCellNum(pPage))
#define TDB_PAGE_USABLE_SIZE(pPage)        ((u8 *)(pPage)->pPageFtr - (pPage)->pCellIdx)
#define TDB_PAGE_FREE_SIZE(pPage)          (*(pPage)->pPageMethods->getFreeBytes)(pPage)
#define TDB_PAGE_PGNO(pPage)               ((pPage)->pgid.pgno)
#define TDB_BYTES_CELL_TAKEN(pPage, pCell) ((*(pPage)->xCellSize)(pPage, pCell) + (pPage)->pPageMethods->szOffset)
#define TDB_PAGE_OFFSET_SIZE(pPage)        ((pPage)->pPageMethods->szOffset)

int  tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t), void *arg);
int  tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg);
void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *));
void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *));
int  tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl);
int  tdbPageDropCell(SPage *pPage, int idx);
int  tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell);
void tdbPageCopy(SPage *pFromPage, SPage *pToPage);
int  tdbPageCapacity(int pageSize, int amHdrSize);

static inline SCell *tdbPageGetCell(SPage *pPage, int idx) {
  SCell *pCell;
  int    iOvfl;
  int    lidx;

  ASSERT(idx >= 0 && idx < TDB_PAGE_TOTAL_CELLS(pPage));

  iOvfl = 0;
  for (; iOvfl < pPage->nOverflow; iOvfl++) {
    if (pPage->aiOvfl[iOvfl] == idx) {
      pCell = pPage->apOvfl[iOvfl];
      return pCell;
    } else if (pPage->aiOvfl[iOvfl] > idx) {
      break;
    }
  }

  lidx = idx - iOvfl;
  ASSERT(lidx >= 0 && lidx < pPage->pPageMethods->getCellNum(pPage));
  pCell = pPage->pData + pPage->pPageMethods->getCellOffset(pPage, lidx);

  return pCell;
}

struct STDB {
  char    *dbName;
  char    *jnName;
  int      jfd;
  SPCache *pCache;
  SPager  *pgrList;
  int      nPager;
  int      nPgrHash;
  SPager **pgrHash;
};

struct SPager {
  char    *dbFileName;
  char    *jFileName;
  int      pageSize;
  uint8_t  fid[TDB_FILE_ID_LEN];
  tdb_fd_t fd;
  tdb_fd_t jfd;
  SPCache *pCache;
  SPgno    dbFileSize;
  SPgno    dbOrigSize;
  SPage   *pDirty;
  u8       inTran;
  SPager  *pNext;      // used by TDB
  SPager  *pHashNext;  // used by TDB
};

#ifdef __cplusplus
}
#endif

#endif /*_TD_TDB_INTERNAL_H_*/