diff options
Diffstat (limited to '')
| -rw-r--r-- | src/librbd/cache/ObjectCacherObjectDispatch.cc | 467 |
1 files changed, 467 insertions, 0 deletions
diff --git a/src/librbd/cache/ObjectCacherObjectDispatch.cc b/src/librbd/cache/ObjectCacherObjectDispatch.cc new file mode 100644 index 000000000..81d5a7188 --- /dev/null +++ b/src/librbd/cache/ObjectCacherObjectDispatch.cc @@ -0,0 +1,467 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include "librbd/cache/ObjectCacherObjectDispatch.h" +#include "include/neorados/RADOS.hpp" +#include "common/errno.h" +#include "librbd/ImageCtx.h" +#include "librbd/Journal.h" +#include "librbd/Utils.h" +#include "librbd/asio/ContextWQ.h" +#include "librbd/cache/ObjectCacherWriteback.h" +#include "librbd/io/ObjectDispatchSpec.h" +#include "librbd/io/ObjectDispatcherInterface.h" +#include "librbd/io/ReadResult.h" +#include "librbd/io/Types.h" +#include "librbd/io/Utils.h" +#include "osd/osd_types.h" +#include "osdc/WritebackHandler.h" +#include <vector> + +#define dout_subsys ceph_subsys_rbd +#undef dout_prefix +#define dout_prefix *_dout << "librbd::cache::ObjectCacherObjectDispatch: " \ + << this << " " << __func__ << ": " + +namespace librbd { +namespace cache { + +using librbd::util::data_object_name; + +namespace { + +typedef std::vector<ObjectExtent> ObjectExtents; + +} // anonymous namespace + +template <typename I> +struct ObjectCacherObjectDispatch<I>::C_InvalidateCache : public Context { + ObjectCacherObjectDispatch* dispatcher; + bool purge_on_error; + Context *on_finish; + + C_InvalidateCache(ObjectCacherObjectDispatch* dispatcher, + bool purge_on_error, Context *on_finish) + : dispatcher(dispatcher), purge_on_error(purge_on_error), + on_finish(on_finish) { + } + + void finish(int r) override { + ceph_assert(ceph_mutex_is_locked(dispatcher->m_cache_lock)); + auto cct = dispatcher->m_image_ctx->cct; + + if (r == -EBLOCKLISTED) { + lderr(cct) << "blocklisted during flush (purging)" << dendl; + dispatcher->m_object_cacher->purge_set(dispatcher->m_object_set); + } else if (r < 0 && purge_on_error) { + lderr(cct) << "failed to invalidate cache (purging): " + << cpp_strerror(r) << dendl; + dispatcher->m_object_cacher->purge_set(dispatcher->m_object_set); + } else if (r != 0) { + lderr(cct) << "failed to invalidate cache: " << cpp_strerror(r) << dendl; + } + + auto unclean = dispatcher->m_object_cacher->release_set( + dispatcher->m_object_set); + if (unclean == 0) { + r = 0; + } else { + lderr(cct) << "could not release all objects from cache: " + << unclean << " bytes remain" << dendl; + if (r == 0) { + r = -EBUSY; + } + } + + on_finish->complete(r); + } +}; + +template <typename I> +ObjectCacherObjectDispatch<I>::ObjectCacherObjectDispatch( + I* image_ctx, size_t max_dirty, bool writethrough_until_flush) + : m_image_ctx(image_ctx), m_max_dirty(max_dirty), + m_writethrough_until_flush(writethrough_until_flush), + m_cache_lock(ceph::make_mutex(util::unique_lock_name( + "librbd::cache::ObjectCacherObjectDispatch::cache_lock", this))) { + ceph_assert(m_image_ctx->data_ctx.is_valid()); +} + +template <typename I> +ObjectCacherObjectDispatch<I>::~ObjectCacherObjectDispatch() { + delete m_object_cacher; + delete m_object_set; + + delete m_writeback_handler; +} + +template <typename I> +void ObjectCacherObjectDispatch<I>::init() { + auto cct = m_image_ctx->cct; + ldout(cct, 5) << dendl; + + m_cache_lock.lock(); + ldout(cct, 5) << "enabling caching..." << dendl; + m_writeback_handler = new ObjectCacherWriteback(m_image_ctx, m_cache_lock); + + auto init_max_dirty = m_max_dirty; + if (m_writethrough_until_flush) { + init_max_dirty = 0; + } + + auto cache_size = + m_image_ctx->config.template get_val<Option::size_t>("rbd_cache_size"); + auto target_dirty = + m_image_ctx->config.template get_val<Option::size_t>("rbd_cache_target_dirty"); + auto max_dirty_age = + m_image_ctx->config.template get_val<double>("rbd_cache_max_dirty_age"); + auto block_writes_upfront = + m_image_ctx->config.template get_val<bool>("rbd_cache_block_writes_upfront"); + auto max_dirty_object = + m_image_ctx->config.template get_val<uint64_t>("rbd_cache_max_dirty_object"); + + ldout(cct, 5) << "Initial cache settings:" + << " size=" << cache_size + << " num_objects=" << 10 + << " max_dirty=" << init_max_dirty + << " target_dirty=" << target_dirty + << " max_dirty_age=" << max_dirty_age << dendl; + + m_object_cacher = new ObjectCacher(cct, m_image_ctx->perfcounter->get_name(), + *m_writeback_handler, m_cache_lock, + nullptr, nullptr, cache_size, + 10, /* reset this in init */ + init_max_dirty, target_dirty, + max_dirty_age, block_writes_upfront); + + // size object cache appropriately + if (max_dirty_object == 0) { + max_dirty_object = std::min<uint64_t>( + 2000, std::max<uint64_t>(10, cache_size / 100 / + sizeof(ObjectCacher::Object))); + } + ldout(cct, 5) << " cache bytes " << cache_size + << " -> about " << max_dirty_object << " objects" << dendl; + m_object_cacher->set_max_objects(max_dirty_object); + + m_object_set = new ObjectCacher::ObjectSet(nullptr, + m_image_ctx->data_ctx.get_id(), 0); + m_object_cacher->start(); + m_cache_lock.unlock(); + + // add ourself to the IO object dispatcher chain + if (m_max_dirty > 0) { + m_image_ctx->disable_zero_copy = true; + } + m_image_ctx->io_object_dispatcher->register_dispatch(this); +} + +template <typename I> +void ObjectCacherObjectDispatch<I>::shut_down(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 5) << dendl; + + // chain shut down in reverse order + + // shut down the cache + on_finish = new LambdaContext([this, on_finish](int r) { + m_object_cacher->stop(); + on_finish->complete(r); + }); + + // ensure we aren't holding the cache lock post-flush + on_finish = util::create_async_context_callback(*m_image_ctx, on_finish); + + // invalidate any remaining cache entries + on_finish = new C_InvalidateCache(this, true, on_finish); + + // flush all pending writeback state + std::lock_guard locker{m_cache_lock}; + m_object_cacher->release_set(m_object_set); + m_object_cacher->flush_set(m_object_set, on_finish); +} + +template <typename I> +bool ObjectCacherObjectDispatch<I>::read( + uint64_t object_no, io::ReadExtents* extents, IOContext io_context, + int op_flags, int read_flags, const ZTracer::Trace &parent_trace, + uint64_t* version, int* object_dispatch_flags, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + // IO chained in reverse order + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "object_no=" << object_no << " " << *extents << dendl; + + if (extents->size() == 0) { + ldout(cct, 20) << "no extents to read" << dendl; + return false; + } + + if (version != nullptr) { + // we currently don't cache read versions + // and don't support reading more than one extent + return false; + } + + // ensure we aren't holding the cache lock post-read + on_dispatched = util::create_async_context_callback(*m_image_ctx, + on_dispatched); + + // embed the RBD-internal read flags in the genenric RADOS op_flags and + op_flags = ((op_flags & ~ObjectCacherWriteback::READ_FLAGS_MASK) | + ((read_flags << ObjectCacherWriteback::READ_FLAGS_SHIFT) & + ObjectCacherWriteback::READ_FLAGS_MASK)); + + ceph::bufferlist* bl; + if (extents->size() > 1) { + auto req = new io::ReadResult::C_ObjectReadMergedExtents( + cct, extents, on_dispatched); + on_dispatched = req; + bl = &req->bl; + } else { + bl = &extents->front().bl; + } + + m_image_ctx->image_lock.lock_shared(); + auto rd = m_object_cacher->prepare_read( + io_context->read_snap().value_or(CEPH_NOSNAP), bl, op_flags); + m_image_ctx->image_lock.unlock_shared(); + + uint64_t off = 0; + for (auto& read_extent: *extents) { + ObjectExtent extent(data_object_name(m_image_ctx, object_no), object_no, + read_extent.offset, read_extent.length, 0); + extent.oloc.pool = m_image_ctx->data_ctx.get_id(); + extent.buffer_extents.push_back({off, read_extent.length}); + rd->extents.push_back(extent); + off += read_extent.length; + } + + ZTracer::Trace trace(parent_trace); + *dispatch_result = io::DISPATCH_RESULT_COMPLETE; + + m_cache_lock.lock(); + int r = m_object_cacher->readx(rd, m_object_set, on_dispatched, &trace); + m_cache_lock.unlock(); + if (r != 0) { + on_dispatched->complete(r); + } + return true; +} + +template <typename I> +bool ObjectCacherObjectDispatch<I>::discard( + uint64_t object_no, uint64_t object_off, uint64_t object_len, + IOContext io_context, int discard_flags, + const ZTracer::Trace &parent_trace, int* object_dispatch_flags, + uint64_t* journal_tid, io::DispatchResult* dispatch_result, + Context** on_finish, Context* on_dispatched) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "object_no=" << object_no << " " << object_off << "~" + << object_len << dendl; + + ObjectExtents object_extents; + object_extents.emplace_back(data_object_name(m_image_ctx, object_no), + object_no, object_off, object_len, 0); + + // discard the cache state after changes are committed to disk (and to + // prevent races w/ readahead) + auto ctx = *on_finish; + *on_finish = new LambdaContext( + [this, object_extents, ctx](int r) { + m_cache_lock.lock(); + m_object_cacher->discard_set(m_object_set, object_extents); + m_cache_lock.unlock(); + + ctx->complete(r); + }); + + // ensure we aren't holding the cache lock post-write + on_dispatched = util::create_async_context_callback(*m_image_ctx, + on_dispatched); + + *dispatch_result = io::DISPATCH_RESULT_CONTINUE; + + // ensure any in-flight writeback is complete before advancing + // the discard request + std::lock_guard locker{m_cache_lock}; + m_object_cacher->discard_writeback(m_object_set, object_extents, + on_dispatched); + return true; +} + +template <typename I> +bool ObjectCacherObjectDispatch<I>::write( + uint64_t object_no, uint64_t object_off, ceph::bufferlist&& data, + IOContext io_context, int op_flags, int write_flags, + std::optional<uint64_t> assert_version, + const ZTracer::Trace &parent_trace, int* object_dispatch_flags, + uint64_t* journal_tid, io::DispatchResult* dispatch_result, + Context** on_finish, Context* on_dispatched) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "object_no=" << object_no << " " << object_off << "~" + << data.length() << dendl; + + // ensure we aren't holding the cache lock post-write + on_dispatched = util::create_async_context_callback(*m_image_ctx, + on_dispatched); + + // cache layer does not handle version checking + if (assert_version.has_value() || + (write_flags & io::OBJECT_WRITE_FLAG_CREATE_EXCLUSIVE) != 0) { + ObjectExtents object_extents; + object_extents.emplace_back(data_object_name(m_image_ctx, object_no), + object_no, object_off, data.length(), 0); + + *dispatch_result = io::DISPATCH_RESULT_CONTINUE; + + // ensure any in-flight writeback is complete before advancing + // the write request + std::lock_guard locker{m_cache_lock}; + m_object_cacher->discard_writeback(m_object_set, object_extents, + on_dispatched); + return true; + } + + SnapContext snapc; + if (io_context->write_snap_context()) { + auto write_snap_context = *io_context->write_snap_context(); + snapc = SnapContext(write_snap_context.first, + {write_snap_context.second.begin(), + write_snap_context.second.end()}); + } + + m_image_ctx->image_lock.lock_shared(); + ObjectCacher::OSDWrite *wr = m_object_cacher->prepare_write( + snapc, data, ceph::real_time::min(), op_flags, *journal_tid); + m_image_ctx->image_lock.unlock_shared(); + + ObjectExtent extent(data_object_name(m_image_ctx, object_no), + object_no, object_off, data.length(), 0); + extent.oloc.pool = m_image_ctx->data_ctx.get_id(); + extent.buffer_extents.push_back({0, data.length()}); + wr->extents.push_back(extent); + + ZTracer::Trace trace(parent_trace); + *dispatch_result = io::DISPATCH_RESULT_COMPLETE; + + std::lock_guard locker{m_cache_lock}; + m_object_cacher->writex(wr, m_object_set, on_dispatched, &trace); + return true; +} + +template <typename I> +bool ObjectCacherObjectDispatch<I>::write_same( + uint64_t object_no, uint64_t object_off, uint64_t object_len, + io::LightweightBufferExtents&& buffer_extents, ceph::bufferlist&& data, + IOContext io_context, int op_flags, + const ZTracer::Trace &parent_trace, int* object_dispatch_flags, + uint64_t* journal_tid, io::DispatchResult* dispatch_result, + Context** on_finish, Context* on_dispatched) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "object_no=" << object_no << " " << object_off << "~" + << object_len << dendl; + + // ObjectCacher doesn't support write-same so convert to regular write + io::LightweightObjectExtent extent(object_no, object_off, object_len, 0); + extent.buffer_extents = std::move(buffer_extents); + + bufferlist ws_data; + io::util::assemble_write_same_extent(extent, data, &ws_data, true); + + return write(object_no, object_off, std::move(ws_data), io_context, op_flags, + 0, std::nullopt, parent_trace, object_dispatch_flags, + journal_tid, dispatch_result, on_finish, on_dispatched); +} + +template <typename I> +bool ObjectCacherObjectDispatch<I>::compare_and_write( + uint64_t object_no, uint64_t object_off, ceph::bufferlist&& cmp_data, + ceph::bufferlist&& write_data, IOContext io_context, int op_flags, + const ZTracer::Trace &parent_trace, uint64_t* mismatch_offset, + int* object_dispatch_flags, uint64_t* journal_tid, + io::DispatchResult* dispatch_result, Context** on_finish, + Context* on_dispatched) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << "object_no=" << object_no << " " << object_off << "~" + << cmp_data.length() << dendl; + + // pass-through the compare-and-write request since it's not a supported + // operation of the ObjectCacher + + // ensure we aren't holding the cache lock post-flush + on_dispatched = util::create_async_context_callback(*m_image_ctx, + on_dispatched); + + // flush any pending writes from the cache + ZTracer::Trace trace(parent_trace); + *dispatch_result = io::DISPATCH_RESULT_CONTINUE; + + ObjectExtents object_extents; + object_extents.emplace_back(data_object_name(m_image_ctx, object_no), + object_no, object_off, cmp_data.length(), 0); + + std::lock_guard cache_locker{m_cache_lock}; + m_object_cacher->flush_set(m_object_set, object_extents, &trace, + on_dispatched); + return true; +} + +template <typename I> +bool ObjectCacherObjectDispatch<I>::flush( + io::FlushSource flush_source, const ZTracer::Trace &parent_trace, + uint64_t* journal_tid, io::DispatchResult* dispatch_result, + Context** on_finish, Context* on_dispatched) { + auto cct = m_image_ctx->cct; + ldout(cct, 20) << dendl; + + // ensure we aren't holding the cache lock post-flush + on_dispatched = util::create_async_context_callback(*m_image_ctx, + on_dispatched); + + std::lock_guard locker{m_cache_lock}; + if (flush_source == io::FLUSH_SOURCE_USER && !m_user_flushed) { + m_user_flushed = true; + if (m_writethrough_until_flush && m_max_dirty > 0) { + m_object_cacher->set_max_dirty(m_max_dirty); + ldout(cct, 5) << "saw first user flush, enabling writeback" << dendl; + } + } + + *dispatch_result = io::DISPATCH_RESULT_CONTINUE; + m_object_cacher->flush_set(m_object_set, on_dispatched); + return true; +} + +template <typename I> +bool ObjectCacherObjectDispatch<I>::invalidate_cache(Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 5) << dendl; + + // ensure we aren't holding the cache lock post-flush + on_finish = util::create_async_context_callback(*m_image_ctx, on_finish); + + // invalidate any remaining cache entries + on_finish = new C_InvalidateCache(this, false, on_finish); + + std::lock_guard locker{m_cache_lock}; + m_object_cacher->release_set(m_object_set); + m_object_cacher->flush_set(m_object_set, on_finish); + return true; +} + +template <typename I> +bool ObjectCacherObjectDispatch<I>::reset_existence_cache( + Context* on_finish) { + auto cct = m_image_ctx->cct; + ldout(cct, 5) << dendl; + + std::lock_guard locker{m_cache_lock}; + m_object_cacher->clear_nonexistence(m_object_set); + return false; +} + +} // namespace cache +} // namespace librbd + +template class librbd::cache::ObjectCacherObjectDispatch<librbd::ImageCtx>; |