/*

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

 */

#include "hashZone.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "numeric.h"

#include "permassert.h"

#include "constants.h"

#include "dataVIO.h"

#include "hashLock.h"

#include "hashLockInternals.h"

#include "pointerMap.h"

#include "ringNode.h"

#include "statistics.h"

#include "threadConfig.h"

#include "types.h"

#include "vdoInternal.h"

enum {

  LOCK_POOL_CAPACITY = MAXIMUM_USER_VIOS,

};

/**

 * These fields are only modified by the locks sharing the hash zone thread,

 * but are queried by other threads.

 **/

typedef struct atomicHashLockStatistics {

  /** Number of times the UDS advice proved correct */

  Atomic64 dedupeAdviceValid;

  /** Number of times the UDS advice proved incorrect */

  Atomic64 dedupeAdviceStale;

  /** Number of writes with the same data as another in-flight write */

  Atomic64 concurrentDataMatches;

  /** Number of writes whose hash collided with an in-flight write */

  Atomic64 concurrentHashCollisions;

} AtomicHashLockStatistics;

struct hashZone {

  /** Which hash zone this is */

  ZoneCount zoneNumber;

  /** The thread ID for this zone */

  ThreadID threadID;

  /** Mapping from chunkName fields to HashLocks */

  PointerMap *hashLockMap;

  /** Ring containing all unused HashLocks */

  RingNode lockPool;

  /** Statistics shared by all hash locks in this zone */

  AtomicHashLockStatistics statistics;

  /** Array of all HashLocks */

  HashLock *lockArray;

};

/**

 * Implements PointerKeyComparator.

 **/

static bool compareKeys(const void *thisKey, const void *thatKey)

{

  // Null keys are not supported.

  return (memcmp(thisKey, thatKey, sizeof(UdsChunkName)) == 0);

}

/**

 * Implements PointerKeyComparator.

 **/

static uint32_t hashKey(const void *key)

{

  const UdsChunkName *name = key;

  /*

   * Use a fragment of the chunk name as a hash code. It must not overlap with

   * fragments used elsewhere to ensure uniform distributions.

   */

  // XXX pick an offset in the chunk name that isn't used elsewhere

  return getUInt32LE(&name->name[4]);

}

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

static inline HashLock *asHashLock(RingNode *poolNode)

{

  STATIC_ASSERT(offsetof(HashLock, poolNode) == 0);

  return (HashLock *) poolNode;

}

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

int makeHashZone(VDO *vdo, ZoneCount zoneNumber, HashZone **zonePtr)

{

  HashZone *zone;

  int result = ALLOCATE(1, HashZone, __func__, &zone);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = makePointerMap(LOCK_MAP_CAPACITY, 0, compareKeys, hashKey,

                          &zone->hashLockMap);

  if (result != VDO_SUCCESS) {

    freeHashZone(&zone);

    return result;

  }

  zone->zoneNumber = zoneNumber;

  zone->threadID   = getHashZoneThread(getThreadConfig(vdo), zoneNumber);

  initializeRing(&zone->lockPool);

  result = ALLOCATE(LOCK_POOL_CAPACITY, HashLock, "HashLock array",

                    &zone->lockArray);

  if (result != VDO_SUCCESS) {

    freeHashZone(&zone);

    return result;

  }

  for (VIOCount i = 0; i < LOCK_POOL_CAPACITY; i++) {

    HashLock *lock = &zone->lockArray[i];

    initializeHashLock(lock);

    pushRingNode(&zone->lockPool, &lock->poolNode);

  }

  *zonePtr = zone;

  return VDO_SUCCESS;

}

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

void freeHashZone(HashZone **zonePtr)

{

  if (*zonePtr == NULL) {

    return;

  }

  HashZone *zone = *zonePtr;

  freePointerMap(&zone->hashLockMap);

  FREE(zone->lockArray);

  FREE(zone);

  *zonePtr = NULL;

}

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

ZoneCount getHashZoneNumber(const HashZone *zone)

{

  return zone->zoneNumber;

}

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

ThreadID getHashZoneThreadID(const HashZone *zone)

{

  return zone->threadID;

}

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

HashLockStatistics getHashZoneStatistics(const HashZone *zone)

{

  const AtomicHashLockStatistics *atoms = &zone->statistics;

  return (HashLockStatistics) {

    .dedupeAdviceValid     = relaxedLoad64(&atoms->dedupeAdviceValid),

    .dedupeAdviceStale     = relaxedLoad64(&atoms->dedupeAdviceStale),

    .concurrentDataMatches = relaxedLoad64(&atoms->concurrentDataMatches),

    .concurrentHashCollisions

      = relaxedLoad64(&atoms->concurrentHashCollisions),

  };

}

/**

 * Return a hash lock to the zone's pool and null out the reference to it.

 *

 * @param [in]     zone     The zone from which the lock was borrowed

 * @param [in,out] lockPtr  The last reference to the lock being returned

 **/

static void returnHashLockToPool(HashZone *zone, HashLock **lockPtr)

{

  HashLock *lock = *lockPtr;

  *lockPtr = NULL;

  memset(lock, 0, sizeof(*lock));

  initializeHashLock(lock);

  pushRingNode(&zone->lockPool, &lock->poolNode);

}

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

