ceph/src/librbd/journal/ObjectDispatch.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "librbd/journal/ObjectDispatch.h"
   5 #include "common/dout.h"
   6 #include "common/WorkQueue.h"
   7 #include "osdc/Striper.h"
   8 #include "librbd/ImageCtx.h"
   9 #include "librbd/Journal.h"
  10 #include "librbd/io/ObjectDispatchSpec.h"
  11 #include "librbd/io/ObjectDispatcher.h"
  12
  13 #define dout_subsys ceph_subsys_rbd
  14 #undef dout_prefix
  15 #define dout_prefix *_dout << "librbd::journal::ObjectDispatch: " << this \
  16                            << " " << __func__ << ": "
  17
  18 namespace librbd {
  19 namespace journal {
  20
  21 namespace {
  22
  23 template <typename I>
  24 struct C_CommitIOEvent : public Context {
  25   I* image_ctx;
  26   Journal<I>* journal;
  27   uint64_t object_no;
  28   uint64_t object_off;
  29   uint64_t object_len;
  30   uint64_t journal_tid;
  31   int object_dispatch_flags;
  32   Context* on_finish;
  33
  34   C_CommitIOEvent(I* image_ctx, Journal<I>* journal, uint64_t object_no,
  35                   uint64_t object_off, uint64_t object_len,
  36                   uint64_t journal_tid, int object_dispatch_flags,
  37                   Context* on_finish)
  38     : image_ctx(image_ctx), journal(journal), object_no(object_no),
  39       object_off(object_off), object_len(object_len), journal_tid(journal_tid),
  40       object_dispatch_flags(object_dispatch_flags), on_finish(on_finish) {
  41   }
  42
  43   void finish(int r) override {
  44     // don't commit the IO extent if a previous dispatch handler will just
  45     // retry the failed IO
  46     if (r >= 0 ||
  47         (object_dispatch_flags &
  48            io::OBJECT_DISPATCH_FLAG_WILL_RETRY_ON_ERROR) == 0) {
  49       io::Extents file_extents;
  50       Striper::extent_to_file(image_ctx->cct, &image_ctx->layout, object_no,
  51                               object_off, object_len, file_extents);
  52       for (auto& extent : file_extents) {
  53         journal->commit_io_event_extent(journal_tid, extent.first,
  54                                         extent.second, r);
  55       }
  56     }
  57
  58     if (on_finish != nullptr) {
  59       on_finish->complete(r);
  60     }
  61   }
  62 };
  63
  64 } // anonymous namespace
  65
  66 template <typename I>
  67 ObjectDispatch<I>::ObjectDispatch(I* image_ctx, Journal<I>* journal)
  68   : m_image_ctx(image_ctx), m_journal(journal) {
  69 }
  70
  71 template <typename I>
  72 void ObjectDispatch<I>::shut_down(Context* on_finish) {
  73   m_image_ctx->op_work_queue->queue(on_finish, 0);
  74 }
  75
  76 template <typename I>
  77 bool ObjectDispatch<I>::discard(
  78     const std::string &oid, uint64_t object_no, uint64_t object_off,
  79     uint64_t object_len, const ::SnapContext &snapc, int discard_flags,
  80     const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
  81     uint64_t* journal_tid, io::DispatchResult* dispatch_result,
  82     Context** on_finish, Context* on_dispatched) {
  83   if (*journal_tid == 0) {
  84     // non-journaled IO
  85     return false;
  86   }
  87
  88   auto cct = m_image_ctx->cct;
  89   ldout(cct, 20) << oid << " " << object_off << "~" << object_len << dendl;
  90
  91   *on_finish = new C_CommitIOEvent<I>(m_image_ctx, m_journal, object_no,
  92                                       object_off, object_len, *journal_tid,
  93                                       *object_dispatch_flags, *on_finish);
  94
  95   *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
  96   wait_or_flush_event(*journal_tid, *object_dispatch_flags, on_dispatched);
  97   return true;
  98 }
  99
 100 template <typename I>
 101 bool ObjectDispatch<I>::write(
 102     const std::string &oid, uint64_t object_no, uint64_t object_off,
 103     ceph::bufferlist&& data, const ::SnapContext &snapc, int op_flags,
 104     const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
 105     uint64_t* journal_tid, io::DispatchResult* dispatch_result,
 106     Context** on_finish, Context* on_dispatched) {
 107   if (*journal_tid == 0) {
 108     // non-journaled IO
 109     return false;
 110   }
 111
 112   auto cct = m_image_ctx->cct;
 113   ldout(cct, 20) << oid << " " << object_off << "~" << data.length() << dendl;
 114
 115   *on_finish = new C_CommitIOEvent<I>(m_image_ctx, m_journal, object_no,
 116                                       object_off, data.length(), *journal_tid,
 117                                       *object_dispatch_flags, *on_finish);
 118
 119   *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
 120   wait_or_flush_event(*journal_tid, *object_dispatch_flags, on_dispatched);
 121   return true;
 122 }
 123
 124 template <typename I>
 125 bool ObjectDispatch<I>::write_same(
 126     const std::string &oid, uint64_t object_no, uint64_t object_off,
 127     uint64_t object_len, io::Extents&& buffer_extents, ceph::bufferlist&& data,
 128     const ::SnapContext &snapc, int op_flags,
 129     const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
 130     uint64_t* journal_tid, io::DispatchResult* dispatch_result,
 131     Context** on_finish, Context* on_dispatched) {
 132   if (*journal_tid == 0) {
 133     // non-journaled IO
 134     return false;
 135   }
 136
 137   auto cct = m_image_ctx->cct;
 138   ldout(cct, 20) << oid << " " << object_off << "~" << object_len << dendl;
 139
 140   *on_finish = new C_CommitIOEvent<I>(m_image_ctx, m_journal, object_no,
 141                                       object_off, object_len, *journal_tid,
 142                                       *object_dispatch_flags, *on_finish);
 143
 144   *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
 145   wait_or_flush_event(*journal_tid, *object_dispatch_flags, on_dispatched);
 146   return true;
 147 }
 148
 149 template <typename I>
 150 bool ObjectDispatch<I>::compare_and_write(
 151     const std::string &oid, uint64_t object_no, uint64_t object_off,
 152     ceph::bufferlist&& cmp_data, ceph::bufferlist&& write_data,
 153     const ::SnapContext &snapc, int op_flags,
 154     const ZTracer::Trace &parent_trace, uint64_t* mismatch_offset,
 155     int* object_dispatch_flags, uint64_t* journal_tid,
 156     io::DispatchResult* dispatch_result, Context** on_finish,
 157     Context* on_dispatched) {
 158   if (*journal_tid == 0) {
 159     // non-journaled IO
 160     return false;
 161   }
 162
 163   auto cct = m_image_ctx->cct;
 164   ldout(cct, 20) << oid << " " << object_off << "~" << write_data.length()
 165                  << dendl;
 166
 167   *on_finish = new C_CommitIOEvent<I>(m_image_ctx, m_journal, object_no,
 168                                       object_off, write_data.length(),
 169                                       *journal_tid, *object_dispatch_flags,
 170                                       *on_finish);
 171
 172   *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
 173   wait_or_flush_event(*journal_tid, *object_dispatch_flags, on_dispatched);
 174   return true;
 175 }
 176
 177 template <typename I>
 178 void ObjectDispatch<I>::extent_overwritten(
 179     uint64_t object_no, uint64_t object_off, uint64_t object_len,
 180     uint64_t journal_tid, uint64_t new_journal_tid) {
 181   auto cct = m_image_ctx->cct;
 182   ldout(cct, 20) << object_no << " " << object_off << "~" << object_len
 183                  << dendl;
 184
 185   auto ctx = new C_CommitIOEvent<I>(m_image_ctx, m_journal, object_no,
 186                                     object_off, object_len, journal_tid, false,
 187                                     nullptr);
 188   if (new_journal_tid != 0) {
 189     // ensure new journal event is safely committed to disk before
 190     // committing old event
 191     m_journal->flush_event(new_journal_tid, ctx);
 192   } else {
 193     ctx->complete(0);
 194   }
 195 }
 196
 197 template <typename I>
 198 void ObjectDispatch<I>::wait_or_flush_event(
 199     uint64_t journal_tid, int object_dispatch_flags, Context* on_dispatched) {
 200   auto cct = m_image_ctx->cct;
 201   ldout(cct, 20) << "journal_tid=" << journal_tid << dendl;
 202
 203   if ((object_dispatch_flags & io::OBJECT_DISPATCH_FLAG_FLUSH) != 0) {
 204     m_journal->flush_event(journal_tid, on_dispatched);
 205   } else {
 206     m_journal->wait_event(journal_tid, on_dispatched);
 207   }
 208 }
 209
 210 } // namespace journal
 211 } // namespace librbd
 212
 213 template class librbd::journal::ObjectDispatch<librbd::ImageCtx>;