/*

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

 */

#include "kvio.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "numUtils.h"

#include "vdo.h"

#include "waitQueue.h"

#include "bio.h"

#include "ioSubmitter.h"

#include "kvdoFlush.h"

/**

 * A function to tell vdo that we have completed the requested async

 * operation for a vio

 *

 * @param item    The work item of the VIO to complete

 **/

static void kvdoHandleVIOCallback(KvdoWorkItem *item)

{

  KVIO *kvio = workItemAsKVIO(item);

  runCallback(vioAsCompletion(kvio->vio));

}

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

void kvdoEnqueueVIOCallback(KVIO *kvio)

{

  enqueueKVIO(kvio, kvdoHandleVIOCallback,

              (KvdoWorkFunction) vioAsCompletion(kvio->vio)->callback,

              REQ_Q_ACTION_VIO_CALLBACK);

}

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

void kvdoContinueKvio(KVIO *kvio, int error)

{

  if (unlikely(error != VDO_SUCCESS)) {

    setCompletionResult(vioAsCompletion(kvio->vio), error);

  }

  kvdoEnqueueVIOCallback(kvio);

}

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

// noinline ensures systemtap can hook in here

static noinline void maybeLogKvioTrace(KVIO *kvio)

{

  if (kvio->layer->traceLogging) {

    logKvioTrace(kvio);

  }

}

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

static void freeKVIO(KVIO **kvioPtr)

{

  KVIO *kvio = *kvioPtr;

  if (kvio == NULL) {

    return;

  }

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

    maybeLogKvioTrace(kvio);

    FREE(kvio->vio->trace);

  }

  freeBio(kvio->bio, kvio->layer);

  FREE(kvio);

  *kvioPtr = NULL;

}

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

void freeMetadataKVIO(MetadataKVIO **metadataKVIOPtr)

{

  freeKVIO((KVIO **) metadataKVIOPtr);

}

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

void freeCompressedWriteKVIO(CompressedWriteKVIO **compressedWriteKVIOPtr)

{

  freeKVIO((KVIO **) compressedWriteKVIOPtr);

}

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

void kvdoWriteCompressedBlock(AllocatingVIO *allocatingVIO)

{

  // This method assumes that compressed writes never set the flush or FUA

  // bits.

  CompressedWriteKVIO *compressedWriteKVIO

    = allocatingVIOAsCompressedWriteKVIO(allocatingVIO);

  KVIO *kvio = compressedWriteKVIOAsKVIO(compressedWriteKVIO);

  BIO  *bio  = kvio->bio;

  resetBio(bio, kvio->layer);

  setBioOperationWrite(bio);

  setBioSector(bio, blockToSector(kvio->layer, kvio->vio->physical));

  submitBio(bio, BIO_Q_ACTION_COMPRESSED_DATA);

}

/**

 * Get the BioQueue action for a metadata VIO based on that VIO's priority.

 *

 * @param vio  The VIO

 *

 * @return The action with which to submit the VIO's BIO.

 **/

static inline BioQAction getMetadataAction(VIO *vio)

{

  return ((vio->priority == VIO_PRIORITY_HIGH)

          ? BIO_Q_ACTION_HIGH : BIO_Q_ACTION_METADATA);

}

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

void kvdoSubmitMetadataVIO(VIO *vio)

{

  KVIO *kvio = metadataKVIOAsKVIO(vioAsMetadataKVIO(vio));

  BIO  *bio  = kvio->bio;

  resetBio(bio, kvio->layer);

  setBioSector(bio, blockToSector(kvio->layer, vio->physical));

  // Metadata I/Os bypass the read cache.

  if (isReadVIO(vio)) {

    ASSERT_LOG_ONLY(!vioRequiresFlushBefore(vio),

                    "read VIO does not require flush before");

    vioAddTraceRecord(vio, THIS_LOCATION("$F;io=readMeta"));

    setBioOperationRead(bio);

  } else {

    KernelLayerState state = getKernelLayerState(kvio->layer);

    ASSERT_LOG_ONLY(((state == LAYER_RUNNING)

                     || (state == LAYER_RESUMING)

                     || (state = LAYER_STARTING)),

                    "write metadata in allowed state %d", state);

    if (vioRequiresFlushBefore(vio)) {

      setBioOperationWrite(bio);

      setBioOperationFlagPreflush(bio);

      vioAddTraceRecord(vio, THIS_LOCATION("$F;io=flushWriteMeta"));

    } else {

      setBioOperationWrite(bio);

      vioAddTraceRecord(vio, THIS_LOCATION("$F;io=writeMeta"));

    }

  }

  if (vioRequiresFlushAfter(vio)) {

    setBioOperationFlagFua(bio);

  }

  submitBio(bio, getMetadataAction(vio));

}

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

