/*

 * 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/indexZone.c#4 $

 */

#include "indexZone.h"

#include "errors.h"

#include "index.h"

#include "indexCheckpoint.h"

#include "indexRouter.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "permassert.h"

#include "request.h"

#include "sparseCache.h"

#include "uds.h"

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

int makeIndexZone(struct index *index, unsigned int zoneNumber)

{

  IndexZone *zone;

  int result = ALLOCATE(1, IndexZone, "index zone", &zone);

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = makeOpenChapter(index->volume->geometry, index->zoneCount,

                           &zone->openChapter);

  if (result != UDS_SUCCESS) {

    freeIndexZone(zone);

    return result;

  }

  result = makeOpenChapter(index->volume->geometry, index->zoneCount,

                           &zone->writingChapter);

  if (result != UDS_SUCCESS) {

    freeIndexZone(zone);

    return result;

  }

  zone->index              = index;

  zone->id                 = zoneNumber;

  index->zones[zoneNumber] = zone;

  return UDS_SUCCESS;

}

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

void freeIndexZone(IndexZone *zone)

{

  if (zone == NULL) {

    return;

  }

  freeOpenChapter(zone->openChapter);

  freeOpenChapter(zone->writingChapter);

  FREE(zone);

}

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

bool isZoneChapterSparse(const IndexZone *zone,

                         uint64_t         virtualChapter)

{

  return isChapterSparse(zone->index->volume->geometry,

                         zone->oldestVirtualChapter,

                         zone->newestVirtualChapter,

                         virtualChapter);

}

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

void setActiveChapters(IndexZone *zone)

{

  zone->oldestVirtualChapter = zone->index->oldestVirtualChapter;

  zone->newestVirtualChapter = zone->index->newestVirtualChapter;

}

/**

 * Swap the open and writing chapters after blocking until there are no active

 * chapter writers on the index.

 *

 * @param zone  The zone swapping chapters

 *

 * @return UDS_SUCCESS or a return code

 **/

static int swapOpenChapter(IndexZone *zone)

{

  // Wait for any currently writing chapter to complete

  int result = finishPreviousChapter(zone->index->chapterWriter,

                                     zone->newestVirtualChapter);

  if (result != UDS_SUCCESS) {

    return result;

  }

  // Swap the writing and open chapters

  OpenChapterZone *tempChapter = zone->openChapter;

  zone->openChapter            = zone->writingChapter;

  zone->writingChapter         = tempChapter;

  return UDS_SUCCESS;

}

/**

 * Advance to a new open chapter, and forget the oldest chapter in the

 * index if necessary.

 *

 * @param zone                 The zone containing the chapter to reap

 *

 * @return UDS_SUCCESS or an error code

 **/

static int reapOldestChapter(IndexZone *zone)

{

  Index *index = zone->index;

  unsigned int chaptersPerVolume = index->volume->geometry->chaptersPerVolume;

  int result

    = ASSERT(((zone->newestVirtualChapter - zone->oldestVirtualChapter)

              <= chaptersPerVolume),

             "newest (%llu) and oldest (%llu) virtual chapters "

             "less than or equal to chapters per volume (%u)",

             zone->newestVirtualChapter, zone->oldestVirtualChapter,

             chaptersPerVolume);

  if (result != UDS_SUCCESS) {

    return result;

  }

  setMasterIndexZoneOpenChapter(index->masterIndex, zone->id,

                                zone->newestVirtualChapter);

  return UDS_SUCCESS;

}

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

int executeSparseCacheBarrierMessage(IndexZone          *zone,

                                     BarrierMessageData *barrier)

{

  /*

   * Check if the chapter index for the virtual chapter is already in the

   * cache, and if it's not, rendezvous with the other zone threads to add the

   * chapter index to the sparse index cache.

   */

  return updateSparseCache(zone, barrier->virtualChapter);

}

