/*

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

 */

/**

 * HashLock controls and coordinates writing, index access, and dedupe among

 * groups of DataVIOs concurrently writing identical blocks, allowing them to

 * deduplicate not only against advice but also against each other. This save

 * on index queries and allows those DataVIOs to concurrently deduplicate

 * against a single block instead of being serialized through a PBN read lock.

 * Only one index query is needed for each HashLock, instead of one for every

 * DataVIO.

 *

 * A HashLock acts like a state machine perhaps more than as a lock. Other

 * than the starting and ending states INITIALIZING and DESTROYING, every

 * state represents and is held for the duration of an asynchronous operation.

 * All state transitions are performed on the thread of the HashZone

 * containing the lock. An asynchronous operation is almost always performed

 * upon entering a state, and the callback from that operation triggers

 * exiting the state and entering a new state.

 *

 * In all states except DEDUPING, there is a single DataVIO, called the lock

 * agent, performing the asynchronous operations on behalf of the lock. The

 * agent will change during the lifetime of the lock if the lock is shared by

 * more than one DataVIO. DataVIOs waiting to deduplicate are kept on a wait

 * queue. Viewed a different way, the agent holds the lock exclusively until

 * the lock enters the DEDUPING state, at which point it becomes a shared lock

 * that all the waiters (and any new DataVIOs that arrive) use to share a PBN

 * lock. In state DEDUPING, there is no agent. When the last DataVIO in the

 * lock calls back in DEDUPING, it becomes the agent and the lock becomes

 * exclusive again. New DataVIOs that arrive in the lock will also go on the

 * wait queue.

 *

 * The existence of lock waiters is a key factor controlling which state the

 * lock transitions to next. When the lock is new or has waiters, it will

 * always try to reach DEDUPING, and when it doesn't, it will try to clean up

 * and exit.

 *

 * Deduping requires holding a PBN lock on a block that is known to contain

 * data identical to the DataVIOs in the lock, so the lock will send the

 * agent to the duplicate zone to acquire the PBN lock (LOCKING), to the

 * kernel I/O threads to read and verify the data (VERIFYING), or to write a

 * new copy of the data to a full data block or a slot in a compressed block

 * (WRITING).

 *

 * Cleaning up consists of updating the index when the data location is

 * different from the initial index query (UPDATING, triggered by stale

 * advice, compression, and rollover), releasing the PBN lock on the duplicate

 * block (UNLOCKING), and releasing the HashLock itself back to the hash zone

 * (DESTROYING).

 *

 * The shortest sequence of states is for non-concurrent writes of new data:

 *   INITIALIZING -> QUERYING -> WRITING -> DESTROYING

 * This sequence is short because no PBN read lock or index update is needed.

 *

 * Non-concurrent, finding valid advice looks like this (endpoints elided):

 *   -> QUERYING -> LOCKING -> VERIFYING -> DEDUPING -> UNLOCKING ->

 * Or with stale advice (endpoints elided):

 *   -> QUERYING -> LOCKING -> VERIFYING -> UNLOCKING -> WRITING -> UPDATING ->

 *

 * When there are not enough available reference count increments available on

 * a PBN for a DataVIO to deduplicate, a new lock is forked and the excess

 * waiters roll over to the new lock (which goes directly to WRITING). The new

 * lock takes the place of the old lock in the lock map so new DataVIOs will

 * be directed to it. The two locks will proceed independently, but only the

 * new lock will have the right to update the index (unless it also forks).

 *

 * Since rollover happens in a lock instance, once a valid data location has

 * been selected, it will not change. QUERYING and WRITING are only performed

 * once per lock lifetime. All other non-endpoint states can be re-entered.

 *

 * XXX still need doc on BYPASSING

 *

 * The function names in this module follow a convention referencing the

 * states and transitions in the state machine diagram for VDOSTORY-190.

 * [XXX link or repository path to it?]

 * For example, for the LOCKING state, there are startLocking() and

 * finishLocking() functions. startLocking() is invoked by the finish function

 * of the state (or states) that transition to LOCKING. It performs the actual

 * lock state change and must be invoked on the hash zone thread.

 * finishLocking() is called by (or continued via callback from) the code

 * actually obtaining the lock. It does any bookkeeping or decision-making

 * required and invokes the appropriate start function of the state being

 * transitioned to after LOCKING.

 **/

