/*

 * Copyright (c) 2020 Red Hat, Inc.

 *

 * This program is free software; you can redistribute it and/or

 * modify it under the terms of the GNU General Public License

 * as published by the Free Software Foundation; either version 2

 * of the License, or (at your option) any later version.

 * 

 * 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.  See the

 * GNU General Public License for more details.

 * 

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

 * 02110-1301, USA. 

 *

 * $Id: //eng/uds-releases/jasper/src/uds/pageCache.c#6 $

 */

#include "pageCache.h"

#include "atomicDefs.h"

#include "cacheCounters.h"

#include "chapterIndex.h"

#include "compiler.h"

#include "errors.h"

#include "geometry.h"

#include "hashUtils.h"

#include "indexConfig.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "permassert.h"

#include "recordPage.h"

#include "stringUtils.h"

#include "threads.h"

#include "zone.h"

/**********************************************************************/

int assertPageInCache(PageCache *cache, CachedPage *page)

{

  int result = ASSERT((page->cp_physicalPage < cache->numIndexEntries),

                      "physicalPage %u is valid (< %u)",

                      page->cp_physicalPage, cache->numIndexEntries);

  if (result != UDS_SUCCESS) {

    return result;

  }

  uint16_t pageIndex = cache->index[page->cp_physicalPage];

  return ASSERT((pageIndex < cache->numCacheEntries)

                && (&cache->cache[pageIndex] == page),

                "page is at expected location in cache");

}

/**

 * Clear a cache page.  Note: this does not clear readPending - a read could

 * still be pending and the read thread needs to be able to proceed and restart

 * the requests regardless. This page will still be marked invalid, but it

 * won't get reused (see getLeastRecentPage()) until the readPending flag

 * is cleared. This is a valid case, e.g. the chapter gets forgotten and

 * replaced with a new one in LRU.  Restarting the requests will lead them to

 * not find the records in the MI.

 *

 * @param cache   the cache

 * @param page    the cached page to clear

 *

 **/

static void clearPage(PageCache *cache, CachedPage *page)

{

  page->cp_physicalPage = cache->numIndexEntries;

  WRITE_ONCE(page->cp_lastUsed, 0);

}

/**

 * Get a page from the cache, but with no stats

 *

 * @param cache        the cache

 * @param physicalPage the physical page to get

 * @param queueIndex   the index of the page in the read queue if

 *                     queued, -1 otherwise

 * @param pagePtr      a pointer to hold the page

 *

 * @return UDS_SUCCESS or an error code

 **/

__attribute__((warn_unused_result))

static int getPageNoStats(PageCache     *cache,

                          unsigned int   physicalPage,

                          int           *queueIndex,

                          CachedPage   **pagePtr)

{

  /*

   * ASSERTION: We are either a zone thread holding a searchPendingCounter,

   *            or we are any thread holding the readThreadsMutex.

   *

   * Holding only a searchPendingCounter is the most frequent case.

   */

  int result = ASSERT((physicalPage < cache->numIndexEntries),

                      "physical page %u is invalid", physicalPage);

  if (result != UDS_SUCCESS) {

    return result;

  }

  /*

   * It would be unlikely that the compiler turns the usage of indexValue into

   * two reads of cache->index, but it would be possible and very bad if those

   * reads did not return the same bits.

   */

  uint16_t indexValue = READ_ONCE(cache->index[physicalPage]);

  bool     queued     = (indexValue & VOLUME_CACHE_QUEUED_FLAG) != 0;

  uint16_t index      = indexValue & ~VOLUME_CACHE_QUEUED_FLAG;

  if (!queued && (index < cache->numCacheEntries)) {

    *pagePtr = &cache->cache[index];

    /*

     * We have acquired access to the cached page, but unless we hold the

     * readThreadsMutex, we need a read memory barrier now.  The corresponding

     * write memory barrier is in putPageInCache.

     */

    smp_rmb();

  } else {

    *pagePtr = NULL;

  }

  if (queueIndex != NULL) {

    *queueIndex = queued ? index : -1;

  }

  return UDS_SUCCESS;

}

/**

 * Wait for all pending searches on a page in the cache to complete

 *

 * @param cache         the page cache

 * @param physicalPage  the page to check searches on

 **/

static void waitForPendingSearches(PageCache *cache, unsigned int physicalPage)

