/*
 * 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/superBlock.c#5 $
 */

#include "superBlock.h"

#include "buffer.h"
#include "logger.h"
#include "memoryAlloc.h"
#include "permassert.h"

#include "completion.h"
#include "constants.h"
#include "header.h"
#include "releaseVersions.h"
#include "statusCodes.h"
#include "types.h"
#include "vio.h"

struct superBlock {
  /** The parent for asynchronous load and save operations */
  VDOCompletion        *parent;
  /** The VIO for reading and writing the super block to disk */
  VIO                  *vio;
  /** The buffer for encoding and decoding component data */
  Buffer               *componentBuffer;
  /**
   * A sector-sized buffer wrapping the first sector of encodedSuperBlock, for
   * encoding and decoding the entire super block.
   **/
  Buffer               *blockBuffer;
  /** A 1-block buffer holding the encoded on-disk super block */
  byte                 *encodedSuperBlock;
  /** The release version number loaded from the volume */
  ReleaseVersionNumber  loadedReleaseVersion;
  /** Whether this super block may not be written */
  bool                  unwriteable;
};

enum {
  SUPER_BLOCK_FIXED_SIZE
    = ENCODED_HEADER_SIZE + sizeof(ReleaseVersionNumber) + CHECKSUM_SIZE,
  MAX_COMPONENT_DATA_SIZE = VDO_SECTOR_SIZE - SUPER_BLOCK_FIXED_SIZE,
};

static const Header SUPER_BLOCK_HEADER_12_0 = {
  .id = SUPER_BLOCK,
  .version = {
    .majorVersion = 12,
    .minorVersion = 0,
  },

  // This is the minimum size, if the super block contains no components.
  .size = SUPER_BLOCK_FIXED_SIZE - ENCODED_HEADER_SIZE,
};

/**
 * Allocate a super block. Callers must free the allocated super block even
 * on error.
 *
 * @param layer  The physical layer which holds the super block on disk
 * @param superBlockPtr  A pointer to hold the new super block
 *
 * @return VDO_SUCCESS or an error
 **/
__attribute__((warn_unused_result))
static int allocateSuperBlock(PhysicalLayer *layer, SuperBlock **superBlockPtr)
{
  int result = ALLOCATE(1, SuperBlock, __func__, superBlockPtr);
  if (result != UDS_SUCCESS) {
    return result;
  }

  SuperBlock *superBlock = *superBlockPtr;
  result = makeBuffer(MAX_COMPONENT_DATA_SIZE, &superBlock->componentBuffer);
  if (result != UDS_SUCCESS) {
    return result;
  }

  result = layer->allocateIOBuffer(layer, VDO_BLOCK_SIZE,
                                   "encoded super block",
                                   (char **) &superBlock->encodedSuperBlock);
  if (result != UDS_SUCCESS) {
    return result;
  }

  // Even though the buffer is a full block, to avoid the potential corruption
  // from a torn write, the entire encoding must fit in the first sector.
  result = wrapBuffer(superBlock->encodedSuperBlock, VDO_SECTOR_SIZE, 0,
                      &superBlock->blockBuffer);
  if (result != UDS_SUCCESS) {
    return result;
  }

  if (layer->createMetadataVIO == NULL) {
    return VDO_SUCCESS;
  }

  return createVIO(layer, VIO_TYPE_SUPER_BLOCK, VIO_PRIORITY_METADATA,
                   superBlock, (char *) superBlock->encodedSuperBlock,
                   &superBlock->vio);
}

/**********************************************************************/
int makeSuperBlock(PhysicalLayer *layer, SuperBlock **superBlockPtr)
{
  SuperBlock *superBlock;
  int         result = allocateSuperBlock(layer, &superBlock);
  if (result != VDO_SUCCESS) {
    freeSuperBlock(&superBlock);
    return result;
  }

  // For a new super block, use the current release.
  superBlock->loadedReleaseVersion = CURRENT_RELEASE_VERSION_NUMBER;
  *superBlockPtr = superBlock;
  return VDO_SUCCESS;
}

/**********************************************************************/
void freeSuperBlock(SuperBlock **superBlockPtr)
{
  if (*superBlockPtr == NULL) {
    return;
  }

  SuperBlock *superBlock = *superBlockPtr;
  freeBuffer(&superBlock->blockBuffer);
  freeBuffer(&superBlock->componentBuffer);
  freeVIO(&superBlock->vio);
  FREE(superBlock->encodedSuperBlock);
  FREE(superBlock);
  *superBlockPtr = NULL;
}

/**
 * Encode a super block into its on-disk representation.
 *
 * @param layer       The physical layer which implements the checksum
 * @param superBlock  The super block to encode
 *
 * @return VDO_SUCCESS or an error
 **/
