/*

 * 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/dataKVIO.c#18 $

 */

#include "dataKVIO.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "murmur/MurmurHash3.h"

#include "dataVIO.h"

#include "compressedBlock.h"

#include "hashLock.h"

#include "lz4.h"

#include "bio.h"

#include "dedupeIndex.h"

#include "kvdoFlush.h"

#include "kvio.h"

#include "ioSubmitter.h"

#include "vdoCommon.h"

#include "verify.h"

static void dumpPooledDataKVIO(void *poolData, void *data);

enum {

  WRITE_PROTECT_FREE_POOL = 0,

  WP_DATA_KVIO_SIZE       = (sizeof(DataKVIO) + PAGE_SIZE - 1

                             - ((sizeof(DataKVIO) + PAGE_SIZE - 1)

                                % PAGE_SIZE))

};

/**

 * Alter the write-access permission to a page of memory, so that

 * objects in the free pool may no longer be modified.

 *

 * To do: Deny read access as well.

 *

 * @param address    The starting address to protect, which must be on a

 *                   page boundary

 * @param byteCount  The number of bytes to protect, which must be a multiple

 *                   of the page size

 * @param mode       The write protection mode (true means read-only)

 **/

static __always_inline void

setWriteProtect(void   *address,

                size_t  byteCount,

                bool    mode __attribute__((unused)))

{

  BUG_ON((((long) address) % PAGE_SIZE) != 0);

  BUG_ON((byteCount % PAGE_SIZE) != 0);

  BUG(); // only works in internal code, sorry

}

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

static void maybeLogDataKVIOTrace(DataKVIO *dataKVIO)

{

  if (dataKVIO->kvio.layer->traceLogging) {

    logKvioTrace(&dataKVIO->kvio);

  }

}

/**

 * First tracing hook for VIO completion.

 *

 * If the SystemTap script vdotrace.stp is in use, it does stage 1 of

 * its processing here. We must not call addTraceRecord between the

 * two tap functions.

 *

 * @param dataKVIO  The VIO we're finishing up

 **/

static void kvioCompletionTap1(DataKVIO *dataKVIO)

{

  /*

   * Ensure that dataKVIO doesn't get optimized out, even under inline

   * expansion. Also, make sure the compiler has to emit debug info

   * for baseTraceLocation, which some of our SystemTap scripts will

   * use here.

   *

   * First, make it look as though all memory could be clobbered; then

   * require that a value be read into a register. That'll force at

   * least one instruction to exist (so SystemTap can hook in) where

   * dataKVIO is live. We use a field that the caller would've

   * accessed recently anyway, so it may be cached.

   */

  barrier();

  __asm__ __volatile__(""

                       :

                       : "g" (dataKVIO), "g" (baseTraceLocation),

                         "r" (dataKVIO->kvio.layer));

}

/**

 * Second tracing hook for VIO completion.

 *

 * The SystemTap script vdotrace.stp splits its VIO-completion work

 * into two stages, to reduce lock contention for script variables.

 * Hence, it needs two hooks in the code.

 *

 * @param dataKVIO  The VIO we're finishing up

 **/

static void kvioCompletionTap2(DataKVIO *dataKVIO)

{

  // Hack to ensure variable doesn't get optimized out.

  barrier();

  __asm__ __volatile__("" : : "g" (dataKVIO), "r" (dataKVIO->kvio.layer));

}

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

static void kvdoAcknowledgeDataKVIO(DataKVIO *dataKVIO)

{

  KernelLayer       *layer             = dataKVIO->kvio.layer;

  ExternalIORequest *externalIORequest = &dataKVIO->externalIORequest;

  BIO               *bio               = externalIORequest->bio;

  if (bio == NULL) {

    return;

  }

  externalIORequest->bio = NULL;

  int error

    = mapToSystemError(dataVIOAsCompletion(&dataKVIO->dataVIO)->result);

  bio->bi_end_io  = externalIORequest->endIO;

  bio->bi_private = externalIORequest->private;

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

  bio->bi_opf     = externalIORequest->rw;

#else

  bio->bi_rw      = externalIORequest->rw;

#endif

  countBios(&layer->biosAcknowledged, bio);

  if (dataKVIO->isPartial) {

    countBios(&layer->biosAcknowledgedPartial, bio);

  }

  dataKVIOAddTraceRecord(dataKVIO, THIS_LOCATION(NULL));

  completeBio(bio, error);

}

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

