/*

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

 */

#include "dataVIO.h"

#include "logger.h"

#include "atomic.h"

#include "blockMap.h"

#include "compressionState.h"

#include "extent.h"

#include "logicalZone.h"

#include "threadConfig.h"

#include "vdoInternal.h"

#include "vioRead.h"

#include "vioWrite.h"

static const char *ASYNC_OPERATION_NAMES[] = {

  "launch",

  "acknowledgeWrite",

  "acquireHashLock",

  "acquireLogicalBlockLock",

  "acquirePBNReadLock",

  "checkForDedupeForRollover",

  "checkForDeduplication",

  "compressData",

  "continueVIOAsync",

  "findBlockMapSlot",

  "getMappedBlock",

  "getMappedBlockForDedupe",

  "getMappedBlockForWrite",

  "hashData",

  "journalDecrementForDedupe",

  "journalDecrementForWrite",

  "journalIncrementForCompression",

  "journalIncrementForDedupe",

  "journalIncrementForWrite",

  "journalMappingForCompression",

  "journalMappingForDedupe",

  "journalMappingForWrite",

  "journalUnmappingForDedupe",

  "journalUnmappingForWrite",

  "attemptPacking",

  "putMappedBlock",

  "putMappedBlockForDedupe",

  "readData",

  "updateIndex",

  "verifyDeduplication",

  "writeData",

};

/**

 * Initialize the LBN lock of a DataVIO. In addition to recording the LBN on

 * which the DataVIO will operate, it will also find the logical zone

 * associated with the LBN.

 *

 * @param dataVIO  The dataVIO to initialize

 * @param lbn      The lbn on which the dataVIO will operate

 **/

static void initializeLBNLock(DataVIO *dataVIO, LogicalBlockNumber lbn)

{

  LBNLock *lock = &dataVIO->logical;

  lock->lbn     = lbn;

  lock->locked  = false;

  initializeWaitQueue(&lock->waiters);

  VDO *vdo   = getVDOFromDataVIO(dataVIO);

  lock->zone = getLogicalZone(vdo->logicalZones, computeLogicalZone(dataVIO));

}

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

void prepareDataVIO(DataVIO            *dataVIO,

                    LogicalBlockNumber  lbn,

                    VIOOperation        operation,

                    bool                isTrim,

                    VDOAction          *callback)

{

  // Clearing the tree lock must happen before initializing the LBN lock,

  // which also adds information to the tree lock.

  memset(&dataVIO->treeLock,  0, sizeof(dataVIO->treeLock));

  initializeLBNLock(dataVIO, lbn);

  initializeRing(&dataVIO->hashLockNode);

  initializeRing(&dataVIO->writeNode);

  resetAllocation(dataVIOAsAllocatingVIO(dataVIO));

  dataVIO->isDuplicate = false;

  memset(&dataVIO->chunkName, 0, sizeof(dataVIO->chunkName));

  memset(&dataVIO->duplicate, 0, sizeof(dataVIO->duplicate));

  VIO *vio       = dataVIOAsVIO(dataVIO);

  vio->operation = operation;

  vio->callback  = callback;

  dataVIO->pageCompletion.completion.enqueueable

    = vioAsCompletion(vio)->enqueueable;

  dataVIO->mapped.state = MAPPING_STATE_UNCOMPRESSED;

  dataVIO->newMapped.state

    = (isTrim ? MAPPING_STATE_UNMAPPED : MAPPING_STATE_UNCOMPRESSED);

  resetCompletion(vioAsCompletion(vio));

  setLogicalCallback(dataVIO, attemptLogicalBlockLock,

                     THIS_LOCATION("$F;cb=acquireLogicalBlockLock"));

}

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

void completeDataVIO(VDOCompletion *completion)

{

  DataVIO *dataVIO = asDataVIO(completion);

  if (completion->result != VDO_SUCCESS) {

    VIO *vio = dataVIOAsVIO(dataVIO);

    updateVIOErrorStats(vio,

                        "Completing %s VIO for LBN %" PRIu64

                        " with error after %s",

                        getVIOReadWriteFlavor(vio), dataVIO->logical.lbn,

                        getOperationName(dataVIO));

  }

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

  if (isReadDataVIO(dataVIO)) {

    cleanupReadDataVIO(dataVIO);

  } else {

    cleanupWriteDataVIO(dataVIO);

  }

}

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

