Merge pull request #35964 from dillaman/wip-neorados-compat

neorados: bridge support to librados cluster connections

Reviewed-by: Adam Emersen <aemerson@redhat.com>

src/include/neorados/RADOS.hpp

@@ -55,6 +55,7 @@
 #include "include/common_fwd.h"
 
 #include "include/buffer.h"
+#include "include/rados/librados_fwd.hpp"
 
 #include "common/ceph_time.h"
 
@@ -73,7 +74,7 @@ struct hash<neorados::IOContext>;
 
 namespace neorados {
 namespace detail {
-class RADOS;
+class Client;
 }
 
 class RADOS;
@@ -531,6 +532,8 @@ public:
     return init.result.get();
   }
 
+  static RADOS make_with_librados(librados::Rados& rados);
+
   RADOS(const RADOS&) = delete;
   RADOS& operator =(const RADOS&) = delete;
 
@@ -944,7 +947,7 @@ private:
 
   friend Builder;
 
-  RADOS(std::unique_ptr<detail::RADOS> impl);
+  RADOS(std::unique_ptr<detail::Client> impl);
   static void make_with_cct(CephContext* cct,
 			    boost::asio::io_context& ioctx,
 		    std::unique_ptr<BuildComp> c);
@@ -1068,9 +1071,8 @@ private:
 			  bool force, std::unique_ptr<SimpleOpComp> c);
 
 
-  // Since detail::RADOS has immovable things inside it, hold a
-  // unique_ptr to it so we can be moved.
-  std::unique_ptr<detail::RADOS> impl;
+  // Proxy object to provide access to low-level RADOS messaging clients
+  std::unique_ptr<detail::Client> impl;
 };
 
 enum class errc {

src/include/rados/librados.hpp

@@ -19,6 +19,8 @@ namespace libradosstriper
   class RadosStriper;
 }
 