static noinline void cleanDataKVIO(DataKVIO *dataKVIO, FreeBufferPointers *fbp)

{

  dataKVIOAddTraceRecord(dataKVIO, THIS_LOCATION(NULL));

  kvdoAcknowledgeDataKVIO(dataKVIO);

  KVIO *kvio = dataKVIOAsKVIO(dataKVIO);

  kvio->bio  = NULL;

  if (unlikely(kvio->vio->trace != NULL)) {

    maybeLogDataKVIOTrace(dataKVIO);

    kvioCompletionTap1(dataKVIO);

    kvioCompletionTap2(dataKVIO);

    freeTraceToPool(kvio->layer, kvio->vio->trace);

  }

  addFreeBufferPointer(fbp, dataKVIO);

}

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

void returnDataKVIOBatchToPool(BatchProcessor *batch, void *closure)

{

  KernelLayer *layer = closure;

  uint32_t     count = 0;

  ASSERT_LOG_ONLY(batch != NULL, "batch not null");

  ASSERT_LOG_ONLY(layer != NULL, "layer not null");

  FreeBufferPointers fbp;

  initFreeBufferPointers(&fbp, layer->dataKVIOPool);

  KvdoWorkItem *item;

  while ((item = nextBatchItem(batch)) != NULL) {

    cleanDataKVIO(workItemAsDataKVIO(item), &fbp);

    condReschedBatchProcessor(batch);

    count++;

  }

  if (fbp.index > 0) {

    freeBufferPointers(&fbp);

  }

  completeManyRequests(layer, count);

}

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

static void kvdoAcknowledgeThenCompleteDataKVIO(KvdoWorkItem *item)

{

  DataKVIO *dataKVIO = workItemAsDataKVIO(item);

  kvdoAcknowledgeDataKVIO(dataKVIO);

  addToBatchProcessor(dataKVIO->kvio.layer->dataKVIOReleaser, item);

}

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

void kvdoCompleteDataKVIO(VDOCompletion *completion)

{

  DataKVIO *dataKVIO = dataVIOAsDataKVIO(asDataVIO(completion));

  dataKVIOAddTraceRecord(dataKVIO, THIS_LOCATION(NULL));

  KernelLayer *layer = getLayerFromDataKVIO(dataKVIO);

  if (useBioAckQueue(layer) && USE_BIO_ACK_QUEUE_FOR_READ

      && (dataKVIO->externalIORequest.bio != NULL)) {

    launchDataKVIOOnBIOAckQueue(dataKVIO, kvdoAcknowledgeThenCompleteDataKVIO,

                                NULL, BIO_ACK_Q_ACTION_ACK);

  } else {

    addToBatchProcessor(layer->dataKVIOReleaser,

                        workItemFromDataKVIO(dataKVIO));

  }

}

/**

 * Copy the uncompressed data from a compressed block read into the user

 * bio which requested the read.

 *

 * @param workItem  The DataKVIO which requested the read

 **/

static void copyReadBlockData(KvdoWorkItem *workItem)

{

  DataKVIO *dataKVIO = workItemAsDataKVIO(workItem);

  // For a read-modify-write, copy the data into the dataBlock buffer so it

  // will be set up for the write phase.

  if (isReadModifyWriteVIO(dataKVIO->kvio.vio)) {

    bioCopyDataOut(getBIOFromDataKVIO(dataKVIO), dataKVIO->readBlock.data);

    kvdoEnqueueDataVIOCallback(dataKVIO);

    return;

  }

  // For a partial read, the callback will copy the requested data from the

  // read block.

  if (dataKVIO->isPartial) {

    kvdoEnqueueDataVIOCallback(dataKVIO);

    return;

  }

  // For a full block read, copy the data to the bio and acknowledge.

  bioCopyDataOut(getBIOFromDataKVIO(dataKVIO), dataKVIO->readBlock.data);

  kvdoAcknowledgeDataVIO(&dataKVIO->dataVIO);

}

