/*

 * 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/vdo.c#21 $

 */

/*

 * This file contains the main entry points for normal operations on a VDO as

 * well as functions for constructing and destroying VDO instances (in memory).

 */

#include "vdoInternal.h"

#include "buffer.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "adminCompletion.h"

#include "blockMap.h"

#include "extent.h"

#include "hashZone.h"

#include "header.h"

#include "logicalZone.h"

#include "numUtils.h"

#include "packer.h"

#include "physicalZone.h"

#include "readOnlyNotifier.h"

#include "recoveryJournal.h"

#include "releaseVersions.h"

#include "slabDepot.h"

#include "slabSummary.h"

#include "statistics.h"

#include "statusCodes.h"

#include "threadConfig.h"

#include "vdoLayout.h"

#include "vioWrite.h"

#include "volumeGeometry.h"

/**

 * The master version of the on-disk format of a VDO. This should be

 * incremented any time the on-disk representation of any VDO structure

 * changes. Changes which require only online upgrade steps should increment

 * the minor version. Changes which require an offline upgrade or which can not

 * be upgraded to at all should increment the major version and set the minor

 * version to 0.

 **/

static const VersionNumber VDO_MASTER_VERSION_67_0 = {

  .majorVersion = 67,

  .minorVersion =  0,

};

/**

 * The current version for the data encoded in the super block. This must

 * be changed any time there is a change to encoding of the component data

 * of any VDO component.

 **/

static const VersionNumber VDO_COMPONENT_DATA_41_0 = {

  .majorVersion = 41,

  .minorVersion =  0,

};

/**

 * This is the structure that captures the VDO fields saved as a SuperBlock

 * component.

 **/

typedef struct {

  VDOState  state;

  uint64_t  completeRecoveries;

  uint64_t  readOnlyRecoveries;

  VDOConfig config;

  Nonce     nonce;

} __attribute__((packed)) VDOComponent41_0;

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

int allocateVDO(PhysicalLayer *layer, VDO **vdoPtr)