{

  /*

   * We hold the readThreadsMutex.  We are waiting for threads that do not hold

   * the readThreadsMutex.  Those threads have "locked" their targeted page by

   * setting the searchPendingCounter.  The corresponding write memory barrier

   * is in beginPendingSearch.

   */

  smp_mb();

  InvalidateCounter initialCounters[MAX_ZONES];

  unsigned int i;

  for (i = 0; i < cache->zoneCount; i++) {

    initialCounters[i] = getInvalidateCounter(cache, i);

  }

  for (i = 0; i < cache->zoneCount; i++) {

    if (searchPending(initialCounters[i])

        && (pageBeingSearched(initialCounters[i]) == physicalPage)) {

      // There is an active search using the physical page.

      // We need to wait for the search to finish.

      while (initialCounters[i] == getInvalidateCounter(cache, i)) {

        yieldScheduler();

      }

    }

  }

}

/**

 * Invalidate a cache page

 *

 * @param cache   the cache

 * @param page    the cached page

 * @param reason  the reason for invalidation, for stats

 *

 * @return UDS_SUCCESS or an error code

 **/

__attribute__((warn_unused_result))

static int invalidatePageInCache(PageCache          *cache,

                                 CachedPage         *page,

                                 InvalidationReason  reason)

{

  // We hold the readThreadsMutex.

  if (page == NULL) {

    return UDS_SUCCESS;

  }

  if (page->cp_physicalPage != cache->numIndexEntries) {

    switch (reason) {

    case INVALIDATION_EVICT:

      cache->counters.evictions++;

      break;

    case INVALIDATION_EXPIRE:

      cache->counters.expirations++;

      break;

    default:

      break;

    }

    if (reason != INVALIDATION_ERROR) {

      int result = assertPageInCache(cache, page);

      if (result != UDS_SUCCESS) {

        return result;

      }

    }

    WRITE_ONCE(cache->index[page->cp_physicalPage], cache->numCacheEntries);

    waitForPendingSearches(cache, page->cp_physicalPage);

  }

  clearPage(cache, page);

  return UDS_SUCCESS;

}

/**********************************************************************/

int findInvalidateAndMakeLeastRecent(PageCache          *cache,

                                     unsigned int        physicalPage,

                                     QueuedRead         *readQueue,

                                     InvalidationReason  reason,

                                     bool                mustFind)

{

  // We hold the readThreadsMutex.

  if (cache == NULL) {

    return UDS_SUCCESS;

  }

  CachedPage *page;

  int queuedIndex = -1;

  int result

    = getPageNoStats(cache, physicalPage,

                     ((readQueue != NULL) ? &queuedIndex : NULL), &page);

  if (result != UDS_SUCCESS) {

    return result;

  }

  if (page == NULL) {

    result = ASSERT(!mustFind, "found page");

    if (result != UDS_SUCCESS) {

      return result;

    }

    if (queuedIndex > -1) {

      logDebug("setting pending read to invalid");

      readQueue[queuedIndex].invalid = true;

    }

    return UDS_SUCCESS;

  }

  // Invalidate the page and unmap it from the cache.

  result = invalidatePageInCache(cache, page, reason);

  if (result != UDS_SUCCESS) {

    return result;

  }

  // Move the cached page to the least recently used end of the list

  // so it will be replaced before any page with valid data.

  WRITE_ONCE(page->cp_lastUsed, 0);

  return UDS_SUCCESS;

}

/**********************************************************************/

__attribute__((warn_unused_result))

static int initializePageCache(PageCache      *cache,

                               const Geometry *geometry,

                               unsigned int    chaptersInCache,

                               unsigned int    readQueueMaxSize,

                               unsigned int    zoneCount)

{

  cache->geometry  = geometry;

  cache->numIndexEntries = geometry->pagesPerVolume + 1;

  cache->numCacheEntries = chaptersInCache * geometry->recordPagesPerChapter;

  cache->readQueueMaxSize = readQueueMaxSize;

  cache->zoneCount = zoneCount;

  atomic64_set(&cache->clock, 1);

  int result = ALLOCATE(readQueueMaxSize, QueuedRead,

                        "volume read queue", &cache->readQueue);

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ALLOCATE(cache->zoneCount, SearchPendingCounter,

                    "Volume Cache Zones", &cache->searchPendingCounters);

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT((cache->numCacheEntries <= VOLUME_CACHE_MAX_ENTRIES),

                  "requested cache size, %u, within limit %u",

                  cache->numCacheEntries, VOLUME_CACHE_MAX_ENTRIES);

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ALLOCATE(cache->numIndexEntries, uint16_t, "page cache index",

                    &cache->index);

  if (result != UDS_SUCCESS) {

    return result;

  }

  // Initialize index values to invalid values.

  unsigned int i;

  for (i = 0; i < cache->numIndexEntries; i++) {

    cache->index[i] = cache->numCacheEntries;

  }

  result = ALLOCATE(cache->numCacheEntries, CachedPage,

                    "page cache cache", &cache->cache);

  if (result != UDS_SUCCESS) {

    return result;

  }

  for (i = 0; i < cache->numCacheEntries; i++) {

    CachedPage *page = &cache->cache[i];

    result = initializeVolumePage(geometry, &page->cp_pageData);

    if (result != UDS_SUCCESS) {

      return result;

    }

    clearPage(cache, page);

  }

  return UDS_SUCCESS;

}

