/*

 * 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/vdo-releases/aluminum/src/c++/vdo/base/slabScrubber.c#6 $

 */

#include "slabScrubberInternals.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "adminState.h"

#include "blockAllocator.h"

#include "constants.h"

#include "readOnlyNotifier.h"

#include "recoveryJournal.h"

#include "refCounts.h"

#include "refCountsInternals.h"

#include "slab.h"

#include "slabJournalInternals.h"

/**

 * Allocate the buffer and extent used for reading the slab journal when

 * scrubbing a slab.

 *

 * @param scrubber         The slab scrubber for which to allocate

 * @param layer            The physical layer on which the scrubber resides

 * @param slabJournalSize  The size of a slab journal

 *

 * @return VDO_SUCCESS or an error

 **/

__attribute__((warn_unused_result))

static int allocateExtentAndBuffer(SlabScrubber  *scrubber,

                                   PhysicalLayer *layer,

                                   BlockCount     slabJournalSize)

{

  size_t bufferSize = VDO_BLOCK_SIZE * slabJournalSize;

  int result = ALLOCATE(bufferSize, char, __func__, &scrubber->journalData);

  if (result != VDO_SUCCESS) {

    return result;

  }

  return createExtent(layer, VIO_TYPE_SLAB_JOURNAL, VIO_PRIORITY_METADATA,

                      slabJournalSize, scrubber->journalData,

                      &scrubber->extent);

}

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

int makeSlabScrubber(PhysicalLayer     *layer,

                     BlockCount         slabJournalSize,

                     ReadOnlyNotifier  *readOnlyNotifier,

                     SlabScrubber     **scrubberPtr)

{

  SlabScrubber *scrubber;

  int result = ALLOCATE(1, SlabScrubber, __func__, &scrubber);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = allocateExtentAndBuffer(scrubber, layer, slabJournalSize);

  if (result != VDO_SUCCESS) {

    freeSlabScrubber(&scrubber);

    return result;

  }

  initializeCompletion(&scrubber->completion, SLAB_SCRUBBER_COMPLETION, layer);

  initializeRing(&scrubber->highPrioritySlabs);

  initializeRing(&scrubber->slabs);

  scrubber->readOnlyNotifier = readOnlyNotifier;

  scrubber->adminState.state = ADMIN_STATE_SUSPENDED;

  *scrubberPtr               = scrubber;

  return VDO_SUCCESS;

}

/**

 * Free the extent and buffer used for reading slab journals.

 *

 * @param scrubber  The scrubber

 **/

static void freeExtentAndBuffer(SlabScrubber *scrubber)

{

  freeExtent(&scrubber->extent);

  if (scrubber->journalData != NULL) {

    FREE(scrubber->journalData);

    scrubber->journalData = NULL;

  }

}

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

void freeSlabScrubber(SlabScrubber **scrubberPtr)

{

  if (*scrubberPtr == NULL) {

    return;

  }

  SlabScrubber *scrubber = *scrubberPtr;

  freeExtentAndBuffer(scrubber);

  FREE(scrubber);

  *scrubberPtr = NULL;

}

/**

 * Get the next slab to scrub.

 *

 * @param scrubber  The slab scrubber

 *

 * @return The next slab to scrub or NULL if there are none

 **/

static Slab *getNextSlab(SlabScrubber *scrubber)

{

  if (!isRingEmpty(&scrubber->highPrioritySlabs)) {

    return slabFromRingNode(scrubber->highPrioritySlabs.next);

  }

  if (!isRingEmpty(&scrubber->slabs)) {

    return slabFromRingNode(scrubber->slabs.next);

  }

  return NULL;

}

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

bool hasSlabsToScrub(SlabScrubber *scrubber)

{

  return (getNextSlab(scrubber) != NULL);

}

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

SlabCount getScrubberSlabCount(const SlabScrubber *scrubber)

{

  return relaxedLoad64(&scrubber->slabCount);

}

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

void registerSlabForScrubbing(SlabScrubber *scrubber,

                              Slab         *slab,

                              bool          highPriority)

