/*

 * 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. 

 *

 * 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/kernel/udsIndex.c#16 $

 */

#include "udsIndex.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "murmur/MurmurHash3.h"

#include "numeric.h"

#include "stringUtils.h"

#include "uds-block.h"

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

typedef struct udsAttribute {

  struct attribute attr;

  const char *(*showString)(DedupeIndex *);

} UDSAttribute;

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

enum { UDS_Q_ACTION };

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

// These are the values in the atomic dedupeContext.requestState field

enum {

  // The UdsRequest object is not in use.

  UR_IDLE = 0,

  // The UdsRequest object is in use, and VDO is waiting for the result.

  UR_BUSY = 1,

  // The UdsRequest object is in use, but has timed out.

  UR_TIMED_OUT = 2,

};

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

typedef enum {

  // The UDS index is closed

  IS_CLOSED = 0,

  // The UDS index session is opening or closing

  IS_CHANGING = 1,

  // The UDS index is open.  There is a UDS index session.

  IS_OPENED = 2,

} IndexState;

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

typedef struct udsIndex {

  DedupeIndex        common;

  struct kobject     dedupeObject;

  RegisteredThread   allocatingThread;

  char              *indexName;

  UdsConfiguration   configuration;

  struct uds_parameters udsParams;

  struct uds_index_session *indexSession;

  atomic_t           active;

  // This spinlock protects the state fields and the starting of dedupe

  // requests.

  spinlock_t         stateLock;

  KvdoWorkItem       workItem;    // protected by stateLock

  KvdoWorkQueue     *udsQueue;    // protected by stateLock

  unsigned int       maximum;     // protected by stateLock

  IndexState         indexState;  // protected by stateLock

  IndexState         indexTarget; // protected by stateLock

  bool               changing;    // protected by stateLock

  bool               createFlag;  // protected by stateLock

  bool               dedupeFlag;  // protected by stateLock

  bool               deduping;    // protected by stateLock

  bool               errorFlag;   // protected by stateLock

  bool               suspended;   // protected by stateLock

  // This spinlock protects the pending list, the pending flag in each KVIO,

  // and the timeout list.

  spinlock_t         pendingLock;

  struct list_head   pendingHead;  // protected by pendingLock

  struct timer_list  pendingTimer; // protected by pendingLock

  bool               startedTimer; // protected by pendingLock

} UDSIndex;

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

// Version 1:  user space albireo index (limited to 32 bytes)

// Version 2:  kernel space albireo index (limited to 16 bytes)

enum {

  UDS_ADVICE_VERSION = 2,

  // version byte + state byte + 64-bit little-endian PBN

  UDS_ADVICE_SIZE    = 1 + 1 + sizeof(uint64_t),

};

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

  // We want to ensure that there is only one copy of the following constants.

static const char *CLOSED    = "closed";

static const char *CLOSING   = "closing";

static const char *ERROR     = "error";

static const char *OFFLINE   = "offline";

static const char *ONLINE    = "online";

static const char *OPENING   = "opening";

static const char *SUSPENDED = "suspended";

static const char *UNKNOWN   = "unknown";

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

static const char *indexStateToString(UDSIndex *index, IndexState state)

{

  if (index->suspended) {

    return SUSPENDED;

  }

  switch (state) {

  case IS_CLOSED:

    // Closed.  The errorFlag tells if it is because of an error.

    return index->errorFlag ? ERROR : CLOSED;

  case IS_CHANGING:

    // The indexTarget tells if we are opening or closing the index.

    return index->indexTarget == IS_OPENED ? OPENING : CLOSING;

  case IS_OPENED:

    // Opened.  The dedupeFlag tells if we are online or offline.

    return index->dedupeFlag ? ONLINE : OFFLINE;

  default:

    return UNKNOWN;

  }

}

/**

 * Encode VDO duplicate advice into the newMetadata field of a UDS request.

 *

 * @param request  The UDS request to receive the encoding

 * @param advice   The advice to encode

 **/

