/*

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

 */

#include "refCounts.h"

#include "refCountsInternals.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "numeric.h"

#include "permassert.h"

#include "adminState.h"

#include "blockAllocatorInternals.h"

#include "completion.h"

#include "extent.h"

#include "header.h"

#include "journalPoint.h"

#include "numUtils.h"

#include "pbnLock.h"

#include "readOnlyNotifier.h"

#include "referenceBlock.h"

#include "referenceOperation.h"

#include "slab.h"

#include "slabJournal.h"

#include "slabJournalInternals.h"

#include "slabSummary.h"

#include "statusCodes.h"

#include "stringUtils.h"

#include "vdo.h"

#include "vioPool.h"

#include "waitQueue.h"

static const uint64_t BYTES_PER_WORD   = sizeof(uint64_t);

static const bool     NORMAL_OPERATION = true;

/**

 * Return the RefCounts from the RefCounts waiter.

 *

 * @param waiter  The waiter to convert

 *

 * @return  The RefCounts

 **/

__attribute__((warn_unused_result))

static inline RefCounts *refCountsFromWaiter(Waiter *waiter)

{

  if (waiter == NULL) {

    return NULL;

  }

  return (RefCounts *)

    ((uintptr_t) waiter - offsetof(RefCounts, slabSummaryWaiter));

}

/**

 * Convert the index of a reference counter back to the block number of the

 * physical block for which it is counting references. The index is assumed to

 * be valid and in-range.

 *

 * @param refCounts  The reference counts object

 * @param index      The array index of the reference counter

 *

 * @return the physical block number corresponding to the index

 **/

static PhysicalBlockNumber indexToPBN(const RefCounts *refCounts,

                                      uint64_t         index)

{

  return (refCounts->slab->start + index);

}

/**

 * Convert a block number to the index of a reference counter for that block.

 * Out of range values are pinned to the beginning or one past the end of the

 * array.

 *

 * @param refCounts  The reference counts object

 * @param pbn        The physical block number

 *

 * @return the index corresponding to the physical block number

 **/

static uint64_t pbnToIndex(const RefCounts *refCounts, PhysicalBlockNumber pbn)

{

  if (pbn < refCounts->slab->start) {

    return 0;

  }

  uint64_t index = (pbn - refCounts->slab->start);

  return minBlock(index, refCounts->blockCount);

}

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

ReferenceStatus referenceCountToStatus(ReferenceCount count)

{

  if (count == EMPTY_REFERENCE_COUNT) {

    return RS_FREE;

  } else if (count == 1) {

    return RS_SINGLE;

  } else if (count == PROVISIONAL_REFERENCE_COUNT) {

    return RS_PROVISIONAL;

  } else {

    return RS_SHARED;

  }

}

/**

 * Reset the free block search back to the first reference counter

 * in the first reference block.

 *

 * @param refCounts  The RefCounts object containing the search cursor

 **/

static void resetSearchCursor(RefCounts *refCounts)

{

  SearchCursor *cursor = &refCounts->searchCursor;

  cursor->block    = cursor->firstBlock;

  cursor->index    = 0;

  // Unit tests have slabs with only one reference block (and it's a runt).

  cursor->endIndex = minBlock(COUNTS_PER_BLOCK, refCounts->blockCount);

}

/**

 * Advance the search cursor to the start of the next reference block,

 * wrapping around to the first reference block if the current block is the

 * last reference block.

 *

 * @param refCounts  The RefCounts object containing the search cursor

 *

 * @return true unless the cursor was at the last reference block

 **/

static bool advanceSearchCursor(RefCounts *refCounts)

{

  SearchCursor *cursor = &refCounts->searchCursor;

  // If we just finished searching the last reference block, then wrap back

  // around to the start of the array.

  if (cursor->block == cursor->lastBlock) {

    resetSearchCursor(refCounts);

    return false;

  }

  // We're not already at the end, so advance to cursor to the next block.

  cursor->block++;

  cursor->index = cursor->endIndex;

  if (cursor->block == cursor->lastBlock) {

    // The last reference block will usually be a runt.

    cursor->endIndex = refCounts->blockCount;

  } else {

    cursor->endIndex += COUNTS_PER_BLOCK;

  }

  return true;

}

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

