/*
 * %Copyright%
 *
 * 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/uds-releases/jasper/src/uds/indexSession.c#10 $
 */

#include "indexSession.h"

#include "indexCheckpoint.h"
#include "indexRouter.h"
#include "logger.h"
#include "memoryAlloc.h"
#include "requestQueue.h"

/**********************************************************************/
static void collectStats(const struct uds_index_session *indexSession,
                         UdsContextStats                *stats)
{
  const SessionStats *sessionStats = &indexSession->stats;

  stats->currentTime = asTimeT(currentTime(CLOCK_REALTIME));

  stats->postsFound         = READ_ONCE(sessionStats->postsFound);
  stats->inMemoryPostsFound = READ_ONCE(sessionStats->postsFoundOpenChapter);
  stats->densePostsFound    = READ_ONCE(sessionStats->postsFoundDense);
  stats->sparsePostsFound   = READ_ONCE(sessionStats->postsFoundSparse);
  stats->postsNotFound      = READ_ONCE(sessionStats->postsNotFound);
  stats->updatesFound       = READ_ONCE(sessionStats->updatesFound);
  stats->updatesNotFound    = READ_ONCE(sessionStats->updatesNotFound);
  stats->deletionsFound     = READ_ONCE(sessionStats->deletionsFound);
  stats->deletionsNotFound  = READ_ONCE(sessionStats->deletionsNotFound);
  stats->queriesFound       = READ_ONCE(sessionStats->queriesFound);
  stats->queriesNotFound    = READ_ONCE(sessionStats->queriesNotFound);
  stats->requests           = READ_ONCE(sessionStats->requests);
}

/**********************************************************************/
static void handleCallbacks(Request *request)
{
  if (request->status == UDS_SUCCESS) {
    // Measure the turnaround time of this request and include that time,
    // along with the rest of the request, in the context's StatCounters.
    updateRequestContextStats(request);
  }

  if (request->callback != NULL) {
    // The request has specified its own callback and does not expect to be
    // freed.
    struct uds_index_session *indexSession = request->session;
    request->found = (request->location != LOC_UNAVAILABLE);
    request->callback((UdsRequest *) request);
    // We do this release after the callback because of the contract of the
    // udsFlushIndexSession method.
    releaseIndexSession(indexSession);
    return;
  }

  // Should not get here, because this is either a control message or it has a
  // callback method.
  freeRequest(request);
}

/**********************************************************************/
int checkIndexSession(struct uds_index_session *indexSession)
{
  lockMutex(&indexSession->requestMutex);
  unsigned int state = indexSession->state;
  unlockMutex(&indexSession->requestMutex);

  if (state == IS_FLAG_LOADED) {
    return UDS_SUCCESS;
  } else if (state & IS_FLAG_DISABLED) {
    return UDS_DISABLED;
  } else if ((state & IS_FLAG_LOADING)
             || (state & IS_FLAG_SUSPENDED)
             || (state & IS_FLAG_WAITING)) {
    return UDS_SUSPENDED;
  }

   return UDS_NO_INDEXSESSION;
}

/**********************************************************************/
int getIndexSession(struct uds_index_session *indexSession)
{
  lockMutex(&indexSession->requestMutex);
  indexSession->requestCount++;
  unlockMutex(&indexSession->requestMutex);

  int result = checkIndexSession(indexSession);
  if (result != UDS_SUCCESS) {
    releaseIndexSession(indexSession);
    return result;
  }
  return UDS_SUCCESS;
}

/**********************************************************************/
void releaseIndexSession(struct uds_index_session *indexSession)
{
  lockMutex(&indexSession->requestMutex);
  if (--indexSession->requestCount == 0) {
    broadcastCond(&indexSession->requestCond);
  }
  unlockMutex(&indexSession->requestMutex);
}

/**********************************************************************/
int startLoadingIndexSession(struct uds_index_session *indexSession)
{
  int result;
  lockMutex(&indexSession->requestMutex);
  if (indexSession->state & IS_FLAG_SUSPENDED) {
    result = UDS_SUSPENDED;
  } else if (indexSession->state != 0) {
    result = UDS_INDEXSESSION_IN_USE;
  } else {
    indexSession->state |= IS_FLAG_LOADING;
    result = UDS_SUCCESS;
  }
  unlockMutex(&indexSession->requestMutex);
  return result;
}

/**********************************************************************/
void finishLoadingIndexSession(struct uds_index_session *indexSession,
                               int                       result)
{
  lockMutex(&indexSession->requestMutex);
  indexSession->state &= ~IS_FLAG_LOADING;
  if (result == UDS_SUCCESS) {
    indexSession->state |= IS_FLAG_LOADED;
  }
  broadcastCond(&indexSession->requestCond);
  unlockMutex(&indexSession->requestMutex);
}