__attribute__((warn_unused_result))
static int encodeSuperBlock(PhysicalLayer *layer, SuperBlock *superBlock)
{
  Buffer *buffer = superBlock->blockBuffer;
  int     result = resetBufferEnd(buffer, 0);
  if (result != VDO_SUCCESS) {
    return result;
  }

  size_t componentDataSize = contentLength(superBlock->componentBuffer);

  // Encode the header.
  Header header = SUPER_BLOCK_HEADER_12_0;
  header.size += componentDataSize;
  result = encodeHeader(&header, buffer);
  if (result != UDS_SUCCESS) {
    return result;
  }

  // Encode the loaded release version.
  result = putUInt32LEIntoBuffer(buffer, superBlock->loadedReleaseVersion);
  if (result != UDS_SUCCESS) {
    return result;
  }

  // Copy the already-encoded component data.
  result = putBytes(buffer, componentDataSize,
                    getBufferContents(superBlock->componentBuffer));
  if (result != UDS_SUCCESS) {
    return result;
  }

  // Compute and encode the checksum.
  CRC32Checksum checksum = layer->updateCRC32(INITIAL_CHECKSUM,
                                              superBlock->encodedSuperBlock,
                                              contentLength(buffer));
  result = putUInt32LEIntoBuffer(buffer, checksum);
  if (result != UDS_SUCCESS) {
    return result;
  }

  return UDS_SUCCESS;
}

/**********************************************************************/
int saveSuperBlock(PhysicalLayer       *layer,
                   SuperBlock          *superBlock,
                   PhysicalBlockNumber  superBlockOffset)
{
  int result = encodeSuperBlock(layer, superBlock);
  if (result != VDO_SUCCESS) {
    return result;
  }

  return layer->writer(layer, superBlockOffset, 1,
                       (char *) superBlock->encodedSuperBlock, NULL);
}

/**
 * Finish the parent of a super block load or save operation. This
 * callback is registered in saveSuperBlockAsync() and loadSuperBlockAsync.
 *
 * @param completion  The super block VIO
 **/
static void finishSuperBlockParent(VDOCompletion *completion)
{
  SuperBlock    *superBlock = completion->parent;
  VDOCompletion *parent     = superBlock->parent;
  superBlock->parent        = NULL;
  finishCompletion(parent, completion->result);
}

/**
 * Log a super block save error. This error handler is registered in
 * saveSuperBlockAsync().
 *
 * @param completion  The super block VIO
 **/
static void handleSaveError(VDOCompletion *completion)
{
  logErrorWithStringError(completion->result, "super block save failed");
  /*
   * Mark the super block as unwritable so that we won't attempt to write it
   * again. This avoids the case where a growth attempt fails writing the
   * super block with the new size, but the subsequent attempt to write out
   * the read-only state succeeds. In this case, writes which happened just
   * before the suspend would not be visible if the VDO is restarted without
   * rebuilding, but, after a read-only rebuild, the effects of those writes
   * would reappear.
   */
  ((SuperBlock *) completion->parent)->unwriteable = true;
  completion->callback(completion);
}

/**********************************************************************/
void saveSuperBlockAsync(SuperBlock          *superBlock,
                         PhysicalBlockNumber  superBlockOffset,
                         VDOCompletion       *parent)
{
  if (superBlock->unwriteable) {
    finishCompletion(parent, VDO_READ_ONLY);
    return;
  }

  if (superBlock->parent != NULL) {
    finishCompletion(parent, VDO_COMPONENT_BUSY);
    return;
  }

  PhysicalLayer *layer = parent->layer;
  int result = encodeSuperBlock(layer, superBlock);
  if (result != VDO_SUCCESS) {
    finishCompletion(parent, result);
    return;
  }

  superBlock->parent                           = parent;
  superBlock->vio->completion.callbackThreadID = parent->callbackThreadID;
  launchWriteMetadataVIOWithFlush(superBlock->vio, superBlockOffset,
                                  finishSuperBlockParent, handleSaveError,
                                  true, true);
}

/**
 * Decode a super block from its on-disk representation.
 *
 * @param layer       The physical layer which implements the checksum
 * @param superBlock  The super block to decode
 *
 * @return VDO_SUCCESS or an error
 **/