/**

 * Finish reading data for a compressed block.

 *

 * @param dataKVIO  The DataKVIO which requested the read

 **/

static void readDataKVIOReadBlockCallback(DataKVIO *dataKVIO)

{

  if (dataKVIO->readBlock.status != VDO_SUCCESS) {

    setCompletionResult(dataVIOAsCompletion(&dataKVIO->dataVIO),

                        dataKVIO->readBlock.status);

    kvdoEnqueueDataVIOCallback(dataKVIO);

    return;

  }

  launchDataKVIOOnCPUQueue(dataKVIO, copyReadBlockData, NULL,

                           CPU_Q_ACTION_COMPRESS_BLOCK);

}

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

/**

 * Complete and reset a bio that was supplied by the user and then used for a

 * read (so that we can complete it with the user's callback).

 *

 * @param bio   The bio to complete

 **/

static void resetUserBio(BIO *bio)

#else

/**

 * Complete and reset a bio that was supplied by the user and then used for a

 * read (so that we can complete it with the user's callback).

 *

 * @param bio   The bio to complete

 * @param error Possible error from underlying block device

 **/

static void resetUserBio(BIO *bio, int error)

#endif

{

#if ((LINUX_VERSION_CODE >= KERNEL_VERSION(3,14,0)) \

     && (LINUX_VERSION_CODE < KERNEL_VERSION(4,2,0)))

  // This is a user bio, and the device just called bio_endio() on it, so

  // we need to re-increment bi_remaining so we too can call bio_endio().

  atomic_inc(&bio->bi_remaining);

#endif

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

  completeAsyncBio(bio);

#else

  completeAsyncBio(bio, error);

#endif

}

/**

 * Uncompress the data that's just been read and then call back the requesting

 * DataKVIO.

 *

 * @param workItem  The DataKVIO requesting the data

 **/

static void uncompressReadBlock(KvdoWorkItem *workItem)

{

  DataKVIO  *dataKVIO  = workItemAsDataKVIO(workItem);

  ReadBlock *readBlock = &dataKVIO->readBlock;

  BlockSize  blockSize = VDO_BLOCK_SIZE;

  // The DataKVIO's scratch block will be used to contain the

  // uncompressed data.

  uint16_t fragmentOffset, fragmentSize;

  char *compressedData = readBlock->data;

  int result = getCompressedBlockFragment(readBlock->mappingState,

                                          compressedData, blockSize,

                                          &fragmentOffset,

                                          &fragmentSize);

  if (result != VDO_SUCCESS) {

    logDebug("%s: frag err %d", __func__, result);

    readBlock->status = result;

    readBlock->callback(dataKVIO);

    return;

  }

  char *fragment = compressedData + fragmentOffset;

  int size = LZ4_uncompress_unknownOutputSize(fragment, dataKVIO->scratchBlock,

                                              fragmentSize, blockSize);

  if (size == blockSize) {

    readBlock->data = dataKVIO->scratchBlock;

  } else {

    logDebug("%s: lz4 error", __func__);

    readBlock->status = VDO_INVALID_FRAGMENT;

  }

  readBlock->callback(dataKVIO);

}

/**

 * Now that we have gotten the data from storage, uncompress the data if

 * necessary and then call back the requesting DataKVIO.

 *

 * @param dataKVIO  The DataKVIO requesting the data

 * @param result    The result of the read operation

 **/

static void completeRead(DataKVIO *dataKVIO, int result)

{

  ReadBlock *readBlock = &dataKVIO->readBlock;

  readBlock->status = result;

  if ((result == VDO_SUCCESS) && isCompressed(readBlock->mappingState)) {

    launchDataKVIOOnCPUQueue(dataKVIO, uncompressReadBlock, NULL,

                             CPU_Q_ACTION_COMPRESS_BLOCK);

    return;

  }

  readBlock->callback(dataKVIO);

}

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

/**

 * Callback for a bio doing a read.

 *

 * @param bio     The bio

 */

static void readBioCallback(BIO *bio)

#else

/**

 * Callback for a bio doing a read.

 *

 * @param bio     The bio

 * @param result  The result of the read operation

 */

static void readBioCallback(BIO *bio, int result)

#endif

