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

#include "readOnlyNotifier.h"

#include "atomic.h"
#include "logger.h"
#include "memoryAlloc.h"
#include "permassert.h"

#include "completion.h"
#include "physicalLayer.h"
#include "threadConfig.h"

/**
 * A ReadOnlyNotifier has a single completion which is used to perform
 * read-only notifications, however, enterReadOnlyMode() may be called from any
 * base thread. A pair of atomic fields are used to control the read-only mode
 * entry process. The first field holds the read-only error. The second is the
 * state field, which may hold any of the four special values enumerated here.
 *
 * When enterReadOnlyMode() is called from some base thread, a compare-and-swap
 * is done on the readOnlyError, setting it to the supplied error if the value
 * was VDO_SUCCESS. If this fails, some other thread has already intiated
 * read-only entry or scheduled a pending entry, so the call exits. Otherwise,
 * a compare-and-swap is done on the state, setting it to NOTIFYING if the
 * value was MAY_NOTIFY. If this succeeds, the caller initiates the
 * notification. If this failed due to notifications being disallowed, the
 * notifier will be in the MAY_NOT_NOTIFY state but readOnlyError will not be
 * VDO_SUCCESS. This configuration will indicate to allowReadOnlyModeEntry()
 * that there is a pending notification to perform.
 **/
enum {
  /** Notifications are allowed but not in progress */
  MAY_NOTIFY = 0,
  /** A notification is in progress */
  NOTIFYING,
  /** Notifications are not allowed */
  MAY_NOT_NOTIFY,
  /** A notification has completed */
  NOTIFIED,
};

/**
 * An object to be notified when the VDO enters read-only mode
 **/
typedef struct readOnlyListener ReadOnlyListener;

struct readOnlyListener {
  /** The listener */
  void                 *listener;
  /** The method to call to notifiy the listener */
  ReadOnlyNotification *notify;
  /** A pointer to the next listener */
  ReadOnlyListener     *next;
};

/**
 * Data associated with each base code thread.
 **/
typedef struct threadData {
  /**
   * Each thread maintains its own notion of whether the VDO is read-only so
   * that the read-only state can be checked from any base thread without
   * worrying about synchronization or thread safety. This does mean that
   * knowledge of the VDO going read-only does not occur simultaneously across
   * the VDO's threads, but that does not seem to cause any problems.
   */
  bool              isReadOnly;
  /**
   * A list of objects waiting to be notified on this thread that the VDO has
   * entered read-only mode.
   **/
  ReadOnlyListener *listeners;
} ThreadData;

struct readOnlyNotifier {
  /** The completion for entering read-only mode */
  VDOCompletion       completion;
  /** A completion waiting for notifications to be drained or enabled */
  VDOCompletion      *waiter;
  /** The code of the error which put the VDO into read-only mode */
  Atomic32            readOnlyError;
  /** The current state of the notifier (values described above) */
  Atomic32            state;
  /** The thread config of the VDO */
  const ThreadConfig *threadConfig;
  /** The array of per-thread data */
  ThreadData          threadData[];
};

/**
 * Convert a generic VDOCompletion to a ReadOnlyNotifier.
 *
 * @param completion The completion to convert
 *
 * @return The completion as a ReadOnlyNotifier
 **/
static inline ReadOnlyNotifier *asNotifier(VDOCompletion *completion)
{
  STATIC_ASSERT(offsetof(ReadOnlyNotifier, completion) == 0);
  assertCompletionType(completion->type, READ_ONLY_MODE_COMPLETION);
  return (ReadOnlyNotifier *) completion;
}

