/*

 * 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/volume.c#23 $

 */

#include "volume.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 "request.h"

#include "sparseCache.h"

#include "stringUtils.h"

#include "threads.h"

enum {

  MAX_BAD_CHAPTERS = 100,           // max number of contiguous bad chapters

  DEFAULT_VOLUME_READ_THREADS = 2,  // Default number of reader threads

  MAX_VOLUME_READ_THREADS = 16,     // Maximum number of reader threads

};

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

static unsigned int getReadThreads(const struct uds_parameters *userParams)

{

  unsigned int readThreads = (userParams == NULL

                              ? DEFAULT_VOLUME_READ_THREADS

                              : userParams->read_threads);

  if (readThreads < 1) {

    readThreads = 1;

  }

  if (readThreads > MAX_VOLUME_READ_THREADS) {

    readThreads = MAX_VOLUME_READ_THREADS;

  }

  return readThreads;

}

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

static INLINE unsigned int mapToPageNumber(Geometry     *geometry,

                                           unsigned int  physicalPage)

{

  return ((physicalPage - 1) % geometry->pagesPerChapter);

}

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

static INLINE unsigned int mapToChapterNumber(Geometry     *geometry,

                                              unsigned int  physicalPage)

{

  return ((physicalPage - 1) / geometry->pagesPerChapter);

}

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

static INLINE bool isRecordPage(Geometry *geometry, unsigned int physicalPage)

{

  return (((physicalPage - 1) % geometry->pagesPerChapter)

          >= geometry->indexPagesPerChapter);

}

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

static INLINE unsigned int getZoneNumber(Request *request)

{

  return (request == NULL) ? 0 : request->zoneNumber;

}

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

int mapToPhysicalPage(const Geometry *geometry, int chapter, int page)

{

  // Page zero is the header page, so the first index page in the

  // first chapter is physical page one.

  return (1 + (geometry->pagesPerChapter * chapter) + page);

}

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

static void waitForReadQueueNotFull(Volume *volume, Request *request)

{

  unsigned int zoneNumber = getZoneNumber(request);

  InvalidateCounter invalidateCounter = getInvalidateCounter(volume->pageCache,

                                                             zoneNumber);

  if (searchPending(invalidateCounter)) {

    // Increment the invalidate counter to avoid deadlock where the reader

    // threads cannot make progress because they are waiting on the counter

    // and the index thread cannot because the read queue is full.

    endPendingSearch(volume->pageCache, zoneNumber);

  }

  while (readQueueIsFull(volume->pageCache)) {

    logDebug("Waiting until read queue not full");

    signalCond(&volume->readThreadsCond);

    waitCond(&volume->readThreadsReadDoneCond, &volume->readThreadsMutex);

  }

  if (searchPending(invalidateCounter)) {

    // Increment again so we get back to an odd value.

    beginPendingSearch(volume->pageCache, pageBeingSearched(invalidateCounter),

                       zoneNumber);

  }

}

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

int enqueuePageRead(Volume *volume, Request *request, int physicalPage)

{

  // Don't allow new requests if we are shutting down, but make sure

  // to process any requests that are still in the pipeline.

  if ((volume->readerState & READER_STATE_EXIT) != 0) {

    logInfo("failed to queue read while shutting down");

    return UDS_SHUTTINGDOWN;

  }

  // Mark the page as queued in the volume cache, for chapter invalidation to

  // be able to cancel a read.

  // If we are unable to do this because the queues are full, flush them first

  int result;

  while ((result = enqueueRead(volume->pageCache, request, physicalPage))

         == UDS_SUCCESS) {

    logDebug("Read queues full, waiting for reads to finish");

    waitForReadQueueNotFull(volume, request);

  }

  if (result == UDS_QUEUED) {

    /* signal a read thread */

    signalCond(&volume->readThreadsCond);

  }

  return result;

}

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

static INLINE void waitToReserveReadQueueEntry(Volume        *volume,

                                               unsigned int  *queuePos,

                                               Request      **requestList,

                                               unsigned int  *physicalPage,

                                               bool          *invalid)

