ceph/src/librbd/io/AioCompletion.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/io/AioCompletion.h"
   5 #include <errno.h>
   6
   7 #include "common/ceph_context.h"
   8 #include "common/dout.h"
   9 #include "common/errno.h"
  10 #include "common/perf_counters.h"
  11 #include "common/WorkQueue.h"
  12
  13 #include "librbd/ImageCtx.h"
  14 #include "librbd/internal.h"
  15 #include "librbd/Journal.h"
  16 #include "librbd/Types.h"
  17
  18 #ifdef WITH_LTTNG
  19 #include "tracing/librbd.h"
  20 #else
  21 #define tracepoint(...)
  22 #endif
  23
  24 #define dout_subsys ceph_subsys_rbd
  25 #undef dout_prefix
  26 #define dout_prefix *_dout << "librbd::io::AioCompletion: " << this \
  27                            << " " << __func__ << ": "
  28
  29 namespace librbd {
  30 namespace io {
  31
  32 int AioCompletion::wait_for_complete() {
  33   tracepoint(librbd, aio_wait_for_complete_enter, this);
  34   lock.Lock();
  35   while (state != AIO_STATE_COMPLETE)
  36     cond.Wait(lock);
  37   lock.Unlock();
  38   tracepoint(librbd, aio_wait_for_complete_exit, 0);
  39   return 0;
  40 }
  41
  42 void AioCompletion::finalize(ssize_t rval)
  43 {
  44   ceph_assert(lock.is_locked());
  45   ceph_assert(ictx != nullptr);
  46   CephContext *cct = ictx->cct;
  47
  48   ldout(cct, 20) << "r=" << rval << dendl;
  49   if (rval >= 0 && aio_type == AIO_TYPE_READ) {
  50     read_result.assemble_result(cct);
  51   }
  52 }
  53
  54 void AioCompletion::complete() {
  55   ceph_assert(lock.is_locked());
  56   ceph_assert(ictx != nullptr);
  57   CephContext *cct = ictx->cct;
  58
  59   tracepoint(librbd, aio_complete_enter, this, rval);
  60   if (ictx->perfcounter != nullptr) {
  61     ceph::timespan elapsed = coarse_mono_clock::now() - start_time;
  62     switch (aio_type) {
  63     case AIO_TYPE_GENERIC:
  64     case AIO_TYPE_OPEN:
  65     case AIO_TYPE_CLOSE:
  66       break;
  67     case AIO_TYPE_READ:
  68       ictx->perfcounter->tinc(l_librbd_rd_latency, elapsed); break;
  69     case AIO_TYPE_WRITE:
  70       ictx->perfcounter->tinc(l_librbd_wr_latency, elapsed); break;
  71     case AIO_TYPE_DISCARD:
  72       ictx->perfcounter->tinc(l_librbd_discard_latency, elapsed); break;
  73     case AIO_TYPE_FLUSH:
  74       ictx->perfcounter->tinc(l_librbd_flush_latency, elapsed); break;
  75     case AIO_TYPE_WRITESAME:
  76       ictx->perfcounter->tinc(l_librbd_ws_latency, elapsed); break;
  77     case AIO_TYPE_COMPARE_AND_WRITE:
  78       ictx->perfcounter->tinc(l_librbd_cmp_latency, elapsed); break;
  79     default:
  80       lderr(cct) << "completed invalid aio_type: " << aio_type << dendl;
  81       break;
  82     }
  83   }
  84
  85   state = AIO_STATE_CALLBACK;
  86   if (complete_cb) {
  87     lock.Unlock();
  88     complete_cb(rbd_comp, complete_arg);
  89     lock.Lock();
  90   }
  91
  92   if (event_notify && ictx->event_socket.is_valid()) {
  93     ictx->completed_reqs_lock.Lock();
  94     ictx->completed_reqs.push_back(&m_xlist_item);
  95     ictx->completed_reqs_lock.Unlock();
  96     ictx->event_socket.notify();
  97   }
  98
  99   state = AIO_STATE_COMPLETE;
 100   cond.Signal();
 101
 102   // note: possible for image to be closed after op marked finished
 103   if (async_op.started()) {
 104     async_op.finish_op();
 105   }
 106   tracepoint(librbd, aio_complete_exit);
 107 }
 108
 109 void AioCompletion::init_time(ImageCtx *i, aio_type_t t) {
 110   Mutex::Locker locker(lock);
 111   if (ictx == nullptr) {
 112     ictx = i;
 113     aio_type = t;
 114     start_time = coarse_mono_clock::now();
 115   }
 116 }
 117
 118 void AioCompletion::start_op(bool ignore_type) {
 119   Mutex::Locker locker(lock);
 120   ceph_assert(ictx != nullptr);
 121   ceph_assert(!async_op.started());
 122   if (state == AIO_STATE_PENDING &&
 123       (ignore_type || aio_type != AIO_TYPE_FLUSH)) {
 124     async_op.start_op(*ictx);
 125   }
 126 }
 127
 128 void AioCompletion::fail(int r)
 129 {
 130   lock.Lock();
 131   ceph_assert(ictx != nullptr);
 132   CephContext *cct = ictx->cct;
 133
 134   lderr(cct) << cpp_strerror(r) << dendl;
 135   ceph_assert(pending_count == 0);
 136   rval = r;
 137   complete();
 138   put_unlock();
 139 }
 140
 141 void AioCompletion::set_request_count(uint32_t count) {
 142   lock.Lock();
 143   ceph_assert(ictx != nullptr);
 144   CephContext *cct = ictx->cct;
 145
 146   ldout(cct, 20) << "pending=" << count << dendl;
 147   ceph_assert(pending_count == 0);
 148   pending_count = count;
 149   lock.Unlock();
 150
 151   // if no pending requests, completion will fire now
 152   unblock();
 153 }
 154
 155 void AioCompletion::complete_request(ssize_t r)
 156 {
 157   lock.Lock();
 158   ceph_assert(ictx != nullptr);
 159   CephContext *cct = ictx->cct;
 160
 161   if (rval >= 0) {
 162     if (r < 0 && r != -EEXIST)
 163       rval = r;
 164     else if (r > 0)
 165       rval += r;
 166   }
 167   ceph_assert(pending_count);
 168   int count = --pending_count;
 169
 170   ldout(cct, 20) << "cb=" << complete_cb << ", "
 171                  << "pending=" << pending_count << dendl;
 172   if (!count && blockers == 0) {
 173     finalize(rval);
 174     complete();
 175   }
 176   put_unlock();
 177 }
 178
 179 bool AioCompletion::is_complete() {
 180   tracepoint(librbd, aio_is_complete_enter, this);
 181   bool done;
 182   {
 183     Mutex::Locker l(lock);
 184     done = this->state == AIO_STATE_COMPLETE;
 185   }
 186   tracepoint(librbd, aio_is_complete_exit, done);
 187   return done;
 188 }
 189
 190 ssize_t AioCompletion::get_return_value() {
 191   tracepoint(librbd, aio_get_return_value_enter, this);
 192   lock.Lock();
 193   ssize_t r = rval;
 194   lock.Unlock();
 195   tracepoint(librbd, aio_get_return_value_exit, r);
 196   return r;
 197 }
 198
 199 } // namespace io
 200 } // namespace librbd