/*

 * 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/kernel/kernelLayer.c#38 $

 */

#include "kernelLayer.h"

#include <linux/backing-dev.h>

#include <linux/blkdev.h>

#include <linux/crc32.h>

#include <linux/delay.h>

#include <linux/module.h>

#include "logger.h"

#include "memoryAlloc.h"

#include "murmur/MurmurHash3.h"

#include "lz4.h"

#include "releaseVersions.h"

#include "volumeGeometry.h"

#include "statistics.h"

#include "vdo.h"

#include "bio.h"

#include "dataKVIO.h"

#include "dedupeIndex.h"

#include "deviceConfig.h"

#include "deviceRegistry.h"

#include "instanceNumber.h"

#include "ioSubmitter.h"

#include "kvdoFlush.h"

#include "kvio.h"

#include "poolSysfs.h"

#include "statusProcfs.h"

#include "stringUtils.h"

#include "verify.h"

enum {

  DEDUPE_TIMEOUT_REPORT_INTERVAL = 1000,

};

static const KvdoWorkQueueType bioAckQType = {

  .actionTable = {

    { .name = "bio_ack",

      .code = BIO_ACK_Q_ACTION_ACK,

      .priority = 0 },

  },

};

static const KvdoWorkQueueType cpuQType = {

  .actionTable = {

    { .name = "cpu_complete_kvio",

      .code = CPU_Q_ACTION_COMPLETE_KVIO,

      .priority = 0 },

    { .name = "cpu_compress_block",

      .code = CPU_Q_ACTION_COMPRESS_BLOCK,

      .priority = 0 },

    { .name = "cpu_hash_block",

      .code = CPU_Q_ACTION_HASH_BLOCK,

      .priority = 0 },

    { .name = "cpu_event_reporter",

      .code = CPU_Q_ACTION_EVENT_REPORTER,

      .priority = 0 },

  },

};

// 2000 is half the number of entries currently in our page cache,

// to allow for each in-progress operation to update two pages.

int defaultMaxRequestsActive = 2000;

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

static CRC32Checksum kvdoUpdateCRC32(CRC32Checksum  crc,

                                     const byte    *buffer,

                                     size_t         length)

{

  /*

   * The kernel's CRC 32 implementation does not do pre- and post-

   * conditioning, so do it ourselves.

   */

  return crc32(crc ^ 0xffffffff, buffer, length) ^ 0xffffffff;

}

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

static BlockCount kvdoGetBlockCount(PhysicalLayer *header)

{

  return asKernelLayer(header)->deviceConfig->physicalBlocks;

}

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

bool layerIsNamed(KernelLayer *layer, void *context)

{

  struct dm_target *ti = layer->deviceConfig->owningTarget;

  const char *deviceName = dm_device_name(dm_table_get_md(ti->table));

  return (strcmp(deviceName, (const char *) context) == 0);

}

/**

 * Implements LayerFilter.

 **/

static bool layerUsesDevice(KernelLayer *layer, void *context)

{

  DeviceConfig *config = context;

  return (layer->deviceConfig->ownedDevice->bdev->bd_dev

          == config->ownedDevice->bdev->bd_dev);

}

int mapToSystemError(int error)

{

  // 0 is success, negative a system error code

  if (likely(error <= 0)) {

    return error;

  }

  if (error < 1024) {

    // errno macro used without negating - may be a minor bug

    return -error;

  }

  // VDO or UDS error

  char errorName[80], errorMessage[ERRBUF_SIZE];

  switch (sansUnrecoverable(error)) {

  case VDO_NO_SPACE:

    return -ENOSPC;

  case VDO_READ_ONLY:

    return -EIO;

  default:

    logInfo("%s: mapping internal status code %d (%s: %s) to EIO",

            __func__, error,

            stringErrorName(error, errorName, sizeof(errorName)),

            stringError(error, errorMessage, sizeof(errorMessage)));

    return -EIO;

  }

}

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

