[ceph.git] / ceph / src / librbd / operation / SnapshotUnprotectRequest.cc

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#include "librbd/operation/SnapshotUnprotectRequest.h"
#include "include/rados/librados.hpp"
#include "include/stringify.h"
#include "common/dout.h"
#include "common/errno.h"
#include "librbd/AsyncObjectThrottle.h"
#include "librbd/ImageCtx.h"
#include "librbd/internal.h"
#include "librbd/Types.h"
#include "librbd/Utils.h"
#include <list>
#include <set>
#include <vector>
#include <boost/lambda/bind.hpp>
#include <boost/lambda/construct.hpp>

#define dout_subsys ceph_subsys_rbd
#undef dout_prefix
#define dout_prefix *_dout << "librbd::SnapshotUnprotectRequest: "

namespace librbd {
namespace operation {

namespace {

typedef std::pair<int64_t, std::string> Pool;
typedef std::vector<Pool> Pools;

template <typename I>
std::ostream& operator<<(std::ostream& os,
                         const typename SnapshotUnprotectRequest<I>::State& state) {
  switch(state) {
  case SnapshotUnprotectRequest<I>::STATE_UNPROTECT_SNAP_START:
    os << "UNPROTECT_SNAP_START";
    break;
  case SnapshotUnprotectRequest<I>::STATE_SCAN_POOL_CHILDREN:
    os << "SCAN_POOL_CHILDREN";
    break;
  case SnapshotUnprotectRequest<I>::STATE_UNPROTECT_SNAP_FINISH:
    os << "UNPROTECT_SNAP_FINISH";
    break;
  case SnapshotUnprotectRequest<I>::STATE_UNPROTECT_SNAP_ROLLBACK:
    os << "UNPROTECT_SNAP_ROLLBACK";
    break;
  default:
    os << "UNKNOWN (" << static_cast<uint32_t>(state) << ")";
    break;
  }
  return os;
}

template <typename I>
class C_ScanPoolChildren : public C_AsyncObjectThrottle<I> {
public:
  C_ScanPoolChildren(AsyncObjectThrottle<I> &throttle, I *image_ctx,
                     const cls::rbd::ParentImageSpec &pspec, const Pools &pools,
                     size_t pool_idx)
    : C_AsyncObjectThrottle<I>(throttle, *image_ctx), m_pspec(pspec),
      m_pool(pools[pool_idx]) {
  }

  int send() override {
    I &image_ctx = this->m_image_ctx;
    ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));

    CephContext *cct = image_ctx.cct;
    ldout(cct, 10) << this << " scanning pool '" << m_pool.second << "'"
                   << dendl;

    librados::Rados rados(image_ctx.md_ctx);
    int64_t base_tier;
    int r = rados.pool_get_base_tier(m_pool.first, &base_tier);
    if (r == -ENOENT) {
      ldout(cct, 1) << "pool '" << m_pool.second << "' no longer exists"
                    << dendl;
      return 1;
    } else if (r < 0) {
      lderr(cct) << "error retrieving base tier for pool '"
                 << m_pool.second << "'" << dendl;
      return r;
    }
    if (m_pool.first != base_tier) {
      // pool is a cache; skip it
      return 1;
    }

    r = util::create_ioctx(image_ctx.md_ctx, "child image", m_pool.first, {},
                           &m_pool_ioctx);
    if (r == -ENOENT) {
      return 1;
    } else if (r < 0) {
      return r;
    }

    librados::ObjectReadOperation op;
    cls_client::get_children_start(&op, m_pspec);