static void encodeUDSAdvice(UdsRequest *request, DataLocation advice)

{

  size_t offset = 0;

  struct udsChunkData *encoding = &request->newMetadata;

  encoding->data[offset++] = UDS_ADVICE_VERSION;

  encoding->data[offset++] = advice.state;

  encodeUInt64LE(encoding->data, &offset, advice.pbn);

  BUG_ON(offset != UDS_ADVICE_SIZE);

}

/**

 * Decode VDO duplicate advice from the oldMetadata field of a UDS request.

 *

 * @param request  The UDS request containing the encoding

 * @param advice   The DataLocation to receive the decoded advice

 *

 * @return true if valid advice was found and decoded

 **/

static bool decodeUDSAdvice(const UdsRequest *request, DataLocation *advice)

{

  if ((request->status != UDS_SUCCESS) || !request->found) {

    return false;

  }

  size_t offset = 0;

  const struct udsChunkData *encoding = &request->oldMetadata;

  byte version = encoding->data[offset++];

  if (version != UDS_ADVICE_VERSION) {

    logError("invalid UDS advice version code %u", version);

    return false;

  }

  advice->state = encoding->data[offset++];

  decodeUInt64LE(encoding->data, &offset, &advice->pbn);

  BUG_ON(offset != UDS_ADVICE_SIZE);

  return true;

}

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

static void finishIndexOperation(UdsRequest *udsRequest)

{

  DataKVIO *dataKVIO = container_of(udsRequest, DataKVIO,

                                    dedupeContext.udsRequest);

  DedupeContext *dedupeContext = &dataKVIO->dedupeContext;

  if (compareAndSwap32(&dedupeContext->requestState, UR_BUSY, UR_IDLE)) {

    KVIO *kvio = dataKVIOAsKVIO(dataKVIO);

    UDSIndex *index = container_of(kvio->layer->dedupeIndex, UDSIndex, common);

    spin_lock_bh(&index->pendingLock);

    if (dedupeContext->isPending) {

      list_del(&dedupeContext->pendingList);

      dedupeContext->isPending = false;

    }

    spin_unlock_bh(&index->pendingLock);

    dedupeContext->status = udsRequest->status;

    if ((udsRequest->type == UDS_POST) || (udsRequest->type == UDS_QUERY)) {

      DataLocation advice;

      if (decodeUDSAdvice(udsRequest, &advice)) {

        setDedupeAdvice(dedupeContext, &advice);

      } else {

        setDedupeAdvice(dedupeContext, NULL);

      }

    }

    invokeDedupeCallback(dataKVIO);

    atomic_dec(&index->active);

  } else {

    compareAndSwap32(&dedupeContext->requestState, UR_TIMED_OUT, UR_IDLE);

  }

}

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

static void startExpirationTimer(UDSIndex *index, DataKVIO *dataKVIO)

{

  if (!index->startedTimer) {

    index->startedTimer = true;

    mod_timer(&index->pendingTimer,

              getAlbireoTimeout(dataKVIO->dedupeContext.submissionTime));

  }

}

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

static void startIndexOperation(KvdoWorkItem *item)

{

  KVIO *kvio = workItemAsKVIO(item);

  DataKVIO *dataKVIO = kvioAsDataKVIO(kvio);

  UDSIndex *index = container_of(kvio->layer->dedupeIndex, UDSIndex, common);

  DedupeContext *dedupeContext = &dataKVIO->dedupeContext;

  spin_lock_bh(&index->pendingLock);

  list_add_tail(&dedupeContext->pendingList, &index->pendingHead);

  dedupeContext->isPending = true;

  startExpirationTimer(index, dataKVIO);

  spin_unlock_bh(&index->pendingLock);

  UdsRequest *udsRequest = &dedupeContext->udsRequest;

  int status = udsStartChunkOperation(udsRequest);

  if (status != UDS_SUCCESS) {

    udsRequest->status = status;

    finishIndexOperation(udsRequest);

  }

}

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