/**

 * Handle the completion of a base-code initiated flush by continuing the flush

 * VIO.

 *

 * @param bio    The bio to complete

 **/

static void completeFlushBio(BIO *bio)

#else

/**

 * Handle the completion of a base-code initiated flush by continuing the flush

 * VIO.

 *

 * @param bio    The bio to complete

 * @param error  Possible error from underlying block device

 **/

static void completeFlushBio(BIO *bio, int error)

#endif

{

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

  int error = getBioResult(bio);

#endif

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

  // XXX This assumes a VDO-created bio around a buffer contains exactly 1

  // page, which we believe is true, but do not assert.

  bio->bi_vcnt = 1;

  // Restore the bio's notion of its own data.

  resetBio(bio, kvio->layer);

  kvdoContinueKvio(kvio, error);

}

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

void kvdoFlushVIO(VIO *vio)

{

  KVIO        *kvio  = metadataKVIOAsKVIO(vioAsMetadataKVIO(vio));

  BIO         *bio   = kvio->bio;

  KernelLayer *layer = kvio->layer;

  resetBio(bio, layer);

  prepareFlushBIO(bio, kvio, getKernelLayerBdev(layer), completeFlushBio);

  submitBio(bio, getMetadataAction(vio));

}

/*

 * Hook for a SystemTap probe to potentially restrict the choices

 * of which VIOs should have their latencies tracked.

 *

 * Normally returns true. Even if true is returned, sampleThisOne may

 * cut down the monitored VIOs by some fraction so as to reduce the

 * impact on system performance.

 *

 * Must be "noinline" so that SystemTap can find the return

 * instruction and modify the return value.

 *

 * @param kvio   The KVIO being initialized

 * @param layer  The kernel layer

 * @param bio    The incoming I/O request

 *

 * @return whether it's useful to track latency for VIOs looking like

 *         this one

 */

static noinline bool

sampleThisVIO(KVIO *kvio, KernelLayer *layer, BIO *bio)

{

  bool result = true;

  // Ensure the arguments and result exist at the same time, for SystemTap.

  __asm__ __volatile__(""

                       : "=g" (result)

                       : "0" (result),

                         "g" (kvio),

                         "g" (layer),

                         "g" (bio)

                       : "memory");

  return result;

}

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

void initializeKVIO(KVIO        *kvio,

                    KernelLayer *layer,

                    VIOType      vioType,

                    VIOPriority  priority,

                    void        *parent,

                    BIO         *bio)

{

  if (layer->vioTraceRecording

      && sampleThisVIO(kvio, layer, bio)

      && sampleThisOne(&layer->traceSampleCounter)) {

    int result = (isDataVIOType(vioType)

                  ? allocTraceFromPool(layer, &kvio->vio->trace)

                  : ALLOCATE(1, Trace, "trace", &kvio->vio->trace));

    if (result != VDO_SUCCESS) {

      logError("trace record allocation failure %d", result);

    }

  }

  kvio->bio   = bio;

  kvio->layer = layer;

  if (bio != NULL) {

    bio->bi_private = kvio;

  }

  initializeVIO(kvio->vio, vioType, priority, parent, getVDO(&layer->kvdo),

                &layer->common);

  // XXX: The "init" label should be replaced depending on the

  // write/read/flush path followed.

  kvioAddTraceRecord(kvio, THIS_LOCATION("$F;io=?init;j=normal"));

  VDOCompletion *completion                = vioAsCompletion(kvio->vio);

  kvio->enqueueable.enqueueable.completion = completion;

  completion->enqueueable                  = &kvio->enqueueable.enqueueable;

}

/**

 * Construct a metadata KVIO.

 *

 * @param [in]  layer            The physical layer

 * @param [in]  vioType          The type of VIO to create

 * @param [in]  priority         The relative priority to assign to the

 *                               MetadataKVIO

 * @param [in]  parent           The parent of the MetadataKVIO completion

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

 * @param [out] metadataKVIOPtr  A pointer to hold the new MetadataKVIO

 *

 * @return VDO_SUCCESS or an error

 **/