{

  int result = registerStatusCodes();

  if (result != VDO_SUCCESS) {

    return result;

  }

  VDO *vdo;

  result = ALLOCATE(1, VDO, __func__, &vdo;;

  if (result != UDS_SUCCESS) {

    return result;

  }

  vdo->layer = layer;

  if (layer->createEnqueueable != NULL) {

    result = initializeAdminCompletion(vdo, &vdo->adminCompletion);

    if (result != VDO_SUCCESS) {

      freeVDO(&vdo;;

      return result;

    }

  }

  *vdoPtr = vdo;

  return VDO_SUCCESS;

}

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

int makeVDO(PhysicalLayer *layer, VDO **vdoPtr)

{

  VDO *vdo;

  int result = allocateVDO(layer, &vdo;;

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = makeZeroThreadConfig(&vdo->loadConfig.threadConfig);

  if (result != VDO_SUCCESS) {

    freeVDO(&vdo;;

    return result;

  }

  *vdoPtr = vdo;

  return VDO_SUCCESS;

}

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

void destroyVDO(VDO *vdo)

{

  freeFlusher(&vdo->flusher);

  freePacker(&vdo->packer);

  freeRecoveryJournal(&vdo->recoveryJournal);

  freeSlabDepot(&vdo->depot);

  freeVDOLayout(&vdo->layout);

  freeSuperBlock(&vdo->superBlock);

  freeBlockMap(&vdo->blockMap);

  const ThreadConfig *threadConfig = getThreadConfig(vdo);

  if (vdo->hashZones != NULL) {

    for (ZoneCount zone = 0; zone < threadConfig->hashZoneCount; zone++) {

      freeHashZone(&vdo->hashZones[zone]);

    }

  }

  FREE(vdo->hashZones);

  vdo->hashZones = NULL;

  freeLogicalZones(&vdo->logicalZones);

  if (vdo->physicalZones != NULL) {

    for (ZoneCount zone = 0; zone < threadConfig->physicalZoneCount; zone++) {

      freePhysicalZone(&vdo->physicalZones[zone]);

    }

  }

  FREE(vdo->physicalZones);

  vdo->physicalZones = NULL;

  uninitializeAdminCompletion(&vdo->adminCompletion);

  freeReadOnlyNotifier(&vdo->readOnlyNotifier);

  freeThreadConfig(&vdo->loadConfig.threadConfig);

}

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

void freeVDO(VDO **vdoPtr)

{

  if (*vdoPtr == NULL) {

    return;

  }

  destroyVDO(*vdoPtr);

  FREE(*vdoPtr);

  *vdoPtr = NULL;

}

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

size_t getComponentDataSize(VDO *vdo)

{

  return (sizeof(VersionNumber)

          + sizeof(VersionNumber)

          + sizeof(VDOComponent41_0)

          + getVDOLayoutEncodedSize(vdo->layout)

          + getRecoveryJournalEncodedSize()

          + getSlabDepotEncodedSize()

          + getBlockMapEncodedSize());

}

/**

 * Encode the VDO master version.

 *

 * @param buffer  The buffer in which to encode the version

 *

 * @return VDO_SUCCESS or an error

 **/

__attribute__((warn_unused_result))

static int encodeMasterVersion(Buffer *buffer)

{

  return encodeVersionNumber(VDO_MASTER_VERSION_67_0, buffer);

}

/**

 * Encode a VDOConfig structure into a buffer.

 *

 * @param config  The config structure to encode

 * @param buffer  A buffer positioned at the start of the encoding

 *

 * @return VDO_SUCCESS or an error

 **/

__attribute__((warn_unused_result))

static int encodeVDOConfig(const VDOConfig *config, Buffer *buffer)

{

  int result = putUInt64LEIntoBuffer(buffer, config->logicalBlocks);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = putUInt64LEIntoBuffer(buffer, config->physicalBlocks);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = putUInt64LEIntoBuffer(buffer, config->slabSize);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = putUInt64LEIntoBuffer(buffer, config->recoveryJournalSize);

  if (result != VDO_SUCCESS) {

    return result;

  }

  return putUInt64LEIntoBuffer(buffer, config->slabJournalBlocks);

}

/**

 * Encode the component data for the VDO itself.

 *

 * @param vdo     The vdo to encode

 * @param buffer  The buffer in which to encode the VDO

 *

 * @return VDO_SUCCESS or an error

 **/

__attribute__((warn_unused_result))

static int encodeVDOComponent(const VDO *vdo, Buffer *buffer)

{

  int result = encodeVersionNumber(VDO_COMPONENT_DATA_41_0, buffer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  size_t initialLength = contentLength(buffer);

  result = putUInt32LEIntoBuffer(buffer, vdo->state);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = putUInt64LEIntoBuffer(buffer, vdo->completeRecoveries);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = putUInt64LEIntoBuffer(buffer, vdo->readOnlyRecoveries);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = encodeVDOConfig(&vdo->config, buffer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = putUInt64LEIntoBuffer(buffer, vdo->nonce);

  if (result != VDO_SUCCESS) {

    return result;

  }

  size_t encodedSize = contentLength(buffer) - initialLength;

  return ASSERT(encodedSize == sizeof(VDOComponent41_0),

                "encoded VDO component size must match structure size");

}

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

static int encodeVDO(VDO *vdo)

{

  Buffer *buffer = getComponentBuffer(vdo->superBlock);

  int result = resetBufferEnd(buffer, 0);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = encodeMasterVersion(buffer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = encodeVDOComponent(vdo, buffer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = encodeVDOLayout(vdo->layout, buffer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = encodeRecoveryJournal(vdo->recoveryJournal, buffer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = encodeSlabDepot(vdo->depot, buffer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = encodeBlockMap(vdo->blockMap, buffer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  ASSERT_LOG_ONLY((contentLength(buffer) == getComponentDataSize(vdo)),

                  "All super block component data was encoded");

  return VDO_SUCCESS;

}

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

int saveVDOComponents(VDO *vdo)

{

  int result = encodeVDO(vdo);

  if (result != VDO_SUCCESS) {

    return result;

  }

  return saveSuperBlock(vdo->layer, vdo->superBlock, getFirstBlockOffset(vdo));

}

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

void saveVDOComponentsAsync(VDO *vdo, VDOCompletion *parent)

{

  int result = encodeVDO(vdo);

  if (result != VDO_SUCCESS) {

    finishCompletion(parent, result);

    return;

  }

  saveSuperBlockAsync(vdo->superBlock, getFirstBlockOffset(vdo), parent);

}

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

int saveReconfiguredVDO(VDO *vdo)

{

  Buffer *buffer         = getComponentBuffer(vdo->superBlock);

  size_t  componentsSize = contentLength(buffer);

  byte *components;

  int   result = copyBytes(buffer, componentsSize, &components);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = resetBufferEnd(buffer, 0);

  if (result != VDO_SUCCESS) {

    FREE(components);

    return result;

  }

  result = encodeMasterVersion(buffer);

  if (result != VDO_SUCCESS) {

    FREE(components);

    return result;

  }

  result = encodeVDOComponent(vdo, buffer);

  if (result != VDO_SUCCESS) {

    FREE(components);

    return result;

  }

  result = putBytes(buffer, componentsSize, components);

  FREE(components);

  if (result != VDO_SUCCESS) {

    return result;

  }

  return saveSuperBlock(vdo->layer, vdo->superBlock, getFirstBlockOffset(vdo));

}

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

int decodeVDOVersion(VDO *vdo)

{

  return decodeVersionNumber(getComponentBuffer(vdo->superBlock),

                             &vdo->loadVersion);

}

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

int validateVDOVersion(VDO *vdo)

{

  int result = decodeVDOVersion(vdo);

  if (result != VDO_SUCCESS) {

    return result;

  }

  ReleaseVersionNumber loadedReleaseVersion

    = getLoadedReleaseVersion(vdo->superBlock);

  if (vdo->loadConfig.releaseVersion != loadedReleaseVersion) {

    return logErrorWithStringError(VDO_UNSUPPORTED_VERSION,

                                   "Geometry release version %" PRIu32 " does "

                                   "not match super block release version %"

                                   PRIu32,

                                   vdo->loadConfig.releaseVersion,

                                   loadedReleaseVersion);

  }

  return validateVersion(VDO_MASTER_VERSION_67_0, vdo->loadVersion, "master");

}

/**

 * Decode a VDOConfig structure from a buffer.

 *

 * @param buffer  A buffer positioned at the start of the encoding

 * @param config  The config structure to receive the decoded values

 *

 * @return UDS_SUCCESS or an error code

 **/

__attribute__((warn_unused_result))

static int decodeVDOConfig(Buffer *buffer, VDOConfig *config)

{

  BlockCount logicalBlocks;

  int result = getUInt64LEFromBuffer(buffer, &logicalBlocks);

  if (result != VDO_SUCCESS) {

    return result;

  }

  BlockCount physicalBlocks;

  result = getUInt64LEFromBuffer(buffer, &physicalBlocks);

  if (result != VDO_SUCCESS) {

    return result;

  }

  BlockCount slabSize;

  result = getUInt64LEFromBuffer(buffer, &slabSize);

  if (result != VDO_SUCCESS) {

    return result;

  }

  BlockCount recoveryJournalSize;

  result = getUInt64LEFromBuffer(buffer, &recoveryJournalSize);

  if (result != VDO_SUCCESS) {

    return result;

  }

  BlockCount slabJournalBlocks;

  result = getUInt64LEFromBuffer(buffer, &slabJournalBlocks);

  if (result != VDO_SUCCESS) {

    return result;

  }

  *config = (VDOConfig) {

    .logicalBlocks       = logicalBlocks,

    .physicalBlocks      = physicalBlocks,

    .slabSize            = slabSize,

    .recoveryJournalSize = recoveryJournalSize,

    .slabJournalBlocks   = slabJournalBlocks,

  };

  return VDO_SUCCESS;

}

/**

 * Decode the version 41.0 component state for the VDO itself from a buffer.

 *

 * @param buffer  A buffer positioned at the start of the encoding

 * @param state   The state structure to receive the decoded values

 *

 * @return VDO_SUCCESS or an error

 **/

__attribute__((warn_unused_result))

  static int decodeVDOComponent_41_0(Buffer *buffer, VDOComponent41_0 *state)

{

  size_t initialLength = contentLength(buffer);

  VDOState vdoState;

  int result = getUInt32LEFromBuffer(buffer, &vdoState);

  if (result != VDO_SUCCESS) {

    return result;

  }

  uint64_t completeRecoveries;

  result = getUInt64LEFromBuffer(buffer, &completeRecoveries);

  if (result != VDO_SUCCESS) {

    return result;

  }

  uint64_t readOnlyRecoveries;

  result = getUInt64LEFromBuffer(buffer, &readOnlyRecoveries);

  if (result != VDO_SUCCESS) {

    return result;

  }

  VDOConfig config;

  result = decodeVDOConfig(buffer, &config);

  if (result != VDO_SUCCESS) {

    return result;

  }

  Nonce nonce;

  result = getUInt64LEFromBuffer(buffer, &nonce);

  if (result != VDO_SUCCESS) {

    return result;

  }

  *state = (VDOComponent41_0) {

    .state              = vdoState,

    .completeRecoveries = completeRecoveries,

    .readOnlyRecoveries = readOnlyRecoveries,

    .config             = config,

    .nonce              = nonce,

  };

  size_t decodedSize = initialLength - contentLength(buffer);

  return ASSERT(decodedSize == sizeof(VDOComponent41_0),

                "decoded VDO component size must match structure size");

}

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

int decodeVDOComponent(VDO *vdo)

{

  Buffer *buffer = getComponentBuffer(vdo->superBlock);

  VersionNumber version;

  int result = decodeVersionNumber(buffer, &version);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = validateVersion(version, VDO_COMPONENT_DATA_41_0,

                           "VDO component data");

  if (result != VDO_SUCCESS) {

    return result;

  }

  VDOComponent41_0 component;

  result = decodeVDOComponent_41_0(buffer, &component);

  if (result != VDO_SUCCESS) {

    return result;

  }

  // Copy the decoded component into the VDO structure.

  vdo->state              = component.state;

  vdo->loadState          = component.state;

  vdo->completeRecoveries = component.completeRecoveries;

  vdo->readOnlyRecoveries = component.readOnlyRecoveries;

  vdo->config             = component.config;

  vdo->nonce              = component.nonce;

  return VDO_SUCCESS;

}

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

int validateVDOConfig(const VDOConfig *config,

                      BlockCount       blockCount,

                      bool             requireLogical)

{

  int result = ASSERT(config->slabSize > 0, "slab size unspecified");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT(isPowerOfTwo(config->slabSize),

                  "slab size must be a power of two");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT(config->slabSize <= (1 << MAX_SLAB_BITS),

                  "slab size must be less than or equal to 2^%d",

                  MAX_SLAB_BITS);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = ASSERT(config->slabJournalBlocks >= MINIMUM_SLAB_JOURNAL_BLOCKS,

                  "slab journal size meets minimum size");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT(config->slabJournalBlocks <= config->slabSize,

                  "slab journal size is within expected bound");

  if (result != UDS_SUCCESS) {

    return result;

  }

  SlabConfig slabConfig;

  result = configureSlab(config->slabSize, config->slabJournalBlocks,

                         &slabConfig);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = ASSERT((slabConfig.dataBlocks >= 1),

                  "slab must be able to hold at least one block");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT(config->physicalBlocks > 0, "physical blocks unspecified");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT(config->physicalBlocks <= MAXIMUM_PHYSICAL_BLOCKS,

                  "physical block count %llu exceeds maximum %llu",

                  config->physicalBlocks, MAXIMUM_PHYSICAL_BLOCKS);

  if (result != UDS_SUCCESS) {

    return VDO_OUT_OF_RANGE;

  }

  // This can't check equality because FileLayer et al can only known about

  // the storage size, which may not match the super block size.

  if (blockCount < config->physicalBlocks) {

    logError("A physical size of %llu blocks was specified,"

             " but that is smaller than the %llu blocks"

             " configured in the VDO super block",

             blockCount, config->physicalBlocks);

    return VDO_PARAMETER_MISMATCH;

  }

  result = ASSERT(!requireLogical || (config->logicalBlocks > 0),

                  "logical blocks unspecified");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT(config->logicalBlocks <= MAXIMUM_LOGICAL_BLOCKS,

                  "logical blocks too large");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT(config->recoveryJournalSize > 0,

                  "recovery journal size unspecified");

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = ASSERT(isPowerOfTwo(config->recoveryJournalSize),

                  "recovery journal size must be a power of two");

  if (result != UDS_SUCCESS) {

    return result;

  }

  return result;

}

/**

 * Notify a VDO that it is going read-only. This will save the read-only state

 * to the super block.

 *

 * 
Implements ReadOnlyNotification.

 *

 * @param listener  The VDO

 * @param parent    The completion to notify in order to acknowledge the

 *                  notification

 **/

static void notifyVDOOfReadOnlyMode(void *listener, VDOCompletion *parent)

{

  VDO *vdo = listener;

  if (inReadOnlyMode(vdo)) {

    completeCompletion(parent);

  }

  vdo->state = VDO_READ_ONLY_MODE;

  saveVDOComponentsAsync(vdo, parent);

}

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

int enableReadOnlyEntry(VDO *vdo)

{

  return registerReadOnlyListener(vdo->readOnlyNotifier, vdo,

                                  notifyVDOOfReadOnlyMode,

                                  getAdminThread(getThreadConfig(vdo)));

}

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

bool inReadOnlyMode(const VDO *vdo)

{

  return (vdo->state == VDO_READ_ONLY_MODE);

}

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

bool isClean(const VDO *vdo)

{

  return ((vdo->state == VDO_CLEAN) || (vdo->state == VDO_NEW));

}

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

bool wasClean(const VDO *vdo)

{

  return ((vdo->loadState == VDO_CLEAN) || (vdo->loadState == VDO_NEW));

}

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

bool wasNew(const VDO *vdo)

{

  return (vdo->loadState == VDO_NEW);

}

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

bool requiresReadOnlyRebuild(const VDO *vdo)

{

  return ((vdo->loadState == VDO_FORCE_REBUILD)

          || (vdo->loadState == VDO_REBUILD_FOR_UPGRADE));

}

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

bool requiresRebuild(const VDO *vdo)

{

  return ((vdo->state == VDO_DIRTY)

          || (vdo->state == VDO_FORCE_REBUILD)

          || (vdo->state == VDO_REPLAYING)

          || (vdo->state == VDO_REBUILD_FOR_UPGRADE));

}

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

bool requiresRecovery(const VDO *vdo)

{

  return ((vdo->loadState == VDO_DIRTY) || (vdo->loadState == VDO_REPLAYING)

          || (vdo->loadState == VDO_RECOVERING));

}

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

bool isReplaying(const VDO *vdo)

{

  return (vdo->state == VDO_REPLAYING);

}

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

bool inRecoveryMode(const VDO *vdo)

{

  return (vdo->state == VDO_RECOVERING);

}

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

void enterRecoveryMode(VDO *vdo)

{

  assertOnAdminThread(vdo, __func__);

  if (inReadOnlyMode(vdo)) {

    return;

  }

  logInfo("Entering recovery mode");

  vdo->state = VDO_RECOVERING;

}

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

void leaveRecoveryMode(VDO *vdo)

{

  assertOnAdminThread(vdo, __func__);

  /*

   * Since scrubbing can be stopped by vdoClose during recovery mode,

   * do not change the VDO state if there are outstanding unrecovered slabs.

   */

  if (inReadOnlyMode(vdo)) {

    return;

  }

  ASSERT_LOG_ONLY(inRecoveryMode(vdo), "VDO is in recovery mode");

  logInfo("Exiting recovery mode");

  vdo->state = VDO_DIRTY;

}

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

void makeVDOReadOnly(VDO *vdo, int errorCode)

{

  enterReadOnlyMode(vdo->readOnlyNotifier, errorCode);

}

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

bool setVDOCompressing(VDO *vdo, bool enableCompression)

{

  bool stateChanged = compareAndSwapBool(&vdo->compressing, !enableCompression,

                                         enableCompression);

  if (stateChanged && !enableCompression) {

    // Flushing the packer is asynchronous, but we don't care when it

    // finishes.

    flushPacker(vdo->packer);

  }

  logInfo("compression is %s", (enableCompression ? "enabled" : "disabled"));

  return (stateChanged ? !enableCompression : enableCompression);

}

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

bool getVDOCompressing(VDO *vdo)

{

  return atomicLoadBool(&vdo->compressing);

}

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

static size_t getBlockMapCacheSize(const VDO *vdo)

{

  return ((size_t) vdo->loadConfig.cacheSize) * VDO_BLOCK_SIZE;

}

/**

 * Tally the hash lock statistics from all the hash zones.

 *

 * @param vdo  The vdo to query

 *

 * @return The sum of the hash lock statistics from all hash zones

 **/

static HashLockStatistics getHashLockStatistics(const VDO *vdo)

{

  HashLockStatistics totals;

  memset(&totals, 0, sizeof(totals));

  const ThreadConfig *threadConfig = getThreadConfig(vdo);

  for (ZoneCount zone = 0; zone < threadConfig->hashZoneCount; zone++) {

    HashLockStatistics stats  = getHashZoneStatistics(vdo->hashZones[zone]);

    totals.dedupeAdviceValid        += stats.dedupeAdviceValid;

    totals.dedupeAdviceStale        += stats.dedupeAdviceStale;

    totals.concurrentDataMatches    += stats.concurrentDataMatches;

    totals.concurrentHashCollisions += stats.concurrentHashCollisions;

  }

  return totals;

}

/**

 * Get the current error statistics from VDO.

 *

 * @param vdo  The vdo to query

 *

 * @return a copy of the current VDO error counters

 **/

static ErrorStatistics getVDOErrorStatistics(const VDO *vdo)

{

  /*

   * The error counts can be incremented from arbitrary threads and so must be

   * incremented atomically, but they are just statistics with no semantics

   * that could rely on memory order, so unfenced reads are sufficient.

   */

  const AtomicErrorStatistics *atoms = &vdo->errorStats;

  return (ErrorStatistics) {

    .invalidAdvicePBNCount = relaxedLoad64(&atoms->invalidAdvicePBNCount),

    .noSpaceErrorCount     = relaxedLoad64(&atoms->noSpaceErrorCount),

    .readOnlyErrorCount    = relaxedLoad64(&atoms->readOnlyErrorCount),

  };

}

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

static const char *describeWritePolicy(WritePolicy policy)

{

  switch (policy) {

  case WRITE_POLICY_ASYNC:

    return "async";

  case WRITE_POLICY_ASYNC_UNSAFE:

    return "async-unsafe";

  case WRITE_POLICY_SYNC:

    return "sync";

  default:

    return "unknown";

  }

}

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

void getVDOStatistics(const VDO *vdo, VDOStatistics *stats)

{

  // These are immutable properties of the VDO object, so it is safe to

  // query them from any thread.

  RecoveryJournal *journal  = vdo->recoveryJournal;