{

  ASSERT_LOG_ONLY((slab->status != SLAB_REBUILT),

                  "slab to be scrubbed is unrecovered");

  if (slab->status != SLAB_REQUIRES_SCRUBBING) {

    return;

  }

  unspliceRingNode(&slab->ringNode);

  if (!slab->wasQueuedForScrubbing) {

    relaxedAdd64(&scrubber->slabCount, 1);

    slab->wasQueuedForScrubbing = true;

  }

  if (highPriority) {

    slab->status = SLAB_REQUIRES_HIGH_PRIORITY_SCRUBBING;

    pushRingNode(&scrubber->highPrioritySlabs, &slab->ringNode);

    return;

  }

  pushRingNode(&scrubber->slabs, &slab->ringNode);

}

/**

 * Stop scrubbing, either because there are no more slabs to scrub or because

 * there's been an error.

 *

 * @param scrubber  The scrubber

 **/

static void finishScrubbing(SlabScrubber *scrubber)

{

  if (!hasSlabsToScrub(scrubber)) {

    freeExtentAndBuffer(scrubber);

  }

  // Inform whoever is waiting that scrubbing has completed.

  completeCompletion(&scrubber->completion);

  bool notify = hasWaiters(&scrubber->waiters);

  // Note that the scrubber has stopped, and inform anyone who might be waiting

  // for that to happen.

  if (!finishDraining(&scrubber->adminState)) {

    scrubber->adminState.state = ADMIN_STATE_SUSPENDED;

  }

  /*

   * We can't notify waiters until after we've finished draining or they'll

   * just requeue. Fortunately if there were waiters, we can't have been freed

   * yet.

   */

  if (notify) {

    notifyAllWaiters(&scrubber->waiters, NULL, NULL);

  }

}

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

static void scrubNextSlab(SlabScrubber *scrubber);

/**

 * Notify the scrubber that a slab has been scrubbed. This callback is

 * registered in applyJournalEntries().

 *

 * @param completion  The slab rebuild completion

 **/

static void slabScrubbed(VDOCompletion *completion)

{

  SlabScrubber *scrubber = completion->parent;

  finishScrubbingSlab(scrubber->slab);

  relaxedAdd64(&scrubber->slabCount, -1);

  scrubNextSlab(scrubber);

}

/**

 * Abort scrubbing due to an error.

 *

 * @param scrubber  The slab scrubber

 * @param result    The error

 **/

static void abortScrubbing(SlabScrubber *scrubber, int result)

{

  enterReadOnlyMode(scrubber->readOnlyNotifier, result);

  setCompletionResult(&scrubber->completion, result);

  scrubNextSlab(scrubber);

}

/**

 * Handle errors while rebuilding a slab.

 *

 * @param completion  The slab rebuild completion

 **/

static void handleScrubberError(VDOCompletion *completion)

{

  abortScrubbing(completion->parent, completion->result);

}

/**

 * Apply all the entries in a block to the reference counts.

 *

 * @param block        A block with entries to apply

 * @param entryCount   The number of entries to apply

 * @param blockNumber  The sequence number of the block

 * @param slab         The slab to apply the entries to

 *

 * @return VDO_SUCCESS or an error code

 **/

static int applyBlockEntries(PackedSlabJournalBlock *block,

                             JournalEntryCount       entryCount,

                             SequenceNumber          blockNumber,

                             Slab                   *slab)

{

  JournalPoint entryPoint = {

    .sequenceNumber = blockNumber,

    .entryCount     = 0,

  };

  SlabBlockNumber maxSBN = slab->end - slab->start;

  while (entryPoint.entryCount < entryCount) {

    SlabJournalEntry entry = decodeSlabJournalEntry(block,

                                                    entryPoint.entryCount);

    if (entry.sbn > maxSBN) {

      // This entry is out of bounds.

      return logErrorWithStringError(VDO_CORRUPT_JOURNAL, "Slab journal entry"

                                     " (%llu, %u) had invalid offset"

                                     " %u in slab (size %u blocks)",

                                     blockNumber, entryPoint.entryCount,

                                     entry.sbn, maxSBN);

    }

    int result = replayReferenceCountChange(slab->referenceCounts, &entryPoint,

                                            entry);

    if (result != VDO_SUCCESS) {

      logErrorWithStringError(result, "Slab journal entry (%llu, %u)"

                              " (%s of offset %" PRIu32 ") could not be"

                              " applied in slab %u",

                              blockNumber, entryPoint.entryCount,

                              getJournalOperationName(entry.operation),

                              entry.sbn, slab->slabNumber);

      return result;

    }

    entryPoint.entryCount++;

  }

  return VDO_SUCCESS;

}