static void setKernelLayerState(KernelLayer *layer, KernelLayerState newState)

{

  atomicStore32(&layer->state, newState);

}

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

void waitForNoRequestsActive(KernelLayer *layer)

{

  // Do nothing if there are no requests active.  This check is not necessary

  // for correctness but does reduce log message traffic.

  if (limiterIsIdle(&layer->requestLimiter)) {

    return;

  }

  // We have to make sure to flush the packer before waiting. We do this

  // by turning off compression, which also means no new entries coming in

  // while waiting will end up in the packer.

  bool wasCompressing = setKVDOCompressing(&layer->kvdo, false);

  // Now wait for there to be no active requests

  limiterWaitForIdle(&layer->requestLimiter);

  // Reset the compression state after all requests are done

  if (wasCompressing) {

    setKVDOCompressing(&layer->kvdo, true);

  }

}

/**

 * Start processing a new data KVIO based on the supplied bio, but from within

 * a VDO thread context, when we're not allowed to block. Using this path at

 * all suggests a bug or erroneous usage, but we special-case it to avoid a

 * deadlock that can apparently result. Message will be logged to alert the

 * administrator that something has gone wrong, while we attempt to continue

 * processing other requests.

 *

 * If a request permit can be acquired immediately, kvdoLaunchDataKVIOFromBio

 * will be called. (If the bio is a discard operation, a permit from the

 * discard limiter will be requested but the call will be made with or without

 * it.) If the request permit is not available, the bio will be saved on a list

 * to be launched later. Either way, this function will not block, and will

 * take responsibility for processing the bio.

 *

 * @param layer        The kernel layer

 * @param bio          The bio to launch

 * @param arrivalTime  The arrival time of the bio

 *

 * @return  DM_MAPIO_SUBMITTED or a system error code

 **/

static int launchDataKVIOFromVDOThread(KernelLayer *layer,

                                       BIO         *bio,

                                       Jiffies      arrivalTime)

{

  logWarning("kvdoMapBio called from within a VDO thread!");

  /*

   * We're not yet entirely sure what circumstances are causing this situation

   * in [ESC-638], but it does appear to be happening and causing VDO to

   * deadlock.

   *

   * Somehow kvdoMapBio is being called from generic_make_request which is

   * being called from the VDO code to pass a flush on down to the underlying

   * storage system; we've got 2000 requests in progress, so we have to wait

   * for one to complete, but none can complete while the bio thread is blocked

   * from passing more I/O requests down. Near as we can tell, the flush bio

   * should always have gotten updated to point to the storage system, so we

   * shouldn't be calling back into VDO unless something's gotten messed up

   * somewhere.

   *

   * To side-step this case, if the limiter says we're busy *and* we're running

   * on one of VDO's own threads, we'll drop the I/O request in a special queue

   * for processing as soon as KVIOs become free.

   *

   * We don't want to do this in general because it leads to unbounded

   * buffering, arbitrarily high latencies, inability to push back in a way the

   * caller can take advantage of, etc. If someone wants huge amounts of

   * buffering on top of VDO, they're welcome to access it through the kernel

   * page cache or roll their own.

   */

  if (!limiterPoll(&layer->requestLimiter)) {

    addToDeadlockQueue(&layer->deadlockQueue, bio, arrivalTime);

    logWarning("queued an I/O request to avoid deadlock!");

    return DM_MAPIO_SUBMITTED;

  }

  bool hasDiscardPermit

    = (isDiscardBio(bio) && limiterPoll(&layer->discardLimiter));

  int result = kvdoLaunchDataKVIOFromBio(layer, bio, arrivalTime,

                                         hasDiscardPermit);

  // Succeed or fail, kvdoLaunchDataKVIOFromBio owns the permit(s) now.

  if (result != VDO_SUCCESS) {

    return result;

  }

  return DM_MAPIO_SUBMITTED;

}

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