void finishDataVIO(DataVIO *dataVIO, int result)

{

  VDOCompletion *completion = dataVIOAsCompletion(dataVIO);

  setCompletionResult(completion, result);

  completeDataVIO(completion);

}

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

const char *getOperationName(DataVIO *dataVIO)

{

  STATIC_ASSERT((MAX_ASYNC_OPERATION_NUMBER - MIN_ASYNC_OPERATION_NUMBER)

                == COUNT_OF(ASYNC_OPERATION_NAMES));

  return ((dataVIO->lastAsyncOperation < MAX_ASYNC_OPERATION_NUMBER)

          ? ASYNC_OPERATION_NAMES[dataVIO->lastAsyncOperation]

          : "unknown async operation");

}

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

void receiveDedupeAdvice(DataVIO *dataVIO, const DataLocation *advice)

{

  /*

   * NOTE: this is called on non-base-code threads. Be very careful to not do

   * anything here that needs a base code thread-local variable, such as

   * trying to get the current thread ID, or that does a lot of work.

   */

  VDO *vdo = getVDOFromDataVIO(dataVIO);

  ZonedPBN duplicate = validateDedupeAdvice(vdo, advice, dataVIO->logical.lbn);

  setDuplicateLocation(dataVIO, duplicate);

}

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

void setDuplicateLocation(DataVIO *dataVIO, const ZonedPBN source)

{

  dataVIO->isDuplicate = (source.pbn != ZERO_BLOCK);

  dataVIO->duplicate   = source;

}

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

void clearMappedLocation(DataVIO *dataVIO)

{

  dataVIO->mapped = (ZonedPBN) { .state = MAPPING_STATE_UNMAPPED };

}

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

int setMappedLocation(DataVIO             *dataVIO,

                      PhysicalBlockNumber  pbn,

                      BlockMappingState    state)

{

  PhysicalZone *zone;

  int result = getPhysicalZone(getVDOFromDataVIO(dataVIO), pbn, &zone);

  if (result != VDO_SUCCESS) {

    return result;

  }

  dataVIO->mapped = (ZonedPBN) {

    .pbn   = pbn,

    .state = state,

    .zone  = zone,

  };

  return VDO_SUCCESS;

}

/**

 * Launch a request which has acquired an LBN lock.

 *

 * @param dataVIO  The DataVIO which has just acquired a lock

 **/

static void launchLockedRequest(DataVIO *dataVIO)

{

  dataVIOAddTraceRecord(dataVIO, THIS_LOCATION(NULL));

  dataVIO->logical.locked = true;

  if (isWriteDataVIO(dataVIO)) {

    launchWriteDataVIO(dataVIO);

  } else {

    launchReadDataVIO(dataVIO);

  }

}

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

void attemptLogicalBlockLock(VDOCompletion *completion)

