/*

 * 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/uds-releases/jasper/src/uds/indexRouter.c#7 $

 */

#include "indexRouter.h"

#include "compiler.h"

#include "indexCheckpoint.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "requestQueue.h"

#include "zone.h"

/**

 * This is the request processing function invoked by the zone's RequestQueue

 * worker thread.

 *

 * @param request  the request to be indexed or executed by the zone worker

 **/

static void executeZoneRequest(Request *request)

{

  executeIndexRouterRequest(request->router, request);

}

/**

 * Construct and enqueue asynchronous control messages to add the chapter

 * index for a given virtual chapter to the sparse chapter index cache.

 *

 * @param router          the router containing the relevant queues

 * @param index           the index with the relevant cache and chapter

 * @param virtualChapter  the virtual chapter number of the chapter to cache

 **/

static void enqueueBarrierMessages(IndexRouter *router,

                                   Index       *index,

                                   uint64_t     virtualChapter)

{

  ZoneMessage barrier = {

    .index = index,

    .data = {

      .barrier = {

        .virtualChapter = virtualChapter,

      }

    }

  };

  unsigned int zone;

  for (zone = 0; zone < router->zoneCount; zone++) {

    int result = launchZoneControlMessage(REQUEST_SPARSE_CACHE_BARRIER,

                                          barrier, zone, router);

    ASSERT_LOG_ONLY((result == UDS_SUCCESS), "barrier message allocation");

  }

}

/**

 * This is the request processing function for the triage stage queue. Each

 * request is resolved in the master index, determining if it is a hook or

 * not, and if a hook, what virtual chapter (if any) it might be found in. If

 * a virtual chapter is found, this enqueues a sparse chapter cache barrier in

 * every zone before enqueueing the request in its zone.

 *

 * @param request  the request to triage

 **/

static void triageRequest(Request *request)

{

  IndexRouter *router = request->router;

  Index *index = router->index;

  // Check if the name is a hook in the index pointing at a sparse chapter.

  uint64_t sparseVirtualChapter = triageIndexRequest(index, request);

  if (sparseVirtualChapter != UINT64_MAX) {

    // Generate and place a barrier request on every zone queue.

    enqueueBarrierMessages(router, index, sparseVirtualChapter);

  }

  enqueueRequest(request, STAGE_INDEX);

}

/**

 * Initialize the zone queues and the triage queue.

 *

 * @param router    the router containing the queues

 * @param geometry  the geometry governing the indexes

 *

 * @return  UDS_SUCCESS or error code

 **/

static int initializeLocalIndexQueues(IndexRouter    *router,

                                      const Geometry *geometry)

{

  unsigned int i;

  for (i = 0; i < router->zoneCount; i++) {

    int result = makeRequestQueue("indexW", &executeZoneRequest,

                                  &router->zoneQueues[i]);

    if (result != UDS_SUCCESS) {

      return result;

    }

  }

  // The triage queue is only needed for sparse multi-zone indexes.

  if ((router->zoneCount > 1) && isSparse(geometry)) {

    int result = makeRequestQueue("triageW", &triageRequest,

                                  &router->triageQueue);

    if (result != UDS_SUCCESS) {

      return result;

    }

  }

  return UDS_SUCCESS;

}

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

static INLINE RequestQueue *getZoneQueue(IndexRouter  *router,

                                         unsigned int  zoneNumber)

{

  return router->zoneQueues[zoneNumber];

}

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

int makeIndexRouter(IndexLayout                  *layout,

                    const Configuration          *config,

                    const struct uds_parameters  *userParams,

                    LoadType                      loadType,

                    IndexLoadContext             *loadContext,

                    IndexRouterCallback           callback,

                    IndexRouter                 **routerPtr)

{

  unsigned int zoneCount = getZoneCount(userParams);

  IndexRouter *router;

  int result = ALLOCATE_EXTENDED(IndexRouter, zoneCount, RequestQueue *,

                                 "index router", &router);

  if (result != UDS_SUCCESS) {

    return result;

  }

  router->callback  = callback;

  router->zoneCount = zoneCount;

  result = initializeLocalIndexQueues(router, config->geometry);

  if (result != UDS_SUCCESS) {

    freeIndexRouter(router);

    return result;

  }

  result = makeIndex(layout, config, userParams, router->zoneCount, loadType,

                     loadContext, &router->index);

  if (result != UDS_SUCCESS) {

    freeIndexRouter(router);

    return logErrorWithStringError(result, "failed to create index");

  }

  router->needToSave = (router->index->loadedType != LOAD_LOAD);

  *routerPtr = router;

  return UDS_SUCCESS;

}

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

int saveIndexRouter(IndexRouter *router)

{

  if (!router->needToSave) {

    return UDS_SUCCESS;

  }

  int result = saveIndex(router->index);

  router->needToSave = (result != UDS_SUCCESS);

  return result;

}

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

void freeIndexRouter(IndexRouter *router)

{

  if (router == NULL) {

    return;

  }

  requestQueueFinish(router->triageQueue);

  unsigned int i;

  for (i = 0; i < router->zoneCount; i++) {

    requestQueueFinish(router->zoneQueues[i]);

  }

  freeIndex(router->index);

  FREE(router);

}

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

RequestQueue *selectIndexRouterQueue(IndexRouter  *router,

                                     Request      *request,

                                     RequestStage  nextStage)

{

  if (request->isControlMessage) {

    return getZoneQueue(router, request->zoneNumber);

  }

  if (nextStage == STAGE_TRIAGE) {

    // The triage queue is only needed for multi-zone sparse indexes and won't

    // be allocated by the router if not needed, so simply check for NULL.

    if (router->triageQueue != NULL) {

      return router->triageQueue;

    }

    // Dense index or single zone, so route it directly to the zone queue.

  } else if (nextStage != STAGE_INDEX) {

    ASSERT_LOG_ONLY(false, "invalid index stage: %d", nextStage);

    return NULL;

  }

  Index *index = router->index;

  request->zoneNumber = getMasterIndexZone(index->masterIndex,

                                           &request->chunkName);

  return getZoneQueue(router, request->zoneNumber);

}

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

void executeIndexRouterRequest(IndexRouter *router, Request *request)

{

  if (request->isControlMessage) {

    int result = dispatchIndexZoneControlRequest(request);

    if (result != UDS_SUCCESS) {

      logErrorWithStringError(result, "error executing control message: %d",

                              request->action);

    }

    request->status = result;

    enterCallbackStage(request);

    return;

  }

  router->needToSave = true;

  if (request->requeued && !isSuccessful(request->status)) {

    request->status = makeUnrecoverable(request->status);

    router->callback(request);

    return;

  }

  Index *index = router->index;

  int result = dispatchIndexRequest(index, request);

  if (result == UDS_QUEUED) {

    // Take the request off the pipeline.

    return;

  }

  request->status = result;

  router->callback(request);

}