/*

 * 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/chapterWriter.c#2 $

 */

#include "chapterWriter.h"

#include "errors.h"

#include "index.h"

#include "indexCheckpoint.h"

#include "indexComponent.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "openChapter.h"

#include "threads.h"

struct chapterWriter {

  /* The index to which we belong */

  Index            *index;

  /* The thread to do the writing */

  Thread            thread;

  /* lock protecting the following fields */

  Mutex             mutex;

  /* condition signalled on state changes */

  CondVar           cond;

  /* Set to true to stop the thread */

  bool              stop;

  /* The result from the most recent write */

  int               result;

  /* The number of bytes allocated by the chapter writer */

  size_t            memoryAllocated;

  /* The number of zones which have submitted a chapter for writing */

  unsigned int      zonesToWrite;

  /* Open chapter index used by closeOpenChapter() */

  OpenChapterIndex *openChapterIndex;

  /* Collated records used by closeOpenChapter() */

  UdsChunkRecord   *collatedRecords;

  /* The chapters to write (one per zone) */

  OpenChapterZone  *chapters[];

};

/**

 * This is the driver function for the writer thread. It loops until

 * terminated, waiting for a chapter to provided to close.

 **/

static void closeChapters(void *arg)

{

  ChapterWriter *writer = arg;

  logDebug("chapter writer starting");

  lockMutex(&writer->mutex);

  for (;;) {

    while (writer->zonesToWrite < writer->index->zoneCount) {

      if (writer->stop && (writer->zonesToWrite == 0)) {

        // We've been told to stop, and all of the zones are in the same

        // open chapter, so we can exit now.

        unlockMutex(&writer->mutex);

        logDebug("chapter writer stopping");

        return;

      }

      waitCond(&writer->cond, &writer->mutex);

    }

    /*

     * Release the lock while closing a chapter. We probably don't need to do

     * this, but it seems safer in principle. It's OK to access the chapter

     * and chapterNumber fields without the lock since those aren't allowed to

     * change until we're done.

     */

    unlockMutex(&writer->mutex);

    if (writer->index->hasSavedOpenChapter) {

      writer->index->hasSavedOpenChapter = false;

      /*

       * Remove the saved open chapter as that chapter is about to be written

       * to the volume.  This matters the first time we close the open chapter

       * after loading from a clean shutdown, or after doing a clean save.

       */

      IndexComponent *oc = findIndexComponent(writer->index->state,

                                              &OPEN_CHAPTER_INFO);

      int result = discardIndexComponent(oc);

      if (result == UDS_SUCCESS) {

        logDebug("Discarding saved open chapter");

      }

    }

    int result = closeOpenChapter(writer->chapters,

                                  writer->index->zoneCount,

                                  writer->index->volume,

                                  writer->openChapterIndex,

                                  writer->collatedRecords,

                                  writer->index->newestVirtualChapter);

    if (result == UDS_SUCCESS) {

      result = processChapterWriterCheckpointSaves(writer->index);

    }

    lockMutex(&writer->mutex);

    // Note that the index is totally finished with the writing chapter

    advanceActiveChapters(writer->index);

    writer->result       = result;

    writer->zonesToWrite = 0;

    broadcastCond(&writer->cond);

  }

}

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

int makeChapterWriter(Index                       *index,

                      const struct index_version  *indexVersion,

                      ChapterWriter              **writerPtr)