int makeRefCounts(BlockCount            blockCount,

                  Slab                 *slab,

                  PhysicalBlockNumber   origin,

                  ReadOnlyNotifier     *readOnlyNotifier,

                  RefCounts           **refCountsPtr)

{

  BlockCount  refBlockCount = getSavedReferenceCountSize(blockCount);

  RefCounts  *refCounts;

  int result = ALLOCATE_EXTENDED(RefCounts, refBlockCount, ReferenceBlock,

                                 "ref counts structure", &refCounts);

  if (result != UDS_SUCCESS) {

    return result;

  }

  // Allocate such that the runt slab has a full-length memory array,

  // plus a little padding so we can word-search even at the very end.

  size_t bytes = ((refBlockCount * COUNTS_PER_BLOCK) + (2 * BYTES_PER_WORD));

  result = ALLOCATE(bytes, ReferenceCount, "ref counts array",

                    &refCounts->counters);

  if (result != UDS_SUCCESS) {

    freeRefCounts(&refCounts);

    return result;

  }

  refCounts->slab                    = slab;

  refCounts->blockCount              = blockCount;

  refCounts->freeBlocks              = blockCount;

  refCounts->origin                  = origin;

  refCounts->referenceBlockCount     = refBlockCount;

  refCounts->readOnlyNotifier        = readOnlyNotifier;

  refCounts->statistics              = &slab->allocator->refCountStatistics;

  refCounts->searchCursor.firstBlock = &refCounts->blocks[0];

  refCounts->searchCursor.lastBlock  = &refCounts->blocks[refBlockCount - 1];

  resetSearchCursor(refCounts);

  for (size_t index = 0; index < refBlockCount; index++) {

    refCounts->blocks[index] = (ReferenceBlock) {

      .refCounts = refCounts,

    };

  }

  *refCountsPtr = refCounts;

  return VDO_SUCCESS;

}

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

void freeRefCounts(RefCounts **refCountsPtr)

{

  RefCounts *refCounts = *refCountsPtr;

  if (refCounts == NULL) {

    return;

  }

  FREE(refCounts->counters);

  FREE(refCounts);

  *refCountsPtr = NULL;

}

/**

 * Check whether a RefCounts has active I/O.

 *

 * @param refCounts  The RefCounts to check

 *

 * @return true if there is reference block I/O or a summary

 *         update in progress

 **/

__attribute__((warn_unused_result))

static bool hasActiveIO(RefCounts *refCounts)

{

  return ((refCounts->activeCount > 0) || refCounts->updatingSlabSummary);

}

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

bool areRefCountsActive(RefCounts *refCounts)

{

  if (hasActiveIO(refCounts)) {

    return true;

  }

    // When not suspending or recovering, the refCounts must be clean.

  AdminStateCode code = refCounts->slab->state.state;

  return (hasWaiters(&refCounts->dirtyBlocks)

          && (code != ADMIN_STATE_SUSPENDING)

          && (code != ADMIN_STATE_RECOVERING));

}

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

static void enterRefCountsReadOnlyMode(RefCounts *refCounts, int result)

{

  enterReadOnlyMode(refCounts->readOnlyNotifier, result);

  checkIfSlabDrained(refCounts->slab);

}

/**

 * Enqueue a block on the dirty queue.

 *

 * @param block  The block to enqueue

 **/

static void enqueueDirtyBlock(ReferenceBlock *block)

{

  int result = enqueueWaiter(&block->refCounts->dirtyBlocks, &block->waiter);

  if (result != VDO_SUCCESS) {

    // This should never happen.

    enterRefCountsReadOnlyMode(block->refCounts, result);

  }

}

/**

 * Mark a reference count block as dirty, potentially adding it to the dirty

 * queue if it wasn't already dirty.

 *

 * @param block  The reference block to mark as dirty

 **/

static void dirtyBlock(ReferenceBlock *block)