int kvdoMapBio(KernelLayer *layer, BIO *bio)

{

  Jiffies          arrivalTime = jiffies;

  KernelLayerState state       = getKernelLayerState(layer);

  ASSERT_LOG_ONLY(state == LAYER_RUNNING,

                  "kvdoMapBio should not be called while in state %d", state);

  // Count all incoming bios.

  countBios(&layer->biosIn, bio);

  // Handle empty bios.  Empty flush bios are not associated with a VIO.

  if (isFlushBio(bio)) {

    if (ASSERT(getBioSize(bio) == 0, "Flush bio is size 0") != VDO_SUCCESS) {

      // We expect flushes to be of size 0.

      return -EINVAL;

    }

    if (shouldProcessFlush(layer)) {

      launchKVDOFlush(layer, bio);

      return DM_MAPIO_SUBMITTED;

    } else {

      // We're not acknowledging this bio now, but we'll never touch it

      // again, so this is the last chance to account for it.

      countBios(&layer->biosAcknowledged, bio);

      atomic64_inc(&layer->flushOut);

      setBioBlockDevice(bio, getKernelLayerBdev(layer));

      return DM_MAPIO_REMAPPED;

    }

  }

  if (ASSERT(getBioSize(bio) != 0, "Data bio is not size 0") != VDO_SUCCESS) {

    // We expect non-flushes to be non-zero in size.

    return -EINVAL;

  }

  if (isDiscardBio(bio) && isReadBio(bio)) {

    // Read and Discard should never occur together

    return -EIO;

  }

  KvdoWorkQueue *currentWorkQueue = getCurrentWorkQueue();

  if ((currentWorkQueue != NULL)

      && (layer == getWorkQueueOwner(currentWorkQueue))) {

    /*

     * This prohibits sleeping during I/O submission to VDO from its own

     * thread.

     */

    return launchDataKVIOFromVDOThread(layer, bio, arrivalTime);

  }

  bool hasDiscardPermit = false;

  if (isDiscardBio(bio)) {

    limiterWaitForOneFree(&layer->discardLimiter);

    hasDiscardPermit = true;

  }

  limiterWaitForOneFree(&layer->requestLimiter);

  int result = kvdoLaunchDataKVIOFromBio(layer, bio, arrivalTime,

                                         hasDiscardPermit);

  // Succeed or fail, kvdoLaunchDataKVIOFromBio owns the permit(s) now.

  if (result != VDO_SUCCESS) {

    return result;

  }

  return DM_MAPIO_SUBMITTED;

}

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

struct block_device *getKernelLayerBdev(const KernelLayer *layer)

{

  return layer->deviceConfig->ownedDevice->bdev;

}

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

void completeManyRequests(KernelLayer *layer, uint32_t count)

{

  // If we had to buffer some requests to avoid deadlock, release them now.

  while (count > 0) {

    Jiffies arrivalTime = 0;

    BIO *bio = pollDeadlockQueue(&layer->deadlockQueue, &arrivalTime);

    if (likely(bio == NULL)) {

      break;

    }

    bool hasDiscardPermit

      = (isDiscardBio(bio) && limiterPoll(&layer->discardLimiter));

    int result = kvdoLaunchDataKVIOFromBio(layer, bio, arrivalTime,

                                           hasDiscardPermit);

    if (result != VDO_SUCCESS) {

      completeBio(bio, result);

    }

    // Succeed or fail, kvdoLaunchDataKVIOFromBio owns the permit(s) now.

    count--;

  }

  // Notify the limiter, so it can wake any blocked processes.

  if (count > 0) {

    limiterReleaseMany(&layer->requestLimiter, count);

  }

}

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

static void reportEvents(PeriodicEventReporter *reporter)