#include "hashLock.h"

#include "hashLockInternals.h"

#include "logger.h"

#include "permassert.h"

#include "compressionState.h"

#include "constants.h"

#include "dataVIO.h"

#include "hashZone.h"

#include "packer.h"

#include "pbnLock.h"

#include "physicalZone.h"

#include "ringNode.h"

#include "slab.h"

#include "slabDepot.h"

#include "trace.h"

#include "types.h"

#include "vdoInternal.h"

#include "vioWrite.h"

#include "waitQueue.h"

static const char *LOCK_STATE_NAMES[] = {

  [HASH_LOCK_BYPASSING]    = "BYPASSING",

  [HASH_LOCK_DEDUPING]     = "DEDUPING",

  [HASH_LOCK_DESTROYING]   = "DESTROYING",

  [HASH_LOCK_INITIALIZING] = "INITIALIZING",

  [HASH_LOCK_LOCKING]      = "LOCKING",

  [HASH_LOCK_QUERYING]     = "QUERYING",

  [HASH_LOCK_UNLOCKING]    = "UNLOCKING",

  [HASH_LOCK_UPDATING]     = "UPDATING",

  [HASH_LOCK_VERIFYING]    = "VERIFYING",

  [HASH_LOCK_WRITING]      = "WRITING",

};

// There are loops in the state diagram, so some forward decl's are needed.

static void startDeduping(HashLock *lock, DataVIO *agent, bool agentIsDone);

static void startLocking(HashLock *lock, DataVIO *agent);

static void startWriting(HashLock *lock, DataVIO *agent);

static void unlockDuplicatePBN(VDOCompletion *completion);

static void transferAllocationLock(DataVIO *dataVIO);

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

PBNLock *getDuplicateLock(DataVIO *dataVIO)

{

  if (dataVIO->hashLock == NULL) {

    return NULL;

  }

  return dataVIO->hashLock->duplicateLock;

}

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

const char *getHashLockStateName(HashLockState state)

{

  // Catch if a state has been added without updating the name array.

  STATIC_ASSERT((HASH_LOCK_DESTROYING + 1) == COUNT_OF(LOCK_STATE_NAMES));

  return (state < COUNT_OF(LOCK_STATE_NAMES)) ? LOCK_STATE_NAMES[state] : NULL;

}

/**

 * Set the current state of a hash lock.

 *

 * @param lock      The lock to update

 * @param newState  The new state

 **/

static void setHashLockState(HashLock *lock, HashLockState newState)

{

  if (false) {

    logWarning("XXX %" PRIptr " %s -> %s", (void *) lock,

               getHashLockStateName(lock->state),

               getHashLockStateName(newState));

  }

  lock->state = newState;

}

/**

 * Assert that a DataVIO is the agent of its hash lock, and that this is being

 * called in the hash zone.

 *

 * @param dataVIO  The DataVIO expected to be the lock agent

 * @param where    A string describing the function making the assertion

 **/

static void assertHashLockAgent(DataVIO *dataVIO, const char *where)

{

  // Not safe to access the agent field except from the hash zone.

  assertInHashZone(dataVIO);

  ASSERT_LOG_ONLY(dataVIO == dataVIO->hashLock->agent,

                  "%s must be for the hash lock agent", where);

}

/**

 * Set or clear the lock agent.

 *

 * @param lock      The hash lock to update

 * @param newAgent  The new lock agent (may be NULL to clear the agent)

 **/

static void setAgent(HashLock *lock, DataVIO *newAgent)

{

  lock->agent = newAgent;

}