/**

 * Find the relevant extent of the slab journal and apply all valid entries.

 * This is a callback registered in startScrubbing().

 *

 * @param completion  The metadata read extent completion

 **/

static void applyJournalEntries(VDOCompletion *completion)

{

  SlabScrubber *scrubber        = completion->parent;

  Slab         *slab            = scrubber->slab;

  SlabJournal  *journal         = slab->journal;

  RefCounts    *referenceCounts = slab->referenceCounts;

  // Find the boundaries of the useful part of the journal.

  SequenceNumber  tail     = journal->tail;

  TailBlockOffset endIndex = getSlabJournalBlockOffset(journal, tail - 1);

  char *endData = scrubber->journalData + (endIndex * VDO_BLOCK_SIZE);

  PackedSlabJournalBlock *endBlock = (PackedSlabJournalBlock *) endData;

  SequenceNumber  head      = getUInt64LE(endBlock->header.fields.head);

  TailBlockOffset headIndex = getSlabJournalBlockOffset(journal, head);

  BlockCount      index     = headIndex;

  JournalPoint refCountsPoint   = referenceCounts->slabJournalPoint;

  JournalPoint lastEntryApplied = refCountsPoint;

  for (SequenceNumber sequence = head; sequence < tail; sequence++) {

    char *blockData = scrubber->journalData + (index * VDO_BLOCK_SIZE);

    PackedSlabJournalBlock *block  = (PackedSlabJournalBlock *) blockData;

    SlabJournalBlockHeader header;

    unpackSlabJournalBlockHeader(&block->header, &header);

    if ((header.nonce != slab->allocator->nonce)

        || (header.metadataType != VDO_METADATA_SLAB_JOURNAL)

        || (header.sequenceNumber != sequence)

        || (header.entryCount > journal->entriesPerBlock)

        || (header.hasBlockMapIncrements

            && (header.entryCount > journal->fullEntriesPerBlock))) {

      // The block is not what we expect it to be.

      logError("Slab journal block for slab %u was invalid",

               slab->slabNumber);

      abortScrubbing(scrubber, VDO_CORRUPT_JOURNAL);

      return;

    }

    int result = applyBlockEntries(block, header.entryCount, sequence, slab);

    if (result != VDO_SUCCESS) {

      abortScrubbing(scrubber, result);

      return;

    }

    lastEntryApplied.sequenceNumber = sequence;

    lastEntryApplied.entryCount     = header.entryCount - 1;

    index++;

    if (index == journal->size) {

      index = 0;

    }

  }

  // At the end of rebuild, the refCounts should be accurate to the end

  // of the journal we just applied.

  int result = ASSERT(!beforeJournalPoint(&lastEntryApplied, &refCountsPoint),

                      "Refcounts are not more accurate than the slab journal");

  if (result != VDO_SUCCESS) {

    abortScrubbing(scrubber, result);

    return;

  }

  // Save out the rebuilt reference blocks.

  prepareCompletion(completion, slabScrubbed, handleScrubberError,

                    completion->callbackThreadID, scrubber);

  startSlabAction(slab, ADMIN_STATE_SAVE_FOR_SCRUBBING, completion);

}

/**

 * Read the current slab's journal from disk now that it has been flushed.

 * This callback is registered in scrubNextSlab().

 *

 * @param completion  The scrubber's extent completion

 **/

static void startScrubbing(VDOCompletion *completion)

{

  SlabScrubber *scrubber = completion->parent;

  Slab         *slab     = scrubber->slab;

  if (getSummarizedCleanliness(slab->allocator->summary, slab->slabNumber)) {

    slabScrubbed(completion);

    return;

  }

  prepareCompletion(&scrubber->extent->completion, applyJournalEntries,

                    handleScrubberError, completion->callbackThreadID,

                    completion->parent);

  readMetadataExtent(scrubber->extent, slab->journalOrigin);

}