__attribute__((warn_unused_result))

static int makeMetadataKVIO(KernelLayer   *layer,

                            VIOType        vioType,

                            VIOPriority    priority,

                            void          *parent,

                            BIO           *bio,

                            MetadataKVIO **metadataKVIOPtr)

{

  // If MetadataKVIO grows past 256 bytes, we'll lose benefits of VDOSTORY-176.

  STATIC_ASSERT(sizeof(MetadataKVIO) <= 256);

  // Metadata VIOs should use direct allocation and not use the buffer pool,

  // which is reserved for submissions from the linux block layer.

  MetadataKVIO *metadataKVIO;

  int result = ALLOCATE(1, MetadataKVIO, __func__, &metadataKVIO);

  if (result != VDO_SUCCESS) {

    logError("metadata KVIO allocation failure %d", result);

    return result;

  }

  KVIO *kvio = &metadataKVIO->kvio;

  kvio->vio  = &metadataKVIO->vio;

  initializeKVIO(kvio, layer, vioType, priority, parent, bio);

  *metadataKVIOPtr = metadataKVIO;

  return VDO_SUCCESS;

}

/**

 * Construct a CompressedWriteKVIO.

 *

 * @param [in]  layer                   The physical layer

 * @param [in]  parent                  The parent of the CompressedWriteKVIO

 *                                      completion

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

 *                                      CompressedWriteKVIO

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

 *                                      CompressedWriteKVIO

 *

 * @return VDO_SUCCESS or an error

 **/

__attribute__((warn_unused_result))

static int

makeCompressedWriteKVIO(KernelLayer          *layer,

                        void                 *parent,

                        BIO                  *bio,

                        CompressedWriteKVIO **compressedWriteKVIOPtr)

{

  // Compressed write VIOs should use direct allocation and not use the buffer

  // pool, which is reserved for submissions from the linux block layer.

  CompressedWriteKVIO *compressedWriteKVIO;

  int result = ALLOCATE(1, CompressedWriteKVIO, __func__,

                        &compressedWriteKVIO);

  if (result != VDO_SUCCESS) {

    logError("compressed write KVIO allocation failure %d", result);

    return result;

  }

  KVIO *kvio = &compressedWriteKVIO->kvio;

  kvio->vio  = allocatingVIOAsVIO(&compressedWriteKVIO->allocatingVIO);

  initializeKVIO(kvio, layer, VIO_TYPE_COMPRESSED_BLOCK,

                 VIO_PRIORITY_COMPRESSED_DATA, parent, bio);

  *compressedWriteKVIOPtr = compressedWriteKVIO;

  return VDO_SUCCESS;

}

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

int kvdoCreateMetadataVIO(PhysicalLayer  *layer,

                          VIOType         vioType,

                          VIOPriority     priority,

                          void           *parent,

                          char           *data,

                          VIO           **vioPtr)

{

  int result = ASSERT(isMetadataVIOType(vioType),

                      "%d is a metadata type", vioType);

  if (result != VDO_SUCCESS) {

    return result;

  }

  BIO *bio;

  KernelLayer *kernelLayer = asKernelLayer(layer);

  result = createBio(kernelLayer, data, &bio;;

  if (result != VDO_SUCCESS) {

    return result;

  }

  MetadataKVIO *metadataKVIO;

  result = makeMetadataKVIO(kernelLayer, vioType, priority, parent, bio,

                            &metadataKVIO);

  if (result != VDO_SUCCESS) {

    freeBio(bio, kernelLayer);

    return result;

  }

  *vioPtr = &metadataKVIO->vio;

  return VDO_SUCCESS;

}

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

int kvdoCreateCompressedWriteVIO(PhysicalLayer  *layer,

                                 void           *parent,

                                 char           *data,

                                 AllocatingVIO **allocatingVIOPtr)

{

  BIO *bio;

  KernelLayer *kernelLayer = asKernelLayer(layer);

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

  if (result != VDO_SUCCESS) {

    return result;

  }

  CompressedWriteKVIO *compressedWriteKVIO;

  result = makeCompressedWriteKVIO(kernelLayer, parent, bio,

                                   &compressedWriteKVIO);

  if (result != VDO_SUCCESS) {

    freeBio(bio, kernelLayer);

    return result;

  }

  *allocatingVIOPtr = &compressedWriteKVIO->allocatingVIO;

  return VDO_SUCCESS;

}