{

  while (((volume->readerState & READER_STATE_EXIT) == 0)

         && (((volume->readerState & READER_STATE_STOP) != 0)

             || !reserveReadQueueEntry(volume->pageCache, queuePos,

                                       requestList, physicalPage, invalid))) {

    waitCond(&volume->readThreadsCond, &volume->readThreadsMutex);

  }

}

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

static int initChapterIndexPage(const Volume   *volume,

                                byte           *indexPage,

                                unsigned int    chapter,

                                unsigned int    indexPageNumber,

                                DeltaIndexPage *chapterIndexPage)

{

  Geometry *geometry = volume->geometry;

  int result = initializeChapterIndexPage(chapterIndexPage, geometry,

                                          indexPage, volume->nonce);

  if (volume->lookupMode == LOOKUP_FOR_REBUILD) {

    return result;

  }

  if (result != UDS_SUCCESS) {

    return logErrorWithStringError(result,

                                   "Reading chapter index page for chapter %u"

                                   " page %u",

                                   chapter, indexPageNumber);

  }

  IndexPageBounds bounds;

  result = getListNumberBounds(volume->indexPageMap, chapter,

                               indexPageNumber, &bounds);

  if (result != UDS_SUCCESS) {

    return result;

  }

  uint64_t     ciVirtual = chapterIndexPage->virtualChapterNumber;

  unsigned int ciChapter = mapToPhysicalChapter(geometry, ciVirtual);

  if ((chapter == ciChapter)

      && (bounds.lowestList == chapterIndexPage->lowestListNumber)

      && (bounds.highestList == chapterIndexPage->highestListNumber)) {

    return UDS_SUCCESS;

  }

  logWarning("Index page map updated to %llu",

             getLastUpdate(volume->indexPageMap));

  logWarning("Page map expects that chapter %u page %u has range %u to %u, "

             "but chapter index page has chapter %" PRIu64

             " with range %u to %u",

             chapter, indexPageNumber, bounds.lowestList, bounds.highestList,

             ciVirtual, chapterIndexPage->lowestListNumber,

             chapterIndexPage->highestListNumber);

  return ASSERT_WITH_ERROR_CODE(false,

                                UDS_CORRUPT_DATA,

                                "index page map mismatch with chapter index");

}

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

static int initializeIndexPage(const Volume *volume,

                               unsigned int  physicalPage,

                               CachedPage   *page)

{

  unsigned int chapter = mapToChapterNumber(volume->geometry, physicalPage);

  unsigned int indexPageNumber = mapToPageNumber(volume->geometry,

                                                 physicalPage);

  int result = initChapterIndexPage(volume, getPageData(&page->cp_pageData),

                                    chapter, indexPageNumber,

                                    &page->cp_indexPage);

  return result;

}

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

static void readThreadFunction(void *arg)

