/*

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

 */

#include "lockCounter.h"

#include "atomic.h"

#include "memoryAlloc.h"

/**

 * LockCounter is intended to keep all of the locks for the blocks in the

 * recovery journal. The per-zone counters are all kept in a single array which

 * is arranged by zone (i.e. zone 0's lock 0 is at index 0, zone 0's lock 1 is

 * at index 1, and zone 1's lock 0 is at index 'locks'.  This arrangement is

 * intended to minimize cache-line contention for counters from different

 * zones.

 *

 * The locks are implemented as a single object instead of as a lock counter

 * per lock both to afford this opportunity to reduce cache line contention and

 * also to eliminate the need to have a completion per lock.

 *

 * Lock sets are laid out with the set for recovery journal first, followed by

 * the logical zones, and then the physical zones.

 **/

typedef enum lockCounterState {

  LOCK_COUNTER_STATE_NOT_NOTIFYING = 0,

  LOCK_COUNTER_STATE_NOTIFYING,

  LOCK_COUNTER_STATE_SUSPENDED,

} LockCounterState;

struct lockCounter {

  /** The completion for notifying the owner of a lock release */

  VDOCompletion  completion;

  /** The number of logical zones which may hold locks */

  ZoneCount      logicalZones;

  /** The number of physical zones which may hold locks */

  ZoneCount      physicalZones;

  /** The number of locks */

  BlockCount     locks;

  /** Whether the lock release notification is in flight */

  Atomic32       state;

  /** The number of logical zones which hold each lock */

  Atomic32      *logicalZoneCounts;

  /** The number of physical zones which hold each lock */

  Atomic32      *physicalZoneCounts;

  /** The per-zone, per-lock counts for the journal zone */

  uint16_t      *journalCounters;

  /** The per-zone, per-lock decrement counts for the journal zone */

  Atomic32      *journalDecrementCounts;

  /** The per-zone, per-lock reference counts for logical zones */

  uint16_t      *logicalCounters;

  /** The per-zone, per-lock reference counts for physical zones */

  uint16_t      *physicalCounters;

};

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

int makeLockCounter(PhysicalLayer  *layer,

                    void           *parent,

                    VDOAction       callback,

                    ThreadID        threadID,

                    ZoneCount       logicalZones,

                    ZoneCount       physicalZones,

                    BlockCount      locks,

                    LockCounter   **lockCounterPtr)

{

  LockCounter *lockCounter;

  int result = ALLOCATE(1, LockCounter, __func__, &lockCounter);

  if (result != VDO_SUCCESS) {

    return result;

  }

  result = ALLOCATE(locks, uint16_t, __func__, &lockCounter->journalCounters);

  if (result != VDO_SUCCESS) {

    freeLockCounter(&lockCounter);

    return result;

  }

  result = ALLOCATE(locks, Atomic32, __func__,

                    &lockCounter->journalDecrementCounts);

  if (result != VDO_SUCCESS) {

    freeLockCounter(&lockCounter);

    return result;

  }

  result = ALLOCATE(locks * logicalZones, uint16_t, __func__,

                    &lockCounter->logicalCounters);

  if (result != VDO_SUCCESS) {

    freeLockCounter(&lockCounter);

    return result;

  }

  result = ALLOCATE(locks, Atomic32, __func__,

                    &lockCounter->logicalZoneCounts);

  if (result != VDO_SUCCESS) {

    freeLockCounter(&lockCounter);

    return result;

  }

  result = ALLOCATE(locks * physicalZones, uint16_t, __func__,

                    &lockCounter->physicalCounters);

  if (result != VDO_SUCCESS) {

    freeLockCounter(&lockCounter);

    return result;

  }

  result = ALLOCATE(locks, Atomic32, __func__,

                    &lockCounter->physicalZoneCounts);

  if (result != VDO_SUCCESS) {

    freeLockCounter(&lockCounter);

    return result;

  }

  result = initializeEnqueueableCompletion(&lockCounter->completion,

                                           LOCK_COUNTER_COMPLETION, layer);

  if (result != VDO_SUCCESS) {

    freeLockCounter(&lockCounter);

    return result;

  }

  setCallbackWithParent(&lockCounter->completion, callback, threadID, parent);

  lockCounter->logicalZones  = logicalZones;

  lockCounter->physicalZones = physicalZones;

  lockCounter->locks         = locks;

  *lockCounterPtr            = lockCounter;

  return VDO_SUCCESS;

}

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

