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