{

  Volume       *volume  = arg;

  unsigned int  queuePos;

  Request      *requestList;

  unsigned int  physicalPage;

  bool          invalid = false;

  logDebug("reader starting");

  lockMutex(&volume->readThreadsMutex);

  while (true) {

    waitToReserveReadQueueEntry(volume, &queuePos, &requestList, &physicalPage,

                                &invalid);

    if ((volume->readerState & READER_STATE_EXIT) != 0) {

      break;

    }

    volume->busyReaderThreads++;

    bool recordPage = isRecordPage(volume->geometry, physicalPage);

    CachedPage *page = NULL;

    int result = UDS_SUCCESS;

    if (!invalid) {

      // Find a place to put the read queue page we reserved above.

      result = selectVictimInCache(volume->pageCache, &page);

      if (result == UDS_SUCCESS) {

        unlockMutex(&volume->readThreadsMutex);

        result = readVolumePage(&volume->volumeStore, physicalPage,

                                &page->cp_pageData);

        if (result != UDS_SUCCESS) {

          logWarning("Error reading page %u from volume", physicalPage);

          cancelPageInCache(volume->pageCache, physicalPage, page);

        }

        lockMutex(&volume->readThreadsMutex);

      } else {

        logWarning("Error selecting cache victim for page read");

      }

      if (result == UDS_SUCCESS) {

        if (!volume->pageCache->readQueue[queuePos].invalid) {

          if (!recordPage) {

            result = initializeIndexPage(volume, physicalPage, page);

            if (result != UDS_SUCCESS) {

              logWarning("Error initializing chapter index page");

              cancelPageInCache(volume->pageCache, physicalPage, page);

            }

          }

          if (result == UDS_SUCCESS) {

            result = putPageInCache(volume->pageCache, physicalPage, page);

            if (result != UDS_SUCCESS) {

              logWarning("Error putting page %u in cache", physicalPage);

              cancelPageInCache(volume->pageCache, physicalPage, page);

            }

          }

        } else {

          logWarning("Page %u invalidated after read", physicalPage);

          cancelPageInCache(volume->pageCache, physicalPage, page);

          invalid = true;

        }

      }

    } else {

      logDebug("Requeuing requests for invalid page");

    }

    if (invalid) {

      result = UDS_SUCCESS;

      page = NULL;

    }

    while (requestList != NULL) {

      Request *request = requestList;

      requestList = request->nextRequest;

      /*

       * If we've read in a record page, we're going to do an immediate search,

       * in an attempt to speed up processing when we requeue the request, so

       * that it doesn't have to go back into the getRecordFromZone code again.

       * However, if we've just read in an index page, we don't want to search.

       * We want the request to be processed again and getRecordFromZone to be

       * run.  We have added new fields in request to allow the index code to

       * know whether it can stop processing before getRecordFromZone is called

       * again.

       */

      if ((result == UDS_SUCCESS) && (page != NULL) && recordPage) {

        if (searchRecordPage(getPageData(&page->cp_pageData),

                             &request->chunkName, volume->geometry,

                             &request->oldMetadata)) {

          request->slLocation = LOC_IN_DENSE;

        } else {

          request->slLocation = LOC_UNAVAILABLE;

        }

        request->slLocationKnown = true;

      }

      // reflect any read failures in the request status

      request->status = result;

      restartRequest(request);

    }

    releaseReadQueueEntry(volume->pageCache, queuePos);

    volume->busyReaderThreads--;

    broadcastCond(&volume->readThreadsReadDoneCond);

  }

  unlockMutex(&volume->readThreadsMutex);

  logDebug("reader done");

}

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

static int readPageLocked(Volume        *volume,

                          Request       *request,

                          unsigned int   physicalPage,

                          bool           syncRead,

                          CachedPage   **pagePtr)

{

  syncRead |= ((volume->lookupMode == LOOKUP_FOR_REBUILD)

               || (request == NULL)

               || (request->session == NULL));

  int result = UDS_SUCCESS;

  CachedPage *page = NULL;

  if (syncRead) {

    // Find a place to put the page.

    result = selectVictimInCache(volume->pageCache, &page);

    if (result != UDS_SUCCESS) {

      logWarning("Error selecting cache victim for page read");

      return result;

    }

    result = readVolumePage(&volume->volumeStore, physicalPage,

                            &page->cp_pageData);

    if (result != UDS_SUCCESS) {

      logWarning("Error reading page %u from volume", physicalPage);

      cancelPageInCache(volume->pageCache, physicalPage, page);

      return result;

    }

    if (!isRecordPage(volume->geometry, physicalPage)) {

      result = initializeIndexPage(volume, physicalPage, page);

      if (result != UDS_SUCCESS) {

        if (volume->lookupMode != LOOKUP_FOR_REBUILD) {

          logWarning("Corrupt index page %u", physicalPage);

        }

        cancelPageInCache(volume->pageCache, physicalPage, page);

        return result;

      }

    }

    result = putPageInCache(volume->pageCache, physicalPage, page);

    if (result != UDS_SUCCESS) {

      logWarning("Error putting page %u in cache", physicalPage);

      cancelPageInCache(volume->pageCache, physicalPage, page);

      return result;

    }

  } else {

    result = enqueuePageRead(volume, request, physicalPage);

    if (result != UDS_SUCCESS) {

      return result;

    }

  }

  *pagePtr = page;

  return UDS_SUCCESS;

}

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

int getPageLocked(Volume          *volume,

                  Request         *request,

                  unsigned int     physicalPage,

                  CacheProbeType   probeType,

                  CachedPage     **pagePtr)