/**

 * Scrub the next slab if there is one.

 *

 * @param scrubber  The scrubber

 **/

static void scrubNextSlab(SlabScrubber *scrubber)

{

  // Note: this notify call is always safe only because scrubbing can only

  // be started when the VDO is quiescent.

  notifyAllWaiters(&scrubber->waiters, NULL, NULL);

  if (isReadOnly(scrubber->readOnlyNotifier)) {

    setCompletionResult(&scrubber->completion, VDO_READ_ONLY);

    finishScrubbing(scrubber);

    return;

  }

  Slab *slab = getNextSlab(scrubber);

  if ((slab == NULL)

      || (scrubber->highPriorityOnly

          && isRingEmpty(&scrubber->highPrioritySlabs))) {

    scrubber->highPriorityOnly = false;

    finishScrubbing(scrubber);

    return;

  }

  if (finishDraining(&scrubber->adminState)) {

    return;

  }

  unspliceRingNode(&slab->ringNode);

  scrubber->slab = slab;

  VDOCompletion *completion = extentAsCompletion(scrubber->extent);

  prepareCompletion(completion, startScrubbing,

                    handleScrubberError, scrubber->completion.callbackThreadID,

                    scrubber);

  startSlabAction(slab, ADMIN_STATE_SCRUBBING, completion);

}

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

void scrubSlabs(SlabScrubber *scrubber,

                void         *parent,

                VDOAction    *callback,

                VDOAction    *errorHandler)

{

  resumeIfQuiescent(&scrubber->adminState);

  ThreadID threadID = getCallbackThreadID();

  prepareCompletion(&scrubber->completion, callback, errorHandler, threadID,

                    parent);

  if (!hasSlabsToScrub(scrubber)) {

    finishScrubbing(scrubber);

    return;

  }

  scrubNextSlab(scrubber);

}

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

void scrubHighPrioritySlabs(SlabScrubber  *scrubber,

                            bool           scrubAtLeastOne,

                            VDOCompletion *parent,

                            VDOAction     *callback,

                            VDOAction     *errorHandler)

{

  if (scrubAtLeastOne && isRingEmpty(&scrubber->highPrioritySlabs)) {

    Slab *slab = getNextSlab(scrubber);

    if (slab != NULL) {

      registerSlabForScrubbing(scrubber, slab, true);

    }

  }

  scrubber->highPriorityOnly = true;

  scrubSlabs(scrubber, parent, callback, errorHandler);

}

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

void stopScrubbing(SlabScrubber *scrubber, VDOCompletion *parent)

{

  if (isQuiescent(&scrubber->adminState)) {

    completeCompletion(parent);

  } else {

    startDraining(&scrubber->adminState, ADMIN_STATE_SUSPENDING, parent, NULL);

  }

}

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

void resumeScrubbing(SlabScrubber *scrubber, VDOCompletion *parent)

{

  if (!hasSlabsToScrub(scrubber)) {

    completeCompletion(parent);

    return;

  }

  int result = resumeIfQuiescent(&scrubber->adminState);

  if (result != VDO_SUCCESS) {

    finishCompletion(parent, result);

    return;

  }

  scrubNextSlab(scrubber);

  completeCompletion(parent);

}

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

int enqueueCleanSlabWaiter(SlabScrubber *scrubber, Waiter *waiter)

{

  if (isReadOnly(scrubber->readOnlyNotifier)) {

    return VDO_READ_ONLY;

  }

  if (isQuiescent(&scrubber->adminState)) {

    return VDO_NO_SPACE;

  }

  return enqueueWaiter(&scrubber->waiters, waiter);

}

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

void dumpSlabScrubber(const SlabScrubber *scrubber)

{

  logInfo("slabScrubber slabCount %u waiters %zu %s%s",

          getScrubberSlabCount(scrubber),

          countWaiters(&scrubber->waiters),

          getAdminStateName(&scrubber->adminState),

          scrubber->highPriorityOnly ? ", highPriorityOnly " : "");

}