ceph/src/librbd/operation/RenameRequest.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/operation/RenameRequest.h"
   5 #include "common/dout.h"
   6 #include "common/errno.h"
   7 #include "include/rados/librados.hpp"
   8 #include "librbd/ImageCtx.h"
   9 #include "librbd/internal.h"
  10 #include "librbd/Utils.h"
  11
  12 #define dout_subsys ceph_subsys_rbd
  13 #undef dout_prefix
  14 #define dout_prefix *_dout << "librbd::operation::RenameRequest: "
  15
  16 namespace librbd {
  17 namespace operation {
  18
  19 namespace {
  20
  21 template <typename I>
  22 std::ostream& operator<<(std::ostream& os,
  23                          const typename RenameRequest<I>::State& state) {
  24   switch(state) {
  25   case RenameRequest<I>::STATE_READ_DIRECTORY:
  26     os << "READ_DIRECTORY";
  27     break;
  28   case RenameRequest<I>::STATE_READ_SOURCE_HEADER:
  29     os << "READ_SOURCE_HEADER";
  30     break;
  31   case RenameRequest<I>::STATE_WRITE_DEST_HEADER:
  32     os << "WRITE_DEST_HEADER";
  33     break;
  34   case RenameRequest<I>::STATE_UPDATE_DIRECTORY:
  35     os << "UPDATE_DIRECTORY";
  36     break;
  37   case RenameRequest<I>::STATE_REMOVE_SOURCE_HEADER:
  38     os << "REMOVE_SOURCE_HEADER";
  39     break;
  40   default:
  41     os << "UNKNOWN (" << static_cast<uint32_t>(state) << ")";
  42     break;
  43   }
  44   return os;
  45 }
  46
  47 } // anonymous namespace
  48
  49 template <typename I>
  50 RenameRequest<I>::RenameRequest(I &image_ctx, Context *on_finish,
  51                                 const std::string &dest_name)
  52   : Request<I>(image_ctx, on_finish), m_dest_name(dest_name),
  53     m_source_oid(image_ctx.old_format ? util::old_header_name(image_ctx.name) :
  54                                         util::id_obj_name(image_ctx.name)),
  55     m_dest_oid(image_ctx.old_format ? util::old_header_name(dest_name) :
  56                                       util::id_obj_name(dest_name)) {
  57 }
  58
  59 template <typename I>
  60 void RenameRequest<I>::send_op() {
  61   I &image_ctx = this->m_image_ctx;
  62   if (image_ctx.old_format) {
  63     send_read_source_header();
  64     return;
  65   }
  66   send_read_directory();
  67 }
  68
  69 template <typename I>
  70 bool RenameRequest<I>::should_complete(int r) {
  71   I &image_ctx = this->m_image_ctx;
  72   CephContext *cct = image_ctx.cct;
  73   ldout(cct, 5) << this << " " << __func__ << ": state=" << m_state << ", "
  74                 << "r=" << r << dendl;
  75   r = filter_return_code(r);
  76   if (r < 0) {
  77     if (r == -EEXIST) {
  78       ldout(cct, 1) << "image already exists" << dendl;
  79     } else {
  80       lderr(cct) << "encountered error: " << cpp_strerror(r) << dendl;
  81     }
  82     return true;
  83   }
  84
  85   if (m_state == STATE_READ_DIRECTORY) {
  86     std::string name;
  87     auto it = m_source_name_bl.cbegin();
  88     r = cls_client::dir_get_name_finish(&it, &name);
  89     if (r < 0) {
  90       lderr(cct) << "could not read directory: " << cpp_strerror(r) << dendl;
  91       return true;
  92     }
  93     bool update = false;
  94     {
  95       std::shared_lock image_locker{image_ctx.image_lock};
  96       update = image_ctx.name != name;
  97     }
  98     if (update) {
  99       image_ctx.set_image_name(name);
 100       m_source_oid = util::id_obj_name(name);
 101     }
 102   } else if (m_state == STATE_UPDATE_DIRECTORY) {
 103     // update in-memory name before removing source header
 104     apply();
 105   } else if (m_state == STATE_REMOVE_SOURCE_HEADER) {
 106     return true;
 107   }
 108
 109   std::shared_lock owner_lock{image_ctx.owner_lock};
 110   switch (m_state) {
 111   case STATE_READ_DIRECTORY:
 112     send_read_source_header();
 113     break;
 114   case STATE_READ_SOURCE_HEADER:
 115     send_write_destination_header();
 116     break;
 117   case STATE_WRITE_DEST_HEADER:
 118     send_update_directory();
 119     break;
 120   case STATE_UPDATE_DIRECTORY:
 121     send_remove_source_header();
 122     break;
 123   default:
 124     ceph_abort();
 125     break;
 126   }
 127   return false;
 128 }
 129
 130 template <typename I>
 131 int RenameRequest<I>::filter_return_code(int r) const {
 132   I &image_ctx = this->m_image_ctx;
 133   CephContext *cct = image_ctx.cct;
 134
 135   if (m_state == STATE_READ_SOURCE_HEADER && r == -ENOENT) {
 136     std::shared_lock image_locker{image_ctx.image_lock};
 137     if (image_ctx.name == m_dest_name) {
 138       // signal that replay raced with itself
 139       return -EEXIST;
 140     }
 141   } else if (m_state == STATE_REMOVE_SOURCE_HEADER && r < 0) {
 142     if (r != -ENOENT) {
 143       lderr(cct) << "warning: couldn't remove old source object ("
 144                  << m_source_oid << ")" << dendl;
 145     }
 146     return 0;
 147   }
 148   return r;
 149 }
 150
 151 template <typename I>
 152 void RenameRequest<I>::send_read_directory() {
 153   I &image_ctx = this->m_image_ctx;
 154   CephContext *cct = image_ctx.cct;
 155   ldout(cct, 5) << this << " " << __func__ << dendl;
 156   m_state = STATE_READ_DIRECTORY;
 157
 158   librados::ObjectReadOperation op;
 159   cls_client::dir_get_name_start(&op, image_ctx.id);
 160
 161   auto comp = this->create_callback_completion();
 162   int r = image_ctx.md_ctx.aio_operate(RBD_DIRECTORY, comp, &op,
 163                                        &m_source_name_bl);
 164   ceph_assert(r == 0);
 165   comp->release();
 166 }
 167
 168 template <typename I>
 169 void RenameRequest<I>::send_read_source_header() {
 170   I &image_ctx = this->m_image_ctx;
 171   CephContext *cct = image_ctx.cct;
 172   ldout(cct, 5) << this << " " << __func__ << dendl;
 173   m_state = STATE_READ_SOURCE_HEADER;
 174
 175   librados::ObjectReadOperation op;
 176   op.read(0, 0, NULL, NULL);
 177
 178   // TODO: old code read omap values but there are no omap values on the
 179   //       old format header nor the new format id object
 180   librados::AioCompletion *rados_completion = this->create_callback_completion();
 181   int r = image_ctx.md_ctx.aio_operate(m_source_oid, rados_completion, &op,
 182                                        &m_header_bl);
 183   ceph_assert(r == 0);
 184   rados_completion->release();
 185 }
 186
 187 template <typename I>
 188 void RenameRequest<I>::send_write_destination_header() {
 189   I &image_ctx = this->m_image_ctx;
 190   CephContext *cct = image_ctx.cct;
 191   ldout(cct, 5) << this << " " << __func__ << dendl;
 192   m_state = STATE_WRITE_DEST_HEADER;
 193
 194   librados::ObjectWriteOperation op;
 195   op.create(true);
 196   op.write_full(m_header_bl);
 197
 198   librados::AioCompletion *rados_completion = this->create_callback_completion();
 199   int r = image_ctx.md_ctx.aio_operate(m_dest_oid, rados_completion, &op);
 200   ceph_assert(r == 0);
 201   rados_completion->release();
 202 }
 203
 204 template <typename I>
 205 void RenameRequest<I>::send_update_directory() {
 206   I &image_ctx = this->m_image_ctx;
 207   CephContext *cct = image_ctx.cct;
 208   ldout(cct, 5) << this << " " << __func__ << dendl;
 209   m_state = STATE_UPDATE_DIRECTORY;
 210
 211   librados::ObjectWriteOperation op;
 212   if (image_ctx.old_format) {
 213     bufferlist cmd_bl;
 214     bufferlist empty_bl;
 215     encode(static_cast<__u8>(CEPH_OSD_TMAP_SET), cmd_bl);
 216     encode(m_dest_name, cmd_bl);
 217     encode(empty_bl, cmd_bl);
 218     encode(static_cast<__u8>(CEPH_OSD_TMAP_RM), cmd_bl);
 219     encode(image_ctx.name, cmd_bl);
 220     op.tmap_update(cmd_bl);
 221   } else {
 222     cls_client::dir_rename_image(&op, image_ctx.name, m_dest_name,
 223                                  image_ctx.id);
 224   }
 225
 226   librados::AioCompletion *rados_completion = this->create_callback_completion();
 227   int r = image_ctx.md_ctx.aio_operate(RBD_DIRECTORY, rados_completion, &op);
 228   ceph_assert(r == 0);
 229   rados_completion->release();
 230 }
 231
 232 template <typename I>
 233 void RenameRequest<I>::send_remove_source_header() {
 234   I &image_ctx = this->m_image_ctx;
 235   CephContext *cct = image_ctx.cct;
 236   ldout(cct, 5) << this << " " << __func__ << dendl;
 237   m_state = STATE_REMOVE_SOURCE_HEADER;
 238
 239   librados::ObjectWriteOperation op;
 240   op.remove();
 241
 242   librados::AioCompletion *rados_completion = this->create_callback_completion();
 243   int r = image_ctx.md_ctx.aio_operate(m_source_oid, rados_completion, &op);
 244   ceph_assert(r == 0);
 245   rados_completion->release();
 246 }
 247
 248 template <typename I>
 249 void RenameRequest<I>::apply() {
 250   I &image_ctx = this->m_image_ctx;
 251   image_ctx.set_image_name(m_dest_name);
 252 }
 253
 254 } // namespace operation
 255 } // namespace librbd
 256
 257 template class librbd::operation::RenameRequest<librbd::ImageCtx>;