osd: refcounting chunks for snapshotted manifest object

src/cls/cas/cls_cas.cc

@@ -107,15 +107,7 @@ static int chunk_create_or_get_ref(cls_method_context_t hctx,
     CLS_LOG(10, "inc ref oid=%s\n",
 	    op.source.oid.name.c_str());
 
-    if (!objr.get(op.source)) {
-      // we could perhaps return -EEXIST here, but instead we will
-      // behave in an idempotent fashion, since we know that it is possible
-      // for references to be leaked.  If A has a ref, tries to drop it and
-      // fails (leaks), and then later tries to take it again, we should
-      // simply succeed if we know we already have it.  This will serve to
-      // silently correct the mistake.
-      return 0;
-    }
+    objr.get(op.source);
 
     ret = chunk_set_refcount(hctx, objr);
     if (ret < 0) {
@@ -148,15 +140,7 @@ static int chunk_get_ref(cls_method_context_t hctx,
   // existing chunk; inc ref
   CLS_LOG(10, "oid=%s\n", op.source.oid.name.c_str());
 
-  if (!objr.get(op.source)) {
-    // we could perhaps return -EEXIST here, but instead we will
-    // behave in an idempotent fashion, since we know that it is possible
-    // for references to be leaked.  If A has a ref, tries to drop it and
-    // fails (leaks), and then later tries to take it again, we should
-    // simply succeed if we know we already have it.  This will serve to
-    // silently correct the mistake.
-    return 0;
-  }
+  objr.get(op.source);
 
   ret = chunk_set_refcount(hctx, objr);
   if (ret < 0) {

src/cls/cas/cls_cas_internal.h

@@ -39,7 +39,7 @@ struct chunk_refs_t {
     virtual uint8_t get_type() const = 0;
     virtual bool empty() const = 0;
     virtual uint64_t count() const = 0;
-    virtual bool get(const hobject_t& o) = 0;
+    virtual void get(const hobject_t& o) = 0;
     virtual bool put(const hobject_t& o) = 0;
     virtual void dump(Formatter *f) const = 0;
     virtual std::string describe_encoding() const {
@@ -72,8 +72,8 @@ struct chunk_refs_t {
     return r->count();
   }
 
-  bool get(const hobject_t& o) {
-    return r->get(o);
+  void get(const hobject_t& o) {
+    r->get(o);
   }
   bool put(const hobject_t& o) {
     bool ret = r->put(o);
@@ -104,7 +104,7 @@ WRITE_CLASS_ENCODER(chunk_refs_t)
 // these are internal and should generally not be used directly
 
 struct chunk_refs_by_object_t : public chunk_refs_t::refs_t {
-  std::set<hobject_t> by_object;
+  std::multiset<hobject_t> by_object;
 
   uint8_t get_type() const {
     return chunk_refs_t::TYPE_BY_OBJECT;
@@ -115,12 +115,8 @@ struct chunk_refs_by_object_t : public chunk_refs_t::refs_t {
   uint64_t count() const override {
     return by_object.size();
   }
-  bool get(const hobject_t& o) override {
-    if (by_object.count(o)) {
-      return false;
-    }
+  void get(const hobject_t& o) override {
     by_object.insert(o);
-    return true;
   }
   bool put(const hobject_t& o) override {
     auto p = by_object.find(o);
@@ -199,10 +195,9 @@ struct chunk_refs_by_hash_t : public chunk_refs_t::refs_t {
   uint64_t count() const override {
     return total;
   }
-  bool get(const hobject_t& o) override {
+  void get(const hobject_t& o) override {
     by_hash[make_pair(o.pool, o.get_hash() & mask())]++;
     ++total;
-    return true;
   }
   bool put(const hobject_t& o) override {
     auto p = by_hash.find(make_pair(o.pool, o.get_hash() & mask()));
@@ -289,10 +284,9 @@ struct chunk_refs_by_pool_t : public chunk_refs_t::refs_t {
   uint64_t count() const override {
     return total;
   }
-  bool get(const hobject_t& o) override {
+  void get(const hobject_t& o) override {
     ++by_pool[o.pool];
     ++total;
-    return true;
   }
   bool put(const hobject_t& o) override {
     auto p = by_pool.find(o.pool);
@@ -367,9 +361,8 @@ struct chunk_refs_count_t : public chunk_refs_t::refs_t {
   uint64_t count() const override {
     return total;
   }
-  bool get(const hobject_t& o) override {
+  void get(const hobject_t& o) override {
     ++total;
-    return true;
   }
   bool put(const hobject_t& o) override {
     if (!total) {

src/include/rados.h

@@ -326,7 +326,7 @@ extern const char *ceph_osd_state_name(int s);
 	f(SET_CHUNK,	__CEPH_OSD_OP(WR, DATA, 40),	"set-chunk")	    \
 	f(TIER_PROMOTE,	__CEPH_OSD_OP(WR, DATA, 41),	"tier-promote")	    \
 	f(UNSET_MANIFEST, __CEPH_OSD_OP(WR, DATA, 42),	"unset-manifest")   \
-	f(TIER_FLUSH, __CEPH_OSD_OP(WR, DATA, 43),	"tier-flush")	    \
+	f(TIER_FLUSH, __CEPH_OSD_OP(CACHE, DATA, 43),	"tier-flush")	    \
 									    \
 	/** attrs **/							    \
 	/* read */							    \

src/include/rados/librados.hpp

@@ -518,7 +518,6 @@ inline namespace v14_2_0 {
                    std::string tgt_oid, uint64_t tgt_offset, int flag = 0);
     void tier_promote();
     void unset_manifest();
-    void tier_flush();
 
 
     friend class IoCtx;
@@ -733,6 +732,12 @@ inline namespace v14_2_0 {
      * triggering a promote on the OSD (that is then evicted).
      */
     void cache_evict();
+
+    /**
+     * flush a manifest tier object to backing tier; will block racing
+     * updates.
+     */
+    void tier_flush();
   };
 
   /* IoCtx : This is a context in which we can perform I/O.

src/librados/librados_cxx.cc

@@ -631,6 +631,13 @@ void librados::ObjectReadOperation::cache_evict()
   o->cache_evict();
 }
 
+void librados::ObjectReadOperation::tier_flush()
+{
+  ceph_assert(impl);
+  ::ObjectOperation *o = &impl->o;
+  o->tier_flush();
+}
+
 void librados::ObjectWriteOperation::set_redirect(const std::string& tgt_obj, 
 						  const IoCtx& tgt_ioctx,
 						  uint64_t tgt_version,
@@ -669,13 +676,6 @@ void librados::ObjectWriteOperation::unset_manifest()
   o->unset_manifest();
 }
 
-void librados::ObjectWriteOperation::tier_flush()
-{
-  ceph_assert(impl);
-  ::ObjectOperation *o = &impl->o;
-  o->tier_flush();
-}
-
 void librados::ObjectWriteOperation::tmap_update(const bufferlist& cmdbl)
 {
   ceph_assert(impl);