/**********************************************************************/
void disableIndexSession(struct uds_index_session *indexSession)
{
  lockMutex(&indexSession->requestMutex);
  indexSession->state |= IS_FLAG_DISABLED;
  unlockMutex(&indexSession->requestMutex);
}

/**********************************************************************/
int makeEmptyIndexSession(struct uds_index_session **indexSessionPtr)
{
  struct uds_index_session *session;
  int result = ALLOCATE(1, struct uds_index_session, __func__, &session);
  if (result != UDS_SUCCESS) {
    return result;
  }

  result = initMutex(&session->requestMutex);
  if (result != UDS_SUCCESS) {
    FREE(session);
    return result;
  }

  result = initCond(&session->requestCond);
  if (result != UDS_SUCCESS) {
    destroyMutex(&session->requestMutex);
    FREE(session);
    return result;
  }

  result = initMutex(&session->loadContext.mutex);
  if (result != UDS_SUCCESS) {
    destroyCond(&session->requestCond);
    destroyMutex(&session->requestMutex);
    FREE(session);
    return result;
  }

  result = initCond(&session->loadContext.cond);
  if (result != UDS_SUCCESS) {
    destroyMutex(&session->loadContext.mutex);
    destroyCond(&session->requestCond);
    destroyMutex(&session->requestMutex);
    FREE(session);
    return result;
  }

  result = makeRequestQueue("callbackW", &handleCallbacks,
                            &session->callbackQueue);
  if (result != UDS_SUCCESS) {
    destroyCond(&session->loadContext.cond);
    destroyMutex(&session->loadContext.mutex);
    destroyCond(&session->requestCond);
    destroyMutex(&session->requestMutex);
    FREE(session);
    return result;
  }

  *indexSessionPtr = session;
  return UDS_SUCCESS;
}

/**********************************************************************/
int udsSuspendIndexSession(struct uds_index_session *session, bool save)
{
  int result;
  bool saveIndex = false;
  bool suspendIndex = false;
  lockMutex(&session->requestMutex);
  // Wait for any pending close operation to complete.
  while (session->state & IS_FLAG_CLOSING) {
    waitCond(&session->requestCond, &session->requestMutex);
  }
  if ((session->state & IS_FLAG_WAITING)
      || (session->state & IS_FLAG_DESTROYING)) {
    result = EBUSY;
  } else if (session->state & IS_FLAG_SUSPENDED) {
    result = UDS_SUCCESS;
  } else if (session->state & IS_FLAG_LOADING) {
    session->state |= IS_FLAG_WAITING;
    suspendIndex = true;
    result = UDS_SUCCESS;
  } else if (!(session->state & IS_FLAG_LOADED)) {
    session->state |= IS_FLAG_SUSPENDED;
    broadcastCond(&session->requestCond);
    result = UDS_SUCCESS;
  } else {
    saveIndex = save;
    if (saveIndex) {
      session->state |= IS_FLAG_WAITING;
    } else {
      session->state |= IS_FLAG_SUSPENDED;
      broadcastCond(&session->requestCond);
    }
    result = UDS_SUCCESS;
  }
  unlockMutex(&session->requestMutex);

  if (!saveIndex && !suspendIndex) {
    return result;
  }

  if (saveIndex) {
    result = udsSaveIndex(session);
    lockMutex(&session->requestMutex);
    session->state &= ~IS_FLAG_WAITING;
    session->state |= IS_FLAG_SUSPENDED;
    broadcastCond(&session->requestCond);
    unlockMutex(&session->requestMutex);
    return result;
  }

  lockMutex(&session->loadContext.mutex);
  switch (session->loadContext.status) {
  case INDEX_OPENING:
    session->loadContext.status = INDEX_SUSPENDING;

    // Wait until the index indicates that it is not replaying.
    while ((session->loadContext.status != INDEX_SUSPENDED)
           && (session->loadContext.status != INDEX_READY)) {
      waitCond(&session->loadContext.cond,
               &session->loadContext.mutex);
    }
    break;

  case INDEX_READY:
    // Index load does not need to be suspended.
    break;

  case INDEX_SUSPENDED:
  case INDEX_SUSPENDING:
  case INDEX_FREEING:
  default:
    // These cases should not happen.
    ASSERT_LOG_ONLY(false, "Bad load context state %u",
                    session->loadContext.status);
    break;
  }
  unlockMutex(&session->loadContext.mutex);

  lockMutex(&session->requestMutex);
  session->state &= ~IS_FLAG_WAITING;
  session->state |= IS_FLAG_SUSPENDED;
  broadcastCond(&session->requestCond);
  unlockMutex(&session->requestMutex);
  return UDS_SUCCESS;
}

