// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#include "librbd/journal/Replay.h"
#include "common/dout.h"
#include "common/errno.h"
#include "common/WorkQueue.h"
#include "librbd/ExclusiveLock.h"
#include "librbd/ImageCtx.h"
#include "librbd/ImageState.h"
#include "librbd/internal.h"
#include "librbd/Operations.h"
#include "librbd/Utils.h"
#include "librbd/io/AioCompletion.h"
#include "librbd/io/ImageRequest.h"

#define dout_subsys ceph_subsys_rbd
#undef dout_prefix
#define dout_prefix *_dout << "librbd::journal::Replay: " << this << " "

namespace librbd {
namespace journal {

namespace {

static const uint64_t IN_FLIGHT_IO_LOW_WATER_MARK(32);
static const uint64_t IN_FLIGHT_IO_HIGH_WATER_MARK(64);

static NoOpProgressContext no_op_progress_callback;

template <typename I, typename E>
struct ExecuteOp : public Context {
  I &image_ctx;
  E event;
  Context *on_op_complete;

  ExecuteOp(I &image_ctx, const E &event, Context *on_op_complete)
    : image_ctx(image_ctx), event(event), on_op_complete(on_op_complete) {
  }

  void execute(const journal::SnapCreateEvent &_) {
    image_ctx.operations->execute_snap_create(event.snap_namespace,
					      event.snap_name,
                                              on_op_complete,
                                              event.op_tid, false);
  }

  void execute(const journal::SnapRemoveEvent &_) {
    image_ctx.operations->execute_snap_remove(event.snap_namespace,
					      event.snap_name,
                                              on_op_complete);
  }

  void execute(const journal::SnapRenameEvent &_) {
    image_ctx.operations->execute_snap_rename(event.snap_id,
                                              event.dst_snap_name,
                                              on_op_complete);
  }

  void execute(const journal::SnapProtectEvent &_) {
    image_ctx.operations->execute_snap_protect(event.snap_namespace,
					       event.snap_name,
                                               on_op_complete);
  }

  void execute(const journal::SnapUnprotectEvent &_) {
    image_ctx.operations->execute_snap_unprotect(event.snap_namespace,
						 event.snap_name,
                                                 on_op_complete);
  }

  void execute(const journal::SnapRollbackEvent &_) {
    image_ctx.operations->execute_snap_rollback(event.snap_namespace,
						event.snap_name,
                                                no_op_progress_callback,
                                                on_op_complete);
  }

  void execute(const journal::RenameEvent &_) {
    image_ctx.operations->execute_rename(event.image_name,
                                         on_op_complete);
  }

  void execute(const journal::ResizeEvent &_) {
    image_ctx.operations->execute_resize(event.size, true, no_op_progress_callback,
                                         on_op_complete, event.op_tid);
  }

  void execute(const journal::FlattenEvent &_) {
    image_ctx.operations->execute_flatten(no_op_progress_callback,
                                          on_op_complete);
  }

  void execute(const journal::SnapLimitEvent &_) {
    image_ctx.operations->execute_snap_set_limit(event.limit, on_op_complete);
  }

  void execute(const journal::UpdateFeaturesEvent &_) {
    image_ctx.operations->execute_update_features(event.features, event.enabled,
						  on_op_complete, event.op_tid);
  }

  void execute(const journal::MetadataSetEvent &_) {
    image_ctx.operations->execute_metadata_set(event.key, event.value,
                                               on_op_complete);
  }

  void execute(const journal::MetadataRemoveEvent &_) {
    image_ctx.operations->execute_metadata_remove(event.key, on_op_complete);
  }

  void finish(int r) override {
    CephContext *cct = image_ctx.cct;
    if (r < 0) {
      lderr(cct) << ": ExecuteOp::" << __func__ << ": r=" << r << dendl;
      on_op_complete->complete(r);
      return;
    }

    ldout(cct, 20) << ": ExecuteOp::" << __func__ << dendl;
    RWLock::RLocker owner_locker(image_ctx.owner_lock);

    if (image_ctx.exclusive_lock == nullptr ||
        !image_ctx.exclusive_lock->accept_ops()) {
      ldout(cct, 5) << ": lost exclusive lock -- skipping op" << dendl;
      on_op_complete->complete(-ECANCELED);
      return;
    }

    execute(event);
  }
};

template <typename I>
struct C_RefreshIfRequired : public Context {
  I &image_ctx;
  Context *on_finish;