__attribute__((warn_unused_result))
static int decodeSuperBlock(PhysicalLayer *layer, SuperBlock *superBlock)
{
  // Reset the block buffer to start decoding the entire first sector.
  Buffer *buffer = superBlock->blockBuffer;
  clearBuffer(buffer);

  // Decode and validate the header.
  Header header;
  int result = decodeHeader(buffer, &header);
  if (result != VDO_SUCCESS) {
    return result;
  }

  result = validateHeader(&SUPER_BLOCK_HEADER_12_0, &header, false, __func__);
  if (result != VDO_SUCCESS) {
    return result;
  }

  if (header.size > contentLength(buffer)) {
    // We can't check release version or checksum until we know the content
    // size, so we have to assume a version mismatch on unexpected values.
    return logErrorWithStringError(VDO_UNSUPPORTED_VERSION,
                                   "super block contents too large: %zu",
                                   header.size);
  }

  // Restrict the buffer to the actual payload bytes that remain.
  result = resetBufferEnd(buffer, uncompactedAmount(buffer) + header.size);
  if (result != VDO_SUCCESS) {
    return result;
  }

  // Decode and store the release version number. It will be checked when the
  // VDO master version is decoded and validated.
  result = getUInt32LEFromBuffer(buffer, &superBlock->loadedReleaseVersion);
  if (result != VDO_SUCCESS) {
    return result;
  }

  // The component data is all the rest, except for the checksum.
  size_t componentDataSize = contentLength(buffer) - sizeof(CRC32Checksum);
  result = putBuffer(superBlock->componentBuffer, buffer, componentDataSize);
  if (result != VDO_SUCCESS) {
    return result;
  }

  // Checksum everything up to but not including the saved checksum itself.
  CRC32Checksum checksum = layer->updateCRC32(INITIAL_CHECKSUM,
                                              superBlock->encodedSuperBlock,
                                              uncompactedAmount(buffer));

  // Decode and verify the saved checksum.
  CRC32Checksum savedChecksum;
  result = getUInt32LEFromBuffer(buffer, &savedChecksum);
  if (result != VDO_SUCCESS) {
    return result;
  }

  result = ASSERT(contentLength(buffer) == 0,
                  "must have decoded entire superblock payload");
  if (result != VDO_SUCCESS) {
    return result;
  }

  return ((checksum != savedChecksum) ? VDO_CHECKSUM_MISMATCH : VDO_SUCCESS);
}

/**********************************************************************/
int loadSuperBlock(PhysicalLayer        *layer,
                   PhysicalBlockNumber   superBlockOffset,
                   SuperBlock          **superBlockPtr)
{
  SuperBlock *superBlock = NULL;
  int         result     = allocateSuperBlock(layer, &superBlock);
  if (result != VDO_SUCCESS) {
    freeSuperBlock(&superBlock);
    return result;
  }

  result = layer->reader(layer, superBlockOffset, 1,
                         (char *) superBlock->encodedSuperBlock, NULL);
  if (result != VDO_SUCCESS) {
    freeSuperBlock(&superBlock);
    return result;
  }

  result = decodeSuperBlock(layer, superBlock);
  if (result != VDO_SUCCESS) {
    freeSuperBlock(&superBlock);
    return result;
  }

  *superBlockPtr = superBlock;
  return result;
}

/**
 * Continue after loading the super block. This callback is registered
 * in loadSuperBlockAsync().
 *
 * @param completion  The super block VIO
 **/
static void finishReadingSuperBlock(VDOCompletion *completion)
{
  SuperBlock    *superBlock = completion->parent;
  VDOCompletion *parent     = superBlock->parent;
  superBlock->parent        = NULL;
  finishCompletion(parent, decodeSuperBlock(completion->layer, superBlock));
}

/**********************************************************************/
void loadSuperBlockAsync(VDOCompletion        *parent,
                         PhysicalBlockNumber   superBlockOffset,
                         SuperBlock          **superBlockPtr)
{
  PhysicalLayer *layer      = parent->layer;
  SuperBlock    *superBlock = NULL;
  int            result     = allocateSuperBlock(layer, &superBlock);
  if (result != VDO_SUCCESS) {
    freeSuperBlock(&superBlock);
    finishCompletion(parent, result);
    return;
  }

  *superBlockPtr = superBlock;

  superBlock->parent                           = parent;
  superBlock->vio->completion.callbackThreadID = parent->callbackThreadID;
  launchReadMetadataVIO(superBlock->vio, superBlockOffset,
                        finishReadingSuperBlock, finishSuperBlockParent);
}

/**********************************************************************/
Buffer *getComponentBuffer(SuperBlock *superBlock)
{
  return superBlock->componentBuffer;
}

/**********************************************************************/
ReleaseVersionNumber getLoadedReleaseVersion(const SuperBlock *superBlock)
{
  return superBlock->loadedReleaseVersion;
}

/**********************************************************************/
size_t getFixedSuperBlockSize(void)
{
  return SUPER_BLOCK_FIXED_SIZE;
}