/*********************************************************************/

int makePageCache(const Geometry  *geometry,

                  unsigned int     chaptersInCache,

                  unsigned int     readQueueMaxSize,

                  unsigned int     zoneCount,

                  PageCache      **cachePtr)

{

  if (chaptersInCache < 1) {

    return logWarningWithStringError(UDS_BAD_STATE,

                                     "cache size must be"

                                     " at least one chapter");

  }

  if (readQueueMaxSize <= 0) {

    return logWarningWithStringError(UDS_INVALID_ARGUMENT,

                                     "read queue max size must be"

                                     " greater than 0");

  }

  if (zoneCount < 1) {

    return logWarningWithStringError(UDS_INVALID_ARGUMENT,

                                     "cache must have at least one zone");

  }

  PageCache *cache;

  int result = ALLOCATE(1, PageCache, "volume cache", &cache);

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = initializePageCache(cache, geometry, chaptersInCache,

                               readQueueMaxSize, zoneCount);

  if (result != UDS_SUCCESS) {

    freePageCache(cache);

    return result;

  }

  *cachePtr = cache;

  return UDS_SUCCESS;

}

/**********************************************************************/

void freePageCache(PageCache *cache)

{

  if (cache == NULL) {

    return;

  }

  if (cache->cache != NULL) {

    unsigned int i;

    for (i = 0; i < cache->numCacheEntries; i++) {

      destroyVolumePage(&cache->cache[i].cp_pageData);

    }

  }

  FREE(cache->index);

  FREE(cache->cache);

  FREE(cache->searchPendingCounters);

  FREE(cache->readQueue);

  FREE(cache);

}

/**********************************************************************/

int invalidatePageCacheForChapter(PageCache          *cache,

                                  unsigned int        chapter,

                                  unsigned int        pagesPerChapter,

                                  InvalidationReason  reason)

{

  // We hold the readThreadsMutex.

  if ((cache == NULL) || (cache->cache == NULL)) {

    return UDS_SUCCESS;

  }

  int result;

  unsigned int i;

  for (i = 0; i < pagesPerChapter; i++) {

    unsigned int physicalPage = 1 + (pagesPerChapter * chapter) + i;

    result = findInvalidateAndMakeLeastRecent(cache, physicalPage,

                                              cache->readQueue, reason, false);

    if (result != UDS_SUCCESS) {

      return result;

    }

  }

  return UDS_SUCCESS;

}

/*********************************************************************/

void makePageMostRecent(PageCache *cache, CachedPage *page)

{

  // ASSERTION: We are either a zone thread holding a searchPendingCounter,

  //            or we are any thread holding the readThreadsMutex.

  if (atomic64_read(&cache->clock) != READ_ONCE(page->cp_lastUsed)) {

    WRITE_ONCE(page->cp_lastUsed, atomic64_inc_return(&cache->clock));

  }

}

/**

 * Get the least recent valid page from the cache.

 *

 * @param cache    the cache

 * @param pagePtr  a pointer to hold the new page (will be set to NULL

 *                 if the page was not found)

 *

 * @return UDS_SUCCESS or an error code

 **/

__attribute__((warn_unused_result))

static int getLeastRecentPage(PageCache *cache, CachedPage **pagePtr)