{

  KVIO *kvio = (KVIO *) bio->bi_private;

  DataKVIO *dataKVIO = kvioAsDataKVIO(kvio);

  dataKVIO->readBlock.data = dataKVIO->readBlock.buffer;

  dataKVIOAddTraceRecord(dataKVIO, THIS_LOCATION(NULL));

  countCompletedBios(bio);

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

  completeRead(dataKVIO, getBioResult(bio));

#else

  completeRead(dataKVIO, result);

#endif

}

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

void kvdoReadBlock(DataVIO             *dataVIO,

                   PhysicalBlockNumber  location,

                   BlockMappingState    mappingState,

                   BioQAction           action,

                   DataKVIOCallback     callback)

{

  dataVIOAddTraceRecord(dataVIO, THIS_LOCATION(NULL));

  DataKVIO    *dataKVIO  = dataVIOAsDataKVIO(dataVIO);

  ReadBlock   *readBlock = &dataKVIO->readBlock;

  KernelLayer *layer     = getLayerFromDataKVIO(dataKVIO);

  readBlock->callback     = callback;

  readBlock->status       = VDO_SUCCESS;

  readBlock->mappingState = mappingState;

  BUG_ON(getBIOFromDataKVIO(dataKVIO)->bi_private != &dataKVIO->kvio);

  // Read the data directly from the device using the read bio.

  BIO *bio = readBlock->bio;

  resetBio(bio, layer);

  setBioSector(bio, blockToSector(layer, location));

  setBioOperationRead(bio);

  bio->bi_end_io = readBioCallback;

  submitBio(bio, action);

}

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

void kvdoReadDataVIO(DataVIO *dataVIO)

{

  ASSERT_LOG_ONLY(!isWriteVIO(dataVIOAsVIO(dataVIO)),

                  "operation set correctly for data read");

  dataVIOAddTraceRecord(dataVIO, THIS_LOCATION("$F;io=readData"));

  if (isCompressed(dataVIO->mapped.state)) {

    kvdoReadBlock(dataVIO, dataVIO->mapped.pbn, dataVIO->mapped.state,

                  BIO_Q_ACTION_COMPRESSED_DATA, readDataKVIOReadBlockCallback);

    return;

  }

  KVIO *kvio = dataVIOAsKVIO(dataVIO);

  BIO  *bio  = kvio->bio;

  bio->bi_end_io = resetUserBio;

  setBioSector(bio, blockToSector(kvio->layer, dataVIO->mapped.pbn));

  submitBio(bio, BIO_Q_ACTION_DATA);

}

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

static void kvdoAcknowledgeDataKVIOThenContinue(KvdoWorkItem *item)

{

  DataKVIO *dataKVIO = workItemAsDataKVIO(item);

  dataKVIOAddTraceRecord(dataKVIO, THIS_LOCATION(NULL));

  kvdoAcknowledgeDataKVIO(dataKVIO);

  // Even if we're not using bio-ack threads, we may be in the wrong

  // base-code thread.

  kvdoEnqueueDataVIOCallback(dataKVIO);

}

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

void kvdoAcknowledgeDataVIO(DataVIO *dataVIO)

{

  DataKVIO    *dataKVIO = dataVIOAsDataKVIO(dataVIO);

  KernelLayer *layer    = getLayerFromDataKVIO(dataKVIO);

  // If the remaining discard work is not completely processed by this VIO,

  // don't acknowledge it yet.

  if (isDiscardBio(dataKVIO->externalIORequest.bio)

      && (dataKVIO->remainingDiscard

          > (VDO_BLOCK_SIZE - dataKVIO->offset))) {

    invokeCallback(dataVIOAsCompletion(dataVIO));

    return;

  }

  // We've finished with the KVIO; acknowledge completion of the bio to the

  // kernel.

  if (useBioAckQueue(layer)) {

    dataVIOAddTraceRecord(dataVIO, THIS_LOCATION(NULL));

    launchDataKVIOOnBIOAckQueue(dataKVIO, kvdoAcknowledgeDataKVIOThenContinue,

                                NULL, BIO_ACK_Q_ACTION_ACK);

  } else {

    kvdoAcknowledgeDataKVIOThenContinue(workItemFromDataKVIO(dataKVIO));

  }

}

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