/**

 * Handle notification that some other zone has closed its open chapter. If

 * the chapter that was closed is still the open chapter for this zone,

 * close it now in order to minimize skew.

 *

 * @param zone          The zone receiving the notification

 * @param chapterClosed The notification

 *

 * @return UDS_SUCCESS or an error code

 **/

static int handleChapterClosed(IndexZone                *zone,

                               ChapterClosedMessageData *chapterClosed)

{

  if (zone->newestVirtualChapter == chapterClosed->virtualChapter) {

    return openNextChapter(zone, NULL);

  }

  return UDS_SUCCESS;

}

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

int dispatchIndexZoneControlRequest(Request *request)

{

  ZoneMessage *message = &request->zoneMessage;

  IndexZone *zone = message->index->zones[request->zoneNumber];

  switch (request->action) {

  case REQUEST_SPARSE_CACHE_BARRIER:

    return executeSparseCacheBarrierMessage(zone, &message->data.barrier);

  case REQUEST_ANNOUNCE_CHAPTER_CLOSED:

    return handleChapterClosed(zone, &message->data.chapterClosed);

  default:

    return ASSERT_FALSE("valid control message type: %d", request->action);

  }

}

/**

 * Announce the closure of the current open chapter to the other zones.

 *

 * @param request       The request which caused the chapter to close

 *                      (may be NULL)

 * @param zone          The zone which first closed the chapter

 * @param closedChapter The chapter which was closed

 *

 * @return UDS_SUCCESS or an error code

 **/

static int announceChapterClosed(Request   *request,

                                 IndexZone *zone,

                                 uint64_t   closedChapter)

{

  IndexRouter *router = ((request != NULL) ? request->router : NULL);

  ZoneMessage zoneMessage = {

    .index = zone->index,

    .data  = {

      .chapterClosed = { .virtualChapter = closedChapter }

    }

  };

  unsigned int i;

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

    if (zone->id == i) {

      continue;

    }

    int result;

    if (router != NULL) {

      result = launchZoneControlMessage(REQUEST_ANNOUNCE_CHAPTER_CLOSED,

                                        zoneMessage, i, router);

    } else {

      // We're in a test which doesn't have zone queues, so we can just

      // call the message function directly.

      result = handleChapterClosed(zone->index->zones[i],

                                   &zoneMessage.data.chapterClosed);

    }

    if (result != UDS_SUCCESS) {

      return result;

    }

  }

  return UDS_SUCCESS;

}

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

int openNextChapter(IndexZone *zone, Request *request)

{

  logDebug("closing chapter %llu of zone %d after %u entries (%u short)",

           zone->newestVirtualChapter, zone->id, zone->openChapter->size,

           zone->openChapter->capacity - zone->openChapter->size);

  int result = swapOpenChapter(zone);

  if (result != UDS_SUCCESS) {

    return result;

  }

  uint64_t closedChapter = zone->newestVirtualChapter++;

  result = reapOldestChapter(zone);

  if (result != UDS_SUCCESS) {

    return logUnrecoverable(result, "reapOldestChapter failed");

  }

  resetOpenChapter(zone->openChapter);

  // begin, continue, or finish the checkpoint processing

  // moved above startClosingChapter because some of the

  // checkpoint processing now done by the chapter writer thread

  result = processCheckpointing(zone->index,

                                zone->id,

                                zone->newestVirtualChapter);

  if (result != UDS_SUCCESS) {

    return result;

  }

  unsigned int finishedZones = startClosingChapter(zone->index->chapterWriter,

                                                   zone->id,

                                                   zone->writingChapter);

  if ((finishedZones == 1) && (zone->index->zoneCount > 1)) {

    // This is the first zone of a multi-zone index to close this chapter,

    // so inform the other zones in order to control zone skew.

    result = announceChapterClosed(request, zone, closedChapter);

    if (result != UDS_SUCCESS) {

      return result;

    }

  }

  // If the chapter being opened won't overwrite the oldest chapter, we're

  // done.

  if (!areSamePhysicalChapter(zone->index->volume->geometry,

                              zone->newestVirtualChapter,

                              zone->oldestVirtualChapter)) {

    return UDS_SUCCESS;

  }

  uint64_t victim = zone->oldestVirtualChapter++;

  if (finishedZones < zone->index->zoneCount) {

    // We are not the last zone to close the chapter, so we're done

    return UDS_SUCCESS;

  }

  /*

   * We are the last zone to close the chapter, so clean up the cache. That

   * it is safe to let the last thread out of the previous chapter to do this

   * relies on the fact that although the new open chapter shadows the oldest

   * chapter in the cache, until we write the new open chapter to disk, we'll

   * never look for it in the cache.

   */

  return forgetChapter(zone->index->volume, victim, INVALIDATION_EXPIRE);

}

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