#if LINUX_VERSION_CODE >= KERNEL_VERSION(4,15,0)

static void timeoutIndexOperations(struct timer_list *t)

#else

static void timeoutIndexOperations(unsigned long arg)

#endif

{

#if LINUX_VERSION_CODE >= KERNEL_VERSION(4,15,0)

  UDSIndex *index = from_timer(index, t, pendingTimer);

#else

  UDSIndex *index = (UDSIndex *) arg;

#endif

  LIST_HEAD(expiredHead);

  uint64_t timeoutJiffies = msecs_to_jiffies(albireoTimeoutInterval);

  unsigned long earliestSubmissionAllowed = jiffies - timeoutJiffies;

  spin_lock_bh(&index->pendingLock);

  index->startedTimer = false;

  while (!list_empty(&index->pendingHead)) {

    DataKVIO *dataKVIO = list_first_entry(&index->pendingHead, DataKVIO,

                                          dedupeContext.pendingList);

    DedupeContext *dedupeContext = &dataKVIO->dedupeContext;

    if (earliestSubmissionAllowed <= dedupeContext->submissionTime) {

      startExpirationTimer(index, dataKVIO);

      break;

    }

    list_del(&dedupeContext->pendingList);

    dedupeContext->isPending = false;

    list_add_tail(&dedupeContext->pendingList, &expiredHead);

  }

  spin_unlock_bh(&index->pendingLock);

  while (!list_empty(&expiredHead)) {

    DataKVIO *dataKVIO = list_first_entry(&expiredHead, DataKVIO,

                                          dedupeContext.pendingList);

    DedupeContext *dedupeContext = &dataKVIO->dedupeContext;

    list_del(&dedupeContext->pendingList);

    if (compareAndSwap32(&dedupeContext->requestState,

                         UR_BUSY, UR_TIMED_OUT)) {

      dedupeContext->status = ETIMEDOUT;

      invokeDedupeCallback(dataKVIO);

      atomic_dec(&index->active);

      kvdoReportDedupeTimeout(dataKVIOAsKVIO(dataKVIO)->layer, 1);

    }

  }

}

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

static void enqueueIndexOperation(DataKVIO        *dataKVIO,

                                  UdsCallbackType  operation)

{

  KVIO *kvio = dataKVIOAsKVIO(dataKVIO);

  DedupeContext *dedupeContext = &dataKVIO->dedupeContext;

  UDSIndex *index = container_of(kvio->layer->dedupeIndex, UDSIndex, common);

  dedupeContext->status         = UDS_SUCCESS;

  dedupeContext->submissionTime = jiffies;

  if (compareAndSwap32(&dedupeContext->requestState, UR_IDLE, UR_BUSY)) {

    UdsRequest *udsRequest = &dataKVIO->dedupeContext.udsRequest;

    udsRequest->chunkName = *dedupeContext->chunkName;

    udsRequest->callback  = finishIndexOperation;

    udsRequest->session   = index->indexSession;

    udsRequest->type      = operation;

    udsRequest->update    = true;

    if ((operation == UDS_POST) || (operation == UDS_UPDATE)) {

      encodeUDSAdvice(udsRequest, getDedupeAdvice(dedupeContext));

    }

    setupWorkItem(&kvio->enqueueable.workItem, startIndexOperation, NULL,

                  UDS_Q_ACTION);

    spin_lock(&index->stateLock);

    if (index->deduping) {

      enqueueWorkQueue(index->udsQueue, &kvio->enqueueable.workItem);

      unsigned int active = atomic_inc_return(&index->active);

      if (active > index->maximum) {

        index->maximum = active;

      }

      kvio = NULL;

    } else {

      atomicStore32(&dedupeContext->requestState, UR_IDLE);

    }

    spin_unlock(&index->stateLock);

  } else {

    // A previous user of the KVIO had a dedupe timeout

    // and its request is still outstanding.

    atomic64_inc(&kvio->layer->dedupeContextBusy);

  }

  if (kvio != NULL) {

    invokeDedupeCallback(dataKVIO);

  }

}

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