/**

 * Set the duplicate lock held by a hash lock. May only be called in the

 * physical zone of the PBN lock.

 *

 * @param hashLock  The hash lock to update

 * @param pbnLock   The PBN read lock to use as the duplicate lock

 **/

static void setDuplicateLock(HashLock *hashLock, PBNLock *pbnLock)

{

  ASSERT_LOG_ONLY((hashLock->duplicateLock == NULL),

                  "hash lock must not already hold a duplicate lock");

  pbnLock->holderCount += 1;

  hashLock->duplicateLock = pbnLock;

}

/**

 * Convert a pointer to the hashLockNode field in a DataVIO to the enclosing

 * DataVIO.

 *

 * @param lockNode The RingNode to convert

 *

 * @return A pointer to the DataVIO containing the RingNode

 **/

static inline DataVIO *dataVIOFromLockNode(RingNode *lockNode)

{

  return (DataVIO *) ((byte *) lockNode - offsetof(DataVIO, hashLockNode));

}

/**

 * Remove the first DataVIO from the lock's wait queue and return it.

 *

 * @param lock  The lock containing the wait queue

 *

 * @return The first (oldest) waiter in the queue, or NULL if

 *         the queue is empty

 **/

static inline DataVIO *dequeueLockWaiter(HashLock *lock)

{

  return waiterAsDataVIO(dequeueNextWaiter(&lock->waiters));

}

/**

 * Continue processing a DataVIO that has been waiting for an event, setting

 * the result from the event, and continuing in a specified callback function.

 *

 * @param dataVIO   The DataVIO to continue

 * @param result    The current result (will not mask older errors)

 * @param callback  The function in which to continue processing

 **/

static void continueDataVIOIn(DataVIO   *dataVIO,

                              int        result,

                              VDOAction *callback)

{

  dataVIOAsCompletion(dataVIO)->callback = callback;

  continueDataVIO(dataVIO, result);

}

/**

 * Set, change, or clear the hash lock a DataVIO is using. Updates the hash

 * lock (or locks) to reflect the change in membership.

 *

 * @param dataVIO  The DataVIO to update

 * @param newLock  The hash lock the DataVIO is joining

 **/

static void setHashLock(DataVIO *dataVIO, HashLock *newLock)

{

  HashLock *oldLock = dataVIO->hashLock;

  if (oldLock != NULL) {

    ASSERT_LOG_ONLY(dataVIO->hashZone != NULL,

                    "must have a hash zone when halding a hash lock");

    ASSERT_LOG_ONLY(!isRingEmpty(&dataVIO->hashLockNode),

                    "must be on a hash lock ring when holding a hash lock");

    ASSERT_LOG_ONLY(oldLock->referenceCount > 0,

                    "hash lock reference must be counted");

    if ((oldLock->state != HASH_LOCK_BYPASSING)

        && (oldLock->state != HASH_LOCK_UNLOCKING)) {

      // If the reference count goes to zero in a non-terminal state, we're

      // most likely leaking this lock.

      ASSERT_LOG_ONLY(oldLock->referenceCount > 1,

                      "hash locks should only become unreferenced in"

                      " a terminal state, not state %s",

                      getHashLockStateName(oldLock->state));

    }

    unspliceRingNode(&dataVIO->hashLockNode);

    oldLock->referenceCount -= 1;

    dataVIO->hashLock = NULL;

  }

  if (newLock != NULL) {

    // Keep all DataVIOs sharing the lock on a ring since they can complete in

    // any order and we'll always need a pointer to one to compare data.

    pushRingNode(&newLock->duplicateRing, &dataVIO->hashLockNode);

    newLock->referenceCount += 1;

    // XXX Not needed for VDOSTORY-190, but useful for checking whether a test

    // is getting concurrent dedupe, and how much.

    if (newLock->maxReferences < newLock->referenceCount) {

      newLock->maxReferences = newLock->referenceCount;

    }

    dataVIO->hashLock = newLock;

  }

}

