ceph/src/librbd/io/AioCompletion.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #ifndef CEPH_LIBRBD_IO_AIO_COMPLETION_H
   5 #define CEPH_LIBRBD_IO_AIO_COMPLETION_H
   6
   7 #include "common/ceph_time.h"
   8 #include "include/common_fwd.h"
   9 #include "include/Context.h"
  10 #include "include/utime.h"
  11 #include "include/rbd/librbd.hpp"
  12
  13 #include "librbd/ImageCtx.h"
  14 #include "librbd/io/AsyncOperation.h"
  15 #include "librbd/io/ReadResult.h"
  16 #include "librbd/io/Types.h"
  17
  18 #include <atomic>
  19 #include <condition_variable>
  20 #include <mutex>
  21
  22 struct Context;
  23
  24 namespace librbd {
  25 namespace io {
  26
  27 /**
  28  * AioCompletion is the overall completion for a single
  29  * rbd I/O request. It may be composed of many AioObjectRequests,
  30  * which each go to a single object.
  31  *
  32  * The retrying of individual requests is handled at a lower level,
  33  * so all AioCompletion cares about is the count of outstanding
  34  * requests. The number of expected individual requests should be
  35  * set initially using set_request_count() prior to issuing the
  36  * requests.  This ensures that the completion will not be completed
  37  * within the caller's thread of execution (instead via a librados
  38  * context or via a thread pool context for cache read hits).
  39  */
  40 struct AioCompletion {
  41   typedef enum {
  42     AIO_STATE_PENDING = 0,
  43     AIO_STATE_CALLBACK,
  44     AIO_STATE_COMPLETE,
  45   } aio_state_t;
  46
  47   mutable std::mutex lock;
  48   std::condition_variable cond;
  49
  50   callback_t complete_cb = nullptr;
  51   void *complete_arg = nullptr;
  52   rbd_completion_t rbd_comp = nullptr;
  53
  54   /// note: only using atomic for built-in memory barrier
  55   std::atomic<aio_state_t> state{AIO_STATE_PENDING};
  56
  57   std::atomic<ssize_t> rval{0};
  58   std::atomic<int> error_rval{0};
  59   std::atomic<uint32_t> ref{1};
  60   std::atomic<uint32_t> pending_count{0};   ///< number of requests/blocks
  61   std::atomic<bool> released{false};
  62
  63   ImageCtx *ictx = nullptr;
  64   coarse_mono_time start_time;
  65   aio_type_t aio_type = AIO_TYPE_NONE;
  66
  67   ReadResult read_result;
  68
  69   AsyncOperation async_op;
  70
  71   bool event_notify = false;
  72   bool was_armed = false;
  73   bool external_callback = false;
  74
  75   Context* image_dispatcher_ctx = nullptr;
  76
  77   template <typename T, void (T::*MF)(int)>
  78   static void callback_adapter(completion_t cb, void *arg) {
  79     AioCompletion *comp = reinterpret_cast<AioCompletion *>(cb);
  80     T *t = reinterpret_cast<T *>(arg);
  81     (t->*MF)(comp->get_return_value());
  82     comp->release();
  83   }
  84
  85   static AioCompletion *create(void *cb_arg, callback_t cb_complete,
  86                                rbd_completion_t rbd_comp) {
  87     AioCompletion *comp = new AioCompletion();
  88     comp->set_complete_cb(cb_arg, cb_complete);
  89     comp->rbd_comp = (rbd_comp != nullptr ? rbd_comp : comp);
  90     return comp;
  91   }
  92
  93   template <typename T, void (T::*MF)(int) = &T::complete>
  94   static AioCompletion *create(T *obj) {
  95     AioCompletion *comp = new AioCompletion();
  96     comp->set_complete_cb(obj, &callback_adapter<T, MF>);
  97     comp->rbd_comp = comp;
  98     return comp;
  99   }
 100
 101   template <typename T, void (T::*MF)(int) = &T::complete>
 102   static AioCompletion *create_and_start(T *obj, ImageCtx *image_ctx,
 103                                          aio_type_t type) {
 104     AioCompletion *comp = create<T, MF>(obj);
 105     comp->init_time(image_ctx, type);
 106     comp->start_op();
 107     return comp;
 108   }
 109
 110   AioCompletion() {
 111   }
 112
 113   ~AioCompletion() {
 114   }
 115
 116   int wait_for_complete();
 117
 118   void finalize();
 119
 120   inline bool is_initialized(aio_type_t type) const {
 121     std::unique_lock<std::mutex> locker(lock);
 122     return ((ictx != nullptr) && (aio_type == type));
 123   }
 124   inline bool is_started() const {
 125     std::unique_lock<std::mutex> locker(lock);
 126     return async_op.started();
 127   }
 128
 129   void block(CephContext* cct);
 130   void unblock(CephContext* cct);
 131
 132   void init_time(ImageCtx *i, aio_type_t t);
 133   void start_op();
 134   void fail(int r);
 135
 136   void complete();
 137
 138   void set_complete_cb(void *cb_arg, callback_t cb) {
 139     complete_cb = cb;
 140     complete_arg = cb_arg;
 141   }
 142
 143   void set_request_count(uint32_t num);
 144   void add_request() {
 145     ceph_assert(pending_count > 0);
 146     get();
 147   }
 148   void complete_request(ssize_t r);
 149
 150   bool is_complete();
 151
 152   ssize_t get_return_value();
 153
 154   void get() {
 155     ceph_assert(ref > 0);
 156     ++ref;
 157   }
 158   void release() {
 159     bool previous_released = released.exchange(true);
 160     ceph_assert(!previous_released);
 161     put();
 162   }
 163   void put() {
 164     uint32_t previous_ref = ref--;
 165     ceph_assert(previous_ref > 0);
 166
 167     if (previous_ref == 1) {
 168       delete this;
 169     }
 170   }
 171
 172   void set_event_notify(bool s) {
 173     event_notify = s;
 174   }
 175
 176   void *get_arg() {
 177     return complete_arg;
 178   }
 179
 180 private:
 181   void queue_complete();
 182   void complete_external_callback();
 183   void complete_event_socket();
 184   void notify_callbacks_complete();
 185 };
 186
 187 class C_AioRequest : public Context {
 188 public:
 189   C_AioRequest(AioCompletion *completion) : m_completion(completion) {
 190     m_completion->add_request();
 191   }
 192   ~C_AioRequest() override {}
 193   void finish(int r) override {
 194     m_completion->complete_request(r);
 195   }
 196 protected:
 197   AioCompletion *m_completion;
 198 };
 199
 200 } // namespace io
 201 } // namespace librbd
 202
 203 #endif // CEPH_LIBRBD_IO_AIO_COMPLETION_H