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