/**

 * Bottleneck for DataVIOs that have written or deduplicated and that are no

 * longer needed to be an agent for the hash lock.

 *

 * @param dataVIO  The DataVIO to complete and send to be cleaned up

 **/

static void exitHashLock(DataVIO *dataVIO)

{

  // XXX trace record?

  // Release the hash lock now, saving a thread transition in cleanup.

  releaseHashLock(dataVIO);

  // Complete the DataVIO and start the clean-up path in vioWrite to release

  // any locks it still holds.

  finishDataVIO(dataVIO, VDO_SUCCESS);

}

/**

 * Retire the active lock agent, replacing it with the first lock waiter, and

 * make the retired agent exit the hash lock.

 *

 * @param lock  The hash lock to update

 *

 * @return The new lock agent (which will be NULL if there was no waiter)

 **/

static DataVIO *retireLockAgent(HashLock *lock)

{

  DataVIO *oldAgent = lock->agent;

  DataVIO *newAgent = dequeueLockWaiter(lock);

  setAgent(lock, newAgent);

  exitHashLock(oldAgent);

  if (newAgent != NULL) {

    setDuplicateLocation(newAgent, lock->duplicate);

  }

  return newAgent;

}

/**

 * Callback to call compressData(), putting a DataVIO back on the write path.

 *

 * @param completion  The DataVIO

 **/

static void compressDataCallback(VDOCompletion *completion)

{

  // XXX VDOSTORY-190 need an error check since compressData doesn't have one.

  compressData(asDataVIO(completion));

}

/**

 * Add a DataVIO to the lock's queue of waiters.

 *

 * @param lock     The hash lock on which to wait

 * @param dataVIO  The DataVIO to add to the queue

 **/

static void waitOnHashLock(HashLock *lock, DataVIO *dataVIO)