void kvdoWriteDataVIO(DataVIO *dataVIO)

{

  ASSERT_LOG_ONLY(isWriteVIO(dataVIOAsVIO(dataVIO)),

                  "kvdoWriteDataVIO() called on write DataVIO");

  dataVIOAddTraceRecord(dataVIO, THIS_LOCATION("$F;io=writeData;j=normal"));

  KVIO *kvio  = dataVIOAsKVIO(dataVIO);

  BIO  *bio   = kvio->bio;

  setBioOperationWrite(bio);

  setBioSector(bio, blockToSector(kvio->layer, dataVIO->newMapped.pbn));

  submitBio(bio, BIO_Q_ACTION_DATA);

}

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

void kvdoModifyWriteDataVIO(DataVIO *dataVIO)

{

  dataVIOAddTraceRecord(dataVIO, THIS_LOCATION(NULL));

  DataKVIO    *dataKVIO = dataVIOAsDataKVIO(dataVIO);

  BIO         *bio      = dataKVIO->externalIORequest.bio;

  KernelLayer *layer    = getLayerFromDataKVIO(dataKVIO);

  resetBio(dataKVIO->dataBlockBio, layer);

  if (!isDiscardBio(bio)) {

    bioCopyDataIn(bio, dataKVIO->dataBlock + dataKVIO->offset);

  } else {

    memset(dataKVIO->dataBlock + dataKVIO->offset, '\0',

           min(dataKVIO->remainingDiscard,

               (DiscardSize) (VDO_BLOCK_SIZE - dataKVIO->offset)));

  }

  dataVIO->isZeroBlock               = bioIsZeroData(dataKVIO->dataBlockBio);

  dataKVIO->dataBlockBio->bi_private = &dataKVIO->kvio;

  copyBioOperationAndFlags(dataKVIO->dataBlockBio, bio);

  // Make the bio a write, not (potentially) a discard.

  setBioOperationWrite(dataKVIO->dataBlockBio);

}

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

void kvdoZeroDataVIO(DataVIO *dataVIO)

{

  dataVIOAddTraceRecord(dataVIO, THIS_LOCATION("zeroDataVIO;io=readData"));

  bioZeroData(dataVIOAsKVIO(dataVIO)->bio);

}

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

void kvdoCopyDataVIO(DataVIO *source, DataVIO *destination)

{

  dataVIOAddTraceRecord(destination, THIS_LOCATION(NULL));

  bioCopyDataOut(dataVIOAsKVIO(destination)->bio,

                 dataVIOAsDataKVIO(source)->dataBlock);

}

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

static void kvdoCompressWork(KvdoWorkItem *item)

{

  DataKVIO    *dataKVIO = workItemAsDataKVIO(item);

  KernelLayer *layer    = getLayerFromDataKVIO(dataKVIO);

  dataKVIOAddTraceRecord(dataKVIO, THIS_LOCATION(NULL));

  char *context = getWorkQueuePrivateData();

  if (unlikely(context == NULL)) {

    uint32_t index = atomicAdd32(&layer->compressionContextIndex, 1) - 1;

    BUG_ON(index >= layer->deviceConfig->threadCounts.cpuThreads);

    context = layer->compressionContext[index];

    setWorkQueuePrivateData(context);

  }

  int size = LZ4_compress_ctx_limitedOutput(context, dataKVIO->dataBlock,

                                            dataKVIO->scratchBlock,

                                            VDO_BLOCK_SIZE,

                                            VDO_BLOCK_SIZE);

  DataVIO *dataVIO = &dataKVIO->dataVIO;

  if (size > 0) {

    // The scratch block will be used to contain the compressed data.

    dataVIO->compression.data = dataKVIO->scratchBlock;

    dataVIO->compression.size = size;

  } else {

    // Use block size plus one as an indicator for uncompressible data.

    dataVIO->compression.size = VDO_BLOCK_SIZE + 1;

  }

  kvdoEnqueueDataVIOCallback(dataKVIO);

}

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

void kvdoCompressDataVIO(DataVIO *dataVIO)