{

  atomic_set(&reporter->workItemQueued, 0);

  uint64_t newValue = atomic64_read(&reporter->value);

  uint64_t difference = newValue - reporter->lastReportedValue;

  if (difference != 0) {

    logDebug(reporter->format, difference);

    reporter->lastReportedValue = newValue;

  }

}

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

static void reportEventsWork(KvdoWorkItem *item)

{

  PeriodicEventReporter *reporter = container_of(item, PeriodicEventReporter,

                                                 workItem);

  reportEvents(reporter);

}

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

static void initPeriodicEventReporter(PeriodicEventReporter *reporter,

                                      const char            *format,

                                      unsigned long          reportingInterval,

                                      KernelLayer           *layer)

{

  setupWorkItem(&reporter->workItem, reportEventsWork, NULL,

                CPU_Q_ACTION_EVENT_REPORTER);

  reporter->format            = format;

  reporter->reportingInterval = msecs_to_jiffies(reportingInterval);

  reporter->layer             = layer;

}

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

static void addEventCount(PeriodicEventReporter *reporter, unsigned int count)

{

  if (count > 0) {

    atomic64_add(count, &reporter->value);

    int oldWorkItemQueued = atomic_xchg(&reporter->workItemQueued, 1);

    if (oldWorkItemQueued == 0) {

      enqueueWorkQueueDelayed(reporter->layer->cpuQueue,

                              &reporter->workItem,

                              jiffies + reporter->reportingInterval);

    }

  }

}

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

static void stopPeriodicEventReporter(PeriodicEventReporter *reporter)

{

  reportEvents(reporter);

}

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

void kvdoReportDedupeTimeout(KernelLayer *layer, unsigned int expiredCount)

{

  addEventCount(&layer->albireoTimeoutReporter, expiredCount);

}

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

static int kvdoCreateEnqueueable(VDOCompletion *completion)

{

  KvdoEnqueueable *kvdoEnqueueable;

  int result = ALLOCATE(1, KvdoEnqueueable, "kvdoEnqueueable",

                        &kvdoEnqueueable);

  if (result != VDO_SUCCESS) {

    logError("kvdoEnqueueable allocation failure %d", result);

    return result;

  }

  kvdoEnqueueable->enqueueable.completion = completion;

  completion->enqueueable                 = &kvdoEnqueueable->enqueueable;

  return VDO_SUCCESS;

}

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

static void kvdoDestroyEnqueueable(Enqueueable **enqueueablePtr)

{

  Enqueueable *enqueueable = *enqueueablePtr;

  if (enqueueable != NULL) {

    KvdoEnqueueable *kvdoEnqueueable

      = container_of(enqueueable, KvdoEnqueueable, enqueueable);

    FREE(kvdoEnqueueable);

    *enqueueablePtr = NULL;

  }

}

/**

 * Implements BufferAllocator.

 **/

static int kvdoAllocateIOBuffer(PhysicalLayer  *layer __attribute__((unused)),

                                size_t          bytes,

                                const char     *why,

                                char          **bufferPtr)

{

  return ALLOCATE(bytes, char, why, bufferPtr);

}

/**

 * Implements ExtentReader. Exists only for the geometry block; is unset after

 * it is read.

 **/

static int kvdoSynchronousRead(PhysicalLayer       *layer,

                               PhysicalBlockNumber  startBlock,

                               size_t               blockCount,

                               char                *buffer,

                               size_t              *blocksRead)