{

  DataVIO *dataVIO = asDataVIO(completion);

  assertInLogicalZone(dataVIO);

  if (dataVIO->logical.lbn

      >= getVDOFromDataVIO(dataVIO)->config.logicalBlocks) {

    finishDataVIO(dataVIO, VDO_OUT_OF_RANGE);

    return;

  }

  DataVIO *lockHolder;

  LBNLock *lock = &dataVIO->logical;

  int result = intMapPut(getLBNLockMap(lock->zone), lock->lbn, dataVIO, false,

                         (void **) &lockHolder);

  if (result != VDO_SUCCESS) {

    finishDataVIO(dataVIO, result);

    return;

  }

  if (lockHolder == NULL) {

    // We got the lock

    launchLockedRequest(dataVIO);

    return;

  }

  result = ASSERT(lockHolder->logical.locked, "logical block lock held");

  if (result != VDO_SUCCESS) {

    finishDataVIO(dataVIO, result);

    return;

  }

  /*

   * If the new request is a pure read request (not read-modify-write) and

   * the lockHolder is writing and has received an allocation (VDO-2683),

   * service the read request immediately by copying data from the lockHolder

   * to avoid having to flush the write out of the packer just to prevent the

   * read from waiting indefinitely. If the lockHolder does not yet have an

   * allocation, prevent it from blocking in the packer and wait on it.

   */

  if (isReadDataVIO(dataVIO) && atomicLoadBool(&lockHolder->hasAllocation)) {

    dataVIOAsCompletion(dataVIO)->layer->copyData(lockHolder, dataVIO);

    finishDataVIO(dataVIO, VDO_SUCCESS);

    return;

  }

  dataVIO->lastAsyncOperation = ACQUIRE_LOGICAL_BLOCK_LOCK;

  result = enqueueDataVIO(&lockHolder->logical.waiters, dataVIO,

                          THIS_LOCATION("$F;cb=logicalBlockLock"));

  if (result != VDO_SUCCESS) {

    finishDataVIO(dataVIO, result);

    return;

  }

  // Prevent writes and read-modify-writes from blocking indefinitely on

  // lock holders in the packer.

  if (!isReadDataVIO(lockHolder) && cancelCompression(lockHolder)) {

    dataVIO->compression.lockHolder = lockHolder;

    launchPackerCallback(dataVIO, removeLockHolderFromPacker,

                         THIS_LOCATION("$F;cb=removeLockHolderFromPacker"));

  }

}

/**

 * Release an uncontended LBN lock.

 *

 * @param dataVIO  The DataVIO holding the lock

 **/

static void releaseLock(DataVIO *dataVIO)

{

  LBNLock *lock    = &dataVIO->logical;

  IntMap  *lockMap = getLBNLockMap(lock->zone);

  if (!lock->locked) {

    // The lock is not locked, so it had better not be registered in the lock

    // map.

    DataVIO *lockHolder = intMapGet(lockMap, lock->lbn);

    ASSERT_LOG_ONLY((dataVIO != lockHolder),

                    "no logical block lock held for block %llu",

                    lock->lbn);

    return;

  }

  // Remove the lock from the logical block lock map, releasing the lock.

  DataVIO *lockHolder = intMapRemove(lockMap, lock->lbn);

  ASSERT_LOG_ONLY((dataVIO == lockHolder),

                  "logical block lock mismatch for block %llu", lock->lbn);

  lock->locked = false;

  return;

}

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

void releaseLogicalBlockLock(DataVIO *dataVIO)

{

  assertInLogicalZone(dataVIO);

  if (!hasWaiters(&dataVIO->logical.waiters)) {

    releaseLock(dataVIO);

    return;

  }

  LBNLock *lock = &dataVIO->logical;

  ASSERT_LOG_ONLY(lock->locked, "LBNLock with waiters is not locked");

  // Another DataVIO is waiting for the lock, so just transfer it in a single

  // lock map operation

  DataVIO *nextLockHolder = waiterAsDataVIO(dequeueNextWaiter(&lock->waiters));

  // Transfer the remaining lock waiters to the next lock holder.

  transferAllWaiters(&lock->waiters, &nextLockHolder->logical.waiters);

  DataVIO *lockHolder;

  int result = intMapPut(getLBNLockMap(lock->zone), lock->lbn, nextLockHolder,

                         true, (void **) &lockHolder);

  if (result != VDO_SUCCESS) {

    finishDataVIO(nextLockHolder, result);

    return;

  }

  ASSERT_LOG_ONLY((lockHolder == dataVIO),

                  "logical block lock mismatch for block %llu", lock->lbn);

  lock->locked = false;

  /*

   * If there are still waiters, other DataVIOs must be trying to get the lock

   * we just transferred. We must ensure that the new lock holder doesn't block

   * in the packer.

   */

  if (hasWaiters(&nextLockHolder->logical.waiters)) {

    cancelCompression(nextLockHolder);

  }

  // Avoid stack overflow on lock transfer.

  // XXX: this is only an issue in the 1 thread config.

  dataVIOAsCompletion(nextLockHolder)->requeue = true;

  launchLockedRequest(nextLockHolder);

}