#4720

Committed 08 Sep 2025 08:43AM UTC coverage: 58.139% (-0.6%) from 58.762%

Build # #4720

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #32881 from taosdata/enh/add-new-windows-ci

fix(ci): update workflow reference to use new Windows CI YAML

Run Details

133181 of 292179 branches covered (45.58%)

Branch coverage included in aggregate %.

201691 of 283811 relevant lines covered (71.07%)

5442780.71 hits per line

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

77.37

/source/libs/tdb/src/db/tdbPCache.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"

// #include <sys/types.h>
// #include <unistd.h>

struct SPCache {
  int         szPage;
  int         nPages;
  SPage     **aPage;
  tdb_mutex_t mutex;
  int         nFree;
  SPage      *pFree;
  int         nPage;
  int         nHash;
  SPage     **pgHash;
  int         nRecyclable;
  SPage       lru;
};

static inline uint32_t tdbPCachePageHash(const SPgid *pPgid) {
  uint32_t *t = (uint32_t *)((pPgid)->fileid);
  return (uint32_t)(t[0] + t[1] + t[2] + t[3] + t[4] + t[5] + (pPgid)->pgno);
}

static int    tdbPCacheOpenImpl(SPCache *pCache);
static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn, bool force, bool* loaded);
static void   tdbPCachePinPage(SPCache *pCache, SPage *pPage);
static void   tdbPCacheRemovePageFromHash(SPCache *pCache, SPage *pPage);
static void   tdbPCacheAddPageToHash(SPCache *pCache, SPage *pPage);
static void   tdbPCacheUnpinPage(SPCache *pCache, SPage *pPage);
static void   tdbPCacheCloseImpl(SPCache *pCache);

static void tdbPCacheInitLock(SPCache *pCache) {
  if (tdbMutexInit(&(pCache->mutex), NULL) != 0) {
    tdbError("tdb/pcache: mutex init failed.");
  }
}

static void tdbPCacheDestroyLock(SPCache *pCache) {
  if (tdbMutexDestroy(&(pCache->mutex)) != 0) {
    tdbError("tdb/pcache: mutex destroy failed.");
  }
}

static void tdbPCacheLock(SPCache *pCache) {
  if (tdbMutexLock(&(pCache->mutex)) != 0) {
    tdbError("tdb/pcache: mutex lock failed.");
  }
}

static void tdbPCacheUnlock(SPCache *pCache) {
  if (tdbMutexUnlock(&(pCache->mutex)) != 0) {
    tdbError("tdb/pcache: mutex unlock failed.");
  }
}