  C_RefreshIfRequired(I &image_ctx, Context *on_finish)
    : image_ctx(image_ctx), on_finish(on_finish) {
  }
  ~C_RefreshIfRequired() override {
    delete on_finish;
  }

  void finish(int r) override {
    CephContext *cct = image_ctx.cct;
    Context *ctx = on_finish;
    on_finish = nullptr;

    if (r < 0) {
      lderr(cct) << ": C_RefreshIfRequired::" << __func__ << ": r=" << r << dendl;
      image_ctx.op_work_queue->queue(ctx, r);
      return;
    }

    if (image_ctx.state->is_refresh_required()) {
      ldout(cct, 20) << ": C_RefreshIfRequired::" << __func__ << ": "
                     << "refresh required" << dendl;
      image_ctx.state->refresh(ctx);
      return;
    }

    image_ctx.op_work_queue->queue(ctx, 0);
  }
};

} // anonymous namespace

#undef dout_prefix
#define dout_prefix *_dout << "librbd::journal::Replay: " << this << " " \
                           << __func__

template <typename I>
Replay<I>::Replay(I &image_ctx)
  : m_image_ctx(image_ctx), m_lock("Replay<I>::m_lock") {
}

template <typename I>
Replay<I>::~Replay() {
  assert(m_in_flight_aio_flush == 0);
  assert(m_in_flight_aio_modify == 0);
  assert(m_aio_modify_unsafe_contexts.empty());
  assert(m_aio_modify_safe_contexts.empty());
  assert(m_op_events.empty());
  assert(m_in_flight_op_events == 0);
}

template <typename I>
int Replay<I>::decode(bufferlist::iterator *it, EventEntry *event_entry) {
  try {
    ::decode(*event_entry, *it);
  } catch (const buffer::error &err) {
    return -EBADMSG;
  }
  return 0;
}

template <typename I>
void Replay<I>::process(const EventEntry &event_entry,
                        Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": on_ready=" << on_ready << ", on_safe=" << on_safe
                 << dendl;

  on_ready = util::create_async_context_callback(m_image_ctx, on_ready);

  RWLock::RLocker owner_lock(m_image_ctx.owner_lock);
  if (m_image_ctx.exclusive_lock == nullptr ||
      !m_image_ctx.exclusive_lock->accept_ops()) {
    ldout(cct, 5) << ": lost exclusive lock -- skipping event" << dendl;
    m_image_ctx.op_work_queue->queue(on_safe, -ECANCELED);
    on_ready->complete(0);
    return;
  }

  boost::apply_visitor(EventVisitor(this, on_ready, on_safe),
                       event_entry.event);
}

template <typename I>
void Replay<I>::shut_down(bool cancel_ops, Context *on_finish) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << dendl;

  io::AioCompletion *flush_comp = nullptr;
  on_finish = util::create_async_context_callback(
    m_image_ctx, on_finish);

  {
    Mutex::Locker locker(m_lock);

    // safely commit any remaining AIO modify operations
    if ((m_in_flight_aio_flush + m_in_flight_aio_modify) != 0) {
      flush_comp = create_aio_flush_completion(nullptr);
      assert(flush_comp != nullptr);
    }

    for (auto &op_event_pair : m_op_events) {
      OpEvent &op_event = op_event_pair.second;
      if (cancel_ops) {
        // cancel ops that are waiting to start (waiting for
        // OpFinishEvent or waiting for ready)
        if (op_event.on_start_ready == nullptr &&
            op_event.on_op_finish_event != nullptr) {
          Context *on_op_finish_event = nullptr;
          std::swap(on_op_finish_event, op_event.on_op_finish_event);
          m_image_ctx.op_work_queue->queue(on_op_finish_event, -ERESTART);
        }
      } else if (op_event.on_op_finish_event != nullptr) {
        // start ops waiting for OpFinishEvent
        Context *on_op_finish_event = nullptr;
        std::swap(on_op_finish_event, op_event.on_op_finish_event);
        m_image_ctx.op_work_queue->queue(on_op_finish_event, 0);
      } else if (op_event.on_start_ready != nullptr) {
        // waiting for op ready
        op_event_pair.second.finish_on_ready = true;
      }
    }

    assert(!m_shut_down);
    m_shut_down = true;

    assert(m_flush_ctx == nullptr);
    if (m_in_flight_op_events > 0 || flush_comp != nullptr) {
      std::swap(m_flush_ctx, on_finish);
    }
  }