IndexRegion computeIndexRegion(const IndexZone *zone,

                               uint64_t         virtualChapter)

{

  if (virtualChapter == zone->newestVirtualChapter) {

    return LOC_IN_OPEN_CHAPTER;

  }

  return (isZoneChapterSparse(zone, virtualChapter)

          ? LOC_IN_SPARSE : LOC_IN_DENSE);

}

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

int getRecordFromZone(IndexZone *zone,

                      Request   *request,

                      bool      *found,

                      uint64_t   virtualChapter)

{

  if (virtualChapter == zone->newestVirtualChapter) {

    searchOpenChapter(zone->openChapter, &request->chunkName,

                      &request->oldMetadata, found);

    return UDS_SUCCESS;

  }

  if ((zone->newestVirtualChapter > 0)

      && (virtualChapter == (zone->newestVirtualChapter - 1))

      && (zone->writingChapter->size > 0)) {

    // Only search the writing chapter if it is full, else look on disk.

    searchOpenChapter(zone->writingChapter, &request->chunkName,

                      &request->oldMetadata, found);

    return UDS_SUCCESS;

  }

  // The slow lane thread has determined the location previously. We don't need

  // to search again. Just return the location.

  if (request->slLocationKnown) {

    *found = request->slLocation != LOC_UNAVAILABLE;

    return UDS_SUCCESS;

  }

  Volume *volume = zone->index->volume;

  if (isZoneChapterSparse(zone, virtualChapter)

      && sparseCacheContains(volume->sparseCache, virtualChapter,

                             request->zoneNumber)) {

    // The named chunk, if it exists, is in a sparse chapter that is cached,

    // so just run the chunk through the sparse chapter cache search.

    return searchSparseCacheInZone(zone, request, virtualChapter, found);

  }

  return searchVolumePageCache(volume, request, &request->chunkName,

                               virtualChapter, &request->oldMetadata, found);

}

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

int putRecordInZone(IndexZone          *zone,

                    Request            *request,

                    const UdsChunkData *metadata)

{

  unsigned int remaining;

  int result = putOpenChapter(zone->openChapter, &request->chunkName, metadata,

                              &remaining);

  if (result != UDS_SUCCESS) {

    return result;

  }

  if (remaining == 0) {

    return openNextChapter(zone, request);

  }

  return UDS_SUCCESS;

}

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

int searchSparseCacheInZone(IndexZone *zone,

                            Request   *request,

                            uint64_t   virtualChapter,

                            bool      *found)

{

  int recordPageNumber;

  int result = searchSparseCache(zone, &request->chunkName, &virtualChapter,

                                 &recordPageNumber);

  if ((result != UDS_SUCCESS) || (virtualChapter == UINT64_MAX)) {

    return result;

  }

  Volume *volume = zone->index->volume;

  // XXX map to physical chapter and validate. It would be nice to just pass

  // the virtual in to the slow lane, since it's tracking invalidations.

  unsigned int chapter

    = mapToPhysicalChapter(volume->geometry, virtualChapter);

  return searchCachedRecordPage(volume, request, &request->chunkName, chapter,

                                recordPageNumber, &request->oldMetadata,

                                found);

}