{

  if (block->isDirty) {

    return;

  }

  block->isDirty = true;

  if (block->isWriting) {

    // The conclusion of the current write will enqueue the block again.

    return;

  }

  enqueueDirtyBlock(block);

}

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

BlockCount getUnreferencedBlockCount(RefCounts *refCounts)

{

  return refCounts->freeBlocks;

}

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

ReferenceBlock *getReferenceBlock(RefCounts *refCounts, SlabBlockNumber index)

{

  return &refCounts->blocks[index / COUNTS_PER_BLOCK];

}

/**

 * Get the reference counter that covers the given physical block number.

 *

 * @param [in]  refCounts       The refcounts object

 * @param [in]  pbn             The physical block number

 * @param [out] counterPtr      A pointer to the reference counter

 **/

static int getReferenceCounter(RefCounts            *refCounts,

                               PhysicalBlockNumber   pbn,

                               ReferenceCount      **counterPtr)

{

  SlabBlockNumber index;

  int result = slabBlockNumberFromPBN(refCounts->slab, pbn, &index);

  if (result != VDO_SUCCESS) {

    return result;

  }

  *counterPtr = &refCounts->counters[index];

  return VDO_SUCCESS;

}

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

uint8_t getAvailableReferences(RefCounts *refCounts, PhysicalBlockNumber pbn)

{

  ReferenceCount *counterPtr = NULL;

  int result = getReferenceCounter(refCounts, pbn, &counterPtr);

  if (result != VDO_SUCCESS) {

    return 0;

  }

  if (*counterPtr == PROVISIONAL_REFERENCE_COUNT) {

    return (MAXIMUM_REFERENCE_COUNT - 1);

  }

  return (MAXIMUM_REFERENCE_COUNT - *counterPtr);

}

/**

 * Increment the reference count for a data block.

 *

 * @param [in]     refCounts          The refCounts responsible for the block

 * @param [in]     block              The reference block which contains the

 *                                    block being updated

 * @param [in]     slabBlockNumber    The block to update

 * @param [in]     oldStatus          The reference status of the data block

 *                                    before this increment

 * @param [in]     lock               The PBNLock associated with this

 *                                    increment (may be NULL)

 * @param [in,out] counterPtr         A pointer to the count for the data block

 * @param [out]    freeStatusChanged  A pointer which will be set to true if

 *                                    this update changed the free status of

 *                                    the block

 *

 * @return VDO_SUCCESS or an error

 **/

static int incrementForData(RefCounts       *refCounts,

                            ReferenceBlock  *block,

                            SlabBlockNumber  slabBlockNumber,

                            ReferenceStatus  oldStatus,

                            PBNLock         *lock,

                            ReferenceCount  *counterPtr,

                            bool            *freeStatusChanged)

{

  switch (oldStatus) {

  case RS_FREE:

    *counterPtr = 1;

    block->allocatedCount++;

    refCounts->freeBlocks--;

    *freeStatusChanged = true;

    break;

  case RS_PROVISIONAL:

    *counterPtr        = 1;

    *freeStatusChanged = false;

    break;

  default:

    // Single or shared

    if (*counterPtr >= MAXIMUM_REFERENCE_COUNT) {

      return logErrorWithStringError(VDO_REF_COUNT_INVALID,

                                     "Incrementing a block already having"

                                     " 254 references (slab %u, offset %"

                                     PRIu32 ")",

                                     refCounts->slab->slabNumber,

                                     slabBlockNumber);

    }

    (*counterPtr)++;

    *freeStatusChanged = false;

  }

  if (lock != NULL) {

    unassignProvisionalReference(lock);

  }

  return VDO_SUCCESS;

}

/**

 * Decrement the reference count for a data block.

 *

 * @param [in]     refCounts          The refCounts responsible for the block

 * @param [in]     block              The reference block which contains the

 *                                    block being updated

 * @param [in]     slabBlockNumber    The block to update

 * @param [in]     oldStatus          The reference status of the data block

 *                                    before this decrement

 * @param [in]     lock               The PBNLock associated with the block

 *                                    being decremented (may be NULL)

 * @param [in,out] counterPtr         A pointer to the count for the data block

 * @param [out]    freeStatusChanged  A pointer which will be set to true if

 *                                    this update changed the free status of

 *                                    the block

 *

 * @return VDO_SUCCESS or an error

 **/