static void closeIndex(UDSIndex *index)

{

  // Change the index state so that getIndexStatistics will not try to

  // use the index session we are closing.

  index->indexState = IS_CHANGING;

  spin_unlock(&index->stateLock);

  int result = udsCloseIndex(index->indexSession);

  if (result != UDS_SUCCESS) {

    logErrorWithStringError(result, "Error closing index %s",

                            index->indexName);

  }

  spin_lock(&index->stateLock);

  index->indexState = IS_CLOSED;

  index->errorFlag |= result != UDS_SUCCESS;

  // ASSERTION: We leave in IS_CLOSED state.

}

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

static void openIndex(UDSIndex *index)

{

  // ASSERTION: We enter in IS_CLOSED state.

  bool createFlag = index->createFlag;

  index->createFlag = false;

  // Change the index state so that the it will be reported to the outside

  // world as "opening".

  index->indexState = IS_CHANGING;

  index->errorFlag = false;

  // Open the index session, while not holding the stateLock

  spin_unlock(&index->stateLock);

  int result = udsOpenIndex(createFlag ? UDS_CREATE : UDS_LOAD,

                            index->indexName, &index->udsParams,

                            index->configuration, index->indexSession);

  if (result != UDS_SUCCESS) {

    logErrorWithStringError(result, "Error opening index %s",

                            index->indexName);

  }

  spin_lock(&index->stateLock);

  if (!createFlag) {

    switch (result) {

    case UDS_CORRUPT_COMPONENT:

    case UDS_NO_INDEX:

      // Either there is no index, or there is no way we can recover the index.

      // We will be called again and try to create a new index.

      index->indexState = IS_CLOSED;

      index->createFlag = true;

      return;

    default:

      break;

    }

  }

  if (result == UDS_SUCCESS) {

    index->indexState  = IS_OPENED;

  } else {

    index->indexState  = IS_CLOSED;

    index->indexTarget = IS_CLOSED;

    index->errorFlag   = true;

    spin_unlock(&index->stateLock);

    logInfo("Setting UDS index target state to error");

    spin_lock(&index->stateLock);

  }

  // ASSERTION: On success, we leave in IS_OPEN state.

  // ASSERTION: On failure, we leave in IS_CLOSED state.

}

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

static void changeDedupeState(KvdoWorkItem *item)

{

  UDSIndex *index = container_of(item, UDSIndex, workItem);

  spin_lock(&index->stateLock);

  // Loop until the index is in the target state and the create flag is

  // clear.

  while (!index->suspended &&

         ((index->indexState != index->indexTarget) ||

          index->createFlag)) {

    if (index->indexState == IS_OPENED) {

      closeIndex(index);

    } else {

      openIndex(index);

    }

  }

  index->changing = false;

  index->deduping = index->dedupeFlag && (index->indexState == IS_OPENED);

  spin_unlock(&index->stateLock);

}

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

static void launchDedupeStateChange(UDSIndex *index)

{

  // ASSERTION: We enter with the state_lock held.

  if (index->changing || index->suspended) {

    // Either a change is already in progress, or changes are

    // not allowed.

    return;

  }

  if (index->createFlag ||

      (index->indexState != index->indexTarget)) {

    index->changing = true;

    index->deduping = false;

    setupWorkItem(&index->workItem,

                  changeDedupeState,

                  NULL,

                  UDS_Q_ACTION);

    enqueueWorkQueue(index->udsQueue, &index->workItem);

    return;

  }

  // Online vs. offline changes happen immediately

  index->deduping = (index->dedupeFlag && !index->suspended &&

                     (index->indexState == IS_OPENED));

  // ASSERTION: We exit with the state_lock held.

}

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