{

  int result = enqueueDataVIO(&lock->waiters, dataVIO, THIS_LOCATION(NULL));

  if (result != VDO_SUCCESS) {

    // This should be impossible, but if it somehow happens, give up on trying

    // to dedupe the data.

    setHashLock(dataVIO, NULL);

    continueDataVIOIn(dataVIO, result, compressDataCallback);

    return;

  }

  // Make sure the agent doesn't block indefinitely in the packer since it now

  // has at least one other DataVIO waiting on it.

  if ((lock->state == HASH_LOCK_WRITING) && cancelCompression(lock->agent)) {

    /*

     * Even though we're waiting, we also have to send ourselves as a one-way

     * message to the packer to ensure the agent continues executing. This is

     * safe because cancelCompression() guarantees the agent won't continue

     * executing until this message arrives in the packer, and because the

     * wait queue link isn't used for sending the message.

     */

    dataVIO->compression.lockHolder = lock->agent;

    launchPackerCallback(dataVIO, removeLockHolderFromPacker,

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

  }

}

/**

 * WaiterCallback function that calls compressData on the DataVIO waiter.

 *

 * @param waiter   The DataVIO's waiter link

 * @param context  Not used

 **/

static void compressWaiter(Waiter *waiter,

                           void   *context __attribute__((unused)))

{

  DataVIO *dataVIO     = waiterAsDataVIO(waiter);

  dataVIO->isDuplicate = false;

  compressData(dataVIO);

}

/**

 * Handle the result of the agent for the lock releasing a read lock on

 * duplicate candidate due to aborting the hash lock. This continuation is

 * registered in unlockDuplicatePBN().

 *

 * @param completion  The completion of the DataVIO acting as the lock's agent

 **/

static void finishBypassing(VDOCompletion *completion)

{

  DataVIO  *agent = asDataVIO(completion);

  assertHashLockAgent(agent, __func__);

  HashLock *lock = agent->hashLock;

  ASSERT_LOG_ONLY(lock->duplicateLock == NULL,

                  "must have released the duplicate lock for the hash lock");

  exitHashLock(agent);

}

/**

 * Stop using the hash lock, resuming the old write path for the lock agent

 * and any DataVIOs waiting on it, and put it in a state where DataVIOs

 * entering the lock will use the old dedupe path instead of waiting.

 *

 * @param lock   The hash lock

 * @param agent  The DataVIO acting as the agent for the lock

 **/

static void startBypassing(HashLock *lock, DataVIO *agent)

{

  setHashLockState(lock, HASH_LOCK_BYPASSING);

  // Ensure we don't attempt to update advice when cleaning up.

  lock->updateAdvice = false;

  ASSERT_LOG_ONLY(((agent != NULL) || !hasWaiters(&lock->waiters)),

                  "should not have waiters without an agent");

  notifyAllWaiters(&lock->waiters, compressWaiter, NULL);

  if (lock->duplicateLock != NULL) {

    if (agent != NULL) {

      // The agent must reference the duplicate zone to launch it.

      agent->duplicate = lock->duplicate;

      launchDuplicateZoneCallback(agent, unlockDuplicatePBN,

                                  THIS_LOCATION(NULL));

      return;

    }

    ASSERT_LOG_ONLY(false, "hash lock holding a PBN lock must have an agent");

  }

  if (agent == NULL) {

    return;

  }

  setAgent(lock, NULL);

  agent->isDuplicate = false;

  compressData(agent);

}

/**

 * Abort processing on this hash lock when noticing an error. Currently, this

 * moves the hash lock to the BYPASSING state, to release all pending DataVIOs.

 *

 * @param lock     The HashLock

 * @param dataVIO  The DataVIO with the error

 **/

static void abortHashLock(HashLock *lock, DataVIO *dataVIO)

{

  // If we've already aborted the lock, don't try to re-abort it; just exit.

  if (lock->state == HASH_LOCK_BYPASSING) {

    exitHashLock(dataVIO);

    return;

  }

  if (dataVIO != lock->agent) {

    if ((lock->agent != NULL) || (lock->referenceCount > 1)) {

      // Other DataVIOs are still sharing the lock (which should be DEDUPING),

      // so just kick this one out of the lock to report its error.

      ASSERT_LOG_ONLY(lock->agent == NULL,

                      "only active agent should call abortHashLock");

      exitHashLock(dataVIO);

      return;

    }

    // Make the lone DataVIO the lock agent so it can abort and clean up.

    setAgent(lock, dataVIO);

  }

  startBypassing(lock, dataVIO);

}

/**

 * Handle the result of the agent for the lock releasing a read lock on

 * duplicate candidate. This continuation is registered in

 * unlockDuplicatePBN().

 *

 * @param completion  The completion of the DataVIO acting as the lock's agent

 **/

static void finishUnlocking(VDOCompletion *completion)

{

  DataVIO  *agent = asDataVIO(completion);

  assertHashLockAgent(agent, __func__);

  HashLock *lock = agent->hashLock;

  ASSERT_LOG_ONLY(lock->duplicateLock == NULL,

                  "must have released the duplicate lock for the hash lock");

  if (completion->result != VDO_SUCCESS) {

    abortHashLock(lock, agent);

    return;

  }

  if (!lock->verified) {

    /*

     * UNLOCKING -> WRITING transition: The lock we released was on an

     * unverified block, so it must have been a lock on advice we were

     * verifying, not on a location that was used for deduplication. Go write

     * (or compress) the block to get a location to dedupe against.

     */

    startWriting(lock, agent);

    return;

  }

  // With the lock released, the verified duplicate block may already have

  // changed and will need to be re-verified if a waiter arrived.

  lock->verified = false;

  if (hasWaiters(&lock->waiters)) {

    /*

     * UNLOCKING -> LOCKING transition: A new DataVIO entered the hash lock

     * while the agent was releasing the PBN lock. The current agent exits and

     * the waiter has to re-lock and re-verify the duplicate location.

     */

    // XXX VDOSTORY-190 If we used the current agent to re-acquire the PBN

    // lock we wouldn't need to re-verify.

    agent = retireLockAgent(lock);

    startLocking(lock, agent);

    return;

  }

  /*

   * UNLOCKING -> DESTROYING transition: The agent is done with the lock

   * and no other DataVIOs reference it, so remove it from the lock map

   * and return it to the pool.

   */

  exitHashLock(agent);

}

/**

 * Release a read lock on the PBN of the block that may or may not have

 * contained duplicate data. This continuation is launched by

 * startUnlocking(), and calls back to finishUnlocking() on the hash zone

 * thread.

 *

 * @param completion  The completion of the DataVIO acting as the lock's agent

 **/

static void unlockDuplicatePBN(VDOCompletion *completion)

{

  DataVIO *agent = asDataVIO(completion);

  assertInDuplicateZone(agent);

  HashLock *lock = agent->hashLock;

  ASSERT_LOG_ONLY(lock->duplicateLock != NULL,

                  "must have a duplicate lock to release");

  releasePBNLock(agent->duplicate.zone, agent->duplicate.pbn,

                 &lock->duplicateLock);

  if (lock->state == HASH_LOCK_BYPASSING) {

    launchHashZoneCallback(agent, finishBypassing, THIS_LOCATION(NULL));

  } else {

    launchHashZoneCallback(agent, finishUnlocking, THIS_LOCATION(NULL));

  }

}

/**

 * Release a read lock on the PBN of the block that may or may not have

 * contained duplicate data.

 *

 * @param lock   The hash lock

 * @param agent  The DataVIO currently acting as the agent for the lock

 **/

static void startUnlocking(HashLock *lock, DataVIO *agent)

{

  setHashLockState(lock, HASH_LOCK_UNLOCKING);

  /*

   * XXX If we arrange to continue on the duplicate zone thread when

   * verification fails, and don't explicitly change lock states (or use an

   * agent-local state, or an atomic), we can avoid a thread transition here.

   */

  launchDuplicateZoneCallback(agent, unlockDuplicatePBN, THIS_LOCATION(NULL));

}

/**

 * Process the result of a UDS update performed by the agent for the lock.

 * This continuation is registered in startQuerying().

 *

 * @param completion  The completion of the DataVIO that performed the update

 **/

static void finishUpdating(VDOCompletion *completion)

{

  DataVIO  *agent = asDataVIO(completion);

  assertHashLockAgent(agent, __func__);

  HashLock *lock = agent->hashLock;

  if (completion->result != VDO_SUCCESS) {

    abortHashLock(lock, agent);

    return;

  }

  // UDS was updated successfully, so don't update again unless the

  // duplicate location changes due to rollover.

  lock->updateAdvice = false;

  if (hasWaiters(&lock->waiters)) {

    /*

     * UPDATING -> DEDUPING transition: A new DataVIO arrived during the UDS

     * update. Send it on the verified dedupe path. The agent is done with the

     * lock, but the lock may still need to use it to clean up after rollover.

     */

    startDeduping(lock, agent, true);

    return;

  }

  if (lock->duplicateLock != NULL) {

    /*

     * UPDATING -> UNLOCKING transition: No one is waiting to dedupe, but we

     * hold a duplicate PBN lock, so go release it.

     */

    startUnlocking(lock, agent);

  } else {

    /*

     * UPDATING -> DESTROYING transition: No one is waiting to dedupe and

     * there's no lock to release.

     */

    // XXX startDestroying(lock, agent);

    startBypassing(lock, NULL);

    exitHashLock(agent);

  }

}

/**

 * Continue deduplication with the last step, updating UDS with the location

 * of the duplicate that should be returned as advice in the future.

 *

 * @param lock   The hash lock

 * @param agent  The DataVIO currently acting as the agent for the lock

 **/

static void startUpdating(HashLock *lock, DataVIO *agent)

{

  setHashLockState(lock, HASH_LOCK_UPDATING);

  ASSERT_LOG_ONLY(lock->verified, "new advice should have been verified");

  ASSERT_LOG_ONLY(lock->updateAdvice, "should only update advice if needed");

  agent->lastAsyncOperation = UPDATE_INDEX;

  setHashZoneCallback(agent, finishUpdating, THIS_LOCATION(NULL));

  dataVIOAsCompletion(agent)->layer->updateAlbireo(agent);

}

/**

 * Handle a DataVIO that has finished deduplicating against the block locked

 * by the hash lock. If there are other DataVIOs still sharing the lock, this

 * will just release the DataVIO's share of the lock and finish processing the

 * DataVIO. If this is the last DataVIO holding the lock, this makes the

 * DataVIO the lock agent and uses it to advance the state of the lock so it

 * can eventually be released.

 *

 * @param lock     The hash lock

 * @param dataVIO  The lock holder that has finished deduplicating

 **/

static void finishDeduping(HashLock *lock, DataVIO *dataVIO)

{

  ASSERT_LOG_ONLY(lock->agent == NULL, "shouldn't have an agent in DEDUPING");

  ASSERT_LOG_ONLY(!hasWaiters(&lock->waiters),

                  "shouldn't have any lock waiters in DEDUPING");

  // Just release the lock reference if other DataVIOs are still deduping.

  if (lock->referenceCount > 1) {

    exitHashLock(dataVIO);

    return;

  }

  // The hash lock must have an agent for all other lock states.

  DataVIO *agent = dataVIO;

  setAgent(lock, agent);

  if (lock->updateAdvice) {

    /*

     * DEDUPING -> UPDATING transition: The location of the duplicate block

     * changed since the initial UDS query because of compression, rollover,

     * or because the query agent didn't have an allocation. The UDS update

     * was delayed in case there was another change in location, but with only

     * this DataVIO using the hash lock, it's time to update the advice.

     */

    startUpdating(lock, agent);

  } else {

    /*

     * DEDUPING -> UNLOCKING transition: Release the PBN read lock on the

     * duplicate location so the hash lock itself can be released (contingent

     * on no new DataVIOs arriving in the lock before the agent returns).

     */

    startUnlocking(lock, agent);

  }

}

/**

 * Implements WaiterCallback. Binds the DataVIO that was waiting to a new hash

 * lock and waits on that lock.

 **/

static void enterForkedLock(Waiter *waiter, void *context)

{

  DataVIO  *dataVIO = waiterAsDataVIO(waiter);

  HashLock *newLock = (HashLock *) context;

  setHashLock(dataVIO, newLock);

  waitOnHashLock(newLock, dataVIO);

}

/**

 * Fork a hash lock because it has run out of increments on the duplicate PBN.

 * Transfers the new agent and any lock waiters to a new hash lock instance

 * which takes the place of the old lock in the lock map. The old lock remains

 * active, but will not update advice.

 *

 * @param oldLock   The hash lock to fork

 * @param newAgent  The DataVIO that will be the agent for the new lock

 **/

static void forkHashLock(HashLock *oldLock, DataVIO *newAgent)

{

  HashLock *newLock;

  int result = acquireHashLockFromZone(newAgent->hashZone,

                                       &newAgent->chunkName,

                                       oldLock, &newLock);

  if (result != VDO_SUCCESS) {

    abortHashLock(oldLock, newAgent);

    return;

  }

  // Only one of the two locks should update UDS. The old lock is out of

  // references, so it would be poor dedupe advice in the short term.

  oldLock->updateAdvice = false;

  newLock->updateAdvice = true;

  setHashLock(newAgent, newLock);

  setAgent(newLock, newAgent);

  notifyAllWaiters(&oldLock->waiters, enterForkedLock, newLock);

  newAgent->isDuplicate = false;

  startWriting(newLock, newAgent);

}

/**

 * Reserve a reference count increment for a DataVIO and launch it on the

 * dedupe path. If no increments are available, this will roll over to a new

 * hash lock and launch the DataVIO as the writing agent for that lock.

 *

 * @param lock      The hash lock

 * @param dataVIO   The DataVIO to deduplicate using the hash lock

 * @param hasClaim  true if the dataVIO already has claimed

 *                  an increment from the duplicate lock

 **/

static void launchDedupe(HashLock *lock, DataVIO *dataVIO, bool hasClaim)

{

  if (!hasClaim && !claimPBNLockIncrement(lock->duplicateLock)) {

    // Out of increments, so must roll over to a new lock.

    forkHashLock(lock, dataVIO);

    return;

  }

  // Deduplicate against the lock's verified location.

  setDuplicateLocation(dataVIO, lock->duplicate);

  launchDuplicateZoneCallback(dataVIO, shareBlock,

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

}

/**

 * Enter the hash lock state where DataVIOs deduplicate in parallel against a

 * true copy of their data on disk. If the agent itself needs to deduplicate,

 * an increment for it must already have been claimed from the duplicate lock,

 * ensuring the hash lock will still have a DataVIO holding it.

 *

 * @param lock         The hash lock

 * @param agent        The DataVIO acting as the agent for the lock

 * @param agentIsDone  true only if the agent has already written

 *                     or deduplicated against its data

 **/

static void startDeduping(HashLock *lock, DataVIO *agent, bool agentIsDone)

{

  setHashLockState(lock, HASH_LOCK_DEDUPING);

  // We don't take the downgraded allocation lock from the agent unless we

  // actually need to deduplicate against it.

  if (lock->duplicateLock == NULL) {

    ASSERT_LOG_ONLY(!isCompressed(agent->newMapped.state),

                    "compression must have shared a lock");

    ASSERT_LOG_ONLY(agentIsDone, "agent must have written the new duplicate");

    transferAllocationLock(agent);

  }

  ASSERT_LOG_ONLY(isPBNReadLock(lock->duplicateLock),

                  "duplicateLock must be a PBN read lock");

  /*

   * This state is not like any of the other states. There is no designated

   * agent--the agent transitioning to this state and all the waiters will be

   * launched to deduplicate in parallel.

   */

  setAgent(lock, NULL);

  /*

   * Launch the agent (if not already deduplicated) and as many lock waiters

   * as we have available increments for on the dedupe path. If we run out of

   * increments, rollover will be triggered and the remaining waiters will be

   * transferred to the new lock.

   */

  if (!agentIsDone) {

    launchDedupe(lock, agent, true);

    agent = NULL;

  }

  while (hasWaiters(&lock->waiters)) {

    launchDedupe(lock, dequeueLockWaiter(lock), false);

  }

  if (agentIsDone) {

    /*

     * In the degenerate case where all the waiters rolled over to a new lock,

     * this will continue to use the old agent to clean up this lock, and

     * otherwise it just lets the agent exit the lock.

     */

    finishDeduping(lock, agent);

  }

}

/**

 * Handle the result of the agent for the lock comparing its data to the

 * duplicate candidate. This continuation is registered in startVerifying().

 *

 * @param completion  The completion of the DataVIO used to verify dedupe

 **/

static void finishVerifying(VDOCompletion *completion)

{

  DataVIO  *agent = asDataVIO(completion);

  assertHashLockAgent(agent, __func__);

  HashLock *lock = agent->hashLock;

  if (completion->result != VDO_SUCCESS) {

    // XXX VDOSTORY-190 should convert verify IO errors to verification failure

    abortHashLock(lock, agent);

    return;

  }

  lock->verified = agent->isDuplicate;

  // Only count the result of the initial verification of the advice as valid

  // or stale, and not any re-verifications due to PBN lock releases.

  if (!lock->verifyCounted) {

    lock->verifyCounted = true;

    if (lock->verified) {

      bumpHashZoneValidAdviceCount(agent->hashZone);

    } else {

      bumpHashZoneStaleAdviceCount(agent->hashZone);

    }

  }

  // Even if the block is a verified duplicate, we can't start to deduplicate

  // unless we can claim a reference count increment for the agent.

  if (lock->verified && !claimPBNLockIncrement(lock->duplicateLock)) {

    agent->isDuplicate = false;

    lock->verified     = false;

  }