    librados::AioCompletion *rados_completion =
      util::create_rados_callback(this);
    r = m_pool_ioctx.aio_operate(RBD_CHILDREN, rados_completion, &op,
                                 &m_children_bl);
    ceph_assert(r == 0);
    rados_completion->release();
    return 0;
  }

protected:
  void finish(int r) override {
    I &image_ctx = this->m_image_ctx;
    CephContext *cct = image_ctx.cct;

    if (r == 0) {
      auto it = m_children_bl.cbegin();
      r= cls_client::get_children_finish(&it, &m_children);
    }

    ldout(cct, 10) << this << " retrieved children: r=" << r << dendl;
    if (r == -ENOENT) {
      // no children -- proceed with unprotect
      r = 0;
    } else if (r < 0) {
      lderr(cct) << "cannot get children for pool '" << m_pool.second << "'"
                 << dendl;
    } else {
      lderr(cct) << "cannot unprotect: at least " << m_children.size() << " "
                 << "child(ren) [" << joinify(m_children.begin(),
                                              m_children.end(),
                                              std::string(",")) << "] "
                 << "in pool '" << m_pool.second << "'" << dendl;
      r = -EBUSY;
    }
    C_AsyncObjectThrottle<I>::finish(r);
  }

private:
  cls::rbd::ParentImageSpec m_pspec;
  Pool m_pool;

  IoCtx m_pool_ioctx;
  std::set<std::string> m_children;
  bufferlist m_children_bl;
};

} // anonymous namespace

template <typename I>
SnapshotUnprotectRequest<I>::SnapshotUnprotectRequest(I &image_ctx,
                                                      Context *on_finish,
                                                      const cls::rbd::SnapshotNamespace &snap_namespace,
						      const std::string &snap_name)
  : Request<I>(image_ctx, on_finish), m_snap_namespace(snap_namespace),
    m_snap_name(snap_name), m_state(STATE_UNPROTECT_SNAP_START),
    m_ret_val(0), m_snap_id(CEPH_NOSNAP) {
}

template <typename I>
void SnapshotUnprotectRequest<I>::send_op() {
  send_unprotect_snap_start();
}

template <typename I>
bool SnapshotUnprotectRequest<I>::should_complete(int r) {
  I &image_ctx = this->m_image_ctx;
  CephContext *cct = image_ctx.cct;
  ldout(cct, 5) << this << " " << __func__ << ": state=" << m_state << ", "
                << "r=" << r << dendl;
  if (r < 0) {
    if (r == -EINVAL) {
      ldout(cct, 1) << "snapshot is already unprotected" << dendl;
    } else {
      lderr(cct) << "encountered error: " << cpp_strerror(r) << dendl;
    }
    if (m_ret_val == 0) {
      m_ret_val = r;
    }
  }

  // use a different state machine once an error is encountered
  if (m_ret_val < 0) {
    return should_complete_error();
  }

  std::shared_lock owner_lock{image_ctx.owner_lock};
  bool finished = false;
  switch (m_state) {
  case STATE_UNPROTECT_SNAP_START:
    send_scan_pool_children();
    break;
  case STATE_SCAN_POOL_CHILDREN:
    send_unprotect_snap_finish();
    break;
  case STATE_UNPROTECT_SNAP_FINISH:
    finished = true;
    break;
  default:
    ceph_abort();
    break;
  }
  return finished;
}

template <typename I>
bool SnapshotUnprotectRequest<I>::should_complete_error() {
  I &image_ctx = this->m_image_ctx;
  std::shared_lock owner_locker{image_ctx.owner_lock};
  CephContext *cct = image_ctx.cct;
  lderr(cct) << this << " " << __func__ << ": "
             << "ret_val=" << m_ret_val << dendl;

  bool finished = true;
  if (m_state == STATE_SCAN_POOL_CHILDREN ||
      m_state == STATE_UNPROTECT_SNAP_FINISH) {
    send_unprotect_snap_rollback();
    finished = false;
  }
  return finished;
}

template <typename I>
void SnapshotUnprotectRequest<I>::send_unprotect_snap_start() {
  I &image_ctx = this->m_image_ctx;
  ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));

  CephContext *cct = image_ctx.cct;
  ldout(cct, 5) << this << " " << __func__ << dendl;

  int r = verify_and_send_unprotect_snap_start();
  if (r < 0) {
    this->async_complete(r);
    return;
  }
}