static void setTargetState(UDSIndex   *index,

                           IndexState  target,

                           bool        changeDedupe,

                           bool        dedupe,

                           bool        setCreate)

{

  spin_lock(&index->stateLock);

  const char *oldState = indexStateToString(index, index->indexTarget);

  if (changeDedupe) {

    index->dedupeFlag = dedupe;

  }

  if (setCreate) {

    index->createFlag = true;

  }

  index->indexTarget = target;

  launchDedupeStateChange(index);

  const char *newState = indexStateToString(index, index->indexTarget);

  spin_unlock(&index->stateLock);

  if (oldState != newState) {

    logInfo("Setting UDS index target state to %s", newState);

  }

}

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

static void suspendUDSIndex(DedupeIndex *dedupeIndex, bool saveFlag)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  spin_lock(&index->stateLock);

  index->suspended = true;

  IndexState indexState = index->indexState;

  spin_unlock(&index->stateLock);

  if (indexState != IS_CLOSED) {

    int result = udsSuspendIndexSession(index->indexSession, saveFlag);

    if (result != UDS_SUCCESS) {

      logErrorWithStringError(result, "Error suspending dedupe index");

    }

  }

}

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

static void resumeUDSIndex(DedupeIndex *dedupeIndex)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  int result = udsResumeIndexSession(index->indexSession);

  if (result != UDS_SUCCESS) {

    logErrorWithStringError(result, "Error resuming dedupe index");

  }

  spin_lock(&index->stateLock);

  index->suspended = false;

  launchDedupeStateChange(index);

  spin_unlock(&index->stateLock);

}

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

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

static void dumpUDSIndex(DedupeIndex *dedupeIndex, bool showQueue)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  spin_lock(&index->stateLock);

  const char *state = indexStateToString(index, index->indexState);

  const char *target = (index->changing

                        ? indexStateToString(index, index->indexTarget)

                        : NULL);

  spin_unlock(&index->stateLock);

  logInfo("UDS index: state: %s", state);

  if (target != NULL) {

    logInfo("UDS index: changing to state: %s", target);

  }

  if (showQueue) {

    dumpWorkQueue(index->udsQueue);

  }

}

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

static void finishUDSIndex(DedupeIndex *dedupeIndex)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  setTargetState(index, IS_CLOSED, false, false, false);

  udsDestroyIndexSession(index->indexSession);

  finishWorkQueue(index->udsQueue);

}

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

static void freeUDSIndex(DedupeIndex *dedupeIndex)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  freeWorkQueue(&index->udsQueue);

  spin_lock_bh(&index->pendingLock);

  if (index->startedTimer) {

    del_timer_sync(&index->pendingTimer);

  }

  spin_unlock_bh(&index->pendingLock);

  kobject_put(&index->dedupeObject);

}

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

static const char *getUDSStateName(DedupeIndex *dedupeIndex)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  spin_lock(&index->stateLock);

  const char *state = indexStateToString(index, index->indexState);

  spin_unlock(&index->stateLock);

  return state;

}

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

static void getUDSStatistics(DedupeIndex *dedupeIndex, IndexStatistics *stats)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  spin_lock(&index->stateLock);

  IndexState      indexState   = index->indexState;

  stats->maxDedupeQueries      = index->maximum;

  spin_unlock(&index->stateLock);

  stats->currDedupeQueries     = atomic_read(&index->active);

  if (indexState == IS_OPENED) {

    UdsIndexStats indexStats;

    int result = udsGetIndexStats(index->indexSession, &indexStats);

    if (result == UDS_SUCCESS) {

      stats->entriesIndexed = indexStats.entriesIndexed;

    } else {

      logErrorWithStringError(result, "Error reading index stats");

    }

    UdsContextStats contextStats;

    result = udsGetIndexSessionStats(index->indexSession, &contextStats);

    if (result == UDS_SUCCESS) {

      stats->postsFound      = contextStats.postsFound;

      stats->postsNotFound   = contextStats.postsNotFound;

      stats->queriesFound    = contextStats.queriesFound;

      stats->queriesNotFound = contextStats.queriesNotFound;

      stats->updatesFound    = contextStats.updatesFound;

      stats->updatesNotFound = contextStats.updatesNotFound;

    } else {

      logErrorWithStringError(result, "Error reading context stats");

    }

  }

}

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