+namespace neorados { class RADOS; }
+
 namespace librados {
 
 using ceph::bufferlist;
@@ -1503,9 +1505,11 @@ inline namespace v14_2_0 {
 						callback_t cb_safe)
       __attribute__ ((deprecated));
     static AioCompletion *aio_create_completion(void *cb_arg, callback_t cb_complete);
-    
+
     friend std::ostream& operator<<(std::ostream &oss, const Rados& r);
   private:
+    friend class neorados::RADOS;
+
     // We don't allow assignment or copying
     Rados(const Rados& rhs);
     const Rados& operator=(const Rados& rhs);

src/librados/RadosClient.h

@@ -40,9 +40,12 @@ class MLog;
 class Messenger;
 class AioCompletionImpl;
 
+namespace neorados { namespace detail { class RadosClient; }}
+
 class librados::RadosClient : public Dispatcher,
 			      public md_config_obs_t
 {
+  friend neorados::detail::RadosClient;
 public:
   using Dispatcher::cct;
 private:

src/neorados/RADOS.cc

@@ -748,9 +748,9 @@ void RADOS::make_with_cct(CephContext* cct,
 			  boost::asio::io_context& ioctx,
 			  std::unique_ptr<BuildComp> c) {
   try {
-    auto r = new detail::RADOS(ioctx, cct);
+    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::RADOS>(r)]() mutable {
+      [c = std::move(c), r = std::unique_ptr<detail::Client>(r)]() mutable {
 	c->dispatch(std::move(c), bs::error_code{},
 		    RADOS{std::move(r)});
       });
@@ -759,10 +759,13 @@ void RADOS::make_with_cct(CephContext* cct,
   }
 }
 
+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::RADOS> impl)
+RADOS::RADOS(std::unique_ptr<detail::Client> impl)
   : impl(std::move(impl)) {}
 
 RADOS::RADOS(RADOS&&) = default;
@@ -873,7 +876,7 @@ void RADOS::lookup_pool(std::string_view name,
   if (ret < 0) {
     impl->objecter->wait_for_latest_osdmap(
       [name = std::string(name), c = std::move(c),
-       objecter = impl->objecter.get()]
+       objecter = impl->objecter]
       (bs::error_code ec) mutable {
 	int64_t ret =
 	  objecter->with_osdmap(std::mem_fn(&OSDMap::lookup_pg_pool_name),
@@ -1175,7 +1178,7 @@ void RADOS::unwatch(uint64_t cookie, const IOContext& _ioc,
 			 ioc->snapc, ceph::real_clock::now(), 0,
 			 Objecter::Op::OpComp::create(
 			   get_executor(),
-			   [objecter = impl->objecter.get(),
+			   [objecter = impl->objecter,
 			    linger_op, c = std::move(c)]
 			   (bs::error_code ec) mutable {
 			     objecter->linger_cancel(linger_op);
@@ -1203,7 +1206,7 @@ void RADOS::unwatch(uint64_t cookie, std::int64_t pool,
 			 {}, ceph::real_clock::now(), 0,
 			 Objecter::Op::OpComp::create(
 			   get_executor(),
-			   [objecter = impl->objecter.get(),
+			   [objecter = impl->objecter,
 			    linger_op, c = std::move(c)]
 			   (bs::error_code ec) mutable {
 			     objecter->linger_cancel(linger_op);
@@ -1279,7 +1282,7 @@ void RADOS::notify(const Object& o, const IOContext& _ioc, bufferlist&& bl,
   auto ioc = reinterpret_cast<const IOContextImpl*>(&_ioc.impl);
   auto linger_op = impl->objecter->linger_register(*oid, ioc->oloc, 0);
 
-  auto cb = std::make_shared<NotifyHandler>(impl->ioctx, impl->objecter.get(),
+  auto cb = std::make_shared<NotifyHandler>(impl->ioctx, impl->objecter,
                                             linger_op, std::move(c));
   linger_op->on_notify_finish =
     Objecter::LingerOp::OpComp::create(
@@ -1318,7 +1321,7 @@ void RADOS::notify(const Object& o, std::int64_t pool, bufferlist&& bl,
     oloc.key = *key;
   auto linger_op = impl->objecter->linger_register(*oid, oloc, 0);
 
-  auto cb = std::make_shared<NotifyHandler>(impl->ioctx, impl->objecter.get(),
+  auto cb = std::make_shared<NotifyHandler>(impl->ioctx, impl->objecter,
                                             linger_op, std::move(c));
   linger_op->on_notify_finish =
     Objecter::LingerOp::OpComp::create(
@@ -1637,7 +1640,7 @@ const bs::error_category& error_category() noexcept {
 }
 
 CephContext* RADOS::cct() {
-  return impl->cct;
+  return impl->cct.get();
 }
 }
 

src/neorados/RADOSImpl.cc

@@ -25,18 +25,17 @@ namespace neorados {
 namespace detail {
 
 RADOS::RADOS(boost::asio::io_context& ioctx,
-	     boost::intrusive_ptr<CephContext> _cct)
-  : Dispatcher(_cct.detach()),
+	     boost::intrusive_ptr<CephContext> cct)
+  : Dispatcher(cct.get()),
     ioctx(ioctx),
-    monclient(cct, ioctx),
-    moncsd(monclient),
-    mgrclient(cct, nullptr, &monclient.monmap),
-    mgrcsd(mgrclient) {
+    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, "radosclient"));
+  messenger.reset(Messenger::create_client_messenger(cct.get(), "radosclient"));
   if (!messenger)
     throw std::bad_alloc();
 
@@ -46,7 +45,7 @@ RADOS::RADOS(boost::asio::io_context& ioctx,
   messenger->set_default_policy(
     Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX));
 
-  objecter.reset(new Objecter(cct, messenger.get(), &monclient,
+  objecter.reset(new Objecter(cct.get(), messenger.get(), &monclient,
 			      ioctx,
 			      cct->_conf->rados_mon_op_timeout,
 			      cct->_conf->rados_osd_op_timeout));
@@ -87,6 +86,20 @@ RADOS::RADOS(boost::asio::io_context& ioctx,
   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()) {
@@ -107,6 +120,5 @@ bool RADOS::ms_handle_refused(Connection *con) {
   return false;
 }
 
-RADOS::~RADOS() = default;
-}
-}
+} // namespace detail
+} // namespace neorados

src/neorados/RADOSImpl.h

@@ -11,8 +11,8 @@
  * Foundation.  See file COPYING.
  *
  */
-#ifndef CEPH_LIBRADOS_RADOSCLIENT_H
-#define CEPH_LIBRADOS_RADOSCLIENT_H
+#ifndef CEPH_NEORADOS_RADOSIMPL_H
+#define CEPH_NEORADOS_RADOSIMPL_H
 
 #include <functional>
 #include <memory>
@@ -24,6 +24,8 @@
 #include "common/ceph_context.h"
 #include "common/ceph_mutex.h"
 
+#include "librados/RadosClient.h"
+
 #include "mon/MonClient.h"
 
 #include "mgr/MgrClient.h"
@@ -31,55 +33,30 @@
 #include "osdc/Objecter.h"
 
 namespace neorados {
-  class RADOS;
+
+class RADOS;
+
 namespace detail {
 
+class NeoClient;
+
 class RADOS : public Dispatcher
 {
   friend ::neorados::RADOS;
-  struct MsgDeleter {
-    void operator()(Messenger* p) const {
-      if (p) {
-	p->shutdown();
-	p->wait();
-      }
-      delete p;
-    }
-  };
-
-  struct ObjDeleter {
-    void operator()(Objecter* p) const {
-      if (p) {
-	p->shutdown();
-      }
-      delete p;
-    }
-  };
-
-  template<typename T>
-  struct scoped_shutdown {
-    T& m;
-    scoped_shutdown(T& m) : m(m) {}
-
-    ~scoped_shutdown() {
-      m.shutdown();
-    }
-  };
+  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, MsgDeleter> messenger;
+  std::unique_ptr<Messenger> messenger;
 
   MonClient monclient;
-  scoped_shutdown<MonClient> moncsd;
-
   MgrClient mgrclient;
-  scoped_shutdown<MgrClient> mgrcsd;
-
-  std::unique_ptr<Objecter, ObjDeleter> objecter;
 
+  std::unique_ptr<Objecter> objecter;
 
 public:
 
@@ -93,11 +70,66 @@ public:
   mon_feature_t get_required_monitor_features() const {
     return monclient.with_monmap(std::mem_fn(&MonMap::get_required_features));
   }
-  int get_instance_id() const {
-    return instance_id;
-  }
 };
-}
-}
+
+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

src/test/neorados/CMakeLists.txt

@@ -1,3 +1,12 @@
+
+add_executable(ceph_test_neorados test_neorados.cc)
+target_link_libraries(ceph_test_neorados global libneorados
+  ${unittest_libs}
+  radostest
+  radostest-cxx
+  librados
+  GTest::GTest)
+
 add_executable(ceph_test_neorados_start_stop start_stop.cc)
 target_link_libraries(ceph_test_neorados_start_stop global libneorados
   ${unittest_libs})

src/test/neorados/test_neorados.cc

@@ -0,0 +1,47 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "include/rados/librados.hpp"
+#include "include/neorados/RADOS.hpp"
+#include "common/async/blocked_completion.h"
+#include "test/librados/test_cxx.h"
+#include "gtest/gtest.h"
+#include <iostream>
+
+namespace neorados {
+
+class TestNeoRADOS : public ::testing::Test {
+public:
+  TestNeoRADOS() {
+  }
+};
+
+TEST_F(TestNeoRADOS, MakeWithLibRADOS) {
+  librados::Rados paleo_rados;
+  auto result = connect_cluster_pp(paleo_rados);
+  ASSERT_EQ("", result);
+
+  auto rados = RADOS::make_with_librados(paleo_rados);
+
+  ReadOp op;
+  bufferlist bl;
+  op.read(0, 0, &bl);
+
+  // provide pool that doesn't exists -- just testing round-trip
+  ASSERT_THROW(
+    rados.execute({"dummy-obj"}, std::numeric_limits<int64_t>::max(),
+                  std::move(op), nullptr, ceph::async::use_blocked),
+    boost::system::system_error);
+}
+
+} // namespace neorados
+
+int main(int argc, char **argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+
+  int seed = getpid();
+  std::cout << "seed " << seed << std::endl;
+  srand(seed);
+
+  return RUN_ALL_TESTS();
+}