template <typename I>
void SnapshotUnprotectRequest<I>::send_scan_pool_children() {
  I &image_ctx = this->m_image_ctx;
  ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));

  CephContext *cct = image_ctx.cct;
  ldout(cct, 5) << this << " " << __func__ << dendl;
  m_state = STATE_SCAN_POOL_CHILDREN;

  // search all pools for children depending on this snapshot
  // TODO add async version of wait_for_latest_osdmap
  librados::Rados rados(image_ctx.md_ctx);
  rados.wait_for_latest_osdmap();

  // protect against pools being renamed/deleted
  std::list<Pool> pool_list;
  rados.pool_list2(pool_list);

  cls::rbd::ParentImageSpec pspec(image_ctx.md_ctx.get_id(),
                                  image_ctx.md_ctx.get_namespace(),
                                  image_ctx.id, m_snap_id);
  Pools pools(pool_list.begin(), pool_list.end());

  Context *ctx = this->create_callback_context();
  typename AsyncObjectThrottle<I>::ContextFactory context_factory(
    boost::lambda::bind(boost::lambda::new_ptr<C_ScanPoolChildren<I> >(),
      boost::lambda::_1, &image_ctx, pspec, pools, boost::lambda::_2));
  AsyncObjectThrottle<I> *throttle = new AsyncObjectThrottle<I>(
    nullptr, image_ctx, context_factory, ctx, NULL, 0, pools.size());
  throttle->start_ops(
    image_ctx.config.template get_val<uint64_t>("rbd_concurrent_management_ops"));
}

template <typename I>
void SnapshotUnprotectRequest<I>::send_unprotect_snap_finish() {
  I &image_ctx = this->m_image_ctx;
  ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));

  CephContext *cct = image_ctx.cct;
  ldout(cct, 5) << this << " " << __func__ << dendl;

  m_state = STATE_UNPROTECT_SNAP_FINISH;

  librados::ObjectWriteOperation op;
  cls_client::set_protection_status(&op, m_snap_id,
                                    RBD_PROTECTION_STATUS_UNPROTECTED);

  librados::AioCompletion *comp = this->create_callback_completion();
  int r = image_ctx.md_ctx.aio_operate(image_ctx.header_oid, comp, &op);
  ceph_assert(r == 0);
  comp->release();
}

template <typename I>
void SnapshotUnprotectRequest<I>::send_unprotect_snap_rollback() {
  I &image_ctx = this->m_image_ctx;
  ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));

  CephContext *cct = image_ctx.cct;
  ldout(cct, 5) << this << " " << __func__ << dendl;

  m_state = STATE_UNPROTECT_SNAP_ROLLBACK;

  librados::ObjectWriteOperation op;
  cls_client::set_protection_status(&op, m_snap_id,
                                    RBD_PROTECTION_STATUS_PROTECTED);

  librados::AioCompletion *comp = this->create_callback_completion();
  int r = image_ctx.md_ctx.aio_operate(image_ctx.header_oid, comp, &op);
  ceph_assert(r == 0);
  comp->release();
}

template <typename I>
int SnapshotUnprotectRequest<I>::verify_and_send_unprotect_snap_start() {
  I &image_ctx = this->m_image_ctx;
  std::shared_lock image_locker{image_ctx.image_lock};

  CephContext *cct = image_ctx.cct;
  if ((image_ctx.features & RBD_FEATURE_LAYERING) == 0) {
    lderr(cct) << "image must support layering" << dendl;
    return -ENOSYS;
  }

  m_snap_id = image_ctx.get_snap_id(m_snap_namespace, m_snap_name);
  if (m_snap_id == CEPH_NOSNAP) {
    return -ENOENT;
  }

  bool is_unprotected;
  int r = image_ctx.is_snap_unprotected(m_snap_id, &is_unprotected);
  if (r < 0) {
    return r;
  }

  if (is_unprotected) {
    lderr(cct) << "snapshot is already unprotected" << dendl;
    return -EINVAL;
  }

  librados::ObjectWriteOperation op;
  cls_client::set_protection_status(&op, m_snap_id,
                                    RBD_PROTECTION_STATUS_UNPROTECTING);

  librados::AioCompletion *comp = this->create_callback_completion();
  r = image_ctx.md_ctx.aio_operate(image_ctx.header_oid, comp, &op);
  ceph_assert(r == 0);
  comp->release();

  // TODO legacy code threw a notification post UNPROTECTING update -- required?
  return 0;
}

} // namespace operation
} // namespace librbd