  // execute the following outside of lock scope
  if (flush_comp != nullptr) {
    RWLock::RLocker owner_locker(m_image_ctx.owner_lock);
    io::ImageRequest<I>::aio_flush(&m_image_ctx, flush_comp, {});
  }
  if (on_finish != nullptr) {
    on_finish->complete(0);
  }
}

template <typename I>
void Replay<I>::flush(Context *on_finish) {
  io::AioCompletion *aio_comp;
  {
    Mutex::Locker locker(m_lock);
    aio_comp = create_aio_flush_completion(
      util::create_async_context_callback(m_image_ctx, on_finish));
    if (aio_comp == nullptr) {
      return;
    }
  }

  RWLock::RLocker owner_locker(m_image_ctx.owner_lock);
  io::ImageRequest<I>::aio_flush(&m_image_ctx, aio_comp, {});
}

template <typename I>
void Replay<I>::replay_op_ready(uint64_t op_tid, Context *on_resume) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": op_tid=" << op_tid << dendl;

  Mutex::Locker locker(m_lock);
  auto op_it = m_op_events.find(op_tid);
  assert(op_it != m_op_events.end());

  OpEvent &op_event = op_it->second;
  assert(op_event.op_in_progress &&
         op_event.on_op_finish_event == nullptr &&
         op_event.on_finish_ready == nullptr &&
         op_event.on_finish_safe == nullptr);

  // resume processing replay events
  Context *on_start_ready = nullptr;
  std::swap(on_start_ready, op_event.on_start_ready);
  on_start_ready->complete(0);

  // cancel has been requested -- send error to paused state machine
  if (!op_event.finish_on_ready && m_flush_ctx != nullptr) {
    m_image_ctx.op_work_queue->queue(on_resume, -ERESTART);
    return;
  }

  // resume the op state machine once the associated OpFinishEvent
  // is processed
  op_event.on_op_finish_event = new FunctionContext(
    [on_resume](int r) {
      on_resume->complete(r);
    });

  // shut down request -- don't expect OpFinishEvent
  if (op_event.finish_on_ready) {
    m_image_ctx.op_work_queue->queue(on_resume, 0);
  }
}

template <typename I>
void Replay<I>::handle_event(const journal::AioDiscardEvent &event,
                             Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": AIO discard event" << dendl;

  bool flush_required;
  auto aio_comp = create_aio_modify_completion(on_ready, on_safe,
                                               io::AIO_TYPE_DISCARD,
                                               &flush_required,
                                               {});
  if (aio_comp == nullptr) {
    return;
  }

  io::ImageRequest<I>::aio_discard(&m_image_ctx, aio_comp, event.offset,
                                   event.length, event.skip_partial_discard,
				   {});
  if (flush_required) {
    m_lock.Lock();
    auto flush_comp = create_aio_flush_completion(nullptr);
    m_lock.Unlock();

    if (flush_comp != nullptr) {
      io::ImageRequest<I>::aio_flush(&m_image_ctx, flush_comp, {});
    }
  }
}

template <typename I>
void Replay<I>::handle_event(const journal::AioWriteEvent &event,
                             Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": AIO write event" << dendl;

  bufferlist data = event.data;
  bool flush_required;
  auto aio_comp = create_aio_modify_completion(on_ready, on_safe,
                                               io::AIO_TYPE_WRITE,
                                               &flush_required,
                                               {});
  if (aio_comp == nullptr) {
    return;
  }

  io::ImageRequest<I>::aio_write(&m_image_ctx, aio_comp,
                                 {{event.offset, event.length}},
                                 std::move(data), 0, {});
  if (flush_required) {
    m_lock.Lock();
    auto flush_comp = create_aio_flush_completion(nullptr);
    m_lock.Unlock();

    if (flush_comp != nullptr) {
      io::ImageRequest<I>::aio_flush(&m_image_ctx, flush_comp, {});
    }
  }
}