{

  if (blockCount != 1) {

    return VDO_NOT_IMPLEMENTED;

  }

  KernelLayer *kernelLayer = asKernelLayer(layer);

  BIO *bio;

  int result = createBio(kernelLayer, buffer, &bio;;

  if (result != VDO_SUCCESS) {

    return result;

  }

  setBioBlockDevice(bio, getKernelLayerBdev(kernelLayer));

  setBioSector(bio, blockToSector(kernelLayer, startBlock));

  setBioOperationRead(bio);

  result = submitBioAndWait(bio);

  if (result != 0) {

    logErrorWithStringError(result, "synchronous read failed");

    result = -EIO;

  }

  freeBio(bio, kernelLayer);

  if (result != VDO_SUCCESS) {

    return result;

  }

  if (blocksRead != NULL) {

    *blocksRead = blockCount;

  }

  return VDO_SUCCESS;

}

/**

 * Implements VIODestructor.

 **/

static void kvdoFreeVIO(VIO **vioPtr)

{

  VIO *vio = *vioPtr;

  if (vio == NULL) {

    return;

  }

  BUG_ON(isDataVIO(vio));

  if (isCompressedWriteVIO(vio)) {

    CompressedWriteKVIO *compressedWriteKVIO

      = allocatingVIOAsCompressedWriteKVIO(vioAsAllocatingVIO(vio));

    freeCompressedWriteKVIO(&compressedWriteKVIO);

  } else {

    MetadataKVIO *metadataKVIO = vioAsMetadataKVIO(vio);

    freeMetadataKVIO(&metadataKVIO);

  }

  *vioPtr = NULL;

}

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

static WritePolicy kvdoGetWritePolicy(PhysicalLayer *common)

{

  KernelLayer *layer = asKernelLayer(common);

  return getKVDOWritePolicy(&layer->kvdo);

}

/**

 * Function that is called when a synchronous operation is completed. We let

 * the waiting thread know it can continue.

 *

 * 
Implements OperationComplete.

 *

 * @param common  The kernel layer

 **/

static void kvdoCompleteSyncOperation(PhysicalLayer *common)

{

  KernelLayer *layer = asKernelLayer(common);

  complete(&layer->callbackSync);

}

/**

 * Wait for a synchronous operation to complete.

 *

 * 
Implements OperationWaiter.

 *

 * @param common  The kernel layer

 **/

static void waitForSyncOperation(PhysicalLayer *common)

{

  KernelLayer *layer = asKernelLayer(common);

  // Using the "interruptible" interface means that Linux will not log a

  // message when we wait for more than 120 seconds.

  while (wait_for_completion_interruptible(&layer->callbackSync) != 0) {

    // However, if we get a signal in a user-mode process, we could

    // spin...

    msleep(1);

  }

}

/**

 * Make the bio set for allocating new bios.

 *

 * @param layer  The kernel layer

 *

 * @returns VDO_SUCCESS if bio set created, error code otherwise

 **/

static int makeDedupeBioSet(KernelLayer *layer)

{

#if LINUX_VERSION_CODE >= KERNEL_VERSION(4,18,0)

  int result = ALLOCATE(1, struct bio_set, "bio set", &layer->bioset);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = bioset_init(layer->bioset, 0, 0, BIOSET_NEED_BVECS);

  if (result != 0) {

    return result;

  }

#else

#if LINUX_VERSION_CODE >= KERNEL_VERSION(4,13,0)

  layer->bioset = bioset_create(0, 0, BIOSET_NEED_BVECS);

#else

  layer->bioset = bioset_create(0, 0);

#endif

  if (layer->bioset == NULL) {

    return -ENOMEM;

  }

#endif

  return VDO_SUCCESS;

}

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

int makeKernelLayer(uint64_t        startingSector,

                    unsigned int    instance,

                    DeviceConfig   *config,

                    struct kobject *parentKobject,

                    ThreadConfig  **threadConfigPointer,

                    char          **reason,

                    KernelLayer   **layerPtr)

{

  // VDO-3769 - Set a generic reason so we don't ever return garbage.

  *reason = "Unspecified error";

  KernelLayer *oldLayer = findLayerMatching(layerUsesDevice, config);

  if (oldLayer != NULL) {

    logError("Existing layer named %s already uses device %s",

             oldLayer->deviceConfig->poolName,

	     oldLayer->deviceConfig->parentDeviceName);

    *reason = "Cannot share storage device with already-running VDO";

    return VDO_BAD_CONFIGURATION;

  }

  /*

   * Part 1 - Allocate the kernel layer, its essential parts, and setup up the

   * sysfs node.  These must come first so that the sysfs node works correctly

   * through the freeing of the kernel layer.  After this part you must use

   * freeKernelLayer.

   */

  KernelLayer *layer;

  int result = ALLOCATE(1, KernelLayer, "VDO configuration", &layer);

  if (result != UDS_SUCCESS) {

    *reason = "Cannot allocate VDO configuration";

    return result;

  }

  // Allow the base VDO to allocate buffers and construct or destroy

  // enqueuables as part of its allocation.

  layer->common.allocateIOBuffer   = kvdoAllocateIOBuffer;

  layer->common.createEnqueueable  = kvdoCreateEnqueueable;

  layer->common.destroyEnqueueable = kvdoDestroyEnqueueable;

  result = allocateVDO(&layer->common, &layer->kvdo.vdo);

  if (result != VDO_SUCCESS) {

    *reason = "Cannot allocate VDO";

    FREE(layer);

    return result;

  }

  // After this point, calling kobject_put on kobj will decrement its

  // reference count, and when the count goes to 0 the KernelLayer will

  // be freed.

  kobject_init(&layer->kobj, &kernelLayerKobjType);

  result = kobject_add(&layer->kobj, parentKobject, config->poolName);

  if (result != 0) {

    *reason = "Cannot add sysfs node";

    kobject_put(&layer->kobj);

    return result;

  }

  kobject_init(&layer->wqDirectory, &workQueueDirectoryKobjType);

  result = kobject_add(&layer->wqDirectory, &layer->kobj, "work_queues");

  if (result != 0) {

    *reason = "Cannot add sysfs node";

    kobject_put(&layer->wqDirectory);

    kobject_put(&layer->kobj);

    return result;

  }

  /*

   * Part 2 - Do all the simple initialization.  These initializations have no

   * order dependencies and can be done in any order, but freeKernelLayer()

   * cannot be called until all the simple layer properties are set.

   *

   * The KernelLayer structure starts as all zeros.  Pointer initializations

   * consist of replacing a NULL pointer with a non-NULL pointer, which can be

   * easily undone by freeing all of the non-NULL pointers (using the proper

   * free routine).

   */

  setKernelLayerState(layer, LAYER_SIMPLE_THINGS_INITIALIZED);

  initializeDeadlockQueue(&layer->deadlockQueue);

  int requestLimit = defaultMaxRequestsActive;

  initializeLimiter(&layer->requestLimiter, requestLimit);

  initializeLimiter(&layer->discardLimiter, requestLimit * 3 / 4);

  layer->allocationsAllowed   = true;

  layer->instance             = instance;

  layer->deviceConfig         = config;

  layer->startingSectorOffset = startingSector;

  initializeRing(&layer->deviceConfigRing);

  layer->common.updateCRC32              = kvdoUpdateCRC32;

  layer->common.getBlockCount            = kvdoGetBlockCount;

  layer->common.getWritePolicy           = kvdoGetWritePolicy;

  layer->common.createMetadataVIO        = kvdoCreateMetadataVIO;

  layer->common.createCompressedWriteVIO = kvdoCreateCompressedWriteVIO;

  layer->common.freeVIO                  = kvdoFreeVIO;

  layer->common.completeFlush            = kvdoCompleteFlush;

  layer->common.enqueue                  = kvdoEnqueue;

  layer->common.waitForAdminOperation    = waitForSyncOperation;

  layer->common.completeAdminOperation   = kvdoCompleteSyncOperation;

  layer->common.getCurrentThreadID       = kvdoGetCurrentThreadID;

  layer->common.zeroDataVIO              = kvdoZeroDataVIO;

  layer->common.compareDataVIOs          = kvdoCompareDataVIOs;

  layer->common.copyData                 = kvdoCopyDataVIO;

  layer->common.readData                 = kvdoReadDataVIO;

  layer->common.writeData                = kvdoWriteDataVIO;

  layer->common.writeCompressedBlock     = kvdoWriteCompressedBlock;

  layer->common.readMetadata             = kvdoSubmitMetadataVIO;

  layer->common.writeMetadata            = kvdoSubmitMetadataVIO;

  layer->common.applyPartialWrite        = kvdoModifyWriteDataVIO;

  layer->common.flush                    = kvdoFlushVIO;

  layer->common.hashData                 = kvdoHashDataVIO;

  layer->common.checkForDuplication      = kvdoCheckForDuplication;

  layer->common.verifyDuplication        = kvdoVerifyDuplication;

  layer->common.acknowledgeDataVIO       = kvdoAcknowledgeDataVIO;

  layer->common.compressDataVIO          = kvdoCompressDataVIO;

  layer->common.updateAlbireo            = kvdoUpdateDedupeAdvice;

  spin_lock_init(&layer->flushLock);

  mutex_init(&layer->statsMutex);

  bio_list_init(&layer->waitingFlushes);

  result = addLayerToDeviceRegistry(layer);

  if (result != VDO_SUCCESS) {

    *reason = "Cannot add layer to device registry";

    freeKernelLayer(layer);

    return result;

  }

  snprintf(layer->threadNamePrefix, sizeof(layer->threadNamePrefix), "%s%u",

           THIS_MODULE->name, instance);

  result = makeThreadConfig(config->threadCounts.logicalZones,

                            config->threadCounts.physicalZones,

                            config->threadCounts.hashZones,

                            threadConfigPointer);

  if (result != VDO_SUCCESS) {

    *reason = "Cannot create thread configuration";

    freeKernelLayer(layer);

    return result;

  }

  logInfo("zones: %d logical, %d physical, %d hash; base threads: %d",

          config->threadCounts.logicalZones,

          config->threadCounts.physicalZones,

          config->threadCounts.hashZones,

          (*threadConfigPointer)->baseThreadCount);

  result = makeBatchProcessor(layer, returnDataKVIOBatchToPool, layer,

                              &layer->dataKVIOReleaser);

  if (result != UDS_SUCCESS) {

    *reason = "Cannot allocate KVIO-freeing batch processor";

    freeKernelLayer(layer);

    return result;

  }

  // Spare KVDOFlush, so that we will always have at least one available

  result = makeKVDOFlush(&layer->spareKVDOFlush);

  if (result != UDS_SUCCESS) {

    *reason = "Cannot allocate KVDOFlush record";

    freeKernelLayer(layer);

    return result;

  }

  // BIO pool (needed before the geometry block)

  result = makeDedupeBioSet(layer);

  if (result != VDO_SUCCESS) {

    *reason = "Cannot allocate dedupe bioset";

    freeKernelLayer(layer);

    return result;

  }

  // Read the geometry block so we know how to set up the index. Allow it to

  // do synchronous reads.

  layer->common.reader = kvdoSynchronousRead;

  result = loadVolumeGeometry(&layer->common, &layer->geometry);

  layer->common.reader = NULL;

  if (result != VDO_SUCCESS) {

    *reason = "Could not load geometry block";

    freeKernelLayer(layer);

    return result;

  }

  // Albireo Timeout Reporter

  initPeriodicEventReporter(&layer->albireoTimeoutReporter,

                            "Albireo timeout on %llu requests",

                            DEDUPE_TIMEOUT_REPORT_INTERVAL, layer);

  // Dedupe Index

  BUG_ON(layer->threadNamePrefix[0] == '\0');

  result = makeDedupeIndex(&layer->dedupeIndex, layer);

  if (result != UDS_SUCCESS) {

    *reason = "Cannot initialize dedupe index";

    freeKernelLayer(layer);

    return result;

  }

  // Compression context storage

  result = ALLOCATE(config->threadCounts.cpuThreads, char *, "LZ4 context",

                    &layer->compressionContext);

  if (result != VDO_SUCCESS) {

    *reason = "cannot allocate LZ4 context";

    freeKernelLayer(layer);

    return result;

  }

  for (int i = 0; i < config->threadCounts.cpuThreads; i++) {

    result = ALLOCATE(LZ4_context_size(), char, "LZ4 context",

                      &layer->compressionContext[i]);

    if (result != VDO_SUCCESS) {

      *reason = "cannot allocate LZ4 context";

      freeKernelLayer(layer);

      return result;

    }

  }

  /*

   * Part 3 - Do initializations that depend upon other previous

   * initializations, but have no order dependencies at freeing time.

   * Order dependencies for initialization are identified using BUG_ON.

   */

  setKernelLayerState(layer, LAYER_BUFFER_POOLS_INITIALIZED);

  // Trace pool

  BUG_ON(layer->requestLimiter.limit <= 0);

  result = traceKernelLayerInit(layer);

  if (result != VDO_SUCCESS) {

    *reason = "Cannot initialize trace data";

    freeKernelLayer(layer);

    return result;

  }

  // KVIO and VIO pool

  BUG_ON(layer->deviceConfig->logicalBlockSize <= 0);

  BUG_ON(layer->requestLimiter.limit <= 0);

  BUG_ON(layer->bioset == NULL);

  BUG_ON(layer->deviceConfig->ownedDevice == NULL);

  result = makeDataKVIOBufferPool(layer, layer->requestLimiter.limit,

                                  &layer->dataKVIOPool);

  if (result != VDO_SUCCESS) {

    *reason = "Cannot allocate vio data";

    freeKernelLayer(layer);

    return result;

  }

  /*

   * Part 4 - Do initializations that depend upon other previous

   * initialization, that may have order dependencies at freeing time.

   * These are mostly starting up the workqueue threads.

   */

  // Base-code thread, etc

  result = initializeKVDO(&layer->kvdo, *threadConfigPointer, reason);

  if (result != VDO_SUCCESS) {

    freeKernelLayer(layer);

    return result;

  }

  setKernelLayerState(layer, LAYER_REQUEST_QUEUE_INITIALIZED);

  // Bio queue

  result = makeIOSubmitter(layer->threadNamePrefix,

                           config->threadCounts.bioThreads,

                           config->threadCounts.bioRotationInterval,

                           layer->requestLimiter.limit,

                           layer,

                           &layer->ioSubmitter);

  if (result != VDO_SUCCESS) {

    // If initialization of the bio-queues failed, they are cleaned

    // up already, so just free the rest of the kernel layer.

    freeKernelLayer(layer);

    *reason = "bio submission initialization failed";

    return result;

  }

  setKernelLayerState(layer, LAYER_BIO_DATA_INITIALIZED);

  // Bio ack queue

  if (useBioAckQueue(layer)) {

    result = makeWorkQueue(layer->threadNamePrefix, "ackQ",

                           &layer->wqDirectory, layer, layer, &bioAckQType,

                           config->threadCounts.bioAckThreads,

                           &layer->bioAckQueue);

    if (result != VDO_SUCCESS) {

      *reason = "bio ack queue initialization failed";

      freeKernelLayer(layer);

      return result;

    }

  }

  setKernelLayerState(layer, LAYER_BIO_ACK_QUEUE_INITIALIZED);

  // CPU Queues

  result = makeWorkQueue(layer->threadNamePrefix, "cpuQ", &layer->wqDirectory,

                         layer, NULL, &cpuQType,

                         config->threadCounts.cpuThreads, &layer->cpuQueue);

  if (result != VDO_SUCCESS) {

    *reason = "Albireo CPU queue initialization failed";

    freeKernelLayer(layer);

    return result;

  }

  setKernelLayerState(layer, LAYER_CPU_QUEUE_INITIALIZED);