template class librbd::operation::SnapshotUnprotectRequest<librbd::ImageCtx>;
Commit	Line	Data
7c673cae FG	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/SnapshotUnprotectRequest.h"
	5	#include "include/rados/librados.hpp"
	6	#include "include/stringify.h"
	7	#include "common/dout.h"
	8	#include "common/errno.h"
	9	#include "librbd/AsyncObjectThrottle.h"
	10	#include "librbd/ImageCtx.h"
	11	#include "librbd/internal.h"
	12	#include "librbd/Types.h"
	13	#include "librbd/Utils.h"
	14	#include <list>
	15	#include <set>
	16	#include <vector>
	17	#include <boost/lambda/bind.hpp>
	18	#include <boost/lambda/construct.hpp>
	19
	20	#define dout_subsys ceph_subsys_rbd
	21	#undef dout_prefix
	22	#define dout_prefix *_dout << "librbd::SnapshotUnprotectRequest: "
	23
	24	namespace librbd {
	25	namespace operation {
	26
	27	namespace {
	28
	29	typedef std::pair<int64_t, std::string> Pool;
	30	typedef std::vector<Pool> Pools;
	31
	32	template <typename I>
	33	std::ostream& operator<<(std::ostream& os,
	34	const typename SnapshotUnprotectRequest<I>::State& state) {
	35	switch(state) {
	36	case SnapshotUnprotectRequest<I>::STATE_UNPROTECT_SNAP_START:
	37	os << "UNPROTECT_SNAP_START";
	38	break;
	39	case SnapshotUnprotectRequest<I>::STATE_SCAN_POOL_CHILDREN:
	40	os << "SCAN_POOL_CHILDREN";
	41	break;
	42	case SnapshotUnprotectRequest<I>::STATE_UNPROTECT_SNAP_FINISH:
	43	os << "UNPROTECT_SNAP_FINISH";
	44	break;
	45	case SnapshotUnprotectRequest<I>::STATE_UNPROTECT_SNAP_ROLLBACK:
	46	os << "UNPROTECT_SNAP_ROLLBACK";
	47	break;
	48	default:
	49	os << "UNKNOWN (" << static_cast<uint32_t>(state) << ")";
	50	break;
	51	}
	52	return os;
	53	}
	54
	55	template <typename I>
	56	class C_ScanPoolChildren : public C_AsyncObjectThrottle<I> {
	57	public:
	58	C_ScanPoolChildren(AsyncObjectThrottle<I> &throttle, I *image_ctx,
11fdf7f2	59	const cls::rbd::ParentImageSpec &pspec, const Pools &pools,
7c673cae FG	60	size_t pool_idx)
	61	: C_AsyncObjectThrottle<I>(throttle, *image_ctx), m_pspec(pspec),
	62	m_pool(pools[pool_idx]) {
	63	}
	64
	65	int send() override {
	66	I &image_ctx = this->m_image_ctx;
9f95a23c	67	ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));
7c673cae FG	68
	69	CephContext *cct = image_ctx.cct;
	70	ldout(cct, 10) << this << " scanning pool '" << m_pool.second << "'"
	71	<< dendl;
	72
	73	librados::Rados rados(image_ctx.md_ctx);
	74	int64_t base_tier;
	75	int r = rados.pool_get_base_tier(m_pool.first, &base_tier);
	76	if (r == -ENOENT) {
	77	ldout(cct, 1) << "pool '" << m_pool.second << "' no longer exists"
	78	<< dendl;
	79	return 1;
	80	} else if (r < 0) {
	81	lderr(cct) << "error retrieving base tier for pool '"
	82	<< m_pool.second << "'" << dendl;
	83	return r;
	84	}
	85	if (m_pool.first != base_tier) {
	86	// pool is a cache; skip it
	87	return 1;
	88	}
	89
11fdf7f2 TL	90	r = util::create_ioctx(image_ctx.md_ctx, "child image", m_pool.first, {},
11fdf7f2 TL	91	&m_pool_ioctx);
7c673cae	92	if (r == -ENOENT) {
7c673cae FG	93	return 1;
7c673cae FG	94	} else if (r < 0) {
7c673cae FG	95	return r;
	96	}
	97
	98	librados::ObjectReadOperation op;
	99	cls_client::get_children_start(&op, m_pspec);
	100
	101	librados::AioCompletion *rados_completion =
	102	util::create_rados_callback(this);
	103	r = m_pool_ioctx.aio_operate(RBD_CHILDREN, rados_completion, &op,
	104	&m_children_bl);
11fdf7f2	105	ceph_assert(r == 0);
7c673cae FG	106	rados_completion->release();
	107	return 0;
	108	}
	109
	110	protected:
	111	void finish(int r) override {
	112	I &image_ctx = this->m_image_ctx;
	113	CephContext *cct = image_ctx.cct;
	114
	115	if (r == 0) {
11fdf7f2	116	auto it = m_children_bl.cbegin();
7c673cae FG	117	r= cls_client::get_children_finish(&it, &m_children);
	118	}
	119
	120	ldout(cct, 10) << this << " retrieved children: r=" << r << dendl;
	121	if (r == -ENOENT) {
	122	// no children -- proceed with unprotect
	123	r = 0;
	124	} else if (r < 0) {
	125	lderr(cct) << "cannot get children for pool '" << m_pool.second << "'"
	126	<< dendl;
	127	} else {
	128	lderr(cct) << "cannot unprotect: at least " << m_children.size() << " "
	129	<< "child(ren) [" << joinify(m_children.begin(),
	130	m_children.end(),
	131	std::string(",")) << "] "
	132	<< "in pool '" << m_pool.second << "'" << dendl;
	133	r = -EBUSY;
	134	}
	135	C_AsyncObjectThrottle<I>::finish(r);
	136	}
	137
	138	private:
11fdf7f2	139	cls::rbd::ParentImageSpec m_pspec;
7c673cae FG	140	Pool m_pool;
	141
	142	IoCtx m_pool_ioctx;
	143	std::set<std::string> m_children;
	144	bufferlist m_children_bl;
	145	};
	146
	147	} // anonymous namespace
	148
	149	template <typename I>
	150	SnapshotUnprotectRequest<I>::SnapshotUnprotectRequest(I &image_ctx,
	151	Context *on_finish,
	152	const cls::rbd::SnapshotNamespace &snap_namespace,
	153	const std::string &snap_name)
	154	: Request<I>(image_ctx, on_finish), m_snap_namespace(snap_namespace),