/**********************************************************************/
int udsResumeIndexSession(struct uds_index_session *session)
{
  lockMutex(&session->requestMutex);
  if (session->state & IS_FLAG_WAITING) {
    unlockMutex(&session->requestMutex);
    return EBUSY;
  }

  /* If not suspended, just succeed */
  if (!(session->state & IS_FLAG_SUSPENDED)) {
    unlockMutex(&session->requestMutex);
    return UDS_SUCCESS;
  }

  if (!(session->state & IS_FLAG_LOADING)) {
    session->state &= ~IS_FLAG_SUSPENDED;
    unlockMutex(&session->requestMutex);
    return UDS_SUCCESS;
  }

  session->state |= IS_FLAG_WAITING;
  unlockMutex(&session->requestMutex);

  lockMutex(&session->loadContext.mutex);
  switch (session->loadContext.status) {
  case INDEX_SUSPENDED:
    session->loadContext.status = INDEX_OPENING;
    // Notify the index to start replaying again.
    broadcastCond(&session->loadContext.cond);
    break;

  case INDEX_READY:
    // There is no index rebuild to resume.
    break;

  case INDEX_OPENING:
  case INDEX_SUSPENDING:
  case INDEX_FREEING:
  default:
    // These cases should not happen; do nothing.
    ASSERT_LOG_ONLY(false, "Bad load context state %u",
                    session->loadContext.status);
    break;
  }
  unlockMutex(&session->loadContext.mutex);

  lockMutex(&session->requestMutex);
  session->state &= ~IS_FLAG_WAITING;
  session->state &= ~IS_FLAG_SUSPENDED;
  broadcastCond(&session->requestCond);
  unlockMutex(&session->requestMutex);
  return UDS_SUCCESS;
}

/**********************************************************************/
static void waitForNoRequestsInProgress(struct uds_index_session *indexSession)
{
  lockMutex(&indexSession->requestMutex);
  while (indexSession->requestCount > 0) {
    waitCond(&indexSession->requestCond, &indexSession->requestMutex);
  }
  unlockMutex(&indexSession->requestMutex);
}

/**********************************************************************/
int saveAndFreeIndex(struct uds_index_session *indexSession)
{
  int result = UDS_SUCCESS;
  IndexRouter *router = indexSession->router;
  if (router != NULL) {
    lockMutex(&indexSession->requestMutex);
    bool suspended = (indexSession->state & IS_FLAG_SUSPENDED);
    unlockMutex(&indexSession->requestMutex);
    if (!suspended) {
      result = saveIndexRouter(router);
      if (result != UDS_SUCCESS) {
        logWarningWithStringError(result, "ignoring error from saveIndexRouter");
      }
    }
    freeIndexRouter(router);
    indexSession->router = NULL;

    // Reset all index state that happens to be in the index session, so it
    // doesn't affect any future index.
    lockMutex(&indexSession->loadContext.mutex);
    indexSession->loadContext.status = INDEX_OPENING;
    unlockMutex(&indexSession->loadContext.mutex);

    lockMutex(&indexSession->requestMutex);
    // Only the suspend bit will remain relevant.
    indexSession->state &= IS_FLAG_SUSPENDED;
    unlockMutex(&indexSession->requestMutex);
  }

  logDebug("Closed index");
  return result;
}

/**********************************************************************/
int udsCloseIndex(struct uds_index_session *indexSession)
{
  lockMutex(&indexSession->requestMutex);

  // Wait for any pending suspend, resume or close operations to complete.
  while ((indexSession->state & IS_FLAG_WAITING)
         || (indexSession->state & IS_FLAG_CLOSING)) {
    waitCond(&indexSession->requestCond, &indexSession->requestMutex);
  }

  int result = UDS_SUCCESS;
  if (indexSession->state & IS_FLAG_SUSPENDED) {
    result = UDS_SUSPENDED;
  } else if ((indexSession->state & IS_FLAG_DESTROYING)
             || !(indexSession->state & IS_FLAG_LOADED)) {
    // The index doesn't exist, hasn't finished loading, or is being destroyed.
    result = UDS_NO_INDEXSESSION;
  } else {
    indexSession->state |= IS_FLAG_CLOSING;
  }
  unlockMutex(&indexSession->requestMutex);
  if (result != UDS_SUCCESS) {
    return result;
  }

  logDebug("Closing index");
  waitForNoRequestsInProgress(indexSession);
  result = saveAndFreeIndex(indexSession);

  lockMutex(&indexSession->requestMutex);
  indexSession->state &= ~IS_FLAG_CLOSING;
  broadcastCond(&indexSession->requestCond);
  unlockMutex(&indexSession->requestMutex);
  return result;
}