void freeLockCounter(LockCounter **lockCounterPtr)

{

  if (*lockCounterPtr == NULL) {

    return;

  }

  LockCounter *lockCounter = *lockCounterPtr;

  destroyEnqueueable(&lockCounter->completion);

  freeVolatile(lockCounter->physicalZoneCounts);

  freeVolatile(lockCounter->logicalZoneCounts);

  freeVolatile(lockCounter->journalDecrementCounts);

  FREE(lockCounter->journalCounters);

  FREE(lockCounter->logicalCounters);

  FREE(lockCounter->physicalCounters);

  FREE(lockCounter);

  *lockCounterPtr = NULL;

}

/**

 * Get a pointer to the zone count for a given lock on a given zone.

 *

 * @param counter     The lock counter

 * @param lockNumber  The lock to get

 * @param zoneType    The zone type whose count is desired

 *

 * @return A pointer to the zone count for the given lock and zone

 **/

static inline Atomic32 *getZoneCountPtr(LockCounter *counter,

                                        BlockCount   lockNumber,

                                        ZoneType     zoneType)

{

  return ((zoneType == ZONE_TYPE_LOGICAL)

          ? &counter->logicalZoneCounts[lockNumber]

          : &counter->physicalZoneCounts[lockNumber]);

}

/**

 * Get the zone counter for a given lock on a given zone.

 *

 * @param counter     The lock counter

 * @param lockNumber  The lock to get

 * @param zoneType    The zone type whose count is desired

 * @param zoneID      The zone index whose count is desired

 *

 * @return The counter for the given lock and zone

 **/

static inline uint16_t *getCounter(LockCounter *counter,

                                   BlockCount   lockNumber,

                                   ZoneType     zoneType,

                                   ZoneCount    zoneID)

{

  BlockCount zoneCounter = (counter->locks * zoneID) + lockNumber;

  if (zoneType == ZONE_TYPE_JOURNAL) {

    return &counter->journalCounters[zoneCounter];

  }

  if (zoneType == ZONE_TYPE_LOGICAL) {

    return &counter->logicalCounters[zoneCounter];

  }

  return &counter->physicalCounters[zoneCounter];

}

/**

 * Check whether the journal zone is locked for a given lock.

 *

 * @param counter     The LockCounter

 * @param lockNumber  The lock to check

 *

 * @return true if the journal zone is locked

 **/

static bool isJournalZoneLocked(LockCounter *counter, BlockCount lockNumber)

{

  uint16_t journalValue

    = *(getCounter(counter, lockNumber, ZONE_TYPE_JOURNAL, 0));

  uint32_t decrements

    = atomicLoad32(&(counter->journalDecrementCounts[lockNumber]));

  ASSERT_LOG_ONLY((decrements <= journalValue),

                  "journal zone lock counter must not underflow");

  return (journalValue != decrements);

}

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

bool isLocked(LockCounter *lockCounter,

              BlockCount   lockNumber,

              ZoneType     zoneType)

{

  ASSERT_LOG_ONLY((zoneType != ZONE_TYPE_JOURNAL),

                  "isLocked() called for non-journal zone");

  return (isJournalZoneLocked(lockCounter, lockNumber)

          || (atomicLoad32(getZoneCountPtr(lockCounter, lockNumber, zoneType))

              != 0));

}

/**

 * Check that we are on the journal thread.

 *

 * @param counter  The LockCounter

 * @param caller   The name of the caller (for logging)

 **/

static void assertOnJournalThread(LockCounter *counter, const char *caller)

{

  ASSERT_LOG_ONLY((getCallbackThreadID()

                   == counter->completion.callbackThreadID),

                  "%s() called from journal zone", caller);

}

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

void initializeLockCount(LockCounter *counter,

                         BlockCount   lockNumber,

                         uint16_t     value)

{

  assertOnJournalThread(counter, __func__);

  uint16_t *journalValue   = getCounter(counter, lockNumber, ZONE_TYPE_JOURNAL,

                                        0);

  Atomic32 *decrementCount = &(counter->journalDecrementCounts[lockNumber]);

  ASSERT_LOG_ONLY((*journalValue == atomicLoad32(decrementCount)),

                  "count to be initialized not in use");

  *journalValue = value;

  atomicStore32(decrementCount, 0);

}

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

