Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

6 files changed, 4173 insertions, 0 deletions

diff --git a/src/neorados/CMakeLists.txt b/src/neorados/CMakeLists.txt
new file mode 100644
index 000000000..8695b48f0
--- /dev/null
+++ b/src/neorados/CMakeLists.txt
@@ -0,0 +1,42 @@
+add_library(neorados_objs OBJECT
+  RADOSImpl.cc)
+compile_with_fmt(neorados_objs)
+add_library(neorados_api_obj OBJECT
+  RADOS.cc)
+compile_with_fmt(neorados_api_obj)
+
+add_library(libneorados STATIC
+  $<TARGET_OBJECTS:neorados_api_obj>
+  $<TARGET_OBJECTS:neorados_objs>)
+target_link_libraries(libneorados PRIVATE
+  osdc ceph-common cls_lock_client fmt::fmt
+  ${BLKID_LIBRARIES} ${CRYPTO_LIBS} ${EXTRALIBS})
+
+# if(ENABLE_SHARED)
+#   add_library(libneorados ${CEPH_SHARED}
+#     $<TARGET_OBJECTS:neorados_api_obj>
+#     $<TARGET_OBJECTS:neorados_objs>
+#     $<TARGET_OBJECTS:common_buffer_obj>)
+#   set_target_properties(libneorados PROPERTIES
+#     OUTPUT_NAME RADOS
+#     VERSION 0.0.1
+#     SOVERSION 1
+#     CXX_VISIBILITY_PRESET hidden
+#     VISIBILITY_INLINES_HIDDEN ON)
+#   if(NOT APPLE)
+#     set_property(TARGET libneorados APPEND_STRING PROPERTY
+#       LINK_FLAGS " -Wl,--exclude-libs,ALL")
+#   endif()
+# else(ENABLE_SHARED)
+#   add_library(libneorados STATIC
+#     $<TARGET_OBJECTS:neorados_api_obj>
+#     $<TARGET_OBJECTS:neorados_objs>)
+# endif(ENABLE_SHARED)
+# target_link_libraries(libneorados PRIVATE
+#   osdc ceph-common cls_lock_client
+#   ${BLKID_LIBRARIES} ${CRYPTO_LIBS} ${EXTRALIBS})
+# target_link_libraries(libneorados ${rados_libs})
+# install(TARGETS libneorados DESTINATION ${CMAKE_INSTALL_LIBDIR})
+add_library(neorados_cls_fifo STATIC cls/fifo.cc)
+target_link_libraries(neorados_cls_fifo PRIVATE
+	libneorados ceph-common fmt::fmt)
diff --git a/src/neorados/RADOS.cc b/src/neorados/RADOS.cc
new file mode 100644
index 000000000..927f995cc
--- /dev/null
+++ b/src/neorados/RADOS.cc
@@ -0,0 +1,1736 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2018 Red Hat <contact@redhat.com>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#define BOOST_BIND_NO_PLACEHOLDERS
+
+#include <optional>
+#include <string_view>
+
+#include <boost/intrusive_ptr.hpp>
+
+#include <fmt/format.h>
+
+#include "include/ceph_fs.h"
+
+#include "common/ceph_context.h"
+#include "common/ceph_argparse.h"
+#include "common/common_init.h"
+#include "common/hobject.h"
+#include "common/EventTrace.h"
+
+#include "global/global_init.h"
+
+#include "osd/osd_types.h"
+#include "osdc/error_code.h"
+
+#include "neorados/RADOSImpl.h"
+#include "include/neorados/RADOS.hpp"
+
+namespace bc = boost::container;
+namespace bs = boost::system;
+namespace ca = ceph::async;
+namespace cb = ceph::buffer;
+
+namespace neorados {
+// Object
+
+Object::Object() {
+  static_assert(impl_size >= sizeof(object_t));
+  new (&impl) object_t();
+}
+
+Object::Object(const char* s) {
+  static_assert(impl_size >= sizeof(object_t));
+  new (&impl) object_t(s);
+}
+
+Object::Object(std::string_view s) {
+  static_assert(impl_size >= sizeof(object_t));
+  new (&impl) object_t(s);
+}
+
+Object::Object(std::string&& s) {
+  static_assert(impl_size >= sizeof(object_t));
+  new (&impl) object_t(std::move(s));
+}
+
+Object::Object(const std::string& s) {
+  static_assert(impl_size >= sizeof(object_t));
+  new (&impl) object_t(s);
+}
+
+Object::~Object() {
+  reinterpret_cast<object_t*>(&impl)->~object_t();
+}
+
+Object::Object(const Object& o) {
+  static_assert(impl_size >= sizeof(object_t));
+  new (&impl) object_t(*reinterpret_cast<const object_t*>(&o.impl));
+}
+Object& Object::operator =(const Object& o) {
+  *reinterpret_cast<object_t*>(&impl) =
+    *reinterpret_cast<const object_t*>(&o.impl);
+  return *this;
+}
+Object::Object(Object&& o) {
+  static_assert(impl_size >= sizeof(object_t));
+  new (&impl) object_t(std::move(*reinterpret_cast<object_t*>(&o.impl)));
+}
+Object& Object::operator =(Object&& o) {
+  *reinterpret_cast<object_t*>(&impl) =
+    std::move(*reinterpret_cast<object_t*>(&o.impl));
+  return *this;
+}
+
+Object::operator std::string_view() const {
+  return std::string_view(reinterpret_cast<const object_t*>(&impl)->name);
+}
+
+bool operator <(const Object& lhs, const Object& rhs) {
+  return (*reinterpret_cast<const object_t*>(&lhs.impl) <
+	  *reinterpret_cast<const object_t*>(&rhs.impl));
+}
+bool operator <=(const Object& lhs, const Object& rhs) {
+  return (*reinterpret_cast<const object_t*>(&lhs.impl) <=
+	  *reinterpret_cast<const object_t*>(&rhs.impl));
+}
+bool operator >=(const Object& lhs, const Object& rhs) {
+  return (*reinterpret_cast<const object_t*>(&lhs.impl) >=
+	  *reinterpret_cast<const object_t*>(&rhs.impl));
+}
+bool operator >(const Object& lhs, const Object& rhs) {
+  return (*reinterpret_cast<const object_t*>(&lhs.impl) >
+	  *reinterpret_cast<const object_t*>(&rhs.impl));
+}
+
+bool operator ==(const Object& lhs, const Object& rhs) {
+  return (*reinterpret_cast<const object_t*>(&lhs.impl) ==
+	  *reinterpret_cast<const object_t*>(&rhs.impl));
+}
+bool operator !=(const Object& lhs, const Object& rhs) {
+  return (*reinterpret_cast<const object_t*>(&lhs.impl) !=
+	  *reinterpret_cast<const object_t*>(&rhs.impl));
+}
+
+std::ostream& operator <<(std::ostream& m, const Object& o) {
+  return (m << *reinterpret_cast<const object_t*>(&o.impl));
+}
+
+// IOContext
+
+struct IOContextImpl {
+  object_locator_t oloc;
+  snapid_t snap_seq = CEPH_NOSNAP;
+  SnapContext snapc;
+  int extra_op_flags = 0;
+};
+
+IOContext::IOContext() {
+  static_assert(impl_size >= sizeof(IOContextImpl));
+  new (&impl) IOContextImpl();
+}
+
+IOContext::IOContext(std::int64_t _pool) : IOContext() {
+  pool(_pool);
+}
+
+IOContext::IOContext(std::int64_t _pool, std::string_view _ns)
+  : IOContext() {
+  pool(_pool);
+  ns(_ns);
+}
+
+IOContext::IOContext(std::int64_t _pool, std::string&& _ns)
+  : IOContext() {
+  pool(_pool);
+  ns(std::move(_ns));
+}
+
+IOContext::~IOContext() {
+  reinterpret_cast<IOContextImpl*>(&impl)->~IOContextImpl();
+}
+
+IOContext::IOContext(const IOContext& rhs) {
+  static_assert(impl_size >= sizeof(IOContextImpl));
+  new (&impl) IOContextImpl(*reinterpret_cast<const IOContextImpl*>(&rhs.impl));
+}
+
+IOContext& IOContext::operator =(const IOContext& rhs) {
+  *reinterpret_cast<IOContextImpl*>(&impl) =
+    *reinterpret_cast<const IOContextImpl*>(&rhs.impl);
+  return *this;
+}
+
+IOContext::IOContext(IOContext&& rhs) {
+  static_assert(impl_size >= sizeof(IOContextImpl));
+  new (&impl) IOContextImpl(
+    std::move(*reinterpret_cast<IOContextImpl*>(&rhs.impl)));
+}
+
+IOContext& IOContext::operator =(IOContext&& rhs) {
+  *reinterpret_cast<IOContextImpl*>(&impl) =
+    std::move(*reinterpret_cast<IOContextImpl*>(&rhs.impl));
+  return *this;
+}
+
+std::int64_t IOContext::pool() const {
+  return reinterpret_cast<const IOContextImpl*>(&impl)->oloc.pool;
+}
+
+void IOContext::pool(std::int64_t _pool) {
+  reinterpret_cast<IOContextImpl*>(&impl)->oloc.pool = _pool;
+}
+
+std::string_view IOContext::ns() const {
+  return reinterpret_cast<const IOContextImpl*>(&impl)->oloc.nspace;
+}
+
+void IOContext::ns(std::string_view _ns) {
+  reinterpret_cast<IOContextImpl*>(&impl)->oloc.nspace = _ns;
+}
+
+void IOContext::ns(std::string&& _ns) {
+  reinterpret_cast<IOContextImpl*>(&impl)->oloc.nspace = std::move(_ns);
+}
+
+std::optional<std::string_view> IOContext::key() const {
+  auto& oloc = reinterpret_cast<const IOContextImpl*>(&impl)->oloc;
+  if (oloc.key.empty())
+    return std::nullopt;
+  else
+    return std::string_view(oloc.key);
+}
+
+void IOContext::key(std::string_view _key) {
+  auto& oloc = reinterpret_cast<IOContextImpl*>(&impl)->oloc;
+  oloc.hash = -1;
+  oloc.key = _key;
+}
+
+void IOContext::key(std::string&&_key) {
+  auto& oloc = reinterpret_cast<IOContextImpl*>(&impl)->oloc;
+  oloc.hash = -1;
+  oloc.key = std::move(_key);
+}
+
+void IOContext::clear_key() {
+  auto& oloc = reinterpret_cast<IOContextImpl*>(&impl)->oloc;
+  oloc.hash = -1;
+  oloc.key.clear();
+}
+
+std::optional<std::int64_t> IOContext::hash() const {
+  auto& oloc = reinterpret_cast<const IOContextImpl*>(&impl)->oloc;
+  if (oloc.hash < 0)
+    return std::nullopt;
+  else
+    return oloc.hash;
+}
+
+void IOContext::hash(std::int64_t _hash) {
+  auto& oloc = reinterpret_cast<IOContextImpl*>(&impl)->oloc;
+  oloc.hash = _hash;
+  oloc.key.clear();
+}
+
+void IOContext::clear_hash() {
+  auto& oloc = reinterpret_cast<IOContextImpl*>(&impl)->oloc;
+  oloc.hash = -1;
+  oloc.key.clear();
+}
+
+
+std::optional<std::uint64_t> IOContext::read_snap() const {
+  auto& snap_seq = reinterpret_cast<const IOContextImpl*>(&impl)->snap_seq;
+  if (snap_seq == CEPH_NOSNAP)
+    return std::nullopt;
+  else
+    return snap_seq;
+}
+void IOContext::read_snap(std::optional<std::uint64_t> _snapid) {
+  auto& snap_seq = reinterpret_cast<IOContextImpl*>(&impl)->snap_seq;
+  snap_seq = _snapid.value_or(CEPH_NOSNAP);
+}
+
+std::optional<
+  std::pair<std::uint64_t,
+	    std::vector<std::uint64_t>>> IOContext::write_snap_context() const {
+  auto& snapc = reinterpret_cast<const IOContextImpl*>(&impl)->snapc;
+  if (snapc.empty()) {
+    return std::nullopt;
+  } else {
+    std::vector<uint64_t> v(snapc.snaps.begin(), snapc.snaps.end());
+    return std::make_optional(std::make_pair(uint64_t(snapc.seq), v));
+  }
+}
+
+void IOContext::write_snap_context(
+  std::optional<std::pair<std::uint64_t, std::vector<std::uint64_t>>> _snapc) {
+  auto& snapc = reinterpret_cast<IOContextImpl*>(&impl)->snapc;
+  if (!_snapc) {
+    snapc.clear();
+  } else {
+    SnapContext n(_snapc->first, { _snapc->second.begin(), _snapc->second.end()});
+    if (!n.is_valid()) {
+      throw bs::system_error(EINVAL,
+			     bs::system_category(),
+			     "Invalid snap context.");
+
+    } else {
+      snapc = n;
+    }
+  }
+}
+
+bool IOContext::full_try() const {
+  const auto ioc = reinterpret_cast<const IOContextImpl*>(&impl);
+  return (ioc->extra_op_flags & CEPH_OSD_FLAG_FULL_TRY) != 0;
+}
+
+void IOContext::full_try(bool _full_try) {
+  auto ioc = reinterpret_cast<IOContextImpl*>(&impl);
+  if (_full_try) {
+    ioc->extra_op_flags |= CEPH_OSD_FLAG_FULL_TRY;
+  } else {
+    ioc->extra_op_flags &= ~CEPH_OSD_FLAG_FULL_TRY;
+  }
+}
+
+bool operator <(const IOContext& lhs, const IOContext& rhs) {
+  const auto l = reinterpret_cast<const IOContextImpl*>(&lhs.impl);
+  const auto r = reinterpret_cast<const IOContextImpl*>(&rhs.impl);
+
+  return (std::tie(l->oloc.pool, l->oloc.nspace, l->oloc.key) <
+	  std::tie(r->oloc.pool, r->oloc.nspace, r->oloc.key));
+}
+
+bool operator <=(const IOContext& lhs, const IOContext& rhs) {
+  const auto l = reinterpret_cast<const IOContextImpl*>(&lhs.impl);
+  const auto r = reinterpret_cast<const IOContextImpl*>(&rhs.impl);
+
+  return (std::tie(l->oloc.pool, l->oloc.nspace, l->oloc.key) <=
+	  std::tie(r->oloc.pool, r->oloc.nspace, r->oloc.key));
+}
+
+bool operator >=(const IOContext& lhs, const IOContext& rhs) {
+  const auto l = reinterpret_cast<const IOContextImpl*>(&lhs.impl);
+  const auto r = reinterpret_cast<const IOContextImpl*>(&rhs.impl);
+
+  return (std::tie(l->oloc.pool, l->oloc.nspace, l->oloc.key) >=
+	  std::tie(r->oloc.pool, r->oloc.nspace, r->oloc.key));
+}
+
+bool operator >(const IOContext& lhs, const IOContext& rhs) {
+  const auto l = reinterpret_cast<const IOContextImpl*>(&lhs.impl);
+  const auto r = reinterpret_cast<const IOContextImpl*>(&rhs.impl);
+
+  return (std::tie(l->oloc.pool, l->oloc.nspace, l->oloc.key) >
+	  std::tie(r->oloc.pool, r->oloc.nspace, r->oloc.key));
+}
+
+bool operator ==(const IOContext& lhs, const IOContext& rhs) {
+  const auto l = reinterpret_cast<const IOContextImpl*>(&lhs.impl);
+  const auto r = reinterpret_cast<const IOContextImpl*>(&rhs.impl);
+
+  return (std::tie(l->oloc.pool, l->oloc.nspace, l->oloc.key) ==
+	  std::tie(r->oloc.pool, r->oloc.nspace, r->oloc.key));
+}
+
+bool operator !=(const IOContext& lhs, const IOContext& rhs) {
+  const auto l = reinterpret_cast<const IOContextImpl*>(&lhs.impl);
+  const auto r = reinterpret_cast<const IOContextImpl*>(&rhs.impl);
+
+  return (std::tie(l->oloc.pool, l->oloc.nspace, l->oloc.key) !=
+	  std::tie(r->oloc.pool, r->oloc.nspace, r->oloc.key));
+}
+
+std::ostream& operator <<(std::ostream& m, const IOContext& o) {
+  const auto l = reinterpret_cast<const IOContextImpl*>(&o.impl);
+  return (m << l->oloc.pool << ":" << l->oloc.nspace << ":" << l->oloc.key);
+}
+
+
+// Op
+
+struct OpImpl {
+  ObjectOperation op;
+  std::optional<ceph::real_time> mtime;
+
+  OpImpl() = default;
+
+  OpImpl(const OpImpl& rhs) = delete;
+  OpImpl(OpImpl&& rhs) = default;
+
+  OpImpl& operator =(const OpImpl& rhs) = delete;
+  OpImpl& operator =(OpImpl&& rhs) = default;
+};
+
+Op::Op() {
+  static_assert(Op::impl_size >= sizeof(OpImpl));
+  new (&impl) OpImpl;
+}
+
+Op::Op(Op&& rhs) {
+  new (&impl) OpImpl(std::move(*reinterpret_cast<OpImpl*>(&rhs.impl)));
+}
+Op& Op::operator =(Op&& rhs) {
+  reinterpret_cast<OpImpl*>(&impl)->~OpImpl();
+  new (&impl) OpImpl(std::move(*reinterpret_cast<OpImpl*>(&rhs.impl)));
+  return *this;
+}
+Op::~Op() {
+  reinterpret_cast<OpImpl*>(&impl)->~OpImpl();
+}
+
+void Op::set_excl() {
+  reinterpret_cast<OpImpl*>(&impl)->op.set_last_op_flags(CEPH_OSD_OP_FLAG_EXCL);
+}
+void Op::set_failok() {
+  reinterpret_cast<OpImpl*>(&impl)->op.set_last_op_flags(
+    CEPH_OSD_OP_FLAG_FAILOK);
+}
+void Op::set_fadvise_random() {
+  reinterpret_cast<OpImpl*>(&impl)->op.set_last_op_flags(
+    CEPH_OSD_OP_FLAG_FADVISE_RANDOM);
+}
+void Op::set_fadvise_sequential() {
+  reinterpret_cast<OpImpl*>(&impl)->op.set_last_op_flags(
+    CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL);
+}
+void Op::set_fadvise_willneed() {
+  reinterpret_cast<OpImpl*>(&impl)->op.set_last_op_flags(
+    CEPH_OSD_OP_FLAG_FADVISE_WILLNEED);
+}
+void Op::set_fadvise_dontneed() {
+  reinterpret_cast<OpImpl*>(&impl)->op.set_last_op_flags(
+    CEPH_OSD_OP_FLAG_FADVISE_DONTNEED);
+}
+void Op::set_fadvise_nocache() {
+  reinterpret_cast<OpImpl*>(&impl)->op.set_last_op_flags(
+    CEPH_OSD_OP_FLAG_FADVISE_NOCACHE);
+}
+
+void Op::cmpext(uint64_t off, bufferlist&& cmp_bl, std::size_t* s) {
+  reinterpret_cast<OpImpl*>(&impl)->op.cmpext(off, std::move(cmp_bl), nullptr,
+					      s);
+}
+void Op::cmpxattr(std::string_view name, cmpxattr_op op, const bufferlist& val) {
+  reinterpret_cast<OpImpl*>(&impl)->
+    op.cmpxattr(name, std::uint8_t(op), CEPH_OSD_CMPXATTR_MODE_STRING, val);
+}
+void Op::cmpxattr(std::string_view name, cmpxattr_op op, std::uint64_t val) {
+  bufferlist bl;
+  encode(val, bl);
+  reinterpret_cast<OpImpl*>(&impl)->
+    op.cmpxattr(name, std::uint8_t(op), CEPH_OSD_CMPXATTR_MODE_U64, bl);
+}
+
+void Op::assert_version(uint64_t ver) {
+  reinterpret_cast<OpImpl*>(&impl)->op.assert_version(ver);
+}
+void Op::assert_exists() {
+  reinterpret_cast<OpImpl*>(&impl)->op.stat(
+    nullptr,
+    static_cast<ceph::real_time*>(nullptr),
+    static_cast<bs::error_code*>(nullptr));
+}
+void Op::cmp_omap(const bc::flat_map<
+		  std::string, std::pair<cb::list,
+		  int>>& assertions) {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_cmp(assertions, nullptr);
+}
+
+void Op::exec(std::string_view cls, std::string_view method,
+	      const bufferlist& inbl,
+	      cb::list* out,
+	      bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.call(cls, method, inbl, ec, out);
+}
+
+void Op::exec(std::string_view cls, std::string_view method,
+	      const bufferlist& inbl,
+	      fu2::unique_function<void(bs::error_code,
+					const cb::list&) &&> f) {
+  reinterpret_cast<OpImpl*>(&impl)->op.call(cls, method, inbl, std::move(f));
+}
+
+void Op::exec(std::string_view cls, std::string_view method,
+	      const bufferlist& inbl,
+	      fu2::unique_function<void(bs::error_code, int,
+					const cb::list&) &&> f) {
+  reinterpret_cast<OpImpl*>(&impl)->op.call(cls, method, inbl, std::move(f));
+}
+
+void Op::exec(std::string_view cls, std::string_view method,
+	      const bufferlist& inbl, bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.call(cls, method, inbl, ec);
+}
+
+void Op::balance_reads() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_BALANCE_READS;
+}
+void Op::localize_reads() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_LOCALIZE_READS;
+}
+void Op::order_reads_writes() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_RWORDERED;
+}
+void Op::ignore_cache() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_IGNORE_CACHE;
+}
+void Op::skiprwlocks() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_SKIPRWLOCKS;
+}
+void Op::ignore_overlay() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_IGNORE_OVERLAY;
+}
+void Op::full_try() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_FULL_TRY;
+}
+void Op::full_force() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_FULL_FORCE;
+}
+void Op::ignore_redirect() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_IGNORE_REDIRECT;
+}
+void Op::ordersnap() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_ORDERSNAP;
+}
+void Op::returnvec() {
+  reinterpret_cast<OpImpl*>(&impl)->op.flags |= CEPH_OSD_FLAG_RETURNVEC;
+}
+
+std::size_t Op::size() const {
+  return reinterpret_cast<const OpImpl*>(&impl)->op.size();
+}
+
+std::ostream& operator <<(std::ostream& m, const Op& o) {
+  return m << reinterpret_cast<const OpImpl*>(&o.impl)->op;
+}
+
+
+// ---
+
+// ReadOp / WriteOp
+
+void ReadOp::read(size_t off, uint64_t len, cb::list* out,
+		  bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.read(off, len, ec, out);
+}
+
+void ReadOp::get_xattr(std::string_view name, cb::list* out,
+		       bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.getxattr(name, ec, out);
+}
+
+void ReadOp::get_omap_header(cb::list* out,
+			     bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_get_header(ec, out);
+}
+
+void ReadOp::sparse_read(uint64_t off, uint64_t len, cb::list* out,
+			 std::vector<std::pair<std::uint64_t,
+			 std::uint64_t>>* extents,
+			 bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.sparse_read(off, len, ec, extents, out);
+}
+
+void ReadOp::stat(std::uint64_t* size, ceph::real_time* mtime,
+		  bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.stat(size, mtime, ec);
+}
+
+void ReadOp::get_omap_keys(std::optional<std::string_view> start_after,
+			   std::uint64_t max_return,
+			   bc::flat_set<std::string>* keys,
+			   bool* done,
+			   bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_get_keys(start_after, max_return,
+						     ec, keys, done);
+}
+
+void ReadOp::get_xattrs(bc::flat_map<std::string,
+			cb::list>* kv,
+			bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.getxattrs(ec, kv);
+}
+
+void ReadOp::get_omap_vals(std::optional<std::string_view> start_after,
+			   std::optional<std::string_view> filter_prefix,
+			   uint64_t max_return,
+			   bc::flat_map<std::string,
+			   cb::list>* kv,
+			   bool* done,
+			   bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_get_vals(start_after, filter_prefix,
+						     max_return, ec, kv, done);
+}
+
+void ReadOp::get_omap_vals_by_keys(
+  const bc::flat_set<std::string>& keys,
+  bc::flat_map<std::string, cb::list>* kv,
+  bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_get_vals_by_keys(keys, ec, kv);
+}
+
+void ReadOp::list_watchers(std::vector<ObjWatcher>* watchers,
+			   bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)-> op.list_watchers(watchers, ec);
+}
+
+void ReadOp::list_snaps(SnapSet* snaps,
+			bs::error_code* ec) {
+  reinterpret_cast<OpImpl*>(&impl)->op.list_snaps(snaps, nullptr, ec);
+}
+
+// WriteOp
+
+void WriteOp::set_mtime(ceph::real_time t) {
+  auto o = reinterpret_cast<OpImpl*>(&impl);
+  o->mtime = t;
+}
+
+void WriteOp::create(bool exclusive) {
+  reinterpret_cast<OpImpl*>(&impl)->op.create(exclusive);
+}
+
+void WriteOp::write(uint64_t off, bufferlist&& bl) {
+  reinterpret_cast<OpImpl*>(&impl)->op.write(off, bl);
+}
+
+void WriteOp::write_full(bufferlist&& bl) {
+  reinterpret_cast<OpImpl*>(&impl)->op.write_full(bl);
+}
+
+void WriteOp::writesame(uint64_t off, uint64_t write_len, bufferlist&& bl) {
+  reinterpret_cast<OpImpl*>(&impl)->op.writesame(off, write_len, bl);
+}
+
+void WriteOp::append(bufferlist&& bl) {
+  reinterpret_cast<OpImpl*>(&impl)->op.append(bl);
+}
+
+void WriteOp::remove() {
+  reinterpret_cast<OpImpl*>(&impl)->op.remove();
+}
+
+void WriteOp::truncate(uint64_t off) {
+  reinterpret_cast<OpImpl*>(&impl)->op.truncate(off);
+}
+
+void WriteOp::zero(uint64_t off, uint64_t len) {
+  reinterpret_cast<OpImpl*>(&impl)->op.zero(off, len);
+}
+
+void WriteOp::rmxattr(std::string_view name) {
+  reinterpret_cast<OpImpl*>(&impl)->op.rmxattr(name);
+}
+
+void WriteOp::setxattr(std::string_view name,
+                       bufferlist&& bl) {
+  reinterpret_cast<OpImpl*>(&impl)->op.setxattr(name, bl);
+}
+
+void WriteOp::rollback(uint64_t snapid) {
+  reinterpret_cast<OpImpl*>(&impl)->op.rollback(snapid);
+}
+
+void WriteOp::set_omap(
+  const bc::flat_map<std::string, cb::list>& map) {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_set(map);
+}
+
+void WriteOp::set_omap_header(bufferlist&& bl) {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_set_header(bl);
+}
+
+void WriteOp::clear_omap() {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_clear();
+}
+
+void WriteOp::rm_omap_keys(
+  const bc::flat_set<std::string>& to_rm) {
+  reinterpret_cast<OpImpl*>(&impl)->op.omap_rm_keys(to_rm);
+}
+
+void WriteOp::set_alloc_hint(uint64_t expected_object_size,
+			     uint64_t expected_write_size,
+			     alloc_hint::alloc_hint_t flags) {
+  using namespace alloc_hint;
+  static_assert(sequential_write ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_SEQUENTIAL_WRITE));
+  static_assert(random_write ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_RANDOM_WRITE));
+  static_assert(sequential_read ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_SEQUENTIAL_READ));
+  static_assert(random_read ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_RANDOM_READ));
+  static_assert(append_only ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_APPEND_ONLY));
+  static_assert(immutable ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_IMMUTABLE));
+  static_assert(shortlived ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_SHORTLIVED));
+  static_assert(longlived ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_LONGLIVED));
+  static_assert(compressible ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE));
+  static_assert(incompressible ==
+		static_cast<int>(CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE));
+
+  reinterpret_cast<OpImpl*>(&impl)->op.set_alloc_hint(expected_object_size,
+						      expected_write_size,
+						      flags);
+}
+
+// RADOS
+
+RADOS::Builder& RADOS::Builder::add_conf_file(std::string_view f) {
+  if (conf_files)
+    *conf_files += (", " + std::string(f));
+  else
+    conf_files = std::string(f);
+  return *this;
+}
+
+void RADOS::Builder::build(boost::asio::io_context& ioctx,
+			   std::unique_ptr<BuildComp> c) {
+  constexpr auto env = CODE_ENVIRONMENT_LIBRARY;
+  CephInitParameters ci(env);
+  if (name)
+    ci.name.set(CEPH_ENTITY_TYPE_CLIENT, *name);
+  else
+    ci.name.set(CEPH_ENTITY_TYPE_CLIENT, "admin");
+  uint32_t flags = 0;
+  if (no_default_conf)
+    flags |= CINIT_FLAG_NO_DEFAULT_CONFIG_FILE;
+  if (no_mon_conf)
+    flags |= CINIT_FLAG_NO_MON_CONFIG;
+
+  CephContext *cct = common_preinit(ci, env, flags);
+  if (cluster)
+    cct->_conf->cluster = *cluster;
+
+  if (no_mon_conf)
+    cct->_conf->no_mon_config = true;
+
+  // TODO: Come up with proper error codes here. Maybe augment the
+  // functions with a default bs::error_code* parameter to
+  // pass back.
+  {
+    std::ostringstream ss;
+    auto r = cct->_conf.parse_config_files(conf_files ? conf_files->data() : nullptr,
+					   &ss, flags);
+    if (r < 0)
+      c->post(std::move(c), ceph::to_error_code(r), RADOS{nullptr});
+  }
+
+  cct->_conf.parse_env(cct->get_module_type());
+
+  for (const auto& [n, v] : configs) {
+    std::stringstream ss;
+    auto r = cct->_conf.set_val(n, v, &ss);
+    if (r < 0)
+      c->post(std::move(c), ceph::to_error_code(-EINVAL), RADOS{nullptr});
+  }
+
+  if (!no_mon_conf) {
+    MonClient mc_bootstrap(cct, ioctx);
+    // TODO This function should return an error code.
+    auto err = mc_bootstrap.get_monmap_and_config();
+    if (err < 0)
+      c->post(std::move(c), ceph::to_error_code(err), RADOS{nullptr});
+  }
+  if (!cct->_log->is_started()) {
+    cct->_log->start();
+  }
+  common_init_finish(cct);
+
+  RADOS::make_with_cct(cct, ioctx, std::move(c));
+}
+
+void RADOS::make_with_cct(CephContext* cct,
+			  boost::asio::io_context& ioctx,
+			  std::unique_ptr<BuildComp> c) {
+  try {
+    auto r = new detail::NeoClient{std::make_unique<detail::RADOS>(ioctx, cct)};
+    r->objecter->wait_for_osd_map(
+      [c = std::move(c), r = std::unique_ptr<detail::Client>(r)]() mutable {
+	c->dispatch(std::move(c), bs::error_code{},
+		    RADOS{std::move(r)});
+      });
+  } catch (const bs::system_error& err) {
+    c->post(std::move(c), err.code(), RADOS{nullptr});
+  }
+}
+
+RADOS RADOS::make_with_librados(librados::Rados& rados) {
+  return RADOS{std::make_unique<detail::RadosClient>(rados.client)};
+}
+
+RADOS::RADOS() = default;
+
+RADOS::RADOS(std::unique_ptr<detail::Client> impl)
+  : impl(std::move(impl)) {}
+
+RADOS::RADOS(RADOS&&) = default;
+RADOS& RADOS::operator =(RADOS&&) = default;
+
+RADOS::~RADOS() = default;
+
+RADOS::executor_type RADOS::get_executor() const {
+  return impl->ioctx.get_executor();
+}
+
+boost::asio::io_context& RADOS::get_io_context() {
+  return impl->ioctx;
+}
+
+void RADOS::execute(const Object& o, const IOContext& _ioc, ReadOp&& _op,
+		    cb::list* bl,
+		    std::unique_ptr<ReadOp::Completion> c, version_t* objver,
+		    const blkin_trace_info *trace_info) {
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  auto ioc = reinterpret_cast<const IOContextImpl*>(&_ioc.impl);
+  auto op = reinterpret_cast<OpImpl*>(&_op.impl);
+  auto flags = op->op.flags | ioc->extra_op_flags;
+
+  ZTracer::Trace trace;
+  if (trace_info) {
+    ZTracer::Trace parent_trace("", nullptr, trace_info);
+    trace.init("rados execute", &impl->objecter->trace_endpoint, &parent_trace);
+  }
+
+  trace.event("init");
+  impl->objecter->read(
+    *oid, ioc->oloc, std::move(op->op), ioc->snap_seq, bl, flags,
+    std::move(c), objver, nullptr /* data_offset */, 0 /* features */, &trace);
+
+  trace.event("submitted");
+}
+
+void RADOS::execute(const Object& o, const IOContext& _ioc, WriteOp&& _op,
+		    std::unique_ptr<WriteOp::Completion> c, version_t* objver,
+		    const blkin_trace_info *trace_info) {
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  auto ioc = reinterpret_cast<const IOContextImpl*>(&_ioc.impl);
+  auto op = reinterpret_cast<OpImpl*>(&_op.impl);
+  auto flags = op->op.flags | ioc->extra_op_flags;
+  ceph::real_time mtime;
+  if (op->mtime)
+    mtime = *op->mtime;
+  else
+    mtime = ceph::real_clock::now();
+
+  ZTracer::Trace trace;
+  if (trace_info) {
+    ZTracer::Trace parent_trace("", nullptr, trace_info);
+    trace.init("rados execute", &impl->objecter->trace_endpoint, &parent_trace);
+  }
+
+  trace.event("init");
+  impl->objecter->mutate(
+    *oid, ioc->oloc, std::move(op->op), ioc->snapc,
+    mtime, flags,
+    std::move(c), objver, osd_reqid_t{}, &trace);
+  trace.event("submitted");
+}
+
+void RADOS::execute(const Object& o, std::int64_t pool, ReadOp&& _op,
+		    cb::list* bl,
+		    std::unique_ptr<ReadOp::Completion> c,
+		    std::optional<std::string_view> ns,
+		    std::optional<std::string_view> key,
+		    version_t* objver) {
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  auto op = reinterpret_cast<OpImpl*>(&_op.impl);
+  auto flags = op->op.flags;
+  object_locator_t oloc;
+  oloc.pool = pool;
+  if (ns)
+    oloc.nspace = *ns;
+  if (key)
+    oloc.key = *key;
+
+  impl->objecter->read(
+    *oid, oloc, std::move(op->op), CEPH_NOSNAP, bl, flags,
+    std::move(c), objver);
+}
+
+void RADOS::execute(const Object& o, std::int64_t pool, WriteOp&& _op,
+		    std::unique_ptr<WriteOp::Completion> c,
+		    std::optional<std::string_view> ns,
+		    std::optional<std::string_view> key,
+		    version_t* objver) {
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  auto op = reinterpret_cast<OpImpl*>(&_op.impl);
+  auto flags = op->op.flags;
+  object_locator_t oloc;
+  oloc.pool = pool;
+  if (ns)
+    oloc.nspace = *ns;
+  if (key)
+    oloc.key = *key;
+
+  ceph::real_time mtime;
+  if (op->mtime)
+    mtime = *op->mtime;
+  else
+    mtime = ceph::real_clock::now();
+
+  impl->objecter->mutate(
+    *oid, oloc, std::move(op->op), {},
+    mtime, flags,
+    std::move(c), objver);
+}
+
+boost::uuids::uuid RADOS::get_fsid() const noexcept {
+  return impl->monclient.get_fsid().uuid;
+}
+
+
+void RADOS::lookup_pool(std::string_view name,
+			std::unique_ptr<LookupPoolComp> c)
+{
+  // I kind of want to make lookup_pg_pool return
+  // std::optional<int64_t> since it can only return one error code.
+  int64_t ret = impl->objecter->with_osdmap(
+    std::mem_fn(&OSDMap::lookup_pg_pool_name),
+    name);
+  if (ret < 0) {
+    impl->objecter->wait_for_latest_osdmap(
+      [name = std::string(name), c = std::move(c),
+       objecter = impl->objecter]
+      (bs::error_code ec) mutable {
+	int64_t ret =
+	  objecter->with_osdmap([&](const OSDMap &osdmap) {
+	    return osdmap.lookup_pg_pool_name(name);
+	  });
+	if (ret < 0)
+	  ca::dispatch(std::move(c), osdc_errc::pool_dne,
+		       std::int64_t(0));
+	else
+	  ca::dispatch(std::move(c), bs::error_code{}, ret);
+      });
+  } else if (ret < 0) {
+    ca::post(std::move(c), osdc_errc::pool_dne,
+		 std::int64_t(0));
+  } else {
+    ca::post(std::move(c), bs::error_code{}, ret);
+  }
+}
+
+
+std::optional<uint64_t> RADOS::get_pool_alignment(int64_t pool_id)
+{
+  return impl->objecter->with_osdmap(
+    [pool_id](const OSDMap &o) -> std::optional<uint64_t> {
+      if (!o.have_pg_pool(pool_id)) {
+	throw bs::system_error(
+	  ENOENT, bs::system_category(),
+	  "Cannot find pool in OSDMap.");
+      } else if (o.get_pg_pool(pool_id)->requires_aligned_append()) {
+	return o.get_pg_pool(pool_id)->required_alignment();
+      } else {
+	return std::nullopt;
+      }
+    });
+}
+
+void RADOS::list_pools(std::unique_ptr<LSPoolsComp> c) {
+  impl->objecter->with_osdmap(
+    [&](OSDMap& o) {
+      std::vector<std::pair<std::int64_t, std::string>> v;
+      for (auto p : o.get_pools())
+	v.push_back(std::make_pair(p.first, o.get_pool_name(p.first)));
+      ca::dispatch(std::move(c), std::move(v));
+    });
+}
+
+void RADOS::create_pool_snap(std::int64_t pool,
+			     std::string_view snapName,
+			     std::unique_ptr<SimpleOpComp> c)
+{
+  impl->objecter->create_pool_snap(
+    pool, snapName,
+    Objecter::PoolOp::OpComp::create(
+      get_executor(),
+      [c = std::move(c)](bs::error_code e, const bufferlist&) mutable {
+	ca::dispatch(std::move(c), e);
+      }));
+}
+
+void RADOS::allocate_selfmanaged_snap(int64_t pool,
+				      std::unique_ptr<SMSnapComp> c) {
+  impl->objecter->allocate_selfmanaged_snap(
+    pool,
+    ca::Completion<void(bs::error_code, snapid_t)>::create(
+      get_executor(),
+      [c = std::move(c)](bs::error_code e, snapid_t snap) mutable {
+	ca::dispatch(std::move(c), e, snap);
+      }));
+}
+
+void RADOS::delete_pool_snap(std::int64_t pool,
+			     std::string_view snapName,
+			     std::unique_ptr<SimpleOpComp> c)
+{
+  impl->objecter->delete_pool_snap(
+    pool, snapName,
+    Objecter::PoolOp::OpComp::create(
+      get_executor(),
+      [c = std::move(c)](bs::error_code e, const bufferlist&) mutable {
+	ca::dispatch(std::move(c), e);
+      }));
+}
+
+void RADOS::delete_selfmanaged_snap(std::int64_t pool,
+				    std::uint64_t snap,
+				    std::unique_ptr<SimpleOpComp> c)
+{
+  impl->objecter->delete_selfmanaged_snap(
+    pool, snap,
+    Objecter::PoolOp::OpComp::create(
+      get_executor(),
+      [c = std::move(c)](bs::error_code e, const bufferlist&) mutable {
+	ca::dispatch(std::move(c), e);
+      }));
+}
+
+void RADOS::create_pool(std::string_view name,
+			std::optional<int> crush_rule,
+			std::unique_ptr<SimpleOpComp> c)
+{
+  impl->objecter->create_pool(
+    name,
+    Objecter::PoolOp::OpComp::create(
+      get_executor(),
+      [c = std::move(c)](bs::error_code e, const bufferlist&) mutable {
+	ca::dispatch(std::move(c), e);
+      }),
+      crush_rule.value_or(-1));
+}
+
+void RADOS::delete_pool(std::string_view name,
+			std::unique_ptr<SimpleOpComp> c)
+{
+  impl->objecter->delete_pool(
+    name,
+    Objecter::PoolOp::OpComp::create(
+      get_executor(),
+      [c = std::move(c)](bs::error_code e, const bufferlist&) mutable {
+	ca::dispatch(std::move(c), e);
+      }));
+}
+
+void RADOS::delete_pool(std::int64_t pool,
+			std::unique_ptr<SimpleOpComp> c)
+{
+  impl->objecter->delete_pool(
+    pool,
+    Objecter::PoolOp::OpComp::create(
+      get_executor(),
+      [c = std::move(c)](bs::error_code e, const bufferlist&) mutable {
+	ca::dispatch(std::move(c), e);
+      }));
+}
+
+void RADOS::stat_pools(const std::vector<std::string>& pools,
+		       std::unique_ptr<PoolStatComp> c) {
+  impl->objecter->get_pool_stats(
+    pools,
+    [c = std::move(c)]
+    (bs::error_code ec,
+     bc::flat_map<std::string, pool_stat_t> rawresult,
+     bool per_pool) mutable {
+      bc::flat_map<std::string, PoolStats> result;
+      for (auto p = rawresult.begin(); p != rawresult.end(); ++p) {
+	auto& pv = result[p->first];
+	auto& pstat = p->second;
+	store_statfs_t &statfs = pstat.store_stats;
+	uint64_t allocated_bytes = pstat.get_allocated_data_bytes(per_pool) +
+	  pstat.get_allocated_omap_bytes(per_pool);
+	// FIXME: raw_used_rate is unknown hence use 1.0 here
+	// meaning we keep net amount aggregated over all replicas
+	// Not a big deal so far since this field isn't exposed
+	uint64_t user_bytes = pstat.get_user_data_bytes(1.0, per_pool) +
+	  pstat.get_user_omap_bytes(1.0, per_pool);
+
+	object_stat_sum_t *sum = &p->second.stats.sum;
+	pv.num_kb = shift_round_up(allocated_bytes, 10);
+	pv.num_bytes = allocated_bytes;
+	pv.num_objects = sum->num_objects;
+	pv.num_object_clones = sum->num_object_clones;
+	pv.num_object_copies = sum->num_object_copies;
+	pv.num_objects_missing_on_primary = sum->num_objects_missing_on_primary;
+	pv.num_objects_unfound = sum->num_objects_unfound;
+	pv.num_objects_degraded = sum->num_objects_degraded;
+	pv.num_rd = sum->num_rd;
+	pv.num_rd_kb = sum->num_rd_kb;
+	pv.num_wr = sum->num_wr;
+	pv.num_wr_kb = sum->num_wr_kb;
+	pv.num_user_bytes = user_bytes;
+	pv.compressed_bytes_orig = statfs.data_compressed_original;
+	pv.compressed_bytes = statfs.data_compressed;
+	pv.compressed_bytes_alloc = statfs.data_compressed_allocated;
+      }
+
+      ca::dispatch(std::move(c), ec, std::move(result), per_pool);
+    });
+}
+
+void RADOS::stat_fs(std::optional<std::int64_t> _pool,
+		    std::unique_ptr<StatFSComp> c) {
+  boost::optional<int64_t> pool;
+  if (_pool)
+    pool = *pool;
+  impl->objecter->get_fs_stats(
+    pool,
+    [c = std::move(c)](bs::error_code ec, const struct ceph_statfs s) mutable {
+      FSStats fso{s.kb, s.kb_used, s.kb_avail, s.num_objects};
+      c->dispatch(std::move(c), ec, std::move(fso));
+    });
+}
+
+// --- Watch/Notify
+
+void RADOS::watch(const Object& o, const IOContext& _ioc,
+		  std::optional<std::chrono::seconds> timeout, WatchCB&& cb,
+		  std::unique_ptr<WatchComp> c) {
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  auto ioc = reinterpret_cast<const IOContextImpl*>(&_ioc.impl);
+
+  ObjectOperation op;
+
+  auto linger_op = impl->objecter->linger_register(*oid, ioc->oloc,
+                                                   ioc->extra_op_flags);
+  uint64_t cookie = linger_op->get_cookie();
+  linger_op->handle = std::move(cb);
+  op.watch(cookie, CEPH_OSD_WATCH_OP_WATCH, timeout.value_or(0s).count());
+  bufferlist bl;
+  impl->objecter->linger_watch(
+    linger_op, op, ioc->snapc, ceph::real_clock::now(), bl,
+    Objecter::LingerOp::OpComp::create(
+      get_executor(),
+      [c = std::move(c), cookie](bs::error_code e, cb::list) mutable {
+	ca::dispatch(std::move(c), e, cookie);
+      }), nullptr);
+}
+
+void RADOS::watch(const Object& o, std::int64_t pool,
+		  std::optional<std::chrono::seconds> timeout, WatchCB&& cb,
+		  std::unique_ptr<WatchComp> c,
+		  std::optional<std::string_view> ns,
+		  std::optional<std::string_view> key) {
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  object_locator_t oloc;
+  oloc.pool = pool;
+  if (ns)
+    oloc.nspace = *ns;
+  if (key)
+    oloc.key = *key;
+
+  ObjectOperation op;
+
+  Objecter::LingerOp *linger_op = impl->objecter->linger_register(*oid, oloc, 0);
+  uint64_t cookie = linger_op->get_cookie();
+  linger_op->handle = std::move(cb);
+  op.watch(cookie, CEPH_OSD_WATCH_OP_WATCH, timeout.value_or(0s).count());
+  bufferlist bl;
+  impl->objecter->linger_watch(
+    linger_op, op, {}, ceph::real_clock::now(), bl,
+    Objecter::LingerOp::OpComp::create(
+      get_executor(),
+      [c = std::move(c), cookie](bs::error_code e, bufferlist) mutable {
+	ca::dispatch(std::move(c), e, cookie);
+      }), nullptr);
+}
+
+void RADOS::notify_ack(const Object& o,
+		       const IOContext& _ioc,
+		       uint64_t notify_id,
+		       uint64_t cookie,
+		       bufferlist&& bl,
+		       std::unique_ptr<SimpleOpComp> c)
+{
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  auto ioc = reinterpret_cast<const IOContextImpl*>(&_ioc.impl);
+
+  ObjectOperation op;
+  op.notify_ack(notify_id, cookie, bl);
+
+  impl->objecter->read(*oid, ioc->oloc, std::move(op), ioc->snap_seq,
+		       nullptr, ioc->extra_op_flags, std::move(c));
+}
+
+void RADOS::notify_ack(const Object& o,
+		       std::int64_t pool,
+		       uint64_t notify_id,
+		       uint64_t cookie,
+		       bufferlist&& bl,
+		       std::unique_ptr<SimpleOpComp> c,
+		       std::optional<std::string_view> ns,
+		       std::optional<std::string_view> key) {
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  object_locator_t oloc;
+  oloc.pool = pool;
+  if (ns)
+    oloc.nspace = *ns;
+  if (key)
+    oloc.key = *key;
+
+  ObjectOperation op;
+  op.notify_ack(notify_id, cookie, bl);
+  impl->objecter->read(*oid, oloc, std::move(op), CEPH_NOSNAP, nullptr, 0,
+		       std::move(c));
+}
+
+tl::expected<ceph::timespan, bs::error_code> RADOS::watch_check(uint64_t cookie)
+{
+  Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
+  return impl->objecter->linger_check(linger_op);
+}
+
+void RADOS::unwatch(uint64_t cookie, const IOContext& _ioc,
+		    std::unique_ptr<SimpleOpComp> c)
+{
+  auto ioc = reinterpret_cast<const IOContextImpl*>(&_ioc.impl);
+
+  Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
+
+  ObjectOperation op;
+  op.watch(cookie, CEPH_OSD_WATCH_OP_UNWATCH);
+  impl->objecter->mutate(linger_op->target.base_oid, ioc->oloc, std::move(op),
+			 ioc->snapc, ceph::real_clock::now(), ioc->extra_op_flags,
+			 Objecter::Op::OpComp::create(
+			   get_executor(),
+			   [objecter = impl->objecter,
+			    linger_op, c = std::move(c)]
+			   (bs::error_code ec) mutable {
+			     objecter->linger_cancel(linger_op);
+			     ca::dispatch(std::move(c), ec);
+			   }));
+}
+
+void RADOS::unwatch(uint64_t cookie, std::int64_t pool,
+		    std::unique_ptr<SimpleOpComp> c,
+		    std::optional<std::string_view> ns,
+		    std::optional<std::string_view> key)
+{
+  object_locator_t oloc;
+  oloc.pool = pool;
+  if (ns)
+    oloc.nspace = *ns;
+  if (key)
+    oloc.key = *key;
+
+  Objecter::LingerOp *linger_op = reinterpret_cast<Objecter::LingerOp*>(cookie);
+
+  ObjectOperation op;
+  op.watch(cookie, CEPH_OSD_WATCH_OP_UNWATCH);
+  impl->objecter->mutate(linger_op->target.base_oid, oloc, std::move(op),
+			 {}, ceph::real_clock::now(), 0,
+			 Objecter::Op::OpComp::create(
+			   get_executor(),
+			   [objecter = impl->objecter,
+			    linger_op, c = std::move(c)]
+			   (bs::error_code ec) mutable {
+			     objecter->linger_cancel(linger_op);
+			     ca::dispatch(std::move(c), ec);
+			   }));
+}
+
+void RADOS::flush_watch(std::unique_ptr<VoidOpComp> c)
+{
+  impl->objecter->linger_callback_flush([c = std::move(c)]() mutable {
+					  ca::post(std::move(c));
+					});
+}
+
+struct NotifyHandler : std::enable_shared_from_this<NotifyHandler> {
+  boost::asio::io_context& ioc;
+  boost::asio::io_context::strand strand;
+  Objecter* objecter;
+  Objecter::LingerOp* op;
+  std::unique_ptr<RADOS::NotifyComp> c;
+
+  bool acked = false;
+  bool finished = false;
+  bs::error_code res;
+  bufferlist rbl;
+
+  NotifyHandler(boost::asio::io_context& ioc,
+		Objecter* objecter,
+		Objecter::LingerOp* op,
+		std::unique_ptr<RADOS::NotifyComp> c)
+    : ioc(ioc), strand(ioc), objecter(objecter), op(op), c(std::move(c)) {}
+
+  // Use bind or a lambda to pass this in.
+  void handle_ack(bs::error_code ec,
+		  bufferlist&&) {
+    boost::asio::post(
+      strand,
+      [this, ec, p = shared_from_this()]() mutable {
+	acked = true;
+	maybe_cleanup(ec);
+      });
+  }
+
+  // Notify finish callback. It can actually own the object's storage.
+
+  void operator()(bs::error_code ec,
+		  bufferlist&& bl) {
+    boost::asio::post(
+      strand,
+      [this, ec, p = shared_from_this()]() mutable {
+	finished = true;
+	maybe_cleanup(ec);
+      });
+  }
+
+  // Should be called from strand.
+  void maybe_cleanup(bs::error_code ec) {
+    if (!res && ec)
+      res = ec;
+    if ((acked && finished) || res) {
+      objecter->linger_cancel(op);
+      ceph_assert(c);
+      ca::dispatch(std::move(c), res, std::move(rbl));
+    }
+  }
+};
+
+void RADOS::notify(const Object& o, const IOContext& _ioc, bufferlist&& bl,
+		   std::optional<std::chrono::milliseconds> timeout,
+		   std::unique_ptr<NotifyComp> c)
+{
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  auto ioc = reinterpret_cast<const IOContextImpl*>(&_ioc.impl);
+  auto linger_op = impl->objecter->linger_register(*oid, ioc->oloc,
+                                                   ioc->extra_op_flags);
+
+  auto cb = std::make_shared<NotifyHandler>(impl->ioctx, impl->objecter,
+                                            linger_op, std::move(c));
+  linger_op->on_notify_finish =
+    Objecter::LingerOp::OpComp::create(
+      get_executor(),
+      [cb](bs::error_code ec, ceph::bufferlist bl) mutable {
+	(*cb)(ec, std::move(bl));
+      });
+  ObjectOperation rd;
+  bufferlist inbl;
+  rd.notify(
+    linger_op->get_cookie(), 1,
+    timeout ? timeout->count() : impl->cct->_conf->client_notify_timeout,
+    bl, &inbl);
+
+  impl->objecter->linger_notify(
+    linger_op, rd, ioc->snap_seq, inbl,
+    Objecter::LingerOp::OpComp::create(
+      get_executor(),
+      [cb](bs::error_code ec, ceph::bufferlist bl) mutable {
+	cb->handle_ack(ec, std::move(bl));
+      }), nullptr);
+}
+
+void RADOS::notify(const Object& o, std::int64_t pool, bufferlist&& bl,
+		   std::optional<std::chrono::milliseconds> timeout,
+		   std::unique_ptr<NotifyComp> c,
+		   std::optional<std::string_view> ns,
+		   std::optional<std::string_view> key)
+{
+  auto oid = reinterpret_cast<const object_t*>(&o.impl);
+  object_locator_t oloc;
+  oloc.pool = pool;
+  if (ns)
+    oloc.nspace = *ns;
+  if (key)
+    oloc.key = *key;
+  auto linger_op = impl->objecter->linger_register(*oid, oloc, 0);
+
+  auto cb = std::make_shared<NotifyHandler>(impl->ioctx, impl->objecter,
+                                            linger_op, std::move(c));
+  linger_op->on_notify_finish =
+    Objecter::LingerOp::OpComp::create(
+      get_executor(),
+      [cb](bs::error_code ec, ceph::bufferlist&& bl) mutable {
+	(*cb)(ec, std::move(bl));
+      });
+  ObjectOperation rd;
+  bufferlist inbl;
+  rd.notify(
+    linger_op->get_cookie(), 1,
+    timeout ? timeout->count() : impl->cct->_conf->client_notify_timeout,
+    bl, &inbl);
+
+  impl->objecter->linger_notify(
+    linger_op, rd, CEPH_NOSNAP, inbl,
+    Objecter::LingerOp::OpComp::create(
+      get_executor(),
+      [cb](bs::error_code ec, bufferlist&& bl) mutable {
+	cb->handle_ack(ec, std::move(bl));
+      }), nullptr);
+}
+
+// Enumeration
+
+Cursor::Cursor() {
+  static_assert(impl_size >= sizeof(hobject_t));
+  new (&impl) hobject_t();
+};
+
+Cursor::Cursor(end_magic_t) {
+  static_assert(impl_size >= sizeof(hobject_t));
+  new (&impl) hobject_t(hobject_t::get_max());
+}
+
+Cursor::Cursor(void* p) {
+  static_assert(impl_size >= sizeof(hobject_t));
+  new (&impl) hobject_t(std::move(*reinterpret_cast<hobject_t*>(p)));
+}
+
+Cursor Cursor::begin() {
+  Cursor e;
+  return e;
+}
+
+Cursor Cursor::end() {
+  Cursor e(end_magic_t{});
+  return e;
+}
+
+Cursor::Cursor(const Cursor& rhs) {
+  static_assert(impl_size >= sizeof(hobject_t));
+  new (&impl) hobject_t(*reinterpret_cast<const hobject_t*>(&rhs.impl));
+}
+
+Cursor& Cursor::operator =(const Cursor& rhs) {
+  static_assert(impl_size >= sizeof(hobject_t));
+  reinterpret_cast<hobject_t*>(&impl)->~hobject_t();
+  new (&impl) hobject_t(*reinterpret_cast<const hobject_t*>(&rhs.impl));
+  return *this;
+}
+
+Cursor::Cursor(Cursor&& rhs) {
+  static_assert(impl_size >= sizeof(hobject_t));
+  new (&impl) hobject_t(std::move(*reinterpret_cast<hobject_t*>(&rhs.impl)));
+}
+
+Cursor& Cursor::operator =(Cursor&& rhs) {
+  static_assert(impl_size >= sizeof(hobject_t));
+  reinterpret_cast<hobject_t*>(&impl)->~hobject_t();
+  new (&impl) hobject_t(std::move(*reinterpret_cast<hobject_t*>(&rhs.impl)));
+  return *this;
+}
+Cursor::~Cursor() {
+  reinterpret_cast<hobject_t*>(&impl)->~hobject_t();
+}
+
+bool operator ==(const Cursor& lhs, const Cursor& rhs) {
+  return (*reinterpret_cast<const hobject_t*>(&lhs.impl) ==
+	  *reinterpret_cast<const hobject_t*>(&rhs.impl));
+}
+
+bool operator !=(const Cursor& lhs, const Cursor& rhs) {
+  return (*reinterpret_cast<const hobject_t*>(&lhs.impl) !=
+	  *reinterpret_cast<const hobject_t*>(&rhs.impl));
+}
+
+bool operator <(const Cursor& lhs, const Cursor& rhs) {
+  return (*reinterpret_cast<const hobject_t*>(&lhs.impl) <
+	  *reinterpret_cast<const hobject_t*>(&rhs.impl));
+}
+
+bool operator <=(const Cursor& lhs, const Cursor& rhs) {
+  return (*reinterpret_cast<const hobject_t*>(&lhs.impl) <=
+	  *reinterpret_cast<const hobject_t*>(&rhs.impl));
+}
+
+bool operator >=(const Cursor& lhs, const Cursor& rhs) {
+  return (*reinterpret_cast<const hobject_t*>(&lhs.impl) >=
+	  *reinterpret_cast<const hobject_t*>(&rhs.impl));
+}
+
+bool operator >(const Cursor& lhs, const Cursor& rhs) {
+  return (*reinterpret_cast<const hobject_t*>(&lhs.impl) >
+	  *reinterpret_cast<const hobject_t*>(&rhs.impl));
+}
+
+std::string Cursor::to_str() const {
+  using namespace std::literals;
+  auto& h = *reinterpret_cast<const hobject_t*>(&impl);
+
+  return h.is_max() ? "MAX"s : h.to_str();
+}
+
+std::optional<Cursor>
+Cursor::from_str(const std::string& s) {
+  Cursor e;
+  auto& h = *reinterpret_cast<hobject_t*>(&e.impl);
+  if (!h.parse(s))
+    return std::nullopt;
+
+  return e;
+}
+
+void RADOS::enumerate_objects(const IOContext& _ioc,
+			      const Cursor& begin,
+			      const Cursor& end,
+			      const std::uint32_t max,
+			      const bufferlist& filter,
+			      std::unique_ptr<EnumerateComp> c) {
+  auto ioc = reinterpret_cast<const IOContextImpl*>(&_ioc.impl);
+
+  impl->objecter->enumerate_objects<Entry>(
+    ioc->oloc.pool,
+    ioc->oloc.nspace,
+    *reinterpret_cast<const hobject_t*>(&begin.impl),
+    *reinterpret_cast<const hobject_t*>(&end.impl),
+    max,
+    filter,
+    [c = std::move(c)]
+    (bs::error_code ec, std::vector<Entry>&& v,
+     hobject_t&& n) mutable {
+      ca::dispatch(std::move(c), ec, std::move(v),
+		   Cursor(static_cast<void*>(&n)));
+    });
+}
+
+void RADOS::enumerate_objects(std::int64_t pool,
+			      const Cursor& begin,
+			      const Cursor& end,
+			      const std::uint32_t max,
+			      const bufferlist& filter,
+			      std::unique_ptr<EnumerateComp> c,
+			      std::optional<std::string_view> ns,
+			      std::optional<std::string_view> key) {
+  impl->objecter->enumerate_objects<Entry>(
+    pool,
+    ns ? *ns : std::string_view{},
+    *reinterpret_cast<const hobject_t*>(&begin.impl),
+    *reinterpret_cast<const hobject_t*>(&end.impl),
+    max,
+    filter,
+    [c = std::move(c)]
+    (bs::error_code ec, std::vector<Entry>&& v,
+     hobject_t&& n) mutable {
+      ca::dispatch(std::move(c), ec, std::move(v),
+		   Cursor(static_cast<void*>(&n)));
+    });
+}
+
+
+void RADOS::osd_command(int osd, std::vector<std::string>&& cmd,
+			ceph::bufferlist&& in, std::unique_ptr<CommandComp> c) {
+  impl->objecter->osd_command(osd, std::move(cmd), std::move(in), nullptr,
+			      [c = std::move(c)]
+			      (bs::error_code ec,
+			       std::string&& s,
+			       ceph::bufferlist&& b) mutable {
+				ca::dispatch(std::move(c), ec,
+					     std::move(s),
+					     std::move(b));
+			      });
+}
+void RADOS::pg_command(PG pg, std::vector<std::string>&& cmd,
+		       ceph::bufferlist&& in, std::unique_ptr<CommandComp> c) {
+  impl->objecter->pg_command(pg_t{pg.seed, pg.pool}, std::move(cmd), std::move(in), nullptr,
+			     [c = std::move(c)]
+			     (bs::error_code ec,
+			      std::string&& s,
+			      ceph::bufferlist&& b) mutable {
+			       ca::dispatch(std::move(c), ec,
+					    std::move(s),
+					    std::move(b));
+			     });
+}
+
+void RADOS::enable_application(std::string_view pool, std::string_view app_name,
+			       bool force, std::unique_ptr<SimpleOpComp> c) {
+  // pre-Luminous clusters will return -EINVAL and application won't be
+  // preserved until Luminous is configured as minimum version.
+  if (!impl->get_required_monitor_features().contains_all(
+	ceph::features::mon::FEATURE_LUMINOUS)) {
+    ca::post(std::move(c), ceph::to_error_code(-EOPNOTSUPP));
+  } else {
+    impl->monclient.start_mon_command(
+      { fmt::format("{{ \"prefix\": \"osd pool application enable\","
+		    "\"pool\": \"{}\", \"app\": \"{}\"{}}}",
+		    pool, app_name,
+		    force ? " ,\"yes_i_really_mean_it\": true" : "")},
+      {}, [c = std::move(c)](bs::error_code e,
+			     std::string, cb::list) mutable {
+	    ca::post(std::move(c), e);
+	  });
+  }
+}
+
+void RADOS::blocklist_add(std::string_view client_address,
+                          std::optional<std::chrono::seconds> expire,
+                          std::unique_ptr<SimpleOpComp> c) {
+  auto expire_arg = (expire ?
+    fmt::format(", \"expire\": \"{}.0\"", expire->count()) : std::string{});
+  impl->monclient.start_mon_command(
+    { fmt::format("{{"
+                  "\"prefix\": \"osd blocklist\", "
+                  "\"blocklistop\": \"add\", "
+                  "\"addr\": \"{}\"{}}}",
+                  client_address, expire_arg) },
+    {},
+    [this, client_address = std::string(client_address), expire_arg,
+     c = std::move(c)](bs::error_code ec, std::string, cb::list) mutable {
+      if (ec != bs::errc::invalid_argument) {
+        ca::post(std::move(c), ec);
+        return;
+      }
+
+      // retry using the legacy command
+      impl->monclient.start_mon_command(
+        { fmt::format("{{"
+                      "\"prefix\": \"osd blacklist\", "
+                      "\"blacklistop\": \"add\", "
+                      "\"addr\": \"{}\"{}}}",
+                      client_address, expire_arg) },
+        {},
+        [c = std::move(c)](bs::error_code ec, std::string, cb::list) mutable {
+          ca::post(std::move(c), ec);
+        });
+    });
+}
+
+void RADOS::wait_for_latest_osd_map(std::unique_ptr<SimpleOpComp> c) {
+  impl->objecter->wait_for_latest_osdmap(std::move(c));
+}
+
+void RADOS::mon_command(std::vector<std::string> command,
+			const cb::list& bl,
+			std::string* outs, cb::list* outbl,
+			std::unique_ptr<SimpleOpComp> c) {
+
+  impl->monclient.start_mon_command(
+    command, bl,
+    [c = std::move(c), outs, outbl](bs::error_code e,
+				    std::string s, cb::list bl) mutable {
+      if (outs)
+	*outs = std::move(s);
+      if (outbl)
+	*outbl = std::move(bl);
+      ca::post(std::move(c), e);
+    });
+}
+
+uint64_t RADOS::instance_id() const {
+  return impl->get_instance_id();
+}
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wnon-virtual-dtor"
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wnon-virtual-dtor"
+class category : public ceph::converting_category {
+public:
+  category() {}
+  const char* name() const noexcept override;
+  const char* message(int ev, char*, std::size_t) const noexcept override;
+  std::string message(int ev) const override;
+  bs::error_condition default_error_condition(int ev) const noexcept
+    override;
+  bool equivalent(int ev, const bs::error_condition& c) const
+    noexcept override;
+  using ceph::converting_category::equivalent;
+  int from_code(int ev) const noexcept override;
+};
+#pragma GCC diagnostic pop
+#pragma clang diagnostic pop
+
+const char* category::name() const noexcept {
+  return "RADOS";
+}
+
+const char* category::message(int ev, char*,
+			      std::size_t) const noexcept {
+  if (ev == 0)
+    return "No error";
+
+  switch (static_cast<errc>(ev)) {
+  case errc::pool_dne:
+    return "Pool does not exist";
+
+  case errc::invalid_snapcontext:
+    return "Invalid snapcontext";
+  }
+
+  return "Unknown error";
+}
+
+std::string category::message(int ev) const {
+  return message(ev, nullptr, 0);
+}
+
+bs::error_condition category::default_error_condition(int ev) const noexcept {
+  switch (static_cast<errc>(ev)) {
+  case errc::pool_dne:
+    return ceph::errc::does_not_exist;
+  case errc::invalid_snapcontext:
+    return bs::errc::invalid_argument;
+  }
+
+  return { ev, *this };
+}
+
+bool category::equivalent(int ev, const bs::error_condition& c) const noexcept {
+  if (static_cast<errc>(ev) == errc::pool_dne) {
+    if (c == bs::errc::no_such_file_or_directory) {
+      return true;
+    }
+  }
+
+  return default_error_condition(ev) == c;
+}
+
+int category::from_code(int ev) const noexcept {
+  switch (static_cast<errc>(ev)) {
+  case errc::pool_dne:
+    return -ENOENT;
+  case errc::invalid_snapcontext:
+    return -EINVAL;
+  }
+  return -EDOM;
+}
+
+const bs::error_category& error_category() noexcept {
+  static const class category c;
+  return c;
+}
+
+CephContext* RADOS::cct() {
+  return impl->cct.get();
+}
+}
+
+namespace std {
+size_t hash<neorados::Object>::operator ()(
+  const neorados::Object& r) const {
+  static constexpr const hash<object_t> H;
+  return H(*reinterpret_cast<const object_t*>(&r.impl));
+}
+
+size_t hash<neorados::IOContext>::operator ()(
+  const neorados::IOContext& r) const {
+  static constexpr const hash<int64_t> H;
+  static constexpr const hash<std::string> G;
+  const auto l = reinterpret_cast<const neorados::IOContextImpl*>(&r.impl);
+  return H(l->oloc.pool) ^ (G(l->oloc.nspace) << 1) ^ (G(l->oloc.key) << 2);
+}
+}
diff --git a/src/neorados/RADOSImpl.cc b/src/neorados/RADOSImpl.cc
new file mode 100644
index 000000000..6c9c210a8
--- /dev/null
+++ b/src/neorados/RADOSImpl.cc
@@ -0,0 +1,121 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2012 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+
+#include <boost/system/system_error.hpp>
+
+#include "common/common_init.h"
+
+#include "global/global_init.h"
+
+#include "RADOSImpl.h"
+
+namespace neorados {
+namespace detail {
+
+RADOS::RADOS(boost::asio::io_context& ioctx,
+	     boost::intrusive_ptr<CephContext> cct)
+  : Dispatcher(cct.get()),
+    ioctx(ioctx),
+    cct(cct),
+    monclient(cct.get(), ioctx),
+    mgrclient(cct.get(), nullptr, &monclient.monmap) {
+  auto err = monclient.build_initial_monmap();
+  if (err < 0)
+    throw std::system_error(ceph::to_error_code(err));
+
+  messenger.reset(Messenger::create_client_messenger(cct.get(), "radosclient"));
+  if (!messenger)
+    throw std::bad_alloc();
+
+  // Require OSDREPLYMUX feature. This means we will fail to talk to
+  // old servers. This is necessary because otherwise we won't know
+  // how to decompose the reply data into its constituent pieces.
+  messenger->set_default_policy(
+    Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX));
+
+  objecter = std::make_unique<Objecter>(cct.get(), messenger.get(), &monclient, ioctx);
+
+  objecter->set_balanced_budget();
+  monclient.set_messenger(messenger.get());
+  mgrclient.set_messenger(messenger.get());
+  objecter->init();
+  messenger->add_dispatcher_head(&mgrclient);
+  messenger->add_dispatcher_tail(objecter.get());
+  messenger->start();
+  monclient.set_want_keys(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_OSD | CEPH_ENTITY_TYPE_MGR);
+  err = monclient.init();
+  if (err) {
+    throw boost::system::system_error(ceph::to_error_code(err));
+  }
+  err = monclient.authenticate(std::chrono::duration<double>(cct->_conf.get_val<std::chrono::seconds>("client_mount_timeout")).count());
+  if (err) {
+    throw boost::system::system_error(ceph::to_error_code(err));
+  }
+  messenger->set_myname(entity_name_t::CLIENT(monclient.get_global_id()));
+  // Detect older cluster, put mgrclient into compatible mode
+  mgrclient.set_mgr_optional(
+      !get_required_monitor_features().contains_all(
+        ceph::features::mon::FEATURE_LUMINOUS));
+
+  // MgrClient needs this (it doesn't have MonClient reference itself)
+  monclient.sub_want("mgrmap", 0, 0);
+  monclient.renew_subs();
+
+  mgrclient.init();
+  objecter->set_client_incarnation(0);
+  objecter->start();
+
+  messenger->add_dispatcher_tail(this);
+
+  std::unique_lock l(lock);
+  instance_id = monclient.get_global_id();
+}
+
+RADOS::~RADOS() {
+  if (objecter && objecter->initialized) {
+    objecter->shutdown();
+  }
+
+  mgrclient.shutdown();
+  monclient.shutdown();
+
+  if (messenger) {
+    messenger->shutdown();
+    messenger->wait();
+  }
+}
+
+bool RADOS::ms_dispatch(Message *m)
+{
+  switch (m->get_type()) {
+  // OSD
+  case CEPH_MSG_OSD_MAP:
+    m->put();
+    return true;
+  }
+  return false;
+}
+
+void RADOS::ms_handle_connect(Connection *con) {}
+bool RADOS::ms_handle_reset(Connection *con) {
+  return false;
+}
+void RADOS::ms_handle_remote_reset(Connection *con) {}
+bool RADOS::ms_handle_refused(Connection *con) {
+  return false;
+}
+
+} // namespace detail
+} // namespace neorados
diff --git a/src/neorados/RADOSImpl.h b/src/neorados/RADOSImpl.h
new file mode 100644
index 000000000..d45ca9481
--- /dev/null
+++ b/src/neorados/RADOSImpl.h
@@ -0,0 +1,135 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2004-2012 Sage Weil <sage@newdream.net>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+#ifndef CEPH_NEORADOS_RADOSIMPL_H
+#define CEPH_NEORADOS_RADOSIMPL_H
+
+#include <functional>
+#include <memory>
+#include <string>
+
+#include <boost/asio.hpp>
+#include <boost/intrusive_ptr.hpp>
+
+#include "common/ceph_context.h"
+#include "common/ceph_mutex.h"
+
+#include "librados/RadosClient.h"
+
+#include "mon/MonClient.h"
+
+#include "mgr/MgrClient.h"
+
+#include "osdc/Objecter.h"
+
+namespace neorados {
+
+class RADOS;
+
+namespace detail {
+
+class NeoClient;
+
+class RADOS : public Dispatcher
+{
+  friend ::neorados::RADOS;
+  friend NeoClient;
+
+  boost::asio::io_context& ioctx;
+  boost::intrusive_ptr<CephContext> cct;
+
+  ceph::mutex lock = ceph::make_mutex("RADOS_unleashed::_::RADOSImpl");
+  int instance_id = -1;
+
+  std::unique_ptr<Messenger> messenger;
+
+  MonClient monclient;
+  MgrClient mgrclient;
+
+  std::unique_ptr<Objecter> objecter;
+
+public:
+
+  RADOS(boost::asio::io_context& ioctx, boost::intrusive_ptr<CephContext> cct);
+  ~RADOS();
+  bool ms_dispatch(Message *m) override;
+  void ms_handle_connect(Connection *con) override;
+  bool ms_handle_reset(Connection *con) override;
+  void ms_handle_remote_reset(Connection *con) override;
+  bool ms_handle_refused(Connection *con) override;
+  mon_feature_t get_required_monitor_features() const {
+    return monclient.with_monmap(std::mem_fn(&MonMap::get_required_features));
+  }
+};
+
+class Client {
+public:
+  Client(boost::asio::io_context& ioctx,
+         boost::intrusive_ptr<CephContext> cct,
+         MonClient& monclient, Objecter* objecter)
+    : ioctx(ioctx), cct(cct), monclient(monclient), objecter(objecter) {
+  }
+  virtual ~Client() {}
+
+  Client(const Client&) = delete;
+  Client& operator=(const Client&) = delete;
+
+  boost::asio::io_context& ioctx;
+
+  boost::intrusive_ptr<CephContext> cct;
+  MonClient& monclient;
+  Objecter* objecter;
+
+  mon_feature_t get_required_monitor_features() const {
+    return monclient.with_monmap(std::mem_fn(&MonMap::get_required_features));
+  }
+
+  virtual int get_instance_id() const = 0;
+};
+
+class NeoClient : public Client {
+public:
+  NeoClient(std::unique_ptr<RADOS>&& rados)
+    : Client(rados->ioctx, rados->cct, rados->monclient,
+             rados->objecter.get()),
+      rados(std::move(rados)) {
+  }
+
+  int get_instance_id() const override {
+    return rados->instance_id;
+  }
+
+private:
+  std::unique_ptr<RADOS> rados;
+};
+
+class RadosClient : public Client {
+public:
+  RadosClient(librados::RadosClient* rados_client)
+    : Client(rados_client->poolctx, {rados_client->cct},
+             rados_client->monclient, rados_client->objecter),
+      rados_client(rados_client) {
+  }
+
+  int get_instance_id() const override {
+    return rados_client->instance_id;
+  }
+
+public:
+  librados::RadosClient* rados_client;
+};
+
+} // namespace detail
+} // namespace neorados
+
+#endif
diff --git a/src/neorados/cls/fifo.cc b/src/neorados/cls/fifo.cc
new file mode 100644
index 000000000..fa99275b2
--- /dev/null
+++ b/src/neorados/cls/fifo.cc
@@ -0,0 +1,385 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2020 Red Hat <contact@redhat.com>
+ * Author: Adam C. Emerson
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include <cstdint>
+#include <numeric>
+#include <optional>
+#include <string_view>
+
+#undef FMT_HEADER_ONLY
+#define FMT_HEADER_ONLY 1
+#include <fmt/format.h>
+
+#include <boost/system/error_code.hpp>
+
+#include "include/neorados/RADOS.hpp"
+
+#include "include/buffer.h"
+
+#include "common/random_string.h"
+
+#include "cls/fifo/cls_fifo_types.h"
+#include "cls/fifo/cls_fifo_ops.h"
+
+#include "fifo.h"
+
+namespace neorados::cls::fifo {
+namespace bs = boost::system;
+namespace cb = ceph::buffer;
+namespace fifo = rados::cls::fifo;
+
+void create_meta(WriteOp& op, std::string_view id,
+		 std::optional<fifo::objv> objv,
+		 std::optional<std::string_view> oid_prefix,
+		 bool exclusive,
+		 std::uint64_t max_part_size,
+		 std::uint64_t max_entry_size)
+{
+  fifo::op::create_meta cm;
+
+  cm.id = id;
+  cm.version = objv;
+  cm.oid_prefix = oid_prefix;
+  cm.max_part_size = max_part_size;
+  cm.max_entry_size = max_entry_size;
+  cm.exclusive = exclusive;
+
+  cb::list in;
+  encode(cm, in);
+  op.exec(fifo::op::CLASS, fifo::op::CREATE_META, in);
+}
+
+void get_meta(ReadOp& op, std::optional<fifo::objv> objv,
+	      bs::error_code* ec_out, fifo::info* info,
+	      std::uint32_t* part_header_size,
+	      std::uint32_t* part_entry_overhead)
+{
+  fifo::op::get_meta gm;
+  gm.version = objv;
+  cb::list in;
+  encode(gm, in);
+  op.exec(fifo::op::CLASS, fifo::op::GET_META, in,
+	  [ec_out, info, part_header_size,
+	   part_entry_overhead](bs::error_code ec, const cb::list& bl) {
+	    fifo::op::get_meta_reply reply;
+	    if (!ec) try {
+		auto iter = bl.cbegin();
+		decode(reply, iter);
+	      } catch (const cb::error& err) {
+		ec = err.code();
+	      }
+	    if (ec_out) *ec_out = ec;
+	    if (info) *info = std::move(reply.info);
+	    if (part_header_size) *part_header_size = reply.part_header_size;
+	    if (part_entry_overhead)
+		*part_entry_overhead = reply.part_entry_overhead;
+	  });
+};
+
+void update_meta(WriteOp& op, const fifo::objv& objv,
+		 const fifo::update& update)
+{
+  fifo::op::update_meta um;
+
+  um.version = objv;
+  um.tail_part_num = update.tail_part_num();
+  um.head_part_num = update.head_part_num();
+  um.min_push_part_num = update.min_push_part_num();
+  um.max_push_part_num = update.max_push_part_num();
+  um.journal_entries_add = std::move(update).journal_entries_add();
+  um.journal_entries_rm = std::move(update).journal_entries_rm();
+
+  cb::list in;
+  encode(um, in);
+  op.exec(fifo::op::CLASS, fifo::op::UPDATE_META, in);
+}
+
+void part_init(WriteOp& op, std::string_view tag,
+	       fifo::data_params params)
+{
+  fifo::op::init_part ip;
+
+  ip.tag = tag;
+  ip.params = params;
+
+  cb::list in;
+  encode(ip, in);
+  op.exec(fifo::op::CLASS, fifo::op::INIT_PART, in);
+}
+
+void push_part(WriteOp& op, std::string_view tag,
+	       std::deque<cb::list> data_bufs,
+	       fu2::unique_function<void(bs::error_code, int)> f)
+{
+  fifo::op::push_part pp;
+
+  pp.tag = tag;
+  pp.data_bufs = data_bufs;
+  pp.total_len = 0;
+
+  for (const auto& bl : data_bufs)
+    pp.total_len += bl.length();
+
+  cb::list in;
+  encode(pp, in);
+  op.exec(fifo::op::CLASS, fifo::op::PUSH_PART, in,
+	  [f = std::move(f)](bs::error_code ec, int r, const cb::list&) mutable {
+	    std::move(f)(ec, r);
+	  });
+  op.returnvec();
+}
+
+void trim_part(WriteOp& op,
+	       std::optional<std::string_view> tag,
+	       std::uint64_t ofs, bool exclusive)
+{
+  fifo::op::trim_part tp;
+
+  tp.tag = tag;
+  tp.ofs = ofs;
+  tp.exclusive = exclusive;
+
+  bufferlist in;
+  encode(tp, in);
+  op.exec(fifo::op::CLASS, fifo::op::TRIM_PART, in);
+}
+
+void list_part(ReadOp& op,
+	       std::optional<string_view> tag,
+	       std::uint64_t ofs,
+	       std::uint64_t max_entries,
+	       bs::error_code* ec_out,
+	       std::vector<fifo::part_list_entry>* entries,
+	       bool* more,
+	       bool* full_part,
+	       std::string* ptag)
+{
+  fifo::op::list_part lp;
+
+  lp.tag = tag;
+  lp.ofs = ofs;
+  lp.max_entries = max_entries;
+
+  bufferlist in;
+  encode(lp, in);
+  op.exec(fifo::op::CLASS, fifo::op::LIST_PART, in,
+	  [entries, more, full_part, ptag, ec_out](bs::error_code ec,
+						   const cb::list& bl) {
+	    if (ec) {
+	      if (ec_out) *ec_out = ec;
+	      return;
+	    }
+
+	    fifo::op::list_part_reply reply;
+	    auto iter = bl.cbegin();
+	    try {
+	      decode(reply, iter);
+	    } catch (const cb::error& err) {
+	      if (ec_out) *ec_out = ec;
+	      return;
+	    }
+
+	    if (entries) *entries = std::move(reply.entries);
+	    if (more) *more = reply.more;
+	    if (full_part) *full_part = reply.full_part;
+	    if (ptag) *ptag = reply.tag;
+	  });
+}
+
+void get_part_info(ReadOp& op,
+		   bs::error_code* out_ec,
+		   fifo::part_header* header)
+{
+  fifo::op::get_part_info gpi;
+
+  bufferlist in;
+  encode(gpi, in);
+  op.exec(fifo::op::CLASS, fifo::op::GET_PART_INFO, in,
+	  [out_ec, header](bs::error_code ec, const cb::list& bl) {
+	    if (ec) {
+	      if (out_ec) *out_ec = ec;
+	    }
+	    fifo::op::get_part_info_reply reply;
+	    auto iter = bl.cbegin();
+	    try {
+	      decode(reply, iter);
+	    } catch (const cb::error& err) {
+	      if (out_ec) *out_ec = ec;
+	      return;
+	    }
+
+	    if (header) *header = std::move(reply.header);
+	  });
+}
+
+std::optional<marker> FIFO::to_marker(std::string_view s) {
+  marker m;
+  if (s.empty()) {
+    m.num = info.tail_part_num;
+    m.ofs = 0;
+    return m;
+  }
+
+  auto pos = s.find(':');
+  if (pos == string::npos) {
+    return std::nullopt;
+  }
+
+  auto num = s.substr(0, pos);
+  auto ofs = s.substr(pos + 1);
+
+  auto n = ceph::parse<decltype(m.num)>(num);
+  if (!n) {
+    return std::nullopt;
+  }
+  m.num = *n;
+  auto o = ceph::parse<decltype(m.ofs)>(ofs);
+  if (!o) {
+    return std::nullopt;
+  }
+  m.ofs = *o;
+  return m;
+}
+
+bs::error_code FIFO::apply_update(fifo::info* info,
+				  const fifo::objv& objv,
+				  const fifo::update& update) {
+  std::unique_lock l(m);
+  auto err = info->apply_update(update);
+  if (objv != info->version) {
+    ldout(r->cct(), 0) << __func__ << "(): Raced locally!" << dendl;
+    return errc::raced;
+  }
+  if (err) {
+    ldout(r->cct(), 0) << __func__ << "(): ERROR: " << err << dendl;
+    return errc::update_failed;
+  }
+
+  ++info->version.ver;
+
+  return {};
+}
+
+std::string FIFO::generate_tag() const
+{
+  static constexpr auto HEADER_TAG_SIZE = 16;
+  return gen_rand_alphanumeric_plain(r->cct(), HEADER_TAG_SIZE);
+}
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wnon-virtual-dtor"
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wnon-virtual-dtor"
+class error_category : public ceph::converting_category {
+public:
+  error_category(){}
+  const char* name() const noexcept override;
+  const char* message(int ev, char*, std::size_t) const noexcept override;
+  std::string message(int ev) const override;
+  bs::error_condition default_error_condition(int ev) const noexcept
+    override;
+  bool equivalent(int ev, const bs::error_condition& c) const
+    noexcept override;
+  using ceph::converting_category::equivalent;
+  int from_code(int ev) const noexcept override;
+};
+#pragma GCC diagnostic pop
+#pragma clang diagnostic pop
+
+const char* error_category::name() const noexcept {
+  return "FIFO";
+}
+
+const char* error_category::message(int ev, char*, std::size_t) const noexcept {
+  if (ev == 0)
+    return "No error";
+
+  switch (static_cast<errc>(ev)) {
+  case errc::raced:
+    return "Retry-race count exceeded";
+
+  case errc::inconsistency:
+    return "Inconsistent result! New head before old head";
+
+  case errc::entry_too_large:
+    return "Pushed entry too large";
+
+  case errc::invalid_marker:
+    return "Invalid marker string";
+
+  case errc::update_failed:
+    return "Update failed";
+  }
+
+  return "Unknown error";
+}
+
+std::string error_category::message(int ev) const {
+  return message(ev, nullptr, 0);
+}
+
+bs::error_condition
+error_category::default_error_condition(int ev) const noexcept {
+  switch (static_cast<errc>(ev)) {
+  case errc::raced:
+    return bs::errc::operation_canceled;
+
+  case errc::inconsistency:
+    return bs::errc::io_error;
+
+  case errc::entry_too_large:
+    return bs::errc::value_too_large;
+
+  case errc::invalid_marker:
+    return bs::errc::invalid_argument;
+
+  case errc::update_failed:
+    return bs::errc::invalid_argument;
+  }
+
+  return { ev, *this };
+}
+
+bool error_category::equivalent(int ev, const bs::error_condition& c) const noexcept {
+  return default_error_condition(ev) == c;
+}
+
+int error_category::from_code(int ev) const noexcept {
+  switch (static_cast<errc>(ev)) {
+  case errc::raced:
+    return -ECANCELED;
+
+  case errc::inconsistency:
+    return -EIO;
+
+  case errc::entry_too_large:
+    return -E2BIG;
+
+  case errc::invalid_marker:
+    return -EINVAL;
+
+  case errc::update_failed:
+    return -EINVAL;
+
+  }
+  return -EDOM;
+}
+
+const bs::error_category& error_category() noexcept {
+  static const class error_category c;
+  return c;
+}
+
+}
diff --git a/src/neorados/cls/fifo.h b/src/neorados/cls/fifo.h
new file mode 100644
index 000000000..05865dcca
--- /dev/null
+++ b/src/neorados/cls/fifo.h
@@ -0,0 +1,1754 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2020 Red Hat <contact@redhat.com>
+ * Author: Adam C. Emerson
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef CEPH_NEORADOS_CLS_FIFIO_H
+#define CEPH_NEORADOS_CLS_FIFIO_H
+
+#include <cstdint>
+#include <deque>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string_view>
+#include <vector>
+
+#include <boost/asio.hpp>
+#include <boost/system/error_code.hpp>
+
+#undef FMT_HEADER_ONLY
+#define FMT_HEADER_ONLY 1
+#include <fmt/format.h>
+
+#include "include/neorados/RADOS.hpp"
+#include "include/buffer.h"
+
+#include "common/allocate_unique.h"
+#include "common/async/bind_handler.h"
+#include "common/async/bind_like.h"
+#include "common/async/completion.h"
+#include "common/async/forward_handler.h"
+
+#include "common/dout.h"
+
+#include "cls/fifo/cls_fifo_types.h"
+#include "cls/fifo/cls_fifo_ops.h"
+
+namespace neorados::cls::fifo {
+namespace ba = boost::asio;
+namespace bs = boost::system;
+namespace ca = ceph::async;
+namespace cb = ceph::buffer;
+namespace fifo = rados::cls::fifo;
+
+inline constexpr auto dout_subsys = ceph_subsys_rados;
+inline constexpr std::uint64_t default_max_part_size = 4 * 1024 * 1024;
+inline constexpr std::uint64_t default_max_entry_size = 32 * 1024;
+inline constexpr auto MAX_RACE_RETRIES = 10;
+
+
+const boost::system::error_category& error_category() noexcept;
+
+enum class errc {
+  raced = 1,
+  inconsistency,
+  entry_too_large,
+  invalid_marker,
+  update_failed
+};
+}
+
+namespace boost::system {
+template<>
+struct is_error_code_enum<::neorados::cls::fifo::errc> {
+  static const bool value = true;
+};
+template<>
+struct is_error_condition_enum<::neorados::cls::fifo::errc> {
+  static const bool value = false;
+};
+}
+
+namespace neorados::cls::fifo {
+//  explicit conversion:
+inline bs::error_code make_error_code(errc e) noexcept {
+  return { static_cast<int>(e), error_category() };
+}
+
+inline bs::error_code make_error_category(errc e) noexcept {
+  return { static_cast<int>(e), error_category() };
+}
+
+void create_meta(WriteOp& op, std::string_view id,
+		 std::optional<fifo::objv> objv,
+		 std::optional<std::string_view> oid_prefix,
+		 bool exclusive = false,
+		 std::uint64_t max_part_size = default_max_part_size,
+		 std::uint64_t max_entry_size = default_max_entry_size);
+void get_meta(ReadOp& op, std::optional<fifo::objv> objv,
+	      bs::error_code* ec_out, fifo::info* info,
+	      std::uint32_t* part_header_size,
+	      std::uint32_t* part_entry_overhead);
+
+void update_meta(WriteOp& op, const fifo::objv& objv,
+		 const fifo::update& desc);
+
+void part_init(WriteOp& op, std::string_view tag,
+	       fifo::data_params params);
+
+void push_part(WriteOp& op, std::string_view tag,
+	       std::deque<cb::list> data_bufs,
+	       fu2::unique_function<void(bs::error_code, int)>);
+void trim_part(WriteOp& op, std::optional<std::string_view> tag,
+	       std::uint64_t ofs,
+	       bool exclusive);
+void list_part(ReadOp& op,
+	       std::optional<std::string_view> tag,
+	       std::uint64_t ofs,
+	       std::uint64_t max_entries,
+	       bs::error_code* ec_out,
+	       std::vector<fifo::part_list_entry>* entries,
+	       bool* more,
+	       bool* full_part,
+	       std::string* ptag);
+void get_part_info(ReadOp& op,
+		   bs::error_code* out_ec,
+		   fifo::part_header* header);
+
+struct marker {
+  std::int64_t num = 0;
+  std::uint64_t ofs = 0;
+
+  marker() = default;
+  marker(std::int64_t num, std::uint64_t ofs) : num(num), ofs(ofs) {}
+  static marker max() {
+    return { std::numeric_limits<decltype(num)>::max(),
+	     std::numeric_limits<decltype(ofs)>::max() };
+  }
+
+  std::string to_string() {
+    return fmt::format("{:0>20}:{:0>20}", num, ofs);
+  }
+};
+
+struct list_entry {
+  cb::list data;
+  std::string marker;
+  ceph::real_time mtime;
+};
+
+using part_info = fifo::part_header;
+
+namespace detail {
+template<typename Handler>
+class JournalProcessor;
+}
+
+/// Completions, Handlers, and CompletionTokens
+/// ===========================================
+///
+/// This class is based on Boost.Asio. For information, see
+/// https://www.boost.org/doc/libs/1_74_0/doc/html/boost_asio.html
+///
+/// As summary, Asio's design is that of functions taking completion
+/// handlers. Every handler has a signature, like
+/// (boost::system::error_code, std::string). The completion handler
+/// receives the result of the function, and the signature is the type
+/// of that result.
+///
+/// The completion handler is specified with a CompletionToken. The
+/// CompletionToken is any type that has a specialization of
+/// async_complete and async_result. See
+/// https://www.boost.org/doc/libs/1_74_0/doc/html/boost_asio/reference/async_completion.html
+/// and https://www.boost.org/doc/libs/1_74_0/doc/html/boost_asio/reference/async_result.html
+///
+/// The return type of a function taking a CompletionToken is
+/// async_result<CompletionToken, Signature>::return_type.
+///
+/// Functions
+/// ---------
+///
+/// The default implementations treat whatever value is described as a
+/// function, whose parameters correspond to the signature, and calls
+/// it upon completion.
+///
+/// EXAMPLE:
+/// Let f be an asynchronous function whose signature is (bs::error_code, int)
+/// Let g be an asynchronous function whose signature is
+/// (bs::error_code, int, std::string).
+///
+///
+///    f([](bs::error_code ec, int i) { ... });
+///    g([](bs::error_code ec, int i, std::string s) { ... });
+///
+/// Will schedule asynchronous tasks, and the provided lambdas will be
+/// called on completion. In this case, f and g return void.
+///
+/// There are other specializations. Commonly used ones are.
+///
+/// Futures
+/// -------
+///
+/// A CompletionToken of boost::asio::use_future will complete with a
+/// promise whose type matches (minus any initial error_code) the
+/// function's signature. The corresponding future is returned. If the
+/// error_code of the result is non-zero, the future is set with an
+/// exception of type boost::asio::system_error.
+///
+/// See https://www.boost.org/doc/libs/1_74_0/doc/html/boost_asio/reference/use_future_t.html
+///
+/// EXAMPLE:
+///
+/// std::future<int> = f(ba::use_future);
+/// std::future<std::tuple<int, std::string> = g(ba::use_future).
+///
+/// Coroutines
+/// ----------
+///
+/// A CompletionToken of type spawn::yield_context suspends execution
+/// of the current coroutine until completion of the operation. See
+/// src/spawn/README.md
+/// https://www.boost.org/doc/libs/1_74_0/doc/html/boost_asio/reference/spawn.html and
+/// https://www.boost.org/doc/libs/1_74_0/doc/html/boost_asio/reference/yield_context.html
+///
+/// Operations given this CompletionToken return their results, modulo
+/// any leading error_code. A non-zero error code will be thrown, by
+/// default, but may be bound to a variable instead with the overload
+/// of the array-subscript oeprator.
+///
+/// EXAMPLE:
+/// // Within a function with a yield_context parameter named y
+///
+/// try {
+///    int i = f(y);
+/// } catch (const bs::system_error& ec) { ... }
+///
+/// bs::error_code ec;
+/// auto [i, s] = g(y[ec]);
+///
+/// Blocking calls
+/// --------------
+///
+/// ceph::async::use_blocked, defined in src/common/async/blocked_completion.h
+/// Suspends the current thread of execution, returning the results of
+/// the operation on resumption. Its calling convention is analogous to
+/// that of yield_context.
+///
+/// EXAMPLE:
+/// try {
+///    int i = f(ca::use_blocked);
+/// } catch (const bs::system_error& e) { ... }
+///
+/// bs::error_code ec;
+/// auto [i, s] = g(ca::use_blocked[ec]);
+///
+/// librados Completions
+/// --------------------
+///
+/// If src/common/async/librados_completion.h is included in the
+/// current translation unit, then librados::AioCompletion* may be used
+/// as a CompletionToken. This is only permitted when the completion
+/// signature is either bs::system_error or void. The return type of
+/// functions provided a CompletionToken of AioCompletion* is void. If
+/// the signature includes an error code and the error code is set,
+/// then the error is translated to an int which is set as the result
+/// of the AioCompletion.
+///
+/// EXAMPLE:
+/// // Assume an asynchronous function h whose signature is bs::error_code.
+///
+/// AioCompletion* c = Rados::aio_create_completion();
+/// h(c);
+/// int r = c.get_return_value();
+///
+/// See also src/test/cls_fifo/bench_cls_fifo.cc for a full, simple
+/// example of a program using this class with coroutines.
+///
+///
+/// Markers
+/// =======
+///
+/// Markers represent a position within the FIFO. Internally, they are
+/// part/offset pairs. Externally, they are ordered but otherwise
+/// opaque strings. Markers that compare lower denote positions closer
+/// to the tail.
+///
+/// A marker is returned with every entry from a list() operation. They
+/// may be supplied to a list operation to resume from a given
+/// position, and must be supplied to trim give the position to which
+/// to trim.
+
+class FIFO {
+public:
+
+  FIFO(const FIFO&) = delete;
+  FIFO& operator =(const FIFO&) = delete;
+  FIFO(FIFO&&) = delete;
+  FIFO& operator =(FIFO&&) = delete;
+
+  /// Open an existing FIFO.
+  /// Signature: (bs::error_code ec, std::unique_ptr<FIFO> f)
+  template<typename CT>
+  static auto open(RADOS& r, //< RADOS handle
+		   const IOContext& ioc, //< Context for pool, namespace, etc.
+		   Object oid, //< OID for the 'main' object of the FIFO
+		   CT&& ct, //< CompletionToken
+		   /// Fail if is not this version
+		   std::optional<fifo::objv> objv = std::nullopt,
+		   /// Default executor. By default use the one
+		   /// associated with the RADOS handle.
+		   std::optional<ba::executor> executor = std::nullopt) {
+    ba::async_completion<CT, void(bs::error_code,
+				  std::unique_ptr<FIFO>)> init(ct);
+    auto e = ba::get_associated_executor(init.completion_handler,
+					 executor.value_or(r.get_executor()));
+    auto a = ba::get_associated_allocator(init.completion_handler);
+    _read_meta_(
+      &r, oid, ioc, objv,
+      ca::bind_ea(
+	e, a,
+	[&r, ioc, oid, executor, handler = std::move(init.completion_handler)]
+	(bs::error_code ec, fifo::info info,
+	 std::uint32_t size, std::uint32_t over) mutable {
+	  std::unique_ptr<FIFO> f(
+	    new FIFO(r, ioc, oid, executor.value_or(r.get_executor())));
+	  f->info = info;
+	  f->part_header_size = size;
+	  f->part_entry_overhead = over;
+	  // If there are journal entries, process them, in case
+	  // someone crashed mid-transaction.
+	  if (!ec && !info.journal.empty()) {
+	    auto e = ba::get_associated_executor(handler, f->get_executor());
+	    auto a = ba::get_associated_allocator(handler);
+	    auto g = f.get();
+	    g->_process_journal(
+	      ca::bind_ea(
+		e, a,
+		[f = std::move(f),
+		 handler = std::move(handler)](bs::error_code ec) mutable {
+		  std::move(handler)(ec, std::move(f));
+		}));
+	    return;
+	  }
+	  std::move(handler)(ec, std::move(f));
+	  return;
+	}));
+    return init.result.get();
+  }
+
+  /// Open an existing or create a new FIFO.
+  /// Signature: (bs::error_code ec, std::unique_ptr<FIFO> f)
+  template<typename CT>
+  static auto create(RADOS& r, /// RADOS handle
+		     const IOContext& ioc, /// Context for pool, namespace, etc.
+		     Object oid, /// OID for the 'main' object of the FIFO
+		     CT&& ct, /// CompletionToken
+		     /// Fail if FIFO exists and is not this version
+		     std::optional<fifo::objv> objv = std::nullopt,
+		     /// Custom prefix for parts
+		     std::optional<std::string_view> oid_prefix = std::nullopt,
+		     /// Fail if FIFO already exists
+		     bool exclusive = false,
+		     /// Size at which a part is considered full
+		     std::uint64_t max_part_size = default_max_part_size,
+		     /// Maximum size of any entry
+		     std::uint64_t max_entry_size = default_max_entry_size,
+		     /// Default executor. By default use the one
+		     /// associated with the RADOS handle.
+		     std::optional<ba::executor> executor = std::nullopt) {
+    ba::async_completion<CT, void(bs::error_code,
+				  std::unique_ptr<FIFO>)> init(ct);
+    WriteOp op;
+    create_meta(op, oid, objv, oid_prefix, exclusive, max_part_size,
+		max_entry_size);
+    auto e = ba::get_associated_executor(init.completion_handler,
+					 executor.value_or(r.get_executor()));
+    auto a = ba::get_associated_allocator(init.completion_handler);
+    r.execute(
+      oid, ioc, std::move(op),
+      ca::bind_ea(
+	e, a,
+	[objv, &r, ioc, oid, executor, handler = std::move(init.completion_handler)]
+	(bs::error_code ec) mutable {
+	  if (ec) {
+	    std::move(handler)(ec, nullptr);
+	    return;
+	  }
+	  auto e = ba::get_associated_executor(
+	    handler, executor.value_or(r.get_executor()));
+	  auto a = ba::get_associated_allocator(handler);
+	  FIFO::_read_meta_(
+	    &r, oid, ioc, objv,
+	    ca::bind_ea(
+	      e, a,
+	      [&r, ioc, executor, oid, handler = std::move(handler)]
+	      (bs::error_code ec, fifo::info info,
+	       std::uint32_t size, std::uint32_t over) mutable {
+		std::unique_ptr<FIFO> f(
+		  new FIFO(r, ioc, oid, executor.value_or(r.get_executor())));
+		f->info = info;
+		f->part_header_size = size;
+		f->part_entry_overhead = over;
+		if (!ec && !info.journal.empty()) {
+		  auto e = ba::get_associated_executor(handler,
+						       f->get_executor());
+		  auto a = ba::get_associated_allocator(handler);
+		  auto g = f.get();
+		  g->_process_journal(
+		    ca::bind_ea(
+		      e, a,
+		      [f = std::move(f), handler = std::move(handler)]
+		      (bs::error_code ec) mutable {
+			std::move(handler)(ec, std::move(f));
+		      }));
+		  return;
+		}
+		std::move(handler)(ec, std::move(f));
+	      }));
+	    }));
+    return init.result.get();
+  }
+
+  /// Force a re-read of FIFO metadata.
+  /// Signature: (bs::error_code ec)
+  template<typename CT>
+  auto read_meta(CT&& ct, //< CompletionToken
+		 /// Fail if FIFO not at this version
+		 std::optional<fifo::objv> objv = std::nullopt) {
+    std::unique_lock l(m);
+    auto version = info.version;
+    l.unlock();
+    ba::async_completion<CT, void(bs::error_code)> init(ct);
+    auto e = ba::get_associated_executor(init.completion_handler,
+					 get_executor());
+    auto a = ba::get_associated_allocator(init.completion_handler);
+    _read_meta_(
+      r, oid, ioc, objv,
+      ca::bind_ea(
+	e, a,
+	[this, version, handler = std::move(init.completion_handler)]
+	(bs::error_code ec, fifo::info newinfo,
+	 std::uint32_t size, std::uint32_t over) mutable {
+	  std::unique_lock l(m);
+	  if (version == info.version) {
+	    info = newinfo;
+	    part_header_size = size;
+	    part_entry_overhead = over;
+	  }
+	  l.unlock();
+	  return std::move(handler)(ec);
+	}));
+    return init.result.get();
+  }
+
+  /// Return a reference to currently known metadata
+  const fifo::info& meta() const {
+    return info;
+  }
+
+  /// Return header size and entry overhead of partitions.
+  std::pair<std::uint32_t, std::uint32_t> get_part_layout_info() {
+    return {part_header_size, part_entry_overhead};
+  }
+
+  /// Push a single entry to the FIFO.
+  /// Signature: (bs::error_code)
+  template<typename CT>
+  auto push(const cb::list& bl, //< Bufferlist holding entry to push
+	    CT&& ct //< CompletionToken
+    ) {
+    return push(std::vector{ bl }, std::forward<CT>(ct));
+  }
+
+  /// Push a many entries to the FIFO.
+  /// Signature: (bs::error_code)
+  template<typename CT>
+  auto push(const std::vector<cb::list>& data_bufs, //< Entries to push
+	    CT&& ct //< CompletionToken
+    ) {
+    ba::async_completion<CT, void(bs::error_code)> init(ct);
+    std::unique_lock l(m);
+    auto max_entry_size = info.params.max_entry_size;
+    auto need_new_head = info.need_new_head();
+    l.unlock();
+    auto e = ba::get_associated_executor(init.completion_handler,
+					 get_executor());
+    auto a = ba::get_associated_allocator(init.completion_handler);
+    if (data_bufs.empty() ) {
+      // Can't fail if you don't try.
+      e.post(ca::bind_handler(std::move(init.completion_handler),
+				  bs::error_code{}), a);
+      return init.result.get();
+    }
+
+    // Validate sizes
+    for (const auto& bl : data_bufs) {
+      if (bl.length() > max_entry_size) {
+	ldout(r->cct(), 10) << __func__ << "(): entry too large: "
+			    << bl.length() << " > "
+			    << info.params.max_entry_size << dendl;
+	e.post(ca::bind_handler(std::move(init.completion_handler),
+				    errc::entry_too_large), a);
+	return init.result.get();
+      }
+    }
+
+    auto p = ca::bind_ea(e, a,
+			 Pusher(this, {data_bufs.begin(), data_bufs.end()},
+				  {}, 0, std::move(init.completion_handler)));
+
+    if (need_new_head) {
+      _prepare_new_head(std::move(p));
+    } else {
+      e.dispatch(std::move(p), a);
+    }
+    return init.result.get();
+  }
+
+  /// List the entries in a FIFO
+  /// Signature(bs::error_code ec, bs::vector<list_entry> entries, bool more)
+  ///
+  /// More is true if entries beyond the last exist.
+  /// The list entries are of the form:
+  /// data - Contents of the entry
+  /// marker - String representing the position of this entry within the FIFO.
+  /// mtime - Time (on the OSD) at which the entry was pushed.
+  template<typename CT>
+  auto list(int max_entries, //< Maximum number of entries to fetch
+	    /// Optionally, a marker indicating the position after
+	    /// which to begin listing. If null, begin at the tail.
+	    std::optional<std::string_view> markstr,
+	    CT&& ct //< CompletionToken
+    ) {
+    ba::async_completion<CT, void(bs::error_code,
+				  std::vector<list_entry>, bool)> init(ct);
+    std::unique_lock l(m);
+    std::int64_t part_num = info.tail_part_num;
+    l.unlock();
+    std::uint64_t ofs = 0;
+    auto a = ba::get_associated_allocator(init.completion_handler);
+    auto e = ba::get_associated_executor(init.completion_handler);
+
+    if (markstr) {
+      auto marker = to_marker(*markstr);
+      if (!marker) {
+	ldout(r->cct(), 0) << __func__
+			   << "(): failed to parse marker (" << *markstr
+			   << ")" << dendl;
+	e.post(ca::bind_handler(std::move(init.completion_handler),
+				errc::invalid_marker,
+				std::vector<list_entry>{}, false), a);
+	return init.result.get();
+      }
+      part_num = marker->num;
+      ofs = marker->ofs;
+    }
+
+    using handler_type = decltype(init.completion_handler);
+    auto ls = ceph::allocate_unique<Lister<handler_type>>(
+      a, this, part_num, ofs, max_entries,
+      std::move(init.completion_handler));
+    ls.release()->list();
+    return init.result.get();
+  }
+
+  /// Trim entries from the tail to the given position
+  /// Signature: (bs::error_code)
+  template<typename CT>
+  auto trim(std::string_view markstr, //< Position to which to trim, inclusive
+	    bool exclusive, //< If true, trim markers up to but NOT INCLUDING
+	                    //< markstr, otherwise trim markstr as well.
+	    CT&& ct //< CompletionToken
+    ) {
+    auto m = to_marker(markstr);
+    ba::async_completion<CT, void(bs::error_code)> init(ct);
+    auto a = ba::get_associated_allocator(init.completion_handler);
+    auto e = ba::get_associated_executor(init.completion_handler);
+    if (!m) {
+      ldout(r->cct(), 0) << __func__ << "(): failed to parse marker: marker="
+			 << markstr << dendl;
+      e.post(ca::bind_handler(std::move(init.completion_handler),
+			      errc::invalid_marker), a);
+      return init.result.get();
+    } else {
+      using handler_type = decltype(init.completion_handler);
+      auto t = ceph::allocate_unique<Trimmer<handler_type>>(
+	a, this, m->num, m->ofs, exclusive, std::move(init.completion_handler));
+      t.release()->trim();
+    }
+    return init.result.get();
+  }
+
+  /// Get information about a specific partition
+  /// Signature: (bs::error_code, part_info)
+  ///
+  /// part_info has the following entries
+  /// tag - A random string identifying this partition. Used internally
+  ///       as a sanity check to make sure operations haven't been misdirected
+  /// params - Data parameters, identical for every partition within a
+  ///          FIFO and the same as what is returned from get_part_layout()
+  /// magic - A random magic number, used internally as a prefix to
+  ///         every entry stored on the OSD to ensure sync
+  /// min_ofs - Offset of the first entry
+  /// max_ofs - Offset of the highest entry
+  /// min_index - Minimum entry index
+  /// max_index - Maximum entry index
+  /// max_time - Time of the latest push
+  ///
+  /// The difference between ofs and index is that ofs is a byte
+  /// offset. Index is a count. Nothing really uses indices, but
+  /// they're tracked and sanity-checked as an invariant on the OSD.
+  ///
+  /// max_ofs and max_time are the two that have been used externally
+  /// so far.
+  template<typename CT>
+  auto get_part_info(int64_t part_num, // The number of the partition
+		     CT&& ct // CompletionToken
+    ) {
+
+    ba::async_completion<CT, void(bs::error_code, part_info)> init(ct);
+    fifo::op::get_part_info gpi;
+    cb::list in;
+    encode(gpi, in);
+    ReadOp op;
+    auto e = ba::get_associated_executor(init.completion_handler,
+					 get_executor());
+    auto a = ba::get_associated_allocator(init.completion_handler);
+    auto reply = ceph::allocate_unique<
+      ExecDecodeCB<fifo::op::get_part_info_reply>>(a);
+
+    op.exec(fifo::op::CLASS, fifo::op::GET_PART_INFO, in,
+	    std::ref(*reply));
+    std::unique_lock l(m);
+    auto part_oid = info.part_oid(part_num);
+    l.unlock();
+    r->execute(part_oid, ioc, std::move(op), nullptr,
+	       ca::bind_ea(e, a,
+			   PartInfoGetter(std::move(init.completion_handler),
+					  std::move(reply))));
+    return init.result.get();
+  }
+
+  using executor_type = ba::executor;
+
+  /// Return the default executor, as specified at creation.
+  ba::executor get_executor() const {
+    return executor;
+  }
+
+private:
+  template<typename Handler>
+  friend class detail::JournalProcessor;
+  RADOS* const r;
+  const IOContext ioc;
+  const Object oid;
+  std::mutex m;
+
+  fifo::info info;
+
+  std::uint32_t part_header_size = 0xdeadbeef;
+  std::uint32_t part_entry_overhead = 0xdeadbeef;
+
+  ba::executor executor;
+
+  std::optional<marker> to_marker(std::string_view s);
+
+  template<typename Handler, typename T>
+  static void assoc_delete(const Handler& handler, T* t) {
+    typename std::allocator_traits<typename ba::associated_allocator<Handler>::type>
+      ::template rebind_alloc<T> a(
+	ba::get_associated_allocator(handler));
+    a.destroy(t);
+    a.deallocate(t, 1);
+  }
+
+  FIFO(RADOS& r,
+       IOContext ioc,
+       Object oid,
+       ba::executor executor)
+    : r(&r), ioc(std::move(ioc)), oid(oid), executor(executor) {}
+
+  std::string generate_tag() const;
+
+  template <typename T>
+  struct ExecDecodeCB {
+    bs::error_code ec;
+    T result;
+    void operator()(bs::error_code e, const cb::list& r) {
+      if (e) {
+        ec = e;
+        return;
+      }
+      try {
+        auto p = r.begin();
+        using ceph::decode;
+        decode(result, p);
+      } catch (const cb::error& err) {
+        ec = err.code();
+      }
+    }
+  };
+
+  template<typename Handler>
+  class MetaReader {
+    Handler handler;
+    using allocator_type = boost::asio::associated_allocator_t<Handler>;
+    using decoder_type = ExecDecodeCB<fifo::op::get_meta_reply>;
+    using decoder_ptr = ceph::allocated_unique_ptr<decoder_type, allocator_type>;
+    decoder_ptr decoder;
+  public:
+    MetaReader(Handler&& handler, decoder_ptr&& decoder)
+      : handler(std::move(handler)), decoder(std::move(decoder)) {}
+
+    void operator ()(bs::error_code ec) {
+      if (!ec) {
+        ec = decoder->ec;
+      }
+      auto reply = std::move(decoder->result);
+      decoder.reset(); // free handler-allocated memory before dispatching
+
+      std::move(handler)(ec, std::move(reply.info),
+			 std::move(reply.part_header_size),
+			 std::move(reply.part_entry_overhead));
+    }
+  };
+
+  // Renamed to get around a compiler bug in Bionic that kept
+  // complaining we weren't capturing 'this' to make a static function call.
+  template<typename Handler>
+  static void _read_meta_(RADOS* r, const Object& oid, const IOContext& ioc,
+			  std::optional<fifo::objv> objv,
+			  Handler&& handler, /* error_code, info, uint64,
+						uint64 */
+			  std::optional<ba::executor> executor = std::nullopt){
+    fifo::op::get_meta gm;
+
+    gm.version = objv;
+
+    cb::list in;
+    encode(gm, in);
+    ReadOp op;
+
+    auto a = ba::get_associated_allocator(handler);
+    auto reply =
+      ceph::allocate_unique<ExecDecodeCB<fifo::op::get_meta_reply>>(a);
+
+    auto e = ba::get_associated_executor(handler);
+    op.exec(fifo::op::CLASS, fifo::op::GET_META, in, std::ref(*reply));
+    r->execute(oid, ioc, std::move(op), nullptr,
+	       ca::bind_ea(e, a, MetaReader(std::move(handler),
+					    std::move(reply))));
+  };
+
+  template<typename Handler>
+  void _read_meta(Handler&& handler /* error_code */) {
+    auto e = ba::get_associated_executor(handler, get_executor());
+    auto a = ba::get_associated_allocator(handler);
+    _read_meta_(r, oid, ioc,
+		nullopt,
+		ca::bind_ea(
+		  e, a,
+		  [this,
+		   handler = std::move(handler)](bs::error_code ec,
+						 fifo::info&& info,
+						 std::uint64_t phs,
+						 std::uint64_t peo) mutable {
+		    std::unique_lock l(m);
+		    if (ec) {
+		      l.unlock();
+		      std::move(handler)(ec);
+		      return;
+		    }
+		    // We have a newer version already!
+		    if (!info.version.same_or_later(this->info.version)) {
+		      l.unlock();
+		      std::move(handler)(bs::error_code{});
+		      return;
+		    }
+		    this->info = std::move(info);
+		    part_header_size = phs;
+		    part_entry_overhead = peo;
+		    l.unlock();
+		    std::move(handler)(bs::error_code{});
+		  }), get_executor());
+  }
+
+  bs::error_code apply_update(fifo::info* info,
+			      const fifo::objv& objv,
+			      const fifo::update& update);
+
+
+  template<typename Handler>
+  void _update_meta(const fifo::update& update,
+		    fifo::objv version,
+		    Handler&& handler /* error_code, bool */) {
+    WriteOp op;
+
+    cls::fifo::update_meta(op, info.version, update);
+
+    auto a = ba::get_associated_allocator(handler);
+    auto e = ba::get_associated_executor(handler, get_executor());
+
+    r->execute(
+      oid, ioc, std::move(op),
+      ca::bind_ea(
+	e, a,
+	[this, e, a, version, update,
+	 handler = std::move(handler)](bs::error_code ec) mutable {
+	  if (ec && ec != bs::errc::operation_canceled) {
+	    std::move(handler)(ec, bool{});
+	    return;
+	  }
+
+	  auto canceled = (ec == bs::errc::operation_canceled);
+
+	  if (!canceled) {
+	    ec = apply_update(&info,
+			      version,
+			      update);
+	    if (ec) {
+	      canceled = true;
+	    }
+	  }
+
+	  if (canceled) {
+	    _read_meta(
+	      ca::bind_ea(
+		e, a,
+		[handler = std::move(handler)](bs::error_code ec) mutable {
+		  std::move(handler)(ec, ec ? false : true);
+		}));
+	    return;
+	  }
+	  std::move(handler)(ec, false);
+	  return;
+	}));
+  }
+
+  template<typename Handler>
+  auto _process_journal(Handler&& handler /* error_code */) {
+    auto a = ba::get_associated_allocator(std::ref(handler));
+    auto j = ceph::allocate_unique<detail::JournalProcessor<Handler>>(
+      a, this, std::move(handler));
+    auto p = j.release();
+    p->process();
+  }
+
+  template<typename Handler>
+  class NewPartPreparer {
+    FIFO* f;
+    Handler handler;
+    std::vector<fifo::journal_entry> jentries;
+    int i;
+    std::int64_t new_head_part_num;
+
+  public:
+
+    void operator ()(bs::error_code ec, bool canceled) {
+      if (ec) {
+	std::move(handler)(ec);
+	return;
+      }
+
+      if (canceled) {
+	std::unique_lock l(f->m);
+	auto iter = f->info.journal.find(jentries.front().part_num);
+	auto max_push_part_num = f->info.max_push_part_num;
+	auto head_part_num = f->info.head_part_num;
+	auto version = f->info.version;
+	auto found = (iter != f->info.journal.end());
+	l.unlock();
+	if ((max_push_part_num >= jentries.front().part_num &&
+	    head_part_num >= new_head_part_num)) {
+	  /* raced, but new part was already written */
+	  std::move(handler)(bs::error_code{});
+	  return;
+	}
+	if (i >= MAX_RACE_RETRIES) {
+	  std::move(handler)(errc::raced);
+	  return;
+	}
+	if (!found) {
+	  auto e = ba::get_associated_executor(handler, f->get_executor());
+	  auto a = ba::get_associated_allocator(handler);
+	  f->_update_meta(fifo::update{}
+			  .journal_entries_add(jentries),
+                          version,
+			  ca::bind_ea(
+			    e, a,
+			    NewPartPreparer(f, std::move(handler),
+					    jentries,
+					    i + 1, new_head_part_num)));
+	  return;
+	}
+	// Fall through. We still need to process the journal.
+      }
+      f->_process_journal(std::move(handler));
+      return;
+    }
+
+    NewPartPreparer(FIFO* f,
+		    Handler&& handler,
+		    std::vector<fifo::journal_entry> jentries,
+		    int i, std::int64_t new_head_part_num)
+      : f(f), handler(std::move(handler)), jentries(std::move(jentries)),
+	i(i), new_head_part_num(new_head_part_num) {}
+  };
+
+  template<typename Handler>
+  void _prepare_new_part(bool is_head,
+			 Handler&& handler /* error_code */) {
+    std::unique_lock l(m);
+    std::vector jentries = { info.next_journal_entry(generate_tag()) };
+    std::int64_t new_head_part_num = info.head_part_num;
+    auto version = info.version;
+
+    if (is_head) {
+      auto new_head_jentry = jentries.front();
+      new_head_jentry.op = fifo::journal_entry::Op::set_head;
+      new_head_part_num = jentries.front().part_num;
+      jentries.push_back(std::move(new_head_jentry));
+    }
+    l.unlock();
+
+    auto e = ba::get_associated_executor(handler, get_executor());
+    auto a = ba::get_associated_allocator(handler);
+    _update_meta(fifo::update{}.journal_entries_add(jentries),
+		 version,
+		 ca::bind_ea(
+		   e, a,
+		   NewPartPreparer(this, std::move(handler),
+				   jentries, 0, new_head_part_num)));
+  }
+
+  template<typename Handler>
+  class NewHeadPreparer {
+    FIFO* f;
+    Handler handler;
+    int i;
+    std::int64_t new_head_num;
+
+  public:
+
+    void operator ()(bs::error_code ec, bool canceled) {
+      std::unique_lock l(f->m);
+      auto head_part_num = f->info.head_part_num;
+      auto version = f->info.version;
+      l.unlock();
+
+      if (ec) {
+	std::move(handler)(ec);
+	return;
+      }
+      if (canceled) {
+	if (i >= MAX_RACE_RETRIES) {
+	  std::move(handler)(errc::raced);
+	  return;
+	}
+
+	// Raced, but there's still work to do!
+	if (head_part_num < new_head_num) {
+	  auto e = ba::get_associated_executor(handler, f->get_executor());
+	  auto a = ba::get_associated_allocator(handler);
+	  f->_update_meta(fifo::update{}.head_part_num(new_head_num),
+			  version,
+			  ca::bind_ea(
+			    e, a,
+			    NewHeadPreparer(f, std::move(handler),
+					    i + 1,
+					    new_head_num)));
+	  return;
+	}
+      }
+      // Either we succeeded, or we were raced by someone who did it for us.
+      std::move(handler)(bs::error_code{});
+      return;
+    }
+
+    NewHeadPreparer(FIFO* f,
+		    Handler&& handler,
+		    int i, std::int64_t new_head_num)
+      : f(f), handler(std::move(handler)), i(i), new_head_num(new_head_num) {}
+  };
+
+  template<typename Handler>
+  void _prepare_new_head(Handler&& handler /* error_code */) {
+    std::unique_lock l(m);
+    int64_t new_head_num = info.head_part_num + 1;
+    auto max_push_part_num = info.max_push_part_num;
+    auto version = info.version;
+    l.unlock();
+
+    if (max_push_part_num < new_head_num) {
+      auto e = ba::get_associated_executor(handler, get_executor());
+      auto a = ba::get_associated_allocator(handler);
+      _prepare_new_part(
+	true,
+	ca::bind_ea(
+	  e, a,
+	  [this, new_head_num,
+	   handler = std::move(handler)](bs::error_code ec) mutable {
+	    if (ec) {
+	      handler(ec);
+	      return;
+	    }
+	    std::unique_lock l(m);
+	    if (info.max_push_part_num < new_head_num) {
+	      l.unlock();
+	      ldout(r->cct(), 0)
+		<< "ERROR: " << __func__
+		<< ": after new part creation: meta_info.max_push_part_num="
+		<< info.max_push_part_num << " new_head_num="
+		<< info.max_push_part_num << dendl;
+	      std::move(handler)(errc::inconsistency);
+	    } else {
+	      l.unlock();
+	      std::move(handler)(bs::error_code{});
+	    }
+	  }));
+      return;
+    }
+    auto e = ba::get_associated_executor(handler, get_executor());
+    auto a = ba::get_associated_allocator(handler);
+    _update_meta(fifo::update{}.head_part_num(new_head_num),
+		 version,
+		 ca::bind_ea(
+		   e, a,
+		   NewHeadPreparer(this, std::move(handler), 0,
+				   new_head_num)));
+  }
+
+  template<typename T>
+  struct ExecHandleCB {
+    bs::error_code ec;
+    T result;
+    void operator()(bs::error_code e, const T& t) {
+      if (e) {
+        ec = e;
+        return;
+      }
+      result = t;
+    }
+  };
+
+  template<typename Handler>
+  class EntryPusher {
+    Handler handler;
+    using allocator_type = boost::asio::associated_allocator_t<Handler>;
+    using decoder_type = ExecHandleCB<int>;
+    using decoder_ptr = ceph::allocated_unique_ptr<decoder_type, allocator_type>;
+    decoder_ptr decoder;
+
+  public:
+
+    EntryPusher(Handler&& handler, decoder_ptr&& decoder)
+      : handler(std::move(handler)), decoder(std::move(decoder)) {}
+
+    void operator ()(bs::error_code ec) {
+      if (!ec) {
+        ec = decoder->ec;
+      }
+      auto reply = std::move(decoder->result);
+      decoder.reset(); // free handler-allocated memory before dispatching
+
+      std::move(handler)(ec, std::move(reply));
+    }
+  };
+
+  template<typename Handler>
+  auto push_entries(const std::deque<cb::list>& data_bufs,
+		    Handler&& handler /* error_code, int */) {
+    WriteOp op;
+    std::unique_lock l(m);
+    auto head_part_num = info.head_part_num;
+    auto tag = info.head_tag;
+    auto oid = info.part_oid(head_part_num);
+    l.unlock();
+
+    auto a = ba::get_associated_allocator(handler);
+    auto reply = ceph::allocate_unique<ExecHandleCB<int>>(a);
+
+    auto e = ba::get_associated_executor(handler, get_executor());
+    push_part(op, tag, data_bufs, std::ref(*reply));
+    return r->execute(oid, ioc, std::move(op),
+		      ca::bind_ea(e, a, EntryPusher(std::move(handler),
+						    std::move(reply))));
+  }
+
+  template<typename CT>
+  auto trim_part(int64_t part_num,
+		 uint64_t ofs,
+		 std::optional<std::string_view> tag,
+		 bool exclusive,
+		 CT&& ct) {
+    WriteOp op;
+    cls::fifo::trim_part(op, tag, ofs, exclusive);
+    return r->execute(info.part_oid(part_num), ioc, std::move(op),
+		      std::forward<CT>(ct));
+  }
+
+
+  template<typename Handler>
+  class Pusher {
+    FIFO* f;
+    std::deque<cb::list> remaining;
+    std::deque<cb::list> batch;
+    int i;
+    Handler handler;
+
+    void prep_then_push(const unsigned successes) {
+      std::unique_lock l(f->m);
+      auto max_part_size = f->info.params.max_part_size;
+      auto part_entry_overhead = f->part_entry_overhead;
+      l.unlock();
+
+      uint64_t batch_len = 0;
+      if (successes > 0) {
+	if (successes == batch.size()) {
+	  batch.clear();
+	} else  {
+	  batch.erase(batch.begin(), batch.begin() + successes);
+	  for (const auto& b : batch) {
+	    batch_len +=  b.length() + part_entry_overhead;
+	  }
+	}
+      }
+
+      if (batch.empty() && remaining.empty()) {
+	std::move(handler)(bs::error_code{});
+	return;
+      }
+
+      while (!remaining.empty() &&
+	     (remaining.front().length() + batch_len <= max_part_size)) {
+
+	/* We can send entries with data_len up to max_entry_size,
+	   however, we want to also account the overhead when
+	   dealing with multiple entries. Previous check doesn't
+	   account for overhead on purpose. */
+	batch_len += remaining.front().length() + part_entry_overhead;
+	batch.push_back(std::move(remaining.front()));
+	remaining.pop_front();
+      }
+      push();
+    }
+
+    void push() {
+      auto e = ba::get_associated_executor(handler, f->get_executor());
+      auto a = ba::get_associated_allocator(handler);
+      f->push_entries(batch,
+		      ca::bind_ea(e, a,
+				  Pusher(f, std::move(remaining),
+					 batch, i,
+					 std::move(handler))));
+    }
+
+  public:
+
+    // Initial call!
+    void operator ()() {
+      prep_then_push(0);
+    }
+
+    // Called with response to push_entries
+    void operator ()(bs::error_code ec, int r) {
+      if (ec == bs::errc::result_out_of_range) {
+	auto e = ba::get_associated_executor(handler, f->get_executor());
+	auto a = ba::get_associated_allocator(handler);
+	f->_prepare_new_head(
+	  ca::bind_ea(e, a,
+		      Pusher(f, std::move(remaining),
+			     std::move(batch), i,
+			     std::move(handler))));
+	return;
+      }
+      if (ec) {
+	std::move(handler)(ec);
+	return;
+      }
+      i = 0; // We've made forward progress, so reset the race counter!
+      prep_then_push(r);
+    }
+
+    // Called with response to prepare_new_head
+    void operator ()(bs::error_code ec) {
+      if (ec == bs::errc::operation_canceled) {
+	if (i == MAX_RACE_RETRIES) {
+	  ldout(f->r->cct(), 0)
+	    << "ERROR: " << __func__
+	    << "(): race check failed too many times, likely a bug" << dendl;
+	  std::move(handler)(make_error_code(errc::raced));
+	  return;
+	}
+	++i;
+      } else if (ec) {
+	std::move(handler)(ec);
+	return;
+      }
+
+      if (batch.empty()) {
+	prep_then_push(0);
+	return;
+      } else {
+	push();
+	return;
+      }
+    }
+
+    Pusher(FIFO* f, std::deque<cb::list>&& remaining,
+	   std::deque<cb::list> batch, int i,
+	   Handler&& handler)
+      : f(f), remaining(std::move(remaining)),
+	batch(std::move(batch)), i(i),
+	handler(std::move(handler)) {}
+  };
+
+  template<typename Handler>
+  class Lister {
+    FIFO* f;
+    std::vector<list_entry> result;
+    bool more = false;
+    std::int64_t part_num;
+    std::uint64_t ofs;
+    int max_entries;
+    bs::error_code ec_out;
+    std::vector<fifo::part_list_entry> entries;
+    bool part_more = false;
+    bool part_full = false;
+    Handler handler;
+
+    void handle(bs::error_code ec) {
+      auto h = std::move(handler);
+      auto m = more;
+      auto r = std::move(result);
+
+      FIFO::assoc_delete(h, this);
+      std::move(h)(ec, std::move(r), m);
+    }
+
+  public:
+    Lister(FIFO* f, std::int64_t part_num, std::uint64_t ofs, int max_entries,
+	   Handler&& handler)
+      : f(f), part_num(part_num), ofs(ofs), max_entries(max_entries),
+	handler(std::move(handler)) {
+      result.reserve(max_entries);
+    }
+
+
+    Lister(const Lister&) = delete;
+    Lister& operator =(const Lister&) = delete;
+    Lister(Lister&&) = delete;
+    Lister& operator =(Lister&&) = delete;
+
+    void list() {
+      if (max_entries > 0) {
+	ReadOp op;
+	ec_out.clear();
+	part_more = false;
+	part_full = false;
+	entries.clear();
+
+	std::unique_lock l(f->m);
+	auto part_oid = f->info.part_oid(part_num);
+	l.unlock();
+
+	list_part(op,
+		  {},
+		  ofs,
+		  max_entries,
+		  &ec_out,
+		  &entries,
+		  &part_more,
+		  &part_full,
+		  nullptr);
+	auto e = ba::get_associated_executor(handler, f->get_executor());
+	auto a = ba::get_associated_allocator(handler);
+	f->r->execute(
+	  part_oid,
+	  f->ioc,
+	  std::move(op),
+	  nullptr,
+	  ca::bind_ea(
+	    e, a,
+	    [t = std::unique_ptr<Lister>(this), this,
+	     part_oid](bs::error_code ec) mutable {
+	      t.release();
+	      if (ec == bs::errc::no_such_file_or_directory) {
+		auto e = ba::get_associated_executor(handler,
+						     f->get_executor());
+		auto a = ba::get_associated_allocator(handler);
+		f->_read_meta(
+		  ca::bind_ea(
+		    e, a,
+		    [this](bs::error_code ec) mutable {
+		      if (ec) {
+			handle(ec);
+			return;
+		      }
+
+		      if (part_num < f->info.tail_part_num) {
+			/* raced with trim? restart */
+			max_entries += result.size();
+			result.clear();
+			part_num = f->info.tail_part_num;
+			ofs = 0;
+			list();
+		      }
+		      /* assuming part was not written yet, so end of data */
+		      more = false;
+		      handle({});
+		      return;
+		    }));
+		return;
+	      }
+	      if (ec) {
+		ldout(f->r->cct(), 0)
+		  << __func__
+		  << "(): list_part() on oid=" << part_oid
+		  << " returned ec=" << ec.message() << dendl;
+		handle(ec);
+		return;
+	      }
+	      if (ec_out) {
+		ldout(f->r->cct(), 0)
+		  << __func__
+		  << "(): list_part() on oid=" << f->info.part_oid(part_num)
+		  << " returned ec=" << ec_out.message() << dendl;
+		handle(ec_out);
+		return;
+	      }
+
+	      more = part_full || part_more;
+	      for (auto& entry : entries) {
+		list_entry e;
+		e.data = std::move(entry.data);
+		e.marker = marker{part_num, entry.ofs}.to_string();
+		e.mtime = entry.mtime;
+		result.push_back(std::move(e));
+	      }
+	      max_entries -= entries.size();
+	      entries.clear();
+	      if (max_entries > 0 &&
+		  part_more) {
+		list();
+		return;
+	      }
+
+	      if (!part_full) { /* head part is not full */
+		handle({});
+		return;
+	      }
+	      ++part_num;
+	      ofs = 0;
+	    list();
+	    }));
+      } else {
+	handle({});
+	return;
+      }
+    }
+  };
+
+  template<typename Handler>
+  class Trimmer {
+    FIFO* f;
+    std::int64_t part_num;
+    std::uint64_t ofs;
+    bool exclusive;
+    Handler handler;
+    std::int64_t pn;
+    int i = 0;
+
+    void handle(bs::error_code ec) {
+      auto h = std::move(handler);
+
+      FIFO::assoc_delete(h, this);
+      return std::move(h)(ec);
+    }
+
+    void update() {
+      std::unique_lock l(f->m);
+      auto objv = f->info.version;
+      l.unlock();
+      auto a = ba::get_associated_allocator(handler);
+      auto e = ba::get_associated_executor(handler, f->get_executor());
+      f->_update_meta(
+	fifo::update{}.tail_part_num(part_num),
+	objv,
+	ca::bind_ea(
+	  e, a,
+	  [this, t = std::unique_ptr<Trimmer>(this)](bs::error_code ec,
+						     bool canceled) mutable {
+	    t.release();
+	    if (canceled)
+	      if (i >= MAX_RACE_RETRIES) {
+		ldout(f->r->cct(), 0)
+		  << "ERROR: " << __func__
+		  << "(): race check failed too many times, likely a bug"
+		  << dendl;
+		handle(errc::raced);
+		return;
+	      }
+	    std::unique_lock l(f->m);
+	    auto tail_part_num = f->info.tail_part_num;
+	    l.unlock();
+	    if (tail_part_num < part_num) {
+	      ++i;
+	      update();
+	      return;
+	    }
+	    handle({});
+	    return;
+	  }));
+    }
+
+  public:
+    Trimmer(FIFO* f, std::int64_t part_num, std::uint64_t ofs,
+	    bool exclusive, Handler&& handler)
+      : f(f), part_num(part_num), ofs(ofs), exclusive(exclusive),
+	handler(std::move(handler)) {
+      std::unique_lock l(f->m);
+      pn = f->info.tail_part_num;
+    }
+
+    void trim() {
+      auto a = ba::get_associated_allocator(handler);
+      auto e = ba::get_associated_executor(handler, f->get_executor());
+      if (pn < part_num) {
+	std::unique_lock l(f->m);
+	auto max_part_size = f->info.params.max_part_size;
+	l.unlock();
+	f->trim_part(
+	  pn, max_part_size, std::nullopt,
+	  false,
+	  ca::bind_ea(
+	    e, a,
+	    [t = std::unique_ptr<Trimmer>(this),
+	     this](bs::error_code ec) mutable {
+	      t.release();
+	      if (ec && ec != bs::errc::no_such_file_or_directory) {
+		ldout(f->r->cct(), 0)
+		  << __func__ << "(): ERROR: trim_part() on part="
+		  << pn << " returned ec=" << ec.message() << dendl;
+		handle(ec);
+		return;
+	      }
+	      ++pn;
+	      trim();
+	    }));
+	return;
+      }
+      f->trim_part(
+	part_num, ofs, std::nullopt, exclusive,
+	ca::bind_ea(
+	  e, a,
+	  [t = std::unique_ptr<Trimmer>(this),
+	    this](bs::error_code ec) mutable {
+	    t.release();
+	    if (ec && ec != bs::errc::no_such_file_or_directory) {
+	      ldout(f->r->cct(), 0)
+		<< __func__ << "(): ERROR: trim_part() on part=" << part_num
+		<< " returned ec=" << ec.message() << dendl;
+	      handle(ec);
+	      return;
+	    }
+	    std::unique_lock l(f->m);
+	    auto tail_part_num = f->info.tail_part_num;
+	    l.unlock();
+	    if (part_num <= tail_part_num) {
+	      /* don't need to modify meta info */
+	      handle({});
+	      return;
+	    }
+	    update();
+	  }));
+    }
+  };
+
+  template<typename Handler>
+  class PartInfoGetter {
+    Handler handler;
+    using allocator_type = boost::asio::associated_allocator_t<Handler>;
+    using decoder_type = ExecDecodeCB<fifo::op::get_part_info_reply>;
+    using decoder_ptr = ceph::allocated_unique_ptr<decoder_type, allocator_type>;
+    decoder_ptr decoder;
+  public:
+    PartInfoGetter(Handler&& handler, decoder_ptr&& decoder)
+      : handler(std::move(handler)), decoder(std::move(decoder)) {}
+
+    void operator ()(bs::error_code ec) {
+      if (!ec) {
+        ec = decoder->ec;
+      }
+      auto reply = std::move(decoder->result);
+      decoder.reset(); // free handler-allocated memory before dispatching
+
+      auto p = ca::bind_handler(std::move(handler),
+				ec, std::move(reply.header));
+      std::move(p)();
+    }
+  };
+
+
+};
+
+namespace detail {
+template<typename Handler>
+class JournalProcessor {
+private:
+  FIFO* const fifo;
+  Handler handler;
+
+  std::vector<fifo::journal_entry> processed;
+  std::multimap<std::int64_t, fifo::journal_entry> journal;
+  std::multimap<std::int64_t, fifo::journal_entry>::iterator iter;
+  std::int64_t new_tail;
+  std::int64_t new_head;
+  std::int64_t new_max;
+  int race_retries = 0;
+
+  template<typename CT>
+  auto create_part(int64_t part_num, std::string_view tag, CT&& ct) {
+    WriteOp op;
+    op.create(false); /* We don't need exclusivity, part_init ensures
+			 we're creating from the  same journal entry. */
+    std::unique_lock l(fifo->m);
+    part_init(op, tag, fifo->info.params);
+    auto oid = fifo->info.part_oid(part_num);
+    l.unlock();
+    return fifo->r->execute(oid, fifo->ioc,
+			    std::move(op), std::forward<CT>(ct));
+  }
+
+  template<typename CT>
+  auto remove_part(int64_t part_num, std::string_view tag, CT&& ct) {
+    WriteOp op;
+    op.remove();
+    std::unique_lock l(fifo->m);
+    auto oid = fifo->info.part_oid(part_num);
+    l.unlock();
+    return fifo->r->execute(oid, fifo->ioc,
+			    std::move(op), std::forward<CT>(ct));
+  }
+
+  template<typename PP>
+  void process_journal_entry(const fifo::journal_entry& entry,
+			     PP&& pp) {
+    switch (entry.op) {
+    case fifo::journal_entry::Op::unknown:
+      std::move(pp)(errc::inconsistency);
+      return;
+      break;
+
+    case fifo::journal_entry::Op::create:
+      create_part(entry.part_num, entry.part_tag, std::move(pp));
+      return;
+      break;
+    case fifo::journal_entry::Op::set_head:
+      ba::post(ba::get_associated_executor(handler, fifo->get_executor()),
+		      [pp = std::move(pp)]() mutable {
+			std::move(pp)(bs::error_code{});
+		      });
+      return;
+      break;
+    case fifo::journal_entry::Op::remove:
+      remove_part(entry.part_num, entry.part_tag, std::move(pp));
+      return;
+      break;
+    }
+    std::move(pp)(errc::inconsistency);
+    return;
+  }
+
+  auto journal_entry_finisher(const fifo::journal_entry& entry) {
+    auto a = ba::get_associated_allocator(handler);
+    auto e = ba::get_associated_executor(handler, fifo->get_executor());
+    return
+      ca::bind_ea(
+	e, a,
+	[t = std::unique_ptr<JournalProcessor>(this), this,
+	 entry](bs::error_code ec) mutable {
+	  t.release();
+	  if (entry.op == fifo::journal_entry::Op::remove &&
+	      ec == bs::errc::no_such_file_or_directory)
+	    ec.clear();
+
+	  if (ec) {
+	    ldout(fifo->r->cct(), 0)
+	      << __func__
+	      << "(): ERROR: failed processing journal entry for part="
+	      << entry.part_num << " with error " << ec.message()
+	      << " Bug or inconsistency." << dendl;
+	    handle(errc::inconsistency);
+	    return;
+	  } else {
+	    switch (entry.op) {
+	    case fifo::journal_entry::Op::unknown:
+	      // Can't happen. Filtered out in process_journal_entry.
+	      abort();
+	      break;
+
+	    case fifo::journal_entry::Op::create:
+	      if (entry.part_num > new_max) {
+		new_max = entry.part_num;
+	      }
+	      break;
+	    case fifo::journal_entry::Op::set_head:
+	      if (entry.part_num > new_head) {
+		new_head = entry.part_num;
+	      }
+	      break;
+	    case fifo::journal_entry::Op::remove:
+	      if (entry.part_num >= new_tail) {
+		new_tail = entry.part_num + 1;
+	      }
+	      break;
+	    }
+	    processed.push_back(entry);
+	  }
+	  ++iter;
+	  process();
+	});
+  }
+
+  struct JournalPostprocessor {
+    std::unique_ptr<JournalProcessor> j_;
+    bool first;
+    void operator ()(bs::error_code ec, bool canceled) {
+      std::optional<int64_t> tail_part_num;
+      std::optional<int64_t> head_part_num;
+      std::optional<int64_t> max_part_num;
+
+      auto j = j_.release();
+
+      if (!first && !ec && !canceled) {
+	j->handle({});
+	return;
+      }
+
+      if (canceled) {
+	if (j->race_retries >= MAX_RACE_RETRIES) {
+	  ldout(j->fifo->r->cct(), 0) << "ERROR: " << __func__ <<
+	    "(): race check failed too many times, likely a bug" << dendl;
+	  j->handle(errc::raced);
+	  return;
+	}
+
+	++j->race_retries;
+
+	std::vector<fifo::journal_entry> new_processed;
+	std::unique_lock l(j->fifo->m);
+	for (auto& e : j->processed) {
+	  auto jiter = j->fifo->info.journal.find(e.part_num);
+	  /* journal entry was already processed */
+	  if (jiter == j->fifo->info.journal.end() ||
+	      !(jiter->second == e)) {
+	    continue;
+	  }
+	  new_processed.push_back(e);
+	}
+	j->processed = std::move(new_processed);
+      }
+
+      std::unique_lock l(j->fifo->m);
+      auto objv = j->fifo->info.version;
+      if (j->new_tail > j->fifo->info.tail_part_num) {
+	tail_part_num = j->new_tail;
+      }
+
+      if (j->new_head > j->fifo->info.head_part_num) {
+	head_part_num = j->new_head;
+      }
+
+      if (j->new_max > j->fifo->info.max_push_part_num) {
+	max_part_num = j->new_max;
+      }
+      l.unlock();
+
+      if (j->processed.empty() &&
+	  !tail_part_num &&
+	  !max_part_num) {
+	/* nothing to update anymore */
+	j->handle({});
+	return;
+      }
+      auto a = ba::get_associated_allocator(j->handler);
+      auto e = ba::get_associated_executor(j->handler, j->fifo->get_executor());
+      j->fifo->_update_meta(fifo::update{}
+			    .tail_part_num(tail_part_num)
+		            .head_part_num(head_part_num)
+		            .max_push_part_num(max_part_num)
+			    .journal_entries_rm(j->processed),
+                            objv,
+                            ca::bind_ea(
+			      e, a,
+			      JournalPostprocessor{j, false}));
+      return;
+    }
+
+    JournalPostprocessor(JournalProcessor* j, bool first)
+      : j_(j), first(first) {}
+  };
+
+  void postprocess() {
+    if (processed.empty()) {
+      handle({});
+      return;
+    }
+    JournalPostprocessor(this, true)({}, false);
+  }
+
+  void handle(bs::error_code ec) {
+    auto e = ba::get_associated_executor(handler, fifo->get_executor());
+    auto a = ba::get_associated_allocator(handler);
+    auto h = std::move(handler);
+    FIFO::assoc_delete(h, this);
+    e.dispatch(ca::bind_handler(std::move(h), ec), a);
+    return;
+  }
+
+public:
+
+  JournalProcessor(FIFO* fifo, Handler&& handler)
+    : fifo(fifo), handler(std::move(handler)) {
+    std::unique_lock l(fifo->m);
+    journal = fifo->info.journal;
+    iter = journal.begin();
+    new_tail = fifo->info.tail_part_num;
+    new_head = fifo->info.head_part_num;
+    new_max = fifo->info.max_push_part_num;
+  }
+
+  JournalProcessor(const JournalProcessor&) = delete;
+  JournalProcessor& operator =(const JournalProcessor&) = delete;
+  JournalProcessor(JournalProcessor&&) = delete;
+  JournalProcessor& operator =(JournalProcessor&&) = delete;
+
+  void process() {
+    if (iter != journal.end()) {
+      const auto entry = iter->second;
+      process_journal_entry(entry,
+			    journal_entry_finisher(entry));
+      return;
+    } else {
+      postprocess();
+      return;
+    }
+  }
+};
+}
+}
+
+#endif // CEPH_RADOS_CLS_FIFIO_H