/**********************************************************************/
int makeReadOnlyNotifier(bool                 isReadOnly,
                         const ThreadConfig  *threadConfig,
                         PhysicalLayer       *layer,
                         ReadOnlyNotifier   **notifierPtr)
{
  ReadOnlyNotifier *notifier;
  int result = ALLOCATE_EXTENDED(ReadOnlyNotifier,
                                 threadConfig->baseThreadCount, ThreadData,
                                 __func__, &notifier);
  if (result != VDO_SUCCESS) {
    return result;
  }

  notifier->threadConfig = threadConfig;
  if (isReadOnly) {
    atomicStore32(&notifier->readOnlyError, (uint32_t) VDO_READ_ONLY);
    atomicStore32(&notifier->state, NOTIFIED);
  } else {
    atomicStore32(&notifier->state, MAY_NOTIFY);
  }
  result = initializeEnqueueableCompletion(&notifier->completion,
                                           READ_ONLY_MODE_COMPLETION, layer);
  if (result != VDO_SUCCESS) {
    freeReadOnlyNotifier(&notifier);
    return result;
  }

  for (ThreadCount id = 0; id < threadConfig->baseThreadCount; id++) {
    notifier->threadData[id].isReadOnly = isReadOnly;
  }

  *notifierPtr = notifier;
  return VDO_SUCCESS;
}

/**********************************************************************/
void freeReadOnlyNotifier(ReadOnlyNotifier **notifierPtr)
{
  ReadOnlyNotifier *notifier = *notifierPtr;
  if (notifier == NULL) {
    return;
  }

  for (ThreadCount id = 0; id < notifier->threadConfig->baseThreadCount;
       id++) {
    ThreadData       *threadData = &notifier->threadData[id];
    ReadOnlyListener *listener   = threadData->listeners;
    while (listener != NULL) {
      ReadOnlyListener *toFree = listener;
      listener = listener->next;
      FREE(toFree);
    }
  }

  destroyEnqueueable(&notifier->completion);
  FREE(notifier);
  *notifierPtr = NULL;
}

/**
 * Check that a function was called on the admin thread.
 *
 * @param notifier  The notifier
 * @param caller    The name of the function (for logging)
 **/
static void assertOnAdminThread(ReadOnlyNotifier *notifier, const char *caller)
{
  ThreadID threadID = getCallbackThreadID();
  ASSERT_LOG_ONLY((getAdminThread(notifier->threadConfig) == threadID),
                  "%s called on admin thread", caller);
}


/**********************************************************************/
void waitUntilNotEnteringReadOnlyMode(ReadOnlyNotifier *notifier,
                                      VDOCompletion    *parent)
{
  if (notifier == NULL) {
    finishCompletion(parent, VDO_SUCCESS);
    return;
  }

  assertOnAdminThread(notifier, __func__);
  if (notifier->waiter != NULL) {
    finishCompletion(parent, VDO_COMPONENT_BUSY);
    return;
  }

  uint32_t state = atomicLoad32(&notifier->state);
  if ((state == MAY_NOT_NOTIFY) || (state == NOTIFIED)) {
    // Notifications are already done or disallowed.
    completeCompletion(parent);
    return;
  }

  if (compareAndSwap32(&notifier->state, MAY_NOTIFY, MAY_NOT_NOTIFY)) {
    // A notification was not in progress, and now they are disallowed.
    completeCompletion(parent);
    return;
  }

  /*
   * A notification is in progress, so wait for it to finish. There is no race
   * here since the notification can't finish while the admin thread is in this
   * method.
   */
  notifier->waiter = parent;
}

/**
 * Complete the process of entering read only mode.
 *
 * @param completion  The read-only mode completion
 **/
static void finishEnteringReadOnlyMode(VDOCompletion *completion)
{
  ReadOnlyNotifier *notifier = asNotifier(completion);
  assertOnAdminThread(notifier, __func__);
  atomicStore32(&notifier->state, NOTIFIED);

  VDOCompletion *waiter = notifier->waiter;
  if (waiter != NULL) {
    notifier->waiter = NULL;
    finishCompletion(waiter, completion->result);
  }
}

/**
 * Inform each thread that the VDO is in read-only mode.
 *
 * @param completion  The read-only mode completion
 **/
