/*

 * 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/kernel/kvdoFlush.c#6 $

 */

#include "kvdoFlush.h"

#include "logger.h"

#include "memoryAlloc.h"

#include "threadConfig.h"

#include "bio.h"

#include "ioSubmitter.h"

/**

 * A specific (concrete) encapsulation of flush requests.

 *

 * 
We attempt to allocate a KVDOFlush objects for each incoming flush bio.

 * In case the allocate fails, a spare object is pre-allocated by and stored

 * in the kernel layer. The first time an allocation fails, the spare is used.

 * If another allocation fails while the spare is in use, it will merely be

 * queued for later processing.

 *

 * 
When a KVDOFlush is complete, it will either be freed, immediately

 * re-used for queued flushes, or stashed in the kernel layer as the new spare

 * object. This ensures that we will always make forward progress.

 **/

struct kvdoFlush {

  KvdoWorkItem     workItem;

  KernelLayer     *layer;

  struct bio_list  bios;

  Jiffies          arrivalTime;  // Time when earliest bio appeared

  VDOFlush         vdoFlush;

};

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

int makeKVDOFlush(KVDOFlush **flushPtr)

{

  return ALLOCATE(1, KVDOFlush, __func__, flushPtr);

}

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

bool shouldProcessFlush(KernelLayer *layer)

{

  return (getKVDOWritePolicy(&layer->kvdo) != WRITE_POLICY_SYNC);

}

/**

 * Function call to handle an empty flush request from the request queue.

 *

 * @param item  The work item representing the flush request

 **/

static void kvdoFlushWork(KvdoWorkItem *item)

{

  KVDOFlush *kvdoFlush = container_of(item, KVDOFlush, workItem);

  flush(kvdoFlush->layer->kvdo.vdo, &kvdoFlush->vdoFlush);

}

/**

 * Initialize a KVDOFlush object, transferring all the bios in the kernel

 * layer's waitingFlushes list to it. The caller MUST already hold the layer's

 * flushLock.

 *

 * @param kvdoFlush  The flush to initialize

 * @param layer      The kernel layer on which the flushLock is held

 **/

static void initializeKVDOFlush(KVDOFlush *kvdoFlush, KernelLayer *layer)

{

  kvdoFlush->layer = layer;

  bio_list_init(&kvdoFlush->bios);

  bio_list_merge(&kvdoFlush->bios, &layer->waitingFlushes);

  bio_list_init(&layer->waitingFlushes);

  kvdoFlush->arrivalTime = layer->flushArrivalTime;

}

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

static void enqueueKVDOFlush(KVDOFlush *kvdoFlush)

{

  setupWorkItem(&kvdoFlush->workItem, kvdoFlushWork, NULL, REQ_Q_ACTION_FLUSH);

  KVDO *kvdo = &kvdoFlush->layer->kvdo;

  enqueueKVDOWork(kvdo, &kvdoFlush->workItem,

                  getPackerZoneThread(getThreadConfig(kvdo->vdo)));

}

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

void launchKVDOFlush(KernelLayer *layer, BIO *bio)

{

  // Try to allocate a KVDOFlush to represent the flush request. If the

  // allocation fails, we'll deal with it later.

  KVDOFlush *kvdoFlush = ALLOCATE_NOWAIT(KVDOFlush, __func__);

  spin_lock(&layer->flushLock);

  // We have a new bio to start.  Add it to the list.  If it becomes the

  // only entry on the list, record the time.

  if (bio_list_empty(&layer->waitingFlushes)) {

    layer->flushArrivalTime = jiffies;

  }

  bio_list_add(&layer->waitingFlushes, bio);

  if (kvdoFlush == NULL) {

    // The KVDOFlush allocation failed. Try to use the spare KVDOFlush object.

    if (layer->spareKVDOFlush == NULL) {

      // The spare is already in use. This bio is on waitingFlushes and it

      // will be handled by a flush completion or by a bio that can allocate.

      spin_unlock(&layer->flushLock);

      return;

    }

    // Take and use the spare KVDOFlush object.

    kvdoFlush = layer->spareKVDOFlush;

    layer->spareKVDOFlush = NULL;

  }

  // We have flushes to start. Capture them in the KVDOFlush object.

  initializeKVDOFlush(kvdoFlush, layer);

  spin_unlock(&layer->flushLock);

  // Finish launching the flushes.

  enqueueKVDOFlush(kvdoFlush);

}