static int decrementForData(RefCounts       *refCounts,

                            ReferenceBlock  *block,

                            SlabBlockNumber  slabBlockNumber,

                            ReferenceStatus  oldStatus,

                            PBNLock         *lock,

                            ReferenceCount  *counterPtr,

                            bool            *freeStatusChanged)

{

  switch (oldStatus) {

  case RS_FREE:

    return logErrorWithStringError(VDO_REF_COUNT_INVALID,

                                   "Decrementing free block at offset %"

                                   PRIu32 " in slab %u", slabBlockNumber,

                                   refCounts->slab->slabNumber);

  case RS_PROVISIONAL:

  case RS_SINGLE:

    if (lock != NULL) {

      // There is a read lock on this block, so the block must not become

      // unreferenced.

      *counterPtr        = PROVISIONAL_REFERENCE_COUNT;

      *freeStatusChanged = false;

      assignProvisionalReference(lock);

    } else {

      *counterPtr = EMPTY_REFERENCE_COUNT;

      block->allocatedCount--;

      refCounts->freeBlocks++;

      *freeStatusChanged = true;

    }

    break;

  default:

    // Shared

    (*counterPtr)--;

    *freeStatusChanged = false;

  }

  return VDO_SUCCESS;

}

/**

 * Increment the reference count for a block map page. All block map increments

 * should be from provisional to MAXIMUM_REFERENCE_COUNT. Since block map blocks

 * never dedupe they should never be adjusted from any other state. The

 * adjustment always results in MAXIMUM_REFERENCE_COUNT as this value is used to

 * prevent dedupe against block map blocks.

 *

 * @param [in]     refCounts          The refCounts responsible for the block

 * @param [in]     block              The reference block which contains the

 *                                    block being updated

 * @param [in]     slabBlockNumber    The block to update

 * @param [in]     oldStatus          The reference status of the block

 *                                    before this increment

 * @param [in]     lock               The PBNLock associated with this

 *                                    increment (may be NULL)

 * @param [in]     normalOperation    Whether we are in normal operation vs.

 *                                    recovery or rebuild

 * @param [in,out] counterPtr         A pointer to the count for the block

 * @param [out]    freeStatusChanged  A pointer which will be set to true if

 *                                    this update changed the free status of the

 *                                    block

 *

 * @return VDO_SUCCESS or an error

 **/

static int incrementForBlockMap(RefCounts       *refCounts,

                                ReferenceBlock  *block,

                                SlabBlockNumber  slabBlockNumber,

                                ReferenceStatus  oldStatus,

                                PBNLock         *lock,

                                bool             normalOperation,

                                ReferenceCount  *counterPtr,

                                bool            *freeStatusChanged)

{

  switch (oldStatus) {

  case RS_FREE:

    if (normalOperation) {

      return logErrorWithStringError(VDO_REF_COUNT_INVALID,

                                     "Incrementing unallocated block map block"

                                     " (slab %u, offset %" PRIu32 ")",

                                     refCounts->slab->slabNumber,

                                     slabBlockNumber);

    }

    *counterPtr = MAXIMUM_REFERENCE_COUNT;

    block->allocatedCount++;

    refCounts->freeBlocks--;

    *freeStatusChanged = true;

    return VDO_SUCCESS;

  case RS_PROVISIONAL:

    if (!normalOperation) {

      return logErrorWithStringError(VDO_REF_COUNT_INVALID,

                                     "Block map block had provisional "

                                     "reference during replay"

                                     " (slab %u, offset %" PRIu32 ")",

                                     refCounts->slab->slabNumber,

                                     slabBlockNumber);

    }

    *counterPtr        = MAXIMUM_REFERENCE_COUNT;

    *freeStatusChanged = false;

    if (lock != NULL) {

      unassignProvisionalReference(lock);

    }

    return VDO_SUCCESS;

  default:

    return logErrorWithStringError(VDO_REF_COUNT_INVALID,

                                   "Incrementing a block map block which is "

                                   "already referenced %u times (slab %u, "

                                   "offset %" PRIu32 ")",

                                   *counterPtr,

                                   refCounts->slab->slabNumber,

                                   slabBlockNumber);

  }

}