void acquireLockCountReference(LockCounter *counter,

                               BlockCount   lockNumber,

                               ZoneType     zoneType,

                               ZoneCount    zoneID)

{

  ASSERT_LOG_ONLY((zoneType != ZONE_TYPE_JOURNAL),

                  "invalid lock count increment from journal zone");

  uint16_t *currentValue = getCounter(counter, lockNumber, zoneType, zoneID);

  ASSERT_LOG_ONLY(*currentValue < UINT16_MAX,

                  "increment of lock counter must not overflow");

  if (*currentValue == 0) {

    // This zone is acquiring this lock for the first time.

    atomicAdd32(getZoneCountPtr(counter, lockNumber, zoneType), 1);

  }

  *currentValue += 1;

}

/**

 * Decrement a non-atomic counter.

 *

 * @param counter     The LockCounter

 * @param lockNumber  Which lock to decrement

 * @param zoneType    The type of the zone releasing the reference

 * @param zoneID      The ID of the zone releasing the reference

 *

 * @return The new value of the counter

 **/

static uint16_t releaseReference(LockCounter *counter,

                                 BlockCount   lockNumber,

                                 ZoneType     zoneType,

                                 ZoneCount    zoneID)

{

  uint16_t *currentValue = getCounter(counter, lockNumber, zoneType, zoneID);

  ASSERT_LOG_ONLY((*currentValue >= 1),

                  "decrement of lock counter must not underflow");

  *currentValue -= 1;

  return *currentValue;

}

/**

 * Attempt to notify the owner of this LockCounter that some lock has been

 * released for some zone type. Will do nothing if another notification is

 * already in progress.

 *

 * @param counter  The LockCounter

 **/

static void attemptNotification(LockCounter *counter)

{

  if (compareAndSwap32(&counter->state,

                       LOCK_COUNTER_STATE_NOT_NOTIFYING,

                       LOCK_COUNTER_STATE_NOTIFYING)) {

    resetCompletion(&counter->completion);

    invokeCallback(&counter->completion);

  }

}

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

void releaseLockCountReference(LockCounter *counter,

                               BlockCount   lockNumber,

                               ZoneType     zoneType,

                               ZoneCount    zoneID)

{

  ASSERT_LOG_ONLY((zoneType != ZONE_TYPE_JOURNAL),

                  "invalid lock count decrement from journal zone");

  if (releaseReference(counter, lockNumber, zoneType, zoneID) != 0) {

    return;

  }

  if (atomicAdd32(getZoneCountPtr(counter, lockNumber, zoneType), -1) == 0) {

    // This zone was the last lock holder of its type, so try to notify the

    // owner.

    attemptNotification(counter);

  }

}

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

void releaseJournalZoneReference(LockCounter *counter, BlockCount lockNumber)

{

  assertOnJournalThread(counter, __func__);

  releaseReference(counter, lockNumber, ZONE_TYPE_JOURNAL, 0);

  if (!isJournalZoneLocked(counter, lockNumber)) {

    // The journal zone is not locked, so try to notify the owner.

    attemptNotification(counter);

  }

}

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

void releaseJournalZoneReferenceFromOtherZone(LockCounter *counter,

                                              BlockCount   lockNumber)

{

  atomicAdd32(&(counter->journalDecrementCounts[lockNumber]), 1);

}

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

void acknowledgeUnlock(LockCounter *counter)

{

  atomicStore32(&counter->state, LOCK_COUNTER_STATE_NOT_NOTIFYING);

}

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

bool suspendLockCounter(LockCounter *counter)

{

  assertOnJournalThread(counter, __func__);

  return ((atomicLoad32(&counter->state) == LOCK_COUNTER_STATE_SUSPENDED)

          || compareAndSwap32(&counter->state,

                              LOCK_COUNTER_STATE_NOT_NOTIFYING,

                              LOCK_COUNTER_STATE_SUSPENDED));

}

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

bool resumeLockCounter(LockCounter *counter)

{

  assertOnJournalThread(counter, __func__);

  return compareAndSwap32(&counter->state,

                          LOCK_COUNTER_STATE_SUSPENDED,

                          LOCK_COUNTER_STATE_NOT_NOTIFYING);

}