{

  CachedPage *page = NULL;

  int result = getPageFromCache(volume->pageCache, physicalPage, probeType,

                                &page);

  if (result != UDS_SUCCESS) {

    return result;

  }

  if (page == NULL) {

    result = readPageLocked(volume, request, physicalPage, true, &page);

    if (result != UDS_SUCCESS) {

      return result;

    }

  } else if (getZoneNumber(request) == 0) {

    // Only 1 zone is responsible for updating LRU

    makePageMostRecent(volume->pageCache, page);

  }

  *pagePtr = page;

  return UDS_SUCCESS;

}

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

int getPageProtected(Volume          *volume,

                     Request         *request,

                     unsigned int     physicalPage,

                     CacheProbeType   probeType,

                     CachedPage     **pagePtr)

{

  CachedPage *page = NULL;

  int result = getPageFromCache(volume->pageCache, physicalPage,

                                probeType | CACHE_PROBE_IGNORE_FAILURE,

                                &page);

  if (result != UDS_SUCCESS) {

    return result;

  }

  unsigned int zoneNumber = getZoneNumber(request);

  // If we didn't find a page we need to enqueue a read for it, in which

  // case we need to grab the mutex.

  if (page == NULL) {

    endPendingSearch(volume->pageCache, zoneNumber);

    lockMutex(&volume->readThreadsMutex);

    /*

     * Do the lookup again while holding the read mutex (no longer the fast

     * case so this should be ok to repeat). We need to do this because an

     * page may have been added to the page map by the reader thread between

     * the time searched above and the time we went to actually try to enqueue

     * it below. This could result in us enqueuing another read for an page

     * which is already in the cache, which would mean we end up with two

     * entries in the cache for the same page.

     */

    result

      = getPageFromCache(volume->pageCache, physicalPage, probeType, &page);

    if (result != UDS_SUCCESS) {

      /*

       * In non-success cases (anything not UDS_SUCCESS, meaning both

       * UDS_QUEUED and "real" errors), the caller doesn't get a

       * handle on a cache page, so it can't continue the search, and

       * we don't need to prevent other threads from messing with the

       * cache.

       *

       * However, we do need to set the "search pending" flag because

       * the callers expect it to always be set on return, even if

       * they can't actually do the search.

       *

       * Doing the calls in this order ought to be faster, since we

       * let other threads have the reader thread mutex (which can

       * require a syscall) ASAP, and set the "search pending" state

       * that can block the reader thread as the last thing.

       */

      unlockMutex(&volume->readThreadsMutex);

      beginPendingSearch(volume->pageCache, physicalPage, zoneNumber);

      return result;

    }

    // If we found the page now, we can release the mutex and proceed

    // as if this were the fast case.

    if (page != NULL) {

      /*

       * If we found a page (*pagePtr != NULL and return

       * UDS_SUCCESS), then we're telling the caller where to look for

       * the cache page, and need to switch to "reader thread

       * unlocked" and "search pending" state in careful order so no

       * other thread can mess with the data before our caller gets to

       * look at it.

       */

      beginPendingSearch(volume->pageCache, physicalPage, zoneNumber);

      unlockMutex(&volume->readThreadsMutex);

    }

  }

  if (page == NULL) {

    result = readPageLocked(volume, request, physicalPage, false, &page);

    if (result != UDS_SUCCESS) {

      /*

       * This code path is used frequently in the UDS_QUEUED case, so

       * the performance gain from unlocking first, while "search

       * pending" mode is off, turns out to be significant in some

       * cases.

       */

      unlockMutex(&volume->readThreadsMutex);

      beginPendingSearch(volume->pageCache, physicalPage, zoneNumber);

      return result;

    }

    // See above re: ordering requirement.

    beginPendingSearch(volume->pageCache, physicalPage, zoneNumber);

    unlockMutex(&volume->readThreadsMutex);

  } else {

    if (getZoneNumber(request) == 0 ) {

      // Only 1 zone is responsible for updating LRU

      makePageMostRecent(volume->pageCache, page);

    }

  }

  *pagePtr = page;

  return UDS_SUCCESS;

}

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