static int processMessage(DedupeIndex *dedupeIndex, const char *name)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  if (strcasecmp(name, "index-close") == 0) {

    setTargetState(index, IS_CLOSED, false, false, false);

    return 0;

  } else if (strcasecmp(name, "index-create") == 0) {

    setTargetState(index, IS_OPENED, false, false, true);

    return 0;

  } else if (strcasecmp(name, "index-disable") == 0) {

    setTargetState(index, IS_OPENED, true, false, false);

    return 0;

  } else if (strcasecmp(name, "index-enable") == 0) {

    setTargetState(index, IS_OPENED, true, true, false);

    return 0;

  }

  return -EINVAL;

}

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

static void udsPost(DataKVIO *dataKVIO)

{

  enqueueIndexOperation(dataKVIO, UDS_POST);

}

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

static void udsQuery(DataKVIO *dataKVIO)

{

  enqueueIndexOperation(dataKVIO, UDS_QUERY);

}

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

static void startUDSIndex(DedupeIndex *dedupeIndex, bool createFlag)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  setTargetState(index, IS_OPENED, true, true, createFlag);

}

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

static void stopUDSIndex(DedupeIndex *dedupeIndex)

{

  UDSIndex *index = container_of(dedupeIndex, UDSIndex, common);

  setTargetState(index, IS_CLOSED, false, false, false);

}

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

static void udsUpdate(DataKVIO *dataKVIO)

{

  enqueueIndexOperation(dataKVIO, UDS_UPDATE);

}

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

static void dedupeKobjRelease(struct kobject *kobj)

{

  UDSIndex *index = container_of(kobj, UDSIndex, dedupeObject);

  udsFreeConfiguration(index->configuration);

  FREE(index->indexName);

  FREE(index);

}

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

static ssize_t dedupeStatusShow(struct kobject   *kobj,

                                struct attribute *attr,

                                char             *buf)

{

  UDSAttribute *ua = container_of(attr, UDSAttribute, attr);

  UDSIndex *index = container_of(kobj, UDSIndex, dedupeObject);

  if (ua->showString != NULL) {

    return sprintf(buf, "%s\n", ua->showString(&index->common));

  } else {

    return -EINVAL;

  }

}

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

static ssize_t dedupeStatusStore(struct kobject   *kobj,

                                 struct attribute *attr,

                                 const char       *buf,

                                 size_t            length)

{

  return -EINVAL;

}

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

static struct sysfs_ops dedupeSysfsOps = {

  .show  = dedupeStatusShow,

  .store = dedupeStatusStore,

};

static UDSAttribute dedupeStatusAttribute = {

  .attr = {.name = "status", .mode = 0444, },

  .showString = getUDSStateName,

};

static struct attribute *dedupeAttributes[] = {

  &dedupeStatusAttribute.attr,

  NULL,

};

static struct kobj_type dedupeKobjType = {

  .release       = dedupeKobjRelease,

  .sysfs_ops     = &dedupeSysfsOps,

  .default_attrs = dedupeAttributes,

};

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

static void startUDSQueue(void *ptr)

{

  /*

   * Allow the UDS dedupe worker thread to do memory allocations.  It will

   * only do allocations during the UDS calls that open or close an index,

   * but those allocations can safely sleep while reserving a large amount

   * of memory.  We could use an allocationsAllowed boolean (like the base

   * threads do), but it would be an unnecessary embellishment.

   */

  UDSIndex *index = ptr;

  registerAllocatingThread(&index->allocatingThread, NULL);

}

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