template <typename I>
void Replay<I>::handle_event(const journal::AioFlushEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": AIO flush event" << dendl;

  io::AioCompletion *aio_comp;
  {
    Mutex::Locker locker(m_lock);
    aio_comp = create_aio_flush_completion(on_safe);
  }

  if (aio_comp != nullptr) {
    io::ImageRequest<I>::aio_flush(&m_image_ctx, aio_comp, {});
  }
  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::AioWriteSameEvent &event,
                             Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": AIO writesame event" << dendl;

  bufferlist data = event.data;
  bool flush_required;
  auto aio_comp = create_aio_modify_completion(on_ready, on_safe,
                                               io::AIO_TYPE_WRITESAME,
                                               &flush_required,
                                               {});
  if (aio_comp == nullptr) {
    return;
  }

  io::ImageRequest<I>::aio_writesame(&m_image_ctx, aio_comp, event.offset,
                                     event.length, std::move(data), 0, {});
  if (flush_required) {
    m_lock.Lock();
    auto flush_comp = create_aio_flush_completion(nullptr);
    m_lock.Unlock();

    if (flush_comp != nullptr) {
      io::ImageRequest<I>::aio_flush(&m_image_ctx, flush_comp, {});
    }
  }
}

 template <typename I>
 void Replay<I>::handle_event(const journal::AioCompareAndWriteEvent &event,
                              Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": AIO CompareAndWrite event" << dendl;

  bufferlist cmp_data = event.cmp_data;
  bufferlist write_data = event.write_data;
  bool flush_required;
  auto aio_comp = create_aio_modify_completion(on_ready, on_safe,
                                               io::AIO_TYPE_COMPARE_AND_WRITE,
                                               &flush_required,
                                               {-EILSEQ});
  io::ImageRequest<I>::aio_compare_and_write(&m_image_ctx, aio_comp,
                                             {{event.offset, event.length}},
                                             std::move(cmp_data),
                                             std::move(write_data),
                                             nullptr, 0, {});
  if (flush_required) {
    m_lock.Lock();
    auto flush_comp = create_aio_flush_completion(nullptr);
    m_lock.Unlock();

    io::ImageRequest<I>::aio_flush(&m_image_ctx, flush_comp, {});
  }
}

template <typename I>
void Replay<I>::handle_event(const journal::OpFinishEvent &event,
                             Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Op finish event: "
                 << "op_tid=" << event.op_tid << dendl;

  bool op_in_progress;
  bool filter_ret_val;
  Context *on_op_complete = nullptr;
  Context *on_op_finish_event = nullptr;
  {
    Mutex::Locker locker(m_lock);
    auto op_it = m_op_events.find(event.op_tid);
    if (op_it == m_op_events.end()) {
      ldout(cct, 10) << ": unable to locate associated op: assuming previously "
                     << "committed." << dendl;
      on_ready->complete(0);
      m_image_ctx.op_work_queue->queue(on_safe, 0);
      return;
    }

    OpEvent &op_event = op_it->second;
    assert(op_event.on_finish_safe == nullptr);
    op_event.on_finish_ready = on_ready;
    op_event.on_finish_safe = on_safe;
    op_in_progress = op_event.op_in_progress;
    std::swap(on_op_complete, op_event.on_op_complete);
    std::swap(on_op_finish_event, op_event.on_op_finish_event);

    // special errors which indicate op never started but was recorded
    // as failed in the journal
    filter_ret_val = (op_event.op_finish_error_codes.count(event.r) != 0);
  }

  if (event.r < 0) {
    if (op_in_progress) {
      // bubble the error up to the in-progress op to cancel it
      on_op_finish_event->complete(event.r);
    } else {
      // op hasn't been started -- bubble the error up since
      // our image is now potentially in an inconsistent state
      // since simple errors should have been caught before
      // creating the op event
      delete on_op_complete;
      delete on_op_finish_event;
      handle_op_complete(event.op_tid, filter_ret_val ? 0 : event.r);
    }
    return;
  }

  // journal recorded success -- apply the op now
  on_op_finish_event->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::SnapCreateEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Snap create event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  // ignore errors caused due to replay
  op_event->ignore_error_codes = {-EEXIST};

  // avoid lock cycles
  m_image_ctx.op_work_queue->queue(new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::SnapCreateEvent>(m_image_ctx, event,
                                                            on_op_complete)),
    0);

  // do not process more events until the state machine is ready
  // since it will affect IO
  op_event->op_in_progress = true;
  op_event->on_start_ready = on_ready;
}