11fdf7f2 TL	155	m_snap_name(snap_name), m_state(STATE_UNPROTECT_SNAP_START),
11fdf7f2 TL	156	m_ret_val(0), m_snap_id(CEPH_NOSNAP) {
7c673cae FG	157	}
	158
	159	template <typename I>
	160	void SnapshotUnprotectRequest<I>::send_op() {
	161	send_unprotect_snap_start();
	162	}
	163
	164	template <typename I>
	165	bool SnapshotUnprotectRequest<I>::should_complete(int r) {
	166	I &image_ctx = this->m_image_ctx;
	167	CephContext *cct = image_ctx.cct;
	168	ldout(cct, 5) << this << " " << __func__ << ": state=" << m_state << ", "
	169	<< "r=" << r << dendl;
	170	if (r < 0) {
	171	if (r == -EINVAL) {
	172	ldout(cct, 1) << "snapshot is already unprotected" << dendl;
	173	} else {
	174	lderr(cct) << "encountered error: " << cpp_strerror(r) << dendl;
	175	}
	176	if (m_ret_val == 0) {
	177	m_ret_val = r;
	178	}
	179	}
	180
	181	// use a different state machine once an error is encountered
	182	if (m_ret_val < 0) {
	183	return should_complete_error();
	184	}
	185
9f95a23c	186	std::shared_lock owner_lock{image_ctx.owner_lock};
7c673cae FG	187	bool finished = false;
	188	switch (m_state) {
	189	case STATE_UNPROTECT_SNAP_START:
	190	send_scan_pool_children();
	191	break;
	192	case STATE_SCAN_POOL_CHILDREN:
	193	send_unprotect_snap_finish();
	194	break;
	195	case STATE_UNPROTECT_SNAP_FINISH:
	196	finished = true;
	197	break;
	198	default:
11fdf7f2	199	ceph_abort();
7c673cae FG	200	break;
	201	}
	202	return finished;
	203	}
	204
	205	template <typename I>
	206	bool SnapshotUnprotectRequest<I>::should_complete_error() {
	207	I &image_ctx = this->m_image_ctx;
9f95a23c	208	std::shared_lock owner_locker{image_ctx.owner_lock};
7c673cae FG	209	CephContext *cct = image_ctx.cct;
	210	lderr(cct) << this << " " << __func__ << ": "
	211	<< "ret_val=" << m_ret_val << dendl;
	212
	213	bool finished = true;
	214	if (m_state == STATE_SCAN_POOL_CHILDREN \|\|
	215	m_state == STATE_UNPROTECT_SNAP_FINISH) {
	216	send_unprotect_snap_rollback();
	217	finished = false;
	218	}
	219	return finished;
	220	}
	221
	222	template <typename I>
	223	void SnapshotUnprotectRequest<I>::send_unprotect_snap_start() {
	224	I &image_ctx = this->m_image_ctx;
9f95a23c	225	ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));
7c673cae FG	226
	227	CephContext *cct = image_ctx.cct;
	228	ldout(cct, 5) << this << " " << __func__ << dendl;
	229