{

  size_t collatedRecordsSize

    = (sizeof(UdsChunkRecord)

       * (1 + index->volume->geometry->recordsPerChapter));

  ChapterWriter *writer;

  int result = ALLOCATE_EXTENDED(ChapterWriter,

                                 index->zoneCount, OpenChapterZone *,

                                 "Chapter Writer", &writer);

  if (result != UDS_SUCCESS) {

    return result;

  }

  writer->index = index;

  result = initMutex(&writer->mutex);

  if (result != UDS_SUCCESS) {

    FREE(writer);

    return result;

  }

  result = initCond(&writer->cond);

  if (result != UDS_SUCCESS) {

    destroyMutex(&writer->mutex);

    FREE(writer);

    return result;

  }

  // Now that we have the mutex+cond, it is safe to call freeChapterWriter.

  result = allocateCacheAligned(collatedRecordsSize, "collated records",

                                &writer->collatedRecords);

  if (result != UDS_SUCCESS) {

    freeChapterWriter(writer);

    return makeUnrecoverable(result);

  }

  result = makeOpenChapterIndex(&writer->openChapterIndex,

                                index->volume->geometry,

                                indexVersion->chapterIndexHeaderNativeEndian,

                                index->volume->nonce);

  if (result != UDS_SUCCESS) {

    freeChapterWriter(writer);

    return makeUnrecoverable(result);

  }

  size_t openChapterIndexMemoryAllocated

    = getOpenChapterIndexMemoryAllocated(writer->openChapterIndex);

  writer->memoryAllocated = (sizeof(ChapterWriter)

                             + index->zoneCount * sizeof(OpenChapterZone *)

                             + collatedRecordsSize

                             + openChapterIndexMemoryAllocated);

  // We're initialized, so now it's safe to start the writer thread.

  result = createThread(closeChapters, writer, "writer", &writer->thread);

  if (result != UDS_SUCCESS) {

    freeChapterWriter(writer);

    return makeUnrecoverable(result);

  }

  *writerPtr = writer;

  return UDS_SUCCESS;

}

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

void freeChapterWriter(ChapterWriter *writer)

{

  if (writer == NULL) {

    return;

  }

  int result __attribute__((unused)) = stopChapterWriter(writer);

  destroyMutex(&writer->mutex);

  destroyCond(&writer->cond);

  freeOpenChapterIndex(writer->openChapterIndex);

  FREE(writer->collatedRecords);

  FREE(writer);

}

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

unsigned int startClosingChapter(ChapterWriter   *writer,

                                 unsigned int     zoneNumber,

                                 OpenChapterZone *chapter)

{

  lockMutex(&writer->mutex);

  unsigned int finishedZones = ++writer->zonesToWrite;

  writer->chapters[zoneNumber] = chapter;

  broadcastCond(&writer->cond);

  unlockMutex(&writer->mutex);

  return finishedZones;

}

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

int finishPreviousChapter(ChapterWriter *writer, uint64_t currentChapterNumber)

{

  int result;

  lockMutex(&writer->mutex);

  while (writer->index->newestVirtualChapter < currentChapterNumber) {

    waitCond(&writer->cond, &writer->mutex);

  }

  result = writer->result;

  unlockMutex(&writer->mutex);

  if (result != UDS_SUCCESS) {

    return logUnrecoverable(result, "Writing of previous open chapter failed");

  }

  return UDS_SUCCESS;

}

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

void waitForIdleChapterWriter(ChapterWriter *writer)

{

  lockMutex(&writer->mutex);

  while (writer->zonesToWrite > 0) {

    // The chapter writer is probably writing a chapter.  If it is not, it will

    // soon wake up and write a chapter.

    waitCond(&writer->cond, &writer->mutex);

  }

  unlockMutex(&writer->mutex);

}

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

int stopChapterWriter(ChapterWriter *writer)

{

  Thread writerThread = 0;

  lockMutex(&writer->mutex);

  if (writer->thread != 0) {

    writerThread = writer->thread;

    writer->thread = 0;

    writer->stop = true;

    broadcastCond(&writer->cond);

  }

  int result = writer->result;

  unlockMutex(&writer->mutex);

  if (writerThread != 0) {

    joinThreads(writerThread);

  }

  if (result != UDS_SUCCESS) {

    return logUnrecoverable(result, "Writing of previous open chapter failed");

  }

  return UDS_SUCCESS;

}

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

size_t getChapterWriterMemoryAllocated(ChapterWriter *writer)

{

  return writer->memoryAllocated;

}