template <typename I>
void Replay<I>::handle_event(const journal::SnapRemoveEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Snap remove event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::SnapRemoveEvent>(m_image_ctx, event,
                                                            on_op_complete));

  // ignore errors caused due to replay
  op_event->ignore_error_codes = {-ENOENT};

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::SnapRenameEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Snap rename event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::SnapRenameEvent>(m_image_ctx, event,
                                                            on_op_complete));

  // ignore errors caused due to replay
  op_event->ignore_error_codes = {-EEXIST};

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::SnapProtectEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Snap protect event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::SnapProtectEvent>(m_image_ctx, event,
                                                             on_op_complete));

  // ignore errors caused due to replay
  op_event->ignore_error_codes = {-EBUSY};

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::SnapUnprotectEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Snap unprotect event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::SnapUnprotectEvent>(m_image_ctx,
                                                               event,
                                                               on_op_complete));

  // ignore errors recorded in the journal
  op_event->op_finish_error_codes = {-EBUSY};

  // ignore errors caused due to replay
  op_event->ignore_error_codes = {-EINVAL};

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::SnapRollbackEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Snap rollback start event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::SnapRollbackEvent>(m_image_ctx,
                                                              event,
                                                              on_op_complete));

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::RenameEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Rename event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::RenameEvent>(m_image_ctx, event,
                                                        on_op_complete));

  // ignore errors caused due to replay
  op_event->ignore_error_codes = {-EEXIST};

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::ResizeEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Resize start event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  // avoid lock cycles
  m_image_ctx.op_work_queue->queue(new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::ResizeEvent>(m_image_ctx, event,
                                                        on_op_complete)), 0);

  // do not process more events until the state machine is ready
  // since it will affect IO
  op_event->op_in_progress = true;
  op_event->on_start_ready = on_ready;
}

template <typename I>
void Replay<I>::handle_event(const journal::FlattenEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Flatten start event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::FlattenEvent>(m_image_ctx, event,
                                                         on_op_complete));

  // ignore errors caused due to replay
  op_event->ignore_error_codes = {-EINVAL};

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::DemotePromoteEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Demote/Promote event" << dendl;
  on_ready->complete(0);
  on_safe->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::SnapLimitEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Snap limit event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::SnapLimitEvent>(m_image_ctx,
							   event,
							   on_op_complete));

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::UpdateFeaturesEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Update features event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  // avoid lock cycles
  m_image_ctx.op_work_queue->queue(new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::UpdateFeaturesEvent>(
      m_image_ctx, event, on_op_complete)), 0);

  // do not process more events until the state machine is ready
  // since it will affect IO
  op_event->op_in_progress = true;
  op_event->on_start_ready = on_ready;
}

template <typename I>
void Replay<I>::handle_event(const journal::MetadataSetEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Metadata set event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::MetadataSetEvent>(
      m_image_ctx, event, on_op_complete));

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::MetadataRemoveEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": Metadata remove event" << dendl;

  Mutex::Locker locker(m_lock);
  OpEvent *op_event;
  Context *on_op_complete = create_op_context_callback(event.op_tid, on_ready,
                                                       on_safe, &op_event);
  if (on_op_complete == nullptr) {
    return;
  }

  op_event->on_op_finish_event = new C_RefreshIfRequired<I>(
    m_image_ctx, new ExecuteOp<I, journal::MetadataRemoveEvent>(
      m_image_ctx, event, on_op_complete));

  // ignore errors caused due to replay
  op_event->ignore_error_codes = {-ENOENT};

  on_ready->complete(0);
}

template <typename I>
void Replay<I>::handle_event(const journal::UnknownEvent &event,
			     Context *on_ready, Context *on_safe) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": unknown event" << dendl;
  on_ready->complete(0);
  on_safe->complete(0);
}

template <typename I>
void Replay<I>::handle_aio_modify_complete(Context *on_ready, Context *on_safe,
                                           int r, std::set<int> &filters) {
  Mutex::Locker locker(m_lock);
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": on_ready=" << on_ready << ", "
                 << "on_safe=" << on_safe << ", r=" << r << dendl;

  if (on_ready != nullptr) {
    on_ready->complete(0);
  }

  if (filters.find(r) != filters.end())
    r = 0;

  if (r < 0) {
    lderr(cct) << ": AIO modify op failed: " << cpp_strerror(r) << dendl;
    on_safe->complete(r);
    return;
  }

  // will be completed after next flush operation completes
  m_aio_modify_safe_contexts.insert(on_safe);
}