7c673cae FG	230	int r = verify_and_send_unprotect_snap_start();
	231	if (r < 0) {
	232	this->async_complete(r);
	233	return;
	234	}
	235	}
	236
	237	template <typename I>
	238	void SnapshotUnprotectRequest<I>::send_scan_pool_children() {
	239	I &image_ctx = this->m_image_ctx;
9f95a23c	240	ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));
7c673cae FG	241
	242	CephContext *cct = image_ctx.cct;
	243	ldout(cct, 5) << this << " " << __func__ << dendl;
	244	m_state = STATE_SCAN_POOL_CHILDREN;
	245
	246	// search all pools for children depending on this snapshot
	247	// TODO add async version of wait_for_latest_osdmap
	248	librados::Rados rados(image_ctx.md_ctx);
	249	rados.wait_for_latest_osdmap();
	250
	251	// protect against pools being renamed/deleted
	252	std::list<Pool> pool_list;
	253	rados.pool_list2(pool_list);
	254
11fdf7f2 TL	255	cls::rbd::ParentImageSpec pspec(image_ctx.md_ctx.get_id(),
	256	image_ctx.md_ctx.get_namespace(),
	257	image_ctx.id, m_snap_id);
7c673cae FG	258	Pools pools(pool_list.begin(), pool_list.end());
	259
	260	Context *ctx = this->create_callback_context();
	261	typename AsyncObjectThrottle<I>::ContextFactory context_factory(
	262	boost::lambda::bind(boost::lambda::new_ptr<C_ScanPoolChildren<I> >(),
	263	boost::lambda::_1, &image_ctx, pspec, pools, boost::lambda::_2));
	264	AsyncObjectThrottle<I> *throttle = new AsyncObjectThrottle<I>(
	265	nullptr, image_ctx, context_factory, ctx, NULL, 0, pools.size());