{

  dataVIOAddTraceRecord(dataVIO,

                        THIS_LOCATION("compressDataVIO;"

                                      "io=compress;cb=compress"));

  /*

   * If the orignal bio was a discard, but we got this far because the discard

   * was a partial one (r/m/w), and it is part of a larger discard, we cannot

   * compress this VIO. We need to make sure the VIO completes ASAP.

   */

  DataKVIO *dataKVIO = dataVIOAsDataKVIO(dataVIO);

  if (isDiscardBio(dataKVIO->externalIORequest.bio)

      && (dataKVIO->remainingDiscard > 0)) {

    dataVIO->compression.size = VDO_BLOCK_SIZE + 1;

    kvdoEnqueueDataVIOCallback(dataKVIO);

    return;

  }

  launchDataKVIOOnCPUQueue(dataKVIO, kvdoCompressWork, NULL,

                           CPU_Q_ACTION_COMPRESS_BLOCK);

}

/**

 * Construct a DataKVIO.

 *

 * @param [in]  layer        The physical layer

 * @param [in]  bio          The bio to associate with this DataKVIO

 * @param [out] dataKVIOPtr  A pointer to hold the new DataKVIO

 *

 * @return VDO_SUCCESS or an error

 **/

__attribute__((warn_unused_result))

static int makeDataKVIO(KernelLayer *layer, BIO *bio, DataKVIO **dataKVIOPtr)

{

  DataKVIO *dataKVIO;

  int result = allocBufferFromPool(layer->dataKVIOPool, (void **) &dataKVIO);

  if (result != VDO_SUCCESS) {

    return logErrorWithStringError(result, "data kvio allocation failure");

  }

  if (WRITE_PROTECT_FREE_POOL) {

    setWriteProtect(dataKVIO, WP_DATA_KVIO_SIZE, false);

  }

  KVIO *kvio = &dataKVIO->kvio;

  kvio->vio = dataVIOAsVIO(&dataKVIO->dataVIO);

  memset(&kvio->enqueueable, 0, sizeof(KvdoEnqueueable));

  memset(&dataKVIO->dedupeContext.pendingList, 0, sizeof(struct list_head));

  memset(&dataKVIO->dataVIO, 0, sizeof(DataVIO));

  kvio->bioToSubmit = NULL;

  bio_list_init(&kvio->biosMerged);

  // The dataBlock is only needed for writes and some partial reads.

  if (isWriteBio(bio) || (getBioSize(bio) < VDO_BLOCK_SIZE)) {

    resetBio(dataKVIO->dataBlockBio, layer);

  }

  initializeKVIO(kvio, layer, VIO_TYPE_DATA, VIO_PRIORITY_DATA, NULL, bio);

  *dataKVIOPtr = dataKVIO;

  return VDO_SUCCESS;

}

/**

 * Creates a new DataVIO structure. A DataVIO represents a single logical

 * block of data. It is what most VDO operations work with. This function also

 * creates a wrapping DataKVIO structure that is used when we want to

 * physically read or write the data associated with the DataVIO.

 *

 * @param [in]  layer        The physical layer

 * @param [in]  bio          The BIO from the request the new DataKVIO will

 *                           service

 * @param [in]  arrivalTime  The arrival time of the BIO

 * @param [out] dataKVIOPtr  A pointer to hold the new DataKVIO

 *

 * @return VDO_SUCCESS or an error

 **/

static int kvdoCreateKVIOFromBio(KernelLayer  *layer,

                                 BIO          *bio,

                                 Jiffies       arrivalTime,

                                 DataKVIO    **dataKVIOPtr)