int getPage(Volume          *volume,

            unsigned int     chapter,

            unsigned int     pageNumber,

            CacheProbeType   probeType,

            byte           **dataPtr,

            DeltaIndexPage **indexPagePtr)

{

  unsigned int physicalPage

    = mapToPhysicalPage(volume->geometry, chapter, pageNumber);

  lockMutex(&volume->readThreadsMutex);

  CachedPage *page = NULL;

  int result = getPageLocked(volume, NULL, physicalPage, probeType, &page);

  unlockMutex(&volume->readThreadsMutex);

  if (dataPtr != NULL) {

    *dataPtr = (page != NULL) ? getPageData(&page->cp_pageData) : NULL;

  }

  if (indexPagePtr != NULL) {

    *indexPagePtr = (page != NULL) ? &page->cp_indexPage : NULL;

  }

  return result;

}

/**

 * Search for a chunk name in a cached index page or chapter index, returning

 * the record page number from a chapter index match.

 *

 * @param volume           the volume containing the index page to search

 * @param request          the request originating the search (may be NULL for

 *                         a direct query from volume replay)

 * @param name             the name of the block or chunk

 * @param chapter          the chapter to search

 * @param indexPageNumber  the index page number of the page to search

 * @param recordPageNumber pointer to return the chapter record page number

 *                         (value will be NO_CHAPTER_INDEX_ENTRY if the name

 *                         was not found)

 *

 * @return UDS_SUCCESS or an error code

 **/

static int searchCachedIndexPage(Volume             *volume,

                                 Request            *request,

                                 const UdsChunkName *name,

                                 unsigned int        chapter,

                                 unsigned int        indexPageNumber,

                                 int                *recordPageNumber)

{

  unsigned int zoneNumber = getZoneNumber(request);

  unsigned int physicalPage

    = mapToPhysicalPage(volume->geometry, chapter, indexPageNumber);

  /*

   * Make sure the invalidate counter is updated before we try and read from

   * the page map.  This prevents this thread from reading a page in the

   * page map which has already been marked for invalidation by the reader

   * thread, before the reader thread has noticed that the invalidateCounter

   * has been incremented.

   */

  beginPendingSearch(volume->pageCache, physicalPage, zoneNumber);

  CachedPage *page = NULL;

  int result = getPageProtected(volume, request, physicalPage,

                                cacheProbeType(request, true), &page);

  if (result != UDS_SUCCESS) {

    endPendingSearch(volume->pageCache, zoneNumber);

    return result;

  }

  result

    = ASSERT_LOG_ONLY(searchPending(getInvalidateCounter(volume->pageCache,

                                                         zoneNumber)),

                      "Search is pending for zone %u", zoneNumber);

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = searchChapterIndexPage(&page->cp_indexPage, volume->geometry, name,

                                  recordPageNumber);

  endPendingSearch(volume->pageCache, zoneNumber);

  return result;

}

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

int searchCachedRecordPage(Volume             *volume,

                           Request            *request,

                           const UdsChunkName *name,

                           unsigned int        chapter,

                           int                 recordPageNumber,

                           UdsChunkData       *duplicate,

                           bool               *found)

{

  *found = false;

  if (recordPageNumber == NO_CHAPTER_INDEX_ENTRY) {

    // No record for that name can exist in the chapter.

    return UDS_SUCCESS;

  }

  Geometry *geometry = volume->geometry;

  int result = ASSERT(((recordPageNumber >= 0)

                       && ((unsigned int) recordPageNumber

                           < geometry->recordPagesPerChapter)),

                      "0 <= %d <= %u",

                      recordPageNumber, geometry->recordPagesPerChapter);

  if (result != UDS_SUCCESS) {

    return result;

  }

  unsigned int pageNumber = geometry->indexPagesPerChapter + recordPageNumber;

  unsigned int zoneNumber = getZoneNumber(request);

  int physicalPage

    = mapToPhysicalPage(volume->geometry, chapter, pageNumber);

  /*

   * Make sure the invalidate counter is updated before we try and read from

   * the page map. This prevents this thread from reading a page in the page

   * map which has already been marked for invalidation by the reader thread,

   * before the reader thread has noticed that the invalidateCounter has been

   * incremented.

   */

  beginPendingSearch(volume->pageCache, physicalPage, zoneNumber);

  CachedPage *recordPage;

  result = getPageProtected(volume, request, physicalPage,

                            cacheProbeType(request, false), &recordPage);

  if (result != UDS_SUCCESS) {

    endPendingSearch(volume->pageCache, zoneNumber);

    return result;

  }

  if (searchRecordPage(getPageData(&recordPage->cp_pageData), name, geometry,

                       duplicate)) {

    *found = true;

  }

  endPendingSearch(volume->pageCache, zoneNumber);

  return UDS_SUCCESS;

}

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

