ceph/src/librbd/AsyncObjectThrottle.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 #include "librbd/AsyncObjectThrottle.h"
   4 #include "common/RWLock.h"
   5 #include "common/WorkQueue.h"
   6 #include "librbd/AsyncRequest.h"
   7 #include "librbd/ImageCtx.h"
   8 #include "librbd/Utils.h"
   9
  10 namespace librbd
  11 {
  12
  13 template <typename T>
  14 AsyncObjectThrottle<T>::AsyncObjectThrottle(
  15     const AsyncRequest<T>* async_request, T &image_ctx,
  16     const ContextFactory& context_factory, Context *ctx,
  17     ProgressContext *prog_ctx, uint64_t object_no, uint64_t end_object_no)
  18   : m_lock(ceph::make_mutex(
  19       util::unique_lock_name("librbd::AsyncThrottle::m_lock", this))),
  20     m_async_request(async_request), m_image_ctx(image_ctx),
  21     m_context_factory(context_factory), m_ctx(ctx), m_prog_ctx(prog_ctx),
  22     m_object_no(object_no), m_end_object_no(end_object_no), m_current_ops(0),
  23     m_ret(0)
  24 {
  25 }
  26
  27 template <typename T>
  28 void AsyncObjectThrottle<T>::start_ops(uint64_t max_concurrent) {
  29   ceph_assert(ceph_mutex_is_locked(m_image_ctx.owner_lock));
  30   bool complete;
  31   {
  32     std::lock_guard l{m_lock};
  33     for (uint64_t i = 0; i < max_concurrent; ++i) {
  34       start_next_op();
  35       if (m_ret < 0 && m_current_ops == 0) {
  36         break;
  37       }
  38     }
  39     complete = (m_current_ops == 0);
  40   }
  41   if (complete) {
  42     // avoid re-entrant callback
  43     m_image_ctx.op_work_queue->queue(m_ctx, m_ret);
  44     delete this;
  45   }
  46 }
  47
  48 template <typename T>
  49 void AsyncObjectThrottle<T>::finish_op(int r) {
  50   bool complete;
  51   {
  52     std::shared_lock owner_locker{m_image_ctx.owner_lock};
  53     std::lock_guard locker{m_lock};
  54     --m_current_ops;
  55     if (r < 0 && r != -ENOENT && m_ret == 0) {
  56       m_ret = r;
  57     }
  58
  59     start_next_op();
  60     complete = (m_current_ops == 0);
  61   }
  62   if (complete) {
  63     m_ctx->complete(m_ret);
  64     delete this;
  65   }
  66 }
  67
  68 template <typename T>
  69 void AsyncObjectThrottle<T>::start_next_op() {
  70   bool done = false;
  71   while (!done) {
  72     if (m_async_request != NULL && m_async_request->is_canceled() &&
  73         m_ret == 0) {
  74       // allow in-flight ops to complete, but don't start new ops
  75       m_ret = -ERESTART;
  76       return;
  77     } else if (m_ret != 0 || m_object_no >= m_end_object_no) {
  78       return;
  79     }
  80
  81     uint64_t ono = m_object_no++;
  82     C_AsyncObjectThrottle<T> *ctx = m_context_factory(*this, ono);
  83
  84     int r = ctx->send();
  85     if (r < 0) {
  86       m_ret = r;
  87       delete ctx;
  88       return;
  89     } else if (r > 0) {
  90       // op completed immediately
  91       delete ctx;
  92     } else {
  93       ++m_current_ops;
  94       done = true;
  95     }
  96     if (m_prog_ctx != NULL) {
  97       r = m_prog_ctx->update_progress(ono, m_end_object_no);
  98       if (r < 0) {
  99         m_ret = r;
 100       }
 101     }
 102   }
 103 }
 104
 105 } // namespace librbd
 106
 107 #ifndef TEST_F
 108 template class librbd::AsyncObjectThrottle<librbd::ImageCtx>;
 109 #endif