/**

 * Update the reference count of a block.

 *

 * @param [in]  refCounts                The refCounts responsible for the

 *                                       block

 * @param [in]  block                    The reference block which contains the

 *                                       block being updated

 * @param [in]  slabBlockNumber          The block to update

 * @param [in]  slabJournalPoint         The slab journal point at which this

 *                                       update is journaled

 * @param [in]  operation                How to update the count

 * @param [in]  normalOperation          Whether we are in normal operation vs.

 *                                       recovery or rebuild

 * @param [out] freeStatusChanged        A pointer which will be set to true if

 *                                       this update changed the free status of

 *                                       the block

 * @param [out] provisionalDecrementPtr  A pointer which will be set to true if

 *                                       this update was a decrement of a

 *                                       provisional reference

 *

 * @return VDO_SUCCESS or an error

 **/

static int updateReferenceCount(RefCounts          *refCounts,

                                ReferenceBlock     *block,

                                SlabBlockNumber     slabBlockNumber,

                                const JournalPoint *slabJournalPoint,

                                ReferenceOperation  operation,

                                bool                normalOperation,

                                bool               *freeStatusChanged,

                                bool               *provisionalDecrementPtr)

{

  ReferenceCount  *counterPtr = &refCounts->counters[slabBlockNumber];

  ReferenceStatus  oldStatus  = referenceCountToStatus(*counterPtr);

  PBNLock         *lock       = getReferenceOperationPBNLock(operation);

  int result;

  switch (operation.type) {

  case DATA_INCREMENT:

    result = incrementForData(refCounts, block, slabBlockNumber, oldStatus,

                              lock, counterPtr, freeStatusChanged);

    break;

  case DATA_DECREMENT:

    result = decrementForData(refCounts, block, slabBlockNumber, oldStatus,

                              lock, counterPtr, freeStatusChanged);

    if ((result == VDO_SUCCESS) && (oldStatus == RS_PROVISIONAL)) {

      if (provisionalDecrementPtr != NULL) {

        *provisionalDecrementPtr = true;

      }

      return VDO_SUCCESS;

    }

    break;

  case BLOCK_MAP_INCREMENT:

    result = incrementForBlockMap(refCounts, block, slabBlockNumber, oldStatus,

                                  lock, normalOperation, counterPtr,

                                  freeStatusChanged);

    break;

  default:

    logError("Unknown reference count operation: %u", operation.type);

    enterRefCountsReadOnlyMode(refCounts, VDO_NOT_IMPLEMENTED);

    result = VDO_NOT_IMPLEMENTED;

  }

  if (result != VDO_SUCCESS) {

    return result;

  }

  if (isValidJournalPoint(slabJournalPoint)) {

    refCounts->slabJournalPoint = *slabJournalPoint;

  }

  return VDO_SUCCESS;

}

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

int adjustReferenceCount(RefCounts          *refCounts,

                         ReferenceOperation  operation,

                         const JournalPoint *slabJournalPoint,

                         bool               *freeStatusChanged)

