crimson/os: support multicores for cyanstore

Signed-off-by: chunmei-liu <chunmei.liu@intel.com>
Signed-off-by: Yingxin Cheng <yingxin.cheng@intel.com>

2 files changed, 406 insertions, 178 deletions

diff --git a/src/crimson/os/cyanstore/cyan_store.cc b/src/crimson/os/cyanstore/cyan_store.cc
index a1d93e8dd2e..5336f1eb971 100644
--- a/src/crimson/os/cyanstore/cyan_store.cc
+++ b/src/crimson/os/cyanstore/cyan_store.cc
@@ -53,11 +53,98 @@ private:
   };
 };
 
-CyanStore::mount_ertr::future<> CyanStore::mount()
+seastar::future<> CyanStore::start()
+{
+  return shard_stores.start(path);
+}
+
+seastar::future<store_statfs_t> CyanStore::stat() const
+{
+  logger().debug("{}", __func__);
+  return shard_stores.map_reduce0(
+    [](const CyanStore::ShardStores &local_store) {
+      return local_store.get_used_bytes();
+    },
+    (uint64_t)0,
+    std::plus<uint64_t>()
+  ).then([](uint64_t used_bytes) {
+    store_statfs_t st;
+    st.total = crimson::common::local_conf().get_val<Option::size_t>("memstore_device_bytes");
+    st.available = st.total - used_bytes;
+    return seastar::make_ready_future<store_statfs_t>(std::move(st));
+  });
+}
+
+
+CyanStore::mkfs_ertr::future<> CyanStore::mkfs(uuid_d new_osd_fsid)
+{
+  static const char read_meta_errmsg[]{"read_meta"};
+  static const char parse_fsid_errmsg[]{"failed to parse fsid"};
+  static const char match_ofsid_errmsg[]{"unmatched osd_fsid"};
+  return read_meta("fsid").then([=, this](auto&& ret) -> mkfs_ertr::future<> {
+    auto& [r, fsid_str] = ret;
+    if (r == -ENOENT) {
+      if (new_osd_fsid.is_zero()) {
+        osd_fsid.generate_random();
+      } else {
+        osd_fsid = new_osd_fsid;
+      }
+      return write_meta("fsid", fmt::format("{}", osd_fsid));
+    } else if (r < 0) {
+      return crimson::stateful_ec{ singleton_ec<read_meta_errmsg>() };
+    } else {
+      logger().info("mkfs already has fsid {}", fsid_str);
+      if (!osd_fsid.parse(fsid_str.c_str())) {
+        return crimson::stateful_ec{ singleton_ec<parse_fsid_errmsg>() };
+      } else if (osd_fsid != new_osd_fsid) {
+        logger().error("on-disk fsid {} != provided {}", osd_fsid, new_osd_fsid);
+        return crimson::stateful_ec{ singleton_ec<match_ofsid_errmsg>() };
+      } else {
+        return mkfs_ertr::now();
+      }
+    }
+  }).safe_then([this]{
+    return write_meta("type", "memstore");
+  }).safe_then([this] {
+    return shard_stores.invoke_on_all(
+      [](auto &local_store) {
+      return local_store.mkfs();
+    });
+  });
+}
+
+seastar::future<> CyanStore::ShardStores::mkfs()
+{
+  std::string fn =
+    path + "/collections" + std::to_string(seastar::this_shard_id());
+  ceph::bufferlist bl;
+  std::set<coll_t> collections;
+  ceph::encode(collections, bl);
+  return crimson::write_file(std::move(bl), fn);
+}
+
+seastar::future<std::vector<coll_t>>
+CyanStore::list_collections()
+{
+  return seastar::do_with(std::vector<coll_t>{}, [this](auto &collections) {
+    return shard_stores.map([](auto &local_store) {
+      return local_store.list_collections();
+    }).then([&collections](std::vector<std::vector<coll_t>> results) {
+      for (auto& colls : results) {
+        collections.insert(collections.end(), colls.begin(), colls.end());
+      }
+      return seastar::make_ready_future<std::vector<coll_t>>(
+        std::move(collections));
+    });
+  });
+}
+
+CyanStore::mount_ertr::future<> CyanStore::ShardStores::mount()
 {
   static const char read_file_errmsg[]{"read_file"};
   ceph::bufferlist bl;
-  std::string fn = path + "/collections";
+  std::string fn =
+    path + "/collections" + std::to_string(seastar::this_shard_id());
   std::string err;
   if (int r = bl.read_file(fn.c_str(), &err); r < 0) {
     return crimson::stateful_ec{ singleton_ec<read_file_errmsg>() };
@@ -68,7 +155,8 @@ CyanStore::mount_ertr::future<> CyanStore::mount()
   ceph::decode(collections, p);
 
   for (auto& coll : collections) {
-    std::string fn = fmt::format("{}/{}", path, coll);
+    std::string fn = fmt::format("{}/{}{}", path, coll,
+      std::to_string(seastar::this_shard_id()));
     ceph::bufferlist cbl;
     if (int r = cbl.read_file(fn.c_str(), &err); r < 0) {
       return crimson::stateful_ec{ singleton_ec<read_file_errmsg>() };
@@ -82,7 +170,7 @@ CyanStore::mount_ertr::future<> CyanStore::mount()
   return mount_ertr::now();
 }
 
-seastar::future<> CyanStore::umount()
+seastar::future<> CyanStore::ShardStores::umount()
 {
   return seastar::do_with(std::set<coll_t>{}, [this](auto& collections) {
     return seastar::do_for_each(coll_map, [&collections, this](auto& coll) {
@@ -91,69 +179,25 @@ seastar::future<> CyanStore::umount()
       ceph::bufferlist bl;
       ceph_assert(ch);
       ch->encode(bl);
-      std::string fn = fmt::format("{}/{}", path, col);
+      std::string fn = fmt::format("{}/{}{}", path, col,
+        std::to_string(seastar::this_shard_id()));
       return crimson::write_file(std::move(bl), fn);
     }).then([&collections, this] {
       ceph::bufferlist bl;
       ceph::encode(collections, bl);
-      std::string fn = fmt::format("{}/collections", path);
+      std::string fn = fmt::format("{}/collections{}",
+        path, std::to_string(seastar::this_shard_id()));
       return crimson::write_file(std::move(bl), fn);
     });
   });
 }
 
-CyanStore::mkfs_ertr::future<> CyanStore::mkfs(uuid_d new_osd_fsid)
-{
-  static const char read_meta_errmsg[]{"read_meta"};
-  static const char parse_fsid_errmsg[]{"failed to parse fsid"};
-  static const char match_ofsid_errmsg[]{"unmatched osd_fsid"};
-  return read_meta("fsid").then([=, this](auto&& ret) -> mkfs_ertr::future<> {
-    auto& [r, fsid_str] = ret;
-    if (r == -ENOENT) {
-      if (new_osd_fsid.is_zero()) {
-        osd_fsid.generate_random();
-      } else {
-        osd_fsid = new_osd_fsid;
-      }
-      return write_meta("fsid", fmt::format("{}", osd_fsid));
-    } else if (r < 0) {
-      return crimson::stateful_ec{ singleton_ec<read_meta_errmsg>() };
-    } else {
-      logger().info("mkfs already has fsid {}", fsid_str);
-      if (!osd_fsid.parse(fsid_str.c_str())) {
-        return crimson::stateful_ec{ singleton_ec<parse_fsid_errmsg>() };
-      } else if (osd_fsid != new_osd_fsid) {
-        logger().error("on-disk fsid {} != provided {}", osd_fsid, new_osd_fsid);
-        return crimson::stateful_ec{ singleton_ec<match_ofsid_errmsg>() };
-      } else {
-	return mkfs_ertr::now();
-      }
-    }
-  }).safe_then([this]{
-    std::string fn = path + "/collections";
-    ceph::bufferlist bl;
-    std::set<coll_t> collections;
-    ceph::encode(collections, bl);
-    return crimson::write_file(std::move(bl), fn);
-  }).safe_then([this] {
-    return write_meta("type", "memstore");
-  });
-}
-
-seastar::future<store_statfs_t> CyanStore::stat() const
-{
-  logger().debug("{}", __func__);
-  store_statfs_t st;
-  st.total = crimson::common::local_conf().get_val<Option::size_t>("memstore_device_bytes");
-  st.available = st.total - used_bytes;
-  return seastar::make_ready_future<store_statfs_t>(std::move(st));
-}
-
 seastar::future<std::tuple<std::vector<ghobject_t>, ghobject_t>>
-CyanStore::list_objects(CollectionRef ch,
-                        const ghobject_t& start,
-                        const ghobject_t& end,
-                        uint64_t limit) const
+CyanStore::ShardStores::list_objects(
+  CollectionRef ch,
+  const ghobject_t& start,
+  const ghobject_t& end,
+  uint64_t limit) const
 {
   auto c = static_cast<Collection*>(ch.get());
   logger().debug("{} {} {} {} {}",
@@ -175,19 +219,22 @@ CyanStore::list_objects(CollectionRef ch,
     std::make_tuple(std::move(objects), next));
 }
 
-seastar::future<CollectionRef> CyanStore::create_new_collection(const coll_t& cid)
+seastar::future<CollectionRef>
+CyanStore::ShardStores::create_new_collection(const coll_t& cid)
 {
   auto c = new Collection{cid};
   new_coll_map[cid] = c;
   return seastar::make_ready_future<CollectionRef>(c);
 }
 
-seastar::future<CollectionRef> CyanStore::open_collection(const coll_t& cid)
+seastar::future<CollectionRef>
+CyanStore::ShardStores::open_collection(const coll_t& cid)
 {
   return seastar::make_ready_future<CollectionRef>(_get_collection(cid));
 }
 
-seastar::future<std::vector<coll_t>> CyanStore::list_collections()
+seastar::future<std::vector<coll_t>>
+CyanStore::ShardStores::list_collections()
 {
   std::vector<coll_t> collections;
   for (auto& coll : coll_map) {
@@ -196,7 +243,8 @@ seastar::future<std::vector<coll_t>> CyanStore::list_collections()
   return seastar::make_ready_future<std::vector<coll_t>>(std::move(collections));
 }
 
-CyanStore::read_errorator::future<ceph::bufferlist> CyanStore::read(
+CyanStore::read_errorator::future<ceph::bufferlist>
+CyanStore::ShardStores::read(
   CollectionRef ch,
   const ghobject_t& oid,
   uint64_t offset,
@@ -223,7 +271,8 @@ CyanStore::read_errorator::future<ceph::bufferlist> CyanStore::read(
   return read_errorator::make_ready_future<ceph::bufferlist>(o->read(offset, l));
 }
 
-CyanStore::read_errorator::future<ceph::bufferlist> CyanStore::readv(
+CyanStore::read_errorator::future<ceph::bufferlist>
+CyanStore::ShardStores::readv(
   CollectionRef ch,
   const ghobject_t& oid,
   interval_set<uint64_t>& m,
@@ -243,8 +292,8 @@ CyanStore::read_errorator::future<ceph::bufferlist> CyanStore::readv(
   });
 }
 
-
-CyanStore::get_attr_errorator::future<ceph::bufferlist> CyanStore::get_attr(
+CyanStore::get_attr_errorator::future<ceph::bufferlist>
+CyanStore::ShardStores::get_attr(
   CollectionRef ch,
   const ghobject_t& oid,
   std::string_view name) const
@@ -263,7 +312,8 @@ CyanStore::get_attr_errorator::future<ceph::bufferlist> CyanStore::get_attr(
   }
 }
 
-CyanStore::get_attrs_ertr::future<CyanStore::attrs_t> CyanStore::get_attrs(
+CyanStore::get_attrs_ertr::future<CyanStore::attrs_t>
+CyanStore::ShardStores::get_attrs(
   CollectionRef ch,
   const ghobject_t& oid)
 {
@@ -277,9 +327,10 @@ CyanStore::get_attrs_ertr::future<CyanStore::attrs_t> CyanStore::get_attrs(
   return get_attrs_ertr::make_ready_future<attrs_t>(o->xattr);
 }
 
-auto CyanStore::omap_get_values(CollectionRef ch,
-				const ghobject_t& oid,
-				const omap_keys_t& keys)
+auto CyanStore::ShardStores::omap_get_values(
+  CollectionRef ch,
+  const ghobject_t& oid,
+  const omap_keys_t& keys)
   -> read_errorator::future<omap_values_t>
 {
   auto c = static_cast<Collection*>(ch.get());
@@ -297,10 +348,10 @@ auto CyanStore::omap_get_values(CollectionRef ch,
   return seastar::make_ready_future<omap_values_t>(std::move(values));
 }
 
-auto
-CyanStore::omap_get_values(CollectionRef ch,
-			   const ghobject_t &oid,
-			   const std::optional<string> &start)
+auto CyanStore::ShardStores::omap_get_values(
+  CollectionRef ch,
+  const ghobject_t &oid,
+  const std::optional<string> &start)
   -> read_errorator::future<std::tuple<bool, omap_values_t>>
 {
   auto c = static_cast<Collection*>(ch.get());
@@ -319,9 +370,9 @@ CyanStore::omap_get_values(CollectionRef ch,
     std::make_tuple(true, std::move(values)));
 }
 
-auto
-CyanStore::omap_get_header(CollectionRef ch,
-			   const ghobject_t& oid)
+auto CyanStore::ShardStores::omap_get_header(
+  CollectionRef ch,
+  const ghobject_t& oid)
   -> get_attr_errorator::future<ceph::bufferlist>
 {
   auto c = static_cast<Collection*>(ch.get());
@@ -334,7 +385,7 @@ CyanStore::omap_get_header(CollectionRef ch,
     o->omap_header);
 }
 
-seastar::future<> CyanStore::do_transaction_no_callbacks(
+seastar::future<> CyanStore::ShardStores::do_transaction_no_callbacks(
   CollectionRef ch,
   ceph::os::Transaction&& t)
 {
@@ -526,7 +577,7 @@ seastar::future<> CyanStore::do_transaction_no_callbacks(
   return seastar::now();
 }
 
-int CyanStore::_remove(const coll_t& cid, const ghobject_t& oid)
+int CyanStore::ShardStores::_remove(const coll_t& cid, const ghobject_t& oid)
 {
   logger().debug("{} cid={} oid={}",
                 __func__, cid, oid);
@@ -543,7 +594,7 @@ int CyanStore::_remove(const coll_t& cid, const ghobject_t& oid)
   return 0;
 }
 
-int CyanStore::_touch(const coll_t& cid, const ghobject_t& oid)
+int CyanStore::ShardStores::_touch(const coll_t& cid, const ghobject_t& oid)
 {
   logger().debug("{} cid={} oid={}",
                 __func__, cid, oid);
@@ -555,9 +606,13 @@ int CyanStore::_touch(const coll_t& cid, const ghobject_t& oid)
   return 0;
 }
 
-int CyanStore::_write(const coll_t& cid, const ghobject_t& oid,
-                      uint64_t offset, size_t len, const ceph::bufferlist& bl,
-                      uint32_t fadvise_flags)
+int CyanStore::ShardStores::_write(
+  const coll_t& cid,
+  const ghobject_t& oid,
+  uint64_t offset,
+  size_t len,
+  const ceph::bufferlist& bl,
+  uint32_t fadvise_flags)
 {
   logger().debug("{} {} {} {} ~ {}",
                 __func__, cid, oid, offset, len);
@@ -577,8 +632,11 @@ int CyanStore::_write(const coll_t& cid, const ghobject_t& oid,
   return 0;
 }
 
-int CyanStore::_zero(const coll_t& cid, const ghobject_t& oid,
-                     uint64_t offset, size_t len)
+int CyanStore::ShardStores::_zero(
+  const coll_t& cid,
+  const ghobject_t& oid,
+  uint64_t offset,
+  size_t len)
 {
   logger().debug("{} {} {} {} ~ {}",
                 __func__, cid, oid, offset, len);
@@ -588,7 +646,7 @@ int CyanStore::_zero(const coll_t& cid, const ghobject_t& oid,
   return _write(cid, oid, offset, len, bl, 0);
 }
 
-int CyanStore::_omap_clear(
+int CyanStore::ShardStores::_omap_clear(
   const coll_t& cid,
   const ghobject_t& oid)
 {
@@ -607,7 +665,7 @@ int CyanStore::_omap_clear(
   return 0;
 }
 
-int CyanStore::_omap_set_values(
+int CyanStore::ShardStores::_omap_set_values(
   const coll_t& cid,
   const ghobject_t& oid,
   std::map<std::string, ceph::bufferlist> &&aset)
@@ -627,7 +685,7 @@ int CyanStore::_omap_set_values(
   return 0;
 }
 
-int CyanStore::_omap_set_header(
+int CyanStore::ShardStores::_omap_set_header(
   const coll_t& cid,
   const ghobject_t& oid,
   const ceph::bufferlist &header)
@@ -645,7 +703,7 @@ int CyanStore::_omap_set_header(
   return 0;
 }
 
-int CyanStore::_omap_rmkeys(
+int CyanStore::ShardStores::_omap_rmkeys(
   const coll_t& cid,
   const ghobject_t& oid,
   const omap_keys_t& aset)
@@ -665,7 +723,7 @@ int CyanStore::_omap_rmkeys(
   return 0;
 }
 
-int CyanStore::_omap_rmkeyrange(
+int CyanStore::ShardStores::_omap_rmkeyrange(
   const coll_t& cid,
   const ghobject_t& oid,
   const std::string &first,
@@ -686,7 +744,10 @@ int CyanStore::_omap_rmkeyrange(
   return 0;
 }
 
-int CyanStore::_truncate(const coll_t& cid, const ghobject_t& oid, uint64_t size)
+int CyanStore::ShardStores::_truncate(
+  const coll_t& cid,
+  const ghobject_t& oid,
+  uint64_t size)
 {
   logger().debug("{} cid={} oid={} size={}",
                 __func__, cid, oid, size);
@@ -705,8 +766,10 @@ int CyanStore::_truncate(const coll_t& cid, const ghobject_t& oid, uint64_t size
   return r;
 }
 
-int CyanStore::_clone(const coll_t& cid, const ghobject_t& oid,
-                      const ghobject_t& noid)
+int CyanStore::ShardStores::_clone(
+  const coll_t& cid,
+  const ghobject_t& oid,
+  const ghobject_t& noid)
 {
   logger().debug("{} cid={} oid={} noid={}",
                 __func__, cid, oid, noid);
@@ -729,8 +792,10 @@ int CyanStore::_clone(const coll_t& cid, const ghobject_t& oid,
   return 0;
 }
 
-int CyanStore::_setattrs(const coll_t& cid, const ghobject_t& oid,
-                         std::map<std::string,bufferlist>&& aset)
+int CyanStore::ShardStores::_setattrs(
+  const coll_t& cid,
+  const ghobject_t& oid,
+  std::map<std::string,bufferlist>&& aset)
 {
   logger().debug("{} cid={} oid={}",
                 __func__, cid, oid);
@@ -747,8 +812,10 @@ int CyanStore::_setattrs(const coll_t& cid, const ghobject_t& oid,
   return 0;
 }
 
-int CyanStore::_rm_attr(const coll_t& cid, const ghobject_t& oid,
-			std::string_view name)
+int CyanStore::ShardStores::_rm_attr(
+  const coll_t& cid,
+  const ghobject_t& oid,
+  std::string_view name)
 {
   logger().debug("{} cid={} oid={} name={}", __func__, cid, oid, name);
   auto c = _get_collection(cid);
@@ -767,7 +834,9 @@ int CyanStore::_rm_attr(const coll_t& cid, const ghobject_t& oid,
   return 0;
 }
 
-int CyanStore::_rm_attrs(const coll_t& cid, const ghobject_t& oid)
+int CyanStore::ShardStores::_rm_attrs(
+  const coll_t& cid,
+  const ghobject_t& oid)
 {
   logger().debug("{} cid={} oid={}", __func__, cid, oid);
   auto c = _get_collection(cid);
@@ -782,7 +851,7 @@ int CyanStore::_rm_attrs(const coll_t& cid, const ghobject_t& oid)
   return 0;
 }
 
-int CyanStore::_create_collection(const coll_t& cid, int bits)
+int CyanStore::ShardStores::_create_collection(const coll_t& cid, int bits)
 {
   auto result = coll_map.try_emplace(cid);
   if (!result.second)
@@ -795,7 +864,8 @@ int CyanStore::_create_collection(const coll_t& cid, int bits)
   return 0;
 }
 
-boost::intrusive_ptr<Collection> CyanStore::_get_collection(const coll_t& cid)
+boost::intrusive_ptr<Collection>
+CyanStore::ShardStores::_get_collection(const coll_t& cid)
 {
   auto cp = coll_map.find(cid);
   if (cp == coll_map.end())
@@ -803,8 +873,9 @@ boost::intrusive_ptr<Collection> CyanStore::_get_collection(const coll_t& cid)
   return cp->second;
 }
 
-seastar::future<> CyanStore::write_meta(const std::string& key,
-					const std::string& value)
+seastar::future<> CyanStore::write_meta(
+  const std::string& key,
+  const std::string& value)
 {
   std::string v = value;
   v += "\n";
@@ -845,11 +916,11 @@ unsigned CyanStore::get_max_attr_name_length() const
 }
 
 CyanStore::read_errorator::future<std::map<uint64_t, uint64_t>>
-CyanStore::fiemap(
-    CollectionRef ch,
-    const ghobject_t& oid,
-    uint64_t off,
-    uint64_t len)
+CyanStore::ShardStores::fiemap(
+  CollectionRef ch,
+  const ghobject_t& oid,
+  uint64_t off,
+  uint64_t len)
 {
   auto c = static_cast<Collection*>(ch.get());
 
@@ -862,7 +933,7 @@ CyanStore::fiemap(
 }
 
 seastar::future<struct stat>
-CyanStore::stat(
+CyanStore::ShardStores::stat(
   CollectionRef ch,
   const ghobject_t& oid)
 {
diff --git a/src/crimson/os/cyanstore/cyan_store.h b/src/crimson/os/cyanstore/cyan_store.h
index 1c8c6f6d01c..738157e8a85 100644
--- a/src/crimson/os/cyanstore/cyan_store.h
+++ b/src/crimson/os/cyanstore/cyan_store.h
@@ -17,6 +17,7 @@
 #include "include/uuid.h"
 
 #include "crimson/os/cyanstore/cyan_object.h"
+#include "crimson/os/cyanstore/cyan_collection.h"
 #include "crimson/os/futurized_store.h"
 
 namespace ceph::os {
@@ -24,131 +25,287 @@ class Transaction;
 }
 
 namespace crimson::os {
-class Collection;
 
 class CyanStore final : public FuturizedStore {
-  const std::string path;
-  std::unordered_map<coll_t, boost::intrusive_ptr<Collection>> coll_map;
-  std::map<coll_t, boost::intrusive_ptr<Collection>> new_coll_map;
-  uint64_t used_bytes = 0;
-  uuid_d osd_fsid;
+  class ShardStores {
+  public:
+    ShardStores(std::string path)
+      :path(path){}
+
+    mount_ertr::future<> mount();
+
+    seastar::future<> umount();
+
+    seastar::future<> mkfs();
+
+    mkfs_ertr::future<> mkcoll(uuid_d new_osd_fsid);
+
+    seastar::future<struct stat> stat(
+      CollectionRef c,
+      const ghobject_t& oid);
+
+    read_errorator::future<ceph::bufferlist> read(
+      CollectionRef c,
+      const ghobject_t& oid,
+      uint64_t offset,
+      size_t len,
+      uint32_t op_flags = 0);
+
+    read_errorator::future<ceph::bufferlist> readv(
+      CollectionRef c,
+      const ghobject_t& oid,
+      interval_set<uint64_t>& m,
+      uint32_t op_flags = 0);
+
+    get_attr_errorator::future<ceph::bufferlist> get_attr(
+      CollectionRef c,
+      const ghobject_t& oid,
+      std::string_view name) const;
+
+    get_attrs_ertr::future<attrs_t> get_attrs(
+      CollectionRef c,
+      const ghobject_t& oid);
+
+    read_errorator::future<omap_values_t> omap_get_values(
+      CollectionRef c,
+      const ghobject_t& oid,
+      const omap_keys_t& keys);
+
+    read_errorator::future<std::tuple<bool, omap_values_t>> omap_get_values(
+      CollectionRef c,           ///< [in] collection
+      const ghobject_t &oid,     ///< [in] oid
+      const std::optional<std::string> &start ///< [in] start, empty for begin
+      );
+
+    seastar::future<std::tuple<std::vector<ghobject_t>, ghobject_t>>
+    list_objects(
+      CollectionRef c,
+      const ghobject_t& start,
+      const ghobject_t& end,
+      uint64_t limit) const;
+
+    get_attr_errorator::future<ceph::bufferlist> omap_get_header(
+      CollectionRef c,
+      const ghobject_t& oid);
+
+    seastar::future<CollectionRef> create_new_collection(const coll_t& cid);
+
+    seastar::future<CollectionRef> open_collection(const coll_t& cid);
+
+    seastar::future<std::vector<coll_t>> list_collections();
+
+    seastar::future<> do_transaction_no_callbacks(
+      CollectionRef ch,
+      ceph::os::Transaction&& txn);
 
+    read_errorator::future<std::map<uint64_t, uint64_t>>
+    fiemap(
+      CollectionRef c,
+      const ghobject_t& oid,
+      uint64_t off,
+      uint64_t len);
+
+    uint64_t get_used_bytes() const { return used_bytes; }
+
+  private:
+    int _remove(const coll_t& cid, const ghobject_t& oid);
+    int _touch(const coll_t& cid, const ghobject_t& oid);
+    int _write(const coll_t& cid, const ghobject_t& oid,
+	       uint64_t offset, size_t len, const ceph::bufferlist& bl,
+	       uint32_t fadvise_flags);
+    int _zero(const coll_t& cid, const ghobject_t& oid,
+	      uint64_t offset, size_t len);
+    int _omap_clear(
+      const coll_t& cid,
+      const ghobject_t& oid);
+    int _omap_set_values(
+      const coll_t& cid,
+      const ghobject_t& oid,
+      std::map<std::string, ceph::bufferlist> &&aset);
+    int _omap_set_header(
+      const coll_t& cid,
+      const ghobject_t& oid,
+      const ceph::bufferlist &header);
+    int _omap_rmkeys(
+      const coll_t& cid,
+      const ghobject_t& oid,
+      const omap_keys_t& aset);
+    int _omap_rmkeyrange(
+      const coll_t& cid,
+      const ghobject_t& oid,
+      const std::string &first,
+      const std::string &last);
+    int _truncate(const coll_t& cid, const ghobject_t& oid, uint64_t size);
+    int _clone(const coll_t& cid, const ghobject_t& oid,
+               const ghobject_t& noid);
+    int _setattrs(const coll_t& cid, const ghobject_t& oid,
+                  std::map<std::string,bufferlist>&& aset);
+    int _rm_attr(const coll_t& cid, const ghobject_t& oid,
+                 std::string_view name);
+    int _rm_attrs(const coll_t& cid, const ghobject_t& oid);
+    int _create_collection(const coll_t& cid, int bits);
+    boost::intrusive_ptr<Collection> _get_collection(const coll_t& cid);
+
+  private:
+    uint64_t used_bytes = 0;
+    const std::string path;
+    std::unordered_map<coll_t, boost::intrusive_ptr<Collection>> coll_map;
+    std::map<coll_t, boost::intrusive_ptr<Collection>> new_coll_map;
+  };
+
+// public interfaces called by main OSD
 public:
   CyanStore(const std::string& path);
   ~CyanStore() final;
 
   seastar::future<> stop() final {
-    return seastar::now();
+    return shard_stores.stop();
   }
-  mount_ertr::future<> mount() final;
-  seastar::future<> umount() final;
+  seastar::future<> start() final;
 
   mkfs_ertr::future<> mkfs(uuid_d new_osd_fsid) final;
+
   seastar::future<store_statfs_t> stat() const final;
+
+  seastar::future<> write_meta(const std::string& key,
+		  const std::string& value) final;
+
+  seastar::future<std::tuple<int, std::string>>
+  read_meta(const std::string& key) final;
+
+  uuid_d get_fsid() const final;
+
+  mount_ertr::future<> mount() final {
+    return shard_stores.invoke_on_all(
+      [](auto &local_store) {
+      return local_store.mount().handle_error(
+      crimson::stateful_ec::handle([](const auto& ec) {
+        crimson::get_logger(ceph_subsys_cyanstore).error(
+	    "error mounting cyanstore: ({}) {}",
+            ec.value(), ec.message());
+        std::exit(EXIT_FAILURE);
+      }));
+    });
+  }
+
+  seastar::future<> umount() final {
+    return shard_stores.invoke_on_all(
+      [](auto &local_store) {
+      return local_store.umount();
+    });
+  }
+
+  seastar::future<std::vector<coll_t>> list_collections() final;
+
+// public interfaces called by each shard osd
+public:
+  unsigned get_max_attr_name_length() const final;
+
   seastar::future<struct stat> stat(
     CollectionRef c,
-    const ghobject_t& oid) final;
+    const ghobject_t& oid) final {
+    return shard_stores.local().stat(
+      c, oid);
+  }
 
   read_errorator::future<ceph::bufferlist> read(
     CollectionRef c,
     const ghobject_t& oid,
     uint64_t offset,
     size_t len,
-    uint32_t op_flags = 0) final;
+    uint32_t op_flags = 0) final {
+    return shard_stores.local().read(
+      c, oid, offset, len, op_flags);
+  }
+
   read_errorator::future<ceph::bufferlist> readv(
     CollectionRef c,
     const ghobject_t& oid,
     interval_set<uint64_t>& m,
-    uint32_t op_flags = 0) final;
+    uint32_t op_flags = 0) final {
+    return shard_stores.local().readv(
+      c, oid, m, op_flags);
+  }
 
   get_attr_errorator::future<ceph::bufferlist> get_attr(
     CollectionRef c,
     const ghobject_t& oid,
-    std::string_view name) const final;
+    std::string_view name) const final {
+    return shard_stores.local().get_attr(
+      c, oid, name);
+  }
+
   get_attrs_ertr::future<attrs_t> get_attrs(
     CollectionRef c,
-    const ghobject_t& oid);
+    const ghobject_t& oid) final {
+    return shard_stores.local().get_attrs(
+      c, oid);
+  }
 
   read_errorator::future<omap_values_t> omap_get_values(
     CollectionRef c,
     const ghobject_t& oid,
-    const omap_keys_t& keys) final;
+    const omap_keys_t& keys) final {
+    return shard_stores.local().omap_get_values(
+      c, oid, keys);
+  }
 
   /// Retrieves paged set of values > start (if present)
   read_errorator::future<std::tuple<bool, omap_values_t>> omap_get_values(
     CollectionRef c,           ///< [in] collection
     const ghobject_t &oid,     ///< [in] oid
     const std::optional<std::string> &start ///< [in] start, empty for begin
-    ) final; ///< @return <done, values> values.empty() iff done
+    ) final {
+    return shard_stores.local().omap_get_values(
+      c, oid, start);
+  }
 
-  seastar::future<std::tuple<std::vector<ghobject_t>, ghobject_t>> list_objects(
+  seastar::future<std::tuple<std::vector<ghobject_t>, ghobject_t>>
+  list_objects(
     CollectionRef c,
     const ghobject_t& start,
     const ghobject_t& end,
-    uint64_t limit) const final;
+    uint64_t limit) const final {
+    return shard_stores.local().list_objects(
+      c, start, end, limit);
+  }
 
   get_attr_errorator::future<ceph::bufferlist> omap_get_header(
     CollectionRef c,
-    const ghobject_t& oid) final;
+    const ghobject_t& oid) final {
+    return shard_stores.local().omap_get_header(
+      c, oid);
+  }
 
-  seastar::future<CollectionRef> create_new_collection(const coll_t& cid) final;
-  seastar::future<CollectionRef> open_collection(const coll_t& cid) final;
-  seastar::future<std::vector<coll_t>> list_collections() final;
+  seastar::future<CollectionRef>
+  create_new_collection(const coll_t& cid) final {
+    return shard_stores.local().create_new_collection(cid);
+  }
+
+  seastar::future<CollectionRef>
+  open_collection(const coll_t& cid) final {
+    return shard_stores.local().open_collection(cid);
+  }
 
   seastar::future<> do_transaction_no_callbacks(
     CollectionRef ch,
-    ceph::os::Transaction&& txn) final;
-
-  seastar::future<> write_meta(const std::string& key,
-		  const std::string& value) final;
-  seastar::future<std::tuple<int, std::string>>
-  read_meta(const std::string& key) final;
-  uuid_d get_fsid() const final;
-  unsigned get_max_attr_name_length() const final;
+    ceph::os::Transaction&& txn) final {
+    return shard_stores.local().do_transaction_no_callbacks(
+      ch, std::move(txn));
+  }
 
-  read_errorator::future<std::map<uint64_t, uint64_t>> fiemap(CollectionRef c,
-						       const ghobject_t& oid,
-						       uint64_t off,
-						       uint64_t len);
+  read_errorator::future<std::map<uint64_t, uint64_t>>
+  fiemap(CollectionRef c,
+         const ghobject_t& oid,
+         uint64_t off,
+         uint64_t len) final {
+    return shard_stores.local().fiemap(
+      c, oid, off, len);
+  }
 
 private:
-  int _remove(const coll_t& cid, const ghobject_t& oid);
-  int _touch(const coll_t& cid, const ghobject_t& oid);
-  int _write(const coll_t& cid, const ghobject_t& oid,
-	     uint64_t offset, size_t len, const ceph::bufferlist& bl,
-	     uint32_t fadvise_flags);
-  int _zero(const coll_t& cid, const ghobject_t& oid,
-	    uint64_t offset, size_t len);
-  int _omap_clear(
-    const coll_t& cid,
-    const ghobject_t& oid);
-  int _omap_set_values(
-    const coll_t& cid,
-    const ghobject_t& oid,
-    std::map<std::string, ceph::bufferlist> &&aset);
-  int _omap_set_header(
-    const coll_t& cid,
-    const ghobject_t& oid,
-    const ceph::bufferlist &header);
-  int _omap_rmkeys(
-    const coll_t& cid,
-    const ghobject_t& oid,
-    const omap_keys_t& aset);
-  int _omap_rmkeyrange(
-    const coll_t& cid,
-    const ghobject_t& oid,
-    const std::string &first,
-    const std::string &last);
-  int _truncate(const coll_t& cid, const ghobject_t& oid, uint64_t size);
-  int _clone(const coll_t& cid, const ghobject_t& oid,
-             const ghobject_t& noid);
-  int _setattrs(const coll_t& cid, const ghobject_t& oid,
-                std::map<std::string,bufferlist>&& aset);
-  int _rm_attr(const coll_t& cid, const ghobject_t& oid,
-               std::string_view name);
-  int _rm_attrs(const coll_t& cid, const ghobject_t& oid);
-  int _create_collection(const coll_t& cid, int bits);
-  boost::intrusive_ptr<Collection> _get_collection(const coll_t& cid);
+  const std::string path;
+  uuid_d osd_fsid;
+  seastar::sharded<ShardStores> shard_stores;
 };
-
 }