{

  // We hold the readThreadsMutex.

  int oldestIndex = 0;

  // Our first candidate is any page that does have a pending read.  We ensure

  // above that there are more entries than read threads, so there must be one.

  unsigned int i;

  for (i = 0;; i++) {

    if (i >= cache->numCacheEntries) {

      // This should never happen.

      return ASSERT(false, "oldest page is not NULL");

    }

    if (!cache->cache[i].cp_readPending) {

      oldestIndex = i;

      break;

    }

  }

  // Now find the least recently used page that does not have a pending read.

  for (i = 0; i < cache->numCacheEntries; i++) {

    if (!cache->cache[i].cp_readPending

        && (READ_ONCE(cache->cache[i].cp_lastUsed)

            <= READ_ONCE(cache->cache[oldestIndex].cp_lastUsed))) {

      oldestIndex = i;

    }

  }

  *pagePtr = &cache->cache[oldestIndex];

  return UDS_SUCCESS;

}

/***********************************************************************/

int getPageFromCache(PageCache     *cache,

                     unsigned int   physicalPage,

                     int            probeType,

                     CachedPage   **pagePtr)

{

  // ASSERTION: We are in a zone thread.

  // ASSERTION: We holding a searchPendingCounter or the readThreadsMutex.

  if (cache == NULL) {

    return logWarningWithStringError(UDS_BAD_STATE,

                                     "cannot get page with NULL cache");

  }

  // Get the cache page from the index

  CachedPage *page;

  int queueIndex = -1;

  int result = getPageNoStats(cache, physicalPage, &queueIndex, &page);

  if (result != UDS_SUCCESS) {

    return result;

  }

  CacheResultKind cacheResult = ((page != NULL)

                                 ? CACHE_RESULT_HIT

                                 : ((queueIndex != -1)

                                    ? CACHE_RESULT_QUEUED

                                    : CACHE_RESULT_MISS));

  incrementCacheCounter(&cache->counters, probeType, cacheResult);

  if (pagePtr != NULL) {

    *pagePtr = page;

  }

  return UDS_SUCCESS;

}

/***********************************************************************/

int enqueueRead(PageCache *cache, Request *request, unsigned int physicalPage)

{

  // We hold the readThreadsMutex.

  uint16_t first = cache->readQueueFirst;

  uint16_t last  = cache->readQueueLast;

  uint16_t next  = (last + 1) % cache->readQueueMaxSize;

  uint16_t readQueuePos;

  if ((cache->index[physicalPage] & VOLUME_CACHE_QUEUED_FLAG) == 0) {

    /* Not seen before, add this to the read queue and mark it as queued */

    if (next == first) {

      /* queue is full */

      return UDS_SUCCESS;

    }

    /* fill the read queue entry */

    cache->readQueue[last].physicalPage = physicalPage;

    cache->readQueue[last].invalid = false;

    /* point the cache index to it */

    readQueuePos = last;

    WRITE_ONCE(cache->index[physicalPage],

               readQueuePos | VOLUME_CACHE_QUEUED_FLAG);

    cache->readQueue[readQueuePos].requestList.first = NULL;

    cache->readQueue[readQueuePos].requestList.last = NULL;

    /* bump the last pointer */

    cache->readQueueLast = next;

  } else {

    /* It's already queued, just add on to it */

    readQueuePos = cache->index[physicalPage] & ~VOLUME_CACHE_QUEUED_FLAG;

  }

  int result = ASSERT((readQueuePos < cache->readQueueMaxSize),

                      "queue is not overfull");

  if (result != UDS_SUCCESS) {

    return result;

  }

  request->nextRequest = NULL;

  if (cache->readQueue[readQueuePos].requestList.first == NULL) {

    cache->readQueue[readQueuePos].requestList.first = request;

  } else {

    cache->readQueue[readQueuePos].requestList.last->nextRequest = request;

  }

  cache->readQueue[readQueuePos].requestList.last = request;

  return UDS_QUEUED;

}

/***********************************************************************/

bool reserveReadQueueEntry(PageCache     *cache,

                           unsigned int  *queuePos,

                           Request      **firstRequest,

                           unsigned int  *physicalPage,

                           bool          *invalid)

{

  // We hold the readThreadsMutex.

  uint16_t lastRead = cache->readQueueLastRead;

  // No items to dequeue

  if (lastRead == cache->readQueueLast) {

    return false;

  }

  unsigned int pageNo    = cache->readQueue[lastRead].physicalPage;

  bool         isInvalid = cache->readQueue[lastRead].invalid;

  uint16_t indexValue = cache->index[pageNo];

  bool     queued     = (indexValue & VOLUME_CACHE_QUEUED_FLAG) != 0;

  // ALB-1429 ... need to check to see if its still queued before resetting

  if (isInvalid && queued) {

    // invalidate cache index slot

    WRITE_ONCE(cache->index[pageNo], cache->numCacheEntries);

  }

  // If a sync read has taken this page, set invalid to true so we don't

  // overwrite, we simply just requeue requests.

  if (!queued) {

    isInvalid = true;

  }

  cache->readQueue[lastRead].reserved = true;

  *queuePos                = lastRead;

  *firstRequest            = cache->readQueue[lastRead].requestList.first;

  *physicalPage            = pageNo;

  *invalid                 = isInvalid;

  cache->readQueueLastRead = (lastRead  + 1) % cache->readQueueMaxSize;

  return true;

}