11fdf7f2 TL	266	throttle->start_ops(
11fdf7f2 TL	267	image_ctx.config.template get_val<uint64_t>("rbd_concurrent_management_ops"));
7c673cae FG	268	}
	269
	270	template <typename I>
	271	void SnapshotUnprotectRequest<I>::send_unprotect_snap_finish() {
	272	I &image_ctx = this->m_image_ctx;
9f95a23c	273	ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));
7c673cae FG	274
	275	CephContext *cct = image_ctx.cct;
	276	ldout(cct, 5) << this << " " << __func__ << dendl;
	277
	278	m_state = STATE_UNPROTECT_SNAP_FINISH;
	279
	280	librados::ObjectWriteOperation op;
	281	cls_client::set_protection_status(&op, m_snap_id,
	282	RBD_PROTECTION_STATUS_UNPROTECTED);
	283
	284	librados::AioCompletion *comp = this->create_callback_completion();
	285	int r = image_ctx.md_ctx.aio_operate(image_ctx.header_oid, comp, &op);
11fdf7f2	286	ceph_assert(r == 0);
7c673cae FG	287	comp->release();
	288	}
	289
	290	template <typename I>
	291	void SnapshotUnprotectRequest<I>::send_unprotect_snap_rollback() {
	292	I &image_ctx = this->m_image_ctx;
9f95a23c	293	ceph_assert(ceph_mutex_is_locked(image_ctx.owner_lock));
7c673cae FG	294
	295	CephContext *cct = image_ctx.cct;
	296	ldout(cct, 5) << this << " " << __func__ << dendl;
	297
	298	m_state = STATE_UNPROTECT_SNAP_ROLLBACK;
	299
	300	librados::ObjectWriteOperation op;
	301	cls_client::set_protection_status(&op, m_snap_id,
	302	RBD_PROTECTION_STATUS_PROTECTED);
	303
	304	librados::AioCompletion *comp = this->create_callback_completion();
	305	int r = image_ctx.md_ctx.aio_operate(image_ctx.header_oid, comp, &op);
11fdf7f2	306	ceph_assert(r == 0);
7c673cae FG	307	comp->release();
	308	}
	309
	310	template <typename I>
	311	int SnapshotUnprotectRequest<I>::verify_and_send_unprotect_snap_start() {
	312	I &image_ctx = this->m_image_ctx;
9f95a23c	313	std::shared_lock image_locker{image_ctx.image_lock};
7c673cae FG	314
	315	CephContext *cct = image_ctx.cct;
	316	if ((image_ctx.features & RBD_FEATURE_LAYERING) == 0) {
	317	lderr(cct) << "image must support layering" << dendl;
	318	return -ENOSYS;
	319	}
	320
	321	m_snap_id = image_ctx.get_snap_id(m_snap_namespace, m_snap_name);
	322	if (m_snap_id == CEPH_NOSNAP) {
	323	return -ENOENT;
	324	}
	325
	326	bool is_unprotected;
	327	int r = image_ctx.is_snap_unprotected(m_snap_id, &is_unprotected);
	328	if (r < 0) {
	329	return r;
	330	}
	331
	332	if (is_unprotected) {
	333	lderr(cct) << "snapshot is already unprotected" << dendl;
	334	return -EINVAL;
	335	}
	336
	337	librados::ObjectWriteOperation op;
	338	cls_client::set_protection_status(&op, m_snap_id,
	339	RBD_PROTECTION_STATUS_UNPROTECTING);
	340
	341	librados::AioCompletion *comp = this->create_callback_completion();
	342	r = image_ctx.md_ctx.aio_operate(image_ctx.header_oid, comp, &op);
11fdf7f2	343	ceph_assert(r == 0);
7c673cae FG	344	comp->release();
	345
	346	// TODO legacy code threw a notification post UNPROTECTING update -- required?
	347	return 0;
	348	}
	349
	350	} // namespace operation
	351	} // namespace librbd
	352
	353	template class librbd::operation::SnapshotUnprotectRequest<librbd::ImageCtx>;