{

  if (!isSlabOpen(refCounts->slab)) {

    return VDO_INVALID_ADMIN_STATE;

  }

  SlabBlockNumber slabBlockNumber;

  int result = slabBlockNumberFromPBN(refCounts->slab, operation.pbn,

                                      &slabBlockNumber);

  if (result != VDO_SUCCESS) {

    return result;

  }

  ReferenceBlock *block = getReferenceBlock(refCounts, slabBlockNumber);

  bool provisionalDecrement = false;

  result = updateReferenceCount(refCounts, block, slabBlockNumber,

                                slabJournalPoint, operation,

                                NORMAL_OPERATION, freeStatusChanged,

                                &provisionalDecrement);

  if ((result != VDO_SUCCESS) || provisionalDecrement) {

    return result;

  }

  if (block->isDirty && (block->slabJournalLock > 0)) {

    /*

     * This block is already dirty and a slab journal entry has been made

     * for it since the last time it was clean. We must release the per-entry

     * slab journal lock for the entry associated with the update we are now

     * doing.

     */

    result = ASSERT(isValidJournalPoint(slabJournalPoint),

                    "Reference count adjustments need slab journal points.");

    if (result != VDO_SUCCESS) {

      return result;

    }

    SequenceNumber entryLock = slabJournalPoint->sequenceNumber;

    adjustSlabJournalBlockReference(refCounts->slab->journal, entryLock, -1);

    return VDO_SUCCESS;

  }

  /*

   * This may be the first time we are applying an update for which there

   * is a slab journal entry to this block since the block was

   * cleaned. Therefore, we convert the per-entry slab journal lock to an

   * uncommitted reference block lock, if there is a per-entry lock.

   */

  if (isValidJournalPoint(slabJournalPoint)) {

    block->slabJournalLock = slabJournalPoint->sequenceNumber;

  } else {

    block->slabJournalLock = 0;

  }

  dirtyBlock(block);

  return VDO_SUCCESS;

}

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

int adjustReferenceCountForRebuild(RefCounts           *refCounts,

                                   PhysicalBlockNumber  pbn,

                                   JournalOperation     operation)

{

  SlabBlockNumber slabBlockNumber;

  int result = slabBlockNumberFromPBN(refCounts->slab, pbn, &slabBlockNumber);

  if (result != VDO_SUCCESS) {

    return result;

  }

  ReferenceBlock *block = getReferenceBlock(refCounts, slabBlockNumber);

  bool unusedFreeStatus;

  ReferenceOperation physicalOperation = {

    .type = operation,

  };

  result = updateReferenceCount(refCounts, block, slabBlockNumber, NULL,

                                physicalOperation, !NORMAL_OPERATION,

                                &unusedFreeStatus, NULL);

  if (result != VDO_SUCCESS) {

    return result;

  }

  dirtyBlock(block);

  return VDO_SUCCESS;

}

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

int replayReferenceCountChange(RefCounts          *refCounts,

                               const JournalPoint *entryPoint,

                               SlabJournalEntry    entry)

{

  ReferenceBlock *block = getReferenceBlock(refCounts, entry.sbn);

  SectorCount sector

    = (entry.sbn % COUNTS_PER_BLOCK) / COUNTS_PER_SECTOR;

  if (!beforeJournalPoint(&block->commitPoints[sector], entryPoint)) {

    // This entry is already reflected in the existing counts, so do nothing.

    return VDO_SUCCESS;

  }

  // This entry is not yet counted in the reference counts.

  bool unusedFreeStatus;

  ReferenceOperation operation = {

    .type = entry.operation

  };

  int result = updateReferenceCount(refCounts, block, entry.sbn,

                                    entryPoint, operation, !NORMAL_OPERATION,

                                    &unusedFreeStatus, NULL);

  if (result != VDO_SUCCESS) {

    return result;

  }

  dirtyBlock(block);

  return VDO_SUCCESS;

}

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

int getReferenceStatus(RefCounts           *refCounts,

                       PhysicalBlockNumber  pbn,

                       ReferenceStatus     *statusPtr)

{

  ReferenceCount *counterPtr = NULL;

  int result = getReferenceCounter(refCounts, pbn, &counterPtr);

  if (result != VDO_SUCCESS) {

    return result;

  }

  *statusPtr = referenceCountToStatus(*counterPtr);

  return VDO_SUCCESS;

}

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

bool areEquivalentReferenceCounters(RefCounts *counterA, RefCounts *counterB)

{

  if ((counterA->blockCount             != counterB->blockCount)

      || (counterA->freeBlocks          != counterB->freeBlocks)

      || (counterA->referenceBlockCount != counterB->referenceBlockCount)) {

    return false;

  }

  for (size_t i = 0; i < counterA->referenceBlockCount; i++) {

    ReferenceBlock *blockA = &counterA->blocks[i];

    ReferenceBlock *blockB = &counterB->blocks[i];

    if (blockA->allocatedCount != blockB->allocatedCount) {

      return false;

    }

  }

  return (memcmp(counterA->counters, counterB->counters,

                 sizeof(ReferenceCount) * counterA->blockCount) == 0);

}