int acquireHashLockFromZone(HashZone            *zone,

                            const UdsChunkName  *hash,

                            HashLock            *replaceLock,

                            HashLock           **lockPtr)

{

  // Borrow and prepare a lock from the pool so we don't have to do two

  // PointerMap accesses in the common case of no lock contention.

  HashLock *newLock = asHashLock(popRingNode(&zone->lockPool));

  int result = ASSERT(newLock != NULL,

                      "never need to wait for a free hash lock");

  if (result != VDO_SUCCESS) {

    return result;

  }

  // Fill in the hash of the new lock so we can map it, since we have to use

  // the hash as the map key.

  newLock->hash = *hash;

  HashLock *lock;

  result = pointerMapPut(zone->hashLockMap, &newLock->hash, newLock,

                         (replaceLock != NULL), (void **) &lock);

  if (result != VDO_SUCCESS) {

    returnHashLockToPool(zone, &newLock);

    return result;

  }

  if (replaceLock != NULL) {

    // XXX on mismatch put the old lock back and return a severe error

    ASSERT_LOG_ONLY(lock == replaceLock,

                    "old lock must have been in the lock map");

    // XXX check earlier and bail out?

    ASSERT_LOG_ONLY(replaceLock->registered,

                    "old lock must have been marked registered");

    replaceLock->registered = false;

  }

  if (lock == replaceLock) {

    lock = newLock;

    lock->registered = true;

  } else {

    // There's already a lock for the hash, so we don't need the borrowed lock.

    returnHashLockToPool(zone, &newLock);

  }

  *lockPtr = lock;

  return VDO_SUCCESS;

}

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

void returnHashLockToZone(HashZone *zone, HashLock **lockPtr)

{

  HashLock *lock = *lockPtr;

  *lockPtr = NULL;

  if (lock->registered) {

    HashLock *removed = pointerMapRemove(zone->hashLockMap, &lock->hash);

    ASSERT_LOG_ONLY(lock == removed,

                    "hash lock being released must have been mapped");

  } else {

    ASSERT_LOG_ONLY(lock != pointerMapGet(zone->hashLockMap, &lock->hash),

                    "unregistered hash lock must not be in the lock map");

  }

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

                  "hash lock returned to zone must have no waiters");

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

                  "hash lock returned to zone must not reference a PBN lock");

  ASSERT_LOG_ONLY((lock->state == HASH_LOCK_DESTROYING),

                  "returned hash lock must not be in use with state %s",

                  getHashLockStateName(lock->state));

  ASSERT_LOG_ONLY(isRingEmpty(&lock->poolNode),

                  "hash lock returned to zone must not be in a pool ring");

  ASSERT_LOG_ONLY(isRingEmpty(&lock->duplicateRing),

                  "hash lock returned to zone must not reference DataVIOs");

  returnHashLockToPool(zone, &lock);

}

/**

 * Dump a compact description of HashLock to the log if the lock is not on the

 * free list.

 *

 * @param lock  The hash lock to dump

 **/

static void dumpHashLock(const HashLock *lock)

{

  if (!isRingEmpty(&lock->poolNode)) {

    // This lock is on the free list.

    return;

  }

  // Necessarily cryptic since we can log a lot of these. First three chars of

  // state is unambiguous. 'U' indicates a lock not registered in the map.

  const char *state = getHashLockStateName(lock->state);

  logInfo("  hl %" PRIptr ": %3.3s %c%llu/%u rc=%u wc=%zu agt=%" PRIptr,

          (const void *) lock,

          state,

          (lock->registered ? 'D' : 'U'),

          lock->duplicate.pbn,

          lock->duplicate.state,

          lock->referenceCount,

          countWaiters(&lock->waiters),

          (void *) lock->agent);

}

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

void bumpHashZoneValidAdviceCount(HashZone *zone)

{

  // Must only be mutated on the hash zone thread.

  relaxedAdd64(&zone->statistics.dedupeAdviceValid, 1);

}

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

void bumpHashZoneStaleAdviceCount(HashZone *zone)

{

  // Must only be mutated on the hash zone thread.

  relaxedAdd64(&zone->statistics.dedupeAdviceStale, 1);

}

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

void bumpHashZoneDataMatchCount(HashZone *zone)

{

  // Must only be mutated on the hash zone thread.

  relaxedAdd64(&zone->statistics.concurrentDataMatches, 1);

}

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

void bumpHashZoneCollisionCount(HashZone *zone)

{

  // Must only be mutated on the hash zone thread.

  relaxedAdd64(&zone->statistics.concurrentHashCollisions, 1);

}

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

void dumpHashZone(const HashZone *zone)

{

  if (zone->hashLockMap == NULL) {

    logInfo("HashZone %u: NULL map", zone->zoneNumber);

    return;

  }

  logInfo("HashZone %u: mapSize=%zu",

          zone->zoneNumber, pointerMapSize(zone->hashLockMap));

  for (VIOCount i = 0; i < LOCK_POOL_CAPACITY; i++) {

    dumpHashLock(&zone->lockArray[i]);

  }

}