/**

 * Release a KVDOFlush object that has completed its work. If there are any

 * pending flush requests whose KVDOFlush allocation failed, they will be

 * launched by immediately re-using the released KVDOFlush. If there is no

 * spare KVDOFlush, the released object will become the spare. Otherwise, the

 * KVDOFlush will be freed.

 *

 * @param kvdoFlush  The completed flush object to re-use or free

 **/

static void releaseKVDOFlush(KVDOFlush *kvdoFlush)

{

  KernelLayer *layer = kvdoFlush->layer;

  bool relaunchFlush = false;

  bool freeFlush     = false;

  spin_lock(&layer->flushLock);

  if (bio_list_empty(&layer->waitingFlushes)) {

    // Nothing needs to be started.  Save one spare KVDOFlush object.

    if (layer->spareKVDOFlush == NULL) {

      // Make the new spare all zero, just like a newly allocated one.

      memset(kvdoFlush, 0, sizeof(*kvdoFlush));

      layer->spareKVDOFlush = kvdoFlush;

    } else {

      freeFlush = true;

    }

  } else {

    // We have flushes to start.  Capture them in the KVDOFlush object.

    initializeKVDOFlush(kvdoFlush, layer);

    relaunchFlush = true;

  }

  spin_unlock(&layer->flushLock);

  if (relaunchFlush) {

    // Finish launching the flushes.

    enqueueKVDOFlush(kvdoFlush);

  } else if (freeFlush) {

    FREE(kvdoFlush);

  }

}

/**

 * Function called to complete and free a flush request

 *

 * @param item    The flush-request work item

 **/

static void kvdoCompleteFlushWork(KvdoWorkItem *item)

{

  KVDOFlush   *kvdoFlush = container_of(item, KVDOFlush, workItem);

  KernelLayer *layer     = kvdoFlush->layer;

  BIO *bio;

  while ((bio = bio_list_pop(&kvdoFlush->bios)) != NULL) {

    // We're not acknowledging this bio now, but we'll never touch it

    // again, so this is the last chance to account for it.

    countBios(&layer->biosAcknowledged, bio);

    // Make sure the bio is a empty flush bio.

    prepareFlushBIO(bio, bio->bi_private, getKernelLayerBdev(layer),

                    bio->bi_end_io);

    atomic64_inc(&layer->flushOut);

    generic_make_request(bio);

  }

  // Release the KVDOFlush object, freeing it, re-using it as the spare, or

  // using it to launch any flushes that had to wait when allocations failed.

  releaseKVDOFlush(kvdoFlush);

}

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

void kvdoCompleteFlush(VDOFlush **kfp)

{

  if (*kfp != NULL) {

    KVDOFlush *kvdoFlush = container_of(*kfp, KVDOFlush, vdoFlush);

    setupWorkItem(&kvdoFlush->workItem, kvdoCompleteFlushWork, NULL,

                  BIO_Q_ACTION_FLUSH);

    enqueueBioWorkItem(kvdoFlush->layer->ioSubmitter,

                       &kvdoFlush->workItem);

    *kfp = NULL;

  }

}

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

int synchronousFlush(KernelLayer *layer)

{

  BIO bio;

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

  bio_init(&bio, 0, 0);

#else

  bio_init(&bio;;

#endif

  int result = 0;

  prepareFlushBIO(&bio, layer, getKernelLayerBdev(layer), NULL);

  result = submitBioAndWait(&bio;;

  atomic64_inc(&layer->flushOut);

  if (result != 0) {

    logErrorWithStringError(result, "synchronous flush failed");

    result = -EIO;

  }

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

  bio_uninit(&bio;;

#endif

  return result;

}