template <typename I>
void Replay<I>::handle_aio_flush_complete(Context *on_flush_safe,
                                          Contexts &on_safe_ctxs, int r) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": r=" << r << dendl;

  if (r < 0) {
    lderr(cct) << ": AIO flush failed: " << cpp_strerror(r) << dendl;
  }

  Context *on_aio_ready = nullptr;
  Context *on_flush = nullptr;
  {
    Mutex::Locker locker(m_lock);
    assert(m_in_flight_aio_flush > 0);
    assert(m_in_flight_aio_modify >= on_safe_ctxs.size());
    --m_in_flight_aio_flush;
    m_in_flight_aio_modify -= on_safe_ctxs.size();

    std::swap(on_aio_ready, m_on_aio_ready);
    if (m_in_flight_op_events == 0 &&
        (m_in_flight_aio_flush + m_in_flight_aio_modify) == 0) {
      on_flush = m_flush_ctx;
    }

    // strip out previously failed on_safe contexts
    for (auto it = on_safe_ctxs.begin(); it != on_safe_ctxs.end(); ) {
      if (m_aio_modify_safe_contexts.erase(*it)) {
        ++it;
      } else {
        it = on_safe_ctxs.erase(it);
      }
    }
  }

  if (on_aio_ready != nullptr) {
    ldout(cct, 10) << ": resuming paused AIO" << dendl;
    on_aio_ready->complete(0);
  }

  if (on_flush_safe != nullptr) {
    on_safe_ctxs.push_back(on_flush_safe);
  }
  for (auto ctx : on_safe_ctxs) {
    ldout(cct, 20) << ": completing safe context: " << ctx << dendl;
    ctx->complete(r);
  }

  if (on_flush != nullptr) {
    ldout(cct, 20) << ": completing flush context: " << on_flush << dendl;
    on_flush->complete(r);
  }
}

template <typename I>
Context *Replay<I>::create_op_context_callback(uint64_t op_tid,
                                               Context *on_ready,
                                               Context *on_safe,
                                               OpEvent **op_event) {
  CephContext *cct = m_image_ctx.cct;
  if (m_shut_down) {
    ldout(cct, 5) << ": ignoring event after shut down" << dendl;
    on_ready->complete(0);
    m_image_ctx.op_work_queue->queue(on_safe, -ESHUTDOWN);
    return nullptr;
  }

  assert(m_lock.is_locked());
  if (m_op_events.count(op_tid) != 0) {
    lderr(cct) << ": duplicate op tid detected: " << op_tid << dendl;

    // on_ready is already async but on failure invoke on_safe async
    // as well
    on_ready->complete(0);
    m_image_ctx.op_work_queue->queue(on_safe, -EINVAL);
    return nullptr;
  }

  ++m_in_flight_op_events;
  *op_event = &m_op_events[op_tid];
  (*op_event)->on_start_safe = on_safe;

  Context *on_op_complete = new C_OpOnComplete(this, op_tid);
  (*op_event)->on_op_complete = on_op_complete;
  return on_op_complete;
}

