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

#include "vio.h"

#include "logger.h"

#include "dataVIO.h"
#include "vdoInternal.h"

#ifdef __KERNEL__
#include <linux/ratelimit.h>
#endif

/**********************************************************************/
void freeVIO(VIO **vioPtr)
{
  VIO *vio = *vioPtr;
  if (vio == NULL) {
    return;
  }

  vio->completion.layer->freeVIO(vioPtr);
}

/**********************************************************************/
void initializeVIO(VIO           *vio,
                   VIOType        type,
                   VIOPriority    priority,
                   VDOCompletion *parent,
                   VDO           *vdo,
                   PhysicalLayer *layer)
{
  vio->vdo      = vdo;
  vio->type     = type;
  vio->priority = priority;

  VDOCompletion *completion = vioAsCompletion(vio);
  initializeCompletion(completion, VIO_COMPLETION, layer);
  completion->parent = parent;
}

/**********************************************************************/
void vioDoneCallback(VDOCompletion *completion)
{
  VIO *vio = asVIO(completion);
  completion->callback     = vio->callback;
  completion->errorHandler = vio->errorHandler;
  completeCompletion(completion);
}

/**********************************************************************/
const char *getVIOReadWriteFlavor(const VIO *vio)
{
  if (isReadVIO(vio)) {
    return "read";
  }
  return (isWriteVIO(vio) ? "write" : "read-modify-write");
}

/**********************************************************************/
void updateVIOErrorStats(VIO *vio, const char *format, ...)
{
  int priority;
  int result = vioAsCompletion(vio)->result;
  switch (result) {
  case VDO_READ_ONLY:
    atomicAdd64(&vio->vdo->errorStats.readOnlyErrorCount, 1);
    return;

  case VDO_NO_SPACE:
    atomicAdd64(&vio->vdo->errorStats.noSpaceErrorCount, 1);
    priority = LOG_DEBUG;
    break;

  default:
    priority = LOG_ERR;
  }

#ifdef __KERNEL__
  static DEFINE_RATELIMIT_STATE(errorLimiter, DEFAULT_RATELIMIT_INTERVAL,
                                DEFAULT_RATELIMIT_BURST);

  if (!__ratelimit(&errorLimiter)) {
    return;
  }
#endif

  va_list args;
  va_start(args, format);
  vLogWithStringError(priority, result, format, args);
  va_end(args);
}

/**
 * Handle an error from a metadata I/O.
 *
 * @param completion  The VIO
 **/
static void handleMetadataIOError(VDOCompletion *completion)
{
  VIO *vio = asVIO(completion);
  updateVIOErrorStats(vio,
                      "Completing %s VIO of type %u for physical block %"
                       PRIu64 " with error",
                       getVIOReadWriteFlavor(vio), vio->type, vio->physical);
  vioDoneCallback(completion);
}

/**********************************************************************/
void launchMetadataVIO(VIO                 *vio,
                       PhysicalBlockNumber  physical,
                       VDOAction           *callback,
                       VDOAction           *errorHandler,
                       VIOOperation         operation)
{
  vio->operation    = operation;
  vio->physical     = physical;
  vio->callback     = callback;
  vio->errorHandler = errorHandler;

  VDOCompletion *completion = vioAsCompletion(vio);
  resetCompletion(completion);
  completion->callback     = vioDoneCallback;
  completion->errorHandler = handleMetadataIOError;

  if (isReadVIO(vio)) {
    completion->layer->readMetadata(vio);
  } else {
    completion->layer->writeMetadata(vio);
  }
}

/**
 * Handle a flush error.
 *
 * @param completion  The flush VIO
 **/
static void handleFlushError(VDOCompletion *completion)
{
  logErrorWithStringError(completion->result, "Error flushing layer");
  completion->errorHandler = asVIO(completion)->errorHandler;
  completeCompletion(completion);
}

/**********************************************************************/
void launchFlush(VIO *vio, VDOAction *callback, VDOAction *errorHandler)
{
  VDOCompletion *completion = vioAsCompletion(vio);
  resetCompletion(completion);
  completion->callback     = callback;
  completion->errorHandler = handleFlushError;
  vio->errorHandler        = errorHandler;
  vio->operation           = VIO_FLUSH_BEFORE;
  vio->physical            = ZERO_BLOCK;

  PhysicalLayer *layer = completion->layer;
  if (layer->getWritePolicy(layer) == WRITE_POLICY_SYNC) {
    // XXX It is dangerous to be subtly dropping flushes possibly
    // needed for correctness in sync mode.
    finishCompletion(completion, VDO_SUCCESS);
    return;
  }

  layer->flush(vio);
}