{

  ExternalIORequest externalIORequest = {

    .bio         = bio,

    .private     = bio->bi_private,

    .endIO       = bio->bi_end_io,

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

    .rw          = bio->bi_opf,

#else

    .rw          = bio->bi_rw,

#endif

  };

  // We will handle FUA at the end of the request (after we restore the

  // bi_rw field from externalIORequest.rw).

  clearBioOperationFlagFua(bio);

  DataKVIO *dataKVIO = NULL;

  int       result   = makeDataKVIO(layer, bio, &dataKVIO);

  if (result != VDO_SUCCESS) {

    return result;

  }

  dataKVIO->externalIORequest = externalIORequest;

  dataKVIO->offset = sectorToBlockOffset(layer, getBioSector(bio));

  dataKVIO->isPartial = ((getBioSize(bio) < VDO_BLOCK_SIZE)

                         || (dataKVIO->offset != 0));

  if (dataKVIO->isPartial) {

    countBios(&layer->biosInPartial, bio);

  } else {

    /*

     * Note that we unconditionally fill in the dataBlock array for

     * non-read operations. There are places like kvdoCopyVIO that may

     * look at kvio->dataBlock for a zero block (and maybe for

     * discards?). We could skip filling in dataBlock for such cases,

     * but only once we're sure all such places are fixed to check the

     * isZeroBlock flag first.

     */

    if (isDiscardBio(bio)) {

      /*

       * This is a discard/trim operation. This is treated much like the zero

       * block, but we keep different stats and distinguish it in the block

       * map.

       */

      memset(dataKVIO->dataBlock, 0, VDO_BLOCK_SIZE);

    } else if (bio_data_dir(bio) == WRITE) {

      dataKVIO->dataVIO.isZeroBlock = bioIsZeroData(bio);

      // Copy the bio data to a char array so that we can continue to use

      // the data after we acknowledge the bio.

      bioCopyDataIn(bio, dataKVIO->dataBlock);

    }

  }

  if (dataKVIO->isPartial || isWriteBio(bio)) {

    /*

     * dataKVIO->bio will point at kvio->dataBlockBio for all writes and

     * partial block I/O so the rest of the kernel code doesn't need to

     * make a decision as to what to use.

     */

    dataKVIO->dataBlockBio->bi_private = &dataKVIO->kvio;

    if (dataKVIO->isPartial && isWriteBio(bio)) {

      clearBioOperationAndFlags(dataKVIO->dataBlockBio);

      setBioOperationRead(dataKVIO->dataBlockBio);

    } else {

      copyBioOperationAndFlags(dataKVIO->dataBlockBio, bio);

    }

    dataKVIOAsKVIO(dataKVIO)->bio = dataKVIO->dataBlockBio;

    dataKVIO->readBlock.data      = dataKVIO->dataBlock;

  }

  setBioBlockDevice(bio, getKernelLayerBdev(layer));

  bio->bi_end_io = completeAsyncBio;

  *dataKVIOPtr   = dataKVIO;

  return VDO_SUCCESS;

}

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

static void launchDataKVIOWork(KvdoWorkItem *item)

{

  runCallback(vioAsCompletion(workItemAsKVIO(item)->vio));

}

/**

 * Continue discard processing for requests that span multiple physical blocks.

 * If all have been processed the KVIO is completed.  If we have already seen

 * an error, we skip the rest of the discard and fail immediately.

 *

 * 
Invoked in a request-queue thread after the discard of a block has

 * completed.

 *

 * @param completion  A completion representing the discard KVIO

 **/

static void kvdoContinueDiscardKVIO(VDOCompletion *completion)

{

  DataVIO     *dataVIO  = asDataVIO(completion);

  DataKVIO    *dataKVIO = dataVIOAsDataKVIO(dataVIO);

  KernelLayer *layer    = getLayerFromDataKVIO(dataKVIO);

  dataKVIO->remainingDiscard

    -= min(dataKVIO->remainingDiscard,

           (DiscardSize) (VDO_BLOCK_SIZE - dataKVIO->offset));

  if ((completion->result != VDO_SUCCESS)

      || (dataKVIO->remainingDiscard == 0)) {

    if (dataKVIO->hasDiscardPermit) {

      limiterRelease(&layer->discardLimiter);

      dataKVIO->hasDiscardPermit = false;

    }

    kvdoCompleteDataKVIO(completion);

    return;

  }

  BIO *bio = getBIOFromDataKVIO(dataKVIO);

  resetBio(bio, layer);

  dataKVIO->isPartial = (dataKVIO->remainingDiscard < VDO_BLOCK_SIZE);

  dataKVIO->offset    = 0;

  VIOOperation operation;

  if (dataKVIO->isPartial) {

    operation  = VIO_READ_MODIFY_WRITE;

    setBioOperationRead(bio);

  } else {

    operation  = VIO_WRITE;

  }

  if (requestorSetFUA(dataKVIO)) {

    operation |= VIO_FLUSH_AFTER;

  }

  prepareDataVIO(dataVIO, dataVIO->logical.lbn + 1, operation,

                 !dataKVIO->isPartial, kvdoContinueDiscardKVIO);

  enqueueDataKVIO(dataKVIO, launchDataKVIOWork, completion->callback,

                  REQ_Q_ACTION_MAP_BIO);

}