int readChapterIndexFromVolume(const Volume       *volume,

                               uint64_t            virtualChapter,

                               struct volume_page  volumePages[],

                               DeltaIndexPage      indexPages[])

{

  const Geometry *geometry = volume->geometry;

  unsigned int physicalChapter = mapToPhysicalChapter(geometry,

                                                      virtualChapter);

  int physicalPage = mapToPhysicalPage(geometry, physicalChapter, 0);

  prefetchVolumePages(&volume->volumeStore, physicalPage,

                      geometry->indexPagesPerChapter);

  unsigned int i;

  struct volume_page volumePage;

  int result = initializeVolumePage(geometry, &volumePage);

  for (i = 0; i < geometry->indexPagesPerChapter; i++) {

    int result = readVolumePage(&volume->volumeStore, physicalPage + i,

                                &volumePages[i]);

    if (result != UDS_SUCCESS) {

      break;

    }

    byte *indexPage = getPageData(&volumePages[i]);

    result = initChapterIndexPage(volume, indexPage, physicalChapter, i,

                                  &indexPages[i]);

    if (result != UDS_SUCCESS) {

      break;

    }

  }

  destroyVolumePage(&volumePage);

  return result;

}

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

int searchVolumePageCache(Volume             *volume,

                          Request            *request,

                          const UdsChunkName *name,

                          uint64_t            virtualChapter,

                          UdsChunkData       *metadata,

                          bool               *found)

{

  unsigned int physicalChapter

    = mapToPhysicalChapter(volume->geometry, virtualChapter);

  unsigned int indexPageNumber;

  int result = findIndexPageNumber(volume->indexPageMap, name, physicalChapter,

                                   &indexPageNumber);

  if (result != UDS_SUCCESS) {

    return result;

  }

  int recordPageNumber;

  result = searchCachedIndexPage(volume, request, name, physicalChapter,

                                 indexPageNumber, &recordPageNumber);

  if (result == UDS_SUCCESS) {

    result = searchCachedRecordPage(volume, request, name, physicalChapter,

                                    recordPageNumber, metadata, found);

  }

  return result;

}

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

int forgetChapter(Volume             *volume,

                  uint64_t            virtualChapter,

                  InvalidationReason  reason)

{

  logDebug("forgetting chapter %llu", virtualChapter);

  unsigned int physicalChapter

    = mapToPhysicalChapter(volume->geometry, virtualChapter);

  lockMutex(&volume->readThreadsMutex);

  int result

    = invalidatePageCacheForChapter(volume->pageCache, physicalChapter,

                                    volume->geometry->pagesPerChapter,

                                    reason);

  unlockMutex(&volume->readThreadsMutex);

  return result;

}

/**

 * Donate index page data to the page cache for an index page that was just

 * written to the volume.  The caller must already hold the reader thread

 * mutex.

 *

 * @param volume           the volume

 * @param physicalChapter  the physical chapter number of the index page

 * @param indexPageNumber  the chapter page number of the index page

 * @param scratchPage      the index page data

 **/

static int donateIndexPageLocked(Volume             *volume,

                                 unsigned int        physicalChapter,

                                 unsigned int        indexPageNumber,

                                 struct volume_page *scratchPage)

{

  unsigned int physicalPage

    = mapToPhysicalPage(volume->geometry, physicalChapter, indexPageNumber);

  // Find a place to put the page.

  CachedPage *page = NULL;

  int result = selectVictimInCache(volume->pageCache, &page);

  if (result != UDS_SUCCESS) {

    return result;

  }

  // Exchange the scratch page with the cache page

  swapVolumePages(&page->cp_pageData, scratchPage);