static void makeThreadReadOnly(VDOCompletion *completion)
{
  ThreadID          threadID   = completion->callbackThreadID;
  ReadOnlyNotifier *notifier   = asNotifier(completion);
  ReadOnlyListener *listener   = completion->parent;
  if (listener == NULL) {
    // This is the first call on this thread
    ThreadData *threadData = &notifier->threadData[threadID];
    threadData->isReadOnly = true;
    listener               = threadData->listeners;
    if (threadID == 0) {
      // Note: This message must be recognizable by Permabit::UserMachine.
      logErrorWithStringError((int) atomicLoad32(&notifier->readOnlyError),
                              "Unrecoverable error, entering read-only mode");
    }
  } else {
    // We've just finished notifying a listener
    listener = listener->next;
  }

  if (listener != NULL) {
    // We have a listener to notify
    prepareCompletion(completion, makeThreadReadOnly, makeThreadReadOnly,
                      threadID, listener);
    listener->notify(listener->listener, completion);
    return;
  }

  // We're done with this thread
  if (++threadID >= notifier->threadConfig->baseThreadCount) {
    // There are no more threads
    prepareCompletion(completion, finishEnteringReadOnlyMode,
                      finishEnteringReadOnlyMode,
                      getAdminThread(notifier->threadConfig), NULL);
  } else {
    prepareCompletion(completion, makeThreadReadOnly, makeThreadReadOnly,
                      threadID, NULL);
  }

  invokeCallback(completion);
}

/**********************************************************************/
void allowReadOnlyModeEntry(ReadOnlyNotifier *notifier, VDOCompletion *parent)
{
  assertOnAdminThread(notifier, __func__);
  if (notifier->waiter != NULL) {
    finishCompletion(parent, VDO_COMPONENT_BUSY);
    return;
  }

   if (!compareAndSwap32(&notifier->state, MAY_NOT_NOTIFY, MAY_NOTIFY)) {
    // Notifications were already allowed or complete
    completeCompletion(parent);
    return;
  }

  if ((int) atomicLoad32(&notifier->readOnlyError) == VDO_SUCCESS) {
    // We're done
    completeCompletion(parent);
    return;
  }

  // There may have been a pending notification
  if (!compareAndSwap32(&notifier->state, MAY_NOTIFY, NOTIFYING)) {
    /*
     * There wasn't, the error check raced with a thread calling
     * enterReadOnlyMode() after we set the state to MAY_NOTIFY. It has already
     * started the notification.
     */
    completeCompletion(parent);
    return;
  }

  // Do the pending notification.
  notifier->waiter = parent;
  makeThreadReadOnly(&notifier->completion);
}

/**********************************************************************/
void enterReadOnlyMode(ReadOnlyNotifier *notifier, int errorCode)
{
  ThreadData *threadData = &notifier->threadData[getCallbackThreadID()];
  if (threadData->isReadOnly) {
    // This thread has already gone read-only.
    return;
  }

  // Record for this thread that the VDO is read-only.
  threadData->isReadOnly = true;

  if (!compareAndSwap32(&notifier->readOnlyError, (uint32_t) VDO_SUCCESS,
                        (uint32_t) errorCode)) {
    // The notifier is already aware of a read-only error
    return;
  }

  if (compareAndSwap32(&notifier->state, MAY_NOTIFY, NOTIFYING)) {
    // Initiate a notification starting on the lowest numbered thread.
    launchCallback(&notifier->completion, makeThreadReadOnly, 0);
  }
}

/**********************************************************************/
bool isReadOnly(ReadOnlyNotifier *notifier)
{
  return notifier->threadData[getCallbackThreadID()].isReadOnly;
}

/**********************************************************************/
bool isOrWillBeReadOnly(ReadOnlyNotifier *notifier)
{
  return (((int) relaxedLoad32(&notifier->readOnlyError)) != VDO_SUCCESS);
}

/**********************************************************************/
int registerReadOnlyListener(ReadOnlyNotifier     *notifier,
                             void                 *listener,
                             ReadOnlyNotification *notification,
                             ThreadID              threadID)
{
  ReadOnlyListener *readOnlyListener;
  int result = ALLOCATE(1, ReadOnlyListener, __func__, &readOnlyListener);
  if (result != VDO_SUCCESS) {
    return result;
  }

  ThreadData *threadData = &notifier->threadData[threadID];
  *readOnlyListener = (ReadOnlyListener) {
    .listener = listener,
    .notify   = notification,
    .next     = threadData->listeners,
  };

  threadData->listeners = readOnlyListener;
  return VDO_SUCCESS;
}