/************************************************************************/

void releaseReadQueueEntry(PageCache *cache, unsigned int queuePos)

{

  // We hold the readThreadsMutex.

  cache->readQueue[queuePos].reserved = false;

  uint16_t lastRead = cache->readQueueLastRead;

  // Move the readQueueFirst pointer along when we can

  while ((cache->readQueueFirst != lastRead)

         && (!cache->readQueue[cache->readQueueFirst].reserved)) {

    cache->readQueueFirst =

      (cache->readQueueFirst + 1) % cache->readQueueMaxSize;

  }

}

/***********************************************************************/

int selectVictimInCache(PageCache   *cache,

                        CachedPage **pagePtr)

{

  // We hold the readThreadsMutex.

  if (cache == NULL) {

    return logWarningWithStringError(UDS_BAD_STATE,

                                     "cannot put page in NULL cache");

  }

  CachedPage *page = NULL;

  int result = getLeastRecentPage(cache, &page);

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT((page != NULL), "least recent page was not NULL");

  if (result != UDS_SUCCESS) {

    return result;

  }

  // If the page is currently being pointed to by the page map, clear

  // it from the page map, and update cache stats

  if (page->cp_physicalPage != cache->numIndexEntries) {

    cache->counters.evictions++;

    WRITE_ONCE(cache->index[page->cp_physicalPage], cache->numCacheEntries);

    waitForPendingSearches(cache, page->cp_physicalPage);

  }

  page->cp_readPending = true;

  *pagePtr = page;

  return UDS_SUCCESS;

}

/***********************************************************************/

int putPageInCache(PageCache    *cache,

                   unsigned int  physicalPage,

                   CachedPage   *page)

{

  // We hold the readThreadsMutex.

  if (cache == NULL) {

    return logWarningWithStringError(UDS_BAD_STATE,

                                     "cannot complete page in NULL cache");

  }

  int result = ASSERT((page != NULL), "page to install exists");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT((page->cp_readPending),

                  "page to install has a pending read");

  if (result != UDS_SUCCESS) {

    return result;

  }

  clearPage(cache, page);

  page->cp_physicalPage = physicalPage;

  // Figure out the index into the cache array using pointer arithmetic

  uint16_t value = page - cache->cache;

  result = ASSERT((value < cache->numCacheEntries), "cache index is valid");

  if (result != UDS_SUCCESS) {

    return result;

  }

  makePageMostRecent(cache, page);

  page->cp_readPending = false;

  /*

   * We hold the readThreadsMutex, but we must have a write memory barrier

   * before making the CachedPage available to the readers that do not hold the

   * mutex.  The corresponding read memory barrier is in getPageNoStats.

   */

  smp_wmb();

  // Point the page map to the new page. Will clear queued flag

  WRITE_ONCE(cache->index[physicalPage], value);

  return UDS_SUCCESS;

}

/***********************************************************************/

void cancelPageInCache(PageCache    *cache,

                       unsigned int  physicalPage,

                       CachedPage   *page)

{

  // We hold the readThreadsMutex.

  if (cache == NULL) {

    logWarning("cannot cancel page in NULL cache");

    return;

  }

  int result = ASSERT((page != NULL), "page to install exists");

  if (result != UDS_SUCCESS) {

    return;

  }

  result = ASSERT((page->cp_readPending),

                  "page to install has a pending read");

  if (result != UDS_SUCCESS) {

    return;

  }

  clearPage(cache, page);

  page->cp_readPending = false;

  // Clear the page map for the new page. Will clear queued flag

  WRITE_ONCE(cache->index[physicalPage], cache->numCacheEntries);

}

/**********************************************************************/

size_t getPageCacheSize(PageCache *cache)

{

  if (cache == NULL) {

    return 0;

  }

  return sizeof(DeltaIndexPage) * cache->numCacheEntries;

}