int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) {
  int32_t  code = 0;
  int32_t  lino;
  SPCache *pCache;
  void    *pPtr;
  SPage   *pPgHdr;

  pCache = (SPCache *)tdbOsCalloc(1, sizeof(*pCache) + sizeof(SPage *) * cacheSize);
  if (pCache == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  pCache->szPage = pageSize;
  pCache->nPages = cacheSize;
  pCache->aPage = (SPage **)tdbOsCalloc(cacheSize, sizeof(SPage *));
  if (pCache->aPage == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  code = tdbPCacheOpenImpl(pCache);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    tdbError("%s failed at %s:%d since %s", __func__, __FILE__, __LINE__, tstrerror(code));
    tdbPCacheClose(pCache);
    *ppCache = NULL;
  } else {
    *ppCache = pCache;
  }
  return code;
}

void tdbPCacheClose(SPCache *pCache) {
  if (pCache) {
    tdbPCacheCloseImpl(pCache);
    tdbOsFree(pCache->aPage);
    tdbOsFree(pCache);
  }
  return;
}

// TODO:
// if (pPage->id >= pCache->nPages) {
//   free(pPage);
//   pCache->aPage[pPage->id] = NULL;
// } else {
//   add to free list
// }

static int tdbPCacheAlterImpl(SPCache *pCache, int32_t nPage) {
  if (pCache->nPages == nPage) {
    return 0;
  } else if (pCache->nPages < nPage) {
    SPage **aPage = tdbOsCalloc(nPage, sizeof(SPage *));
    if (aPage == NULL) {
      return terrno;
    }

    for (int32_t iPage = pCache->nPages; iPage < nPage; iPage++) {
      int32_t code = tdbPageCreate(pCache->szPage, &aPage[iPage], tdbDefaultMalloc, NULL);
      if (code) {
        tdbOsFree(aPage);
        return code;
      }

      // pPage->pgid = 0;
      aPage[iPage]->isAnchor = 0;
      aPage[iPage]->isLocal = 1;
      aPage[iPage]->nRef = 0;
      aPage[iPage]->pHashNext = NULL;
      aPage[iPage]->pLruNext = NULL;
      aPage[iPage]->pLruPrev = NULL;
      aPage[iPage]->pDirtyNext = NULL;

      // add to local list
      aPage[iPage]->id = iPage;
    }

    // add page to free list
    for (int32_t iPage = pCache->nPages; iPage < nPage; iPage++) {
      aPage[iPage]->pFreeNext = pCache->pFree;
      pCache->pFree = aPage[iPage];
      pCache->nFree++;
    }

    for (int32_t iPage = 0; iPage < pCache->nPages; iPage++) {
      aPage[iPage] = pCache->aPage[iPage];
    }

    tdbOsFree(pCache->aPage);
    pCache->aPage = aPage;
  } else {
    for (SPage **ppPage = &pCache->pFree; *ppPage;) {
      int32_t iPage = (*ppPage)->id;

      if (iPage >= nPage) {
        SPage *pPage = *ppPage;
        *ppPage = pPage->pFreeNext;
        pCache->aPage[pPage->id] = NULL;
        tdbPageDestroy(pPage, tdbDefaultFree, NULL);
        pCache->nFree--;
      } else {
        ppPage = &(*ppPage)->pFreeNext;
      }
    }
  }

  pCache->nPages = nPage;
  return 0;
}

int tdbPCacheAlter(SPCache *pCache, int32_t nPage) {
  int code;
  tdbPCacheLock(pCache);
  code = tdbPCacheAlterImpl(pCache, nPage);
  tdbPCacheUnlock(pCache);
  return code;
}

SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn, bool force, bool* loaded) {
  SPage *pPage;
  i32    nRef = 0;

  tdbPCacheLock(pCache);

  pPage = tdbPCacheFetchImpl(pCache, pPgid, pTxn, force, loaded);
  if (pPage) {
    nRef = tdbRefPage(pPage);
  }

  tdbPCacheUnlock(pCache);

  if (pPage) {
    tdbTrace("pcache/fetch page %p/%d/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id, nRef);
  } else {
    tdbTrace("pcache/fetch page %p", pPage);
  }

  return pPage;
}

void tdbPCacheMarkFree(SPCache *pCache, SPage *pPage) {
  tdbPCacheLock(pCache);
  tdbPCacheRemovePageFromHash(pCache, pPage);
  pPage->isFree = 1;
  tdbPCacheUnlock(pCache);
}

static void tdbPCacheFreePage(SPCache *pCache, SPage *pPage) {
  if (pPage->id < pCache->nPages) {
    pPage->pFreeNext = pCache->pFree;
    pCache->pFree = pPage;
    pPage->isFree = 0;
    ++pCache->nFree;
    tdbTrace("pcache/free page %p/%d, pgno:%d, ", pPage, pPage->id, TDB_PAGE_PGNO(pPage));
  } else {
    tdbTrace("pcache/free2 page: %p/%d, pgno:%d, ", pPage, pPage->id, TDB_PAGE_PGNO(pPage));

    tdbPCacheRemovePageFromHash(pCache, pPage);
    tdbPageDestroy(pPage, tdbDefaultFree, NULL);
  }
}

void tdbPCacheInvalidatePage(SPCache *pCache, SPager *pPager, SPgno pgno) {
  SPgid        pgid;
  const SPgid *pPgid = &pgid;
  SPage       *pPage = NULL;

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

  pPage = pCache->pgHash[tdbPCachePageHash(pPgid) % pCache->nHash];
  while (pPage) {
    if (pPage->pgid.pgno == pPgid->pgno && memcmp(pPage->pgid.fileid, pPgid->fileid, TDB_FILE_ID_LEN) == 0) break;
    pPage = pPage->pHashNext;
  }

  if (pPage) {
    bool moveToFreeList = false;
    if (pPage->pLruNext) {
      tdbPCachePinPage(pCache, pPage);
      moveToFreeList = true;
    }
    tdbPCacheRemovePageFromHash(pCache, pPage);
    if (moveToFreeList) {
      tdbPCacheFreePage(pCache, pPage);
    }
  }
}

void tdbPCacheRelease(SPCache *pCache, SPage *pPage, TXN *pTxn) {
  i32 nRef;

  if (!pTxn) {
    tdbError("tdb/pcache: null ptr pTxn, release failed.");
    return;
  }

  tdbPCacheLock(pCache);
  nRef = tdbUnrefPage(pPage);
  tdbTrace("pcache/release page %p/%d/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id, nRef);
  if (nRef == 0) {
    // test the nRef again to make sure
    // it is safe th handle the page
    // nRef = tdbGetPageRef(pPage);
    // if (nRef == 0) {
    if (pPage->isLocal) {
      if (!pPage->isFree) {
        tdbPCacheUnpinPage(pCache, pPage);
      } else {
        tdbPCacheFreePage(pCache, pPage);
      }
    } else {
      if (TDB_TXN_IS_WRITE(pTxn)) {
        // remove from hash
        tdbPCacheRemovePageFromHash(pCache, pPage);
      }

      tdbPageDestroy(pPage, pTxn->xFree, pTxn->xArg);
    }
    // }
  }
  tdbPCacheUnlock(pCache);
}

int tdbPCacheGetPageSize(SPCache *pCache) { return pCache->szPage; }

static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn, bool force, bool* loaded) {
  int    ret = 0;
  SPage *pPage = NULL;
  SPage *pPageH = NULL;

  if (!pTxn) {
    tdbError("tdb/pcache: null ptr pTxn, fetch impl failed.");
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  if (loaded) {
    *loaded = false;
  }

  // 1. Search the hash table
  pPage = pCache->pgHash[tdbPCachePageHash(pPgid) % pCache->nHash];
  while (pPage) {
    if (pPage->pgid.pgno == pPgid->pgno && memcmp(pPage->pgid.fileid, pPgid->fileid, TDB_FILE_ID_LEN) == 0) break;
    pPage = pPage->pHashNext;
  }

  if (pPage) {
    if (pPage->isLocal || TDB_TXN_IS_WRITE(pTxn)) {
      tdbPCachePinPage(pCache, pPage);
      if (loaded) {
        *loaded = true;
      }
      return pPage;
    }
  }

  // 1. pPage == NULL
  // 2. pPage && !pPage->isLocal == 0 && !TDB_TXN_IS_WRITE(pTxn)
  pPageH = pPage;
  pPage = NULL;

  // 2. Try to allocate a new page from the free list
  if (pCache->pFree) {
    pPage = pCache->pFree;
    pCache->pFree = pPage->pFreeNext;
    pCache->nFree--;
    pPage->pLruNext = NULL;
  }

  // 3. Try to Recycle a page
  if (!pPageH && !pPage && !pCache->lru.pLruPrev->isAnchor) {
    pPage = pCache->lru.pLruPrev;
    tdbPCacheRemovePageFromHash(pCache, pPage);
    tdbPCachePinPage(pCache, pPage);
  }

  // 4. Try a create new page
  if (!pPage && pTxn->xMalloc != NULL) {
    if (force) {
      ret = tdbPageCreate(pCache->szPage, &pPage, &tdbDefaultMalloc, pTxn->xArg);
    } else {
      ret = tdbPageCreate(pCache->szPage, &pPage, pTxn->xMalloc, pTxn->xArg);
    }
    if (ret < 0 || pPage == NULL) {
      // when allocating from bufpool failed, it's time to flush cache.
      // tdbError("tdb/pcache: ret: %" PRId32 " pPage: %p, page create failed.", ret, pPage);

      terrno = ret;
      return NULL;
    }

    // init the page fields
    pPage->isAnchor = 0;
    pPage->isLocal = 0;
    pPage->nRef = 0;
    pPage->id = -1;
  }

  // 5. Page here are just created from a free list
  // or by recycling or allocated streesly,
  // need to initialize it
  if (pPage) {
    if (pPageH) {
      // copy the page content
      memcpy(&(pPage->pgid), pPgid, sizeof(*pPgid));

      for (int nLoops = 0;;) {
        if (pPageH->pPager) break;
        if (++nLoops > 1000) {
          (void)sched_yield();
          nLoops = 0;
        }
      }

      pPage->pLruNext = NULL;
      pPage->pPager = pPageH->pPager;

      memcpy(pPage->pData, pPageH->pData, pPage->pageSize);
      // tdbDebug("pcache/pPageH: %p %ld %p %p %u", pPageH, pPageH->pPageHdr - pPageH->pData, pPageH->xCellSize, pPage,
      //         TDB_PAGE_PGNO(pPageH));
      tdbPageInit(pPage, pPageH->pPageHdr - pPageH->pData, pPageH->xCellSize);
      pPage->kLen = pPageH->kLen;
      pPage->vLen = pPageH->vLen;
      pPage->maxLocal = pPageH->maxLocal;
      pPage->minLocal = pPageH->minLocal;
    } else {
      memcpy(&(pPage->pgid), pPgid, sizeof(*pPgid));
      pPage->pLruNext = NULL;
      pPage->pPager = NULL;

      if (pPage->isLocal || TDB_TXN_IS_WRITE(pTxn)) {
        tdbPCacheAddPageToHash(pCache, pPage);
      }
    }
  }

  return pPage;
}

static void tdbPCachePinPage(SPCache *pCache, SPage *pPage) {
  if (pPage->pLruNext != NULL) {
    int32_t nRef = tdbGetPageRef(pPage);
    if (nRef != 0) {
      tdbError("tdb/pcache: pin page's ref not zero: %" PRId32, nRef);
      return;
    }

    pPage->pLruPrev->pLruNext = pPage->pLruNext;
    pPage->pLruNext->pLruPrev = pPage->pLruPrev;
    pPage->pLruNext = NULL;

    pCache->nRecyclable--;

    tdbTrace("pcache/pin page %p/%d, pgno:%d, ", pPage, pPage->id, TDB_PAGE_PGNO(pPage));
  }
}

static void tdbPCacheUnpinPage(SPCache *pCache, SPage *pPage) {
  i32 nRef = tdbGetPageRef(pPage);
  if (nRef != 0) {
    tdbError("tdb/pcache: unpin page's ref not zero: %" PRId32, nRef);
    return;
  }
  if (!pPage->isLocal) {
    tdbError("tdb/pcache: unpin page's not local: %" PRIu8, pPage->isLocal);
    return;
  }
  if (pPage->isDirty) {
    tdbError("tdb/pcache: unpin page's dirty: %" PRIu8, pPage->isDirty);
    return;
  }
  if (NULL != pPage->pLruNext) {
    tdbError("tdb/pcache: unpin page's pLruNext not null.");
    return;
  }

  tdbTrace("pCache:%p unpin page %p/%d, nPages:%d, pgno:%d, ", pCache, pPage, pPage->id, pCache->nPages,
           TDB_PAGE_PGNO(pPage));
  if (pPage->id < pCache->nPages) {
    pPage->pLruPrev = &(pCache->lru);
    pPage->pLruNext = pCache->lru.pLruNext;
    pCache->lru.pLruNext->pLruPrev = pPage;
    pCache->lru.pLruNext = pPage;

    pCache->nRecyclable++;

    // printf("unpin page %d pgno %d pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
    tdbTrace("pcache/unpin page %p/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id);
  } else {
    tdbTrace("pcache destroy page: %p/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id);

    tdbPCacheRemovePageFromHash(pCache, pPage);
    tdbPageDestroy(pPage, tdbDefaultFree, NULL);
  }
}

static void tdbPCacheRemovePageFromHash(SPCache *pCache, SPage *pPage) {
  uint32_t h = tdbPCachePageHash(&(pPage->pgid)) % pCache->nHash;

  SPage **ppPage = &(pCache->pgHash[h]);
  for (; (*ppPage) && *ppPage != pPage; ppPage = &((*ppPage)->pHashNext))
    ;

  if (*ppPage) {
    *ppPage = pPage->pHashNext;
    pCache->nPage--;
    // printf("rmv page %d to hash, pgno %d, pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
  }

  tdbTrace("pcache/remove page %p/%d from hash %" PRIu32 " pgno:%d, ", pPage, pPage->id, h, TDB_PAGE_PGNO(pPage));
}

static void tdbPCacheAddPageToHash(SPCache *pCache, SPage *pPage) {
  uint32_t h = tdbPCachePageHash(&(pPage->pgid)) % pCache->nHash;

  pPage->pHashNext = pCache->pgHash[h];
  pCache->pgHash[h] = pPage;

  pCache->nPage++;

  tdbTrace("pcache/add page %p/%d to hash %" PRIu32 " pgno:%d, ", pPage, pPage->id, h, TDB_PAGE_PGNO(pPage));
}

static int tdbPCacheOpenImpl(SPCache *pCache) {
  SPage *pPage;
  u8    *pPtr;
  int    tsize;
  int    ret;

  tdbPCacheInitLock(pCache);

  // Open the free list
  pCache->nFree = 0;
  pCache->pFree = NULL;
  for (int i = 0; i < pCache->nPages; i++) {
    ret = tdbPageCreate(pCache->szPage, &pPage, tdbDefaultMalloc, NULL);
    if (ret) return ret;

    // pPage->pgid = 0;
    pPage->isAnchor = 0;
    pPage->isLocal = 1;
    pPage->nRef = 0;
    pPage->pHashNext = NULL;
    pPage->pLruNext = NULL;
    pPage->pLruPrev = NULL;
    pPage->pDirtyNext = NULL;

    // add page to free list
    pPage->pFreeNext = pCache->pFree;
    pCache->pFree = pPage;
    pCache->nFree++;

    // add to local list
    pPage->id = i;
    pCache->aPage[i] = pPage;
  }

  // Open the hash table
  pCache->nPage = 0;
  pCache->nHash = pCache->nPages < 8 ? 8 : pCache->nPages;
  pCache->pgHash = (SPage **)tdbOsCalloc(pCache->nHash, sizeof(SPage *));
  if (pCache->pgHash == NULL) {
    return terrno;
  }

  // Open LRU list
  pCache->nRecyclable = 0;
  pCache->lru.isAnchor = 1;
  pCache->lru.pLruNext = &(pCache->lru);
  pCache->lru.pLruPrev = &(pCache->lru);

  return 0;
}

static void tdbPCacheCloseImpl(SPCache *pCache) {
  // free free page
  for (SPage *pPage = pCache->pFree; pPage;) {
    SPage *pPageT = pPage->pFreeNext;
    tdbPageDestroy(pPage, tdbDefaultFree, NULL);
    pPage = pPageT;
  }

  for (int32_t iBucket = 0; iBucket < pCache->nHash; iBucket++) {
    for (SPage *pPage = pCache->pgHash[iBucket]; pPage;) {
      SPage *pPageT = pPage->pHashNext;
      tdbPageDestroy(pPage, tdbDefaultFree, NULL);
      pPage = pPageT;
    }
  }

  tdbOsFree(pCache->pgHash);
  tdbPCacheDestroyLock(pCache);
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15	#include "tdbInt.h"
16
17	// #include <sys/types.h>
18	// #include <unistd.h>
19
20	struct SPCache {
21	int szPage;
22	int nPages;
23	SPage **aPage;
24	tdb_mutex_t mutex;
25	int nFree;
26	SPage *pFree;
27	int nPage;
28	int nHash;
29	SPage **pgHash;
30	int nRecyclable;
31	SPage lru;
32	};
33
34	static inline uint32_t tdbPCachePageHash(const SPgid *pPgid) {	34,747,883✔
35	uint32_t t = (uint32_t )((pPgid)->fileid);	34,747,883✔
36	return (uint32_t)(t[0] + t[1] + t[2] + t[3] + t[4] + t[5] + (pPgid)->pgno);	34,747,883✔
37	}
38
39	static int tdbPCacheOpenImpl(SPCache *pCache);
40	static SPage tdbPCacheFetchImpl(SPCache pCache, const SPgid pPgid, TXN pTxn, bool force, bool* loaded);
41	static void tdbPCachePinPage(SPCache pCache, SPage pPage);
42	static void tdbPCacheRemovePageFromHash(SPCache pCache, SPage pPage);
43	static void tdbPCacheAddPageToHash(SPCache pCache, SPage pPage);
44	static void tdbPCacheUnpinPage(SPCache pCache, SPage pPage);
45	static void tdbPCacheCloseImpl(SPCache *pCache);
46
47	static void tdbPCacheInitLock(SPCache *pCache) {	24,833✔
48	if (tdbMutexInit(&(pCache->mutex), NULL) != 0) {	24,833!
49	tdbError("tdb/pcache: mutex init failed.");	×
50	}
51	}	24,840✔
52
53	static void tdbPCacheDestroyLock(SPCache *pCache) {	24,847✔
54	if (tdbMutexDestroy(&(pCache->mutex)) != 0) {	24,847!
55	tdbError("tdb/pcache: mutex destroy failed.");	×
56	}
57	}	24,849✔
58
59	static void tdbPCacheLock(SPCache *pCache) {	58,947,101✔
60	if (tdbMutexLock(&(pCache->mutex)) != 0) {	58,947,101!
61	tdbError("tdb/pcache: mutex lock failed.");	×
62	}
63	}	59,069,685✔
64
65	static void tdbPCacheUnlock(SPCache *pCache) {	59,074,404✔
66	if (tdbMutexUnlock(&(pCache->mutex)) != 0) {	59,074,404!
67	tdbError("tdb/pcache: mutex unlock failed.");	×
68	}
69	}	59,059,923✔
70
71	int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) {	24,717✔
72	int32_t code = 0;	24,717✔
73	int32_t lino;
74	SPCache *pCache;
75	void *pPtr;
76	SPage *pPgHdr;
77
78	pCache = (SPCache )tdbOsCalloc(1, sizeof(pCache) + sizeof(SPage ) cacheSize);	24,717!
79	if (pCache == NULL) {	24,841!
80	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
81	}
82
83	pCache->szPage = pageSize;	24,841✔
84	pCache->nPages = cacheSize;	24,841✔
85	pCache->aPage = (SPage *)tdbOsCalloc(cacheSize, sizeof(SPage ));	24,841!
86	if (pCache->aPage == NULL) {	24,845!
87	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
88	}
89
90	code = tdbPCacheOpenImpl(pCache);	24,845✔
91	TSDB_CHECK_CODE(code, lino, _exit);	24,839!
92
93	_exit:	24,839✔
94	if (code) {	24,839!
95	tdbError("%s failed at %s:%d since %s", __func__, __FILE__, __LINE__, tstrerror(code));	×
96	tdbPCacheClose(pCache);	×
97	*ppCache = NULL;	×
98	} else {
99	*ppCache = pCache;	24,839✔
100	}
101	return code;	24,839✔
102	}
103
104	void tdbPCacheClose(SPCache *pCache) {	24,847✔
105	if (pCache) {	24,847!
106	tdbPCacheCloseImpl(pCache);	24,848✔
107	tdbOsFree(pCache->aPage);	24,848!
108	tdbOsFree(pCache);	24,848!
109	}
110	return;	24,849✔
111	}
112
113	// TODO:
114	// if (pPage->id >= pCache->nPages) {
115	// free(pPage);
116	// pCache->aPage[pPage->id] = NULL;
117	// } else {
118	// add to free list
119	// }
120
121	static int tdbPCacheAlterImpl(SPCache *pCache, int32_t nPage) {	6✔
122	if (pCache->nPages == nPage) {	6!
123	return 0;	×
124	} else if (pCache->nPages < nPage) {	6✔
125	SPage *aPage = tdbOsCalloc(nPage, sizeof(SPage ));	4!
126	if (aPage == NULL) {	4!
127	return terrno;	×
128	}
129
130	for (int32_t iPage = pCache->nPages; iPage < nPage; iPage++) {	900✔
131	int32_t code = tdbPageCreate(pCache->szPage, &aPage[iPage], tdbDefaultMalloc, NULL);	896✔
132	if (code) {	896!
133	tdbOsFree(aPage);	×
134	return code;	×
135	}
136
137	// pPage->pgid = 0;
138	aPage[iPage]->isAnchor = 0;	896✔
139	aPage[iPage]->isLocal = 1;	896✔
140	aPage[iPage]->nRef = 0;	896✔
141	aPage[iPage]->pHashNext = NULL;	896✔
142	aPage[iPage]->pLruNext = NULL;	896✔
143	aPage[iPage]->pLruPrev = NULL;	896✔
144	aPage[iPage]->pDirtyNext = NULL;	896✔
145
146	// add to local list
147	aPage[iPage]->id = iPage;	896✔
148	}
149
150	// add page to free list
151	for (int32_t iPage = pCache->nPages; iPage < nPage; iPage++) {	900✔
152	aPage[iPage]->pFreeNext = pCache->pFree;	896✔
153	pCache->pFree = aPage[iPage];	896✔
154	pCache->nFree++;	896✔
155	}
156
157	for (int32_t iPage = 0; iPage < pCache->nPages; iPage++) {	388✔
158	aPage[iPage] = pCache->aPage[iPage];	384✔
159	}
160
161	tdbOsFree(pCache->aPage);	4!
162	pCache->aPage = aPage;	4✔
163	} else {
164	for (SPage *ppPage = &pCache->pFree; ppPage;) {	486✔
165	int32_t iPage = (*ppPage)->id;	484✔
166
167	if (iPage >= nPage) {	484✔
168	SPage pPage = ppPage;	356✔
169	*ppPage = pPage->pFreeNext;	356✔
170	pCache->aPage[pPage->id] = NULL;	356✔
171	tdbPageDestroy(pPage, tdbDefaultFree, NULL);	356✔
172	pCache->nFree--;	356✔
173	} else {
174	ppPage = &(*ppPage)->pFreeNext;	128✔
175	}
176	}
177	}
178
179	pCache->nPages = nPage;	6✔
180	return 0;	6✔
181	}
182
183	int tdbPCacheAlter(SPCache *pCache, int32_t nPage) {	6✔
184	int code;
185	tdbPCacheLock(pCache);	6✔
186	code = tdbPCacheAlterImpl(pCache, nPage);	6✔
187	tdbPCacheUnlock(pCache);	6✔
188	return code;	6✔
189	}
190
191	SPage tdbPCacheFetch(SPCache pCache, const SPgid pPgid, TXN pTxn, bool force, bool* loaded) {	29,158,952✔
192	SPage *pPage;
193	i32 nRef = 0;	29,158,952✔
194
195	tdbPCacheLock(pCache);	29,158,952✔
196
197	pPage = tdbPCacheFetchImpl(pCache, pPgid, pTxn, force, loaded);	29,205,136✔
198	if (pPage) {	29,179,203✔
199	nRef = tdbRefPage(pPage);	28,010,407✔
200	}
201
202	tdbPCacheUnlock(pCache);	29,211,671✔
203
204	if (pPage) {	29,191,057✔
205	tdbTrace("pcache/fetch page %p/%d/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id, nRef);	28,022,223✔
206	} else {
207	tdbTrace("pcache/fetch page %p", pPage);	1,168,834!
208	}
209
210	return pPage;	29,191,285✔
211	}
212
213	void tdbPCacheMarkFree(SPCache pCache, SPage pPage) {	48,649✔
214	tdbPCacheLock(pCache);	48,649✔
215	tdbPCacheRemovePageFromHash(pCache, pPage);	48,649✔
216	pPage->isFree = 1;	48,649✔
217	tdbPCacheUnlock(pCache);	48,649✔
218	}	48,649✔
219
220	static void tdbPCacheFreePage(SPCache pCache, SPage pPage) {	33,440✔
221	if (pPage->id < pCache->nPages) {	33,440!
222	pPage->pFreeNext = pCache->pFree;	33,440✔
223	pCache->pFree = pPage;	33,440✔
224	pPage->isFree = 0;	33,440✔
225	++pCache->nFree;	33,440✔
226	tdbTrace("pcache/free page %p/%d, pgno:%d, ", pPage, pPage->id, TDB_PAGE_PGNO(pPage));	33,440✔
227	} else {
228	tdbTrace("pcache/free2 page: %p/%d, pgno:%d, ", pPage, pPage->id, TDB_PAGE_PGNO(pPage));	×
229
230	tdbPCacheRemovePageFromHash(pCache, pPage);	×
231	tdbPageDestroy(pPage, tdbDefaultFree, NULL);	×
232	}
233	}	33,440✔
234
235	void tdbPCacheInvalidatePage(SPCache pCache, SPager pPager, SPgno pgno) {	28,294✔
236	SPgid pgid;
237	const SPgid *pPgid = &pgid;	28,294✔
238	SPage *pPage = NULL;	28,294✔
239
240	memcpy(&pgid, pPager->fid, TDB_FILE_ID_LEN);	28,294✔
241	pgid.pgno = pgno;	28,294✔
242
243	pPage = pCache->pgHash[tdbPCachePageHash(pPgid) % pCache->nHash];	28,294✔
244	while (pPage) {	29,024✔
245	if (pPage->pgid.pgno == pPgid->pgno && memcmp(pPage->pgid.fileid, pPgid->fileid, TDB_FILE_ID_LEN) == 0) break;	29,012!
246	pPage = pPage->pHashNext;	730✔
247	}
248
249	if (pPage) {	28,294✔
250	bool moveToFreeList = false;	28,282✔
251	if (pPage->pLruNext) {	28,282!
252	tdbPCachePinPage(pCache, pPage);	×
253	moveToFreeList = true;	×
254	}
255	tdbPCacheRemovePageFromHash(pCache, pPage);	28,282✔
256	if (moveToFreeList) {	28,282!
257	tdbPCacheFreePage(pCache, pPage);	×
258	}
259	}
260	}	28,294✔
261
262	void tdbPCacheRelease(SPCache pCache, SPage pPage, TXN *pTxn) {	29,810,344✔
263	i32 nRef;
264
265	if (!pTxn) {	29,810,344!
266	tdbError("tdb/pcache: null ptr pTxn, release failed.");	×
267	return;	×
268	}
269
270	tdbPCacheLock(pCache);	29,810,344✔
271	nRef = tdbUnrefPage(pPage);	29,839,467✔
272	tdbTrace("pcache/release page %p/%d/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id, nRef);	29,849,124✔
273	if (nRef == 0) {	29,849,124✔
274	// test the nRef again to make sure
275	// it is safe th handle the page
276	// nRef = tdbGetPageRef(pPage);
277	// if (nRef == 0) {
278	if (pPage->isLocal) {	6,744,004✔
279	if (!pPage->isFree) {	3,630,549✔
280	tdbPCacheUnpinPage(pCache, pPage);	3,597,185✔
281	} else {
282	tdbPCacheFreePage(pCache, pPage);	33,364✔
283	}
284	} else {
285	if (TDB_TXN_IS_WRITE(pTxn)) {	3,113,455✔
286	// remove from hash
287	tdbPCacheRemovePageFromHash(pCache, pPage);	2,395,259✔
288	}
289
290	tdbPageDestroy(pPage, pTxn->xFree, pTxn->xArg);	3,113,450✔
291	}
292	// }
293	}
294	tdbPCacheUnlock(pCache);	29,848,225✔
295	}
296
297	int tdbPCacheGetPageSize(SPCache *pCache) { return pCache->szPage; }	24,823✔
298
299	static SPage tdbPCacheFetchImpl(SPCache pCache, const SPgid pPgid, TXN pTxn, bool force, bool* loaded) {	29,202,240✔
300	int ret = 0;	29,202,240✔
301	SPage *pPage = NULL;	29,202,240✔
302	SPage *pPageH = NULL;	29,202,240✔
303
304	if (!pTxn) {	29,202,240!
305	tdbError("tdb/pcache: null ptr pTxn, fetch impl failed.");	×
306	terrno = TSDB_CODE_INVALID_PARA;	×
307	return NULL;	×
308	}
309
310	if (loaded) {	29,202,240!
311	*loaded = false;	×
312	}
313
314	// 1. Search the hash table
315	pPage = pCache->pgHash[tdbPCachePageHash(pPgid) % pCache->nHash];	29,202,240✔
316	while (pPage) {	177,840,777✔
317	if (pPage->pgid.pgno == pPgid->pgno && memcmp(pPage->pgid.fileid, pPgid->fileid, TDB_FILE_ID_LEN) == 0) break;	173,827,071✔
318	pPage = pPage->pHashNext;	148,649,664✔
319	}
320
321	if (pPage) {	29,191,113✔
322	if (pPage->isLocal \|\| TDB_TXN_IS_WRITE(pTxn)) {	25,177,722✔
323	tdbPCachePinPage(pCache, pPage);	24,485,162✔
324	if (loaded) {	24,478,027!
325	*loaded = true;	×
326	}
327	return pPage;	24,478,027✔
328	}
329	}
330
331	// 1. pPage == NULL
332	// 2. pPage && !pPage->isLocal == 0 && !TDB_TXN_IS_WRITE(pTxn)
333	pPageH = pPage;	4,705,951✔
334	pPage = NULL;	4,705,951✔
335
336	// 2. Try to allocate a new page from the free list
337	if (pCache->pFree) {	4,705,951✔
338	pPage = pCache->pFree;	191,012✔
339	pCache->pFree = pPage->pFreeNext;	191,012✔
340	pCache->nFree--;	191,012✔
341	pPage->pLruNext = NULL;	191,012✔
342	}
343
344	// 3. Try to Recycle a page
345	if (!pPageH && !pPage && !pCache->lru.pLruPrev->isAnchor) {	4,705,951✔
346	pPage = pCache->lru.pLruPrev;	233,031✔
347	tdbPCacheRemovePageFromHash(pCache, pPage);	233,031✔
348	tdbPCachePinPage(pCache, pPage);	233,027✔
349	}
350
351	// 4. Try a create new page
352	if (!pPage && pTxn->xMalloc != NULL) {	4,706,604✔
353	if (force) {	4,280,207!
354	ret = tdbPageCreate(pCache->szPage, &pPage, &tdbDefaultMalloc, pTxn->xArg);	×
355	} else {
356	ret = tdbPageCreate(pCache->szPage, &pPage, pTxn->xMalloc, pTxn->xArg);	4,280,207✔
357	}
358	if (ret < 0 \|\| pPage == NULL) {	4,280,199!
359	// when allocating from bufpool failed, it's time to flush cache.
360	// tdbError("tdb/pcache: ret: %" PRId32 " pPage: %p, page create failed.", ret, pPage);
361
362	terrno = ret;	1,166,907✔
363	return NULL;	1,166,907✔
364	}
365
366	// init the page fields
367	pPage->isAnchor = 0;	3,113,292✔
368	pPage->isLocal = 0;	3,113,292✔
369	pPage->nRef = 0;	3,113,292✔
370	pPage->id = -1;	3,113,292✔
371	}
372
373	// 5. Page here are just created from a free list
374	// or by recycling or allocated streesly,
375	// need to initialize it
376	if (pPage) {	3,539,689✔
377	if (pPageH) {	3,537,288✔
378	// copy the page content
379	memcpy(&(pPage->pgid), pPgid, sizeof(*pPgid));	692,525✔
380
381	for (int nLoops = 0;;) {	692,525✔
382	if (pPageH->pPager) break;	692,527!
383	if (++nLoops > 1000) {	×
384	(void)sched_yield();	×
385	nLoops = 0;	2✔
386	}
387	}
388
389	pPage->pLruNext = NULL;	692,527✔
390	pPage->pPager = pPageH->pPager;	692,527✔
391
392	memcpy(pPage->pData, pPageH->pData, pPage->pageSize);	692,527✔
393	// tdbDebug("pcache/pPageH: %p %ld %p %p %u", pPageH, pPageH->pPageHdr - pPageH->pData, pPageH->xCellSize, pPage,
394	// TDB_PAGE_PGNO(pPageH));
395	tdbPageInit(pPage, pPageH->pPageHdr - pPageH->pData, pPageH->xCellSize);	692,527✔
396	pPage->kLen = pPageH->kLen;	692,477✔
397	pPage->vLen = pPageH->vLen;	692,477✔
398	pPage->maxLocal = pPageH->maxLocal;	692,477✔
399	pPage->minLocal = pPageH->minLocal;	692,477✔
400	} else {
401	memcpy(&(pPage->pgid), pPgid, sizeof(*pPgid));	2,844,763✔
402	pPage->pLruNext = NULL;	2,844,763✔
403	pPage->pPager = NULL;	2,844,763✔
404
405	if (pPage->isLocal \|\| TDB_TXN_IS_WRITE(pTxn)) {	2,844,763✔
406	tdbPCacheAddPageToHash(pCache, pPage);	2,819,249✔
407	}
408	}
409	}
410
411	return pPage;	3,539,572✔
412	}
413
414	static void tdbPCachePinPage(SPCache pCache, SPage pPage) {	24,716,589✔
415	if (pPage->pLruNext != NULL) {	24,716,589✔
416	int32_t nRef = tdbGetPageRef(pPage);	3,438,419✔
417	if (nRef != 0) {	3,438,027!
418	tdbError("tdb/pcache: pin page's ref not zero: %" PRId32, nRef);	×
419	return;	×
420	}
421
422	pPage->pLruPrev->pLruNext = pPage->pLruNext;	3,438,027✔
423	pPage->pLruNext->pLruPrev = pPage->pLruPrev;	3,438,027✔
424	pPage->pLruNext = NULL;	3,438,027✔
425
426	pCache->nRecyclable--;	3,438,027✔
427
428	tdbTrace("pcache/pin page %p/%d, pgno:%d, ", pPage, pPage->id, TDB_PAGE_PGNO(pPage));	3,438,027✔
429	}
430	}
431
432	static void tdbPCacheUnpinPage(SPCache pCache, SPage pPage) {	3,596,812✔
433	i32 nRef = tdbGetPageRef(pPage);	3,596,812✔
434	if (nRef != 0) {	3,596,620!
435	tdbError("tdb/pcache: unpin page's ref not zero: %" PRId32, nRef);	×
436	return;	×
437	}
438	if (!pPage->isLocal) {	3,596,623!
439	tdbError("tdb/pcache: unpin page's not local: %" PRIu8, pPage->isLocal);	×
440	return;	×
441	}
442	if (pPage->isDirty) {	3,596,623!
443	tdbError("tdb/pcache: unpin page's dirty: %" PRIu8, pPage->isDirty);	×
444	return;	×
445	}
446	if (NULL != pPage->pLruNext) {	3,596,623!
447	tdbError("tdb/pcache: unpin page's pLruNext not null.");	×
448	return;	×
449	}
450
451	tdbTrace("pCache:%p unpin page %p/%d, nPages:%d, pgno:%d, ", pCache, pPage, pPage->id, pCache->nPages,	3,596,623✔
452	TDB_PAGE_PGNO(pPage));
453	if (pPage->id < pCache->nPages) {	3,596,489✔
454	pPage->pLruPrev = &(pCache->lru);	3,596,487✔
455	pPage->pLruNext = pCache->lru.pLruNext;	3,596,487✔
456	pCache->lru.pLruNext->pLruPrev = pPage;	3,596,487✔
457	pCache->lru.pLruNext = pPage;	3,596,487✔
458
459	pCache->nRecyclable++;	3,596,487✔
460
461	// printf("unpin page %d pgno %d pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
462	tdbTrace("pcache/unpin page %p/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id);	3,596,487✔
463	} else {
464	tdbTrace("pcache destroy page: %p/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id);	2!
465
466	tdbPCacheRemovePageFromHash(pCache, pPage);	2✔
467	tdbPageDestroy(pPage, tdbDefaultFree, NULL);	2✔
468	}
469	}
470
471	static void tdbPCacheRemovePageFromHash(SPCache pCache, SPage pPage) {	2,705,178✔
472	uint32_t h = tdbPCachePageHash(&(pPage->pgid)) % pCache->nHash;	2,705,178✔
473
474	SPage **ppPage = &(pCache->pgHash[h]);	2,705,183✔
475	for (; (ppPage) && ppPage != pPage; ppPage = &((*ppPage)->pHashNext))	41,101,937✔
476	;
477
478	if (*ppPage) {	2,705,183✔
479	*ppPage = pPage->pHashNext;	2,661,722✔
480	pCache->nPage--;	2,661,722✔
481	// printf("rmv page %d to hash, pgno %d, pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
482	}
483
484	tdbTrace("pcache/remove page %p/%d from hash %" PRIu32 " pgno:%d, ", pPage, pPage->id, h, TDB_PAGE_PGNO(pPage));	2,705,183✔
485	}	2,705,183✔
486
487	static void tdbPCacheAddPageToHash(SPCache pCache, SPage pPage) {	2,819,173✔
488	uint32_t h = tdbPCachePageHash(&(pPage->pgid)) % pCache->nHash;	2,819,173✔
489
490	pPage->pHashNext = pCache->pgHash[h];	2,819,175✔
491	pCache->pgHash[h] = pPage;	2,819,175✔
492
493	pCache->nPage++;	2,819,175✔
494
495	tdbTrace("pcache/add page %p/%d to hash %" PRIu32 " pgno:%d, ", pPage, pPage->id, h, TDB_PAGE_PGNO(pPage));	2,819,175✔
496	}	2,819,175✔
497
498	static int tdbPCacheOpenImpl(SPCache *pCache) {	24,836✔
499	SPage *pPage;
500	u8 *pPtr;
501	int tsize;
502	int ret;
503
504	tdbPCacheInitLock(pCache);	24,836✔
505
506	// Open the free list
507	pCache->nFree = 0;	24,839✔
508	pCache->pFree = NULL;	24,839✔
509	for (int i = 0; i < pCache->nPages; i++) {	3,194,287✔
510	ret = tdbPageCreate(pCache->szPage, &pPage, tdbDefaultMalloc, NULL);	3,165,776✔
511	if (ret) return ret;	3,169,448!
512
513	// pPage->pgid = 0;
514	pPage->isAnchor = 0;	3,169,448✔
515	pPage->isLocal = 1;	3,169,448✔
516	pPage->nRef = 0;	3,169,448✔
517	pPage->pHashNext = NULL;	3,169,448✔
518	pPage->pLruNext = NULL;	3,169,448✔
519	pPage->pLruPrev = NULL;	3,169,448✔
520	pPage->pDirtyNext = NULL;	3,169,448✔
521
522	// add page to free list
523	pPage->pFreeNext = pCache->pFree;	3,169,448✔
524	pCache->pFree = pPage;	3,169,448✔
525	pCache->nFree++;	3,169,448✔
526
527	// add to local list
528	pPage->id = i;	3,169,448✔
529	pCache->aPage[i] = pPage;	3,169,448✔
530	}
531
532	// Open the hash table
533	pCache->nPage = 0;	28,511✔
534	pCache->nHash = pCache->nPages < 8 ? 8 : pCache->nPages;	28,511✔
535	pCache->pgHash = (SPage *)tdbOsCalloc(pCache->nHash, sizeof(SPage ));	28,511!
536	if (pCache->pgHash == NULL) {	24,843!
537	return terrno;	×
538	}
539
540	// Open LRU list
541	pCache->nRecyclable = 0;	24,843✔
542	pCache->lru.isAnchor = 1;	24,843✔
543	pCache->lru.pLruNext = &(pCache->lru);	24,843✔
544	pCache->lru.pLruPrev = &(pCache->lru);	24,843✔
545
546	return 0;	24,843✔
547	}
548
549	static void tdbPCacheCloseImpl(SPCache *pCache) {	24,846✔
550	// free free page
551	for (SPage *pPage = pCache->pFree; pPage;) {	3,058,725✔
552	SPage *pPageT = pPage->pFreeNext;	3,033,890✔
553	tdbPageDestroy(pPage, tdbDefaultFree, NULL);	3,033,890✔
554	pPage = pPageT;	3,033,879✔
555	}
556
557	for (int32_t iBucket = 0; iBucket < pCache->nHash; iBucket++) {	3,304,889✔
558	for (SPage *pPage = pCache->pgHash[iBucket]; pPage;) {	3,437,661✔
559	SPage *pPageT = pPage->pHashNext;	157,607✔
560	tdbPageDestroy(pPage, tdbDefaultFree, NULL);	157,607✔
561	pPage = pPageT;	157,604✔
562	}
563	}
564
565	tdbOsFree(pCache->pgHash);	24,832!
566	tdbPCacheDestroyLock(pCache);	24,849✔
567	}	24,848✔

taosdata / TDengine / #4720

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