  result = initChapterIndexPage(volume, getPageData(&page->cp_pageData),

                                physicalChapter, indexPageNumber,

                                &page->cp_indexPage);

  if (result != UDS_SUCCESS) {

    logWarning("Error initialize chapter index page");

    cancelPageInCache(volume->pageCache, physicalPage, page);

    return result;

  }

  result = putPageInCache(volume->pageCache, physicalPage, page);

  if (result != UDS_SUCCESS) {

    logWarning("Error putting page %u in cache", physicalPage);

    cancelPageInCache(volume->pageCache, physicalPage, page);

    return result;

  }

  return UDS_SUCCESS;

}

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

int writeIndexPages(Volume            *volume,

                    int                physicalPage,

                    OpenChapterIndex  *chapterIndex,

                    byte             **pages)

{

  Geometry *geometry = volume->geometry;

  unsigned int physicalChapterNumber

    = mapToPhysicalChapter(geometry, chapterIndex->virtualChapterNumber);

  unsigned int deltaListNumber = 0;

  unsigned int indexPageNumber;

  for (indexPageNumber = 0;

       indexPageNumber < geometry->indexPagesPerChapter;

       indexPageNumber++) {

    int result = prepareToWriteVolumePage(&volume->volumeStore,

                                          physicalPage + indexPageNumber,

                                          &volume->scratchPage);

    if (result != UDS_SUCCESS) {

      return logWarningWithStringError(result, "failed to prepare index page");

    }

    // Pack as many delta lists into the index page as will fit.

    unsigned int listsPacked;

    bool lastPage = ((indexPageNumber + 1) == geometry->indexPagesPerChapter);

    result = packOpenChapterIndexPage(chapterIndex,

                                      getPageData(&volume->scratchPage),

                                      deltaListNumber, lastPage, &listsPacked);

    if (result != UDS_SUCCESS) {

      return logWarningWithStringError(result, "failed to pack index page");

    }

    result = writeVolumePage(&volume->volumeStore,

                             physicalPage + indexPageNumber,

                             &volume->scratchPage);

    if (result != UDS_SUCCESS) {

      return logWarningWithStringError(result,

                                       "failed to write chapter index page");

    }

    if (pages != NULL) {

      memcpy(pages[indexPageNumber], getPageData(&volume->scratchPage),

             geometry->bytesPerPage);

    }

    // Tell the index page map the list number of the last delta list that was

    // packed into the index page.

    if (listsPacked == 0) {

      logDebug("no delta lists packed on chapter %u page %u",

               physicalChapterNumber, indexPageNumber);

    } else {

      deltaListNumber += listsPacked;

    }

    result = updateIndexPageMap(volume->indexPageMap,

                                chapterIndex->virtualChapterNumber,

                                physicalChapterNumber,

                                indexPageNumber, deltaListNumber - 1);

    if (result != UDS_SUCCESS) {

      return logErrorWithStringError(result,

                                     "failed to update index page map");

    }

    // Donate the page data for the index page to the page cache.

    lockMutex(&volume->readThreadsMutex);

    result = donateIndexPageLocked(volume, physicalChapterNumber,

                                   indexPageNumber, &volume->scratchPage);

    unlockMutex(&volume->readThreadsMutex);

    if (result != UDS_SUCCESS) {

      return result;

    }

  }

  return UDS_SUCCESS;

}

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

int writeRecordPages(Volume                *volume,

                     int                    physicalPage,

                     const UdsChunkRecord   records[],

                     byte                 **pages)

{

  Geometry *geometry = volume->geometry;

  // Skip over the index pages, which come before the record pages

  physicalPage += geometry->indexPagesPerChapter;

  // The record array from the open chapter is 1-based.

  const UdsChunkRecord *nextRecord = &records[1];

  unsigned int recordPageNumber;

  for (recordPageNumber = 0;

       recordPageNumber < geometry->recordPagesPerChapter;

       recordPageNumber++) {

    int result = prepareToWriteVolumePage(&volume->volumeStore,

                                          physicalPage + recordPageNumber,

                                          &volume->scratchPage);

    if (result != UDS_SUCCESS) {

      return logWarningWithStringError(result,