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
  23 namespace librbd {
  24 namespace io {
  25
  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   template <typename T, void (T::*MF)(int)>
  76   static void callback_adapter(completion_t cb, void *arg) {
  77     AioCompletion *comp = reinterpret_cast<AioCompletion *>(cb);
  78     T *t = reinterpret_cast<T *>(arg);
  79     (t->*MF)(comp->get_return_value());
  80     comp->release();
  81   }
  82
  83   static AioCompletion *create(void *cb_arg, callback_t cb_complete,
  84                                rbd_completion_t rbd_comp) {
  85     AioCompletion *comp = new AioCompletion();
  86     comp->set_complete_cb(cb_arg, cb_complete);
  87     comp->rbd_comp = (rbd_comp != nullptr ? rbd_comp : comp);
  88     return comp;
  89   }
  90
  91   template <typename T, void (T::*MF)(int) = &T::complete>
  92   static AioCompletion *create(T *obj) {
  93     AioCompletion *comp = new AioCompletion();
  94     comp->set_complete_cb(obj, &callback_adapter<T, MF>);
  95     comp->rbd_comp = comp;
  96     return comp;
  97   }
  98
  99   template <typename T, void (T::*MF)(int) = &T::complete>
 100   static AioCompletion *create_and_start(T *obj, ImageCtx *image_ctx,
 101                                          aio_type_t type) {
 102     AioCompletion *comp = create<T, MF>(obj);
 103     comp->init_time(image_ctx, type);
 104     comp->start_op();
 105     return comp;
 106   }
 107
 108   AioCompletion() {
 109   }
 110
 111   ~AioCompletion() {
 112   }
 113
 114   int wait_for_complete();
 115
 116   void finalize();
 117
 118   inline bool is_initialized(aio_type_t type) const {
 119     std::unique_lock<std::mutex> locker(lock);
 120     return ((ictx != nullptr) && (aio_type == type));
 121   }
 122   inline bool is_started() const {
 123     std::unique_lock<std::mutex> locker(lock);
 124     return async_op.started();
 125   }
 126
 127   void block(CephContext* cct);
 128   void unblock(CephContext* cct);
 129
 130   void init_time(ImageCtx *i, aio_type_t t);
 131   void start_op();
 132   void fail(int r);
 133
 134   void complete();
 135
 136   void set_complete_cb(void *cb_arg, callback_t cb) {
 137     complete_cb = cb;
 138     complete_arg = cb_arg;
 139   }
 140
 141   void set_request_count(uint32_t num);
 142   void add_request() {
 143     ceph_assert(pending_count > 0);
 144     get();
 145   }
 146   void complete_request(ssize_t r);
 147
 148   bool is_complete();
 149
 150   ssize_t get_return_value();
 151
 152   void get() {
 153     ceph_assert(ref > 0);
 154     ++ref;
 155   }
 156   void release() {
 157     bool previous_released = released.exchange(true);
 158     ceph_assert(!previous_released);
 159     put();
 160   }
 161   void put() {
 162     uint32_t previous_ref = ref--;
 163     ceph_assert(previous_ref > 0);
 164
 165     if (previous_ref == 1) {
 166       delete this;
 167     }
 168   }
 169
 170   void set_event_notify(bool s) {
 171     event_notify = s;
 172   }
 173
 174   void *get_arg() {
 175     return complete_arg;
 176   }
 177
 178 private:
 179   void queue_complete();
 180   void complete_external_callback();
 181   void complete_event_socket();
 182   void notify_callbacks_complete();
 183 };
 184
 185 class C_AioRequest : public Context {
 186 public:
 187   C_AioRequest(AioCompletion *completion) : m_completion(completion) {
 188     m_completion->add_request();
 189   }
 190   ~C_AioRequest() override {}
 191   void finish(int r) override {
 192     m_completion->complete_request(r);
 193   }
 194 protected:
 195   AioCompletion *m_completion;
 196 };
 197
 198 } // namespace io
 199 } // namespace librbd
 200
 201 #endif // CEPH_LIBRBD_IO_AIO_COMPLETION_H