/**********************************************************************/
int udsDestroyIndexSession(struct uds_index_session *indexSession)
{
  logDebug("Destroying index session");

  bool loadPending = false;
  lockMutex(&indexSession->requestMutex);

  // Wait for any pending suspend, resume, or close operations to complete.
  while ((indexSession->state & IS_FLAG_WAITING)
         || (indexSession->state & IS_FLAG_CLOSING)) {
    waitCond(&indexSession->requestCond, &indexSession->requestMutex);
  }

  if (indexSession->state & IS_FLAG_DESTROYING) {
    unlockMutex(&indexSession->requestMutex);
    return EBUSY;
  }

  indexSession->state |= IS_FLAG_DESTROYING;
  loadPending = ((indexSession->state & IS_FLAG_LOADING)
                 && (indexSession->state & IS_FLAG_SUSPENDED));
  unlockMutex(&indexSession->requestMutex);

  if (loadPending) {
    // Tell the index to terminate the rebuild.
    lockMutex(&indexSession->loadContext.mutex);
    if (indexSession->loadContext.status == INDEX_SUSPENDED) {
      indexSession->loadContext.status = INDEX_FREEING;
      broadcastCond(&indexSession->loadContext.cond);
    }
    unlockMutex(&indexSession->loadContext.mutex);

    // Wait until the load exits before proceeding.
    lockMutex(&indexSession->requestMutex);
    while (indexSession->state & IS_FLAG_LOADING) {
      waitCond(&indexSession->requestCond, &indexSession->requestMutex);
    }
    unlockMutex(&indexSession->requestMutex);
  }

  waitForNoRequestsInProgress(indexSession);
  int result = saveAndFreeIndex(indexSession);
  requestQueueFinish(indexSession->callbackQueue);
  indexSession->callbackQueue = NULL;
  destroyCond(&indexSession->loadContext.cond);
  destroyMutex(&indexSession->loadContext.mutex);
  destroyCond(&indexSession->requestCond);
  destroyMutex(&indexSession->requestMutex);
  logDebug("Destroyed index session");
  FREE(indexSession);
  return result;
}

/**********************************************************************/
int udsFlushIndexSession(struct uds_index_session *indexSession)
{
  waitForNoRequestsInProgress(indexSession);
  // Wait until any open chapter writes are complete
  waitForIdleIndexRouter(indexSession->router);
  return UDS_SUCCESS;
}

/**********************************************************************/
int udsSaveIndex(struct uds_index_session *indexSession)
{
  waitForNoRequestsInProgress(indexSession);
  // saveIndexRouter waits for open chapter writes to complete
  return saveIndexRouter(indexSession->router);
}

/**********************************************************************/
int udsSetCheckpointFrequency(struct uds_index_session *indexSession,
                              unsigned int              frequency)
{
  setIndexCheckpointFrequency(indexSession->router->index->checkpoint,
                              frequency);
  return UDS_SUCCESS;
}

/**********************************************************************/
int udsGetIndexConfiguration(struct uds_index_session *indexSession,
                             UdsConfiguration         *conf)
{
  if (conf == NULL) {
    return logErrorWithStringError(UDS_CONF_PTR_REQUIRED,
                                   "received a NULL config pointer");
  }
  int result = ALLOCATE(1, struct udsConfiguration, __func__, conf);
  if (result == UDS_SUCCESS) {
    **conf = indexSession->userConfig;
  }
  return result;
}

/**********************************************************************/
int udsGetIndexStats(struct uds_index_session *indexSession,
                     UdsIndexStats            *stats)
{
  if (stats == NULL) {
    return logErrorWithStringError(UDS_INDEX_STATS_PTR_REQUIRED,
                                   "received a NULL index stats pointer");
  }
  getIndexStats(indexSession->router->index, stats);
  return UDS_SUCCESS;
}

/**********************************************************************/
int udsGetIndexSessionStats(struct uds_index_session *indexSession,
                            UdsContextStats          *stats)
{
  if (stats == NULL) {
    return logWarningWithStringError(UDS_CONTEXT_STATS_PTR_REQUIRED,
                                     "received a NULL context stats pointer");
  }
  collectStats(indexSession, stats);
  return UDS_SUCCESS;
}