fix memcheck leak issue (#4308)

Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

core/src/db/snapshot/ReferenceProxy.h

@@ -18,10 +18,10 @@
 
 namespace milvus::engine::snapshot {
 
-using OnNoRefCBF = std::function<void(void)>;
-
-class ReferenceProxy {
+class ReferenceProxy : public std::enable_shared_from_this<ReferenceProxy> {
  public:
+    using Ptr = std::shared_ptr<ReferenceProxy>;
+    using OnNoRefCBF = std::function<void(Ptr)>;
     ReferenceProxy() = default;
     virtual ~ReferenceProxy() = default;
 
@@ -42,7 +42,7 @@ class ReferenceProxy {
         }
         if (ref_count_.fetch_sub(1) == 1) {
             for (auto& cb : on_no_ref_cbs_) {
-                cb();
+                cb(this->shared_from_this());
             }
         }
     }

core/src/db/snapshot/ResourceHolder.h

@@ -123,7 +123,7 @@ class ResourceHolder {
             return false;
         }
         id_map_[resource->GetID()] = resource;
-        resource->RegisterOnNoRefCB(std::bind(&Derived::OnNoRefCallBack, this, resource));
+        resource->RegisterOnNoRefCB(std::bind(&Derived::OnNoRefCallBack, this, std::placeholders::_1));
         return true;
     }
 
@@ -138,10 +138,14 @@ class ResourceHolder {
     }
 
     virtual void
-    OnNoRefCallBack(ResourcePtr resource) {
-        resource->Deactivate();
-        Release(resource->GetID());
-        auto evt_ptr = std::make_shared<ResourceGCEvent<ResourceT>>(resource);
+    OnNoRefCallBack(ReferenceProxy::Ptr resource) {
+        auto res = std::dynamic_pointer_cast<ResourceT>(resource);
+        if (!res) {
+            return;
+        }
+        res->Deactivate();
+        Release(res->GetID());
+        auto evt_ptr = std::make_shared<ResourceGCEvent<ResourceT>>(res);
         EventExecutor::GetInstance().Submit(evt_ptr);
     }
 

core/unittest/db/test_snapshot.cpp

@@ -44,25 +44,25 @@ TEST_F(SnapshotTest, ResourcesTest) {
 TEST_F(SnapshotTest, ReferenceProxyTest) {
     std::string status("raw");
     const std::string CALLED = "CALLED";
-    auto callback = [&]() {
+    auto callback = [&](ReferenceProxy::Ptr) {
         status = CALLED;
     };
 
-    auto proxy = ReferenceProxy();
-    ASSERT_EQ(proxy.ref_count(), 0);
+    auto proxy = std::make_shared<ReferenceProxy>();
+    ASSERT_EQ(proxy->ref_count(), 0);
 
     int refcnt = 3;
     for (auto i = 0; i < refcnt; ++i) {
-        proxy.Ref();
+        proxy->Ref();
     }
-    ASSERT_EQ(proxy.ref_count(), refcnt);
+    ASSERT_EQ(proxy->ref_count(), refcnt);
 
-    proxy.RegisterOnNoRefCB(callback);
+    proxy->RegisterOnNoRefCB(callback);
 
     for (auto i = 0; i < refcnt; ++i) {
-        proxy.UnRef();
+        proxy->UnRef();
     }
-    ASSERT_EQ(proxy.ref_count(), 0);
+    ASSERT_EQ(proxy->ref_count(), 0);
     ASSERT_EQ(status, CALLED);
 }