static void finishUDSQueue(void *ptr)

{

  unregisterAllocatingThread();

}

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

int makeUDSIndex(KernelLayer *layer, DedupeIndex **indexPtr)

{

  UDSIndex *index;

  int result = ALLOCATE(1, UDSIndex, "UDS index data", &index);

  if (result != UDS_SUCCESS) {

    return result;

  }

  result = allocSprintf("index name", &index->indexName,

                        "dev=%s offset=4096 size=%llu",

                        layer->deviceConfig->parentDeviceName,

                        getIndexRegionSize(layer->geometry) * VDO_BLOCK_SIZE);

  if (result != UDS_SUCCESS) {

    logError("Creating index name failed (%d)", result);

    FREE(index);

    return result;

  }

  index->udsParams = (struct uds_parameters) UDS_PARAMETERS_INITIALIZER;

  indexConfigToUdsParameters(&layer->geometry.indexConfig, &index->udsParams);

  result = indexConfigToUdsConfiguration(&layer->geometry.indexConfig,

                                         &index->configuration);

  if (result != VDO_SUCCESS) {

    FREE(index->indexName);

    FREE(index);

    return result;

  }

  udsConfigurationSetNonce(index->configuration,

                           (UdsNonce) layer->geometry.nonce);

  result = udsCreateIndexSession(&index->indexSession);

  if (result != UDS_SUCCESS) {

    udsFreeConfiguration(index->configuration);

    FREE(index->indexName);

    FREE(index);

    return result;

  }

  static const KvdoWorkQueueType udsQueueType = {

    .start        = startUDSQueue,

    .finish       = finishUDSQueue,

    .actionTable  = {

      { .name = "uds_action", .code = UDS_Q_ACTION, .priority = 0 },

    },

  };

  result = makeWorkQueue(layer->threadNamePrefix, "dedupeQ",

                         &layer->wqDirectory, layer, index, &udsQueueType, 1,

                         &index->udsQueue);

  if (result != VDO_SUCCESS) {

    logError("UDS index queue initialization failed (%d)", result);

    udsDestroyIndexSession(index->indexSession);

    udsFreeConfiguration(index->configuration);

    FREE(index->indexName);

    FREE(index);

    return result;

  }

  kobject_init(&index->dedupeObject, &dedupeKobjType);

  result = kobject_add(&index->dedupeObject, &layer->kobj, "dedupe");

  if (result != VDO_SUCCESS) {

    freeWorkQueue(&index->udsQueue);

    udsDestroyIndexSession(index->indexSession);

    udsFreeConfiguration(index->configuration);

    FREE(index->indexName);

    FREE(index);

    return result;

  }

  index->common.dump                      = dumpUDSIndex;

  index->common.free                      = freeUDSIndex;

  index->common.getDedupeStateName        = getUDSStateName;

  index->common.getStatistics             = getUDSStatistics;

  index->common.message                   = processMessage;

  index->common.post                      = udsPost;

  index->common.query                     = udsQuery;

  index->common.resume                    = resumeUDSIndex;

  index->common.start                     = startUDSIndex;

  index->common.stop                      = stopUDSIndex;

  index->common.suspend                   = suspendUDSIndex;

  index->common.finish                    = finishUDSIndex;

  index->common.update                    = udsUpdate;

  INIT_LIST_HEAD(&index->pendingHead);

  spin_lock_init(&index->pendingLock);

  spin_lock_init(&index->stateLock);

#if LINUX_VERSION_CODE >= KERNEL_VERSION(4,15,0)

  timer_setup(&index->pendingTimer, timeoutIndexOperations, 0);

#else

  setup_timer(&index->pendingTimer, timeoutIndexOperations,

              (unsigned long) index);

#endif

  *indexPtr = &index->common;

  return VDO_SUCCESS;

}