/**

 * Find the array index of the first zero byte in word-sized range of

 * reference counters. The search does no bounds checking; the function relies

 * on the array being sufficiently padded.

 *

 * @param wordPtr     A pointer to the eight counter bytes to check

 * @param startIndex  The array index corresponding to wordPtr[0]

 * @param failIndex   The array index to return if no zero byte is found

 * @return the array index of the first zero byte in the word, or

 *         the value passed as failIndex if no zero byte was found

 **/

static inline SlabBlockNumber findZeroByteInWord(const byte      *wordPtr,

                                                 SlabBlockNumber  startIndex,

                                                 SlabBlockNumber  failIndex)

{

  uint64_t word = getUInt64LE(wordPtr);

  // This looks like a loop, but GCC will unroll the eight iterations for us.

  for (unsigned int offset = 0; offset < BYTES_PER_WORD; offset++) {

    // Assumes little-endian byte order, which we have on X86.

    if ((word & 0xFF) == 0) {

      return (startIndex + offset);

    }

    word >>= 8;

  }

  return failIndex;

}

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

bool findFreeBlock(const RefCounts *refCounts,

                   SlabBlockNumber  startIndex,

                   SlabBlockNumber  endIndex,

                   SlabBlockNumber *indexPtr)

{

  SlabBlockNumber  zeroIndex;

  SlabBlockNumber  nextIndex   = startIndex;

  byte            *nextCounter = &refCounts->counters[nextIndex];

  byte            *endCounter  = &refCounts->counters[endIndex];

  // Search every byte of the first unaligned word. (Array is padded so

  // reading past end is safe.)

  zeroIndex = findZeroByteInWord(nextCounter, nextIndex, endIndex);

  if (zeroIndex < endIndex) {

    *indexPtr = zeroIndex;

    return true;

  }

  // On architectures where unaligned word access is expensive, this

  // would be a good place to advance to an alignment boundary.

  nextIndex   += BYTES_PER_WORD;

  nextCounter += BYTES_PER_WORD;

  // Now we're word-aligned; check an word at a time until we find a word

  // containing a zero. (Array is padded so reading past end is safe.)

  while (nextCounter < endCounter) {

    /*

     * The following code is currently an exact copy of the code preceding the

     * loop, but if you try to merge them by using a do loop, it runs slower

     * because a jump instruction gets added at the start of the iteration.

     */

    zeroIndex = findZeroByteInWord(nextCounter, nextIndex, endIndex);

    if (zeroIndex < endIndex) {

      *indexPtr = zeroIndex;

      return true;

    }

    nextIndex   += BYTES_PER_WORD;

    nextCounter += BYTES_PER_WORD;

  }

  return false;

}

/**

 * Search the reference block currently saved in the search cursor for a

 * reference count of zero, starting at the saved counter index.

 *

 * @param [in]  refCounts     The RefCounts object to search

 * @param [out] freeIndexPtr  A pointer to receive the array index of the

 *                            zero reference count

 *

 * @return true if an unreferenced counter was found

 **/

static bool searchCurrentReferenceBlock(const RefCounts *refCounts,

                                        SlabBlockNumber *freeIndexPtr)

{

  // Don't bother searching if the current block is known to be full.

  return ((refCounts->searchCursor.block->allocatedCount < COUNTS_PER_BLOCK)

          && findFreeBlock(refCounts, refCounts->searchCursor.index,

                           refCounts->searchCursor.endIndex, freeIndexPtr));

}

/**

 * Search each reference block for a reference count of zero, starting at the

 * reference block and counter index saved in the search cursor and searching

 * up to the end of the last reference block. The search does not wrap.

 *

 * @param [in]  refCounts     The RefCounts object to search

 * @param [out] freeIndexPtr  A pointer to receive the array index of the

 *                            zero reference count

 *

 * @return true if an unreferenced counter was found

 **/

static bool searchReferenceBlocks(RefCounts       *refCounts,

                                  SlabBlockNumber *freeIndexPtr)

{