/**

 * Finish a partial read.

 *

 * @param completion  The partial read KVIO

 **/

static void kvdoCompletePartialRead(VDOCompletion *completion)

{

  DataKVIO *dataKVIO = dataVIOAsDataKVIO(asDataVIO(completion));

  dataKVIOAddTraceRecord(dataKVIO, THIS_LOCATION(NULL));

  bioCopyDataOut(dataKVIO->externalIORequest.bio,

                 dataKVIO->readBlock.data + dataKVIO->offset);

  kvdoCompleteDataKVIO(completion);

  return;

}

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

int kvdoLaunchDataKVIOFromBio(KernelLayer *layer,

                              BIO         *bio,

                              uint64_t     arrivalTime,

                              bool         hasDiscardPermit)

{

  DataKVIO *dataKVIO = NULL;

  int result = kvdoCreateKVIOFromBio(layer, bio, arrivalTime, &dataKVIO);

  if (unlikely(result != VDO_SUCCESS)) {

    logInfo("%s: KVIO allocation failure", __func__);

    if (hasDiscardPermit) {

      limiterRelease(&layer->discardLimiter);

    }

    limiterRelease(&layer->requestLimiter);

    return mapToSystemError(result);

  }

  /*

   * Discards behave very differently than other requests when coming

   * in from device-mapper. We have to be able to handle any size discards

   * and with various sector offsets within a block.

   */

  KVIO         *kvio      = &dataKVIO->kvio;

  VDOAction    *callback  = kvdoCompleteDataKVIO;

  VIOOperation  operation = VIO_WRITE;

  bool          isTrim    = false;

  if (isDiscardBio(bio)) {

    dataKVIO->hasDiscardPermit = hasDiscardPermit;

    dataKVIO->remainingDiscard = getBioSize(bio);

    callback                   = kvdoContinueDiscardKVIO;

    if (dataKVIO->isPartial) {

      operation = VIO_READ_MODIFY_WRITE;

    } else {

      isTrim = true;

    }

  } else if (dataKVIO->isPartial) {

    if (bio_data_dir(bio) == READ) {

      callback  = kvdoCompletePartialRead;

      operation = VIO_READ;

    } else {

      operation = VIO_READ_MODIFY_WRITE;

    }

  } else if (bio_data_dir(bio) == READ) {

    operation = VIO_READ;

  }

  if (requestorSetFUA(dataKVIO)) {

    operation |= VIO_FLUSH_AFTER;

  }

  LogicalBlockNumber lbn

    = sectorToBlock(layer, getBioSector(bio) - layer->startingSectorOffset);

  prepareDataVIO(&dataKVIO->dataVIO, lbn, operation, isTrim, callback);

  enqueueKVIO(kvio, launchDataKVIOWork, vioAsCompletion(kvio->vio)->callback,

              REQ_Q_ACTION_MAP_BIO);

  return VDO_SUCCESS;

}

/**

 * Hash a DataKVIO and set its chunk name.

 *

 * @param item  The DataKVIO to be hashed

 **/

static void kvdoHashDataWork(KvdoWorkItem *item)

{

  DataKVIO *dataKVIO = workItemAsDataKVIO(item);

  DataVIO  *dataVIO  = &dataKVIO->dataVIO;

  dataVIOAddTraceRecord(dataVIO, THIS_LOCATION(NULL));

  MurmurHash3_x64_128(dataKVIO->dataBlock, VDO_BLOCK_SIZE, 0x62ea60be,

                      &dataVIO->chunkName);

  dataKVIO->dedupeContext.chunkName = &dataVIO->chunkName;

  kvdoEnqueueDataVIOCallback(dataKVIO);

}

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

void kvdoHashDataVIO(DataVIO *dataVIO)

{

  dataVIOAddTraceRecord(dataVIO, THIS_LOCATION(NULL));