rbd: use the ordered throttle for the export action

This allows exports to STDOUT to use multiple concurrent operations
and also fixes a potential race condition with concurrent callbacks
and file seeking.

Fixes: http://tracker.ceph.com/issues/40435
Signed-off-by: Jason Dillaman <dillaman@redhat.com>
(cherry picked from commit 751bf6957020ad02b3a29b005b9792b9c6547f96)

Conflicts:
	src/tools/rbd/action/Export.cc
- mimic has g_conf->get_val and int64_t "rbd_concurrent_management_ops"

diff --git a/src/tools/rbd/action/Export.cc b/src/tools/rbd/action/Export.cc
index 9aae6f0ebcead7973d831d427e5722c38a0763a5..9a821a43651eff70a55dc81da0dbff1811f725f7 100644 (file)
--- a/src/tools/rbd/action/Export.cc
+++ b/src/tools/rbd/action/Export.cc
@@ -314,26 +314,25 @@ Shell::Action action_diff(
 class C_Export : public Context
 {
 public:
-  C_Export(SimpleThrottle &simple_throttle, librbd::Image &image,
+  C_Export(OrderedThrottle &ordered_throttle, librbd::Image &image,
           uint64_t fd_offset, uint64_t offset, uint64_t length, int fd)
-    : m_aio_completion(
-        new librbd::RBD::AioCompletion(this, &utils::aio_context_callback)),
-      m_throttle(simple_throttle), m_image(image), m_dest_offset(fd_offset),
+    : m_throttle(ordered_throttle), m_image(image), m_dest_offset(fd_offset),
       m_offset(offset), m_length(length), m_fd(fd)
   {
   }
 
   void send()
   {
-    m_throttle.start_op();
-
+    auto ctx = m_throttle.start_op(this);
+    auto aio_completion = new librbd::RBD::AioCompletion(
+      ctx, &utils::aio_context_callback);
     int op_flags = LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL |
                    LIBRADOS_OP_FLAG_FADVISE_NOCACHE;
     int r = m_image.aio_read2(m_offset, m_length, m_bufferlist,
-                              m_aio_completion, op_flags);
+                              aio_completion, op_flags);
     if (r < 0) {
       cerr << "rbd: error requesting read from source image" << std::endl;
-      m_aio_completion->release();
+      aio_completion->release();
       m_throttle.end_op(r);
     }
   }
@@ -374,8 +373,7 @@ public:
   }
 
 private:
-  librbd::RBD::AioCompletion *m_aio_completion;
-  SimpleThrottle &m_throttle;
+  OrderedThrottle &m_throttle;
   librbd::Image &m_image;
   bufferlist m_bufferlist;
   uint64_t m_dest_offset;
@@ -515,19 +513,21 @@ static int do_export_v2(librbd::Image& image, librbd::image_info_t &info, int fd
   return r;
 }
 
-static int do_export_v1(librbd::Image& image, librbd::image_info_t &info, int fd,
-                       uint64_t period, int max_concurrent_ops, utils::ProgressContext &pc)
+static int do_export_v1(librbd::Image& image, librbd::image_info_t &info,
+                        int fd, uint64_t period, int max_concurrent_ops,
+                        utils::ProgressContext &pc)
 {
   int r = 0;
   size_t file_size = 0;
-  SimpleThrottle throttle(max_concurrent_ops, false);
+  OrderedThrottle throttle(max_concurrent_ops, false);
   for (uint64_t offset = 0; offset < info.size; offset += period) {
     if (throttle.pending_error()) {
       break;
     }
 
     uint64_t length = min(period, info.size - offset);
-    C_Export *ctx = new C_Export(throttle, image, file_size + offset, offset, length, fd);
+    C_Export *ctx = new C_Export(throttle, image, file_size + offset, offset,
+                                 length, fd);
     ctx->send();
 
     pc.update_progress(offset, info.size);
@@ -549,7 +549,8 @@ static int do_export_v1(librbd::Image& image, librbd::image_info_t &info, int fd
   return r;
 }
 
-static int do_export(librbd::Image& image, const char *path, bool no_progress, int export_format)
+static int do_export(librbd::Image& image, const char *path, bool no_progress,
+                     int export_format)
 {
   librbd::image_info_t info;
   int64_t r = image.stat(info, sizeof(info));
@@ -557,13 +558,11 @@ static int do_export(librbd::Image& image, const char *path, bool no_progress, i
     return r;
 
   int fd;
-  int max_concurrent_ops;
+  int max_concurrent_ops = g_conf->get_val<int64_t>("rbd_concurrent_management_ops");
   bool to_stdout = (strcmp(path, "-") == 0);
   if (to_stdout) {
     fd = STDOUT_FILENO;
-    max_concurrent_ops = 1;
   } else {
-    max_concurrent_ops = g_conf->get_val<int64_t>("rbd_concurrent_management_ops");
     fd = open(path, O_WRONLY | O_CREAT | O_EXCL, 0644);
     if (fd < 0) {
       return -errno;