template <typename I>
void Replay<I>::handle_op_complete(uint64_t op_tid, int r) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 20) << ": op_tid=" << op_tid << ", "
                 << "r=" << r << dendl;

  OpEvent op_event;
  bool shutting_down = false;
  {
    Mutex::Locker locker(m_lock);
    auto op_it = m_op_events.find(op_tid);
    assert(op_it != m_op_events.end());

    op_event = std::move(op_it->second);
    m_op_events.erase(op_it);

    if (m_shut_down) {
      assert(m_flush_ctx != nullptr);
      shutting_down = true;
    }
  }

  assert(op_event.on_start_ready == nullptr || (r < 0 && r != -ERESTART));
  if (op_event.on_start_ready != nullptr) {
    // blocking op event failed before it became ready
    assert(op_event.on_finish_ready == nullptr &&
           op_event.on_finish_safe == nullptr);

    op_event.on_start_ready->complete(0);
  } else {
    // event kicked off by OpFinishEvent
    assert((op_event.on_finish_ready != nullptr &&
            op_event.on_finish_safe != nullptr) || shutting_down);
  }

  if (op_event.on_op_finish_event != nullptr) {
    op_event.on_op_finish_event->complete(r);
  }

  if (op_event.on_finish_ready != nullptr) {
    op_event.on_finish_ready->complete(0);
  }

  // filter out errors caused by replay of the same op
  if (r < 0 && op_event.ignore_error_codes.count(r) != 0) {
    r = 0;
  }

  op_event.on_start_safe->complete(r);
  if (op_event.on_finish_safe != nullptr) {
    op_event.on_finish_safe->complete(r);
  }

  // shut down request might have occurred while lock was
  // dropped -- handle if pending
  Context *on_flush = nullptr;
  {
    Mutex::Locker locker(m_lock);
    assert(m_in_flight_op_events > 0);
    --m_in_flight_op_events;
    if (m_in_flight_op_events == 0 &&
        (m_in_flight_aio_flush + m_in_flight_aio_modify) == 0) {
      on_flush = m_flush_ctx;
    }
  }
  if (on_flush != nullptr) {
    m_image_ctx.op_work_queue->queue(on_flush, 0);
  }
}

template <typename I>
io::AioCompletion *
Replay<I>::create_aio_modify_completion(Context *on_ready,
                                        Context *on_safe,
                                        io::aio_type_t aio_type,
                                        bool *flush_required,
                                        std::set<int> &&filters) {
  Mutex::Locker locker(m_lock);
  CephContext *cct = m_image_ctx.cct;
  assert(m_on_aio_ready == nullptr);

  if (m_shut_down) {
    ldout(cct, 5) << ": ignoring event after shut down" << dendl;
    on_ready->complete(0);
    m_image_ctx.op_work_queue->queue(on_safe, -ESHUTDOWN);
    return nullptr;
  }

  ++m_in_flight_aio_modify;
  m_aio_modify_unsafe_contexts.push_back(on_safe);

  // FLUSH if we hit the low-water mark -- on_safe contexts are
  // completed by flushes-only so that we don't move the journal
  // commit position until safely on-disk

  *flush_required = (m_aio_modify_unsafe_contexts.size() ==
                       IN_FLIGHT_IO_LOW_WATER_MARK);
  if (*flush_required) {
    ldout(cct, 10) << ": hit AIO replay low-water mark: scheduling flush"
                   << dendl;
  }

  // READY for more events if:
  // * not at high-water mark for IO
  // * in-flight ops are at a consistent point (snap create has IO flushed,
  //   shrink has adjusted clip boundary, etc) -- should have already been
  //   flagged not-ready
  if (m_in_flight_aio_modify == IN_FLIGHT_IO_HIGH_WATER_MARK) {
    ldout(cct, 10) << ": hit AIO replay high-water mark: pausing replay"
                   << dendl;
    assert(m_on_aio_ready == nullptr);
    std::swap(m_on_aio_ready, on_ready);
  }

  // when the modification is ACKed by librbd, we can process the next
  // event. when flushed, the completion of the next flush will fire the
  // on_safe callback
  auto aio_comp = io::AioCompletion::create_and_start<Context>(
    new C_AioModifyComplete(this, on_ready, on_safe, std::move(filters)),
    util::get_image_ctx(&m_image_ctx), aio_type);
  return aio_comp;
}

template <typename I>
io::AioCompletion *Replay<I>::create_aio_flush_completion(Context *on_safe) {
  assert(m_lock.is_locked());

  CephContext *cct = m_image_ctx.cct;
  if (m_shut_down) {
    ldout(cct, 5) << ": ignoring event after shut down" << dendl;
    if (on_safe != nullptr) {
      m_image_ctx.op_work_queue->queue(on_safe, -ESHUTDOWN);
    }
    return nullptr;
  }

  ++m_in_flight_aio_flush;

  // associate all prior write/discard ops to this flush request
  auto aio_comp = io::AioCompletion::create_and_start<Context>(
      new C_AioFlushComplete(this, on_safe,
                             std::move(m_aio_modify_unsafe_contexts)),
      util::get_image_ctx(&m_image_ctx), io::AIO_TYPE_FLUSH);
  m_aio_modify_unsafe_contexts.clear();
  return aio_comp;
}

} // namespace journal
} // namespace librbd

template class librbd::journal::Replay<librbd::ImageCtx>;